You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by ka...@apache.org on 2017/12/08 06:33:24 UTC
[1/3] storm git commit: STORM-2796: Implement support for static
factory methods
Repository: storm
Updated Branches:
refs/heads/master c8b306dc3 -> 67f13b530
http://git-wip-us.apache.org/repos/asf/storm/blob/cfb7d263/flux/flux-examples/src/main/java/org/apache/storm/flux/examples/WordCountClient.java
----------------------------------------------------------------------
diff --git a/flux/flux-examples/src/main/java/org/apache/storm/flux/examples/WordCountClient.java b/flux/flux-examples/src/main/java/org/apache/storm/flux/examples/WordCountClient.java
index eb4fb7a..6e732ae 100644
--- a/flux/flux-examples/src/main/java/org/apache/storm/flux/examples/WordCountClient.java
+++ b/flux/flux-examples/src/main/java/org/apache/storm/flux/examples/WordCountClient.java
@@ -15,8 +15,12 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.storm.flux.examples;
+import java.io.FileInputStream;
+import java.util.Properties;
+
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Get;
@@ -24,15 +28,12 @@ import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;
-import java.io.FileInputStream;
-import java.util.Properties;
-
/**
* Connects to the 'WordCount' HBase table and prints counts for each word.
- *
+ * <p/>
* Assumes you have run (or are running) the YAML topology definition in
* <code>simple_hbase.yaml</code>
- *
+ * <p/>
* You will also need to modify `src/main/resources/hbase-site.xml`
* to point to your HBase instance, and then repackage with `mvn package`.
* This is a known issue.
@@ -40,13 +41,18 @@ import java.util.Properties;
*/
public class WordCountClient {
+ /**
+ * Entry point for WordCountClient.
+ * @param args command line arguments
+ * @throws Exception if an unexpected error occurs
+ */
public static void main(String[] args) throws Exception {
Configuration config = HBaseConfiguration.create();
- if(args.length == 1){
+ if (args.length == 1) {
Properties props = new Properties();
props.load(new FileInputStream(args[0]));
System.out.println("HBase configuration:");
- for(Object key : props.keySet()) {
+ for (Object key : props.keySet()) {
System.out.println(key + "=" + props.get(key));
config.set((String)key, props.getProperty((String)key));
}
http://git-wip-us.apache.org/repos/asf/storm/blob/cfb7d263/flux/flux-examples/src/main/java/org/apache/storm/flux/examples/WordCounter.java
----------------------------------------------------------------------
diff --git a/flux/flux-examples/src/main/java/org/apache/storm/flux/examples/WordCounter.java b/flux/flux-examples/src/main/java/org/apache/storm/flux/examples/WordCounter.java
index 94f65ba..f15af95 100644
--- a/flux/flux-examples/src/main/java/org/apache/storm/flux/examples/WordCounter.java
+++ b/flux/flux-examples/src/main/java/org/apache/storm/flux/examples/WordCounter.java
@@ -15,8 +15,13 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.storm.flux.examples;
+import static org.apache.storm.utils.Utils.tuple;
+
+import java.util.Map;
+
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
@@ -26,14 +31,10 @@ import org.apache.storm.tuple.Tuple;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.Map;
-
-import static org.apache.storm.utils.Utils.tuple;
-
/**
* This bolt is used by the HBase example. It simply emits the first field
* found in the incoming tuple as "word", with a "count" of `1`.
- *
+ * <p/>
* In this case, the downstream HBase bolt handles the counting, so a value
* of `1` will just increment the HBase counter by one.
*/
http://git-wip-us.apache.org/repos/asf/storm/blob/cfb7d263/flux/flux-wrappers/src/main/java/org/apache/storm/flux/wrappers/bolts/FluxShellBolt.java
----------------------------------------------------------------------
diff --git a/flux/flux-wrappers/src/main/java/org/apache/storm/flux/wrappers/bolts/FluxShellBolt.java b/flux/flux-wrappers/src/main/java/org/apache/storm/flux/wrappers/bolts/FluxShellBolt.java
index 05b8e7a..8878107 100644
--- a/flux/flux-wrappers/src/main/java/org/apache/storm/flux/wrappers/bolts/FluxShellBolt.java
+++ b/flux/flux-wrappers/src/main/java/org/apache/storm/flux/wrappers/bolts/FluxShellBolt.java
@@ -15,52 +15,52 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.storm.flux.wrappers.bolts;
-import org.apache.storm.task.ShellBolt;
-import org.apache.storm.topology.IRichBolt;
-import org.apache.storm.topology.OutputFieldsDeclarer;
-import org.apache.storm.tuple.Fields;
+package org.apache.storm.flux.wrappers.bolts;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import org.apache.storm.task.ShellBolt;
+import org.apache.storm.topology.IRichBolt;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.tuple.Fields;
+
/**
* A generic `ShellBolt` implementation that allows you specify output fields
* and even streams without having to subclass `ShellBolt` to do so.
*
*/
-public class FluxShellBolt extends ShellBolt implements IRichBolt{
+public class FluxShellBolt extends ShellBolt implements IRichBolt {
private Map<String, String[]> outputFields;
private Map<String, Object> componentConfig;
/**
- * Create a ShellBolt with command line arguments
+ * Create a ShellBolt with command line arguments.
* @param command Command line arguments for the bolt
*/
- public FluxShellBolt(String[] command){
+ 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){
+ public FluxShellBolt(String[] command, String[] outputFields) {
this(command);
this.setDefaultStream(outputFields);
}
/**
* Add configuration for this bolt. This method is called from YAML file:
- *
+ * <p/>
* ```
* className: "org.apache.storm.flux.wrappers.bolts.FluxShellBolt"
* constructorArgs:
@@ -73,8 +73,8 @@ public class FluxShellBolt extends ShellBolt implements IRichBolt{
* args: ["publisher.data_paths", "actions"]
* ```
*
- * @param key
- * @param value
+ * @param key config key
+ * @param value config value
*/
public void addComponentConfig(String key, Object value) {
if (this.componentConfig == null) {
@@ -85,7 +85,7 @@ public class FluxShellBolt extends ShellBolt implements IRichBolt{
/**
* Add configuration for this bolt. This method is called from YAML file:
- *
+ * <p/>
* ```
* className: "org.apache.storm.flux.wrappers.bolts.FluxShellBolt"
* constructorArgs:
@@ -100,8 +100,8 @@ public class FluxShellBolt extends ShellBolt implements IRichBolt{
* - ["actions"]
* ```
*
- * @param key
- * @param values
+ * @param key config key
+ * @param values config values
*/
public void addComponentConfig(String key, List<Object> values) {
if (this.componentConfig == null) {
@@ -112,7 +112,7 @@ public class FluxShellBolt extends ShellBolt implements IRichBolt{
/**
* Set default stream outputFields, this method is called from YAML file:
- *
+ * <p/>
* ```
* bolts:
* - className: org.apache.storm.flux.wrappers.bolts.FluxShellBolt
@@ -133,7 +133,7 @@ public class FluxShellBolt extends ShellBolt implements IRichBolt{
/**
* Set custom *named* stream outputFields, this method is called from YAML file:
- *
+ * <p/>
* ```
* bolts:
* - className: org.apache.storm.flux.wrappers.bolts.FluxShellBolt
@@ -160,7 +160,7 @@ public class FluxShellBolt extends ShellBolt implements IRichBolt{
Map.Entry entryTuple = (Map.Entry)it.next();
String key = (String)entryTuple.getKey();
String[] value = (String[])entryTuple.getValue();
- if(key.equals("default")) {
+ if (key.equals("default")) {
declarer.declare(new Fields(value));
} else {
declarer.declareStream(key, new Fields(value));
http://git-wip-us.apache.org/repos/asf/storm/blob/cfb7d263/flux/flux-wrappers/src/main/java/org/apache/storm/flux/wrappers/bolts/LogInfoBolt.java
----------------------------------------------------------------------
diff --git a/flux/flux-wrappers/src/main/java/org/apache/storm/flux/wrappers/bolts/LogInfoBolt.java b/flux/flux-wrappers/src/main/java/org/apache/storm/flux/wrappers/bolts/LogInfoBolt.java
index 5f0e84b..3e99791 100644
--- a/flux/flux-wrappers/src/main/java/org/apache/storm/flux/wrappers/bolts/LogInfoBolt.java
+++ b/flux/flux-wrappers/src/main/java/org/apache/storm/flux/wrappers/bolts/LogInfoBolt.java
@@ -34,7 +34,7 @@ public class LogInfoBolt extends BaseBasicBolt {
@Override
public void execute(Tuple tuple, BasicOutputCollector basicOutputCollector) {
- LOG.info("{}", tuple);
+ LOG.info("{}", tuple);
}
@Override
http://git-wip-us.apache.org/repos/asf/storm/blob/cfb7d263/flux/flux-wrappers/src/main/java/org/apache/storm/flux/wrappers/spouts/FluxShellSpout.java
----------------------------------------------------------------------
diff --git a/flux/flux-wrappers/src/main/java/org/apache/storm/flux/wrappers/spouts/FluxShellSpout.java b/flux/flux-wrappers/src/main/java/org/apache/storm/flux/wrappers/spouts/FluxShellSpout.java
index 5fd378d..51ad2c8 100644
--- a/flux/flux-wrappers/src/main/java/org/apache/storm/flux/wrappers/spouts/FluxShellSpout.java
+++ b/flux/flux-wrappers/src/main/java/org/apache/storm/flux/wrappers/spouts/FluxShellSpout.java
@@ -15,17 +15,19 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.storm.flux.wrappers.spouts;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
import org.apache.storm.spout.ShellSpout;
import org.apache.storm.topology.IRichSpout;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
-import java.util.List;
-import java.util.Map;
-import java.util.HashMap;
-import java.util.Iterator;
/**
* A generic `ShellSpout` implementation that allows you specify output fields
@@ -37,30 +39,30 @@ public class FluxShellSpout extends ShellSpout implements IRichSpout {
private Map<String, Object> componentConfig;
/**
- * Create a ShellSpout with command line arguments
+ * Create a ShellSpout with command line arguments.
* @param command Command line arguments for the bolt
*/
- public FluxShellSpout(String[] command){
+ public FluxShellSpout(String[] command) {
super(command);
this.outputFields = new HashMap<String, String[]>();
}
/**
* Create a ShellSpout with command line arguments and output fields
- *
+ * <p/>
* 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){
+ public FluxShellSpout(String[] args, String[] outputFields) {
this(args);
this.setDefaultStream(outputFields);
}
/**
* Add configuration for this spout. This method is called from YAML file:
- *
+ * <p></p>
* ```
* className: "org.apache.storm.flux.wrappers.bolts.FluxShellSpout"
* constructorArgs:
@@ -73,8 +75,8 @@ public class FluxShellSpout extends ShellSpout implements IRichSpout {
* args: ["publisher.data_paths", "actions"]
* ```
*
- * @param key
- * @param value
+ * @param key config key
+ * @param value config value
*/
public void addComponentConfig(String key, Object value) {
if (this.componentConfig == null) {
@@ -85,7 +87,7 @@ public class FluxShellSpout extends ShellSpout implements IRichSpout {
/**
* Add configuration for this spout. This method is called from YAML file:
- *
+ * <p/>
* ```
* className: "org.apache.storm.flux.wrappers.bolts.FluxShellSpout"
* constructorArgs:
@@ -100,8 +102,8 @@ public class FluxShellSpout extends ShellSpout implements IRichSpout {
* - ["actions"]
* ```
*
- * @param key
- * @param values
+ * @param key config key
+ * @param values config values
*/
public void addComponentConfig(String key, List<Object> values) {
if (this.componentConfig == null) {
@@ -112,7 +114,7 @@ public class FluxShellSpout extends ShellSpout implements IRichSpout {
/**
* Set default stream outputFields, this method is called from YAML file:
- *
+ * <p/>
* ```
* spouts:
* - className: org.apache.storm.flux.wrappers.bolts.FluxShellSpout
@@ -133,7 +135,7 @@ public class FluxShellSpout extends ShellSpout implements IRichSpout {
/**
* Set custom *named* stream outputFields, this method is called from YAML file:
- *
+ * <p/>
* ```
* spouts:
* - className: org.apache.storm.flux.wrappers.bolts.FluxShellSpout
@@ -160,7 +162,7 @@ public class FluxShellSpout extends ShellSpout implements IRichSpout {
Map.Entry entryTuple = (Map.Entry)it.next();
String key = (String)entryTuple.getKey();
String[] value = (String[])entryTuple.getValue();
- if(key.equals("default")) {
+ if (key.equals("default")) {
declarer.declare(new Fields(value));
} else {
declarer.declareStream(key, new Fields(value));
http://git-wip-us.apache.org/repos/asf/storm/blob/cfb7d263/flux/pom.xml
----------------------------------------------------------------------
diff --git a/flux/pom.xml b/flux/pom.xml
index 4e04fa8..834cfa7 100644
--- a/flux/pom.xml
+++ b/flux/pom.xml
@@ -41,6 +41,7 @@
<modules>
<module>flux-wrappers</module>
<module>flux-core</module>
+ <module>flux-examples</module>
</modules>
<dependencies>
[3/3] storm git commit: Merge branch 'STORM-2796-merge'
Posted by ka...@apache.org.
Merge branch 'STORM-2796-merge'
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/67f13b53
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/67f13b53
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/67f13b53
Branch: refs/heads/master
Commit: 67f13b530577e2da4993709a62ec3b2aa86b376c
Parents: c8b306d cfb7d26
Author: Jungtaek Lim <ka...@gmail.com>
Authored: Fri Dec 8 15:33:12 2017 +0900
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Fri Dec 8 15:33:12 2017 +0900
----------------------------------------------------------------------
flux/flux-core/pom.xml | 2 +-
.../main/java/org/apache/storm/flux/Flux.java | 69 +++---
.../java/org/apache/storm/flux/FluxBuilder.java | 234 ++++++++++++-------
.../apache/storm/flux/api/TopologySource.java | 8 +-
.../org/apache/storm/flux/model/BeanDef.java | 1 +
.../storm/flux/model/BeanListReference.java | 5 +-
.../apache/storm/flux/model/BeanReference.java | 5 +-
.../org/apache/storm/flux/model/BoltDef.java | 1 +
.../storm/flux/model/ConfigMethodDef.java | 13 +-
.../storm/flux/model/ExecutionContext.java | 23 +-
.../apache/storm/flux/model/GroupingDef.java | 3 +-
.../org/apache/storm/flux/model/IncludeDef.java | 3 +-
.../org/apache/storm/flux/model/ObjectDef.java | 64 ++++-
.../apache/storm/flux/model/PropertyDef.java | 17 +-
.../org/apache/storm/flux/model/SpoutDef.java | 1 +
.../org/apache/storm/flux/model/StreamDef.java | 3 +-
.../apache/storm/flux/model/TopologyDef.java | 114 ++++++---
.../storm/flux/model/TopologySourceDef.java | 3 +-
.../org/apache/storm/flux/model/VertexDef.java | 3 +-
.../apache/storm/flux/parser/FluxParser.java | 155 +++++++-----
.../java/org/apache/storm/flux/TCKTest.java | 4 +
.../org/apache/storm/flux/test/TestBolt.java | 20 ++
.../resources/configs/config-methods-test.yaml | 38 +++
flux/flux-examples/pom.xml | 2 +-
.../flux/examples/StatefulWordCounter.java | 7 +-
.../storm/flux/examples/TestPrintBolt.java | 3 +-
.../storm/flux/examples/TestWindowBolt.java | 5 +-
.../storm/flux/examples/WordCountClient.java | 20 +-
.../apache/storm/flux/examples/WordCounter.java | 11 +-
.../flux/wrappers/bolts/FluxShellBolt.java | 38 +--
.../storm/flux/wrappers/bolts/LogInfoBolt.java | 2 +-
.../flux/wrappers/spouts/FluxShellSpout.java | 36 +--
flux/pom.xml | 1 +
33 files changed, 611 insertions(+), 303 deletions(-)
----------------------------------------------------------------------
[2/3] storm git commit: STORM-2796: Implement support for static
factory methods
Posted by ka...@apache.org.
STORM-2796: Implement support for static factory methods
* add test for factory args
* improve handling of non-primitive numbers
* merge/fix patch from @roshannaik
* Address checkstyle errors.
Squashed by Jungtaek Lim <ka...@gmail.com>
This closes #2409
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/cfb7d263
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/cfb7d263
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/cfb7d263
Branch: refs/heads/master
Commit: cfb7d263d2c5f4b48f106f930ed654b0b539674d
Parents: c8b306d
Author: P. Taylor Goetz <pt...@gmail.com>
Authored: Fri Nov 3 23:17:46 2017 -0400
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Fri Dec 8 15:31:29 2017 +0900
----------------------------------------------------------------------
flux/flux-core/pom.xml | 2 +-
.../main/java/org/apache/storm/flux/Flux.java | 69 +++---
.../java/org/apache/storm/flux/FluxBuilder.java | 234 ++++++++++++-------
.../apache/storm/flux/api/TopologySource.java | 8 +-
.../org/apache/storm/flux/model/BeanDef.java | 1 +
.../storm/flux/model/BeanListReference.java | 5 +-
.../apache/storm/flux/model/BeanReference.java | 5 +-
.../org/apache/storm/flux/model/BoltDef.java | 1 +
.../storm/flux/model/ConfigMethodDef.java | 13 +-
.../storm/flux/model/ExecutionContext.java | 23 +-
.../apache/storm/flux/model/GroupingDef.java | 3 +-
.../org/apache/storm/flux/model/IncludeDef.java | 3 +-
.../org/apache/storm/flux/model/ObjectDef.java | 64 ++++-
.../apache/storm/flux/model/PropertyDef.java | 17 +-
.../org/apache/storm/flux/model/SpoutDef.java | 1 +
.../org/apache/storm/flux/model/StreamDef.java | 3 +-
.../apache/storm/flux/model/TopologyDef.java | 114 ++++++---
.../storm/flux/model/TopologySourceDef.java | 3 +-
.../org/apache/storm/flux/model/VertexDef.java | 3 +-
.../apache/storm/flux/parser/FluxParser.java | 155 +++++++-----
.../java/org/apache/storm/flux/TCKTest.java | 4 +
.../org/apache/storm/flux/test/TestBolt.java | 20 ++
.../resources/configs/config-methods-test.yaml | 38 +++
flux/flux-examples/pom.xml | 2 +-
.../flux/examples/StatefulWordCounter.java | 7 +-
.../storm/flux/examples/TestPrintBolt.java | 3 +-
.../storm/flux/examples/TestWindowBolt.java | 5 +-
.../storm/flux/examples/WordCountClient.java | 20 +-
.../apache/storm/flux/examples/WordCounter.java | 11 +-
.../flux/wrappers/bolts/FluxShellBolt.java | 38 +--
.../storm/flux/wrappers/bolts/LogInfoBolt.java | 2 +-
.../flux/wrappers/spouts/FluxShellSpout.java | 36 +--
flux/pom.xml | 1 +
33 files changed, 611 insertions(+), 303 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/cfb7d263/flux/flux-core/pom.xml
----------------------------------------------------------------------
diff --git a/flux/flux-core/pom.xml b/flux/flux-core/pom.xml
index 7df517b..acdf805 100644
--- a/flux/flux-core/pom.xml
+++ b/flux/flux-core/pom.xml
@@ -112,7 +112,7 @@
<artifactId>maven-checkstyle-plugin</artifactId>
<!--Note - the version would be inherited-->
<configuration>
- <maxAllowedViolations>285</maxAllowedViolations>
+ <maxAllowedViolations>0</maxAllowedViolations>
</configuration>
</plugin>
</plugins>
http://git-wip-us.apache.org/repos/asf/storm/blob/cfb7d263/flux/flux-core/src/main/java/org/apache/storm/flux/Flux.java
----------------------------------------------------------------------
diff --git a/flux/flux-core/src/main/java/org/apache/storm/flux/Flux.java b/flux/flux-core/src/main/java/org/apache/storm/flux/Flux.java
index 1d38ada..e88f4db 100644
--- a/flux/flux-core/src/main/java/org/apache/storm/flux/Flux.java
+++ b/flux/flux-core/src/main/java/org/apache/storm/flux/Flux.java
@@ -15,6 +15,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.storm.flux;
import java.io.BufferedReader;
@@ -46,7 +47,6 @@ import org.slf4j.LoggerFactory;
/**
* Flux entry point.
- *
*/
public class Flux {
private static final Logger LOG = LoggerFactory.getLogger(Flux.class);
@@ -63,6 +63,11 @@ public class Flux {
private static final String OPTION_FILTER = "filter";
private static final String OPTION_ENV_FILTER = "env-filter";
+ /**
+ * Flux main entry point.
+ * @param args command line arguments
+ * @throws Exception if parsing/topology creation fails
+ */
public static void main(String[] args) throws Exception {
Options options = new Options();
@@ -72,11 +77,11 @@ public class Flux {
options.addOption(option(0, "R", OPTION_RESOURCE, "Treat the supplied path as a classpath resource instead of a file."));
- options.addOption(option(1, "s", OPTION_SLEEP, "ms", "When running locally, the amount of time to sleep (in ms.) " +
- "before killing the topology and shutting down the local cluster."));
+ options.addOption(option(1, "s", OPTION_SLEEP, "ms", "When running locally, the amount of time to sleep (in ms.) "
+ + "before killing the topology and shutting down the local cluster."));
- options.addOption(option(0, "d", OPTION_DRY_RUN, "Do not run or deploy the topology. Just build, validate, " +
- "and print information about the topology."));
+ options.addOption(option(0, "d", OPTION_DRY_RUN, "Do not run or deploy the topology. Just build, validate, "
+ + "and print information about the topology."));
options.addOption(option(0, "q", OPTION_NO_DETAIL, "Suppress the printing of topology details."));
@@ -84,15 +89,15 @@ public class Flux {
options.addOption(option(0, "i", OPTION_INACTIVE, "Deploy the topology, but do not activate it."));
- options.addOption(option(1, "z", OPTION_ZOOKEEPER, "host:port", "When running in local mode, use the ZooKeeper at the " +
- "specified <host>:<port> instead of the in-process ZooKeeper. (requires Storm 0.9.3 or later)"));
+ options.addOption(option(1, "z", OPTION_ZOOKEEPER, "host:port", "When running in local mode, use the ZooKeeper at the "
+ + "specified <host>:<port> instead of the in-process ZooKeeper. (requires Storm 0.9.3 or later)"));
- options.addOption(option(1, "f", OPTION_FILTER, "file", "Perform property substitution. Use the specified file " +
- "as a source of properties, and replace keys identified with {$[property name]} with the value defined " +
- "in the properties file."));
+ options.addOption(option(1, "f", OPTION_FILTER, "file", "Perform property substitution. Use the specified file "
+ + "as a source of properties, and replace keys identified with {$[property name]} with the value defined "
+ + "in the properties file."));
- options.addOption(option(0, "e", OPTION_ENV_FILTER, "Perform environment variable substitution. Replace keys" +
- "identified with `${ENV-[NAME]}` will be replaced with the corresponding `NAME` environment value"));
+ options.addOption(option(0, "e", OPTION_ENV_FILTER, "Perform environment variable substitution. Replace keys"
+ + "identified with `${ENV-[NAME]}` will be replaced with the corresponding `NAME` environment value"));
CommandLineParser parser = new BasicParser();
CommandLine cmd = parser.parse(options, args);
@@ -104,11 +109,11 @@ public class Flux {
runCli(cmd);
}
- private static Option option(int argCount, String shortName, String longName, String description){
- return option(argCount, shortName, longName, longName, description);
+ private static Option option(int argCount, String shortName, String longName, String description) {
+ return option(argCount, shortName, longName, longName, description);
}
- private static Option option(int argCount, String shortName, String longName, String argName, String description){
+ private static Option option(int argCount, String shortName, String longName, String argName, String description) {
Option option = OptionBuilder.hasArgs(argCount)
.withArgName(argName)
.withLongOpt(longName)
@@ -119,30 +124,30 @@ public class Flux {
private static void usage(Options options) {
HelpFormatter formatter = new HelpFormatter();
- formatter.printHelp("storm jar <my_topology_uber_jar.jar> " +
- Flux.class.getName() +
- " [options] <topology-config.yaml>", options);
+ formatter.printHelp("storm jar <my_topology_uber_jar.jar> "
+ + Flux.class.getName()
+ + " [options] <topology-config.yaml>", options);
}
- private static void runCli(CommandLine cmd)throws Exception {
- if(!cmd.hasOption(OPTION_NO_SPLASH)) {
+ private static void runCli(CommandLine cmd) throws Exception {
+ if (!cmd.hasOption(OPTION_NO_SPLASH)) {
printSplash();
}
boolean dumpYaml = cmd.hasOption("dump-yaml");
TopologyDef topologyDef = null;
- String filePath = (String)cmd.getArgList().get(0);
+ String filePath = (String) cmd.getArgList().get(0);
// TODO conditionally load properties from a file our resource
String filterProps = null;
- if(cmd.hasOption(OPTION_FILTER)){
+ if (cmd.hasOption(OPTION_FILTER)) {
filterProps = cmd.getOptionValue(OPTION_FILTER);
}
boolean envFilter = cmd.hasOption(OPTION_ENV_FILTER);
- if(cmd.hasOption(OPTION_RESOURCE)){
+ if (cmd.hasOption(OPTION_RESOURCE)) {
printf("Parsing classpath resource: %s", filePath);
topologyDef = FluxParser.parseResource(filePath, dumpYaml, true, filterProps, envFilter);
} else {
@@ -158,16 +163,16 @@ public class Flux {
ExecutionContext context = new ExecutionContext(topologyDef, conf);
StormTopology topology = FluxBuilder.buildTopology(context);
- if(!cmd.hasOption(OPTION_NO_DETAIL)){
+ if (!cmd.hasOption(OPTION_NO_DETAIL)) {
printTopologyInfo(context);
}
- if(!cmd.hasOption(OPTION_DRY_RUN)) {
+ if (!cmd.hasOption(OPTION_DRY_RUN)) {
if (cmd.hasOption(OPTION_REMOTE)) {
LOG.info("Running remotely...");
// should the topology be active or inactive
SubmitOptions submitOptions = null;
- if(cmd.hasOption(OPTION_INACTIVE)){
+ if (cmd.hasOption(OPTION_INACTIVE)) {
LOG.info("Deploying topology in an INACTIVE state...");
submitOptions = new SubmitOptions(TopologyInitialStatus.INACTIVE);
} else {
@@ -182,9 +187,9 @@ public class Flux {
}
}
- static void printTopologyInfo(ExecutionContext ctx){
+ static void printTopologyInfo(ExecutionContext ctx) {
TopologyDef t = ctx.getTopologyDef();
- if(t.isDslTopology()) {
+ if (t.isDslTopology()) {
print("---------- TOPOLOGY DETAILS ----------");
printf("Topology Name: %s", t.getName());
@@ -206,22 +211,22 @@ public class Flux {
}
// save a little typing
- private static void printf(String format, Object... args){
+ private static void printf(String format, Object... args) {
print(String.format(format, args));
}
- private static void print(String string){
+ private static void print(String string) {
System.out.println(string);
}
private static void printSplash() throws IOException {
// banner
InputStream is = Flux.class.getResourceAsStream("/splash.txt");
- if(is != null){
+ if (is != null) {
InputStreamReader isr = new InputStreamReader(is, "UTF-8");
BufferedReader br = new BufferedReader(isr);
String line = null;
- while((line = br.readLine()) != null){
+ while ((line = br.readLine()) != null) {
System.out.println(line);
}
}
http://git-wip-us.apache.org/repos/asf/storm/blob/cfb7d263/flux/flux-core/src/main/java/org/apache/storm/flux/FluxBuilder.java
----------------------------------------------------------------------
diff --git a/flux/flux-core/src/main/java/org/apache/storm/flux/FluxBuilder.java b/flux/flux-core/src/main/java/org/apache/storm/flux/FluxBuilder.java
index 1758a37..20bde18 100644
--- a/flux/flux-core/src/main/java/org/apache/storm/flux/FluxBuilder.java
+++ b/flux/flux-core/src/main/java/org/apache/storm/flux/FluxBuilder.java
@@ -15,29 +15,56 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.storm.flux;
+import java.lang.reflect.Array;
+import java.lang.reflect.Constructor;
+import java.lang.reflect.Field;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
import org.apache.storm.Config;
-import org.apache.storm.flux.model.*;
+import org.apache.storm.flux.model.BeanDef;
+import org.apache.storm.flux.model.BeanListReference;
+import org.apache.storm.flux.model.BeanReference;
+import org.apache.storm.flux.model.BoltDef;
+import org.apache.storm.flux.model.ConfigMethodDef;
+import org.apache.storm.flux.model.ExecutionContext;
+import org.apache.storm.flux.model.GroupingDef;
+import org.apache.storm.flux.model.ObjectDef;
+import org.apache.storm.flux.model.PropertyDef;
+import org.apache.storm.flux.model.SpoutDef;
+import org.apache.storm.flux.model.StreamDef;
+import org.apache.storm.flux.model.TopologyDef;
import org.apache.storm.generated.StormTopology;
import org.apache.storm.grouping.CustomStreamGrouping;
-import org.apache.storm.topology.*;
+import org.apache.storm.topology.BoltDeclarer;
+import org.apache.storm.topology.IBasicBolt;
+import org.apache.storm.topology.IRichBolt;
+import org.apache.storm.topology.IRichSpout;
+import org.apache.storm.topology.IStatefulBolt;
+import org.apache.storm.topology.IWindowedBolt;
+import org.apache.storm.topology.SpoutDeclarer;
+import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;
import org.apache.storm.utils.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.lang.reflect.*;
-import java.util.*;
-
public class FluxBuilder {
private static Logger LOG = LoggerFactory.getLogger(FluxBuilder.class);
+
/**
* Given a topology definition, return a populated `org.apache.storm.Config` instance.
- *
- * @param topologyDef
- * @return
+ * @param topologyDef topology definition
+ * @return a Storm Config object
*/
public static Config buildConfig(TopologyDef topologyDef) {
// merge contents of `config` into topology config
@@ -48,14 +75,14 @@ public class FluxBuilder {
/**
* Given a topology definition, return a Storm topology that can be run either locally or remotely.
- *
- * @param context
- * @return
- * @throws IllegalAccessException
- * @throws InstantiationException
- * @throws ClassNotFoundException
- * @throws NoSuchMethodException
- * @throws InvocationTargetException
+ * @param context execution context
+ * @return A runable Storm topology
+ * @throws IllegalAccessException if security policy disallows operation
+ * @throws InstantiationException if a class can't be instantiated
+ * @throws ClassNotFoundException if a class can't be found
+ * @throws NoSuchMethodException if a method can't be found
+ * @throws InvocationTargetException if method invocation fails
+ * @throws NoSuchFieldException if a referenced field does not exist
*/
public static StormTopology buildTopology(ExecutionContext context) throws IllegalAccessException,
InstantiationException, ClassNotFoundException, NoSuchMethodException, InvocationTargetException, NoSuchFieldException {
@@ -63,9 +90,9 @@ public class FluxBuilder {
StormTopology topology = null;
TopologyDef topologyDef = context.getTopologyDef();
- if(!topologyDef.validate()){
- throw new IllegalArgumentException("Invalid topology config. Spouts, bolts and streams cannot be " +
- "defined in the same configuration as a topologySource.");
+ if (!topologyDef.validate()) {
+ throw new IllegalArgumentException("Invalid topology config. Spouts, bolts and streams cannot be "
+ + "defined in the same configuration as a topologySource.");
}
// build components that may be referenced by spouts, bolts, etc.
@@ -73,7 +100,7 @@ public class FluxBuilder {
// constructed class instance
buildComponents(context);
- if(topologyDef.isDslTopology()) {
+ if (topologyDef.isDslTopology()) {
// This is a DSL (YAML, etc.) topology...
LOG.info("Detected DSL topology...");
@@ -105,33 +132,33 @@ public class FluxBuilder {
* parameter: `java.util.Map` or `org.apache.storm.Config`.
*
* @param topologySource object to inspect for the specified method
- * @param methodName name of the method to look for
- * @return
- * @throws NoSuchMethodException
+ * @param methodName name of the method to look for
+ * @return a Method object that returns a storm topology
+ * @throws NoSuchMethodException if no such method exists
*/
private static Method findGetTopologyMethod(Object topologySource, String methodName) throws NoSuchMethodException {
Class clazz = topologySource.getClass();
- Method[] methods = clazz.getMethods();
+ Method[] methods = clazz.getMethods();
ArrayList<Method> candidates = new ArrayList<Method>();
- for(Method method : methods){
- if(!method.getName().equals(methodName)){
+ for (Method method : methods) {
+ if (!method.getName().equals(methodName)) {
continue;
}
- if(!method.getReturnType().equals(StormTopology.class)){
+ if (!method.getReturnType().equals(StormTopology.class)) {
continue;
}
Class[] paramTypes = method.getParameterTypes();
- if(paramTypes.length != 1){
+ if (paramTypes.length != 1) {
continue;
}
- if(paramTypes[0].isAssignableFrom(Map.class) || paramTypes[0].isAssignableFrom(Config.class)){
+ if (paramTypes[0].isAssignableFrom(Map.class) || paramTypes[0].isAssignableFrom(Config.class)) {
candidates.add(method);
}
}
- if(candidates.size() == 0){
+ if (candidates.size() == 0) {
throw new IllegalArgumentException("Unable to find method '" + methodName + "' method in class: " + clazz.getName());
- } else if (candidates.size() > 1){
+ } else if (candidates.size() > 1) {
LOG.warn("Found multiple candidate methods in class '" + clazz.getName() + "'. Using the first one found");
}
@@ -139,8 +166,9 @@ public class FluxBuilder {
}
/**
- * @param context
- * @param builder
+ * Builds stream definitions.
+ * @param context context
+ * @param builder builder
*/
private static void buildStreamDefinitions(ExecutionContext context, TopologyBuilder builder)
throws ClassNotFoundException, NoSuchMethodException, InvocationTargetException, InstantiationException,
@@ -152,14 +180,14 @@ public class FluxBuilder {
Object boltObj = context.getBolt(stream.getTo());
BoltDeclarer declarer = declarers.get(stream.getTo());
if (boltObj instanceof IRichBolt) {
- if(declarer == null) {
+ if (declarer == null) {
declarer = builder.setBolt(stream.getTo(),
(IRichBolt) boltObj,
topologyDef.parallelismForBolt(stream.getTo()));
declarers.put(stream.getTo(), declarer);
}
} else if (boltObj instanceof IBasicBolt) {
- if(declarer == null) {
+ if (declarer == null) {
declarer = builder.setBolt(
stream.getTo(),
(IBasicBolt) boltObj,
@@ -167,7 +195,7 @@ public class FluxBuilder {
declarers.put(stream.getTo(), declarer);
}
} else if (boltObj instanceof IWindowedBolt) {
- if(declarer == null) {
+ if (declarer == null) {
declarer = builder.setBolt(
stream.getTo(),
(IWindowedBolt) boltObj,
@@ -175,7 +203,7 @@ public class FluxBuilder {
declarers.put(stream.getTo(), declarer);
}
} else if (boltObj instanceof IStatefulBolt) {
- if(declarer == null) {
+ if (declarer == null) {
declarer = builder.setBolt(
stream.getTo(),
(IStatefulBolt) boltObj,
@@ -183,8 +211,8 @@ public class FluxBuilder {
declarers.put(stream.getTo(), declarer);
}
} else {
- throw new IllegalArgumentException("Class does not appear to be a bolt: " +
- boltObj.getClass().getName());
+ throw new IllegalArgumentException("Class does not appear to be a bolt: "
+ + boltObj.getClass().getName());
}
BoltDef boltDef = topologyDef.getBoltDef(stream.getTo());
@@ -288,11 +316,11 @@ public class FluxBuilder {
private static List<Object> resolveReferences(List<Object> args, ExecutionContext context) {
LOG.debug("Checking arguments for references.");
- List<Object> cArgs = new ArrayList<Object>();
+ List<Object> constructorArgs = new ArrayList<Object>();
// resolve references
for (Object arg : args) {
if (arg instanceof BeanReference) {
- cArgs.add(context.getComponent(((BeanReference) arg).getId()));
+ constructorArgs.add(context.getComponent(((BeanReference) arg).getId()));
} else if (arg instanceof BeanListReference) {
List<Object> components = new ArrayList<>();
BeanListReference ref = (BeanListReference) arg;
@@ -301,12 +329,12 @@ public class FluxBuilder {
}
LOG.debug("BeanListReference resolved as {}", components);
- cArgs.add(components);
+ constructorArgs.add(components);
} else {
- cArgs.add(arg);
+ constructorArgs.add(arg);
}
}
- return cArgs;
+ return constructorArgs;
}
private static Object buildObject(ObjectDef def, ExecutionContext context) throws ClassNotFoundException,
@@ -315,20 +343,40 @@ public class FluxBuilder {
Object obj = null;
if (def.hasConstructorArgs()) {
LOG.debug("Found constructor arguments in definition: " + def.getConstructorArgs().getClass().getName());
- List<Object> cArgs = def.getConstructorArgs();
- if(def.hasReferences()){
- cArgs = resolveReferences(cArgs, context);
+ List<Object> constructorArgs = def.getConstructorArgs();
+ if (def.hasReferences()) {
+ constructorArgs = resolveReferences(constructorArgs, context);
}
- Constructor con = findCompatibleConstructor(cArgs, clazz);
+ Constructor con = findCompatibleConstructor(constructorArgs, clazz);
if (con != null) {
LOG.debug("Found something seemingly compatible, attempting invocation...");
- obj = con.newInstance(getArgsWithListCoercian(cArgs, con.getParameterTypes()));
+ obj = con.newInstance(getArgsWithListCoercian(constructorArgs, con.getParameterTypes()));
} else {
String msg = String.format("Couldn't find a suitable constructor for class '%s' with arguments '%s'.",
clazz.getName(),
- cArgs);
+ constructorArgs);
+ throw new IllegalArgumentException(msg);
+ }
+ } else if (def.hasFactory()) {
+ Method method = null;
+ List<Object> methodArgs = new ArrayList<>(); // empty if no factoryArgs
+ if (def.hasFactoryArgs()) {
+ methodArgs = def.getFactoryArgs();
+ if (def.hasReferences()) {
+ methodArgs = resolveReferences(methodArgs, context);
+ }
+ }
+ method = findCompatibleMethod(methodArgs, clazz, def.getFactory());
+ if (method != null) {
+ obj = method.invoke(null, getArgsWithListCoercian(methodArgs, method.getParameterTypes()));
+ } else {
+ String msg = String.format("Couldn't find a suitable static method '%s' for class '%s' with arguments '%s'.",
+ def.getFactory(),
+ clazz.getName(),
+ methodArgs);
throw new IllegalArgumentException(msg);
}
+
} else {
obj = clazz.newInstance();
}
@@ -345,7 +393,7 @@ public class FluxBuilder {
String methodName = context.getTopologyDef().getTopologySource().getMethodName();
Method getTopology = findGetTopologyMethod(topologySource, methodName);
- if(getTopology.getParameterTypes()[0].equals(Config.class)){
+ if (getTopology.getParameterTypes()[0].equals(Config.class)) {
Config config = new Config();
config.putAll(context.getTopologyDef().getConfig());
return (StormTopology) getTopology.invoke(topologySource, config);
@@ -358,7 +406,7 @@ public class FluxBuilder {
throws ClassNotFoundException,
IllegalAccessException, InstantiationException, NoSuchMethodException, InvocationTargetException, NoSuchFieldException {
Object grouping = buildObject(def, context);
- return (CustomStreamGrouping)grouping;
+ return (CustomStreamGrouping) grouping;
}
/**
@@ -367,9 +415,9 @@ public class FluxBuilder {
*/
private static void buildComponents(ExecutionContext context) throws ClassNotFoundException, NoSuchMethodException,
IllegalAccessException, InvocationTargetException, InstantiationException, NoSuchFieldException {
- Collection<BeanDef> cDefs = context.getTopologyDef().getComponents();
- if (cDefs != null) {
- for (BeanDef bean : cDefs) {
+ Collection<BeanDef> beanDefs = context.getTopologyDef().getComponents();
+ if (beanDefs != null) {
+ for (BeanDef bean : beanDefs) {
Object obj = buildObject(bean, context);
context.addComponent(bean.getId(), obj);
}
@@ -407,7 +455,7 @@ public class FluxBuilder {
*/
private static IRichSpout buildSpout(SpoutDef def, ExecutionContext context) throws ClassNotFoundException,
IllegalAccessException, InstantiationException, NoSuchMethodException, InvocationTargetException, NoSuchFieldException {
- return (IRichSpout)buildObject(def, context);
+ return (IRichSpout) buildObject(def, context);
}
/**
@@ -425,7 +473,6 @@ public class FluxBuilder {
/**
* Given a list of constructor arguments, and a target class, attempt to find a suitable constructor.
- *
*/
private static Constructor findCompatibleConstructor(List<Object> args, Class target) throws NoSuchMethodException {
Constructor retval = null;
@@ -456,25 +503,33 @@ public class FluxBuilder {
}
+ /**
+ * Invokes configuration methods on an class instance.
+ * @param bean the bean/component definition
+ * @param instance the class instance being operated on
+ * @param context execution context
+ * @throws InvocationTargetException if method invocation fails
+ * @throws IllegalAccessException if security policy prefents invocation
+ */
public static void invokeConfigMethods(ObjectDef bean, Object instance, ExecutionContext context)
throws InvocationTargetException, IllegalAccessException {
List<ConfigMethodDef> methodDefs = bean.getConfigMethods();
- if(methodDefs == null || methodDefs.size() == 0){
+ if (methodDefs == null || methodDefs.size() == 0) {
return;
}
Class clazz = instance.getClass();
- for(ConfigMethodDef methodDef : methodDefs){
+ for (ConfigMethodDef methodDef : methodDefs) {
List<Object> args = methodDef.getArgs();
- if (args == null){
+ if (args == null) {
args = new ArrayList();
}
- if(methodDef.hasReferences()){
+ if (methodDef.hasReferences()) {
args = resolveReferences(args, context);
}
String methodName = methodDef.getName();
Method method = findCompatibleMethod(args, clazz, methodName);
- if(method != null) {
+ if (method != null) {
Object[] methodArgs = getArgsWithListCoercian(args, method.getParameterTypes());
method.invoke(instance, methodArgs);
} else {
@@ -485,7 +540,7 @@ public class FluxBuilder {
}
}
- private static Method findCompatibleMethod(List<Object> args, Class target, String methodName){
+ private static Method findCompatibleMethod(List<Object> args, Class target, String methodName) {
Method retval = null;
int eligibleCount = 0;
@@ -497,7 +552,7 @@ public class FluxBuilder {
if (paramClasses.length == args.size() && method.getName().equals(methodName)) {
LOG.debug("found constructor with same number of args..");
boolean invokable = false;
- if (args.size() == 0){
+ if (args.size() == 0) {
// it's a method with zero args
invokable = true;
} else {
@@ -513,9 +568,9 @@ public class FluxBuilder {
}
}
if (eligibleCount > 1) {
- LOG.warn("Found multiple invokable methods for class {}, method {}, given arguments {}. " +
- "Using the last one found.",
- new Object[]{target, methodName, args});
+ LOG.warn("Found multiple invokable methods for class {}, method {}, given arguments {}. "
+ + "Using the last one found.",
+ new Object[]{target, methodName, args});
}
return retval;
}
@@ -526,7 +581,6 @@ public class FluxBuilder {
* to be coerced from a List to an Array, do so.
*/
private static Object[] getArgsWithListCoercian(List<Object> args, Class[] parameterTypes) {
-// Class[] parameterTypes = constructor.getParameterTypes();
if (parameterTypes.length != args.size()) {
throw new IllegalArgumentException("Contructor parameter count does not egual argument size.");
}
@@ -550,26 +604,27 @@ public class FluxBuilder {
constructorParams[i] = args.get(i);
continue;
}
- if (isPrimitiveBoolean(paramType) && Boolean.class.isAssignableFrom(objectType)){
+ if (isPrimitiveBoolean(paramType) && Boolean.class.isAssignableFrom(objectType)) {
LOG.debug("Its a primitive boolean.");
- Boolean bool = (Boolean)args.get(i);
+ Boolean bool = (Boolean) args.get(i);
constructorParams[i] = bool.booleanValue();
continue;
}
- if(isPrimitiveNumber(paramType) && Number.class.isAssignableFrom(objectType)){
- LOG.debug("Its a primitive number.");
- Number num = (Number)args.get(i);
- if(paramType == Float.TYPE){
+ if ((isPrimitiveNumber(paramType) || Number.class.isAssignableFrom(paramType))
+ && Number.class.isAssignableFrom(objectType)) {
+ LOG.debug("Its a number.");
+ Number num = (Number) args.get(i);
+ if (paramType == Float.TYPE || paramType == Float.class) {
constructorParams[i] = num.floatValue();
- } else if (paramType == Double.TYPE) {
+ } else if (paramType == Double.TYPE || paramType == Double.class) {
constructorParams[i] = num.doubleValue();
- } else if (paramType == Long.TYPE) {
+ } else if (paramType == Long.TYPE || paramType == Long.class) {
constructorParams[i] = num.longValue();
- } else if (paramType == Integer.TYPE) {
+ } else if (paramType == Integer.TYPE || paramType == Integer.class) {
constructorParams[i] = num.intValue();
- } else if (paramType == Short.TYPE) {
+ } else if (paramType == Short.TYPE || paramType == Short.class) {
constructorParams[i] = num.shortValue();
- } else if (paramType == Byte.TYPE) {
+ } else if (paramType == Byte.TYPE || paramType == Byte.class) {
constructorParams[i] = num.byteValue();
} else {
constructorParams[i] = args.get(i);
@@ -578,9 +633,9 @@ public class FluxBuilder {
}
// enum conversion
- if(paramType.isEnum() && objectType.equals(String.class)){
+ if (paramType.isEnum() && objectType.equals(String.class)) {
LOG.debug("Yes, will convert a String to enum");
- constructorParams[i] = Enum.valueOf(paramType, (String)args.get(i));
+ constructorParams[i] = Enum.valueOf(paramType, (String) args.get(i));
continue;
}
@@ -609,9 +664,9 @@ public class FluxBuilder {
* Determine if the given constructor/method parameter types are compatible given arguments List. Consider if
* list coercian can make it possible.
*
- * @param args
- * @param parameterTypes
- * @return
+ * @param args arguments
+ * @param parameterTypes parameter types
+ * @return true if parameter types and args are compatible
*/
private static boolean canInvokeWithArgs(List<Object> args, Class[] parameterTypes) {
if (parameterTypes.length != args.size()) {
@@ -632,11 +687,12 @@ public class FluxBuilder {
LOG.debug("Yes, they are the same class.");
} else if (paramType.isAssignableFrom(objectType)) {
LOG.debug("Yes, assignment is possible.");
- } else if (isPrimitiveBoolean(paramType) && Boolean.class.isAssignableFrom(objectType)){
+ } else if (isPrimitiveBoolean(paramType) && Boolean.class.isAssignableFrom(objectType)) {
LOG.debug("Yes, assignment is possible.");
- } else if(isPrimitiveNumber(paramType) && Number.class.isAssignableFrom(objectType)){
+ } else if (isPrimitiveNumber(paramType) || Number.class.isAssignableFrom(paramType)
+ && Number.class.isAssignableFrom(objectType)) {
LOG.debug("Yes, assignment is possible.");
- } else if(paramType.isEnum() && objectType.equals(String.class)){
+ } else if (paramType.isEnum() && objectType.equals(String.class)) {
LOG.debug("Yes, will convert a String to enum");
} else if (paramType.isArray() && List.class.isAssignableFrom(objectType)) {
// TODO more collection content type checking
@@ -649,11 +705,11 @@ public class FluxBuilder {
return true;
}
- public static boolean isPrimitiveNumber(Class clazz){
+ public static boolean isPrimitiveNumber(Class clazz) {
return clazz.isPrimitive() && !clazz.equals(boolean.class);
}
- public static boolean isPrimitiveBoolean(Class clazz){
+ public static boolean isPrimitiveBoolean(Class clazz) {
return clazz.isPrimitive() && clazz.equals(boolean.class);
}
}
http://git-wip-us.apache.org/repos/asf/storm/blob/cfb7d263/flux/flux-core/src/main/java/org/apache/storm/flux/api/TopologySource.java
----------------------------------------------------------------------
diff --git a/flux/flux-core/src/main/java/org/apache/storm/flux/api/TopologySource.java b/flux/flux-core/src/main/java/org/apache/storm/flux/api/TopologySource.java
index 2777854..1944c38 100644
--- a/flux/flux-core/src/main/java/org/apache/storm/flux/api/TopologySource.java
+++ b/flux/flux-core/src/main/java/org/apache/storm/flux/api/TopologySource.java
@@ -15,20 +15,20 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.storm.flux.api;
+import java.util.Map;
import org.apache.storm.generated.StormTopology;
-import java.util.Map;
-
/**
* Marker interface for objects that can produce `StormTopology` objects.
- *
+ * <p/>
* If a `topology-source` class implements the `getTopology()` method, Flux will
* call that method. Otherwise, it will introspect the given class and look for a
* similar method that produces a `StormTopology` instance.
- *
+ * <p/>
* Note that it is not strictly necessary for a class to implement this interface.
* If a class defines a method with a similar signature, Flux should be able to find
* and invoke it.
http://git-wip-us.apache.org/repos/asf/storm/blob/cfb7d263/flux/flux-core/src/main/java/org/apache/storm/flux/model/BeanDef.java
----------------------------------------------------------------------
diff --git a/flux/flux-core/src/main/java/org/apache/storm/flux/model/BeanDef.java b/flux/flux-core/src/main/java/org/apache/storm/flux/model/BeanDef.java
index f0247ed..199f4d2 100644
--- a/flux/flux-core/src/main/java/org/apache/storm/flux/model/BeanDef.java
+++ b/flux/flux-core/src/main/java/org/apache/storm/flux/model/BeanDef.java
@@ -15,6 +15,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.storm.flux.model;
/**
http://git-wip-us.apache.org/repos/asf/storm/blob/cfb7d263/flux/flux-core/src/main/java/org/apache/storm/flux/model/BeanListReference.java
----------------------------------------------------------------------
diff --git a/flux/flux-core/src/main/java/org/apache/storm/flux/model/BeanListReference.java b/flux/flux-core/src/main/java/org/apache/storm/flux/model/BeanListReference.java
index 652210c..8b9fa3e 100644
--- a/flux/flux-core/src/main/java/org/apache/storm/flux/model/BeanListReference.java
+++ b/flux/flux-core/src/main/java/org/apache/storm/flux/model/BeanListReference.java
@@ -15,6 +15,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.storm.flux.model;
import java.util.List;
@@ -25,9 +26,9 @@ import java.util.List;
public class BeanListReference {
public List<String> ids;
- public BeanListReference(){}
+ public BeanListReference() {}
- public BeanListReference(List<String> ids){
+ public BeanListReference(List<String> ids) {
this.ids = ids;
}
http://git-wip-us.apache.org/repos/asf/storm/blob/cfb7d263/flux/flux-core/src/main/java/org/apache/storm/flux/model/BeanReference.java
----------------------------------------------------------------------
diff --git a/flux/flux-core/src/main/java/org/apache/storm/flux/model/BeanReference.java b/flux/flux-core/src/main/java/org/apache/storm/flux/model/BeanReference.java
index bd236f1..f2f7cf9 100644
--- a/flux/flux-core/src/main/java/org/apache/storm/flux/model/BeanReference.java
+++ b/flux/flux-core/src/main/java/org/apache/storm/flux/model/BeanReference.java
@@ -15,6 +15,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.storm.flux.model;
/**
@@ -23,9 +24,9 @@ package org.apache.storm.flux.model;
public class BeanReference {
public String id;
- public BeanReference(){}
+ public BeanReference() {}
- public BeanReference(String id){
+ public BeanReference(String id) {
this.id = id;
}
http://git-wip-us.apache.org/repos/asf/storm/blob/cfb7d263/flux/flux-core/src/main/java/org/apache/storm/flux/model/BoltDef.java
----------------------------------------------------------------------
diff --git a/flux/flux-core/src/main/java/org/apache/storm/flux/model/BoltDef.java b/flux/flux-core/src/main/java/org/apache/storm/flux/model/BoltDef.java
index 362abf1..6969492 100644
--- a/flux/flux-core/src/main/java/org/apache/storm/flux/model/BoltDef.java
+++ b/flux/flux-core/src/main/java/org/apache/storm/flux/model/BoltDef.java
@@ -15,6 +15,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.storm.flux.model;
/**
http://git-wip-us.apache.org/repos/asf/storm/blob/cfb7d263/flux/flux-core/src/main/java/org/apache/storm/flux/model/ConfigMethodDef.java
----------------------------------------------------------------------
diff --git a/flux/flux-core/src/main/java/org/apache/storm/flux/model/ConfigMethodDef.java b/flux/flux-core/src/main/java/org/apache/storm/flux/model/ConfigMethodDef.java
index 69cabc3..d100598 100644
--- a/flux/flux-core/src/main/java/org/apache/storm/flux/model/ConfigMethodDef.java
+++ b/flux/flux-core/src/main/java/org/apache/storm/flux/model/ConfigMethodDef.java
@@ -15,6 +15,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.storm.flux.model;
import java.util.ArrayList;
@@ -39,13 +40,17 @@ public class ConfigMethodDef {
return args;
}
+ /**
+ * Set the method arguments.
+ * @param args method parameters
+ */
public void setArgs(List<Object> args) {
List<Object> newVal = new ArrayList<Object>();
- for(Object obj : args){
- if(obj instanceof LinkedHashMap){
+ for (Object obj : args) {
+ if (obj instanceof LinkedHashMap) {
Map map = (Map)obj;
- if(map.containsKey("ref") && map.size() == 1){
+ if (map.containsKey("ref") && map.size() == 1) {
newVal.add(new BeanReference((String)map.get("ref")));
this.hasReferences = true;
} else if (map.containsKey("reflist") && map.size() == 1) {
@@ -61,7 +66,7 @@ public class ConfigMethodDef {
this.args = newVal;
}
- public boolean hasReferences(){
+ public boolean hasReferences() {
return this.hasReferences;
}
}
http://git-wip-us.apache.org/repos/asf/storm/blob/cfb7d263/flux/flux-core/src/main/java/org/apache/storm/flux/model/ExecutionContext.java
----------------------------------------------------------------------
diff --git a/flux/flux-core/src/main/java/org/apache/storm/flux/model/ExecutionContext.java b/flux/flux-core/src/main/java/org/apache/storm/flux/model/ExecutionContext.java
index 1520006..ec0529e 100644
--- a/flux/flux-core/src/main/java/org/apache/storm/flux/model/ExecutionContext.java
+++ b/flux/flux-core/src/main/java/org/apache/storm/flux/model/ExecutionContext.java
@@ -15,16 +15,17 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.storm.flux.model;
-import org.apache.storm.Config;
-import org.apache.storm.task.IBolt;
-import org.apache.storm.topology.IRichSpout;
+package org.apache.storm.flux.model;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import org.apache.storm.Config;
+import org.apache.storm.task.IBolt;
+import org.apache.storm.topology.IRichSpout;
+
/**
* Container for all the objects required to instantiate a topology.
*/
@@ -45,32 +46,32 @@ public class ExecutionContext {
private List<IBolt> bolts;
private Map<String, Object> boltMap = new HashMap<String, Object>();
- public ExecutionContext(TopologyDef topologyDef, Config config){
+ public ExecutionContext(TopologyDef topologyDef, Config config) {
this.topologyDef = topologyDef;
this.config = config;
}
- public TopologyDef getTopologyDef(){
+ public TopologyDef getTopologyDef() {
return this.topologyDef;
}
- public void addSpout(String id, IRichSpout spout){
+ public void addSpout(String id, IRichSpout spout) {
this.spoutMap.put(id, spout);
}
- public void addBolt(String id, Object bolt){
+ public void addBolt(String id, Object bolt) {
this.boltMap.put(id, bolt);
}
- public Object getBolt(String id){
+ public Object getBolt(String id) {
return this.boltMap.get(id);
}
- public void addComponent(String id, Object value){
+ public void addComponent(String id, Object value) {
this.componentMap.put(id, value);
}
- public Object getComponent(String id){
+ public Object getComponent(String id) {
return this.componentMap.get(id);
}
http://git-wip-us.apache.org/repos/asf/storm/blob/cfb7d263/flux/flux-core/src/main/java/org/apache/storm/flux/model/GroupingDef.java
----------------------------------------------------------------------
diff --git a/flux/flux-core/src/main/java/org/apache/storm/flux/model/GroupingDef.java b/flux/flux-core/src/main/java/org/apache/storm/flux/model/GroupingDef.java
index e4fac8e..f191e3c 100644
--- a/flux/flux-core/src/main/java/org/apache/storm/flux/model/GroupingDef.java
+++ b/flux/flux-core/src/main/java/org/apache/storm/flux/model/GroupingDef.java
@@ -15,6 +15,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.storm.flux.model;
import java.util.List;
@@ -25,7 +26,7 @@ import java.util.List;
public class GroupingDef {
/**
- * Types of stream groupings Storm allows
+ * Types of stream groupings Storm allows.
*/
public static enum Type {
ALL,
http://git-wip-us.apache.org/repos/asf/storm/blob/cfb7d263/flux/flux-core/src/main/java/org/apache/storm/flux/model/IncludeDef.java
----------------------------------------------------------------------
diff --git a/flux/flux-core/src/main/java/org/apache/storm/flux/model/IncludeDef.java b/flux/flux-core/src/main/java/org/apache/storm/flux/model/IncludeDef.java
index 23fd9d2..52e9ac3 100644
--- a/flux/flux-core/src/main/java/org/apache/storm/flux/model/IncludeDef.java
+++ b/flux/flux-core/src/main/java/org/apache/storm/flux/model/IncludeDef.java
@@ -15,11 +15,12 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.storm.flux.model;
/**
* Represents an include. Includes can be either a file or a classpath resource.
- *
+ *<p/>
* If an include is marked as `override=true` then existing properties will be replaced.
*
*/
http://git-wip-us.apache.org/repos/asf/storm/blob/cfb7d263/flux/flux-core/src/main/java/org/apache/storm/flux/model/ObjectDef.java
----------------------------------------------------------------------
diff --git a/flux/flux-core/src/main/java/org/apache/storm/flux/model/ObjectDef.java b/flux/flux-core/src/main/java/org/apache/storm/flux/model/ObjectDef.java
index 04a7e8a..170ee4f 100644
--- a/flux/flux-core/src/main/java/org/apache/storm/flux/model/ObjectDef.java
+++ b/flux/flux-core/src/main/java/org/apache/storm/flux/model/ObjectDef.java
@@ -15,9 +15,8 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.storm.flux.model;
-import org.apache.storm.Config;
+package org.apache.storm.flux.model;
import java.util.ArrayList;
import java.util.LinkedHashMap;
@@ -34,6 +33,8 @@ public class ObjectDef {
private boolean hasReferences;
private List<PropertyDef> properties;
private List<ConfigMethodDef> configMethods;
+ private String factory;
+ private List<Object> factoryArgs;
public String getClassName() {
return className;
@@ -47,13 +48,17 @@ public class ObjectDef {
return constructorArgs;
}
+ /**
+ * Sets the arguments for the constructor and checks for references.
+ * @param constructorArgs Constructor arguments
+ */
public void setConstructorArgs(List<Object> constructorArgs) {
List<Object> newVal = new ArrayList<Object>();
- for(Object obj : constructorArgs){
- if(obj instanceof LinkedHashMap){
+ for (Object obj : constructorArgs) {
+ if (obj instanceof LinkedHashMap) {
Map map = (Map)obj;
- if(map.containsKey("ref") && map.size() == 1) {
+ if (map.containsKey("ref") && map.size() == 1) {
newVal.add(new BeanReference((String) map.get("ref")));
this.hasReferences = true;
} else if (map.containsKey("reflist") && map.size() == 1) {
@@ -69,11 +74,11 @@ public class ObjectDef {
this.constructorArgs = newVal;
}
- public boolean hasConstructorArgs(){
+ public boolean hasConstructorArgs() {
return this.constructorArgs != null && this.constructorArgs.size() > 0;
}
- public boolean hasReferences(){
+ public boolean hasReferences() {
return this.hasReferences;
}
@@ -92,4 +97,49 @@ public class ObjectDef {
public void setConfigMethods(List<ConfigMethodDef> configMethods) {
this.configMethods = configMethods;
}
+
+ public boolean hasFactory() {
+ return this.factory != null && !this.factory.isEmpty();
+ }
+
+ public boolean hasFactoryArgs() {
+ return this.factoryArgs != null && this.factoryArgs.size() > 0;
+ }
+
+ public String getFactory() {
+ return this.factory;
+ }
+
+ public void setFactory(String factory) {
+ this.factory = factory;
+ }
+
+ public List<Object> getFactoryArgs() {
+ return this.factoryArgs;
+ }
+
+ /**
+ * Sets factory method arguments and checks for references.
+ * @param factoryArgs factory method arguments
+ */
+ public void setFactoryArgs(List<Object> factoryArgs) {
+ List<Object> newVal = new ArrayList<Object>();
+ for (Object obj : factoryArgs) {
+ if (obj instanceof LinkedHashMap) {
+ Map map = (Map)obj;
+ if (map.containsKey("ref") && map.size() == 1) {
+ newVal.add(new BeanReference((String) map.get("ref")));
+ this.hasReferences = true;
+ } else if (map.containsKey("reflist") && map.size() == 1) {
+ newVal.add(new BeanListReference((List<String>) map.get("reflist")));
+ this.hasReferences = true;
+ } else {
+ newVal.add(obj);
+ }
+ } else {
+ newVal.add(obj);
+ }
+ }
+ this.factoryArgs = newVal;
+ }
}
http://git-wip-us.apache.org/repos/asf/storm/blob/cfb7d263/flux/flux-core/src/main/java/org/apache/storm/flux/model/PropertyDef.java
----------------------------------------------------------------------
diff --git a/flux/flux-core/src/main/java/org/apache/storm/flux/model/PropertyDef.java b/flux/flux-core/src/main/java/org/apache/storm/flux/model/PropertyDef.java
index f3d7704..3debba9 100644
--- a/flux/flux-core/src/main/java/org/apache/storm/flux/model/PropertyDef.java
+++ b/flux/flux-core/src/main/java/org/apache/storm/flux/model/PropertyDef.java
@@ -15,6 +15,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.storm.flux.model;
public class PropertyDef {
@@ -34,8 +35,13 @@ public class PropertyDef {
return value;
}
+ /**
+ * Sets the value of this property. Throws IllegalArgumentException if a reference has
+ * already been set.
+ * @param value property value
+ */
public void setValue(Object value) {
- if(this.ref != null){
+ if (this.ref != null) {
throw new IllegalStateException("A property can only have a value OR a reference, not both.");
}
this.value = value;
@@ -45,14 +51,19 @@ public class PropertyDef {
return ref;
}
+ /**
+ * Sets the value of this property to a reference. Throws IllegalArgumentException if a value has
+ * already been set.
+ * @param ref property reference
+ */
public void setRef(String ref) {
- if(this.value != null){
+ if (this.value != null) {
throw new IllegalStateException("A property can only have a value OR a reference, not both.");
}
this.ref = ref;
}
- public boolean isReference(){
+ public boolean isReference() {
return this.ref != null;
}
}
http://git-wip-us.apache.org/repos/asf/storm/blob/cfb7d263/flux/flux-core/src/main/java/org/apache/storm/flux/model/SpoutDef.java
----------------------------------------------------------------------
diff --git a/flux/flux-core/src/main/java/org/apache/storm/flux/model/SpoutDef.java b/flux/flux-core/src/main/java/org/apache/storm/flux/model/SpoutDef.java
index 277c601..df17b2a 100644
--- a/flux/flux-core/src/main/java/org/apache/storm/flux/model/SpoutDef.java
+++ b/flux/flux-core/src/main/java/org/apache/storm/flux/model/SpoutDef.java
@@ -15,6 +15,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.storm.flux.model;
/**
http://git-wip-us.apache.org/repos/asf/storm/blob/cfb7d263/flux/flux-core/src/main/java/org/apache/storm/flux/model/StreamDef.java
----------------------------------------------------------------------
diff --git a/flux/flux-core/src/main/java/org/apache/storm/flux/model/StreamDef.java b/flux/flux-core/src/main/java/org/apache/storm/flux/model/StreamDef.java
index da80f1c..2a1b87c 100644
--- a/flux/flux-core/src/main/java/org/apache/storm/flux/model/StreamDef.java
+++ b/flux/flux-core/src/main/java/org/apache/storm/flux/model/StreamDef.java
@@ -15,11 +15,12 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.storm.flux.model;
/**
* Represents a stream of tuples from one Storm component (Spout or Bolt) to another (an edge in the topology DAG).
- *
+ * <p/>
* Required fields are `from` and `to`, which define the source and destination, and the stream `grouping`.
*
*/
http://git-wip-us.apache.org/repos/asf/storm/blob/cfb7d263/flux/flux-core/src/main/java/org/apache/storm/flux/model/TopologyDef.java
----------------------------------------------------------------------
diff --git a/flux/flux-core/src/main/java/org/apache/storm/flux/model/TopologyDef.java b/flux/flux-core/src/main/java/org/apache/storm/flux/model/TopologyDef.java
index 86614f1..a0b4a03 100644
--- a/flux/flux-core/src/main/java/org/apache/storm/flux/model/TopologyDef.java
+++ b/flux/flux-core/src/main/java/org/apache/storm/flux/model/TopologyDef.java
@@ -15,16 +15,21 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.storm.flux.model;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.*;
-
/**
* Bean represenation of a topology.
- *
+ * <p/>
* It consists of the following:
* 1. The topology name
* 2. A `java.util.Map` representing the `org.apache.storm.config` for the topology
@@ -58,36 +63,57 @@ public class TopologyDef {
this.name = name;
}
- public void setName(String name, boolean override){
- if(this.name == null || override){
+ /**
+ * Sets the name of the topology.
+ * @param name topology name
+ * @param override whether to override if already set
+ */
+ public void setName(String name, boolean override) {
+ if (this.name == null || override) {
this.name = name;
} else {
LOG.warn("Ignoring attempt to set property 'name' with override == false.");
}
}
+ /**
+ * Returns all spout definitions.
+ * @return spout definitions.
+ */
public List<SpoutDef> getSpouts() {
ArrayList<SpoutDef> retval = new ArrayList<SpoutDef>();
retval.addAll(this.spoutMap.values());
return retval;
}
+ /**
+ * Set spout definitions.
+ * @param spouts spout definitions
+ */
public void setSpouts(List<SpoutDef> spouts) {
this.spoutMap = new LinkedHashMap<String, SpoutDef>();
- for(SpoutDef spout : spouts){
+ for (SpoutDef spout : spouts) {
this.spoutMap.put(spout.getId(), spout);
}
}
+ /**
+ * Returns bolt definitions.
+ * @return bolt definitions
+ */
public List<BoltDef> getBolts() {
ArrayList<BoltDef> retval = new ArrayList<BoltDef>();
retval.addAll(this.boltMap.values());
return retval;
}
+ /**
+ * Sets bolt definitions.
+ * @param bolts bolt definitions
+ */
public void setBolts(List<BoltDef> bolts) {
- this.boltMap = new LinkedHashMap<String, BoltDef>();
- for(BoltDef bolt : bolts){
+ this.boltMap = new LinkedHashMap<>();
+ for (BoltDef bolt : bolts) {
this.boltMap.put(bolt.getId(), bolt);
}
}
@@ -108,15 +134,23 @@ public class TopologyDef {
this.config = config;
}
+ /**
+ * Returns a list of all component definitions.
+ * @return components
+ */
public List<BeanDef> getComponents() {
ArrayList<BeanDef> retval = new ArrayList<BeanDef>();
retval.addAll(this.componentMap.values());
return retval;
}
+ /**
+ * Sets the list of component definitions.
+ * @param components components definitions
+ */
public void setComponents(List<BeanDef> components) {
- this.componentMap = new LinkedHashMap<String, BeanDef>();
- for(BeanDef component : components){
+ this.componentMap = new LinkedHashMap<>();
+ for (BeanDef component : components) {
this.componentMap.put(component.getId(), component);
}
}
@@ -130,27 +164,32 @@ public class TopologyDef {
}
// utility methods
- public int parallelismForBolt(String boltId){
+ public int parallelismForBolt(String boltId) {
return this.boltMap.get(boltId).getParallelism();
}
- public BoltDef getBoltDef(String id){
+ public BoltDef getBoltDef(String id) {
return this.boltMap.get(id);
}
- public SpoutDef getSpoutDef(String id){
+ public SpoutDef getSpoutDef(String id) {
return this.spoutMap.get(id);
}
- public BeanDef getComponent(String id){
+ public BeanDef getComponent(String id) {
return this.componentMap.get(id);
}
- // used by includes implementation
- public void addAllBolts(List<BoltDef> bolts, boolean override){
- for(BoltDef bolt : bolts){
+ /**
+ * Adds a list of bolt definitions. Optionally overriding existing definitions
+ * if one with the same ID already exists.
+ * @param bolts bolt definitions
+ * @param override whether or not to override existing definitions
+ */
+ public void addAllBolts(List<BoltDef> bolts, boolean override) {
+ for (BoltDef bolt : bolts) {
String id = bolt.getId();
- if(this.boltMap.get(id) == null || override) {
+ if (this.boltMap.get(id) == null || override) {
this.boltMap.put(bolt.getId(), bolt);
} else {
LOG.warn("Ignoring attempt to create bolt '{}' with override == false.", id);
@@ -158,10 +197,16 @@ public class TopologyDef {
}
}
- public void addAllSpouts(List<SpoutDef> spouts, boolean override){
- for(SpoutDef spout : spouts){
+ /**
+ * Adds a list of spout definitions. Optionally overriding existing definitions
+ * if one with the same ID already exists.
+ * @param spouts spout definitions
+ * @param override whether or not to override existing definitions
+ */
+ public void addAllSpouts(List<SpoutDef> spouts, boolean override) {
+ for (SpoutDef spout : spouts) {
String id = spout.getId();
- if(this.spoutMap.get(id) == null || override) {
+ if (this.spoutMap.get(id) == null || override) {
this.spoutMap.put(spout.getId(), spout);
} else {
LOG.warn("Ignoring attempt to create spout '{}' with override == false.", id);
@@ -169,10 +214,16 @@ public class TopologyDef {
}
}
+ /**
+ * Adds a list of component definitions. Optionally overriding existing definitions
+ * if one with the same ID already exists.
+ * @param components component definitions
+ * @param override whether or not to override existing definitions
+ */
public void addAllComponents(List<BeanDef> components, boolean override) {
- for(BeanDef bean : components){
+ for (BeanDef bean : components) {
String id = bean.getId();
- if(this.componentMap.get(id) == null || override) {
+ if (this.componentMap.get(id) == null || override) {
this.componentMap.put(bean.getId(), bean);
} else {
LOG.warn("Ignoring attempt to create component '{}' with override == false.", id);
@@ -180,6 +231,12 @@ public class TopologyDef {
}
}
+ /**
+ * Adds a list of stream definitions. Optionally overriding existing definitions
+ * if one with the same ID already exists.
+ * @param streams stream definitions
+ * @param override whether or not to override existing definitions (currently ignored)
+ */
public void addAllStreams(List<StreamDef> streams, boolean override) {
//TODO figure out how we want to deal with overrides. Users may want to add streams even when overriding other
// properties. For now we just add them blindly which could lead to a potentially invalid topology.
@@ -194,21 +251,24 @@ public class TopologyDef {
this.topologySource = topologySource;
}
- public boolean isDslTopology(){
+ public boolean isDslTopology() {
return this.topologySource == null;
}
- public boolean validate(){
+ /**
+ * Determines is this represents a valid Topology.
+ * @return true if valid
+ */
+ public boolean validate() {
boolean hasSpouts = this.spoutMap != null && this.spoutMap.size() > 0;
boolean hasBolts = this.boltMap != null && this.boltMap.size() > 0;
boolean hasStreams = this.streams != null && this.streams.size() > 0;
- boolean hasSpoutsBoltsStreams = hasStreams && hasBolts && hasSpouts;
// you cant define a topologySource and a DSL topology at the same time...
if (!isDslTopology() && ((hasSpouts || hasBolts || hasStreams))) {
return false;
}
- if(isDslTopology() && (hasSpouts && hasBolts && hasStreams)) {
+ if (isDslTopology() && (hasSpouts && hasBolts && hasStreams)) {
return true;
}
return true;
http://git-wip-us.apache.org/repos/asf/storm/blob/cfb7d263/flux/flux-core/src/main/java/org/apache/storm/flux/model/TopologySourceDef.java
----------------------------------------------------------------------
diff --git a/flux/flux-core/src/main/java/org/apache/storm/flux/model/TopologySourceDef.java b/flux/flux-core/src/main/java/org/apache/storm/flux/model/TopologySourceDef.java
index d6a2f57..17904a0 100644
--- a/flux/flux-core/src/main/java/org/apache/storm/flux/model/TopologySourceDef.java
+++ b/flux/flux-core/src/main/java/org/apache/storm/flux/model/TopologySourceDef.java
@@ -15,6 +15,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.storm.flux.model;
public class TopologySourceDef extends ObjectDef {
@@ -22,7 +23,7 @@ public class TopologySourceDef extends ObjectDef {
private String methodName;
- public TopologySourceDef(){
+ public TopologySourceDef() {
this.methodName = DEFAULT_METHOD_NAME;
}
http://git-wip-us.apache.org/repos/asf/storm/blob/cfb7d263/flux/flux-core/src/main/java/org/apache/storm/flux/model/VertexDef.java
----------------------------------------------------------------------
diff --git a/flux/flux-core/src/main/java/org/apache/storm/flux/model/VertexDef.java b/flux/flux-core/src/main/java/org/apache/storm/flux/model/VertexDef.java
index 8651819..cd09d05 100644
--- a/flux/flux-core/src/main/java/org/apache/storm/flux/model/VertexDef.java
+++ b/flux/flux-core/src/main/java/org/apache/storm/flux/model/VertexDef.java
@@ -15,10 +15,11 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.storm.flux.model;
/**
- * Abstract parent class of component definitions
+ * Abstract parent class of component definitions.
* (spouts/bolts)
*/
public abstract class VertexDef extends BeanDef {
http://git-wip-us.apache.org/repos/asf/storm/blob/cfb7d263/flux/flux-core/src/main/java/org/apache/storm/flux/parser/FluxParser.java
----------------------------------------------------------------------
diff --git a/flux/flux-core/src/main/java/org/apache/storm/flux/parser/FluxParser.java b/flux/flux-core/src/main/java/org/apache/storm/flux/parser/FluxParser.java
index 1bb018a..cc23fb8 100644
--- a/flux/flux-core/src/main/java/org/apache/storm/flux/parser/FluxParser.java
+++ b/flux/flux-core/src/main/java/org/apache/storm/flux/parser/FluxParser.java
@@ -15,8 +15,16 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.storm.flux.parser;
+import java.io.ByteArrayOutputStream;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Map;
+import java.util.Properties;
+
import org.apache.storm.flux.model.BoltDef;
import org.apache.storm.flux.model.IncludeDef;
import org.apache.storm.flux.model.SpoutDef;
@@ -27,74 +35,102 @@ import org.yaml.snakeyaml.TypeDescription;
import org.yaml.snakeyaml.Yaml;
import org.yaml.snakeyaml.constructor.Constructor;
-import java.io.ByteArrayOutputStream;
-import java.io.FileInputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.util.Map;
-import java.util.Properties;
-
+/**
+ * Static utility methods for parsing flux YAML.
+ */
public class FluxParser {
private static final Logger LOG = LoggerFactory.getLogger(FluxParser.class);
- private FluxParser(){}
+ private FluxParser() {
+ }
// TODO refactor input stream processing (see parseResource() method).
+
+ /**
+ * Parse a flux topology definition.
+ * @param inputFile source YAML file
+ * @param dumpYaml if true, dump the parsed YAML to stdout
+ * @param processIncludes whether or not to process includes
+ * @param propertiesFile properties file for variable substitution
+ * @param envSub whether or not to perform environment variable substitution
+ * @return resulting topologuy definition
+ * @throws IOException if there is a problem reading file(s)
+ */
public static TopologyDef parseFile(String inputFile, boolean dumpYaml, boolean processIncludes,
- String propertiesFile, boolean envSub) throws IOException {
-
+ String propertiesFile, boolean envSub) throws IOException {
+
FileInputStream in = new FileInputStream(inputFile);
TopologyDef topology = parseInputStream(in, dumpYaml, processIncludes, propertiesFile, envSub);
in.close();
-
+
return topology;
}
+ /**
+ * Parse a flux topology definition from a classpath resource..
+ * @param resource YAML resource
+ * @param dumpYaml if true, dump the parsed YAML to stdout
+ * @param processIncludes whether or not to process includes
+ * @param propertiesFile properties file for variable substitution
+ * @param envSub whether or not to perform environment variable substitution
+ * @return resulting topologuy definition
+ * @throws IOException if there is a problem reading file(s)
+ */
public static TopologyDef parseResource(String resource, boolean dumpYaml, boolean processIncludes,
- String propertiesFile, boolean envSub) throws IOException {
-
+ String propertiesFile, boolean envSub) throws IOException {
+
InputStream in = FluxParser.class.getResourceAsStream(resource);
TopologyDef topology = parseInputStream(in, dumpYaml, processIncludes, propertiesFile, envSub);
in.close();
-
+
return topology;
}
-
+
+ /**
+ * Parse a flux topology definition.
+ * @param inputStream InputStream representation of YAML file
+ * @param dumpYaml if true, dump the parsed YAML to stdout
+ * @param processIncludes whether or not to process includes
+ * @param propertiesFile properties file for variable substitution
+ * @param envSub whether or not to perform environment variable substitution
+ * @return resulting topologuy definition
+ * @throws IOException if there is a problem reading file(s)
+ */
public static TopologyDef parseInputStream(InputStream inputStream, boolean dumpYaml, boolean processIncludes,
- String propertiesFile, boolean envSub) throws IOException {
-
- Yaml yaml = yaml();
-
- if (inputStream == null) {
- LOG.error("Unable to load input stream");
- System.exit(1);
- }
-
- TopologyDef topology = loadYaml(yaml, inputStream, propertiesFile, envSub);
-
- if (dumpYaml) {
- dumpYaml(topology, yaml);
- }
-
- if (processIncludes) {
- return processIncludes(yaml, topology, propertiesFile, envSub);
- } else {
- return topology;
- }
+ String propertiesFile, boolean envSub) throws IOException {
+
+ Yaml yaml = yaml();
+
+ if (inputStream == null) {
+ LOG.error("Unable to load input stream");
+ System.exit(1);
+ }
+
+ TopologyDef topology = loadYaml(yaml, inputStream, propertiesFile, envSub);
+
+ if (dumpYaml) {
+ dumpYaml(topology, yaml);
+ }
+
+ if (processIncludes) {
+ return processIncludes(yaml, topology, propertiesFile, envSub);
+ } else {
+ return topology;
+ }
}
private static TopologyDef loadYaml(Yaml yaml, InputStream in, String propsFile, boolean envSubstitution) throws IOException {
ByteArrayOutputStream bos = new ByteArrayOutputStream();
LOG.info("loading YAML from input stream...");
int b = -1;
- while((b = in.read()) != -1){
+ while ((b = in.read()) != -1) {
bos.write(b);
}
// TODO substitution implementation is not exactly efficient or kind to memory...
String str = bos.toString();
// properties file substitution
- if(propsFile != null){
+ if (propsFile != null) {
LOG.info("Performing property substitution.");
try (InputStream propsIn = new FileInputStream(propsFile)) {
Properties props = new Properties();
@@ -108,46 +144,46 @@ public class FluxParser {
}
// environment variable substitution
- if(envSubstitution){
+ if (envSubstitution) {
LOG.info("Performing environment variable substitution...");
Map<String, String> envs = System.getenv();
- for(String key : envs.keySet()){
+ for (String key : envs.keySet()) {
str = str.replace("${ENV-" + key + "}", envs.get(key));
}
} else {
LOG.info("Not performing environment variable substitution.");
}
- return (TopologyDef)yaml.load(str);
+ return (TopologyDef) yaml.load(str);
}
- private static void dumpYaml(TopologyDef topology, Yaml yaml){
+ private static void dumpYaml(TopologyDef topology, Yaml yaml) {
System.out.println("Configuration (interpreted): \n" + yaml.dump(topology));
}
- private static Yaml yaml(){
- Constructor constructor = new Constructor(TopologyDef.class);
-
+ private static Yaml yaml() {
TypeDescription topologyDescription = new TypeDescription(TopologyDef.class);
topologyDescription.putListPropertyType("spouts", SpoutDef.class);
topologyDescription.putListPropertyType("bolts", BoltDef.class);
topologyDescription.putListPropertyType("includes", IncludeDef.class);
+
+ Constructor constructor = new Constructor(TopologyDef.class);
constructor.addTypeDescription(topologyDescription);
- Yaml yaml = new Yaml(constructor);
+ Yaml yaml = new Yaml(constructor);
return yaml;
}
/**
- *
- * @param yaml the yaml parser for parsing the include file(s)
+ * Process includes contained within a yaml file.
+ * @param yaml the yaml parser for parsing the include file(s)
* @param topologyDef the topology definition containing (possibly zero) includes
* @return The TopologyDef with includes resolved.
*/
private static TopologyDef processIncludes(Yaml yaml, TopologyDef topologyDef, String propsFile, boolean envSub)
throws IOException {
//TODO support multiple levels of includes
- if(topologyDef.getIncludes() != null) {
- for (IncludeDef include : topologyDef.getIncludes()){
+ if (topologyDef.getIncludes() != null) {
+ for (IncludeDef include : topologyDef.getIncludes()) {
TopologyDef includeTopologyDef = null;
if (include.isResource()) {
LOG.info("Loading includes from resource: {}", include.getFile());
@@ -160,23 +196,22 @@ public class FluxParser {
// if overrides are disabled, we won't replace anything that already exists
boolean override = include.isOverride();
// name
- if(includeTopologyDef.getName() != null){
+ if (includeTopologyDef.getName() != null) {
topologyDef.setName(includeTopologyDef.getName(), override);
}
// config
- if(includeTopologyDef.getConfig() != null) {
+ if (includeTopologyDef.getConfig() != null) {
//TODO move this logic to the model class
Map<String, Object> config = topologyDef.getConfig();
Map<String, Object> includeConfig = includeTopologyDef.getConfig();
- if(override) {
+ if (override) {
config.putAll(includeTopologyDef.getConfig());
} else {
- for(String key : includeConfig.keySet()){
- if(config.containsKey(key)){
+ for (String key : includeConfig.keySet()) {
+ if (config.containsKey(key)) {
LOG.warn("Ignoring attempt to set topology config property '{}' with override == false", key);
- }
- else {
+ } else {
config.put(key, includeConfig.get(key));
}
}
@@ -184,20 +219,20 @@ public class FluxParser {
}
//component overrides
- if(includeTopologyDef.getComponents() != null){
+ if (includeTopologyDef.getComponents() != null) {
topologyDef.addAllComponents(includeTopologyDef.getComponents(), override);
}
//bolt overrides
- if(includeTopologyDef.getBolts() != null){
+ if (includeTopologyDef.getBolts() != null) {
topologyDef.addAllBolts(includeTopologyDef.getBolts(), override);
}
//spout overrides
- if(includeTopologyDef.getSpouts() != null) {
+ if (includeTopologyDef.getSpouts() != null) {
topologyDef.addAllSpouts(includeTopologyDef.getSpouts(), override);
}
//stream overrides
//TODO streams should be uniquely identifiable
- if(includeTopologyDef.getStreams() != null) {
+ if (includeTopologyDef.getStreams() != null) {
topologyDef.addAllStreams(includeTopologyDef.getStreams(), override);
}
} // end include processing
http://git-wip-us.apache.org/repos/asf/storm/blob/cfb7d263/flux/flux-core/src/test/java/org/apache/storm/flux/TCKTest.java
----------------------------------------------------------------------
diff --git a/flux/flux-core/src/test/java/org/apache/storm/flux/TCKTest.java b/flux/flux-core/src/test/java/org/apache/storm/flux/TCKTest.java
index da38969..3cc82fe 100644
--- a/flux/flux-core/src/test/java/org/apache/storm/flux/TCKTest.java
+++ b/flux/flux-core/src/test/java/org/apache/storm/flux/TCKTest.java
@@ -229,6 +229,10 @@ public class TCKTest {
assertTrue(bolt.getFoo().equals("foo"));
assertTrue(bolt.getBar().equals("bar"));
assertTrue(bolt.getFooBar().equals("foobar"));
+
+ assertNotNull(context.getBolt("bolt-2"));
+ assertNotNull(context.getBolt("bolt-3"));
+ assertNotNull(context.getBolt("bolt-4"));
assertArrayEquals(new TestBolt.TestClass[] {new TestBolt.TestClass("foo"), new TestBolt.TestClass("bar"), new TestBolt.TestClass("baz")}, bolt.getClasses());
}
http://git-wip-us.apache.org/repos/asf/storm/blob/cfb7d263/flux/flux-core/src/test/java/org/apache/storm/flux/test/TestBolt.java
----------------------------------------------------------------------
diff --git a/flux/flux-core/src/test/java/org/apache/storm/flux/test/TestBolt.java b/flux/flux-core/src/test/java/org/apache/storm/flux/test/TestBolt.java
index 28d11b6..373553f 100644
--- a/flux/flux-core/src/test/java/org/apache/storm/flux/test/TestBolt.java
+++ b/flux/flux-core/src/test/java/org/apache/storm/flux/test/TestBolt.java
@@ -25,6 +25,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.Serializable;
+import java.time.Duration;
public class TestBolt extends BaseBasicBolt {
@@ -85,6 +86,8 @@ public class TestBolt extends BaseBasicBolt {
}
+ public TestBolt(Long l){}
+
@Override
public void execute(Tuple tuple, BasicOutputCollector basicOutputCollector) {
LOG.info("{}", tuple);
@@ -114,6 +117,14 @@ public class TestBolt extends BaseBasicBolt {
this.classes = classes;
}
+ public void setTimeLen(Duration x) { x.toMillis(); }
+ public void setTimeLenArr(Duration[] x) { x.toString(); }
+
+ public void withDuration(Duration x) { x.toMillis(); }
+
+ public void withDurationArr(Duration[] x) { x.toString(); }
+
+
public String getFoo(){
return this.foo;
}
@@ -128,4 +139,13 @@ public class TestBolt extends BaseBasicBolt {
public TestClass[] getClasses() {
return classes;
}
+
+ // Factory methods
+ public static TestBolt newInstance() {
+ return newInstance(TestEnum.FOO);
+ }
+
+ public static TestBolt newInstance(TestEnum te){
+ return new TestBolt(te);
+ }
}
http://git-wip-us.apache.org/repos/asf/storm/blob/cfb7d263/flux/flux-core/src/test/resources/configs/config-methods-test.yaml
----------------------------------------------------------------------
diff --git a/flux/flux-core/src/test/resources/configs/config-methods-test.yaml b/flux/flux-core/src/test/resources/configs/config-methods-test.yaml
index 7c4ffb3..baa6589 100644
--- a/flux/flux-core/src/test/resources/configs/config-methods-test.yaml
+++ b/flux/flux-core/src/test/resources/configs/config-methods-test.yaml
@@ -35,6 +35,13 @@ components:
constructorArgs:
- "baz"
+ - id: "time1"
+ className: "java.time.Duration"
+ factory: "ofHours"
+ factoryArgs:
+ - 10
+
+
# spout definitions
spouts:
- id: "spout-1"
@@ -70,8 +77,39 @@ bolts:
- "foo"
- "bar"
- "baz"
+ - name: "withDuration"
+ args:
+ - ref: "time1"
+ - name: "withDurationArr"
+ args:
+ - reflist:
+ - "time1"
+ - "time1"
+ - name: "setTimeLenArr"
+ args:
+ - reflist: ["time1", "time2"]
+ properties:
+ - name: "timeLen"
+ ref: "time1"
+ - id: "bolt-2"
+ className: "org.apache.storm.flux.test.TestBolt"
+ factory: "newInstance"
+ parallelism: 1
+
+ - id: "bolt-3"
+ className: "org.apache.storm.flux.test.TestBolt"
+ factory: "newInstance"
+ factoryArgs:
+ - FOO
+ parallelism: 1
+
+ - id: "bolt-4"
+ className: "org.apache.storm.flux.test.TestBolt"
+ constructorArgs:
+ - 10
+ parallelism: 1
#stream definitions
# stream definitions define connections between spouts and bolts.
# note that such connections can be cyclical
http://git-wip-us.apache.org/repos/asf/storm/blob/cfb7d263/flux/flux-examples/pom.xml
----------------------------------------------------------------------
diff --git a/flux/flux-examples/pom.xml b/flux/flux-examples/pom.xml
index 3a15f2e..a9d9c1e 100644
--- a/flux/flux-examples/pom.xml
+++ b/flux/flux-examples/pom.xml
@@ -150,7 +150,7 @@
<artifactId>maven-checkstyle-plugin</artifactId>
<!--Note - the version would be inherited-->
<configuration>
- <maxAllowedViolations>1</maxAllowedViolations>
+ <maxAllowedViolations>0</maxAllowedViolations>
</configuration>
</plugin>
</plugins>
http://git-wip-us.apache.org/repos/asf/storm/blob/cfb7d263/flux/flux-examples/src/main/java/org/apache/storm/flux/examples/StatefulWordCounter.java
----------------------------------------------------------------------
diff --git a/flux/flux-examples/src/main/java/org/apache/storm/flux/examples/StatefulWordCounter.java b/flux/flux-examples/src/main/java/org/apache/storm/flux/examples/StatefulWordCounter.java
index cc3ef38..f1f4a6c 100644
--- a/flux/flux-examples/src/main/java/org/apache/storm/flux/examples/StatefulWordCounter.java
+++ b/flux/flux-examples/src/main/java/org/apache/storm/flux/examples/StatefulWordCounter.java
@@ -15,19 +15,20 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.storm.flux.examples;
+import java.util.Map;
+
import org.apache.storm.state.KeyValueState;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
-import org.apache.storm.topology.base.BaseStatefulBolt;
import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.topology.base.BaseStatefulBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
-import java.util.Map;
-
public class StatefulWordCounter extends BaseStatefulBolt<KeyValueState<String, Long>> {
private KeyValueState<String, Long> wordCounts;
http://git-wip-us.apache.org/repos/asf/storm/blob/cfb7d263/flux/flux-examples/src/main/java/org/apache/storm/flux/examples/TestPrintBolt.java
----------------------------------------------------------------------
diff --git a/flux/flux-examples/src/main/java/org/apache/storm/flux/examples/TestPrintBolt.java b/flux/flux-examples/src/main/java/org/apache/storm/flux/examples/TestPrintBolt.java
index 137e354..2d28c7f 100644
--- a/flux/flux-examples/src/main/java/org/apache/storm/flux/examples/TestPrintBolt.java
+++ b/flux/flux-examples/src/main/java/org/apache/storm/flux/examples/TestPrintBolt.java
@@ -15,6 +15,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.storm.flux.examples;
import org.apache.storm.topology.BasicOutputCollector;
@@ -23,7 +24,7 @@ import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.tuple.Tuple;
/**
- * Prints the tuples to stdout
+ * Prints the tuples to stdout.
*/
public class TestPrintBolt extends BaseBasicBolt {
http://git-wip-us.apache.org/repos/asf/storm/blob/cfb7d263/flux/flux-examples/src/main/java/org/apache/storm/flux/examples/TestWindowBolt.java
----------------------------------------------------------------------
diff --git a/flux/flux-examples/src/main/java/org/apache/storm/flux/examples/TestWindowBolt.java b/flux/flux-examples/src/main/java/org/apache/storm/flux/examples/TestWindowBolt.java
index f8c140b..57826b5 100644
--- a/flux/flux-examples/src/main/java/org/apache/storm/flux/examples/TestWindowBolt.java
+++ b/flux/flux-examples/src/main/java/org/apache/storm/flux/examples/TestWindowBolt.java
@@ -15,8 +15,11 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.storm.flux.examples;
+import java.util.Map;
+
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
@@ -25,8 +28,6 @@ import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;
import org.apache.storm.windowing.TupleWindow;
-import java.util.Map;
-
public class TestWindowBolt extends BaseWindowedBolt {
private OutputCollector collector;