You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by pt...@apache.org on 2016/05/24 20:21:45 UTC
[1/5] storm git commit: FluxShellBolt can emit into custom stream.
Repository: storm
Updated Branches:
refs/heads/master f818c7d6a -> a70254659
FluxShellBolt can emit into custom stream.
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/25dd41ed
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/25dd41ed
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/25dd41ed
Branch: refs/heads/master
Commit: 25dd41ed1e03a02aeb82bea7bc7e3f11da3867a6
Parents: 3fe482e
Author: Pavel Grochal <da...@gmail.com>
Authored: Tue May 17 15:11:29 2016 +0200
Committer: Pavel Grochal <da...@gmail.com>
Committed: Tue May 17 15:11:29 2016 +0200
----------------------------------------------------------------------
.../flux/wrappers/bolts/FluxShellBolt.java | 77 ++++++++++++++++++--
1 file changed, 71 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/25dd41ed/external/flux/flux-wrappers/src/main/java/org/apache/storm/flux/wrappers/bolts/FluxShellBolt.java
----------------------------------------------------------------------
diff --git a/external/flux/flux-wrappers/src/main/java/org/apache/storm/flux/wrappers/bolts/FluxShellBolt.java b/external/flux/flux-wrappers/src/main/java/org/apache/storm/flux/wrappers/bolts/FluxShellBolt.java
index 1af1f42..7c96c64 100644
--- a/external/flux/flux-wrappers/src/main/java/org/apache/storm/flux/wrappers/bolts/FluxShellBolt.java
+++ b/external/flux/flux-wrappers/src/main/java/org/apache/storm/flux/wrappers/bolts/FluxShellBolt.java
@@ -26,27 +26,92 @@ import java.util.Map;
/**
* A generic `ShellBolt` implementation that allows you specify output fields
- * without having to subclass `ShellBolt` to do so.
+ * and even streams without having to subclass `ShellBolt` to do so.
*
*/
public class FluxShellBolt extends ShellBolt implements IRichBolt{
- private String[] outputFields;
+ private Map<String, String[]> outputFields;
private Map<String, Object> componentConfig;
+
+ /**
+ * Create a ShellBolt with command line arguments
+ * @param command Command line arguments for the bolt
+ */
+ public FluxShellBolt(String[] command){
+ super(command);
+ this.outputFields = new HashMap<String, String[]>();
+ }
/**
* Create a ShellBolt with command line arguments and output fields
+ *
+ * Keep this constructor for backward compatibility.
+ *
* @param command Command line arguments for the bolt
* @param outputFields Names of fields the bolt will emit (if any).
*/
-
public FluxShellBolt(String[] command, String[] outputFields){
- super(command);
- this.outputFields = outputFields;
+ this(command);
+ this.setDefaultStream(outputFields);
+ }
+
+
+ /**
+ * Set default stream outputFields, this method is called from YAML file:
+ *
+ * ```
+ * bolts:
+ * - className: org.apache.storm.flux.wrappers.bolts.FluxShellBolt
+ * id: my_bolt
+ * constructorArgs:
+ * - [python, my_bolt.py]
+ * configMethods:
+ * - name: setDefaultStream
+ * args:
+ * - [word, count]
+ * ```
+ *
+ * @param outputFields Names of fields the bolt will emit (if any) in default stream.
+ */
+ public void setDefaultStream(String[] outputFields) {
+ this.setNamedStream("default", outputFields);
+ }
+
+ /**
+ * Set custom *named* stream outputFields, this method is called from YAML file:
+ *
+ * ```
+ * bolts:
+ * - className: org.apache.storm.flux.wrappers.bolts.FluxShellBolt
+ * id: my_bolt
+ * constructorArgs:
+ * - [python, my_bolt.py]
+ * configMethods:
+ * - name: setNamedStream
+ * args:
+ * - first
+ * - [word, count]
+ * ```
+ * @param name Name of stream the bolt will emit into.
+ * @param outputFields Names of fields the bolt will emit in custom *named* stream.
+ */
+ public void setNamedStream(String name, String[] outputFields) {
+ this.outputFields.put(name, outputFields);
}
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
- declarer.declare(new Fields(this.outputFields));
+ Iterator it = this.outputFields.entrySet().iterator();
+ while (it.hasNext()) {
+ Map.Entry entryTuple = (Map.Entry)it.next();
+ String key = (String)entryTuple.getKey();
+ String[] value = (String[])entryTuple.getValue();
+ if(key.equals("default")) {
+ declarer.declare(new Fields(value));
+ } else {
+ declarer.declareStream(key, new Fields(value));
+ }
+ }
}
@Override
[2/5] storm git commit: Fixed imports
Posted by pt...@apache.org.
Fixed imports
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/8a175701
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/8a175701
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/8a175701
Branch: refs/heads/master
Commit: 8a175701fa797d4bb4c90814e49f7dbac287caa3
Parents: 25dd41e
Author: Pavel Grochal <da...@gmail.com>
Authored: Tue May 17 15:12:13 2016 +0200
Committer: Pavel Grochal <da...@gmail.com>
Committed: Tue May 17 15:12:13 2016 +0200
----------------------------------------------------------------------
.../java/org/apache/storm/flux/wrappers/bolts/FluxShellBolt.java | 2 ++
1 file changed, 2 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/8a175701/external/flux/flux-wrappers/src/main/java/org/apache/storm/flux/wrappers/bolts/FluxShellBolt.java
----------------------------------------------------------------------
diff --git a/external/flux/flux-wrappers/src/main/java/org/apache/storm/flux/wrappers/bolts/FluxShellBolt.java b/external/flux/flux-wrappers/src/main/java/org/apache/storm/flux/wrappers/bolts/FluxShellBolt.java
index 7c96c64..457b8dd 100644
--- a/external/flux/flux-wrappers/src/main/java/org/apache/storm/flux/wrappers/bolts/FluxShellBolt.java
+++ b/external/flux/flux-wrappers/src/main/java/org/apache/storm/flux/wrappers/bolts/FluxShellBolt.java
@@ -23,6 +23,8 @@ import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import java.util.Map;
+import java.util.HashMap;
+import java.util.Iterator;
/**
* A generic `ShellBolt` implementation that allows you specify output fields
[3/5] storm git commit: FluxShellSpout can emit into streams.
Posted by pt...@apache.org.
FluxShellSpout can emit into streams.
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/d97cab91
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/d97cab91
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/d97cab91
Branch: refs/heads/master
Commit: d97cab9118268b2af51cec77ae31ca6dc2a62220
Parents: 8a17570
Author: Pavel Grochal <da...@gmail.com>
Authored: Tue May 17 15:17:50 2016 +0200
Committer: Pavel Grochal <da...@gmail.com>
Committed: Tue May 17 15:17:50 2016 +0200
----------------------------------------------------------------------
.../flux/wrappers/spouts/FluxShellSpout.java | 77 ++++++++++++++++++--
1 file changed, 72 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/d97cab91/external/flux/flux-wrappers/src/main/java/org/apache/storm/flux/wrappers/spouts/FluxShellSpout.java
----------------------------------------------------------------------
diff --git a/external/flux/flux-wrappers/src/main/java/org/apache/storm/flux/wrappers/spouts/FluxShellSpout.java b/external/flux/flux-wrappers/src/main/java/org/apache/storm/flux/wrappers/spouts/FluxShellSpout.java
index 1951c1e..d855f12 100644
--- a/external/flux/flux-wrappers/src/main/java/org/apache/storm/flux/wrappers/spouts/FluxShellSpout.java
+++ b/external/flux/flux-wrappers/src/main/java/org/apache/storm/flux/wrappers/spouts/FluxShellSpout.java
@@ -23,29 +23,96 @@ import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import java.util.Map;
+import java.util.HashMap;
+import java.util.Iterator;
/**
* A generic `ShellSpout` implementation that allows you specify output fields
- * without having to subclass `ShellSpout` to do so.
+ * and even streams without having to subclass `ShellSpout` to do so.
*
*/
public class FluxShellSpout extends ShellSpout implements IRichSpout {
- private String[] outputFields;
+ private Map<String, String[]> outputFields;
private Map<String, Object> componentConfig;
+
+ /**
+ * Create a ShellSpout with command line arguments
+ * @param command Command line arguments for the bolt
+ */
+ public FluxShellSpout(String[] command){
+ super(command);
+ this.outputFields = new HashMap<String, String[]>();
+ }
/**
* Create a ShellSpout with command line arguments and output fields
+ *
+ * Keep this constructor for backward compatibility.
+ *
* @param args Command line arguments for the spout
* @param outputFields Names of fields the spout will emit.
*/
public FluxShellSpout(String[] args, String[] outputFields){
- super(args);
- this.outputFields = outputFields;
+ this(args);
+ this.setDefaultStream(outputFields);
+ }
+
+ /**
+ * Set default stream outputFields, this method is called from YAML file:
+ *
+ * ```
+ * spouts:
+ * - className: org.apache.storm.flux.wrappers.bolts.FluxShellSpout
+ * id: my_spout
+ * constructorArgs:
+ * - [python, my_spout.py]
+ * configMethods:
+ * - name: setDefaultStream
+ * args:
+ * - [word, count]
+ * ```
+ *
+ * @param outputFields Names of fields the spout will emit (if any) in default stream.
+ */
+ public void setDefaultStream(String[] outputFields) {
+ this.setNamedStream("default", outputFields);
+ }
+
+ /**
+ * Set custom *named* stream outputFields, this method is called from YAML file:
+ *
+ * ```
+ * spouts:
+ * - className: org.apache.storm.flux.wrappers.bolts.FluxShellSpout
+ * id: my_spout
+ * constructorArgs:
+ * - [python, my_spout.py]
+ * configMethods:
+ * - name: setNamedStream
+ * args:
+ * - first
+ * - [word, count]
+ * ```
+ * @param name Name of stream the spout will emit into.
+ * @param outputFields Names of fields the spout will emit in custom *named* stream.
+ */
+ public void setNamedStream(String name, String[] outputFields) {
+ this.outputFields.put(name, outputFields);
}
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
- declarer.declare(new Fields(this.outputFields));
+ Iterator it = this.outputFields.entrySet().iterator();
+ while (it.hasNext()) {
+ Map.Entry entryTuple = (Map.Entry)it.next();
+ String key = (String)entryTuple.getKey();
+ String[] value = (String[])entryTuple.getValue();
+ if(key.equals("default")) {
+ declarer.declare(new Fields(value));
+ } else {
+ declarer.declareStream(key, new Fields(value));
+ }
+ }
}
@Override
[5/5] storm git commit: add STORM-1862 to changelog
Posted by pt...@apache.org.
add STORM-1862 to changelog
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/a7025465
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/a7025465
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/a7025465
Branch: refs/heads/master
Commit: a702546598dac65401a979b6cb5af11428b44605
Parents: abe8d2a
Author: P. Taylor Goetz <pt...@gmail.com>
Authored: Tue May 24 16:19:04 2016 -0400
Committer: P. Taylor Goetz <pt...@gmail.com>
Committed: Tue May 24 16:19:04 2016 -0400
----------------------------------------------------------------------
CHANGELOG.md | 1 +
1 file changed, 1 insertion(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/a7025465/CHANGELOG.md
----------------------------------------------------------------------
diff --git a/CHANGELOG.md b/CHANGELOG.md
index db0f324..0f8c1f7 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,4 +1,5 @@
## 2.0.0
+ * STORM-1862: Flux ShellSpout and ShellBolt can't emit to named streams
* Storm-1728: TransactionalTridentKafkaSpout error
* STORM-1850: State Checkpointing Documentation update
* STORM-1674: Idle KafkaSpout consumes more bandwidth than needed
[4/5] storm git commit: Merge branch 'master' of
github.com:Darkless012/storm
Posted by pt...@apache.org.
Merge branch 'master' of github.com:Darkless012/storm
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/abe8d2a6
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/abe8d2a6
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/abe8d2a6
Branch: refs/heads/master
Commit: abe8d2a65e9906c3dc6cb81f4784865483a8ca54
Parents: f818c7d d97cab9
Author: P. Taylor Goetz <pt...@gmail.com>
Authored: Tue May 24 16:14:53 2016 -0400
Committer: P. Taylor Goetz <pt...@gmail.com>
Committed: Tue May 24 16:14:53 2016 -0400
----------------------------------------------------------------------
.../flux/wrappers/bolts/FluxShellBolt.java | 79 ++++++++++++++++++--
.../flux/wrappers/spouts/FluxShellSpout.java | 77 +++++++++++++++++--
2 files changed, 145 insertions(+), 11 deletions(-)
----------------------------------------------------------------------