ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From vkuliche...@apache.org
Subject incubator-ignite git commit: # IGNITE-45 - Examples
Date Sun, 22 Mar 2015 04:02:00 GMT
Repository: incubator-ignite
Updated Branches:
  refs/heads/ignite-45 4230e03a1 -> cc3c939d2


# IGNITE-45 - Examples


Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/cc3c939d
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/cc3c939d
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/cc3c939d

Branch: refs/heads/ignite-45
Commit: cc3c939d2978d54d254c5a0aa887529d63ef75c0
Parents: 4230e03
Author: Valentin Kulichenko <vkulichenko@gridgain.com>
Authored: Sat Mar 21 21:03:35 2015 -0700
Committer: Valentin Kulichenko <vkulichenko@gridgain.com>
Committed: Sat Mar 21 21:03:35 2015 -0700

----------------------------------------------------------------------
 .../streaming/numbers/StreamRandomNumbers.java  |  9 +++----
 .../streaming/numbers/StreamRandomNumbers.java  |  2 +-
 .../apache/ignite/stream/StreamTransformer.java | 26 +++++++++++---------
 3 files changed, 19 insertions(+), 18 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/cc3c939d/examples/src/main/java/org/apache/ignite/examples/streaming/numbers/StreamRandomNumbers.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java/org/apache/ignite/examples/streaming/numbers/StreamRandomNumbers.java
b/examples/src/main/java/org/apache/ignite/examples/streaming/numbers/StreamRandomNumbers.java
index efb5f28..f1e55d2 100644
--- a/examples/src/main/java/org/apache/ignite/examples/streaming/numbers/StreamRandomNumbers.java
+++ b/examples/src/main/java/org/apache/ignite/examples/streaming/numbers/StreamRandomNumbers.java
@@ -18,7 +18,6 @@
 package org.apache.ignite.examples.streaming.numbers;
 
 import org.apache.ignite.*;
-import org.apache.ignite.cache.*;
 import org.apache.ignite.examples.*;
 import org.apache.ignite.stream.*;
 
@@ -59,16 +58,16 @@ public class StreamRandomNumbers {
                 stmr.allowOverwrite(true);
 
                 // Configure data transformation to count instances of the same word.
-                stmr.receiver(new StreamTransformer<>(new CacheEntryProcessor<Integer,
Long, Object>() {
-                    @Override
-                    public Object process(MutableEntry<Integer, Long> e, Object...
arg) {
+                stmr.receiver(new StreamTransformer<Integer, Long>() {
+                    @Override public Object process(MutableEntry<Integer, Long> e,
Object... objects)
+                        throws EntryProcessorException {
                         Long val = e.getValue();
 
                         e.setValue(val == null ? 1L : val + 1);
 
                         return null;
                     }
-                }));
+                });
 
                 // Stream random numbers into the streamer cache.
                 while (true)

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/cc3c939d/examples/src/main/java8/org/apache/ignite/examples/java8/streaming/numbers/StreamRandomNumbers.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java8/org/apache/ignite/examples/java8/streaming/numbers/StreamRandomNumbers.java
b/examples/src/main/java8/org/apache/ignite/examples/java8/streaming/numbers/StreamRandomNumbers.java
index 25ab99b..630f4e3 100644
--- a/examples/src/main/java8/org/apache/ignite/examples/java8/streaming/numbers/StreamRandomNumbers.java
+++ b/examples/src/main/java8/org/apache/ignite/examples/java8/streaming/numbers/StreamRandomNumbers.java
@@ -57,7 +57,7 @@ public class StreamRandomNumbers {
                 stmr.allowOverwrite(true);
 
                 // Configure data transformation to count instances of the same word.
-                stmr.receiver(new StreamTransformer<>((e, arg) -> {
+                stmr.receiver(StreamTransformer.from((e, arg) -> {
                     Long val = e.getValue();
 
                     e.setValue(val == null ? 1L : val + 1);

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/cc3c939d/modules/core/src/main/java/org/apache/ignite/stream/StreamTransformer.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/stream/StreamTransformer.java b/modules/core/src/main/java/org/apache/ignite/stream/StreamTransformer.java
index 8c85e5d..3f22299 100644
--- a/modules/core/src/main/java/org/apache/ignite/stream/StreamTransformer.java
+++ b/modules/core/src/main/java/org/apache/ignite/stream/StreamTransformer.java
@@ -27,25 +27,27 @@ import java.util.*;
  * Convenience adapter to transform update existing values in streaming cache
  * based on the previously cached value.
  */
-public class StreamTransformer<K, V> implements StreamReceiver<K, V> {
+public abstract class StreamTransformer<K, V> implements StreamReceiver<K, V>,
EntryProcessor<K, V, Object> {
     /** */
     private static final long serialVersionUID = 0L;
 
-    /** Entry processor. */
-    private EntryProcessor<K, V, Object> ep;
+    /** {@inheritDoc} */
+    @Override public void receive(IgniteCache<K, V> cache, Collection<Map.Entry<K,
V>> entries) throws IgniteException {
+        for (Map.Entry<K, V> entry : entries)
+            cache.invoke(entry.getKey(), this);
+    }
 
     /**
-     * Entry processor to update cache values based on the previously cached value.
+     * Creates a new transformer based on instance of {@link CacheEntryProcessor}.
      *
      * @param ep Entry processor.
+     * @return Stream transformer.
      */
-    public StreamTransformer(CacheEntryProcessor<K, V, Object> ep) {
-        this.ep = ep;
-    }
-
-    /** {@inheritDoc} */
-    @Override public void receive(IgniteCache<K, V> cache, Collection<Map.Entry<K,
V>> entries) throws IgniteException {
-        for (Map.Entry<K, V> entry : entries)
-            cache.invoke(entry.getKey(), ep);
+    public static <K, V> StreamTransformer<K, V> from(final CacheEntryProcessor<K,
V, Object> ep) {
+        return new StreamTransformer<K, V>() {
+            @Override public Object process(MutableEntry<K, V> entry, Object... args)
throws EntryProcessorException {
+                return ep.process(entry, args);
+            }
+        };
     }
 }


Mime
View raw message