You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by pt...@apache.org on 2015/06/03 19:35:49 UTC
[35/50] [abbrv] storm git commit: integrate flux with Storm build
integrate flux with Storm build
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/2094a08e
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/2094a08e
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/2094a08e
Branch: refs/heads/master
Commit: 2094a08ea6ccdf81126c103930d9ec3e77fcd5c1
Parents: b21a98d
Author: P. Taylor Goetz <pt...@gmail.com>
Authored: Wed May 6 14:13:13 2015 -0400
Committer: P. Taylor Goetz <pt...@gmail.com>
Committed: Wed May 6 14:13:13 2015 -0400
----------------------------------------------------------------------
external/flux/README.md | 15 +-
external/flux/flux-core/pom.xml | 13 +-
external/flux/flux-examples/pom.xml | 13 +-
external/flux/flux-wrappers/pom.xml | 24 +-
.../main/resources/resources/randomsentence.js | 93 -----
.../main/resources/resources/splitsentence.py | 24 --
.../src/main/resources/resources/storm.js | 373 -------------------
.../src/main/resources/resources/storm.py | 260 -------------
external/flux/pom.xml | 19 +-
pom.xml | 1 +
10 files changed, 41 insertions(+), 794 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/2094a08e/external/flux/README.md
----------------------------------------------------------------------
diff --git a/external/flux/README.md b/external/flux/README.md
index 6f27219..d09a73c 100644
--- a/external/flux/README.md
+++ b/external/flux/README.md
@@ -829,17 +829,6 @@ topologySource:
methodName: "getTopologyWithDifferentMethodName"
```
-## Author
-P. Taylor Goetz
+## Committer Sponsors
-## Contributors
-
-
-## Contributing
-
-Contributions in any form are more than welcome.
-
-The intent of this project is that it will be donated to Apache Storm.
-
-By offering any contributions to this project, you should be willing and able to submit an
-[Apache ICLA](http://www.apache.org/licenses/icla.txt), if you have not done so already.
\ No newline at end of file
+ * P. Taylor Goetz ([ptgoetz@apache.org](mailto:ptgoetz@apache.org))
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/storm/blob/2094a08e/external/flux/flux-core/pom.xml
----------------------------------------------------------------------
diff --git a/external/flux/flux-core/pom.xml b/external/flux/flux-core/pom.xml
index 600613d..c3842bd 100644
--- a/external/flux/flux-core/pom.xml
+++ b/external/flux/flux-core/pom.xml
@@ -19,13 +19,12 @@
<modelVersion>4.0.0</modelVersion>
<parent>
- <groupId>com.github.ptgoetz</groupId>
+ <groupId>org.apache.storm</groupId>
<artifactId>flux</artifactId>
- <version>0.3.1-SNAPSHOT</version>
+ <version>0.11.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
- <groupId>com.github.ptgoetz</groupId>
<artifactId>flux-core</artifactId>
<packaging>jar</packaging>
@@ -34,26 +33,26 @@
<dependencies>
<dependency>
- <groupId>com.github.ptgoetz</groupId>
+ <groupId>org.apache.storm</groupId>
<artifactId>flux-wrappers</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-kafka</artifactId>
- <version>${storm.version}</version>
+ <version>${project.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-hdfs</artifactId>
- <version>${storm.version}</version>
+ <version>${project.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-hbase</artifactId>
- <version>${storm.version}</version>
+ <version>${project.version}</version>
<scope>test</scope>
</dependency>
</dependencies>
http://git-wip-us.apache.org/repos/asf/storm/blob/2094a08e/external/flux/flux-examples/pom.xml
----------------------------------------------------------------------
diff --git a/external/flux/flux-examples/pom.xml b/external/flux/flux-examples/pom.xml
index 0b9796e..709b20b 100644
--- a/external/flux/flux-examples/pom.xml
+++ b/external/flux/flux-examples/pom.xml
@@ -19,13 +19,12 @@
<modelVersion>4.0.0</modelVersion>
<parent>
- <groupId>com.github.ptgoetz</groupId>
+ <groupId>org.apache.storm</groupId>
<artifactId>flux</artifactId>
- <version>0.3.1-SNAPSHOT</version>
+ <version>0.11.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
- <groupId>com.github.ptgoetz</groupId>
<artifactId>flux-examples</artifactId>
<packaging>jar</packaging>
@@ -34,12 +33,12 @@
<dependencies>
<dependency>
- <groupId>com.github.ptgoetz</groupId>
+ <groupId>org.apache.storm</groupId>
<artifactId>flux-core</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
- <groupId>com.github.ptgoetz</groupId>
+ <groupId>org.apache.storm</groupId>
<artifactId>flux-wrappers</artifactId>
<version>${project.version}</version>
</dependency>
@@ -47,12 +46,12 @@
<dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-hdfs</artifactId>
- <version>${storm.version}</version>
+ <version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-hbase</artifactId>
- <version>${storm.version}</version>
+ <version>${project.version}</version>
</dependency>
</dependencies>
http://git-wip-us.apache.org/repos/asf/storm/blob/2094a08e/external/flux/flux-wrappers/pom.xml
----------------------------------------------------------------------
diff --git a/external/flux/flux-wrappers/pom.xml b/external/flux/flux-wrappers/pom.xml
index 6784141..be042ff 100644
--- a/external/flux/flux-wrappers/pom.xml
+++ b/external/flux/flux-wrappers/pom.xml
@@ -19,17 +19,33 @@
<modelVersion>4.0.0</modelVersion>
<parent>
- <groupId>com.github.ptgoetz</groupId>
+ <groupId>org.apache.storm</groupId>
<artifactId>flux</artifactId>
- <version>0.3.1-SNAPSHOT</version>
+ <version>0.11.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
- <groupId>com.github.ptgoetz</groupId>
<artifactId>flux-wrappers</artifactId>
<packaging>jar</packaging>
<name>flux-wrappers</name>
- <url>https://github.com/ptgoetz/flux</url>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.storm</groupId>
+ <artifactId>multilang-javascript</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.storm</groupId>
+ <artifactId>multilang-ruby</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.storm</groupId>
+ <artifactId>multilang-python</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ </dependencies>
</project>
http://git-wip-us.apache.org/repos/asf/storm/blob/2094a08e/external/flux/flux-wrappers/src/main/resources/resources/randomsentence.js
----------------------------------------------------------------------
diff --git a/external/flux/flux-wrappers/src/main/resources/resources/randomsentence.js b/external/flux/flux-wrappers/src/main/resources/resources/randomsentence.js
deleted file mode 100644
index 36fc5f5..0000000
--- a/external/flux/flux-wrappers/src/main/resources/resources/randomsentence.js
+++ /dev/null
@@ -1,93 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-/**
- * Example for storm spout. Emits random sentences.
- * The original class in java - storm.starter.spout.RandomSentenceSpout.
- *
- */
-
-var storm = require('./storm');
-var Spout = storm.Spout;
-
-
-var SENTENCES = [
- "the cow jumped over the moon",
- "an apple a day keeps the doctor away",
- "four score and seven years ago",
- "snow white and the seven dwarfs",
- "i am at two with nature"]
-
-function RandomSentenceSpout(sentences) {
- Spout.call(this);
- this.runningTupleId = 0;
- this.sentences = sentences;
- this.pending = {};
-};
-
-RandomSentenceSpout.prototype = Object.create(Spout.prototype);
-RandomSentenceSpout.prototype.constructor = RandomSentenceSpout;
-
-RandomSentenceSpout.prototype.getRandomSentence = function() {
- return this.sentences[getRandomInt(0, this.sentences.length - 1)];
-}
-
-RandomSentenceSpout.prototype.nextTuple = function(done) {
- var self = this;
- var sentence = this.getRandomSentence();
- var tup = [sentence];
- var id = this.createNextTupleId();
- this.pending[id] = tup;
- //This timeout can be removed if TOPOLOGY_SLEEP_SPOUT_WAIT_STRATEGY_TIME_MS is configured to 100
- setTimeout(function() {
- self.emit({tuple: tup, id: id}, function(taskIds) {
- self.log(tup + ' sent to task ids - ' + taskIds);
- });
- done();
- },100);
-}
-
-RandomSentenceSpout.prototype.createNextTupleId = function() {
- var id = this.runningTupleId;
- this.runningTupleId++;
- return id;
-}
-
-RandomSentenceSpout.prototype.ack = function(id, done) {
- this.log('Received ack for - ' + id);
- delete this.pending[id];
- done();
-}
-
-RandomSentenceSpout.prototype.fail = function(id, done) {
- var self = this;
- this.log('Received fail for - ' + id + '. Retrying.');
- this.emit({tuple: this.pending[id], id:id}, function(taskIds) {
- self.log(self.pending[id] + ' sent to task ids - ' + taskIds);
- });
- done();
-}
-
-/**
- * Returns a random integer between min (inclusive) and max (inclusive)
- */
-function getRandomInt(min, max) {
- return Math.floor(Math.random() * (max - min + 1)) + min;
-}
-
-new RandomSentenceSpout(SENTENCES).run();
http://git-wip-us.apache.org/repos/asf/storm/blob/2094a08e/external/flux/flux-wrappers/src/main/resources/resources/splitsentence.py
----------------------------------------------------------------------
diff --git a/external/flux/flux-wrappers/src/main/resources/resources/splitsentence.py b/external/flux/flux-wrappers/src/main/resources/resources/splitsentence.py
deleted file mode 100644
index 300105f..0000000
--- a/external/flux/flux-wrappers/src/main/resources/resources/splitsentence.py
+++ /dev/null
@@ -1,24 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements. See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership. The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License. You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-import storm
-
-class SplitSentenceBolt(storm.BasicBolt):
- def process(self, tup):
- words = tup.values[0].split(" ")
- for word in words:
- storm.emit([word])
-
-SplitSentenceBolt().run()
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/storm/blob/2094a08e/external/flux/flux-wrappers/src/main/resources/resources/storm.js
----------------------------------------------------------------------
diff --git a/external/flux/flux-wrappers/src/main/resources/resources/storm.js b/external/flux/flux-wrappers/src/main/resources/resources/storm.js
deleted file mode 100755
index 355c2d2..0000000
--- a/external/flux/flux-wrappers/src/main/resources/resources/storm.js
+++ /dev/null
@@ -1,373 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-/**
- * Base classes in node-js for storm Bolt and Spout.
- * Implements the storm multilang protocol for nodejs.
- */
-
-
-var fs = require('fs');
-
-function Storm() {
- this.messagePart = "";
- this.taskIdsCallbacks = [];
- this.isFirstMessage = true;
- this.separator = '\nend\n';
-}
-
-Storm.prototype.sendMsgToParent = function(msg) {
- var str = JSON.stringify(msg);
- process.stdout.write(str + this.separator);
-}
-
-Storm.prototype.sync = function() {
- this.sendMsgToParent({"command":"sync"});
-}
-
-Storm.prototype.sendPid = function(heartbeatdir) {
- var pid = process.pid;
- fs.closeSync(fs.openSync(heartbeatdir + "/" + pid, "w"));
- this.sendMsgToParent({"pid": pid})
-}
-
-Storm.prototype.log = function(msg) {
- this.sendMsgToParent({"command": "log", "msg": msg});
-}
-
-Storm.prototype.initSetupInfo = function(setupInfo) {
- var self = this;
- var callback = function() {
- self.sendPid(setupInfo['pidDir']);
- }
- this.initialize(setupInfo['conf'], setupInfo['context'], callback);
-}
-
-Storm.prototype.startReadingInput = function() {
- var self = this;
- process.stdin.on('readable', function() {
- var chunk = process.stdin.read();
- var messages = self.handleNewChunk(chunk);
- messages.forEach(function(message) {
- self.handleNewMessage(message);
- })
-
- });
-}
-
-/**
- * receives a new string chunk and returns a list of new messages with the separator removed
- * stores state in this.messagePart
- * @param chunk
- */
-Storm.prototype.handleNewChunk = function(chunk) {
- //invariant: this.messagePart has no separator otherwise we would have parsed it already
- var messages = [];
- if (chunk && chunk.length !== 0) {
- //"{}".split("\nend\n") ==> ['{}']
- //"\nend\n".split("\nend\n") ==> ['' , '']
- //"{}\nend\n".split("\nend\n") ==> ['{}', '']
- //"\nend\n{}".split("\nend\n") ==> ['' , '{}']
- // "{}\nend\n{}".split("\nend\n") ==> ['{}', '{}' ]
- this.messagePart = this.messagePart + chunk;
- var newMessageParts = this.messagePart.split(this.separator);
- while (newMessageParts.length > 0) {
- var potentialMessage = newMessageParts.shift();
- var anotherMessageAhead = newMessageParts.length > 0;
- if (!anotherMessageAhead) {
- this.messagePart = potentialMessage;
- }
- else if (potentialMessage.length > 0) {
- messages.push(potentialMessage);
- }
- }
- }
- return messages;
-}
-
-Storm.prototype.isTaskIds = function(msg) {
- return (msg instanceof Array);
-}
-
-Storm.prototype.handleNewMessage = function(msg) {
- var parsedMsg = JSON.parse(msg);
-
- if (this.isFirstMessage) {
- this.initSetupInfo(parsedMsg);
- this.isFirstMessage = false;
- } else if (this.isTaskIds(parsedMsg)) {
- this.handleNewTaskId(parsedMsg);
- } else {
- this.handleNewCommand(parsedMsg);
- }
-}
-
-Storm.prototype.handleNewTaskId = function(taskIds) {
- //When new list of task ids arrives, the callback that was passed with the corresponding emit should be called.
- //Storm assures that the task ids will be sent in the same order as their corresponding emits so it we can simply
- //take the first callback in the list and be sure it is the right one.
-
- var callback = this.taskIdsCallbacks.shift();
- if (callback) {
- callback(taskIds);
- } else {
- throw new Error('Something went wrong, we off the split of task id callbacks');
- }
-}
-
-
-
-/**
- *
- * @param messageDetails json with the emit details.
- *
- * For bolt, the json must contain the required fields:
- * - tuple - the value to emit
- * - anchorTupleId - the value of the anchor tuple (the input tuple that lead to this emit). Used to track the source
- * tuple and return ack when all components successfully finished to process it.
- * and may contain the optional fields:
- * - stream (if empty - emit to default stream)
- *
- * For spout, the json must contain the required fields:
- * - tuple - the value to emit
- *
- * and may contain the optional fields:
- * - id - pass id for reliable emit (and receive ack/fail later).
- * - stream - if empty - emit to default stream.
- *
- * @param onTaskIds function than will be called with list of task ids the message was emitted to (when received).
- */
-Storm.prototype.emit = function(messageDetails, onTaskIds) {
- //Every emit triggers a response - list of task ids to which the tuple was emitted. The task ids are accessible
- //through the callback (will be called when the response arrives). The callback is stored in a list until the
- //corresponding task id list arrives.
- if (messageDetails.task) {
- throw new Error('Illegal input - task. To emit to specific task use emit direct!');
- }
-
- if (!onTaskIds) {
- throw new Error('You must pass a onTaskIds callback when using emit!')
- }
-
- this.taskIdsCallbacks.push(onTaskIds);
- this.__emit(messageDetails);;
-}
-
-
-/**
- * Emit message to specific task.
- * @param messageDetails json with the emit details.
- *
- * For bolt, the json must contain the required fields:
- * - tuple - the value to emit
- * - anchorTupleId - the value of the anchor tuple (the input tuple that lead to this emit). Used to track the source
- * tuple and return ack when all components successfully finished to process it.
- * - task - indicate the task to send the tuple to.
- * and may contain the optional fields:
- * - stream (if empty - emit to default stream)
- *
- * For spout, the json must contain the required fields:
- * - tuple - the value to emit
- * - task - indicate the task to send the tuple to.
- * and may contain the optional fields:
- * - id - pass id for reliable emit (and receive ack/fail later).
- * - stream - if empty - emit to default stream.
- *
- * @param onTaskIds function than will be called with list of task ids the message was emitted to (when received).
- */
-Storm.prototype.emitDirect = function(commandDetails) {
- if (!commandDetails.task) {
- throw new Error("Emit direct must receive task id!")
- }
- this.__emit(commandDetails);
-}
-
-/**
- * Initialize storm component according to the configuration received.
- * @param conf configuration object accrding to storm protocol.
- * @param context context object according to storm protocol.
- * @param done callback. Call this method when finished initializing.
- */
-Storm.prototype.initialize = function(conf, context, done) {
- done();
-}
-
-Storm.prototype.run = function() {
- process.stdout.setEncoding('utf8');
- process.stdin.setEncoding('utf8');
- this.startReadingInput();
-}
-
-function Tuple(id, component, stream, task, values) {
- this.id = id;
- this.component = component;
- this.stream = stream;
- this.task = task;
- this.values = values;
-}
-
-/**
- * Base class for storm bolt.
- * To create a bolt implement 'process' method.
- * You may also implement initialize method to
- */
-function BasicBolt() {
- Storm.call(this);
- this.anchorTuple = null;
-};
-
-BasicBolt.prototype = Object.create(Storm.prototype);
-BasicBolt.prototype.constructor = BasicBolt;
-
-/**
- * Emit message.
- * @param commandDetails json with the required fields:
- * - tuple - the value to emit
- * - anchorTupleId - the value of the anchor tuple (the input tuple that lead to this emit). Used to track the source
- * tuple and return ack when all components successfully finished to process it.
- * and the optional fields:
- * - stream (if empty - emit to default stream)
- * - task (pass only to emit to specific task)
- */
-BasicBolt.prototype.__emit = function(commandDetails) {
- var self = this;
-
- var message = {
- command: "emit",
- tuple: commandDetails.tuple,
- stream: commandDetails.stream,
- task: commandDetails.task,
- anchors: [commandDetails.anchorTupleId]
- };
-
- this.sendMsgToParent(message);
-}
-
-BasicBolt.prototype.handleNewCommand = function(command) {
- var self = this;
- var tup = new Tuple(command["id"], command["comp"], command["stream"], command["task"], command["tuple"]);
-
- if (tup.task === -1 && tup.stream === "__heartbeat") {
- self.sync();
- return;
- }
-
- var callback = function(err) {
- if (err) {
- self.fail(tup, err);
- return;
- }
- self.ack(tup);
- }
- this.process(tup, callback);
-}
-
-/**
- * Implement this method when creating a bolt. This is the main method that provides the logic of the bolt (what
- * should it do?).
- * @param tuple the input of the bolt - what to process.
- * @param done call this method when done processing.
- */
-BasicBolt.prototype.process = function(tuple, done) {};
-
-BasicBolt.prototype.ack = function(tup) {
- this.sendMsgToParent({"command": "ack", "id": tup.id});
-}
-
-BasicBolt.prototype.fail = function(tup, err) {
- this.sendMsgToParent({"command": "fail", "id": tup.id});
-}
-
-
-/**
- * Base class for storm spout.
- * To create a spout implement the following methods: nextTuple, ack and fail (nextTuple - mandatory, ack and fail
- * can stay empty).
- * You may also implement initialize method.
- *
- */
-function Spout() {
- Storm.call(this);
-};
-
-Spout.prototype = Object.create(Storm.prototype);
-
-Spout.prototype.constructor = Spout;
-
-/**
- * This method will be called when an ack is received for preciously sent tuple. One may implement it.
- * @param id The id of the tuple.
- * @param done Call this method when finished and ready to receive more tuples.
- */
-Spout.prototype.ack = function(id, done) {};
-
-/**
- * This method will be called when an fail is received for preciously sent tuple. One may implement it (for example -
- * log the failure or send the tuple again).
- * @param id The id of the tuple.
- * @param done Call this method when finished and ready to receive more tuples.
- */
-Spout.prototype.fail = function(id, done) {};
-
-/**
- * Method the indicates its time to emit the next tuple.
- * @param done call this method when done sending the output.
- */
-Spout.prototype.nextTuple = function(done) {};
-
-Spout.prototype.handleNewCommand = function(command) {
- var self = this;
- var callback = function() {
- self.sync();
- }
-
- if (command["command"] === "next") {
- this.nextTuple(callback);
- }
-
- if (command["command"] === "ack") {
- this.ack(command["id"], callback);
- }
-
- if (command["command"] === "fail") {
- this.fail(command["id"], callback);
- }
-}
-
-/**
- * @param commandDetails json with the required fields:
- * - tuple - the value to emit.
- * and the optional fields:
- * - id - pass id for reliable emit (and receive ack/fail later).
- * - stream - if empty - emit to default stream.
- * - task - pass only to emit to specific task.
- */
-Spout.prototype.__emit = function(commandDetails) {
- var message = {
- command: "emit",
- tuple: commandDetails.tuple,
- id: commandDetails.id,
- stream: commandDetails.stream,
- task: commandDetails.task
- };
-
- this.sendMsgToParent(message);
-}
-
-module.exports.BasicBolt = BasicBolt;
-module.exports.Spout = Spout;
http://git-wip-us.apache.org/repos/asf/storm/blob/2094a08e/external/flux/flux-wrappers/src/main/resources/resources/storm.py
----------------------------------------------------------------------
diff --git a/external/flux/flux-wrappers/src/main/resources/resources/storm.py b/external/flux/flux-wrappers/src/main/resources/resources/storm.py
deleted file mode 100644
index 642c393..0000000
--- a/external/flux/flux-wrappers/src/main/resources/resources/storm.py
+++ /dev/null
@@ -1,260 +0,0 @@
-# -*- coding: utf-8 -*-
-
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements. See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership. The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License. You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-import sys
-import os
-import traceback
-from collections import deque
-
-try:
- import simplejson as json
-except ImportError:
- import json
-
-json_encode = lambda x: json.dumps(x)
-json_decode = lambda x: json.loads(x)
-
-#reads lines and reconstructs newlines appropriately
-def readMsg():
- msg = ""
- while True:
- line = sys.stdin.readline()
- if not line:
- raise Exception('Read EOF from stdin')
- if line[0:-1] == "end":
- break
- msg = msg + line
- return json_decode(msg[0:-1])
-
-MODE = None
-ANCHOR_TUPLE = None
-
-#queue up commands we read while trying to read taskids
-pending_commands = deque()
-
-def readTaskIds():
- if pending_taskids:
- return pending_taskids.popleft()
- else:
- msg = readMsg()
- while type(msg) is not list:
- pending_commands.append(msg)
- msg = readMsg()
- return msg
-
-#queue up taskids we read while trying to read commands/tuples
-pending_taskids = deque()
-
-def readCommand():
- if pending_commands:
- return pending_commands.popleft()
- else:
- msg = readMsg()
- while type(msg) is list:
- pending_taskids.append(msg)
- msg = readMsg()
- return msg
-
-def readTuple():
- cmd = readCommand()
- return Tuple(cmd["id"], cmd["comp"], cmd["stream"], cmd["task"], cmd["tuple"])
-
-def sendMsgToParent(msg):
- print json_encode(msg)
- print "end"
- sys.stdout.flush()
-
-def sync():
- sendMsgToParent({'command':'sync'})
-
-def sendpid(heartbeatdir):
- pid = os.getpid()
- sendMsgToParent({'pid':pid})
- open(heartbeatdir + "/" + str(pid), "w").close()
-
-def emit(*args, **kwargs):
- __emit(*args, **kwargs)
- return readTaskIds()
-
-def emitDirect(task, *args, **kwargs):
- kwargs["directTask"] = task
- __emit(*args, **kwargs)
-
-def __emit(*args, **kwargs):
- global MODE
- if MODE == Bolt:
- emitBolt(*args, **kwargs)
- elif MODE == Spout:
- emitSpout(*args, **kwargs)
-
-def emitBolt(tup, stream=None, anchors = [], directTask=None):
- global ANCHOR_TUPLE
- if ANCHOR_TUPLE is not None:
- anchors = [ANCHOR_TUPLE]
- m = {"command": "emit"}
- if stream is not None:
- m["stream"] = stream
- m["anchors"] = map(lambda a: a.id, anchors)
- if directTask is not None:
- m["task"] = directTask
- m["tuple"] = tup
- sendMsgToParent(m)
-
-def emitSpout(tup, stream=None, id=None, directTask=None):
- m = {"command": "emit"}
- if id is not None:
- m["id"] = id
- if stream is not None:
- m["stream"] = stream
- if directTask is not None:
- m["task"] = directTask
- m["tuple"] = tup
- sendMsgToParent(m)
-
-def ack(tup):
- sendMsgToParent({"command": "ack", "id": tup.id})
-
-def fail(tup):
- sendMsgToParent({"command": "fail", "id": tup.id})
-
-def reportError(msg):
- sendMsgToParent({"command": "error", "msg": msg})
-
-def log(msg, level=2):
- sendMsgToParent({"command": "log", "msg": msg, "level":level})
-
-def logTrace(msg):
- log(msg, 0)
-
-def logDebug(msg):
- log(msg, 1)
-
-def logInfo(msg):
- log(msg, 2)
-
-def logWarn(msg):
- log(msg, 3)
-
-def logError(msg):
- log(msg, 4)
-
-def rpcMetrics(name, params):
- sendMsgToParent({"command": "metrics", "name": name, "params": params})
-
-def initComponent():
- setupInfo = readMsg()
- sendpid(setupInfo['pidDir'])
- return [setupInfo['conf'], setupInfo['context']]
-
-class Tuple(object):
- def __init__(self, id, component, stream, task, values):
- self.id = id
- self.component = component
- self.stream = stream
- self.task = task
- self.values = values
-
- def __repr__(self):
- return '<%s%s>' % (
- self.__class__.__name__,
- ''.join(' %s=%r' % (k, self.__dict__[k]) for k in sorted(self.__dict__.keys())))
-
- def is_heartbeat_tuple(self):
- return self.task == -1 and self.stream == "__heartbeat"
-
-class Bolt(object):
- def initialize(self, stormconf, context):
- pass
-
- def process(self, tuple):
- pass
-
- def run(self):
- global MODE
- MODE = Bolt
- conf, context = initComponent()
- try:
- self.initialize(conf, context)
- while True:
- tup = readTuple()
- if tup.is_heartbeat_tuple():
- sync()
- else:
- self.process(tup)
- except Exception, e:
- reportError(traceback.format_exc(e))
-
-class BasicBolt(object):
- def initialize(self, stormconf, context):
- pass
-
- def process(self, tuple):
- pass
-
- def run(self):
- global MODE
- MODE = Bolt
- global ANCHOR_TUPLE
- conf, context = initComponent()
- try:
- self.initialize(conf, context)
- while True:
- tup = readTuple()
- if tup.is_heartbeat_tuple():
- sync()
- else:
- ANCHOR_TUPLE = tup
- try:
- self.process(tup)
- ack(tup)
- except Exception, e:
- reportError(traceback.format_exc(e))
- fail(tup)
- except Exception, e:
- reportError(traceback.format_exc(e))
-
-class Spout(object):
- def initialize(self, conf, context):
- pass
-
- def ack(self, id):
- pass
-
- def fail(self, id):
- pass
-
- def nextTuple(self):
- pass
-
- def run(self):
- global MODE
- MODE = Spout
- conf, context = initComponent()
- try:
- self.initialize(conf, context)
- while True:
- msg = readCommand()
- if msg["command"] == "next":
- self.nextTuple()
- if msg["command"] == "ack":
- self.ack(msg["id"])
- if msg["command"] == "fail":
- self.fail(msg["id"])
- sync()
- except Exception, e:
- reportError(traceback.format_exc(e))
http://git-wip-us.apache.org/repos/asf/storm/blob/2094a08e/external/flux/pom.xml
----------------------------------------------------------------------
diff --git a/external/flux/pom.xml b/external/flux/pom.xml
index 5ea1b40..bf975cb 100644
--- a/external/flux/pom.xml
+++ b/external/flux/pom.xml
@@ -18,23 +18,17 @@
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
- <groupId>com.github.ptgoetz</groupId>
<artifactId>flux</artifactId>
- <version>0.3.1-SNAPSHOT</version>
<packaging>pom</packaging>
<name>flux</name>
- <url>https://github.com/ptgoetz/flux</url>
+
<parent>
- <groupId>org.sonatype.oss</groupId>
- <artifactId>oss-parent</artifactId>
- <version>7</version>
+ <artifactId>storm</artifactId>
+ <groupId>org.apache.storm</groupId>
+ <version>0.11.0-SNAPSHOT</version>
+ <relativePath>../../pom.xml</relativePath>
</parent>
- <scm>
- <connection>scm:git:git@github.com:ptgoetz/flux.git</connection>
- <developerConnection>scm:git:git@github.com:ptgoetz/flux.git</developerConnection>
- <url>:git@github.com:ptgoetz/flux.git</url>
- </scm>
<developers>
<developer>
@@ -46,7 +40,6 @@
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
- <storm.version>0.9.3</storm.version>
<!-- see comment below... This fixes an annoyance with intellij -->
<provided.scope>provided</provided.scope>
</properties>
@@ -75,7 +68,7 @@
<dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-core</artifactId>
- <version>${storm.version}</version>
+ <version>${project.version}</version>
<scope>${provided.scope}</scope>
</dependency>
<dependency>
http://git-wip-us.apache.org/repos/asf/storm/blob/2094a08e/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 2e0c898..02417b8 100644
--- a/pom.xml
+++ b/pom.xml
@@ -169,6 +169,7 @@
<module>external/storm-jdbc</module>
<module>external/storm-redis</module>
<module>external/storm-eventhubs</module>
+ <module>external/flux</module>
</modules>
<scm>