You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sb...@apache.org on 2015/05/15 03:09:18 UTC
[1/9] incubator-ignite git commit: # ignite-669 - streaming design.
Repository: incubator-ignite
Updated Branches:
refs/heads/ignite-sprint-5 56e67e8f7 -> 896b426bf
# ignite-669 - streaming design.
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/0cbe3c68
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/0cbe3c68
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/0cbe3c68
Branch: refs/heads/ignite-sprint-5
Commit: 0cbe3c68dcea09a39fd0e97909bbbaa6b1091a6b
Parents: 56e67e8
Author: Dmitiry Setrakyan <ds...@gridgain.com>
Authored: Thu Apr 2 02:45:48 2015 -0700
Committer: agura <ag...@gridgain.com>
Committed: Fri May 15 03:44:15 2015 +0300
----------------------------------------------------------------------
.../ignite/stream/adapters/StreamAdapter.java | 73 ++++++++++++++++++++
.../stream/adapters/StreamTupleExtractor.java | 33 +++++++++
.../ignite/stream/adapters/package-info.java | 21 ++++++
3 files changed, 127 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0cbe3c68/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
new file mode 100644
index 0000000..02ae795
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/stream/adapters/StreamAdapter.java
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.stream.adapters;
+
+import org.apache.ignite.*;
+
+import java.util.*;
+
+/**
+ * Convenience adapter for streamers. Adapters are optional components for
+ * streaming from different data sources. The purpose of adapters is to
+ * convert different message formats into Ignite stream key-value tuples
+ * and feed the tuples into the provided {@link org.apache.ignite.IgniteDataStreamer}.
+ */
+public abstract class StreamAdapter<T, K, V> {
+ /** Tuple extractor. */
+ private final StreamTupleExtractor<T, K, V> extractor;
+
+ /** Streamer. */
+ private final IgniteDataStreamer<K, V> stmr;
+
+ /**
+ * Stream adapter.
+ *
+ * @param stmr Streamer.
+ * @param extractor Tuple extractor.
+ */
+ protected StreamAdapter(IgniteDataStreamer<K, V> stmr, StreamTupleExtractor<T, K, V> extractor) {
+ this.stmr = stmr;
+ this.extractor = extractor;
+ }
+
+ /**
+ * @return Provided data streamer.
+ */
+ public IgniteDataStreamer<K, V> streamer() {
+ return stmr;
+ }
+
+ /**
+ * @return Provided tuple extractor.
+ */
+ public StreamTupleExtractor<T, K, V> converter() {
+ return extractor;
+ }
+
+ /**
+ * Converts given message to a tuple and adds it to the underlying streamer.
+ *
+ * @param msg Message to convert.
+ */
+ protected void addMessage(T msg) {
+ Map.Entry<K, V> e = extractor.extract(msg);
+
+ if (e != null)
+ stmr.addData(e);
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0cbe3c68/modules/core/src/main/java/org/apache/ignite/stream/adapters/StreamTupleExtractor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/stream/adapters/StreamTupleExtractor.java b/modules/core/src/main/java/org/apache/ignite/stream/adapters/StreamTupleExtractor.java
new file mode 100644
index 0000000..9b0c395
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/stream/adapters/StreamTupleExtractor.java
@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.stream.adapters;
+
+import java.util.*;
+
+/**
+ * Stream tuple extractor to convert messages to Ignite key-value tuples.
+ */
+public interface StreamTupleExtractor<T, K, V> {
+ /**
+ * Extracts a key-value tuple from a message.
+ *
+ * @param msg Message.
+ * @return Key-value tuple.
+ */
+ public Map.Entry<K, V> extract(T msg);
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0cbe3c68/modules/core/src/main/java/org/apache/ignite/stream/adapters/package-info.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/stream/adapters/package-info.java b/modules/core/src/main/java/org/apache/ignite/stream/adapters/package-info.java
new file mode 100644
index 0000000..a69ffc0
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/stream/adapters/package-info.java
@@ -0,0 +1,21 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * Contains Ignite stream adapters.
+ */
+package org.apache.ignite.stream.adapters;
[5/9] incubator-ignite git commit: ignite-430 Implement
IgniteSocketStreamer to stream data from TCP socket.
Posted by sb...@apache.org.
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/53995dcb
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/53995dcb
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/53995dcb
Branch: refs/heads/ignite-sprint-5
Commit: 53995dcb3470de07df73d8dfd284da0dbb8df8dd
Parents: b0fbfa0
Author: agura <ag...@gridgain.com>
Authored: Mon Apr 13 18:28:40 2015 +0300
Committer: agura <ag...@gridgain.com>
Committed: Fri May 15 03:44:27 2015 +0300
----------------------------------------------------------------------
.../streaming/socket/SocketStreamerExample.java | 158 ++++++++++
.../socket/ZStringsSocketStreamerExample.java | 151 +++++++++
.../examples/streaming/socket/package-info.java | 21 ++
.../internal/util/nio/GridBufferedParser.java | 4 -
.../internal/util/nio/GridDelimitedParser.java | 91 ++++++
.../util/nio/GridNioDelimitedBuffer.java | 106 +++++++
.../ignite/stream/adapters/StreamAdapter.java | 17 +
.../stream/socket/IgniteSocketStreamer.java | 217 +++++++++++++
.../stream/socket/SocketMessageConverter.java | 31 ++
.../ignite/stream/socket/package-info.java | 21 ++
.../util/nio/GridNioDelimitedBufferTest.java | 112 +++++++
.../socket/IgniteSocketStreamerSelfTest.java | 315 +++++++++++++++++++
.../ignite/stream/socket/package-info.java | 21 ++
.../testsuites/IgniteStreamTestSuite.java | 39 +++
.../testsuites/IgniteUtilSelfTestSuite.java | 1 +
15 files changed, 1301 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/53995dcb/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
new file mode 100644
index 0000000..73cb970
--- /dev/null
+++ b/examples/src/main/java/org/apache/ignite/examples/streaming/socket/SocketStreamerExample.java
@@ -0,0 +1,158 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.examples.streaming.socket;
+
+import org.apache.ignite.*;
+import org.apache.ignite.examples.*;
+import org.apache.ignite.examples.streaming.numbers.*;
+import org.apache.ignite.lang.*;
+import org.apache.ignite.stream.*;
+import org.apache.ignite.stream.adapters.*;
+import org.apache.ignite.stream.socket.*;
+
+import javax.cache.processor.*;
+import java.io.*;
+import java.net.*;
+import java.util.*;
+
+/**
+ * Streams random numbers into the streaming cache using {@link IgniteSocketStreamer}.
+ * To start the example, you should:
+ * <ul>
+ * <li>Start a few nodes using {@link ExampleNodeStartup} or by starting remote nodes as specified below.</li>
+ * <li>Start streaming using {@link SocketStreamerExample}.</li>
+ * <li>Start querying popular numbers using {@link QueryPopularNumbers}.</li>
+ * </ul>
+ * <p>
+ * You should start remote nodes by running {@link ExampleNodeStartup} in another JVM.
+ */
+public class SocketStreamerExample {
+ /** Random number generator. */
+ private static final Random RAND = new Random();
+
+ /** Range within which to generate numbers. */
+ private static final int RANGE = 1000;
+
+ /** Port. */
+ private static final int PORT = 5555;
+
+ /**
+ * @param args Args.
+ */
+ public static void main(String[] args) throws InterruptedException, IOException {
+ // Mark this cluster member as client.
+ Ignition.setClientMode(true);
+
+ try (Ignite ignite = Ignition.start("examples/config/example-ignite.xml")) {
+ if (!ExamplesUtils.hasServerNodes(ignite))
+ return;
+
+ // The cache is configured with sliding window holding 1 second of the streaming data.
+ IgniteCache<Integer, Long> stmCache = ignite.getOrCreateCache(CacheConfig.randomNumbersCache());
+
+ try (IgniteDataStreamer<Integer, Long> stmr = ignite.dataStreamer(stmCache.getName())) {
+ // Allow data updates.
+ stmr.allowOverwrite(true);
+
+ // Configure data transformation to count instances of the same word.
+ 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;
+ }
+ });
+
+ InetAddress addr = InetAddress.getLocalHost();
+
+ IgniteSocketStreamer<Tuple, Integer, Long> sockStmr = new IgniteSocketStreamer<>();
+
+ sockStmr.setAddr(addr);
+
+ sockStmr.setPort(PORT);
+
+ sockStmr.setIgnite(ignite);
+
+ sockStmr.setStreamer(stmr);
+
+ sockStmr.setTupleExtractor(new StreamTupleExtractor<Tuple, Integer, Long>() {
+ @Override public Map.Entry<Integer, Long> extract(Tuple tuple) {
+ return new IgniteBiTuple<>(tuple.key, tuple.cnt);
+ }
+ });
+
+ sockStmr.start();
+
+ sendData(addr, PORT);
+ }
+ }
+ }
+
+ /**
+ * @param addr Address.
+ * @param port Port.
+ */
+ private static void sendData(InetAddress addr, int port) throws IOException, InterruptedException {
+ try (Socket sock = new Socket(addr, port);
+ OutputStream oos = new BufferedOutputStream(sock.getOutputStream())) {
+ while (true) {
+ try (ByteArrayOutputStream bos = new ByteArrayOutputStream();
+ ObjectOutputStream out = new ObjectOutputStream(bos)) {
+ Tuple tuple = new Tuple(RAND.nextInt(RANGE), 1L);
+
+ out.writeObject(tuple);
+
+ byte[] arr = bos.toByteArray();
+
+ oos.write(arr.length >>> 24);
+ oos.write(arr.length >>> 16);
+ oos.write(arr.length >>> 8);
+ oos.write(arr.length);
+
+ oos.write(arr);
+ }
+ }
+ }
+ }
+
+ /**
+ * Tuple.
+ */
+ private static class Tuple implements Serializable {
+ /** Serial version uid. */
+ private static final long serialVersionUID = 0;
+
+ /** Key. */
+ private final int key;
+
+ /** Count. */
+ private final long cnt;
+
+ /**
+ * @param key Key.
+ * @param cnt Count.
+ */
+ public Tuple(int key, long cnt) {
+ this.key = key;
+ this.cnt = cnt;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/53995dcb/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
new file mode 100644
index 0000000..a535c73
--- /dev/null
+++ b/examples/src/main/java/org/apache/ignite/examples/streaming/socket/ZStringsSocketStreamerExample.java
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.examples.streaming.socket;
+
+import org.apache.ignite.*;
+import org.apache.ignite.examples.*;
+import org.apache.ignite.examples.streaming.numbers.*;
+import org.apache.ignite.lang.*;
+import org.apache.ignite.stream.*;
+import org.apache.ignite.stream.adapters.*;
+import org.apache.ignite.stream.socket.*;
+
+import javax.cache.processor.*;
+import java.io.*;
+import java.net.*;
+import java.util.*;
+
+/**
+ * Stream random numbers into the streaming cache using {@link IgniteSocketStreamer}.
+ * <p>
+ * Example illustrates usage of TCP socket streamer in case of non-Java clients. In this example client streams
+ * zero-terminated strings.
+ * <p>
+ * To start the example, you should:
+ * <ul>
+ * <li>Start a few nodes using {@link ExampleNodeStartup} or by starting remote nodes as specified below.</li>
+ * <li>Start streaming using {@link ZStringsSocketStreamerExample}.</li>
+ * <li>Start querying popular numbers using {@link QueryPopularNumbers}.</li>
+ * </ul>
+ * <p>
+ * You should start remote nodes by running {@link ExampleNodeStartup} in another JVM.
+ */
+public class ZStringsSocketStreamerExample {
+ /** Random number generator. */
+ private static final Random RAND = new Random();
+
+ /** Range within which to generate numbers. */
+ private static final int RANGE = 1000;
+
+ /** Port. */
+ private static final int PORT = 5555;
+
+ /** Delimiter. */
+ private static final byte[] DELIM = new byte[] {0};
+
+ /**
+ * @param args Args.
+ */
+ public static void main(String[] args) throws InterruptedException, IOException {
+ // Mark this cluster member as client.
+ Ignition.setClientMode(true);
+
+ try (Ignite ignite = Ignition.start("examples/config/example-ignite.xml")) {
+ if (!ExamplesUtils.hasServerNodes(ignite))
+ return;
+
+ // The cache is configured with sliding window holding 1 second of the streaming data.
+ IgniteCache<Integer, Long> stmCache = ignite.getOrCreateCache(CacheConfig.randomNumbersCache());
+
+ try (IgniteDataStreamer<Integer, Long> stmr = ignite.dataStreamer(stmCache.getName())) {
+ // Allow data updates.
+ stmr.allowOverwrite(true);
+
+ // Configure data transformation to count instances of the same word.
+ 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;
+ }
+ });
+
+ InetAddress addr = InetAddress.getLocalHost();
+
+ IgniteSocketStreamer<String, Integer, Long> sockStmr = new IgniteSocketStreamer<>();
+
+ sockStmr.setAddr(addr);
+
+ sockStmr.setPort(PORT);
+
+ sockStmr.setDelimiter(DELIM);
+
+ sockStmr.setIgnite(ignite);
+
+ sockStmr.setStreamer(stmr);
+
+ // Converter from zero-terminated string to Java strings.
+ sockStmr.setConverter(new SocketMessageConverter<String>() {
+ @Override public String convert(byte[] msg) {
+ try {
+ return new String(msg, "ASCII");
+ }
+ catch (UnsupportedEncodingException e) {
+ throw new IgniteException(e);
+ }
+ }
+ });
+
+ sockStmr.setTupleExtractor(new StreamTupleExtractor<String, Integer, Long>() {
+ @Override public Map.Entry<Integer, Long> extract(String input) {
+ String[] pair = input.split("=");
+ return new IgniteBiTuple<>(Integer.parseInt(pair[0]), Long.parseLong(pair[1]));
+ }
+ });
+
+ sockStmr.start();
+
+ sendData(addr, PORT);
+ }
+ }
+ }
+
+ /**
+ * @param addr Address.
+ * @param port Port.
+ */
+ private static void sendData(InetAddress addr, int port) throws IOException, InterruptedException {
+ try (Socket sock = new Socket(addr, port);
+ OutputStream oos = new BufferedOutputStream(sock.getOutputStream())) {
+
+ while (true) {
+ int key = RAND.nextInt(RANGE);
+
+ String str = key + "=1";
+
+ byte[] arr = str.getBytes("ASCII");
+
+ oos.write(arr);
+ oos.write(DELIM);
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/53995dcb/examples/src/main/java/org/apache/ignite/examples/streaming/socket/package-info.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java/org/apache/ignite/examples/streaming/socket/package-info.java b/examples/src/main/java/org/apache/ignite/examples/streaming/socket/package-info.java
new file mode 100644
index 0000000..ae7bdf9
--- /dev/null
+++ b/examples/src/main/java/org/apache/ignite/examples/streaming/socket/package-info.java
@@ -0,0 +1,21 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * Contains {@link org.apache.ignite.stream.socket.IgniteSocketStreamer} usage examples.
+ */
+package org.apache.ignite.examples.streaming.socket;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/53995dcb/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/53995dcb/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridDelimitedParser.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridDelimitedParser.java b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridDelimitedParser.java
new file mode 100644
index 0000000..256597c
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridDelimitedParser.java
@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.util.nio;
+
+import org.apache.ignite.*;
+
+import java.io.*;
+import java.nio.*;
+
+/**
+ * This class implements stream parser based on {@link GridNioDelimitedBuffer}.
+ * <p>
+ * The rule for this parser is that every message sent over the stream is appended with
+ * delimiter (bytes array). So, the stream structure is as follows:
+ * <pre>
+ * +--+--+...+--+--+--+--+--+--+--+...+--+--+--+--+--+-
+ * | MESSAGE | DELIMITER | MESSAGE | DELIMITER |
+ * +--+--+...+--+--+--+--+--+--+--+...+--+--+--+--+--+-
+ * </pre>
+ */
+public class GridDelimitedParser implements GridNioParser {
+ /** Buffer metadata key. */
+ private static final int BUF_META_KEY = GridNioSessionMetaKey.nextUniqueKey();
+
+ /** Delimiter. */
+ private final byte[] delim;
+
+ /** Direct buffer. */
+ private final boolean directBuf;
+
+ /**
+ * @param delim Delimiter.
+ * @param directBuf Direct buffer.
+ */
+ public GridDelimitedParser(byte[] delim, boolean directBuf) {
+ this.delim = delim;
+ this.directBuf = directBuf;
+ }
+
+ /** {@inheritDoc} */
+ @Override public byte[] decode(GridNioSession ses, ByteBuffer buf) throws IOException, IgniteCheckedException {
+ GridNioDelimitedBuffer nioBuf = ses.meta(BUF_META_KEY);
+
+ // Decode for a given session is called per one thread, so there should not be any concurrency issues.
+ // However, we make some additional checks.
+ if (nioBuf == null) {
+ nioBuf = new GridNioDelimitedBuffer(delim);
+
+ GridNioDelimitedBuffer old = ses.addMeta(BUF_META_KEY, nioBuf);
+
+ assert old == null;
+ }
+
+ return nioBuf.read(buf);
+ }
+
+ /** {@inheritDoc} */
+ @Override public ByteBuffer encode(GridNioSession ses, Object msg) throws IOException, IgniteCheckedException {
+ byte[] msg0 = (byte[])msg;
+
+ int cap = msg0.length + delim.length;
+ ByteBuffer res = directBuf ? ByteBuffer.allocateDirect(cap) : ByteBuffer.allocate(cap);
+
+ res.put(msg0);
+ res.put(delim);
+
+ res.flip();
+
+ return res;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return this.getClass().getSimpleName();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/53995dcb/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
new file mode 100644
index 0000000..2b764ec
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioDelimitedBuffer.java
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.util.nio;
+
+import org.jetbrains.annotations.*;
+
+import java.nio.*;
+import java.util.*;
+
+/**
+ * Buffer with message delimiter support.
+ */
+public class GridNioDelimitedBuffer {
+ /** Delimiter. */
+ private final byte[] delim;
+
+ /** Data. */
+ private byte[] data = new byte[16384];
+
+ /** Count. */
+ private int cnt;
+
+ /** Index. */
+ private int idx;
+
+ /**
+ * @param delim Delimiter.
+ */
+ public GridNioDelimitedBuffer(byte[] delim) {
+ assert delim != null;
+ assert delim.length > 0;
+
+ this.delim = delim;
+
+ reset();
+ }
+
+ /**
+ * Resets buffer state.
+ */
+ private void reset() {
+ cnt = 0;
+ idx = 0;
+ }
+
+ /**
+ * @param buf Buffer.
+ * @return Message bytes or {@code null} if message is not fully read yet.
+ */
+ @Nullable public byte[] read(ByteBuffer buf) {
+ while(buf.hasRemaining()) {
+ if (cnt == data.length)
+ data = Arrays.copyOf(data, data.length * 2);
+
+ byte b = buf.get();
+
+ data[cnt++] = b;
+
+ if (b == delim[idx])
+ idx++;
+ else if (idx > 0) {
+ int pos = cnt - idx;
+
+ idx = 0;
+
+ for (int i = pos; i < cnt; i++) {
+ if (data[pos] == delim[idx]) {
+ pos++;
+
+ idx++;
+ }
+ else {
+ pos = cnt - idx;
+
+ idx = 0;
+ }
+ }
+ }
+
+ if (idx == delim.length) {
+ byte[] bytes = Arrays.copyOfRange(data, 0, cnt - delim.length);
+
+ reset();
+
+ return bytes;
+ }
+ }
+
+ return null;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/53995dcb/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/53995dcb/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
new file mode 100644
index 0000000..66369ea
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/stream/socket/IgniteSocketStreamer.java
@@ -0,0 +1,217 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.stream.socket;
+
+import org.apache.ignite.*;
+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.*;
+
+/**
+ * Server that receives data from TCP socket, converts it to key-value pairs using {@link StreamTupleExtractor} and
+ * streams into {@link IgniteDataStreamer} instance.
+ * <p>
+ * By default server uses size-based message processing. That is every message sent over the socket is prepended with
+ * 4-byte integer header containing message size. If message delimiter is defined (see {@link #setDelimiter}) then
+ * delimiter-based message processing will be used. That is every message sent over the socket is appended with
+ * provided delimiter.
+ * <p>
+ * Received messages through socket converts to Java object using standard serialization. Conversion functionality
+ * can be customized via user defined {@link SocketMessageConverter} (e.g. in order to convert messages from
+ * non Java clients).
+ */
+public class IgniteSocketStreamer<T, K, V> extends StreamAdapter<T, K, V> {
+ /** Default threads. */
+ private static final int DFLT_THREADS = Runtime.getRuntime().availableProcessors();
+
+ /** Logger. */
+ private IgniteLogger log;
+
+ /** Address. */
+ private InetAddress addr;
+
+ /** Server port. */
+ private int port;
+
+ /** Threads number. */
+ private int threads = DFLT_THREADS;
+
+ /** Direct mode. */
+ private boolean directMode;
+
+ /** Delimiter. */
+ private byte[] delim;
+
+ /** Converter. */
+ private SocketMessageConverter<T> converter;
+
+ /** Server. */
+ private GridNioServer<byte[]> srv;
+
+ /**
+ * Sets server address.
+ *
+ * @param addr Address.
+ */
+ public void setAddr(InetAddress addr) {
+ this.addr = addr;
+ }
+
+ /**
+ * Sets port number.
+ *
+ * @param port Port.
+ */
+ public void setPort(int port) {
+ this.port = port;
+ }
+
+ /**
+ * Sets threadds amount.
+ *
+ * @param threads Threads.
+ */
+ public void setThreads(int threads) {
+ this.threads = threads;
+ }
+
+ /**
+ * Sets direct mode flag.
+ *
+ * @param directMode Direct mode.
+ */
+ public void setDirectMode(boolean directMode) {
+ this.directMode = directMode;
+ }
+
+ /**
+ * Sets message delimiter.
+ *
+ * @param delim Delimiter.
+ */
+ public void setDelimiter(byte[] delim) {
+ this.delim = delim;
+ }
+
+ /**
+ * Sets message converter.
+ *
+ * @param converter Converter.
+ */
+ public void setConverter(SocketMessageConverter<T> converter) {
+ this.converter = converter;
+ }
+
+ /**
+ * Starts streamer.
+ *
+ * @throws IgniteException If failed.
+ */
+ public void start() {
+ A.notNull(getTupleExtractor(), "tupleExtractor");
+ A.notNull(getStreamer(), "streamer");
+ A.notNull(getIgnite(), "ignite");
+ A.ensure(threads > 0, "threads > 0");
+
+ log = getIgnite().log();
+
+ GridNioServerListener<byte[]> lsnr = new GridNioServerListenerAdapter<byte[]>() {
+ @Override public void onConnected(GridNioSession ses) {
+ assert ses.accepted();
+
+ if (log.isDebugEnabled())
+ log.debug("Accepted connection: " + ses.remoteAddress());
+ }
+
+ @Override public void onDisconnected(GridNioSession ses, @Nullable Exception e) {
+ if (e != null)
+ log.error("Connection failed with exception", e);
+ }
+
+ @Override public void onMessage(GridNioSession ses, byte[] msg) {
+ addMessage(converter.convert(msg));
+ }
+ };
+
+ ByteOrder byteOrder = ByteOrder.BIG_ENDIAN;
+
+ GridNioParser parser = F.isEmpty(delim) ? new GridBufferedParser(directMode, byteOrder) :
+ new GridDelimitedParser(delim, directMode);
+
+ if (converter == null)
+ converter = new DefaultConverter<>();
+
+ GridNioFilter codec = new GridNioCodecFilter(parser, log, directMode);
+
+ GridNioFilter[] filters = new GridNioFilter[] {codec};
+
+ try {
+ srv = new GridNioServer.Builder<byte[]>()
+ .address(addr == null ? InetAddress.getLocalHost() : addr)
+ .port(port)
+ .listener(lsnr)
+ .logger(log)
+ .selectorCount(threads)
+ .byteOrder(byteOrder)
+ .filters(filters)
+ .build();
+ }
+ catch (IgniteCheckedException | UnknownHostException e) {
+ throw new IgniteException(e);
+ }
+
+ srv.start();
+
+ if (log.isDebugEnabled())
+ log.debug("Socket streaming server started on " + addr + ':' + port);
+ }
+
+ /**
+ * Stops streamer.
+ */
+ public void stop() {
+ srv.stop();
+
+ if (log.isDebugEnabled())
+ log.debug("Socket streaming server stopped");
+ }
+
+ /**
+ * Converts message to Java object using Jdk marshaller.
+ */
+ private static class DefaultConverter<T> implements SocketMessageConverter<T> {
+ /** Marshaller. */
+ private static final JdkMarshaller MARSH = new JdkMarshaller();
+
+ /** {@inheritDoc} */
+ @Override public T convert(byte[] msg) {
+ try {
+ return MARSH.unmarshal(msg, null);
+ }
+ catch (IgniteCheckedException e) {
+ throw new IgniteException(e);
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/53995dcb/modules/core/src/main/java/org/apache/ignite/stream/socket/SocketMessageConverter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/stream/socket/SocketMessageConverter.java b/modules/core/src/main/java/org/apache/ignite/stream/socket/SocketMessageConverter.java
new file mode 100644
index 0000000..8161d86
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/stream/socket/SocketMessageConverter.java
@@ -0,0 +1,31 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.stream.socket;
+
+/**
+ * Socket message converter.
+ */
+public interface SocketMessageConverter<T> {
+ /**
+ * Converter message represented by array of bytes to object.
+ *
+ * @param msg Message.
+ * @return Converted object.
+ */
+ public T convert(byte[] msg);
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/53995dcb/modules/core/src/main/java/org/apache/ignite/stream/socket/package-info.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/stream/socket/package-info.java b/modules/core/src/main/java/org/apache/ignite/stream/socket/package-info.java
new file mode 100644
index 0000000..e1cef65
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/stream/socket/package-info.java
@@ -0,0 +1,21 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * Contains socket streamer implementation.
+ */
+package org.apache.ignite.stream.socket;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/53995dcb/modules/core/src/test/java/org/apache/ignite/internal/util/nio/GridNioDelimitedBufferTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/util/nio/GridNioDelimitedBufferTest.java b/modules/core/src/test/java/org/apache/ignite/internal/util/nio/GridNioDelimitedBufferTest.java
new file mode 100644
index 0000000..a0dd2e5
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/util/nio/GridNioDelimitedBufferTest.java
@@ -0,0 +1,112 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.util.nio;
+
+import junit.framework.TestCase;
+
+import java.nio.*;
+import java.util.*;
+
+/**
+ * Tests for {@link GridNioDelimitedBuffer}.
+ */
+public class GridNioDelimitedBufferTest extends TestCase {
+ /** */
+ private static final String ASCII = "ASCII";
+
+ /**
+ * Tests simple delimiter (excluded from alphabet)
+ */
+ public void testReadZString() throws Exception {
+ Random rnd = new Random();
+
+ int buffSize = 0;
+
+ byte[] delim = new byte[] {0};
+
+ List<String> strs = new ArrayList<>(50);
+
+ for (int i = 0; i < 50; i++) {
+ int len = rnd.nextInt(128) + 1;
+
+ buffSize += len + delim.length;
+
+ StringBuilder sb = new StringBuilder(len);
+
+ for (int j = 0; j < len; j++)
+ sb.append((char)(rnd.nextInt(26) + 'a'));
+
+
+ strs.add(sb.toString());
+ }
+
+ ByteBuffer buff = ByteBuffer.allocate(buffSize);
+
+ for (String str : strs) {
+ buff.put(str.getBytes(ASCII));
+ buff.put(delim);
+ }
+
+ buff.flip();
+
+ byte[] msg;
+
+ GridNioDelimitedBuffer delimBuff = new GridNioDelimitedBuffer(delim);
+
+ List<String> res = new ArrayList<>(strs.size());
+
+ while ((msg = delimBuff.read(buff)) != null)
+ res.add(new String(msg, ASCII));
+
+ assertEquals(strs, res);
+ }
+
+ /**
+ * Tests compound delimiter (included to alphabet)
+ */
+ public void testDelim() throws Exception {
+ byte[] delim = "aabb".getBytes(ASCII);
+
+ List<String> strs = Arrays.asList("za", "zaa", "zaab", "zab", "zaabaababbbbabaab");
+
+ int buffSize = 0;
+
+ for (String str : strs)
+ buffSize += str.length() + delim.length;
+
+ ByteBuffer buff = ByteBuffer.allocate(buffSize);
+
+ for (String str : strs) {
+ buff.put(str.getBytes(ASCII));
+ buff.put(delim);
+ }
+
+ buff.flip();
+
+ byte[] msg;
+
+ GridNioDelimitedBuffer delimBuff = new GridNioDelimitedBuffer(delim);
+
+ List<String> res = new ArrayList<>(strs.size());
+
+ while ((msg = delimBuff.read(buff)) != null)
+ res.add(new String(msg, ASCII));
+
+ assertEquals(strs, res);
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/53995dcb/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
new file mode 100644
index 0000000..19852ce
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/stream/socket/IgniteSocketStreamerSelfTest.java
@@ -0,0 +1,315 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.stream.socket;
+
+import org.apache.ignite.*;
+import org.apache.ignite.cache.*;
+import org.apache.ignite.configuration.*;
+import org.apache.ignite.events.*;
+import org.apache.ignite.lang.*;
+import org.apache.ignite.marshaller.*;
+import org.apache.ignite.marshaller.jdk.*;
+import org.apache.ignite.spi.discovery.tcp.*;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.*;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.*;
+import org.apache.ignite.stream.adapters.*;
+import org.apache.ignite.testframework.junits.common.*;
+
+import org.jetbrains.annotations.*;
+
+import java.io.*;
+import java.net.*;
+import java.util.*;
+import java.util.concurrent.*;
+
+import static org.apache.ignite.events.EventType.*;
+
+/**
+ * Tests {@link IgniteSocketStreamer}.
+ */
+public class IgniteSocketStreamerSelfTest extends GridCommonAbstractTest {
+ /** IP finder. */
+ private static final TcpDiscoveryIpFinder IP_FINDER = new TcpDiscoveryVmIpFinder(true);
+
+ /** Grid count. */
+ private final static int GRID_CNT = 3;
+
+ /** Count. */
+ private static final int CNT = 500;
+
+ /** Delimiter. */
+ private static final byte[] DELIM = new byte[] {0, 1, 2, 3, 4, 5, 4, 3, 2, 1, 0};
+
+ /** Port. */
+ private static int port;
+
+ /** Ignite. */
+ private static Ignite ignite;
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration() throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration();
+
+ CacheConfiguration ccfg = cacheConfiguration(cfg, null);
+
+ cfg.setCacheConfiguration(ccfg);
+
+ TcpDiscoverySpi discoSpi = new TcpDiscoverySpi();
+
+ discoSpi.setIpFinder(IP_FINDER);
+
+ cfg.setDiscoverySpi(discoSpi);
+
+ return cfg;
+ }
+
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ ignite = startGrids(GRID_CNT);
+ ignite.<Integer, String>getOrCreateCache(defaultCacheConfiguration());
+
+ try (ServerSocket sock = new ServerSocket(0)) {
+ port = sock.getLocalPort();
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTestsStopped() throws Exception {
+ stopAllGrids();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTest() throws Exception {
+ ignite.cache(null).clear();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testSizeBasedDefaultConverter() throws Exception {
+ test(null, null, new Runnable() {
+ @Override public void run() {
+ try (Socket sock = new Socket(InetAddress.getLocalHost(), port);
+ OutputStream os = new BufferedOutputStream(sock.getOutputStream())) {
+ Marshaller marsh = new JdkMarshaller();
+
+ for (int i = 0; i < CNT; i++) {
+ byte[] msg = marsh.marshal(new Tuple(i));
+
+ os.write(msg.length >>> 24);
+ os.write(msg.length >>> 16);
+ os.write(msg.length >>> 8);
+ os.write(msg.length);
+
+ os.write(msg);
+ }
+ }
+ catch (IOException | IgniteCheckedException e) {
+ throw new IgniteException(e);
+ }
+ }
+ });
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testSizeBasedCustomConverter() throws Exception {
+ SocketMessageConverter<Tuple> converter = new SocketMessageConverter<Tuple>() {
+ @Override public Tuple convert(byte[] msg) {
+ int i = (msg[0] & 0xFF) << 24;
+ i |= (msg[1] & 0xFF) << 16;
+ i |= (msg[2] & 0xFF) << 8;
+ i |= msg[3] & 0xFF;
+
+ return new Tuple(i);
+ }
+ };
+
+ test(converter, null, new Runnable() {
+ @Override public void run() {
+ try(Socket sock = new Socket(InetAddress.getLocalHost(), port);
+ OutputStream os = new BufferedOutputStream(sock.getOutputStream())) {
+
+ for (int i = 0; i < CNT; i++) {
+ os.write(0);
+ os.write(0);
+ os.write(0);
+ os.write(4);
+
+ os.write(i >>> 24);
+ os.write(i >>> 16);
+ os.write(i >>> 8);
+ os.write(i);
+ }
+ }
+ catch (IOException e) {
+ throw new IgniteException(e);
+ }
+ }
+ });
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testDelimiterBasedDefaultConverter() throws Exception {
+ test(null, DELIM, new Runnable() {
+ @Override public void run() {
+ try(Socket sock = new Socket(InetAddress.getLocalHost(), port);
+ OutputStream os = new BufferedOutputStream(sock.getOutputStream())) {
+ Marshaller marsh = new JdkMarshaller();
+
+ for (int i = 0; i < CNT; i++) {
+ byte[] msg = marsh.marshal(new Tuple(i));
+
+ os.write(msg);
+ os.write(DELIM);
+ }
+ }
+ catch (IOException | IgniteCheckedException e) {
+ throw new IgniteException(e);
+ }
+ }
+ });
+
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testDelimiterBasedCustomConverter() throws Exception {
+ SocketMessageConverter<Tuple> converter = new SocketMessageConverter<Tuple>() {
+ @Override public Tuple convert(byte[] msg) {
+ int i = (msg[0] & 0xFF) << 24;
+ i |= (msg[1] & 0xFF) << 16;
+ i |= (msg[2] & 0xFF) << 8;
+ i |= msg[3] & 0xFF;
+
+ return new Tuple(i);
+ }
+ };
+
+ test(converter, DELIM, new Runnable() {
+ @Override public void run() {
+ try(Socket sock = new Socket(InetAddress.getLocalHost(), port);
+ OutputStream os = new BufferedOutputStream(sock.getOutputStream())) {
+
+ for (int i = 0; i < CNT; i++) {
+ os.write(i >>> 24);
+ os.write(i >>> 16);
+ os.write(i >>> 8);
+ os.write(i);
+
+ os.write(DELIM);
+ }
+ }
+ catch (IOException e) {
+ throw new IgniteException(e);
+ }
+ }
+ });
+ }
+
+ /**
+ * @param converter Converter.
+ * @param r Runnable..
+ */
+ private void test(@Nullable SocketMessageConverter<Tuple> converter, @Nullable byte[] delim, Runnable r) throws Exception
+ {
+ IgniteSocketStreamer<Tuple, Integer, String> sockStmr = null;
+
+ try (IgniteDataStreamer<Integer, String> stmr = ignite.dataStreamer(null)) {
+
+ stmr.allowOverwrite(true);
+ stmr.autoFlushFrequency(10);
+
+ sockStmr = new IgniteSocketStreamer<>();
+
+ IgniteCache<Integer, String> cache = ignite.cache(null);
+
+ sockStmr.setIgnite(ignite);
+
+ sockStmr.setStreamer(stmr);
+
+ sockStmr.setPort(port);
+
+ sockStmr.setDelimiter(delim);
+
+ sockStmr.setTupleExtractor(new StreamTupleExtractor<Tuple, Integer, String>() {
+ @Override public Map.Entry<Integer, String> extract(Tuple msg) {
+ return new IgniteBiTuple<>(msg.key, msg.val);
+ }
+ });
+
+ if (converter != null)
+ sockStmr.setConverter(converter);
+
+ final CountDownLatch latch = new CountDownLatch(CNT);
+
+ IgniteBiPredicate<UUID, CacheEvent> locLsnr = new IgniteBiPredicate<UUID, CacheEvent>() {
+ @Override public boolean apply(UUID uuid, CacheEvent evt) {
+ latch.countDown();
+
+ return true;
+ }
+ };
+
+ ignite.events(ignite.cluster().forCacheNodes(null)).remoteListen(locLsnr, null, EVT_CACHE_OBJECT_PUT);
+
+ sockStmr.start();
+
+ r.run();
+
+ latch.await();
+
+ assertEquals(CNT, cache.size(CachePeekMode.PRIMARY));
+
+ for (int i = 0; i < CNT; i++)
+ assertEquals(Integer.toString(i), cache.get(i));
+ }
+ finally {
+ if (sockStmr != null)
+ sockStmr.stop();
+ }
+
+ }
+
+ /**
+ * Tuple.
+ */
+ private static class Tuple implements Serializable {
+ /** Serial version uid. */
+ private static final long serialVersionUID = 0L;
+
+ /** Key. */
+ private final int key;
+
+ /** Value. */
+ private final String val;
+
+ /**
+ * @param key Key.
+ */
+ Tuple(int key) {
+ this.key = key;
+ this.val = Integer.toString(key);
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/53995dcb/modules/core/src/test/java/org/apache/ignite/stream/socket/package-info.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/stream/socket/package-info.java b/modules/core/src/test/java/org/apache/ignite/stream/socket/package-info.java
new file mode 100644
index 0000000..2e28469
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/stream/socket/package-info.java
@@ -0,0 +1,21 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * Contains tests for socket streamer.
+ */
+package org.apache.ignite.stream.socket;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/53995dcb/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteStreamTestSuite.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteStreamTestSuite.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteStreamTestSuite.java
new file mode 100644
index 0000000..87bbfbb
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteStreamTestSuite.java
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.testsuites;
+
+import org.apache.ignite.stream.socket.*;
+
+import junit.framework.*;
+
+/**
+ * Stream test suite.
+ */
+public class IgniteStreamTestSuite extends TestSuite {
+ /**
+ * @return Stream tests suite.
+ * @throws Exception If failed.
+ */
+ public static TestSuite suite() throws Exception {
+ TestSuite suite = new TestSuite("Ignite Stream Test Suite");
+
+ suite.addTest(new TestSuite(IgniteSocketStreamerSelfTest.class));
+
+ return suite;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/53995dcb/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteUtilSelfTestSuite.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteUtilSelfTestSuite.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteUtilSelfTestSuite.java
index 941b06e..32cd038 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteUtilSelfTestSuite.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteUtilSelfTestSuite.java
@@ -67,6 +67,7 @@ public class IgniteUtilSelfTestSuite extends TestSuite {
suite.addTestSuite(GridNioSelfTest.class);
suite.addTestSuite(GridNioFilterChainSelfTest.class);
suite.addTestSuite(GridNioSslSelfTest.class);
+ suite.addTestSuite(GridNioDelimitedBufferTest.class);
return suite;
}
[2/9] incubator-ignite git commit: # ignite-669 - streaming design.
Posted by sb...@apache.org.
# ignite-669 - streaming design.
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/be64e1dd
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/be64e1dd
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/be64e1dd
Branch: refs/heads/ignite-sprint-5
Commit: be64e1dd1f7aba02b664c4be6f2753cdafbbdba6
Parents: 0cbe3c6
Author: Dmitiry Setrakyan <ds...@gridgain.com>
Authored: Thu Apr 2 03:10:54 2015 -0700
Committer: agura <ag...@gridgain.com>
Committed: Fri May 15 03:44:18 2015 +0300
----------------------------------------------------------------------
.../ignite/stream/adapters/StreamAdapter.java | 29 +++++++++++++++++---
1 file changed, 25 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/be64e1dd/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 02ae795..f2e0da9 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
@@ -29,10 +29,17 @@ import java.util.*;
*/
public abstract class StreamAdapter<T, K, V> {
/** Tuple extractor. */
- private final StreamTupleExtractor<T, K, V> extractor;
+ private StreamTupleExtractor<T, K, V> extractor;
/** Streamer. */
- private final IgniteDataStreamer<K, V> stmr;
+ private IgniteDataStreamer<K, V> stmr;
+
+ /**
+ * Empty constructor.
+ */
+ public StreamAdapter() {
+ // No-op.
+ }
/**
* Stream adapter.
@@ -48,18 +55,32 @@ public abstract class StreamAdapter<T, K, V> {
/**
* @return Provided data streamer.
*/
- public IgniteDataStreamer<K, V> streamer() {
+ public IgniteDataStreamer<K, V> getStreamer() {
return stmr;
}
/**
+ * @param stmr Ignite data streamer.
+ */
+ public void setStreamer(IgniteDataStreamer<K, V> stmr) {
+ this.stmr = stmr;
+ }
+
+ /**
* @return Provided tuple extractor.
*/
- public StreamTupleExtractor<T, K, V> converter() {
+ public StreamTupleExtractor<T, K, V> getConverter() {
return extractor;
}
/**
+ * @param extractor Extractor for key-value tuples from messages.
+ */
+ public void setExtractor(StreamTupleExtractor<T, K, V> extractor) {
+ this.extractor = extractor;
+ }
+
+ /**
* Converts given message to a tuple and adds it to the underlying streamer.
*
* @param msg Message to convert.
[4/9] incubator-ignite git commit: # ignite-669 - streaming design.
Posted by sb...@apache.org.
# ignite-669 - streaming design.
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/b0fbfa05
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/b0fbfa05
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/b0fbfa05
Branch: refs/heads/ignite-sprint-5
Commit: b0fbfa05cb1cb712fe217da7166d945c7c2ec597
Parents: 877eb76
Author: Dmitiry Setrakyan <ds...@gridgain.com>
Authored: Thu Apr 2 03:13:04 2015 -0700
Committer: agura <ag...@gridgain.com>
Committed: Fri May 15 03:44:24 2015 +0300
----------------------------------------------------------------------
.../main/java/org/apache/ignite/stream/adapters/StreamAdapter.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b0fbfa05/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 9d4772f..c729362 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
@@ -37,7 +37,7 @@ public abstract class StreamAdapter<T, K, V> {
/**
* Empty constructor.
*/
- public StreamAdapter() {
+ protected StreamAdapter() {
// No-op.
}
[7/9] incubator-ignite git commit: ignite-430 Words count Socket
streamer examples
Posted by sb...@apache.org.
ignite-430 Words count Socket streamer 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/7ee85179
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/7ee85179
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/7ee85179
Branch: refs/heads/ignite-sprint-5
Commit: 7ee85179103df637ba594fab68755dee71b69997
Parents: d87efce
Author: agura <ag...@gridgain.com>
Authored: Wed May 13 20:56:22 2015 +0300
Committer: agura <ag...@gridgain.com>
Committed: Fri May 15 03:44:34 2015 +0300
----------------------------------------------------------------------
.../ignite/examples/streaming/package-info.java | 1 -
.../streaming/socket/SocketStreamerExample.java | 128 --------
.../socket/WordsSocketStreamerClient.java | 86 +++++
.../socket/WordsSocketStreamerServer.java | 93 ++++++
.../socket/ZStringsSocketStreamerExample.java | 141 ---------
.../socket/ZWordsSocketStreamerClient.java | 81 +++++
.../socket/ZWordsSocketStreamerServer.java | 111 +++++++
.../examples/streaming/socket/package-info.java | 3 +-
.../streaming/wordcount/CacheConfig.java | 2 +-
.../streaming/wordcount/QueryWords.java | 2 +-
.../streaming/wordcount/StreamWords.java | 2 +-
.../streaming/wordcount/package-info.java | 1 -
.../org/apache/ignite/stream/StreamAdapter.java | 111 +++++++
.../ignite/stream/StreamTupleExtractor.java | 33 ++
.../ignite/stream/adapters/StreamAdapter.java | 111 -------
.../stream/adapters/StreamTupleExtractor.java | 33 --
.../ignite/stream/adapters/package-info.java | 21 --
.../stream/socket/IgniteSocketStreamer.java | 217 -------------
.../ignite/stream/socket/SocketStreamer.java | 218 +++++++++++++
.../socket/IgniteSocketStreamerSelfTest.java | 315 -------------------
.../stream/socket/SocketStreamerSelfTest.java | 315 +++++++++++++++++++
.../testsuites/IgniteStreamTestSuite.java | 2 +-
22 files changed, 1053 insertions(+), 974 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/7ee85179/examples/src/main/java/org/apache/ignite/examples/streaming/package-info.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java/org/apache/ignite/examples/streaming/package-info.java b/examples/src/main/java/org/apache/ignite/examples/streaming/package-info.java
index 43dea13..43fbab3 100644
--- a/examples/src/main/java/org/apache/ignite/examples/streaming/package-info.java
+++ b/examples/src/main/java/org/apache/ignite/examples/streaming/package-info.java
@@ -16,7 +16,6 @@
*/
/**
- * <!-- Package description. -->
* Demonstrates usage of data streamer.
*/
package org.apache.ignite.examples.streaming;
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/7ee85179/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
deleted file mode 100644
index 487572a..0000000
--- a/examples/src/main/java/org/apache/ignite/examples/streaming/socket/SocketStreamerExample.java
+++ /dev/null
@@ -1,128 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.examples.streaming.socket;
-
-import org.apache.ignite.*;
-import org.apache.ignite.cache.affinity.*;
-import org.apache.ignite.examples.*;
-import org.apache.ignite.examples.streaming.wordcount.*;
-import org.apache.ignite.lang.*;
-import org.apache.ignite.stream.adapters.*;
-import org.apache.ignite.stream.socket.*;
-
-import java.io.*;
-import java.net.*;
-import java.util.*;
-
-/**
- * Stream words into Ignite cache through socket using {@link IgniteSocketStreamer} and message size based protocol.
- * <p>
- * To start the example, you should:
- * <ul>
- * <li>Start a few nodes using {@link ExampleNodeStartup} or by starting remote nodes as specified below.</li>
- * <li>Start streaming using {@link SocketStreamerExample}.</li>
- * <li>Start querying popular numbers using {@link QueryWords}.</li>
- * </ul>
- * <p>
- * You should start remote nodes by running {@link ExampleNodeStartup} in another JVM.
- */
-public class SocketStreamerExample {
- /** Port. */
- private static final int PORT = 5555;
-
- /**
- * @param args Args.
- */
- public static void main(String[] args) throws InterruptedException, IOException {
- // Mark this cluster member as client.
- Ignition.setClientMode(true);
-
- try (Ignite ignite = Ignition.start("examples/config/example-ignite.xml")) {
- if (!ExamplesUtils.hasServerNodes(ignite))
- return;
-
- // The cache is configured with sliding window holding 1 second of the streaming data.
- IgniteCache<AffinityUuid, String> stmCache = ignite.getOrCreateCache(CacheConfig.wordCache());
-
- try (IgniteDataStreamer<AffinityUuid, String> stmr = ignite.dataStreamer(stmCache.getName())) {
- InetAddress addr = InetAddress.getLocalHost();
-
- // Configure socket streamer
- IgniteSocketStreamer<String, AffinityUuid, String> sockStmr = new IgniteSocketStreamer<>();
-
- sockStmr.setAddr(addr);
-
- sockStmr.setPort(PORT);
-
- sockStmr.setIgnite(ignite);
-
- sockStmr.setStreamer(stmr);
-
- sockStmr.setTupleExtractor(new StreamTupleExtractor<String, AffinityUuid, String>() {
- @Override public Map.Entry<AffinityUuid, String> extract(String word) {
- // By using AffinityUuid we ensure that identical
- // words are processed on the same cluster node.
- return new IgniteBiTuple<>(new AffinityUuid(word), word);
- }
- });
-
- sockStmr.start();
-
- sendData(addr, PORT);
- }
- }
- }
-
- /**
- * @param addr Address.
- * @param port Port.
- */
- private static void sendData(InetAddress addr, int port) throws IOException, InterruptedException {
- try (Socket sock = new Socket(addr, port);
- OutputStream oos = new BufferedOutputStream(sock.getOutputStream())) {
- while (true) {
- try (InputStream in = StreamWords.class.getResourceAsStream("../wordcount/alice-in-wonderland.txt");
- LineNumberReader rdr = new LineNumberReader(new InputStreamReader(in))) {
- for (String line = rdr.readLine(); line != null; line = rdr.readLine()) {
- for (String word : line.split(" ")) {
- if (!word.isEmpty()) {
- // Stream words into Ignite through socket.
- try (ByteArrayOutputStream bos = new ByteArrayOutputStream();
- ObjectOutputStream out = new ObjectOutputStream(bos)) {
-
- // Write message
- out.writeObject(word);
-
- byte[] arr = bos.toByteArray();
-
- // Write message length
- oos.write(arr.length >>> 24);
- oos.write(arr.length >>> 16);
- oos.write(arr.length >>> 8);
- oos.write(arr.length);
-
- oos.write(arr);
- }
- }
- }
- }
- }
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/7ee85179/examples/src/main/java/org/apache/ignite/examples/streaming/socket/WordsSocketStreamerClient.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java/org/apache/ignite/examples/streaming/socket/WordsSocketStreamerClient.java b/examples/src/main/java/org/apache/ignite/examples/streaming/socket/WordsSocketStreamerClient.java
new file mode 100644
index 0000000..c5ec079
--- /dev/null
+++ b/examples/src/main/java/org/apache/ignite/examples/streaming/socket/WordsSocketStreamerClient.java
@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.examples.streaming.socket;
+
+import org.apache.ignite.examples.*;
+import org.apache.ignite.examples.streaming.wordcount.*;
+import org.apache.ignite.stream.socket.*;
+
+import java.io.*;
+import java.net.*;
+
+/**
+ * Sends words to socket server based on {@link SocketStreamer} using message size based protocol.
+ * <p>
+ * To start the example, you should:
+ * <ul>
+ * <li>Start a few nodes using {@link ExampleNodeStartup} or by starting remote nodes as specified below.</li>
+ * <li>Start socket server using {@link WordsSocketStreamerServer}.</li>
+ * <li>Start a few socket clients using {@link WordsSocketStreamerClient}.</li>
+ * <li>Start querying popular words using {@link QueryWords}.</li>
+ * </ul>
+ * <p>
+ * You should start remote nodes by running {@link ExampleNodeStartup} in another JVM.
+ */
+public class WordsSocketStreamerClient {
+ /** Port. */
+ private static final int PORT = 5555;
+
+ /**
+ * @param args Args.
+ */
+ public static void main(String[] args) throws IOException {
+ InetAddress addr = InetAddress.getLocalHost();
+
+ try (Socket sock = new Socket(addr, PORT);
+ OutputStream oos = new BufferedOutputStream(sock.getOutputStream())) {
+
+ System.out.println("Words streaming started.");
+
+ while (true) {
+ try (InputStream in = StreamWords.class.getResourceAsStream("../wordcount/alice-in-wonderland.txt");
+ LineNumberReader rdr = new LineNumberReader(new InputStreamReader(in))) {
+ for (String line = rdr.readLine(); line != null; line = rdr.readLine()) {
+ for (String word : line.split(" ")) {
+ if (!word.isEmpty()) {
+ // Stream words into Ignite through socket.
+ try (ByteArrayOutputStream bos = new ByteArrayOutputStream();
+ ObjectOutputStream out = new ObjectOutputStream(bos)) {
+
+ // Write message
+ out.writeObject(word);
+
+ byte[] arr = bos.toByteArray();
+
+ // Write message length
+ oos.write(arr.length >>> 24);
+ oos.write(arr.length >>> 16);
+ oos.write(arr.length >>> 8);
+ oos.write(arr.length);
+
+ oos.write(arr);
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/7ee85179/examples/src/main/java/org/apache/ignite/examples/streaming/socket/WordsSocketStreamerServer.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java/org/apache/ignite/examples/streaming/socket/WordsSocketStreamerServer.java b/examples/src/main/java/org/apache/ignite/examples/streaming/socket/WordsSocketStreamerServer.java
new file mode 100644
index 0000000..5af746d
--- /dev/null
+++ b/examples/src/main/java/org/apache/ignite/examples/streaming/socket/WordsSocketStreamerServer.java
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.examples.streaming.socket;
+
+import org.apache.ignite.*;
+import org.apache.ignite.cache.affinity.*;
+import org.apache.ignite.examples.*;
+import org.apache.ignite.examples.streaming.wordcount.*;
+import org.apache.ignite.lang.*;
+import org.apache.ignite.stream.*;
+import org.apache.ignite.stream.socket.*;
+
+import java.io.*;
+import java.net.*;
+import java.util.*;
+
+/**
+ * Receives words through socket using {@link SocketStreamer} and message size based protocol
+ * and streams them into Ignite cache.
+ * <p>
+ * To start the example, you should:
+ * <ul>
+ * <li>Start a few nodes using {@link ExampleNodeStartup} or by starting remote nodes as specified below.</li>
+ * <li>Start socket server using {@link WordsSocketStreamerServer}.</li>
+ * <li>Start a few socket clients using {@link WordsSocketStreamerClient}.</li>
+ * <li>Start querying popular words using {@link QueryWords}.</li>
+ * </ul>
+ * <p>
+ * You should start remote nodes by running {@link ExampleNodeStartup} in another JVM.
+ */
+public class WordsSocketStreamerServer {
+ /** Port. */
+ private static final int PORT = 5555;
+
+ /**
+ * @param args Args.
+ */
+ public static void main(String[] args) throws InterruptedException, IOException {
+ // Mark this cluster member as client.
+ Ignition.setClientMode(true);
+
+ Ignite ignite = Ignition.start("examples/config/example-ignite.xml");
+
+ if (!ExamplesUtils.hasServerNodes(ignite)) {
+ ignite.close();
+
+ return;
+ }
+
+ // The cache is configured with sliding window holding 1 second of the streaming data.
+ IgniteCache<AffinityUuid, String> stmCache = ignite.getOrCreateCache(CacheConfig.wordCache());
+
+ IgniteDataStreamer<AffinityUuid, String> stmr = ignite.dataStreamer(stmCache.getName());
+
+ InetAddress addr = InetAddress.getLocalHost();
+
+ // Configure socket streamer
+ SocketStreamer<String, AffinityUuid, String> sockStmr = new SocketStreamer<>();
+
+ sockStmr.setAddr(addr);
+
+ sockStmr.setPort(PORT);
+
+ sockStmr.setIgnite(ignite);
+
+ sockStmr.setStreamer(stmr);
+
+ sockStmr.setTupleExtractor(new StreamTupleExtractor<String, AffinityUuid, String>() {
+ @Override public Map.Entry<AffinityUuid, String> extract(String word) {
+ // By using AffinityUuid we ensure that identical
+ // words are processed on the same cluster node.
+ return new IgniteBiTuple<>(new AffinityUuid(word), word);
+ }
+ });
+
+ sockStmr.start();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/7ee85179/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
deleted file mode 100644
index fa5aa28..0000000
--- a/examples/src/main/java/org/apache/ignite/examples/streaming/socket/ZStringsSocketStreamerExample.java
+++ /dev/null
@@ -1,141 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.examples.streaming.socket;
-
-import org.apache.ignite.*;
-import org.apache.ignite.cache.affinity.*;
-import org.apache.ignite.examples.*;
-import org.apache.ignite.examples.streaming.wordcount.*;
-import org.apache.ignite.lang.*;
-import org.apache.ignite.stream.adapters.*;
-import org.apache.ignite.stream.socket.*;
-
-import java.io.*;
-import java.net.*;
-import java.util.*;
-
-/**
- * Stream words into Ignite cache through socket using {@link IgniteSocketStreamer} and message delimiter based
- * protocol.
- * <p>
- * Example illustrates usage of TCP socket streamer in case of non-Java clients. In this example client streams
- * zero-terminated strings.
- * <p>
- * To start the example, you should:
- * <ul>
- * <li>Start a few nodes using {@link ExampleNodeStartup} or by starting remote nodes as specified below.</li>
- * <li>Start streaming using {@link ZStringsSocketStreamerExample}.</li>
- * <li>Start querying popular numbers using {@link QueryWords}.</li>
- * </ul>
- * <p>
- * You should start remote nodes by running {@link ExampleNodeStartup} in another JVM.
- */
-public class ZStringsSocketStreamerExample {
- /** Port. */
- private static final int PORT = 5555;
-
- /** Delimiter. */
- private static final byte[] DELIM = new byte[] {0};
-
- /**
- * @param args Args.
- */
- public static void main(String[] args) throws InterruptedException, IOException {
- // Mark this cluster member as client.
- Ignition.setClientMode(true);
-
- try (Ignite ignite = Ignition.start("examples/config/example-ignite.xml")) {
- if (!ExamplesUtils.hasServerNodes(ignite))
- return;
-
- // The cache is configured with sliding window holding 1 second of the streaming data.
- IgniteCache<AffinityUuid, String> stmCache = ignite.getOrCreateCache(CacheConfig.wordCache());
-
- try (IgniteDataStreamer<AffinityUuid, String> stmr = ignite.dataStreamer(stmCache.getName())) {
- InetAddress addr = InetAddress.getLocalHost();
-
- // Configure socket streamer
- IgniteSocketStreamer<String, AffinityUuid, String> sockStmr = new IgniteSocketStreamer<>();
-
- sockStmr.setAddr(addr);
-
- sockStmr.setPort(PORT);
-
- sockStmr.setDelimiter(DELIM);
-
- sockStmr.setIgnite(ignite);
-
- sockStmr.setStreamer(stmr);
-
- // Converter from zero-terminated string to Java strings.
- sockStmr.setConverter(new SocketMessageConverter<String>() {
- @Override public String convert(byte[] msg) {
- try {
- return new String(msg, "ASCII");
- }
- catch (UnsupportedEncodingException e) {
- throw new IgniteException(e);
- }
- }
- });
-
- sockStmr.setTupleExtractor(new StreamTupleExtractor<String, AffinityUuid, String>() {
- @Override public Map.Entry<AffinityUuid, String> extract(String word) {
- // By using AffinityUuid we ensure that identical
- // words are processed on the same cluster node.
- return new IgniteBiTuple<>(new AffinityUuid(word), word);
- }
- });
-
- sockStmr.start();
-
- sendData(addr, PORT);
- }
- }
- }
-
- /**
- * @param addr Address.
- * @param port Port.
- */
- private static void sendData(InetAddress addr, int port) throws IOException, InterruptedException {
- try (Socket sock = new Socket(addr, port);
- OutputStream oos = new BufferedOutputStream(sock.getOutputStream())) {
-
- while (true) {
- try (InputStream in = StreamWords.class.getResourceAsStream("../wordcount/alice-in-wonderland.txt");
- LineNumberReader rdr = new LineNumberReader(new InputStreamReader(in))) {
- for (String line = rdr.readLine(); line != null; line = rdr.readLine()) {
- for (String word : line.split(" ")) {
- if (!word.isEmpty()) {
- // Stream words into Ignite through socket.
- byte[] arr = word.getBytes("ASCII");
-
- // Write message
- oos.write(arr);
-
- // Write message delimiter
- oos.write(DELIM);
- }
- }
- }
- }
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/7ee85179/examples/src/main/java/org/apache/ignite/examples/streaming/socket/ZWordsSocketStreamerClient.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java/org/apache/ignite/examples/streaming/socket/ZWordsSocketStreamerClient.java b/examples/src/main/java/org/apache/ignite/examples/streaming/socket/ZWordsSocketStreamerClient.java
new file mode 100644
index 0000000..c17ccdc
--- /dev/null
+++ b/examples/src/main/java/org/apache/ignite/examples/streaming/socket/ZWordsSocketStreamerClient.java
@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.examples.streaming.socket;
+
+import org.apache.ignite.examples.*;
+import org.apache.ignite.examples.streaming.wordcount.*;
+import org.apache.ignite.stream.socket.*;
+
+import java.io.*;
+import java.net.*;
+
+/**
+ * Sends words to socket server based on {@link SocketStreamer} using message delimiter based protocol.
+ * Example illustrates usage of TCP socket streamer in case of non-Java clients.
+ * In this example words are zero-terminated strings.
+ * <p>
+ * To start the example, you should:
+ * <ul>
+ * <li>Start a few nodes using {@link ExampleNodeStartup} or by starting remote nodes as specified below.</li>
+ * <li>Start socket server using {@link ZWordsSocketStreamerServer}.</li>
+ * <li>Start a few socket clients using {@link ZWordsSocketStreamerClient}.</li>
+ * <li>Start querying popular words using {@link QueryWords}.</li>
+ * </ul>
+ * <p>
+ * You should start remote nodes by running {@link ExampleNodeStartup} in another JVM.
+ */
+public class ZWordsSocketStreamerClient {
+ /** Port. */
+ private static final int PORT = 5555;
+
+ /** Delimiter. */
+ private static final byte[] DELIM = new byte[] {0};
+
+ /**
+ * @param args Args.
+ */
+ public static void main(String[] args) throws IOException {
+ InetAddress addr = InetAddress.getLocalHost();
+
+ try (Socket sock = new Socket(addr, PORT);
+ OutputStream oos = new BufferedOutputStream(sock.getOutputStream())) {
+
+ System.out.println("Words streaming started.");
+
+ while (true) {
+ try (InputStream in = StreamWords.class.getResourceAsStream("../wordcount/alice-in-wonderland.txt");
+ LineNumberReader rdr = new LineNumberReader(new InputStreamReader(in))) {
+ for (String line = rdr.readLine(); line != null; line = rdr.readLine()) {
+ for (String word : line.split(" ")) {
+ if (!word.isEmpty()) {
+ // Stream words into Ignite through socket.
+ byte[] arr = word.getBytes("ASCII");
+
+ // Write message
+ oos.write(arr);
+
+ // Write message delimiter
+ oos.write(DELIM);
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/7ee85179/examples/src/main/java/org/apache/ignite/examples/streaming/socket/ZWordsSocketStreamerServer.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java/org/apache/ignite/examples/streaming/socket/ZWordsSocketStreamerServer.java b/examples/src/main/java/org/apache/ignite/examples/streaming/socket/ZWordsSocketStreamerServer.java
new file mode 100644
index 0000000..a0ef9da
--- /dev/null
+++ b/examples/src/main/java/org/apache/ignite/examples/streaming/socket/ZWordsSocketStreamerServer.java
@@ -0,0 +1,111 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.examples.streaming.socket;
+
+import org.apache.ignite.*;
+import org.apache.ignite.cache.affinity.*;
+import org.apache.ignite.examples.*;
+import org.apache.ignite.examples.streaming.wordcount.*;
+import org.apache.ignite.lang.*;
+import org.apache.ignite.stream.*;
+import org.apache.ignite.stream.socket.*;
+
+import java.io.*;
+import java.net.*;
+import java.util.*;
+
+/**
+ * Receives words through socket using {@link SocketStreamer} and message delimiter based protocol
+ * and streams them into Ignite cache. Example illustrates usage of TCP socket streamer in case of non-Java clients.
+ * In this example words are zero-terminated strings.
+ * <p>
+ * To start the example, you should:
+ * <ul>
+ * <li>Start a few nodes using {@link ExampleNodeStartup} or by starting remote nodes as specified below.</li>
+ * <li>Start socket server using {@link ZWordsSocketStreamerServer}.</li>
+ * <li>Start a few socket clients using {@link ZWordsSocketStreamerClient}.</li>
+ * <li>Start querying popular words using {@link QueryWords}.</li>
+ * </ul>
+ * <p>
+ * You should start remote nodes by running {@link ExampleNodeStartup} in another JVM.
+ */
+public class ZWordsSocketStreamerServer {
+ /** Port. */
+ private static final int PORT = 5555;
+
+ /** Delimiter. */
+ private static final byte[] DELIM = new byte[] {0};
+
+ /**
+ * @param args Args.
+ */
+ public static void main(String[] args) throws InterruptedException, IOException {
+ // Mark this cluster member as client.
+ Ignition.setClientMode(true);
+
+ Ignite ignite = Ignition.start("examples/config/example-ignite.xml");
+
+ if (!ExamplesUtils.hasServerNodes(ignite)) {
+ ignite.close();
+
+ return;
+ }
+
+ // The cache is configured with sliding window holding 1 second of the streaming data.
+ IgniteCache<AffinityUuid, String> stmCache = ignite.getOrCreateCache(CacheConfig.wordCache());
+
+ IgniteDataStreamer<AffinityUuid, String> stmr = ignite.dataStreamer(stmCache.getName());
+
+ InetAddress addr = InetAddress.getLocalHost();
+
+ // Configure socket streamer
+ SocketStreamer<String, AffinityUuid, String> sockStmr = new SocketStreamer<>();
+
+ sockStmr.setAddr(addr);
+
+ sockStmr.setPort(PORT);
+
+ sockStmr.setDelimiter(DELIM);
+
+ sockStmr.setIgnite(ignite);
+
+ sockStmr.setStreamer(stmr);
+
+ // Converter from zero-terminated string to Java strings.
+ sockStmr.setConverter(new SocketMessageConverter<String>() {
+ @Override public String convert(byte[] msg) {
+ try {
+ return new String(msg, "ASCII");
+ }
+ catch (UnsupportedEncodingException e) {
+ throw new IgniteException(e);
+ }
+ }
+ });
+
+ sockStmr.setTupleExtractor(new StreamTupleExtractor<String, AffinityUuid, String>() {
+ @Override public Map.Entry<AffinityUuid, String> extract(String word) {
+ // By using AffinityUuid we ensure that identical
+ // words are processed on the same cluster node.
+ return new IgniteBiTuple<>(new AffinityUuid(word), word);
+ }
+ });
+
+ sockStmr.start();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/7ee85179/examples/src/main/java/org/apache/ignite/examples/streaming/socket/package-info.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java/org/apache/ignite/examples/streaming/socket/package-info.java b/examples/src/main/java/org/apache/ignite/examples/streaming/socket/package-info.java
index d0a480a..c516ab4 100644
--- a/examples/src/main/java/org/apache/ignite/examples/streaming/socket/package-info.java
+++ b/examples/src/main/java/org/apache/ignite/examples/streaming/socket/package-info.java
@@ -16,7 +16,6 @@
*/
/**
- * <!-- Package description. -->
- * Contains {@link org.apache.ignite.stream.socket.IgniteSocketStreamer} usage examples.
+ * Contains {@link org.apache.ignite.stream.socket.SocketStreamer} usage examples.
*/
package org.apache.ignite.examples.streaming.socket;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/7ee85179/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/CacheConfig.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/CacheConfig.java b/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/CacheConfig.java
index 58704ca..d17b97d 100644
--- a/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/CacheConfig.java
+++ b/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/CacheConfig.java
@@ -26,7 +26,7 @@ import javax.cache.expiry.*;
import static java.util.concurrent.TimeUnit.*;
/**
- * Configuration for the streaming cache to store the stream of random numbers.
+ * Configuration for the streaming cache to store the stream of words.
* This cache is configured with sliding window of 1 second, which means that
* data older than 1 second will be automatically removed from the cache.
*/
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/7ee85179/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/QueryWords.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/QueryWords.java b/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/QueryWords.java
index 3bd9d3d..149aa79 100644
--- a/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/QueryWords.java
+++ b/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/QueryWords.java
@@ -30,7 +30,7 @@ import java.util.*;
* <ul>
* <li>Start a few nodes using {@link ExampleNodeStartup} or by starting remote nodes as specified below.</li>
* <li>Start streaming using {@link StreamWords}.</li>
- * <li>Start querying popular numbers using {@link QueryWords}.</li>
+ * <li>Start querying popular words using {@link QueryWords}.</li>
* </ul>
* <p>
* You should start remote nodes by running {@link ExampleNodeStartup} in another JVM.
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/7ee85179/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/StreamWords.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/StreamWords.java b/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/StreamWords.java
index c59fa51..cc3c0cb 100644
--- a/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/StreamWords.java
+++ b/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/StreamWords.java
@@ -29,7 +29,7 @@ import java.io.*;
* <ul>
* <li>Start a few nodes using {@link ExampleNodeStartup} or by starting remote nodes as specified below.</li>
* <li>Start streaming using {@link StreamWords}.</li>
- * <li>Start querying popular numbers using {@link QueryWords}.</li>
+ * <li>Start querying popular words using {@link QueryWords}.</li>
* </ul>
* <p>
* You should start remote nodes by running {@link ExampleNodeStartup} in another JVM.
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/7ee85179/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/package-info.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/package-info.java b/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/package-info.java
index 010f86a..5d48ae3 100644
--- a/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/package-info.java
+++ b/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/package-info.java
@@ -16,7 +16,6 @@
*/
/**
- * <!-- Package description. -->
* Streaming word count example.
*/
package org.apache.ignite.examples.streaming.wordcount;
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/7ee85179/modules/core/src/main/java/org/apache/ignite/stream/StreamAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/stream/StreamAdapter.java b/modules/core/src/main/java/org/apache/ignite/stream/StreamAdapter.java
new file mode 100644
index 0000000..0c4e2d1
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/stream/StreamAdapter.java
@@ -0,0 +1,111 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.stream;
+
+import org.apache.ignite.*;
+
+import java.util.*;
+
+/**
+ * Convenience adapter for streamers. Adapters are optional components for
+ * streaming from different data sources. The purpose of adapters is to
+ * convert different message formats into Ignite stream key-value tuples
+ * and feed the tuples into the provided {@link org.apache.ignite.IgniteDataStreamer}.
+ */
+public abstract class StreamAdapter<T, K, V> {
+ /** Tuple extractor. */
+ private StreamTupleExtractor<T, K, V> extractor;
+
+ /** Streamer. */
+ private IgniteDataStreamer<K, V> stmr;
+
+ /** Ignite. */
+ private Ignite ignite;
+
+ /**
+ * Empty constructor.
+ */
+ protected StreamAdapter() {
+ // No-op.
+ }
+
+ /**
+ * Stream adapter.
+ *
+ * @param stmr Streamer.
+ * @param extractor Tuple extractor.
+ */
+ protected StreamAdapter(IgniteDataStreamer<K, V> stmr, StreamTupleExtractor<T, K, V> extractor) {
+ this.stmr = stmr;
+ this.extractor = extractor;
+ }
+
+ /**
+ * @return Provided data streamer.
+ */
+ public IgniteDataStreamer<K, V> getStreamer() {
+ return stmr;
+ }
+
+ /**
+ * @param stmr Ignite data streamer.
+ */
+ public void setStreamer(IgniteDataStreamer<K, V> stmr) {
+ this.stmr = stmr;
+ }
+
+ /**
+ * @return Provided tuple extractor.
+ */
+ public StreamTupleExtractor<T, K, V> getTupleExtractor() {
+ return extractor;
+ }
+
+ /**
+ * @param extractor Extractor for key-value tuples from messages.
+ */
+ public void setTupleExtractor(StreamTupleExtractor<T, K, V> extractor) {
+ this.extractor = extractor;
+ }
+
+ /**
+ * @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.
+ */
+ protected void addMessage(T msg) {
+ Map.Entry<K, V> e = extractor.extract(msg);
+
+ if (e != null)
+ stmr.addData(e);
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/7ee85179/modules/core/src/main/java/org/apache/ignite/stream/StreamTupleExtractor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/stream/StreamTupleExtractor.java b/modules/core/src/main/java/org/apache/ignite/stream/StreamTupleExtractor.java
new file mode 100644
index 0000000..d2a4ede
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/stream/StreamTupleExtractor.java
@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.stream;
+
+import java.util.*;
+
+/**
+ * Stream tuple extractor to convert messages to Ignite key-value tuples.
+ */
+public interface StreamTupleExtractor<T, K, V> {
+ /**
+ * Extracts a key-value tuple from a message.
+ *
+ * @param msg Message.
+ * @return Key-value tuple.
+ */
+ public Map.Entry<K, V> extract(T msg);
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/7ee85179/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
deleted file mode 100644
index b99521a..0000000
--- a/modules/core/src/main/java/org/apache/ignite/stream/adapters/StreamAdapter.java
+++ /dev/null
@@ -1,111 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.stream.adapters;
-
-import org.apache.ignite.*;
-
-import java.util.*;
-
-/**
- * Convenience adapter for streamers. Adapters are optional components for
- * streaming from different data sources. The purpose of adapters is to
- * convert different message formats into Ignite stream key-value tuples
- * and feed the tuples into the provided {@link org.apache.ignite.IgniteDataStreamer}.
- */
-public abstract class StreamAdapter<T, K, V> {
- /** Tuple extractor. */
- private StreamTupleExtractor<T, K, V> extractor;
-
- /** Streamer. */
- private IgniteDataStreamer<K, V> stmr;
-
- /** Ignite. */
- private Ignite ignite;
-
- /**
- * Empty constructor.
- */
- protected StreamAdapter() {
- // No-op.
- }
-
- /**
- * Stream adapter.
- *
- * @param stmr Streamer.
- * @param extractor Tuple extractor.
- */
- protected StreamAdapter(IgniteDataStreamer<K, V> stmr, StreamTupleExtractor<T, K, V> extractor) {
- this.stmr = stmr;
- this.extractor = extractor;
- }
-
- /**
- * @return Provided data streamer.
- */
- public IgniteDataStreamer<K, V> getStreamer() {
- return stmr;
- }
-
- /**
- * @param stmr Ignite data streamer.
- */
- public void setStreamer(IgniteDataStreamer<K, V> stmr) {
- this.stmr = stmr;
- }
-
- /**
- * @return Provided tuple extractor.
- */
- public StreamTupleExtractor<T, K, V> getTupleExtractor() {
- return extractor;
- }
-
- /**
- * @param extractor Extractor for key-value tuples from messages.
- */
- public void setTupleExtractor(StreamTupleExtractor<T, K, V> extractor) {
- this.extractor = extractor;
- }
-
- /**
- * @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.
- */
- protected void addMessage(T msg) {
- Map.Entry<K, V> e = extractor.extract(msg);
-
- if (e != null)
- stmr.addData(e);
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/7ee85179/modules/core/src/main/java/org/apache/ignite/stream/adapters/StreamTupleExtractor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/stream/adapters/StreamTupleExtractor.java b/modules/core/src/main/java/org/apache/ignite/stream/adapters/StreamTupleExtractor.java
deleted file mode 100644
index 9b0c395..0000000
--- a/modules/core/src/main/java/org/apache/ignite/stream/adapters/StreamTupleExtractor.java
+++ /dev/null
@@ -1,33 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.stream.adapters;
-
-import java.util.*;
-
-/**
- * Stream tuple extractor to convert messages to Ignite key-value tuples.
- */
-public interface StreamTupleExtractor<T, K, V> {
- /**
- * Extracts a key-value tuple from a message.
- *
- * @param msg Message.
- * @return Key-value tuple.
- */
- public Map.Entry<K, V> extract(T msg);
-}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/7ee85179/modules/core/src/main/java/org/apache/ignite/stream/adapters/package-info.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/stream/adapters/package-info.java b/modules/core/src/main/java/org/apache/ignite/stream/adapters/package-info.java
deleted file mode 100644
index a69ffc0..0000000
--- a/modules/core/src/main/java/org/apache/ignite/stream/adapters/package-info.java
+++ /dev/null
@@ -1,21 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-/**
- * Contains Ignite stream adapters.
- */
-package org.apache.ignite.stream.adapters;
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/7ee85179/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
deleted file mode 100644
index 66369ea..0000000
--- a/modules/core/src/main/java/org/apache/ignite/stream/socket/IgniteSocketStreamer.java
+++ /dev/null
@@ -1,217 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.stream.socket;
-
-import org.apache.ignite.*;
-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.*;
-
-/**
- * Server that receives data from TCP socket, converts it to key-value pairs using {@link StreamTupleExtractor} and
- * streams into {@link IgniteDataStreamer} instance.
- * <p>
- * By default server uses size-based message processing. That is every message sent over the socket is prepended with
- * 4-byte integer header containing message size. If message delimiter is defined (see {@link #setDelimiter}) then
- * delimiter-based message processing will be used. That is every message sent over the socket is appended with
- * provided delimiter.
- * <p>
- * Received messages through socket converts to Java object using standard serialization. Conversion functionality
- * can be customized via user defined {@link SocketMessageConverter} (e.g. in order to convert messages from
- * non Java clients).
- */
-public class IgniteSocketStreamer<T, K, V> extends StreamAdapter<T, K, V> {
- /** Default threads. */
- private static final int DFLT_THREADS = Runtime.getRuntime().availableProcessors();
-
- /** Logger. */
- private IgniteLogger log;
-
- /** Address. */
- private InetAddress addr;
-
- /** Server port. */
- private int port;
-
- /** Threads number. */
- private int threads = DFLT_THREADS;
-
- /** Direct mode. */
- private boolean directMode;
-
- /** Delimiter. */
- private byte[] delim;
-
- /** Converter. */
- private SocketMessageConverter<T> converter;
-
- /** Server. */
- private GridNioServer<byte[]> srv;
-
- /**
- * Sets server address.
- *
- * @param addr Address.
- */
- public void setAddr(InetAddress addr) {
- this.addr = addr;
- }
-
- /**
- * Sets port number.
- *
- * @param port Port.
- */
- public void setPort(int port) {
- this.port = port;
- }
-
- /**
- * Sets threadds amount.
- *
- * @param threads Threads.
- */
- public void setThreads(int threads) {
- this.threads = threads;
- }
-
- /**
- * Sets direct mode flag.
- *
- * @param directMode Direct mode.
- */
- public void setDirectMode(boolean directMode) {
- this.directMode = directMode;
- }
-
- /**
- * Sets message delimiter.
- *
- * @param delim Delimiter.
- */
- public void setDelimiter(byte[] delim) {
- this.delim = delim;
- }
-
- /**
- * Sets message converter.
- *
- * @param converter Converter.
- */
- public void setConverter(SocketMessageConverter<T> converter) {
- this.converter = converter;
- }
-
- /**
- * Starts streamer.
- *
- * @throws IgniteException If failed.
- */
- public void start() {
- A.notNull(getTupleExtractor(), "tupleExtractor");
- A.notNull(getStreamer(), "streamer");
- A.notNull(getIgnite(), "ignite");
- A.ensure(threads > 0, "threads > 0");
-
- log = getIgnite().log();
-
- GridNioServerListener<byte[]> lsnr = new GridNioServerListenerAdapter<byte[]>() {
- @Override public void onConnected(GridNioSession ses) {
- assert ses.accepted();
-
- if (log.isDebugEnabled())
- log.debug("Accepted connection: " + ses.remoteAddress());
- }
-
- @Override public void onDisconnected(GridNioSession ses, @Nullable Exception e) {
- if (e != null)
- log.error("Connection failed with exception", e);
- }
-
- @Override public void onMessage(GridNioSession ses, byte[] msg) {
- addMessage(converter.convert(msg));
- }
- };
-
- ByteOrder byteOrder = ByteOrder.BIG_ENDIAN;
-
- GridNioParser parser = F.isEmpty(delim) ? new GridBufferedParser(directMode, byteOrder) :
- new GridDelimitedParser(delim, directMode);
-
- if (converter == null)
- converter = new DefaultConverter<>();
-
- GridNioFilter codec = new GridNioCodecFilter(parser, log, directMode);
-
- GridNioFilter[] filters = new GridNioFilter[] {codec};
-
- try {
- srv = new GridNioServer.Builder<byte[]>()
- .address(addr == null ? InetAddress.getLocalHost() : addr)
- .port(port)
- .listener(lsnr)
- .logger(log)
- .selectorCount(threads)
- .byteOrder(byteOrder)
- .filters(filters)
- .build();
- }
- catch (IgniteCheckedException | UnknownHostException e) {
- throw new IgniteException(e);
- }
-
- srv.start();
-
- if (log.isDebugEnabled())
- log.debug("Socket streaming server started on " + addr + ':' + port);
- }
-
- /**
- * Stops streamer.
- */
- public void stop() {
- srv.stop();
-
- if (log.isDebugEnabled())
- log.debug("Socket streaming server stopped");
- }
-
- /**
- * Converts message to Java object using Jdk marshaller.
- */
- private static class DefaultConverter<T> implements SocketMessageConverter<T> {
- /** Marshaller. */
- private static final JdkMarshaller MARSH = new JdkMarshaller();
-
- /** {@inheritDoc} */
- @Override public T convert(byte[] msg) {
- try {
- return MARSH.unmarshal(msg, null);
- }
- catch (IgniteCheckedException e) {
- throw new IgniteException(e);
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/7ee85179/modules/core/src/main/java/org/apache/ignite/stream/socket/SocketStreamer.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/stream/socket/SocketStreamer.java b/modules/core/src/main/java/org/apache/ignite/stream/socket/SocketStreamer.java
new file mode 100644
index 0000000..07ce77e
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/stream/socket/SocketStreamer.java
@@ -0,0 +1,218 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.stream.socket;
+
+import org.apache.ignite.*;
+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.*;
+
+import org.jetbrains.annotations.*;
+
+import java.net.*;
+import java.nio.*;
+
+/**
+ * Server that receives data from TCP socket, converts it to key-value pairs using {@link StreamTupleExtractor} and
+ * streams into {@link IgniteDataStreamer} instance.
+ * <p>
+ * By default server uses size-based message processing. That is every message sent over the socket is prepended with
+ * 4-byte integer header containing message size. If message delimiter is defined (see {@link #setDelimiter}) then
+ * delimiter-based message processing will be used. That is every message sent over the socket is appended with
+ * provided delimiter.
+ * <p>
+ * Received messages through socket converts to Java object using standard serialization. Conversion functionality
+ * can be customized via user defined {@link SocketMessageConverter} (e.g. in order to convert messages from
+ * non Java clients).
+ */
+public class SocketStreamer<T, K, V> extends StreamAdapter<T, K, V> {
+ /** Default threads. */
+ private static final int DFLT_THREADS = Runtime.getRuntime().availableProcessors();
+
+ /** Logger. */
+ private IgniteLogger log;
+
+ /** Address. */
+ private InetAddress addr;
+
+ /** Server port. */
+ private int port;
+
+ /** Threads number. */
+ private int threads = DFLT_THREADS;
+
+ /** Direct mode. */
+ private boolean directMode;
+
+ /** Delimiter. */
+ private byte[] delim;
+
+ /** Converter. */
+ private SocketMessageConverter<T> converter;
+
+ /** Server. */
+ private GridNioServer<byte[]> srv;
+
+ /**
+ * Sets server address.
+ *
+ * @param addr Address.
+ */
+ public void setAddr(InetAddress addr) {
+ this.addr = addr;
+ }
+
+ /**
+ * Sets port number.
+ *
+ * @param port Port.
+ */
+ public void setPort(int port) {
+ this.port = port;
+ }
+
+ /**
+ * Sets threadds amount.
+ *
+ * @param threads Threads.
+ */
+ public void setThreads(int threads) {
+ this.threads = threads;
+ }
+
+ /**
+ * Sets direct mode flag.
+ *
+ * @param directMode Direct mode.
+ */
+ public void setDirectMode(boolean directMode) {
+ this.directMode = directMode;
+ }
+
+ /**
+ * Sets message delimiter.
+ *
+ * @param delim Delimiter.
+ */
+ public void setDelimiter(byte[] delim) {
+ this.delim = delim;
+ }
+
+ /**
+ * Sets message converter.
+ *
+ * @param converter Converter.
+ */
+ public void setConverter(SocketMessageConverter<T> converter) {
+ this.converter = converter;
+ }
+
+ /**
+ * Starts streamer.
+ *
+ * @throws IgniteException If failed.
+ */
+ public void start() {
+ A.notNull(getTupleExtractor(), "tupleExtractor");
+ A.notNull(getStreamer(), "streamer");
+ A.notNull(getIgnite(), "ignite");
+ A.ensure(threads > 0, "threads > 0");
+
+ log = getIgnite().log();
+
+ GridNioServerListener<byte[]> lsnr = new GridNioServerListenerAdapter<byte[]>() {
+ @Override public void onConnected(GridNioSession ses) {
+ assert ses.accepted();
+
+ if (log.isDebugEnabled())
+ log.debug("Accepted connection: " + ses.remoteAddress());
+ }
+
+ @Override public void onDisconnected(GridNioSession ses, @Nullable Exception e) {
+ if (e != null)
+ log.error("Connection failed with exception", e);
+ }
+
+ @Override public void onMessage(GridNioSession ses, byte[] msg) {
+ addMessage(converter.convert(msg));
+ }
+ };
+
+ ByteOrder byteOrder = ByteOrder.BIG_ENDIAN;
+
+ GridNioParser parser = F.isEmpty(delim) ? new GridBufferedParser(directMode, byteOrder) :
+ new GridDelimitedParser(delim, directMode);
+
+ if (converter == null)
+ converter = new DefaultConverter<>();
+
+ GridNioFilter codec = new GridNioCodecFilter(parser, log, directMode);
+
+ GridNioFilter[] filters = new GridNioFilter[] {codec};
+
+ try {
+ srv = new GridNioServer.Builder<byte[]>()
+ .address(addr == null ? InetAddress.getLocalHost() : addr)
+ .port(port)
+ .listener(lsnr)
+ .logger(log)
+ .selectorCount(threads)
+ .byteOrder(byteOrder)
+ .filters(filters)
+ .build();
+ }
+ catch (IgniteCheckedException | UnknownHostException e) {
+ throw new IgniteException(e);
+ }
+
+ srv.start();
+
+ if (log.isDebugEnabled())
+ log.debug("Socket streaming server started on " + addr + ':' + port);
+ }
+
+ /**
+ * Stops streamer.
+ */
+ public void stop() {
+ srv.stop();
+
+ if (log.isDebugEnabled())
+ log.debug("Socket streaming server stopped");
+ }
+
+ /**
+ * Converts message to Java object using Jdk marshaller.
+ */
+ private static class DefaultConverter<T> implements SocketMessageConverter<T> {
+ /** Marshaller. */
+ private static final JdkMarshaller MARSH = new JdkMarshaller();
+
+ /** {@inheritDoc} */
+ @Override public T convert(byte[] msg) {
+ try {
+ return MARSH.unmarshal(msg, null);
+ }
+ catch (IgniteCheckedException e) {
+ throw new IgniteException(e);
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/7ee85179/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
deleted file mode 100644
index 19852ce..0000000
--- a/modules/core/src/test/java/org/apache/ignite/stream/socket/IgniteSocketStreamerSelfTest.java
+++ /dev/null
@@ -1,315 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.stream.socket;
-
-import org.apache.ignite.*;
-import org.apache.ignite.cache.*;
-import org.apache.ignite.configuration.*;
-import org.apache.ignite.events.*;
-import org.apache.ignite.lang.*;
-import org.apache.ignite.marshaller.*;
-import org.apache.ignite.marshaller.jdk.*;
-import org.apache.ignite.spi.discovery.tcp.*;
-import org.apache.ignite.spi.discovery.tcp.ipfinder.*;
-import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.*;
-import org.apache.ignite.stream.adapters.*;
-import org.apache.ignite.testframework.junits.common.*;
-
-import org.jetbrains.annotations.*;
-
-import java.io.*;
-import java.net.*;
-import java.util.*;
-import java.util.concurrent.*;
-
-import static org.apache.ignite.events.EventType.*;
-
-/**
- * Tests {@link IgniteSocketStreamer}.
- */
-public class IgniteSocketStreamerSelfTest extends GridCommonAbstractTest {
- /** IP finder. */
- private static final TcpDiscoveryIpFinder IP_FINDER = new TcpDiscoveryVmIpFinder(true);
-
- /** Grid count. */
- private final static int GRID_CNT = 3;
-
- /** Count. */
- private static final int CNT = 500;
-
- /** Delimiter. */
- private static final byte[] DELIM = new byte[] {0, 1, 2, 3, 4, 5, 4, 3, 2, 1, 0};
-
- /** Port. */
- private static int port;
-
- /** Ignite. */
- private static Ignite ignite;
-
- /** {@inheritDoc} */
- @Override protected IgniteConfiguration getConfiguration() throws Exception {
- IgniteConfiguration cfg = super.getConfiguration();
-
- CacheConfiguration ccfg = cacheConfiguration(cfg, null);
-
- cfg.setCacheConfiguration(ccfg);
-
- TcpDiscoverySpi discoSpi = new TcpDiscoverySpi();
-
- discoSpi.setIpFinder(IP_FINDER);
-
- cfg.setDiscoverySpi(discoSpi);
-
- return cfg;
- }
-
-
- /** {@inheritDoc} */
- @Override protected void beforeTestsStarted() throws Exception {
- ignite = startGrids(GRID_CNT);
- ignite.<Integer, String>getOrCreateCache(defaultCacheConfiguration());
-
- try (ServerSocket sock = new ServerSocket(0)) {
- port = sock.getLocalPort();
- }
- }
-
- /** {@inheritDoc} */
- @Override protected void afterTestsStopped() throws Exception {
- stopAllGrids();
- }
-
- /** {@inheritDoc} */
- @Override protected void beforeTest() throws Exception {
- ignite.cache(null).clear();
- }
-
- /**
- * @throws Exception If failed.
- */
- public void testSizeBasedDefaultConverter() throws Exception {
- test(null, null, new Runnable() {
- @Override public void run() {
- try (Socket sock = new Socket(InetAddress.getLocalHost(), port);
- OutputStream os = new BufferedOutputStream(sock.getOutputStream())) {
- Marshaller marsh = new JdkMarshaller();
-
- for (int i = 0; i < CNT; i++) {
- byte[] msg = marsh.marshal(new Tuple(i));
-
- os.write(msg.length >>> 24);
- os.write(msg.length >>> 16);
- os.write(msg.length >>> 8);
- os.write(msg.length);
-
- os.write(msg);
- }
- }
- catch (IOException | IgniteCheckedException e) {
- throw new IgniteException(e);
- }
- }
- });
- }
-
- /**
- * @throws Exception If failed.
- */
- public void testSizeBasedCustomConverter() throws Exception {
- SocketMessageConverter<Tuple> converter = new SocketMessageConverter<Tuple>() {
- @Override public Tuple convert(byte[] msg) {
- int i = (msg[0] & 0xFF) << 24;
- i |= (msg[1] & 0xFF) << 16;
- i |= (msg[2] & 0xFF) << 8;
- i |= msg[3] & 0xFF;
-
- return new Tuple(i);
- }
- };
-
- test(converter, null, new Runnable() {
- @Override public void run() {
- try(Socket sock = new Socket(InetAddress.getLocalHost(), port);
- OutputStream os = new BufferedOutputStream(sock.getOutputStream())) {
-
- for (int i = 0; i < CNT; i++) {
- os.write(0);
- os.write(0);
- os.write(0);
- os.write(4);
-
- os.write(i >>> 24);
- os.write(i >>> 16);
- os.write(i >>> 8);
- os.write(i);
- }
- }
- catch (IOException e) {
- throw new IgniteException(e);
- }
- }
- });
- }
-
- /**
- * @throws Exception If failed.
- */
- public void testDelimiterBasedDefaultConverter() throws Exception {
- test(null, DELIM, new Runnable() {
- @Override public void run() {
- try(Socket sock = new Socket(InetAddress.getLocalHost(), port);
- OutputStream os = new BufferedOutputStream(sock.getOutputStream())) {
- Marshaller marsh = new JdkMarshaller();
-
- for (int i = 0; i < CNT; i++) {
- byte[] msg = marsh.marshal(new Tuple(i));
-
- os.write(msg);
- os.write(DELIM);
- }
- }
- catch (IOException | IgniteCheckedException e) {
- throw new IgniteException(e);
- }
- }
- });
-
- }
-
- /**
- * @throws Exception If failed.
- */
- public void testDelimiterBasedCustomConverter() throws Exception {
- SocketMessageConverter<Tuple> converter = new SocketMessageConverter<Tuple>() {
- @Override public Tuple convert(byte[] msg) {
- int i = (msg[0] & 0xFF) << 24;
- i |= (msg[1] & 0xFF) << 16;
- i |= (msg[2] & 0xFF) << 8;
- i |= msg[3] & 0xFF;
-
- return new Tuple(i);
- }
- };
-
- test(converter, DELIM, new Runnable() {
- @Override public void run() {
- try(Socket sock = new Socket(InetAddress.getLocalHost(), port);
- OutputStream os = new BufferedOutputStream(sock.getOutputStream())) {
-
- for (int i = 0; i < CNT; i++) {
- os.write(i >>> 24);
- os.write(i >>> 16);
- os.write(i >>> 8);
- os.write(i);
-
- os.write(DELIM);
- }
- }
- catch (IOException e) {
- throw new IgniteException(e);
- }
- }
- });
- }
-
- /**
- * @param converter Converter.
- * @param r Runnable..
- */
- private void test(@Nullable SocketMessageConverter<Tuple> converter, @Nullable byte[] delim, Runnable r) throws Exception
- {
- IgniteSocketStreamer<Tuple, Integer, String> sockStmr = null;
-
- try (IgniteDataStreamer<Integer, String> stmr = ignite.dataStreamer(null)) {
-
- stmr.allowOverwrite(true);
- stmr.autoFlushFrequency(10);
-
- sockStmr = new IgniteSocketStreamer<>();
-
- IgniteCache<Integer, String> cache = ignite.cache(null);
-
- sockStmr.setIgnite(ignite);
-
- sockStmr.setStreamer(stmr);
-
- sockStmr.setPort(port);
-
- sockStmr.setDelimiter(delim);
-
- sockStmr.setTupleExtractor(new StreamTupleExtractor<Tuple, Integer, String>() {
- @Override public Map.Entry<Integer, String> extract(Tuple msg) {
- return new IgniteBiTuple<>(msg.key, msg.val);
- }
- });
-
- if (converter != null)
- sockStmr.setConverter(converter);
-
- final CountDownLatch latch = new CountDownLatch(CNT);
-
- IgniteBiPredicate<UUID, CacheEvent> locLsnr = new IgniteBiPredicate<UUID, CacheEvent>() {
- @Override public boolean apply(UUID uuid, CacheEvent evt) {
- latch.countDown();
-
- return true;
- }
- };
-
- ignite.events(ignite.cluster().forCacheNodes(null)).remoteListen(locLsnr, null, EVT_CACHE_OBJECT_PUT);
-
- sockStmr.start();
-
- r.run();
-
- latch.await();
-
- assertEquals(CNT, cache.size(CachePeekMode.PRIMARY));
-
- for (int i = 0; i < CNT; i++)
- assertEquals(Integer.toString(i), cache.get(i));
- }
- finally {
- if (sockStmr != null)
- sockStmr.stop();
- }
-
- }
-
- /**
- * Tuple.
- */
- private static class Tuple implements Serializable {
- /** Serial version uid. */
- private static final long serialVersionUID = 0L;
-
- /** Key. */
- private final int key;
-
- /** Value. */
- private final String val;
-
- /**
- * @param key Key.
- */
- Tuple(int key) {
- this.key = key;
- this.val = Integer.toString(key);
- }
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/7ee85179/modules/core/src/test/java/org/apache/ignite/stream/socket/SocketStreamerSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/stream/socket/SocketStreamerSelfTest.java b/modules/core/src/test/java/org/apache/ignite/stream/socket/SocketStreamerSelfTest.java
new file mode 100644
index 0000000..752e43c
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/stream/socket/SocketStreamerSelfTest.java
@@ -0,0 +1,315 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.stream.socket;
+
+import org.apache.ignite.*;
+import org.apache.ignite.cache.*;
+import org.apache.ignite.configuration.*;
+import org.apache.ignite.events.*;
+import org.apache.ignite.lang.*;
+import org.apache.ignite.marshaller.*;
+import org.apache.ignite.marshaller.jdk.*;
+import org.apache.ignite.spi.discovery.tcp.*;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.*;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.*;
+import org.apache.ignite.stream.*;
+import org.apache.ignite.testframework.junits.common.*;
+
+import org.jetbrains.annotations.*;
+
+import java.io.*;
+import java.net.*;
+import java.util.*;
+import java.util.concurrent.*;
+
+import static org.apache.ignite.events.EventType.*;
+
+/**
+ * Tests {@link SocketStreamer}.
+ */
+public class SocketStreamerSelfTest extends GridCommonAbstractTest {
+ /** IP finder. */
+ private static final TcpDiscoveryIpFinder IP_FINDER = new TcpDiscoveryVmIpFinder(true);
+
+ /** Grid count. */
+ private final static int GRID_CNT = 3;
+
+ /** Count. */
+ private static final int CNT = 500;
+
+ /** Delimiter. */
+ private static final byte[] DELIM = new byte[] {0, 1, 2, 3, 4, 5, 4, 3, 2, 1, 0};
+
+ /** Port. */
+ private static int port;
+
+ /** Ignite. */
+ private static Ignite ignite;
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration() throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration();
+
+ CacheConfiguration ccfg = cacheConfiguration(cfg, null);
+
+ cfg.setCacheConfiguration(ccfg);
+
+ TcpDiscoverySpi discoSpi = new TcpDiscoverySpi();
+
+ discoSpi.setIpFinder(IP_FINDER);
+
+ cfg.setDiscoverySpi(discoSpi);
+
+ return cfg;
+ }
+
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ ignite = startGrids(GRID_CNT);
+ ignite.<Integer, String>getOrCreateCache(defaultCacheConfiguration());
+
+ try (ServerSocket sock = new ServerSocket(0)) {
+ port = sock.getLocalPort();
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTestsStopped() throws Exception {
+ stopAllGrids();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTest() throws Exception {
+ ignite.cache(null).clear();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testSizeBasedDefaultConverter() throws Exception {
+ test(null, null, new Runnable() {
+ @Override public void run() {
+ try (Socket sock = new Socket(InetAddress.getLocalHost(), port);
+ OutputStream os = new BufferedOutputStream(sock.getOutputStream())) {
+ Marshaller marsh = new JdkMarshaller();
+
+ for (int i = 0; i < CNT; i++) {
+ byte[] msg = marsh.marshal(new Tuple(i));
+
+ os.write(msg.length >>> 24);
+ os.write(msg.length >>> 16);
+ os.write(msg.length >>> 8);
+ os.write(msg.length);
+
+ os.write(msg);
+ }
+ }
+ catch (IOException | IgniteCheckedException e) {
+ throw new IgniteException(e);
+ }
+ }
+ });
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testSizeBasedCustomConverter() throws Exception {
+ SocketMessageConverter<Tuple> converter = new SocketMessageConverter<Tuple>() {
+ @Override public Tuple convert(byte[] msg) {
+ int i = (msg[0] & 0xFF) << 24;
+ i |= (msg[1] & 0xFF) << 16;
+ i |= (msg[2] & 0xFF) << 8;
+ i |= msg[3] & 0xFF;
+
+ return new Tuple(i);
+ }
+ };
+
+ test(converter, null, new Runnable() {
+ @Override public void run() {
+ try(Socket sock = new Socket(InetAddress.getLocalHost(), port);
+ OutputStream os = new BufferedOutputStream(sock.getOutputStream())) {
+
+ for (int i = 0; i < CNT; i++) {
+ os.write(0);
+ os.write(0);
+ os.write(0);
+ os.write(4);
+
+ os.write(i >>> 24);
+ os.write(i >>> 16);
+ os.write(i >>> 8);
+ os.write(i);
+ }
+ }
+ catch (IOException e) {
+ throw new IgniteException(e);
+ }
+ }
+ });
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testDelimiterBasedDefaultConverter() throws Exception {
+ test(null, DELIM, new Runnable() {
+ @Override public void run() {
+ try(Socket sock = new Socket(InetAddress.getLocalHost(), port);
+ OutputStream os = new BufferedOutputStream(sock.getOutputStream())) {
+ Marshaller marsh = new JdkMarshaller();
+
+ for (int i = 0; i < CNT; i++) {
+ byte[] msg = marsh.marshal(new Tuple(i));
+
+ os.write(msg);
+ os.write(DELIM);
+ }
+ }
+ catch (IOException | IgniteCheckedException e) {
+ throw new IgniteException(e);
+ }
+ }
+ });
+
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testDelimiterBasedCustomConverter() throws Exception {
+ SocketMessageConverter<Tuple> converter = new SocketMessageConverter<Tuple>() {
+ @Override public Tuple convert(byte[] msg) {
+ int i = (msg[0] & 0xFF) << 24;
+ i |= (msg[1] & 0xFF) << 16;
+ i |= (msg[2] & 0xFF) << 8;
+ i |= msg[3] & 0xFF;
+
+ return new Tuple(i);
+ }
+ };
+
+ test(converter, DELIM, new Runnable() {
+ @Override public void run() {
+ try(Socket sock = new Socket(InetAddress.getLocalHost(), port);
+ OutputStream os = new BufferedOutputStream(sock.getOutputStream())) {
+
+ for (int i = 0; i < CNT; i++) {
+ os.write(i >>> 24);
+ os.write(i >>> 16);
+ os.write(i >>> 8);
+ os.write(i);
+
+ os.write(DELIM);
+ }
+ }
+ catch (IOException e) {
+ throw new IgniteException(e);
+ }
+ }
+ });
+ }
+
+ /**
+ * @param converter Converter.
+ * @param r Runnable..
+ */
+ private void test(@Nullable SocketMessageConverter<Tuple> converter, @Nullable byte[] delim, Runnable r) throws Exception
+ {
+ SocketStreamer<Tuple, Integer, String> sockStmr = null;
+
+ try (IgniteDataStreamer<Integer, String> stmr = ignite.dataStreamer(null)) {
+
+ stmr.allowOverwrite(true);
+ stmr.autoFlushFrequency(10);
+
+ sockStmr = new SocketStreamer<>();
+
+ IgniteCache<Integer, String> cache = ignite.cache(null);
+
+ sockStmr.setIgnite(ignite);
+
+ sockStmr.setStreamer(stmr);
+
+ sockStmr.setPort(port);
+
+ sockStmr.setDelimiter(delim);
+
+ sockStmr.setTupleExtractor(new StreamTupleExtractor<Tuple, Integer, String>() {
+ @Override public Map.Entry<Integer, String> extract(Tuple msg) {
+ return new IgniteBiTuple<>(msg.key, msg.val);
+ }
+ });
+
+ if (converter != null)
+ sockStmr.setConverter(converter);
+
+ final CountDownLatch latch = new CountDownLatch(CNT);
+
+ IgniteBiPredicate<UUID, CacheEvent> locLsnr = new IgniteBiPredicate<UUID, CacheEvent>() {
+ @Override public boolean apply(UUID uuid, CacheEvent evt) {
+ latch.countDown();
+
+ return true;
+ }
+ };
+
+ ignite.events(ignite.cluster().forCacheNodes(null)).remoteListen(locLsnr, null, EVT_CACHE_OBJECT_PUT);
+
+ sockStmr.start();
+
+ r.run();
+
+ latch.await();
+
+ assertEquals(CNT, cache.size(CachePeekMode.PRIMARY));
+
+ for (int i = 0; i < CNT; i++)
+ assertEquals(Integer.toString(i), cache.get(i));
+ }
+ finally {
+ if (sockStmr != null)
+ sockStmr.stop();
+ }
+
+ }
+
+ /**
+ * Tuple.
+ */
+ private static class Tuple implements Serializable {
+ /** Serial version uid. */
+ private static final long serialVersionUID = 0L;
+
+ /** Key. */
+ private final int key;
+
+ /** Value. */
+ private final String val;
+
+ /**
+ * @param key Key.
+ */
+ Tuple(int key) {
+ this.key = key;
+ this.val = Integer.toString(key);
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/7ee85179/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteStreamTestSuite.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteStreamTestSuite.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteStreamTestSuite.java
index 87bbfbb..61be976 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteStreamTestSuite.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteStreamTestSuite.java
@@ -32,7 +32,7 @@ public class IgniteStreamTestSuite extends TestSuite {
public static TestSuite suite() throws Exception {
TestSuite suite = new TestSuite("Ignite Stream Test Suite");
- suite.addTest(new TestSuite(IgniteSocketStreamerSelfTest.class));
+ suite.addTest(new TestSuite(SocketStreamerSelfTest.class));
return suite;
}
[8/9] incubator-ignite git commit: ignite-430 review
Posted by sb...@apache.org.
ignite-430 review
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/fe78d42c
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/fe78d42c
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/fe78d42c
Branch: refs/heads/ignite-sprint-5
Commit: fe78d42cfe77626b7250875e0c3f608b27dc9367
Parents: 7ee8517
Author: Yakov Zhdanov <yz...@gridgain.com>
Authored: Thu May 14 16:36:06 2015 +0300
Committer: agura <ag...@gridgain.com>
Committed: Fri May 15 03:44:37 2015 +0300
----------------------------------------------------------------------
.../socket/WordsSocketStreamerClient.java | 86 --------------
.../socket/ZWordsSocketStreamerClient.java | 81 --------------
.../socket/ZWordsSocketStreamerServer.java | 111 -------------------
.../examples/streaming/socket/package-info.java | 21 ----
.../streaming/wordcount/QueryWords.java | 6 +
.../streaming/wordcount/StreamWords.java | 6 +
.../socket/WordsSocketStreamerClient.java | 82 ++++++++++++++
.../socket/WordsSocketStreamerServer.java | 111 +++++++++++++++++++
.../wordcount/socket/package-info.java | 21 ++++
9 files changed, 226 insertions(+), 299 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/fe78d42c/examples/src/main/java/org/apache/ignite/examples/streaming/socket/WordsSocketStreamerClient.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java/org/apache/ignite/examples/streaming/socket/WordsSocketStreamerClient.java b/examples/src/main/java/org/apache/ignite/examples/streaming/socket/WordsSocketStreamerClient.java
deleted file mode 100644
index c5ec079..0000000
--- a/examples/src/main/java/org/apache/ignite/examples/streaming/socket/WordsSocketStreamerClient.java
+++ /dev/null
@@ -1,86 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.examples.streaming.socket;
-
-import org.apache.ignite.examples.*;
-import org.apache.ignite.examples.streaming.wordcount.*;
-import org.apache.ignite.stream.socket.*;
-
-import java.io.*;
-import java.net.*;
-
-/**
- * Sends words to socket server based on {@link SocketStreamer} using message size based protocol.
- * <p>
- * To start the example, you should:
- * <ul>
- * <li>Start a few nodes using {@link ExampleNodeStartup} or by starting remote nodes as specified below.</li>
- * <li>Start socket server using {@link WordsSocketStreamerServer}.</li>
- * <li>Start a few socket clients using {@link WordsSocketStreamerClient}.</li>
- * <li>Start querying popular words using {@link QueryWords}.</li>
- * </ul>
- * <p>
- * You should start remote nodes by running {@link ExampleNodeStartup} in another JVM.
- */
-public class WordsSocketStreamerClient {
- /** Port. */
- private static final int PORT = 5555;
-
- /**
- * @param args Args.
- */
- public static void main(String[] args) throws IOException {
- InetAddress addr = InetAddress.getLocalHost();
-
- try (Socket sock = new Socket(addr, PORT);
- OutputStream oos = new BufferedOutputStream(sock.getOutputStream())) {
-
- System.out.println("Words streaming started.");
-
- while (true) {
- try (InputStream in = StreamWords.class.getResourceAsStream("../wordcount/alice-in-wonderland.txt");
- LineNumberReader rdr = new LineNumberReader(new InputStreamReader(in))) {
- for (String line = rdr.readLine(); line != null; line = rdr.readLine()) {
- for (String word : line.split(" ")) {
- if (!word.isEmpty()) {
- // Stream words into Ignite through socket.
- try (ByteArrayOutputStream bos = new ByteArrayOutputStream();
- ObjectOutputStream out = new ObjectOutputStream(bos)) {
-
- // Write message
- out.writeObject(word);
-
- byte[] arr = bos.toByteArray();
-
- // Write message length
- oos.write(arr.length >>> 24);
- oos.write(arr.length >>> 16);
- oos.write(arr.length >>> 8);
- oos.write(arr.length);
-
- oos.write(arr);
- }
- }
- }
- }
- }
- }
- }
-
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/fe78d42c/examples/src/main/java/org/apache/ignite/examples/streaming/socket/ZWordsSocketStreamerClient.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java/org/apache/ignite/examples/streaming/socket/ZWordsSocketStreamerClient.java b/examples/src/main/java/org/apache/ignite/examples/streaming/socket/ZWordsSocketStreamerClient.java
deleted file mode 100644
index c17ccdc..0000000
--- a/examples/src/main/java/org/apache/ignite/examples/streaming/socket/ZWordsSocketStreamerClient.java
+++ /dev/null
@@ -1,81 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.examples.streaming.socket;
-
-import org.apache.ignite.examples.*;
-import org.apache.ignite.examples.streaming.wordcount.*;
-import org.apache.ignite.stream.socket.*;
-
-import java.io.*;
-import java.net.*;
-
-/**
- * Sends words to socket server based on {@link SocketStreamer} using message delimiter based protocol.
- * Example illustrates usage of TCP socket streamer in case of non-Java clients.
- * In this example words are zero-terminated strings.
- * <p>
- * To start the example, you should:
- * <ul>
- * <li>Start a few nodes using {@link ExampleNodeStartup} or by starting remote nodes as specified below.</li>
- * <li>Start socket server using {@link ZWordsSocketStreamerServer}.</li>
- * <li>Start a few socket clients using {@link ZWordsSocketStreamerClient}.</li>
- * <li>Start querying popular words using {@link QueryWords}.</li>
- * </ul>
- * <p>
- * You should start remote nodes by running {@link ExampleNodeStartup} in another JVM.
- */
-public class ZWordsSocketStreamerClient {
- /** Port. */
- private static final int PORT = 5555;
-
- /** Delimiter. */
- private static final byte[] DELIM = new byte[] {0};
-
- /**
- * @param args Args.
- */
- public static void main(String[] args) throws IOException {
- InetAddress addr = InetAddress.getLocalHost();
-
- try (Socket sock = new Socket(addr, PORT);
- OutputStream oos = new BufferedOutputStream(sock.getOutputStream())) {
-
- System.out.println("Words streaming started.");
-
- while (true) {
- try (InputStream in = StreamWords.class.getResourceAsStream("../wordcount/alice-in-wonderland.txt");
- LineNumberReader rdr = new LineNumberReader(new InputStreamReader(in))) {
- for (String line = rdr.readLine(); line != null; line = rdr.readLine()) {
- for (String word : line.split(" ")) {
- if (!word.isEmpty()) {
- // Stream words into Ignite through socket.
- byte[] arr = word.getBytes("ASCII");
-
- // Write message
- oos.write(arr);
-
- // Write message delimiter
- oos.write(DELIM);
- }
- }
- }
- }
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/fe78d42c/examples/src/main/java/org/apache/ignite/examples/streaming/socket/ZWordsSocketStreamerServer.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java/org/apache/ignite/examples/streaming/socket/ZWordsSocketStreamerServer.java b/examples/src/main/java/org/apache/ignite/examples/streaming/socket/ZWordsSocketStreamerServer.java
deleted file mode 100644
index a0ef9da..0000000
--- a/examples/src/main/java/org/apache/ignite/examples/streaming/socket/ZWordsSocketStreamerServer.java
+++ /dev/null
@@ -1,111 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.examples.streaming.socket;
-
-import org.apache.ignite.*;
-import org.apache.ignite.cache.affinity.*;
-import org.apache.ignite.examples.*;
-import org.apache.ignite.examples.streaming.wordcount.*;
-import org.apache.ignite.lang.*;
-import org.apache.ignite.stream.*;
-import org.apache.ignite.stream.socket.*;
-
-import java.io.*;
-import java.net.*;
-import java.util.*;
-
-/**
- * Receives words through socket using {@link SocketStreamer} and message delimiter based protocol
- * and streams them into Ignite cache. Example illustrates usage of TCP socket streamer in case of non-Java clients.
- * In this example words are zero-terminated strings.
- * <p>
- * To start the example, you should:
- * <ul>
- * <li>Start a few nodes using {@link ExampleNodeStartup} or by starting remote nodes as specified below.</li>
- * <li>Start socket server using {@link ZWordsSocketStreamerServer}.</li>
- * <li>Start a few socket clients using {@link ZWordsSocketStreamerClient}.</li>
- * <li>Start querying popular words using {@link QueryWords}.</li>
- * </ul>
- * <p>
- * You should start remote nodes by running {@link ExampleNodeStartup} in another JVM.
- */
-public class ZWordsSocketStreamerServer {
- /** Port. */
- private static final int PORT = 5555;
-
- /** Delimiter. */
- private static final byte[] DELIM = new byte[] {0};
-
- /**
- * @param args Args.
- */
- public static void main(String[] args) throws InterruptedException, IOException {
- // Mark this cluster member as client.
- Ignition.setClientMode(true);
-
- Ignite ignite = Ignition.start("examples/config/example-ignite.xml");
-
- if (!ExamplesUtils.hasServerNodes(ignite)) {
- ignite.close();
-
- return;
- }
-
- // The cache is configured with sliding window holding 1 second of the streaming data.
- IgniteCache<AffinityUuid, String> stmCache = ignite.getOrCreateCache(CacheConfig.wordCache());
-
- IgniteDataStreamer<AffinityUuid, String> stmr = ignite.dataStreamer(stmCache.getName());
-
- InetAddress addr = InetAddress.getLocalHost();
-
- // Configure socket streamer
- SocketStreamer<String, AffinityUuid, String> sockStmr = new SocketStreamer<>();
-
- sockStmr.setAddr(addr);
-
- sockStmr.setPort(PORT);
-
- sockStmr.setDelimiter(DELIM);
-
- sockStmr.setIgnite(ignite);
-
- sockStmr.setStreamer(stmr);
-
- // Converter from zero-terminated string to Java strings.
- sockStmr.setConverter(new SocketMessageConverter<String>() {
- @Override public String convert(byte[] msg) {
- try {
- return new String(msg, "ASCII");
- }
- catch (UnsupportedEncodingException e) {
- throw new IgniteException(e);
- }
- }
- });
-
- sockStmr.setTupleExtractor(new StreamTupleExtractor<String, AffinityUuid, String>() {
- @Override public Map.Entry<AffinityUuid, String> extract(String word) {
- // By using AffinityUuid we ensure that identical
- // words are processed on the same cluster node.
- return new IgniteBiTuple<>(new AffinityUuid(word), word);
- }
- });
-
- sockStmr.start();
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/fe78d42c/examples/src/main/java/org/apache/ignite/examples/streaming/socket/package-info.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java/org/apache/ignite/examples/streaming/socket/package-info.java b/examples/src/main/java/org/apache/ignite/examples/streaming/socket/package-info.java
deleted file mode 100644
index c516ab4..0000000
--- a/examples/src/main/java/org/apache/ignite/examples/streaming/socket/package-info.java
+++ /dev/null
@@ -1,21 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-/**
- * Contains {@link org.apache.ignite.stream.socket.SocketStreamer} usage examples.
- */
-package org.apache.ignite.examples.streaming.socket;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/fe78d42c/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/QueryWords.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/QueryWords.java b/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/QueryWords.java
index 149aa79..faf8b51 100644
--- a/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/QueryWords.java
+++ b/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/QueryWords.java
@@ -36,6 +36,12 @@ import java.util.*;
* You should start remote nodes by running {@link ExampleNodeStartup} in another JVM.
*/
public class QueryWords {
+ /**
+ * Schedules words query execution.
+ *
+ * @param args Command line arguments (none required).
+ * @throws Exception If failed.
+ */
public static void main(String[] args) throws Exception {
// Mark this cluster member as client.
Ignition.setClientMode(true);
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/fe78d42c/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/StreamWords.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/StreamWords.java b/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/StreamWords.java
index cc3c0cb..26be178 100644
--- a/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/StreamWords.java
+++ b/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/StreamWords.java
@@ -35,6 +35,12 @@ import java.io.*;
* You should start remote nodes by running {@link ExampleNodeStartup} in another JVM.
*/
public class StreamWords {
+ /**
+ * Starts words streaming.
+ *
+ * @param args Command line arguments (none required).
+ * @throws Exception If failed.
+ */
public static void main(String[] args) throws Exception {
// Mark this cluster member as client.
Ignition.setClientMode(true);
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/fe78d42c/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/socket/WordsSocketStreamerClient.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/socket/WordsSocketStreamerClient.java b/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/socket/WordsSocketStreamerClient.java
new file mode 100644
index 0000000..ea3beaa
--- /dev/null
+++ b/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/socket/WordsSocketStreamerClient.java
@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.examples.streaming.wordcount.socket;
+
+import org.apache.ignite.examples.*;
+import org.apache.ignite.examples.streaming.wordcount.*;
+import org.apache.ignite.stream.socket.*;
+
+import java.io.*;
+import java.net.*;
+
+/**
+ * Sends words to socket server based on {@link SocketStreamer} using message delimiter based protocol.
+ * Example illustrates usage of TCP socket streamer in case of non-Java clients.
+ * In this example words are zero-terminated strings.
+ * <p>
+ * To start the example, you should:
+ * <ul>
+ * <li>Start a few nodes using {@link ExampleNodeStartup} or by starting remote nodes as specified below.</li>
+ * <li>Start socket server using {@link WordsSocketStreamerServer}.</li>
+ * <li>Start a few socket clients using {@link WordsSocketStreamerClient}.</li>
+ * <li>Start querying popular words using {@link QueryWords}.</li>
+ * </ul>
+ * <p>
+ * You should start remote nodes by running {@link ExampleNodeStartup} in another JVM.
+ */
+public class WordsSocketStreamerClient {
+ /** Port. */
+ private static final int PORT = 5555;
+
+ /** Delimiter. */
+ private static final byte[] DELIM = new byte[] {0};
+
+ /**
+ * @param args Args.
+ */
+ public static void main(String[] args) throws IOException {
+ InetAddress addr = InetAddress.getLocalHost();
+
+ try (
+ Socket sock = new Socket(addr, PORT);
+ OutputStream oos = new BufferedOutputStream(sock.getOutputStream())
+ ) {
+ System.out.println("Words streaming started.");
+
+ while (true) {
+ try (InputStream in = StreamWords.class.getResourceAsStream("../wordcount/alice-in-wonderland.txt");
+ LineNumberReader rdr = new LineNumberReader(new InputStreamReader(in))) {
+ for (String line = rdr.readLine(); line != null; line = rdr.readLine()) {
+ for (String word : line.split(" ")) {
+ if (!word.isEmpty()) {
+ // Stream words into Ignite through socket.
+ byte[] arr = word.getBytes("ASCII");
+
+ // Write message
+ oos.write(arr);
+
+ // Write message delimiter
+ oos.write(DELIM);
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/fe78d42c/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/socket/WordsSocketStreamerServer.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/socket/WordsSocketStreamerServer.java b/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/socket/WordsSocketStreamerServer.java
new file mode 100644
index 0000000..259c925
--- /dev/null
+++ b/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/socket/WordsSocketStreamerServer.java
@@ -0,0 +1,111 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.examples.streaming.wordcount.socket;
+
+import org.apache.ignite.*;
+import org.apache.ignite.cache.affinity.*;
+import org.apache.ignite.examples.*;
+import org.apache.ignite.examples.streaming.wordcount.*;
+import org.apache.ignite.lang.*;
+import org.apache.ignite.stream.*;
+import org.apache.ignite.stream.socket.*;
+
+import java.io.*;
+import java.net.*;
+import java.util.*;
+
+/**
+ * Receives words through socket using {@link SocketStreamer} and message delimiter based protocol
+ * and streams them into Ignite cache. Example illustrates usage of TCP socket streamer in case of non-Java clients.
+ * In this example words are zero-terminated strings.
+ * <p>
+ * To start the example, you should:
+ * <ul>
+ * <li>Start a few nodes using {@link ExampleNodeStartup} or by starting remote nodes as specified below.</li>
+ * <li>Start socket server using {@link WordsSocketStreamerServer}.</li>
+ * <li>Start a few socket clients using {@link WordsSocketStreamerClient}.</li>
+ * <li>Start querying popular words using {@link QueryWords}.</li>
+ * </ul>
+ * <p>
+ * You should start remote nodes by running {@link ExampleNodeStartup} in another JVM.
+ */
+public class WordsSocketStreamerServer {
+ /** Port. */
+ private static final int PORT = 5555;
+
+ /** Delimiter. */
+ private static final byte[] DELIM = new byte[] {0};
+
+ /**
+ * @param args Args.
+ */
+ public static void main(String[] args) throws InterruptedException, IOException {
+ // Mark this cluster member as client.
+ Ignition.setClientMode(true);
+
+ Ignite ignite = Ignition.start("examples/config/example-ignite.xml");
+
+ if (!ExamplesUtils.hasServerNodes(ignite)) {
+ ignite.close();
+
+ return;
+ }
+
+ // The cache is configured with sliding window holding 1 second of the streaming data.
+ IgniteCache<AffinityUuid, String> stmCache = ignite.getOrCreateCache(CacheConfig.wordCache());
+
+ IgniteDataStreamer<AffinityUuid, String> stmr = ignite.dataStreamer(stmCache.getName());
+
+ InetAddress addr = InetAddress.getLocalHost();
+
+ // Configure socket streamer
+ SocketStreamer<String, AffinityUuid, String> sockStmr = new SocketStreamer<>();
+
+ sockStmr.setAddr(addr);
+
+ sockStmr.setPort(PORT);
+
+ sockStmr.setDelimiter(DELIM);
+
+ sockStmr.setIgnite(ignite);
+
+ sockStmr.setStreamer(stmr);
+
+ // Converter from zero-terminated string to Java strings.
+ sockStmr.setConverter(new SocketMessageConverter<String>() {
+ @Override public String convert(byte[] msg) {
+ try {
+ return new String(msg, "ASCII");
+ }
+ catch (UnsupportedEncodingException e) {
+ throw new IgniteException(e);
+ }
+ }
+ });
+
+ sockStmr.setTupleExtractor(new StreamTupleExtractor<String, AffinityUuid, String>() {
+ @Override public Map.Entry<AffinityUuid, String> extract(String word) {
+ // By using AffinityUuid we ensure that identical
+ // words are processed on the same cluster node.
+ return new IgniteBiTuple<>(new AffinityUuid(word), word);
+ }
+ });
+
+ sockStmr.start();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/fe78d42c/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/socket/package-info.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/socket/package-info.java b/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/socket/package-info.java
new file mode 100644
index 0000000..048299f
--- /dev/null
+++ b/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/socket/package-info.java
@@ -0,0 +1,21 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * Contains {@link org.apache.ignite.stream.socket.SocketStreamer} usage examples.
+ */
+package org.apache.ignite.examples.streaming.wordcount.socket;
[9/9] incubator-ignite git commit: ignite-430 Words count Socket
streamer examples
Posted by sb...@apache.org.
ignite-430 Words count Socket streamer 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/896b426b
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/896b426b
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/896b426b
Branch: refs/heads/ignite-sprint-5
Commit: 896b426bff3e37aeebbb00f54e179492543f1a1d
Parents: fe78d42
Author: agura <ag...@gridgain.com>
Authored: Thu May 14 20:29:33 2015 +0300
Committer: agura <ag...@gridgain.com>
Committed: Fri May 15 03:44:40 2015 +0300
----------------------------------------------------------------------
.../streaming/StreamTransformerExample.java | 4 +-
.../streaming/StreamVisitorExample.java | 4 +-
.../socket/WordsSocketStreamerServer.java | 93 --------------------
.../streaming/wordcount/QueryWords.java | 4 +-
.../streaming/wordcount/StreamWords.java | 4 +-
.../socket/WordsSocketStreamerClient.java | 14 +--
.../socket/WordsSocketStreamerServer.java | 27 ++++--
7 files changed, 31 insertions(+), 119 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/896b426b/examples/src/main/java/org/apache/ignite/examples/streaming/StreamTransformerExample.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java/org/apache/ignite/examples/streaming/StreamTransformerExample.java b/examples/src/main/java/org/apache/ignite/examples/streaming/StreamTransformerExample.java
index 5e95892..966fce2 100644
--- a/examples/src/main/java/org/apache/ignite/examples/streaming/StreamTransformerExample.java
+++ b/examples/src/main/java/org/apache/ignite/examples/streaming/StreamTransformerExample.java
@@ -30,11 +30,9 @@ import java.util.*;
* Stream random numbers into the streaming cache.
* To start the example, you should:
* <ul>
- * <li>Start a few nodes using {@link ExampleNodeStartup} or by starting remote nodes as specified below.</li>
+ * <li>Start a few nodes using {@link ExampleNodeStartup}.</li>
* <li>Start streaming using {@link StreamTransformerExample}.</li>
* </ul>
- * <p>
- * You should start remote nodes by running {@link ExampleNodeStartup} in another JVM.
*/
public class StreamTransformerExample {
/** Random number generator. */
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/896b426b/examples/src/main/java/org/apache/ignite/examples/streaming/StreamVisitorExample.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java/org/apache/ignite/examples/streaming/StreamVisitorExample.java b/examples/src/main/java/org/apache/ignite/examples/streaming/StreamVisitorExample.java
index 0fbce68..baae5af 100644
--- a/examples/src/main/java/org/apache/ignite/examples/streaming/StreamVisitorExample.java
+++ b/examples/src/main/java/org/apache/ignite/examples/streaming/StreamVisitorExample.java
@@ -31,11 +31,9 @@ import java.util.*;
* Stream random numbers into the streaming cache.
* To start the example, you should:
* <ul>
- * <li>Start a few nodes using {@link ExampleNodeStartup} or by starting remote nodes as specified below.</li>
+ * <li>Start a few nodes using {@link ExampleNodeStartup}.</li>
* <li>Start streaming using {@link StreamVisitorExample}.</li>
* </ul>
- * <p>
- * You should start remote nodes by running {@link ExampleNodeStartup} in another JVM.
*/
public class StreamVisitorExample {
/** Random number generator. */
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/896b426b/examples/src/main/java/org/apache/ignite/examples/streaming/socket/WordsSocketStreamerServer.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java/org/apache/ignite/examples/streaming/socket/WordsSocketStreamerServer.java b/examples/src/main/java/org/apache/ignite/examples/streaming/socket/WordsSocketStreamerServer.java
deleted file mode 100644
index 5af746d..0000000
--- a/examples/src/main/java/org/apache/ignite/examples/streaming/socket/WordsSocketStreamerServer.java
+++ /dev/null
@@ -1,93 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.examples.streaming.socket;
-
-import org.apache.ignite.*;
-import org.apache.ignite.cache.affinity.*;
-import org.apache.ignite.examples.*;
-import org.apache.ignite.examples.streaming.wordcount.*;
-import org.apache.ignite.lang.*;
-import org.apache.ignite.stream.*;
-import org.apache.ignite.stream.socket.*;
-
-import java.io.*;
-import java.net.*;
-import java.util.*;
-
-/**
- * Receives words through socket using {@link SocketStreamer} and message size based protocol
- * and streams them into Ignite cache.
- * <p>
- * To start the example, you should:
- * <ul>
- * <li>Start a few nodes using {@link ExampleNodeStartup} or by starting remote nodes as specified below.</li>
- * <li>Start socket server using {@link WordsSocketStreamerServer}.</li>
- * <li>Start a few socket clients using {@link WordsSocketStreamerClient}.</li>
- * <li>Start querying popular words using {@link QueryWords}.</li>
- * </ul>
- * <p>
- * You should start remote nodes by running {@link ExampleNodeStartup} in another JVM.
- */
-public class WordsSocketStreamerServer {
- /** Port. */
- private static final int PORT = 5555;
-
- /**
- * @param args Args.
- */
- public static void main(String[] args) throws InterruptedException, IOException {
- // Mark this cluster member as client.
- Ignition.setClientMode(true);
-
- Ignite ignite = Ignition.start("examples/config/example-ignite.xml");
-
- if (!ExamplesUtils.hasServerNodes(ignite)) {
- ignite.close();
-
- return;
- }
-
- // The cache is configured with sliding window holding 1 second of the streaming data.
- IgniteCache<AffinityUuid, String> stmCache = ignite.getOrCreateCache(CacheConfig.wordCache());
-
- IgniteDataStreamer<AffinityUuid, String> stmr = ignite.dataStreamer(stmCache.getName());
-
- InetAddress addr = InetAddress.getLocalHost();
-
- // Configure socket streamer
- SocketStreamer<String, AffinityUuid, String> sockStmr = new SocketStreamer<>();
-
- sockStmr.setAddr(addr);
-
- sockStmr.setPort(PORT);
-
- sockStmr.setIgnite(ignite);
-
- sockStmr.setStreamer(stmr);
-
- sockStmr.setTupleExtractor(new StreamTupleExtractor<String, AffinityUuid, String>() {
- @Override public Map.Entry<AffinityUuid, String> extract(String word) {
- // By using AffinityUuid we ensure that identical
- // words are processed on the same cluster node.
- return new IgniteBiTuple<>(new AffinityUuid(word), word);
- }
- });
-
- sockStmr.start();
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/896b426b/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/QueryWords.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/QueryWords.java b/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/QueryWords.java
index faf8b51..58c6ef2 100644
--- a/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/QueryWords.java
+++ b/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/QueryWords.java
@@ -28,12 +28,10 @@ import java.util.*;
* Periodically query popular numbers from the streaming cache.
* To start the example, you should:
* <ul>
- * <li>Start a few nodes using {@link ExampleNodeStartup} or by starting remote nodes as specified below.</li>
+ * <li>Start a few nodes using {@link ExampleNodeStartup}.</li>
* <li>Start streaming using {@link StreamWords}.</li>
* <li>Start querying popular words using {@link QueryWords}.</li>
* </ul>
- * <p>
- * You should start remote nodes by running {@link ExampleNodeStartup} in another JVM.
*/
public class QueryWords {
/**
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/896b426b/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/StreamWords.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/StreamWords.java b/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/StreamWords.java
index 26be178..6024c4b 100644
--- a/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/StreamWords.java
+++ b/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/StreamWords.java
@@ -27,12 +27,10 @@ import java.io.*;
* Stream words into Ignite cache.
* To start the example, you should:
* <ul>
- * <li>Start a few nodes using {@link ExampleNodeStartup} or by starting remote nodes as specified below.</li>
+ * <li>Start a few nodes using {@link ExampleNodeStartup}.</li>
* <li>Start streaming using {@link StreamWords}.</li>
* <li>Start querying popular words using {@link QueryWords}.</li>
* </ul>
- * <p>
- * You should start remote nodes by running {@link ExampleNodeStartup} in another JVM.
*/
public class StreamWords {
/**
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/896b426b/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/socket/WordsSocketStreamerClient.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/socket/WordsSocketStreamerClient.java b/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/socket/WordsSocketStreamerClient.java
index ea3beaa..c4d7b8c 100644
--- a/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/socket/WordsSocketStreamerClient.java
+++ b/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/socket/WordsSocketStreamerClient.java
@@ -25,19 +25,19 @@ import java.io.*;
import java.net.*;
/**
- * Sends words to socket server based on {@link SocketStreamer} using message delimiter based protocol.
- * Example illustrates usage of TCP socket streamer in case of non-Java clients.
- * In this example words are zero-terminated strings.
+ * Example demonstrates streaming of data from external components into Ignite cache.
+ * <p>
+ * {@code WordsSocketStreamerClient} is simple socket streaming client implementation that sends words to socket server
+ * based on {@link SocketStreamer} using message delimiter based protocol. Example illustrates usage of TCP socket
+ * streamer in case of non-Java clients. In this example words are zero-terminated strings.
* <p>
* To start the example, you should:
* <ul>
- * <li>Start a few nodes using {@link ExampleNodeStartup} or by starting remote nodes as specified below.</li>
+ * <li>Start a few nodes using {@link ExampleNodeStartup}.</li>
* <li>Start socket server using {@link WordsSocketStreamerServer}.</li>
* <li>Start a few socket clients using {@link WordsSocketStreamerClient}.</li>
* <li>Start querying popular words using {@link QueryWords}.</li>
* </ul>
- * <p>
- * You should start remote nodes by running {@link ExampleNodeStartup} in another JVM.
*/
public class WordsSocketStreamerClient {
/** Port. */
@@ -59,7 +59,7 @@ public class WordsSocketStreamerClient {
System.out.println("Words streaming started.");
while (true) {
- try (InputStream in = StreamWords.class.getResourceAsStream("../wordcount/alice-in-wonderland.txt");
+ try (InputStream in = WordsSocketStreamerClient.class.getResourceAsStream("../alice-in-wonderland.txt");
LineNumberReader rdr = new LineNumberReader(new InputStreamReader(in))) {
for (String line = rdr.readLine(); line != null; line = rdr.readLine()) {
for (String word : line.split(" ")) {
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/896b426b/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/socket/WordsSocketStreamerServer.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/socket/WordsSocketStreamerServer.java b/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/socket/WordsSocketStreamerServer.java
index 259c925..6a8911c 100644
--- a/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/socket/WordsSocketStreamerServer.java
+++ b/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount/socket/WordsSocketStreamerServer.java
@@ -30,19 +30,20 @@ import java.net.*;
import java.util.*;
/**
- * Receives words through socket using {@link SocketStreamer} and message delimiter based protocol
+ * Example demonstrates streaming of data from external components into Ignite cache.
+ * <p>
+ * {@code WordsSocketStreamerServer} is simple socket streaming server implementation that
+ * receives words from socket using {@link SocketStreamer} and message delimiter based protocol
* and streams them into Ignite cache. Example illustrates usage of TCP socket streamer in case of non-Java clients.
* In this example words are zero-terminated strings.
* <p>
* To start the example, you should:
* <ul>
- * <li>Start a few nodes using {@link ExampleNodeStartup} or by starting remote nodes as specified below.</li>
+ * <li>Start a few nodes using {@link ExampleNodeStartup}.</li>
* <li>Start socket server using {@link WordsSocketStreamerServer}.</li>
* <li>Start a few socket clients using {@link WordsSocketStreamerClient}.</li>
* <li>Start querying popular words using {@link QueryWords}.</li>
* </ul>
- * <p>
- * You should start remote nodes by running {@link ExampleNodeStartup} in another JVM.
*/
public class WordsSocketStreamerServer {
/** Port. */
@@ -52,9 +53,12 @@ public class WordsSocketStreamerServer {
private static final byte[] DELIM = new byte[] {0};
/**
- * @param args Args.
+ * Starts socket streaming server.
+ *
+ * @param args Command line arguments (none required).
+ * @throws Exception If failed.
*/
- public static void main(String[] args) throws InterruptedException, IOException {
+ public static void main(String[] args) throws Exception {
// Mark this cluster member as client.
Ignition.setClientMode(true);
@@ -106,6 +110,15 @@ public class WordsSocketStreamerServer {
}
});
- sockStmr.start();
+ try {
+ sockStmr.start();
+ }
+ catch (IgniteException e) {
+ System.out.println("Streaming server didn't start due to an error: ");
+
+ e.printStackTrace();
+
+ ignite.close();
+ }
}
}
[6/9] incubator-ignite git commit: ignite-430 Words count Socket
streamer examples
Posted by sb...@apache.org.
ignite-430 Words count Socket streamer 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/d87efce0
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/d87efce0
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/d87efce0
Branch: refs/heads/ignite-sprint-5
Commit: d87efce0af1ad269ea1a4182329f441b6ff32122
Parents: 53995dc
Author: agura <ag...@gridgain.com>
Authored: Wed May 13 18:38:48 2015 +0300
Committer: agura <ag...@gridgain.com>
Committed: Fri May 15 03:44:31 2015 +0300
----------------------------------------------------------------------
.../streaming/socket/SocketStreamerExample.java | 112 +++++++------------
.../socket/ZStringsSocketStreamerExample.java | 76 ++++++-------
.../examples/streaming/socket/package-info.java | 1 +
3 files changed, 75 insertions(+), 114 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/d87efce0/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 73cb970..487572a 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
@@ -18,36 +18,30 @@
package org.apache.ignite.examples.streaming.socket;
import org.apache.ignite.*;
+import org.apache.ignite.cache.affinity.*;
import org.apache.ignite.examples.*;
-import org.apache.ignite.examples.streaming.numbers.*;
+import org.apache.ignite.examples.streaming.wordcount.*;
import org.apache.ignite.lang.*;
-import org.apache.ignite.stream.*;
import org.apache.ignite.stream.adapters.*;
import org.apache.ignite.stream.socket.*;
-import javax.cache.processor.*;
import java.io.*;
import java.net.*;
import java.util.*;
/**
- * Streams random numbers into the streaming cache using {@link IgniteSocketStreamer}.
+ * Stream words into Ignite cache through socket using {@link IgniteSocketStreamer} and message size based protocol.
+ * <p>
* To start the example, you should:
* <ul>
- * <li>Start a few nodes using {@link ExampleNodeStartup} or by starting remote nodes as specified below.</li>
- * <li>Start streaming using {@link SocketStreamerExample}.</li>
- * <li>Start querying popular numbers using {@link QueryPopularNumbers}.</li>
+ * <li>Start a few nodes using {@link ExampleNodeStartup} or by starting remote nodes as specified below.</li>
+ * <li>Start streaming using {@link SocketStreamerExample}.</li>
+ * <li>Start querying popular numbers using {@link QueryWords}.</li>
* </ul>
* <p>
* You should start remote nodes by running {@link ExampleNodeStartup} in another JVM.
*/
public class SocketStreamerExample {
- /** Random number generator. */
- private static final Random RAND = new Random();
-
- /** Range within which to generate numbers. */
- private static final int RANGE = 1000;
-
/** Port. */
private static final int PORT = 5555;
@@ -63,27 +57,13 @@ public class SocketStreamerExample {
return;
// The cache is configured with sliding window holding 1 second of the streaming data.
- IgniteCache<Integer, Long> stmCache = ignite.getOrCreateCache(CacheConfig.randomNumbersCache());
-
- try (IgniteDataStreamer<Integer, Long> stmr = ignite.dataStreamer(stmCache.getName())) {
- // Allow data updates.
- stmr.allowOverwrite(true);
-
- // Configure data transformation to count instances of the same word.
- 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;
- }
- });
+ IgniteCache<AffinityUuid, String> stmCache = ignite.getOrCreateCache(CacheConfig.wordCache());
+ try (IgniteDataStreamer<AffinityUuid, String> stmr = ignite.dataStreamer(stmCache.getName())) {
InetAddress addr = InetAddress.getLocalHost();
- IgniteSocketStreamer<Tuple, Integer, Long> sockStmr = new IgniteSocketStreamer<>();
+ // Configure socket streamer
+ IgniteSocketStreamer<String, AffinityUuid, String> sockStmr = new IgniteSocketStreamer<>();
sockStmr.setAddr(addr);
@@ -93,9 +73,11 @@ public class SocketStreamerExample {
sockStmr.setStreamer(stmr);
- sockStmr.setTupleExtractor(new StreamTupleExtractor<Tuple, Integer, Long>() {
- @Override public Map.Entry<Integer, Long> extract(Tuple tuple) {
- return new IgniteBiTuple<>(tuple.key, tuple.cnt);
+ sockStmr.setTupleExtractor(new StreamTupleExtractor<String, AffinityUuid, String>() {
+ @Override public Map.Entry<AffinityUuid, String> extract(String word) {
+ // By using AffinityUuid we ensure that identical
+ // words are processed on the same cluster node.
+ return new IgniteBiTuple<>(new AffinityUuid(word), word);
}
});
@@ -114,45 +96,33 @@ public class SocketStreamerExample {
try (Socket sock = new Socket(addr, port);
OutputStream oos = new BufferedOutputStream(sock.getOutputStream())) {
while (true) {
- try (ByteArrayOutputStream bos = new ByteArrayOutputStream();
- ObjectOutputStream out = new ObjectOutputStream(bos)) {
- Tuple tuple = new Tuple(RAND.nextInt(RANGE), 1L);
-
- out.writeObject(tuple);
-
- byte[] arr = bos.toByteArray();
-
- oos.write(arr.length >>> 24);
- oos.write(arr.length >>> 16);
- oos.write(arr.length >>> 8);
- oos.write(arr.length);
-
- oos.write(arr);
+ try (InputStream in = StreamWords.class.getResourceAsStream("../wordcount/alice-in-wonderland.txt");
+ LineNumberReader rdr = new LineNumberReader(new InputStreamReader(in))) {
+ for (String line = rdr.readLine(); line != null; line = rdr.readLine()) {
+ for (String word : line.split(" ")) {
+ if (!word.isEmpty()) {
+ // Stream words into Ignite through socket.
+ try (ByteArrayOutputStream bos = new ByteArrayOutputStream();
+ ObjectOutputStream out = new ObjectOutputStream(bos)) {
+
+ // Write message
+ out.writeObject(word);
+
+ byte[] arr = bos.toByteArray();
+
+ // Write message length
+ oos.write(arr.length >>> 24);
+ oos.write(arr.length >>> 16);
+ oos.write(arr.length >>> 8);
+ oos.write(arr.length);
+
+ oos.write(arr);
+ }
+ }
+ }
+ }
}
}
}
}
-
- /**
- * Tuple.
- */
- private static class Tuple implements Serializable {
- /** Serial version uid. */
- private static final long serialVersionUID = 0;
-
- /** Key. */
- private final int key;
-
- /** Count. */
- private final long cnt;
-
- /**
- * @param key Key.
- * @param cnt Count.
- */
- public Tuple(int key, long cnt) {
- this.key = key;
- this.cnt = cnt;
- }
- }
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/d87efce0/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 a535c73..fa5aa28 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
@@ -18,40 +18,34 @@
package org.apache.ignite.examples.streaming.socket;
import org.apache.ignite.*;
+import org.apache.ignite.cache.affinity.*;
import org.apache.ignite.examples.*;
-import org.apache.ignite.examples.streaming.numbers.*;
+import org.apache.ignite.examples.streaming.wordcount.*;
import org.apache.ignite.lang.*;
-import org.apache.ignite.stream.*;
import org.apache.ignite.stream.adapters.*;
import org.apache.ignite.stream.socket.*;
-import javax.cache.processor.*;
import java.io.*;
import java.net.*;
import java.util.*;
/**
- * Stream random numbers into the streaming cache using {@link IgniteSocketStreamer}.
+ * Stream words into Ignite cache through socket using {@link IgniteSocketStreamer} and message delimiter based
+ * protocol.
* <p>
* Example illustrates usage of TCP socket streamer in case of non-Java clients. In this example client streams
* zero-terminated strings.
* <p>
* To start the example, you should:
* <ul>
- * <li>Start a few nodes using {@link ExampleNodeStartup} or by starting remote nodes as specified below.</li>
- * <li>Start streaming using {@link ZStringsSocketStreamerExample}.</li>
- * <li>Start querying popular numbers using {@link QueryPopularNumbers}.</li>
+ * <li>Start a few nodes using {@link ExampleNodeStartup} or by starting remote nodes as specified below.</li>
+ * <li>Start streaming using {@link ZStringsSocketStreamerExample}.</li>
+ * <li>Start querying popular numbers using {@link QueryWords}.</li>
* </ul>
* <p>
* You should start remote nodes by running {@link ExampleNodeStartup} in another JVM.
*/
public class ZStringsSocketStreamerExample {
- /** Random number generator. */
- private static final Random RAND = new Random();
-
- /** Range within which to generate numbers. */
- private static final int RANGE = 1000;
-
/** Port. */
private static final int PORT = 5555;
@@ -70,27 +64,13 @@ public class ZStringsSocketStreamerExample {
return;
// The cache is configured with sliding window holding 1 second of the streaming data.
- IgniteCache<Integer, Long> stmCache = ignite.getOrCreateCache(CacheConfig.randomNumbersCache());
-
- try (IgniteDataStreamer<Integer, Long> stmr = ignite.dataStreamer(stmCache.getName())) {
- // Allow data updates.
- stmr.allowOverwrite(true);
-
- // Configure data transformation to count instances of the same word.
- 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;
- }
- });
+ IgniteCache<AffinityUuid, String> stmCache = ignite.getOrCreateCache(CacheConfig.wordCache());
+ try (IgniteDataStreamer<AffinityUuid, String> stmr = ignite.dataStreamer(stmCache.getName())) {
InetAddress addr = InetAddress.getLocalHost();
- IgniteSocketStreamer<String, Integer, Long> sockStmr = new IgniteSocketStreamer<>();
+ // Configure socket streamer
+ IgniteSocketStreamer<String, AffinityUuid, String> sockStmr = new IgniteSocketStreamer<>();
sockStmr.setAddr(addr);
@@ -114,10 +94,11 @@ public class ZStringsSocketStreamerExample {
}
});
- sockStmr.setTupleExtractor(new StreamTupleExtractor<String, Integer, Long>() {
- @Override public Map.Entry<Integer, Long> extract(String input) {
- String[] pair = input.split("=");
- return new IgniteBiTuple<>(Integer.parseInt(pair[0]), Long.parseLong(pair[1]));
+ sockStmr.setTupleExtractor(new StreamTupleExtractor<String, AffinityUuid, String>() {
+ @Override public Map.Entry<AffinityUuid, String> extract(String word) {
+ // By using AffinityUuid we ensure that identical
+ // words are processed on the same cluster node.
+ return new IgniteBiTuple<>(new AffinityUuid(word), word);
}
});
@@ -137,14 +118,23 @@ public class ZStringsSocketStreamerExample {
OutputStream oos = new BufferedOutputStream(sock.getOutputStream())) {
while (true) {
- int key = RAND.nextInt(RANGE);
-
- String str = key + "=1";
-
- byte[] arr = str.getBytes("ASCII");
-
- oos.write(arr);
- oos.write(DELIM);
+ try (InputStream in = StreamWords.class.getResourceAsStream("../wordcount/alice-in-wonderland.txt");
+ LineNumberReader rdr = new LineNumberReader(new InputStreamReader(in))) {
+ for (String line = rdr.readLine(); line != null; line = rdr.readLine()) {
+ for (String word : line.split(" ")) {
+ if (!word.isEmpty()) {
+ // Stream words into Ignite through socket.
+ byte[] arr = word.getBytes("ASCII");
+
+ // Write message
+ oos.write(arr);
+
+ // Write message delimiter
+ oos.write(DELIM);
+ }
+ }
+ }
+ }
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/d87efce0/examples/src/main/java/org/apache/ignite/examples/streaming/socket/package-info.java
----------------------------------------------------------------------
diff --git a/examples/src/main/java/org/apache/ignite/examples/streaming/socket/package-info.java b/examples/src/main/java/org/apache/ignite/examples/streaming/socket/package-info.java
index ae7bdf9..d0a480a 100644
--- a/examples/src/main/java/org/apache/ignite/examples/streaming/socket/package-info.java
+++ b/examples/src/main/java/org/apache/ignite/examples/streaming/socket/package-info.java
@@ -16,6 +16,7 @@
*/
/**
+ * <!-- Package description. -->
* Contains {@link org.apache.ignite.stream.socket.IgniteSocketStreamer} usage examples.
*/
package org.apache.ignite.examples.streaming.socket;
\ No newline at end of file
[3/9] incubator-ignite git commit: # ignite-669 - streaming design.
Posted by sb...@apache.org.
# ignite-669 - streaming design.
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/877eb764
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/877eb764
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/877eb764
Branch: refs/heads/ignite-sprint-5
Commit: 877eb7649e61ca53765c42b4244a7d46a39d1179
Parents: be64e1d
Author: Dmitiry Setrakyan <ds...@gridgain.com>
Authored: Thu Apr 2 03:11:54 2015 -0700
Committer: agura <ag...@gridgain.com>
Committed: Fri May 15 03:44:21 2015 +0300
----------------------------------------------------------------------
.../java/org/apache/ignite/stream/adapters/StreamAdapter.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/877eb764/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 f2e0da9..9d4772f 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
@@ -69,14 +69,14 @@ public abstract class StreamAdapter<T, K, V> {
/**
* @return Provided tuple extractor.
*/
- public StreamTupleExtractor<T, K, V> getConverter() {
+ public StreamTupleExtractor<T, K, V> getTupleExtractor() {
return extractor;
}
/**
* @param extractor Extractor for key-value tuples from messages.
*/
- public void setExtractor(StreamTupleExtractor<T, K, V> extractor) {
+ public void setTupleExtractor(StreamTupleExtractor<T, K, V> extractor) {
this.extractor = extractor;
}