ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject [23/50] [abbrv] incubator-ignite git commit: # ignite-669 - streaming design.
Date Tue, 19 May 2015 17:55:37 GMT
# 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-648
Commit: 0cbe3c68dcea09a39fd0e97909bbbaa6b1091a6b
Parents: 56e67e8
Author: Dmitiry Setrakyan <dsetrakyan@gridgain.com>
Authored: Thu Apr 2 02:45:48 2015 -0700
Committer: agura <agura@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;


Mime
View raw message