ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject incubator-ignite git commit: ignite-430 Implement IgniteSocketStreamer to stream data from TCP socket.
Date Mon, 20 Apr 2015 21:50:36 GMT
Repository: incubator-ignite
Updated Branches:
  refs/heads/ignite-430-1 262e50393 -> 5f4a5ccae


ignite-430 Implement IgniteSocketStreamer to stream data from TCP socket.


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

Branch: refs/heads/ignite-430-1
Commit: 5f4a5ccaebfc00971f3425b129722426810e1d1e
Parents: 262e503
Author: agura <agura@gridgain.com>
Authored: Tue Apr 21 00:04:53 2015 +0300
Committer: agura <agura@gridgain.com>
Committed: Tue Apr 21 00:04:53 2015 +0300

----------------------------------------------------------------------
 .../streaming/socket/SocketStreamerExample.java |  2 +
 .../socket/ZStringsSocketStreamerExample.java   |  2 +
 .../internal/util/nio/GridBufferedParser.java   |  4 --
 .../util/nio/GridNioDelimitedBuffer.java        | 41 +++++++-------------
 .../ignite/stream/adapters/StreamAdapter.java   | 17 ++++++++
 .../stream/socket/IgniteSocketStreamer.java     | 20 +++-------
 .../socket/IgniteSocketStreamerSelfTest.java    |  2 +
 7 files changed, 43 insertions(+), 45 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/5f4a5cca/examples/src/main/java/org/apache/ignite/examples/streaming/socket/SocketStreamerExample.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java/org/apache/ignite/examples/streaming/socket/SocketStreamerExample.java
b/examples/src/main/java/org/apache/ignite/examples/streaming/socket/SocketStreamerExample.java
index 1ee916f..73cb970 100644
--- a/examples/src/main/java/org/apache/ignite/examples/streaming/socket/SocketStreamerExample.java
+++ b/examples/src/main/java/org/apache/ignite/examples/streaming/socket/SocketStreamerExample.java
@@ -89,6 +89,8 @@ public class SocketStreamerExample {
 
                 sockStmr.setPort(PORT);
 
+                sockStmr.setIgnite(ignite);
+
                 sockStmr.setStreamer(stmr);
 
                 sockStmr.setTupleExtractor(new StreamTupleExtractor<Tuple, Integer, Long>()
{

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/5f4a5cca/examples/src/main/java/org/apache/ignite/examples/streaming/socket/ZStringsSocketStreamerExample.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java/org/apache/ignite/examples/streaming/socket/ZStringsSocketStreamerExample.java
b/examples/src/main/java/org/apache/ignite/examples/streaming/socket/ZStringsSocketStreamerExample.java
index 9fd229e..a535c73 100644
--- a/examples/src/main/java/org/apache/ignite/examples/streaming/socket/ZStringsSocketStreamerExample.java
+++ b/examples/src/main/java/org/apache/ignite/examples/streaming/socket/ZStringsSocketStreamerExample.java
@@ -98,6 +98,8 @@ public class ZStringsSocketStreamerExample {
 
                 sockStmr.setDelimiter(DELIM);
 
+                sockStmr.setIgnite(ignite);
+
                 sockStmr.setStreamer(stmr);
 
                 // Converter from zero-terminated string to Java strings.

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/5f4a5cca/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridBufferedParser.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridBufferedParser.java
b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridBufferedParser.java
index 3f81dc4..a03d2c8 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridBufferedParser.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridBufferedParser.java
@@ -18,7 +18,6 @@
 package org.apache.ignite.internal.util.nio;
 
 import org.apache.ignite.*;
-import org.apache.ignite.internal.util.typedef.internal.*;
 
 import java.io.*;
 import java.nio.*;
@@ -33,9 +32,6 @@ import java.nio.*;
  *     | MSG_SIZE  |   MESSAGE  | MSG_SIZE  |   MESSAGE  |
  *     +--+--+--+--+--+--+...+--+--+--+--+--+--+--+...+--+
  * </pre>
- * <p>
- * It expects that first 4 bytes in stream are {@link U#IGNITE_HEADER}. If beginning of a
stream,
- * isn't equal to these bytes than exception will be thrown.
  */
 public class GridBufferedParser implements GridNioParser {
     /** Buffer metadata key. */

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/5f4a5cca/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioDelimitedBuffer.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioDelimitedBuffer.java
b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioDelimitedBuffer.java
index bbcedf2..230297d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioDelimitedBuffer.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioDelimitedBuffer.java
@@ -26,18 +26,18 @@ import java.util.*;
  * Buffer with message delimiter support.
  */
 public class GridNioDelimitedBuffer {
-    /** Buffer size. */
-    private static final int BUFFER_SIZE = 512;
-
     /** Delimiter. */
     private final byte[] delim;
 
     /** Data. */
-    private byte[] data;
+    private byte[] data = new byte[512];
 
     /** Count. */
     private int cnt;
 
+    /** Index. */
+    private int idx;
+
     /**
      * @param delim Delimiter.
      */
@@ -55,8 +55,7 @@ public class GridNioDelimitedBuffer {
      */
     private void reset() {
         cnt = 0;
-
-        data = new byte[BUFFER_SIZE];
+        idx = 0;
     }
 
     /**
@@ -64,14 +63,20 @@ public class GridNioDelimitedBuffer {
      * @return Message bytes or {@code null} if message is not fully read yet.
      */
     @Nullable public byte[] read(ByteBuffer buf) {
-        for(; buf.hasRemaining();) {
-
+        while(buf.hasRemaining()) {
             if (cnt == data.length)
                 data = Arrays.copyOf(data, data.length * 2);
 
-            data[cnt++] = buf.get();
+            byte b = buf.get();
 
-            if (cnt >= delim.length && found()) {
+            data[cnt++] = b;
+
+            if (b == delim[idx])
+                idx++;
+            else
+                idx = (b == delim[0]) ? 1 : 0;
+
+            if (idx == delim.length) {
                 byte[] bytes = Arrays.copyOfRange(data, 0, cnt - delim.length);
 
                 reset();
@@ -82,20 +87,4 @@ public class GridNioDelimitedBuffer {
 
         return null;
     }
-
-    /**
-     * Tries find delimiter in buffer.
-     *
-     * @return {@code True} if delimiter found, {@code false} - otherwise.
-     */
-    private boolean found() {
-        int from = cnt - delim.length;
-
-        for (int i = 0; i < delim.length ; i++) {
-            if (delim[i] != data[from + i])
-                return false;
-        }
-
-        return true;
-    }
 }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/5f4a5cca/modules/core/src/main/java/org/apache/ignite/stream/adapters/StreamAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/stream/adapters/StreamAdapter.java
b/modules/core/src/main/java/org/apache/ignite/stream/adapters/StreamAdapter.java
index c729362..b99521a 100644
--- a/modules/core/src/main/java/org/apache/ignite/stream/adapters/StreamAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/stream/adapters/StreamAdapter.java
@@ -34,6 +34,9 @@ public abstract class StreamAdapter<T, K, V> {
     /** Streamer. */
     private IgniteDataStreamer<K, V> stmr;
 
+    /** Ignite. */
+    private Ignite ignite;
+
     /**
      * Empty constructor.
      */
@@ -81,6 +84,20 @@ public abstract class StreamAdapter<T, K, V> {
     }
 
     /**
+     * @return Provided {@link Ignite} instance.
+     */
+    public Ignite getIgnite() {
+        return ignite;
+    }
+
+    /**
+     * @param ignite {@link Ignite} instance.
+     */
+    public void setIgnite(Ignite ignite) {
+        this.ignite = ignite;
+    }
+
+    /**
      * Converts given message to a tuple and adds it to the underlying streamer.
      *
      * @param msg Message to convert.

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/5f4a5cca/modules/core/src/main/java/org/apache/ignite/stream/socket/IgniteSocketStreamer.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/stream/socket/IgniteSocketStreamer.java
b/modules/core/src/main/java/org/apache/ignite/stream/socket/IgniteSocketStreamer.java
index f506af3..66369ea 100644
--- a/modules/core/src/main/java/org/apache/ignite/stream/socket/IgniteSocketStreamer.java
+++ b/modules/core/src/main/java/org/apache/ignite/stream/socket/IgniteSocketStreamer.java
@@ -18,18 +18,15 @@
 package org.apache.ignite.stream.socket;
 
 import org.apache.ignite.*;
-import org.apache.ignite.internal.processors.datastreamer.*;
 import org.apache.ignite.internal.util.nio.*;
 import org.apache.ignite.internal.util.typedef.*;
 import org.apache.ignite.internal.util.typedef.internal.*;
 import org.apache.ignite.marshaller.jdk.*;
 import org.apache.ignite.stream.adapters.*;
-
 import org.jetbrains.annotations.*;
 
 import java.net.*;
 import java.nio.*;
-import java.util.*;
 
 /**
  * Server that receives data from TCP socket, converts it to key-value pairs using {@link
StreamTupleExtractor} and
@@ -132,15 +129,12 @@ public class IgniteSocketStreamer<T, K, V> extends StreamAdapter<T,
K, V> {
      * @throws IgniteException If failed.
      */
     public void start() {
-        A.notNull(getTupleExtractor(), "tupleExtractor is null");
-        A.notNull(getStreamer(), "streamer is null");
+        A.notNull(getTupleExtractor(), "tupleExtractor");
+        A.notNull(getStreamer(), "streamer");
+        A.notNull(getIgnite(), "ignite");
         A.ensure(threads > 0, "threads > 0");
 
-        UUID uuid = ((DataStreamerImpl)getStreamer()).cacheObjectContext().kernalContext().localNodeId();
-
-        Ignite ignite = Ignition.ignite(uuid);
-
-        log = ignite.log();
+        log = getIgnite().log();
 
         GridNioServerListener<byte[]> lsnr = new GridNioServerListenerAdapter<byte[]>()
{
             @Override public void onConnected(GridNioSession ses) {
@@ -156,11 +150,7 @@ public class IgniteSocketStreamer<T, K, V> extends StreamAdapter<T,
K, V> {
             }
 
             @Override public void onMessage(GridNioSession ses, byte[] msg) {
-                T obj = converter.convert(msg);
-
-                Map.Entry<K, V> e = getTupleExtractor().extract(obj);
-
-                getStreamer().addData(e);
+                addMessage(converter.convert(msg));
             }
         };
 

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/5f4a5cca/modules/core/src/test/java/org/apache/ignite/stream/socket/IgniteSocketStreamerSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/stream/socket/IgniteSocketStreamerSelfTest.java
b/modules/core/src/test/java/org/apache/ignite/stream/socket/IgniteSocketStreamerSelfTest.java
index 6d8a9b4..19852ce 100644
--- a/modules/core/src/test/java/org/apache/ignite/stream/socket/IgniteSocketStreamerSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/stream/socket/IgniteSocketStreamerSelfTest.java
@@ -244,6 +244,8 @@ public class IgniteSocketStreamerSelfTest extends GridCommonAbstractTest
{
 
             IgniteCache<Integer, String> cache = ignite.cache(null);
 
+            sockStmr.setIgnite(ignite);
+
             sockStmr.setStreamer(stmr);
 
             sockStmr.setPort(port);


Mime
View raw message