Merge branch 'STORM-519' of github.com:Parth-Brahmbhatt/incubator-storm
diff --git a/CHANGELOG.md b/CHANGELOG.md
index c2a9a70..360979a 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,4 +1,8 @@
-## 0.9.3-incubating
+## 0.9.3
+ * STORM-488: Exit with 254 error code if storm CLI is run with unknown command
+ * STORM-506: Do not count bolt acks & fails in total stats
+ * STORM-490: fix build under Windows
+ * STORM-439: Replace purl.js qith jquery URL plugin
* STORM-499: Document and clean up shaded dependncy resolution with maven
* STORM-210: Add storm-hbase module
* STORM-507: Topology visualization should not block ui
diff --git a/LICENSE b/LICENSE
index d15be6c..e30786b 100644
--- a/LICENSE
+++ b/LICENSE
@@ -371,16 +371,6 @@
-----------------------------------------------------------------------
-For Purl (storm-core/src/ui/public/js/purl.js)
-
-Purl (A JavaScript URL parser) v2.3.1
-Developed and maintanined by Mark Perkins, mark@allmarkedup.com
-Source repository: https://github.com/allmarkedup/jQuery-URL-Parser
-Licensed under an MIT-style license. See https://github.com/allmarkedup/jQuery-URL-Parser/blob/master/LICENSE for details.
-
-
------------------------------------------------------------------------
-
For mustache.js (storm-core/src/ui/public/js/jquery.mustache.js)
The MIT License
@@ -433,8 +423,18 @@
FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
OTHER DEALINGS IN THE SOFTWARE.
+
-----------------------------------------------------------------------
+For Jquery url plugin (storm-core/src/ui/public/js/url.min.js)
+
+Jquery Url (A Jquery plugin for URL parser) v1.8.6
+Source repository: https://github.com/websanova/js-url
+Licensed under an MIT-style license. Seehttps://github.com/websanova/js-url#license for details.
+
+-----------------------------------------------------------------------
+
+
For jquery.blockUI.min.js (storm-core/src/ui/public/js/jquery.blockUI.min.js)
jQuery BlockUI; v20131009
diff --git a/bin/storm b/bin/storm
index 2c75e58..fee548a 100755
--- a/bin/storm
+++ b/bin/storm
@@ -467,6 +467,7 @@
def unknown_command(*args):
print("Unknown command: [storm %s]" % ' '.join(sys.argv[1:]))
print_usage()
+ sys.exit(254)
COMMANDS = {"jar": jar, "kill": kill, "shell": shell, "nimbus": nimbus, "ui": ui, "logviewer": logviewer,
"drpc": drpc, "supervisor": supervisor, "localconfvalue": print_localconfvalue,
diff --git a/storm-core/src/clj/backtype/storm/daemon/supervisor.clj b/storm-core/src/clj/backtype/storm/daemon/supervisor.clj
index 8c18e1c..cce89bc 100644
--- a/storm-core/src/clj/backtype/storm/daemon/supervisor.clj
+++ b/storm-core/src/clj/backtype/storm/daemon/supervisor.clj
@@ -502,9 +502,7 @@
:distributed [supervisor storm-id port worker-id]
(let [conf (:conf supervisor)
storm-home (System/getProperty "storm.home")
- storm-options (System/getProperty "storm.options")
- storm-conf-file (System/getProperty "storm.conf.file")
- storm-log-dir (or (System/getProperty "storm.log.dir") (str storm-home "/logs"))
+ storm-log-dir (or (System/getProperty "storm.log.dir") (str storm-home file-path-separator "logs"))
stormroot (supervisor-stormdist-root conf storm-id)
jlp (jlp stormroot conf)
stormjar (supervisor-stormjar-path stormroot)
@@ -530,10 +528,8 @@
[(str "-Djava.library.path=" jlp)
(str "-Dlogfile.name=" logfilename)
(str "-Dstorm.home=" storm-home)
- (str "-Dstorm.conf.file=" storm-conf-file)
- (str "-Dstorm.options=" storm-options)
(str "-Dstorm.log.dir=" storm-log-dir)
- (str "-Dlogback.configurationFile=" storm-home "/logback/cluster.xml")
+ (str "-Dlogback.configurationFile=" storm-home file-path-separator "logback" file-path-separator "cluster.xml")
(str "-Dstorm.id=" storm-id)
(str "-Dworker.id=" worker-id)
(str "-Dworker.port=" port)
@@ -629,4 +625,4 @@
))))
(defn -main []
- (-launch (standalone-supervisor)))
+ (-launch (standalone-supervisor)))
\ No newline at end of file
diff --git a/storm-core/src/clj/backtype/storm/ui/core.clj b/storm-core/src/clj/backtype/storm/ui/core.clj
index 9937607..969b514 100644
--- a/storm-core/src/clj/backtype/storm/ui/core.clj
+++ b/storm-core/src/clj/backtype/storm/ui/core.clj
@@ -341,10 +341,11 @@
(merge-with + s1 s2))
(select-keys
agg-bolt-stats
- [:emitted :transferred :acked :failed :complete-latencies])
- (select-keys
- agg-spout-stats
- [:emitted :transferred :acked :failed :complete-latencies]))))
+ ;; Include only keys that will be used. We want to count acked and
+ ;; failed only for the "tuple trees," so we do not include those keys
+ ;; from the bolt executors.
+ [:emitted :transferred])
+ agg-spout-stats)))
(defn stats-times
[stats-map]
@@ -518,7 +519,9 @@
([summs]
{"topologies"
(for [^TopologySummary t summs]
- {"id" (.get_id t)
+ {
+ "id" (.get_id t)
+ "encodedId" (url-encode (.get_id t))
"name" (.get_name t)
"status" (.get_status t)
"uptime" (pretty-uptime-sec (.get_uptime_secs t))
@@ -550,6 +553,7 @@
error-host (get-error-host last-error)
error-port (get-error-port last-error error-host top-id) ]]
{"spoutId" id
+ "encodedSpoutId" (url-encode id)
"executors" (count summs)
"tasks" (sum-tasks summs)
"emitted" (get-in stats [:emitted window])
@@ -573,6 +577,7 @@
error-host (get-error-host last-error)
error-port (get-error-port last-error error-host top-id) ]]
{"boltId" id
+ "encodedBoltId" (url-encode id)
"executors" (count summs)
"tasks" (sum-tasks summs)
"emitted" (get-in stats [:emitted window])
@@ -594,6 +599,7 @@
workers (set (for [^ExecutorSummary e executors]
[(.get_host e) (.get_port e)]))]
{"id" (.get_id summ)
+ "encodedId" (url-encode (.get_id summ))
"name" (.get_name summ)
"status" (.get_status summ)
"uptime" (pretty-uptime-sec (.get_uptime_secs summ))
@@ -671,6 +677,7 @@
swap-map-order
(get window)))]]
{"id" (pretty-executor-info (.get_executor_info e))
+ "encodedId" (url-encode (pretty-executor-info (.get_executor_info e)))
"uptime" (pretty-uptime-sec (.get_uptime_secs e))
"host" (.get_host e)
"port" (.get_port e)
@@ -747,6 +754,7 @@
swap-map-order)]
(for [[^GlobalStreamId s stats] stream-summary]
{"component" (.get_componentId s)
+ "encodedComponent" (url-encode (.get_componentId s))
"stream" (.get_streamId s)
"executeLatency" (float-str (:execute-latencies stats))
"processLatency" (float-str (:execute-latencies stats))
@@ -765,6 +773,7 @@
swap-map-order
(get window)))]]
{"id" (pretty-executor-info (.get_executor_info e))
+ "encodedId" (url-encode (pretty-executor-info (.get_executor_info e)))
"uptime" (pretty-uptime-sec (.get_uptime_secs e))
"host" (.get_host e)
"port" (.get_port e)
@@ -802,11 +811,13 @@
(= type :bolt) (bolt-stats window summ component summs include-sys?))
errors (component-errors (get (.get_errors summ) component) topology-id)]
(merge
- {"id" component
+ { "id" component
+ "encodedId" (url-encode component)
"name" (.get_name summ)
"executors" (count summs)
"tasks" (sum-tasks summs)
"topologyId" topology-id
+ "encodedTopologyId" (url-encode topology-id)
"window" window
"componentType" (name type)
"windowHint" (window-hint window)}
@@ -839,51 +850,44 @@
(GET "/api/v1/topology/summary" [& m]
(json-response (all-topologies-summary) (:callback m)))
(GET "/api/v1/topology/:id" [id & m]
- (let [id (url-decode id)]
- (json-response (topology-page id (:window m) (check-include-sys? (:sys m))) (:callback m))))
+ (json-response (topology-page id (:window m) (check-include-sys? (:sys m))) (:callback m)))
(GET "/api/v1/topology/:id/visualization" [:as {:keys [cookies servlet-request]} id & m]
(json-response (mk-visualization-data id (:window m) (check-include-sys? (:sys m))) (:callback m)))
(GET "/api/v1/topology/:id/component/:component" [id component & m]
- (let [id (url-decode id)
- component (url-decode component)]
- (json-response (component-page id component (:window m) (check-include-sys? (:sys m))) (:callback m))))
+ (json-response (component-page id component (:window m) (check-include-sys? (:sys m))) (:callback m)))
(POST "/api/v1/topology/:id/activate" [id]
(with-nimbus nimbus
- (let [id (url-decode id)
- tplg (.getTopologyInfo ^Nimbus$Client nimbus id)
+ (let [tplg (.getTopologyInfo ^Nimbus$Client nimbus id)
name (.get_name tplg)]
(.activate nimbus name)
(log-message "Activating topology '" name "'")))
- (resp/redirect (str "/api/v1/topology/" id)))
+ (resp/redirect (str "/api/v1/topology/" (url-encode id))))
(POST "/api/v1/topology/:id/deactivate" [id]
(with-nimbus nimbus
- (let [id (url-decode id)
- tplg (.getTopologyInfo ^Nimbus$Client nimbus id)
+ (let [tplg (.getTopologyInfo ^Nimbus$Client nimbus id)
name (.get_name tplg)]
(.deactivate nimbus name)
(log-message "Deactivating topology '" name "'")))
- (resp/redirect (str "/api/v1/topology/" id)))
+ (resp/redirect (str "/api/v1/topology/" (url-encode id))))
(POST "/api/v1/topology/:id/rebalance/:wait-time" [id wait-time]
(with-nimbus nimbus
- (let [id (url-decode id)
- tplg (.getTopologyInfo ^Nimbus$Client nimbus id)
+ (let [tplg (.getTopologyInfo ^Nimbus$Client nimbus id)
name (.get_name tplg)
options (RebalanceOptions.)]
(.set_wait_secs options (Integer/parseInt wait-time))
(.rebalance nimbus name options)
(log-message "Rebalancing topology '" name "' with wait time: " wait-time " secs")))
- (resp/redirect (str "/api/v1/topology/" id)))
+ (resp/redirect (str "/api/v1/topology/" (url-encode id))))
(POST "/api/v1/topology/:id/kill/:wait-time" [id wait-time]
(with-nimbus nimbus
- (let [id (url-decode id)
- tplg (.getTopologyInfo ^Nimbus$Client nimbus id)
+ (let [tplg (.getTopologyInfo ^Nimbus$Client nimbus id)
name (.get_name tplg)
options (KillOptions.)]
(.set_wait_secs options (Integer/parseInt wait-time))
(.killTopologyWithOpts nimbus name options)
(log-message "Killing topology '" name "' with wait time: " wait-time " secs")))
- (resp/redirect (str "/api/v1/topology/" id)))
+ (resp/redirect (str "/api/v1/topology/" (url-encode id))))
(GET "/" [:as {cookies :cookies}]
(resp/redirect "/index.html"))
diff --git a/storm-core/src/dev/resources/storm.py b/storm-core/src/dev/resources/storm.py
deleted file mode 120000
index 5e73111..0000000
--- a/storm-core/src/dev/resources/storm.py
+++ /dev/null
@@ -1 +0,0 @@
-../../multilang/py/storm.py
\ No newline at end of file
diff --git a/storm-core/src/dev/resources/storm.py b/storm-core/src/dev/resources/storm.py
new file mode 100755
index 0000000..d2a3082
--- /dev/null
+++ b/storm-core/src/dev/resources/storm.py
@@ -0,0 +1,247 @@
+# -*- 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())))
+
+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()
+ 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()
+ ANCHOR_TUPLE = tup
+ self.process(tup)
+ ack(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))
diff --git a/storm-core/src/dev/resources/storm.rb b/storm-core/src/dev/resources/storm.rb
deleted file mode 120000
index 96db018..0000000
--- a/storm-core/src/dev/resources/storm.rb
+++ /dev/null
@@ -1 +0,0 @@
-../../multilang/rb/storm.rb
\ No newline at end of file
diff --git a/storm-core/src/dev/resources/storm.rb b/storm-core/src/dev/resources/storm.rb
new file mode 100644
index 0000000..17232d1
--- /dev/null
+++ b/storm-core/src/dev/resources/storm.rb
@@ -0,0 +1,227 @@
+# -*- 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.
+
+require "rubygems"
+require "json"
+
+module Storm
+ module Protocol
+ class << self
+ attr_accessor :mode, :pending_taskids, :pending_commands
+ end
+
+ self.pending_taskids = []
+ self.pending_commands = []
+
+ def read_message
+ msg = ""
+ loop do
+ line = STDIN.readline.chomp
+ break if line == "end"
+ msg << line
+ msg << "\n"
+ end
+ JSON.parse msg.chomp
+ end
+
+ def read_task_ids
+ Storm::Protocol.pending_taskids.shift ||
+ begin
+ msg = read_message
+ until msg.is_a? Array
+ Storm::Protocol.pending_commands.push(msg)
+ msg = read_message
+ end
+ msg
+ end
+ end
+
+ def read_command
+ Storm::Protocol.pending_commands.shift ||
+ begin
+ msg = read_message
+ while msg.is_a? Array
+ Storm::Protocol.pending_taskids.push(msg)
+ msg = read_message
+ end
+ msg
+ end
+ end
+
+ def send_msg_to_parent(msg)
+ puts msg.to_json
+ puts "end"
+ STDOUT.flush
+ end
+
+ def sync
+ send_msg_to_parent({'command' => 'sync'})
+ end
+
+ def send_pid(heartbeat_dir)
+ pid = Process.pid
+ send_msg_to_parent({'pid' => pid})
+ File.open("#{heartbeat_dir}/#{pid}", "w").close
+ end
+
+ def emit_bolt(tup, args = {})
+ stream = args[:stream]
+ anchors = args[:anchors] || args[:anchor] || []
+ anchors = [anchors] unless anchors.is_a? Enumerable
+ direct = args[:direct_task]
+ m = {:command => :emit, :anchors => anchors.map(&:id), :tuple => tup}
+ m[:stream] = stream if stream
+ m[:task] = direct if direct
+ send_msg_to_parent m
+ read_task_ids unless direct
+ end
+
+ def emit_spout(tup, args = {})
+ stream = args[:stream]
+ id = args[:id]
+ direct = args[:direct_task]
+ m = {:command => :emit, :tuple => tup}
+ m[:id] = id if id
+ m[:stream] = stream if stream
+ m[:task] = direct if direct
+ send_msg_to_parent m
+ read_task_ids unless direct
+ end
+
+ def emit(*args)
+ case Storm::Protocol.mode
+ when 'spout'
+ emit_spout(*args)
+ when 'bolt'
+ emit_bolt(*args)
+ end
+ end
+
+ def ack(tup)
+ send_msg_to_parent :command => :ack, :id => tup.id
+ end
+
+ def fail(tup)
+ send_msg_to_parent :command => :fail, :id => tup.id
+ end
+
+ def reportError(msg)
+ send_msg_to_parent :command => :error, :msg => msg.to_s
+ end
+
+ def log(msg, level=2)
+ send_msg_to_parent :command => :log, :msg => msg.to_s, :level => level
+ end
+
+ def logTrace(msg)
+ log(msg, 0)
+ end
+
+ def logDebug(msg)
+ log(msg, 1)
+ end
+
+ def logInfo(msg)
+ log(msg, 2)
+ end
+
+ def logWarn(msg)
+ log(msg, 3)
+ end
+
+ def logError(msg)
+ log(msg, 4)
+ end
+
+ def handshake
+ setup_info = read_message
+ send_pid setup_info['pidDir']
+ [setup_info['conf'], setup_info['context']]
+ end
+ end
+
+ class Tuple
+ attr_accessor :id, :component, :stream, :task, :values
+
+ def initialize(id, component, stream, task, values)
+ @id = id
+ @component = component
+ @stream = stream
+ @task = task
+ @values = values
+ end
+
+ def self.from_hash(hash)
+ Tuple.new(*hash.values_at("id", "comp", "stream", "task", "tuple"))
+ end
+ end
+
+ class Bolt
+ include Storm::Protocol
+
+ def prepare(conf, context); end
+
+ def process(tuple); end
+
+ def run
+ Storm::Protocol.mode = 'bolt'
+ prepare(*handshake)
+ begin
+ while true
+ process Tuple.from_hash(read_command)
+ end
+ rescue Exception => e
+ reportError 'Exception in bolt: ' + e.message + ' - ' + e.backtrace.join('\n')
+ end
+ end
+ end
+
+ class Spout
+ include Storm::Protocol
+
+ def open(conf, context); end
+
+ def nextTuple; end
+
+ def ack(id); end
+
+ def fail(id); end
+
+ def run
+ Storm::Protocol.mode = 'spout'
+ open(*handshake)
+
+ begin
+ while true
+ msg = read_command
+ case msg['command']
+ when 'next'
+ nextTuple
+ when 'ack'
+ ack(msg['id'])
+ when 'fail'
+ fail(msg['id'])
+ end
+ sync
+ end
+ rescue Exception => e
+ reportError 'Exception in spout: ' + e.message + ' - ' + e.backtrace.join('\n')
+ end
+ end
+ end
+end
diff --git a/storm-core/src/multilang/js/storm.js b/storm-core/src/multilang/js/storm.js
new file mode 100755
index 0000000..5c78072
--- /dev/null
+++ b/storm-core/src/multilang/js/storm.js
@@ -0,0 +1,349 @@
+/**
+ * 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"]);
+ 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;
diff --git a/storm-core/src/ui/public/component.html b/storm-core/src/ui/public/component.html
index 6fa998f..fddb9ef 100644
--- a/storm-core/src/ui/public/component.html
+++ b/storm-core/src/ui/public/component.html
@@ -24,7 +24,7 @@
<script src="/js/jquery.tablesorter.min.js" type="text/javascript"></script>
<script src="/js/jquery.cookies.2.2.0.min.js" type="text/javascript"></script>
<script src="/js/jquery.mustache.js" type="text/javascript"></script>
-<script src="/js/purl.js" type="text/javascript"></script>
+<script src="/js/url.min.js" type="text/javascript"></script>
<script src="/js/bootstrap-twipsy.js" type="text/javascript"></script>
<script src="/js/jquery.blockUI.min.js" type="text/javascript"></script>
<script src="/js/script.js" type="text/javascript"></script>
@@ -53,9 +53,9 @@
$.blockUI({ message: '<img src="images/spinner.gif" /> <h3>Loading component summary...</h3>'});
});
$(document).ready(function() {
- var componentId = $.url().param("id");
- var topologyId = $.url().param("topology_id");
- var window = $.url().param("window");
+ var componentId = $.url("?id");
+ var topologyId = $.url("?topology_id");
+ var window = $.url("?window");
var sys = $.cookies.get("sys") || "false";
var url = "/api/v1/topology/"+topologyId+"/component/"+componentId+"?sys="+sys;
if(window) url += "&window="+window;
diff --git a/storm-core/src/ui/public/js/purl.js b/storm-core/src/ui/public/js/purl.js
deleted file mode 100644
index b5799c6..0000000
--- a/storm-core/src/ui/public/js/purl.js
+++ /dev/null
@@ -1,267 +0,0 @@
-/*
- * Purl (A JavaScript URL parser) v2.3.1
- * Developed and maintanined by Mark Perkins, mark@allmarkedup.com
- * Source repository: https://github.com/allmarkedup/jQuery-URL-Parser
- * Licensed under an MIT-style license. See https://github.com/allmarkedup/jQuery-URL-Parser/blob/master/LICENSE for details.
- */
-
-;(function(factory) {
- if (typeof define === 'function' && define.amd) {
- define(factory);
- } else {
- window.purl = factory();
- }
-})(function() {
-
- var tag2attr = {
- a : 'href',
- img : 'src',
- form : 'action',
- base : 'href',
- script : 'src',
- iframe : 'src',
- link : 'href',
- embed : 'src',
- object : 'data'
- },
-
- key = ['source', 'protocol', 'authority', 'userInfo', 'user', 'password', 'host', 'port', 'relative', 'path', 'directory', 'file', 'query', 'fragment'], // keys available to query
-
- aliases = { 'anchor' : 'fragment' }, // aliases for backwards compatability
-
- parser = {
- strict : /^(?:([^:\/?#]+):)?(?:\/\/((?:(([^:@]*):?([^:@]*))?@)?([^:\/?#]*)(?::(\d*))?))?((((?:[^?#\/]*\/)*)([^?#]*))(?:\?([^#]*))?(?:#(.*))?)/, //less intuitive, more accurate to the specs
- loose : /^(?:(?![^:@]+:[^:@\/]*@)([^:\/?#.]+):)?(?:\/\/)?((?:(([^:@]*):?([^:@]*))?@)?([^:\/?#]*)(?::(\d*))?)(((\/(?:[^?#](?![^?#\/]*\.[^?#\/.]+(?:[?#]|$)))*\/?)?([^?#\/]*))(?:\?([^#]*))?(?:#(.*))?)/ // more intuitive, fails on relative paths and deviates from specs
- },
-
- isint = /^[0-9]+$/;
-
- function parseUri( url, strictMode ) {
- var str = decodeURI( url ),
- res = parser[ strictMode || false ? 'strict' : 'loose' ].exec( str ),
- uri = { attr : {}, param : {}, seg : {} },
- i = 14;
-
- while ( i-- ) {
- uri.attr[ key[i] ] = res[i] || '';
- }
-
- // build query and fragment parameters
- uri.param['query'] = parseString(uri.attr['query']);
- uri.param['fragment'] = parseString(uri.attr['fragment']);
-
- // split path and fragement into segments
- uri.seg['path'] = uri.attr.path.replace(/^\/+|\/+$/g,'').split('/');
- uri.seg['fragment'] = uri.attr.fragment.replace(/^\/+|\/+$/g,'').split('/');
-
- // compile a 'base' domain attribute
- uri.attr['base'] = uri.attr.host ? (uri.attr.protocol ? uri.attr.protocol+'://'+uri.attr.host : uri.attr.host) + (uri.attr.port ? ':'+uri.attr.port : '') : '';
-
- return uri;
- }
-
- function getAttrName( elm ) {
- var tn = elm.tagName;
- if ( typeof tn !== 'undefined' ) return tag2attr[tn.toLowerCase()];
- return tn;
- }
-
- function promote(parent, key) {
- if (parent[key].length === 0) return parent[key] = {};
- var t = {};
- for (var i in parent[key]) t[i] = parent[key][i];
- parent[key] = t;
- return t;
- }
-
- function parse(parts, parent, key, val) {
- var part = parts.shift();
- if (!part) {
- if (isArray(parent[key])) {
- parent[key].push(val);
- } else if ('object' == typeof parent[key]) {
- parent[key] = val;
- } else if ('undefined' == typeof parent[key]) {
- parent[key] = val;
- } else {
- parent[key] = [parent[key], val];
- }
- } else {
- var obj = parent[key] = parent[key] || [];
- if (']' == part) {
- if (isArray(obj)) {
- if ('' !== val) obj.push(val);
- } else if ('object' == typeof obj) {
- obj[keys(obj).length] = val;
- } else {
- obj = parent[key] = [parent[key], val];
- }
- } else if (~part.indexOf(']')) {
- part = part.substr(0, part.length - 1);
- if (!isint.test(part) && isArray(obj)) obj = promote(parent, key);
- parse(parts, obj, part, val);
- // key
- } else {
- if (!isint.test(part) && isArray(obj)) obj = promote(parent, key);
- parse(parts, obj, part, val);
- }
- }
- }
-
- function merge(parent, key, val) {
- if (~key.indexOf(']')) {
- var parts = key.split('[');
- parse(parts, parent, 'base', val);
- } else {
- if (!isint.test(key) && isArray(parent.base)) {
- var t = {};
- for (var k in parent.base) t[k] = parent.base[k];
- parent.base = t;
- }
- if (key !== '') {
- set(parent.base, key, val);
- }
- }
- return parent;
- }
-
- function parseString(str) {
- return reduce(String(str).split(/&|;/), function(ret, pair) {
- try {
- pair = decodeURIComponent(pair.replace(/\+/g, ' '));
- } catch(e) {
- // ignore
- }
- var eql = pair.indexOf('='),
- brace = lastBraceInKey(pair),
- key = pair.substr(0, brace || eql),
- val = pair.substr(brace || eql, pair.length);
-
- val = val.substr(val.indexOf('=') + 1, val.length);
-
- if (key === '') {
- key = pair;
- val = '';
- }
-
- return merge(ret, key, val);
- }, { base: {} }).base;
- }
-
- function set(obj, key, val) {
- var v = obj[key];
- if (typeof v === 'undefined') {
- obj[key] = val;
- } else if (isArray(v)) {
- v.push(val);
- } else {
- obj[key] = [v, val];
- }
- }
-
- function lastBraceInKey(str) {
- var len = str.length,
- brace,
- c;
- for (var i = 0; i < len; ++i) {
- c = str[i];
- if (']' == c) brace = false;
- if ('[' == c) brace = true;
- if ('=' == c && !brace) return i;
- }
- }
-
- function reduce(obj, accumulator){
- var i = 0,
- l = obj.length >> 0,
- curr = arguments[2];
- while (i < l) {
- if (i in obj) curr = accumulator.call(undefined, curr, obj[i], i, obj);
- ++i;
- }
- return curr;
- }
-
- function isArray(vArg) {
- return Object.prototype.toString.call(vArg) === "[object Array]";
- }
-
- function keys(obj) {
- var key_array = [];
- for ( var prop in obj ) {
- if ( obj.hasOwnProperty(prop) ) key_array.push(prop);
- }
- return key_array;
- }
-
- function purl( url, strictMode ) {
- if ( arguments.length === 1 && url === true ) {
- strictMode = true;
- url = undefined;
- }
- strictMode = strictMode || false;
- url = url || window.location.toString();
-
- return {
-
- data : parseUri(url, strictMode),
-
- // get various attributes from the URI
- attr : function( attr ) {
- attr = aliases[attr] || attr;
- return typeof attr !== 'undefined' ? this.data.attr[attr] : this.data.attr;
- },
-
- // return query string parameters
- param : function( param ) {
- return typeof param !== 'undefined' ? this.data.param.query[param] : this.data.param.query;
- },
-
- // return fragment parameters
- fparam : function( param ) {
- return typeof param !== 'undefined' ? this.data.param.fragment[param] : this.data.param.fragment;
- },
-
- // return path segments
- segment : function( seg ) {
- if ( typeof seg === 'undefined' ) {
- return this.data.seg.path;
- } else {
- seg = seg < 0 ? this.data.seg.path.length + seg : seg - 1; // negative segments count from the end
- return this.data.seg.path[seg];
- }
- },
-
- // return fragment segments
- fsegment : function( seg ) {
- if ( typeof seg === 'undefined' ) {
- return this.data.seg.fragment;
- } else {
- seg = seg < 0 ? this.data.seg.fragment.length + seg : seg - 1; // negative segments count from the end
- return this.data.seg.fragment[seg];
- }
- }
-
- };
-
- }
-
- purl.jQuery = function($){
- if ($ != null) {
- $.fn.url = function( strictMode ) {
- var url = '';
- if ( this.length ) {
- url = $(this).attr( getAttrName(this[0]) ) || '';
- }
- return purl( url, strictMode );
- };
-
- $.url = purl;
- }
- };
-
- purl.jQuery(window.jQuery);
-
- return purl;
-
-});
diff --git a/storm-core/src/ui/public/js/script.js b/storm-core/src/ui/public/js/script.js
index 46354b8..12c7fca 100644
--- a/storm-core/src/ui/public/js/script.js
+++ b/storm-core/src/ui/public/js/script.js
@@ -140,9 +140,10 @@
}
}
-function topologyActionJson(id,name,status,msgTimeout) {
+function topologyActionJson(id, encodedId, name,status,msgTimeout) {
var jsonData = {};
jsonData["id"] = id;
+ jsonData["encodedId"] = encodedId;
jsonData["name"] = name;
jsonData["msgTimeout"] = msgTimeout;
jsonData["activateStatus"] = (status === "ACTIVE") ? "disabled" : "enabled";
diff --git a/storm-core/src/ui/public/js/url.min.js b/storm-core/src/ui/public/js/url.min.js
new file mode 100755
index 0000000..8057e0a
--- /dev/null
+++ b/storm-core/src/ui/public/js/url.min.js
@@ -0,0 +1 @@
+/*! url - v1.8.6 - 2013-11-22 */window.url=function(){function a(a){return!isNaN(parseFloat(a))&&isFinite(a)}return function(b,c){var d=c||window.location.toString();if(!b)return d;b=b.toString(),"//"===d.substring(0,2)?d="http:"+d:1===d.split("://").length&&(d="http://"+d),c=d.split("/");var e={auth:""},f=c[2].split("@");1===f.length?f=f[0].split(":"):(e.auth=f[0],f=f[1].split(":")),e.protocol=c[0],e.hostname=f[0],e.port=f[1]||("https"===e.protocol.split(":")[0].toLowerCase()?"443":"80"),e.pathname=(c.length>3?"/":"")+c.slice(3,c.length).join("/").split("?")[0].split("#")[0];var g=e.pathname;"/"===g.charAt(g.length-1)&&(g=g.substring(0,g.length-1));var h=e.hostname,i=h.split("."),j=g.split("/");if("hostname"===b)return h;if("domain"===b)return/^(([0-9]|[1-9][0-9]|1[0-9]{2}|2[0-4][0-9]|25[0-5])\.){3}([0-9]|[1-9][0-9]|1[0-9]{2}|2[0-4][0-9]|25[0-5])$/.test(h)?h:i.slice(-2).join(".");if("sub"===b)return i.slice(0,i.length-2).join(".");if("port"===b)return e.port;if("protocol"===b)return e.protocol.split(":")[0];if("auth"===b)return e.auth;if("user"===b)return e.auth.split(":")[0];if("pass"===b)return e.auth.split(":")[1]||"";if("path"===b)return e.pathname;if("."===b.charAt(0)){if(b=b.substring(1),a(b))return b=parseInt(b,10),i[0>b?i.length+b:b-1]||""}else{if(a(b))return b=parseInt(b,10),j[0>b?j.length+b:b]||"";if("file"===b)return j.slice(-1)[0];if("filename"===b)return j.slice(-1)[0].split(".")[0];if("fileext"===b)return j.slice(-1)[0].split(".")[1]||"";if("?"===b.charAt(0)||"#"===b.charAt(0)){var k=d,l=null;if("?"===b.charAt(0)?k=(k.split("?")[1]||"").split("#")[0]:"#"===b.charAt(0)&&(k=k.split("#")[1]||""),!b.charAt(1))return k;b=b.substring(1),k=k.split("&");for(var m=0,n=k.length;n>m;m++)if(l=k[m].split("="),l[0]===b)return l[1]||"";return null}}return""}}(),"undefined"!=typeof jQuery&&jQuery.extend({url:function(a,b){return window.url(a,b)}});
\ No newline at end of file
diff --git a/storm-core/src/ui/public/js/visualization.js b/storm-core/src/ui/public/js/visualization.js
index 7cec697..1e1c4d6 100644
--- a/storm-core/src/ui/public/js/visualization.js
+++ b/storm-core/src/ui/public/js/visualization.js
@@ -392,7 +392,7 @@
var update = function(should_rechoose) {
if(should_update) {
$.ajax({
- url: "/api/v1/topology/" + $.url().param("id") + "/visualization",
+ url: "/api/v1/topology/"+$.url("?id")+"/visualization",
success: function (data, status, jqXHR) {
topology_data = data;
update_data(topology_data, sys);
diff --git a/storm-core/src/ui/public/templates/component-page-template.html b/storm-core/src/ui/public/templates/component-page-template.html
index f718967..1e916e3 100644
--- a/storm-core/src/ui/public/templates/component-page-template.html
+++ b/storm-core/src/ui/public/templates/component-page-template.html
@@ -44,7 +44,7 @@
<tbody>
<tr>
<td>{{id}}</td>
- <td><a href="/topology.html?id={{topologyId}}">{{name}}</a></td>
+ <td><a href="/topology.html?id={{encodedTopologyId}}">{{name}}</a></td>
<td>{{executors}}</td>
<td>{{tasks}}</td>
</tbody>
@@ -90,7 +90,7 @@
<tbody>
{{#spoutSummary}}
<tr>
- <td><a href="/component.html?id={{id}}&topology_id={{topologyId}}&window={{window}}">{{windowPretty}}</a></td>
+ <td><a href="/component.html?id={{encodedId}}&topology_id={{encodedTopologyId}}&window={{window}}">{{windowPretty}}</a></td>
<td>{{transferred}}</td>
<td>{{emitted}}</td>
<td>{{completeLatency}}</td>
@@ -271,7 +271,7 @@
<tbody>
{{#boltStats}}
<tr>
- <td><a href="/component.html?id={{id}}&topology_id={{topologyId}}&window={{window}}">{{windowPretty}}</a></td>
+ <td><a href="/component.html?id={{encodedId}}&topology_id={{encodedTopologyId}}&window={{window}}">{{windowPretty}}</a></td>
<td>{{emitted}}</td>
<td>{{transferred}}</td>
<td>{{executeLatency}}</td>
diff --git a/storm-core/src/ui/public/templates/index-page-template.html b/storm-core/src/ui/public/templates/index-page-template.html
index d249fac..e3aab5e 100644
--- a/storm-core/src/ui/public/templates/index-page-template.html
+++ b/storm-core/src/ui/public/templates/index-page-template.html
@@ -118,7 +118,7 @@
<tbody>
{{#topologies}}
<tr>
- <td><a href="/topology.html?id={{id}}">{{name}}</a></td>
+ <td><a href="/topology.html?id={{encodedId}}">{{name}}</a></td>
<td>{{id}}</td>
<td>{{status}}</td>
<td>{{uptime}}</td>
diff --git a/storm-core/src/ui/public/templates/topology-page-template.html b/storm-core/src/ui/public/templates/topology-page-template.html
index b977071..3479fc1 100644
--- a/storm-core/src/ui/public/templates/topology-page-template.html
+++ b/storm-core/src/ui/public/templates/topology-page-template.html
@@ -108,7 +108,7 @@
<tbody>
{{#topologyStats}}
<tr>
- <td><a href="/topology.html?id={{id}}&window={{window}}">{{windowPretty}}</td>
+ <td><a href="/topology.html?id={{encodedId}}&window={{window}}">{{windowPretty}}</td>
<td>{{emitted}}</td>
<td>{{transferred}}</td>
<td>{{completeLatency}}</td>
@@ -219,7 +219,7 @@
<tbody>
{{#spouts}}
<tr>
- <td><a href="/component.html?id={{spoutId}}&topology_id={{id}}">{{spoutId}}</a></td>
+ <td><a href="/component.html?id={{encodedSpoutId}}&topology_id={{encodedId}}">{{spoutId}}</a></td>
<td>{{executors}}</td>
<td>{{tasks}}</td>
<td>{{emitted}}</td>
@@ -305,7 +305,7 @@
<tbody>
{{#bolts}}
<tr>
- <td><a href="/component.html?id={{boltId}}&topology_id={{id}}">{{boltId}}</a></td>
+ <td><a href="/component.html?id={{encodedBoltId}}&topology_id={{encodedId}}">{{boltId}}</a></td>
<td>{{executors}}</td>
<td>{{tasks}}</td>
<td>{{emitted}}</td>
@@ -326,8 +326,8 @@
</script>
<script id="topology-actions-template" type="text/html">
- <input {{activateStatus}} onclick="confirmAction('{{id}}', '{{name}}', 'activate', false, 0)" type="button" value="Activate">
- <input {{deactivateStatus}} onclick="confirmAction('{{id}}', '{{name}}', 'deactivate', false, 0)" type="button" value="Deactivate">
- <input {{rebalanceStatus}} onclick="confirmAction('{{id}}', '{{name}}', 'rebalance', true, {{msgTimeout}})" type="button" value="Rebalance">
- <input {{killStatus}} onclick="confirmAction('{{id}}', '{{name}}', 'kill', true, 30)" type="button" value="Kill">
+ <input {{activateStatus}} onclick="confirmAction('{{encodedId}}', '{{name}}', 'activate', false, 0)" type="button" value="Activate">
+ <input {{deactivateStatus}} onclick="confirmAction('{{encodedId}}', '{{name}}', 'deactivate', false, 0)" type="button" value="Deactivate">
+ <input {{rebalanceStatus}} onclick="confirmAction('{{encodedId}}', '{{name}}', 'rebalance', true, {{msgTimeout}})" type="button" value="Rebalance">
+ <input {{killStatus}} onclick="confirmAction('{{encodedId}}', '{{name}}', 'kill', true, 30)" type="button" value="Kill">
</script>
diff --git a/storm-core/src/ui/public/topology.html b/storm-core/src/ui/public/topology.html
index 477d3b0..a6598ef 100644
--- a/storm-core/src/ui/public/topology.html
+++ b/storm-core/src/ui/public/topology.html
@@ -25,7 +25,7 @@
<script src="/js/jquery.tablesorter.min.js" type="text/javascript"></script>
<script src="/js/jquery.cookies.2.2.0.min.js" type="text/javascript"></script>
<script src="/js/jquery.mustache.js" type="text/javascript"></script>
-<script src="/js/purl.js" type="text/javascript"></script>
+<script src="/js/url.min.js" type="text/javascript"></script>
<script src="/js/bootstrap-twipsy.js" type="text/javascript"></script>
<script src="/js/jquery.blockUI.min.js" type="text/javascript"></script>
<script src="/js/script.js" type="text/javascript"></script>
@@ -64,8 +64,8 @@
}
});
$(document).ready(function() {
- var topologyId = $.url().param("id");
- var window = $.url().param("window");
+ var topologyId = $.url("?id");
+ var window = $.url("?window");
var sys = $.cookies.get("sys") || "false";
var url = "/api/v1/topology/"+topologyId+"?sys="+sys;
if(window) url += "&window="+window;
@@ -88,7 +88,7 @@
var topologyActions = $("#topology-actions");
var topologyVisualization = $("#topology-visualization")
var formattedConfig = formatConfigData(response["configuration"]);
- var buttonJsonData = topologyActionJson(response["id"],response["name"],response["status"],response["msgTimeout"]);
+ var buttonJsonData = topologyActionJson(response["id"],response["encodedId"],response["name"],response["status"],response["msgTimeout"]);
$.get("/templates/topology-page-template.html", function(template) {
topologySummary.append(Mustache.render($(template).filter("#topology-summary-template").html(),response));
topologyActions.append(Mustache.render($(template).filter("#topology-actions-template").html(),buttonJsonData));
diff --git a/storm-core/test/clj/backtype/storm/supervisor_test.clj b/storm-core/test/clj/backtype/storm/supervisor_test.clj
index a61a2ae..ba74d88 100644
--- a/storm-core/test/clj/backtype/storm/supervisor_test.clj
+++ b/storm-core/test/clj/backtype/storm/supervisor_test.clj
@@ -247,7 +247,7 @@
(let [mock-port "42"
mock-storm-id "fake-storm-id"
mock-worker-id "fake-worker-id"
- mock-cp "/base:/stormjar.jar"
+ mock-cp (str file-path-separator "base" class-path-separator file-path-separator "stormjar.jar")
exp-args-fn (fn [opts topo-opts classpath]
(concat [(supervisor/java-cmd) "-server"]
opts
@@ -255,10 +255,8 @@
["-Djava.library.path="
(str "-Dlogfile.name=worker-" mock-port ".log")
"-Dstorm.home="
- "-Dstorm.conf.file="
- "-Dstorm.options="
- "-Dstorm.log.dir=/logs"
- "-Dlogback.configurationFile=/logback/cluster.xml"
+ (str "-Dstorm.log.dir=" file-path-separator "logs")
+ (str "-Dlogback.configurationFile=" file-path-separator "logback" file-path-separator "cluster.xml")
(str "-Dstorm.id=" mock-storm-id)
(str "-Dworker.id=" mock-worker-id)
(str "-Dworker.port=" mock-port)
@@ -308,14 +306,14 @@
[0]
exp-args))))
(testing "testing topology.classpath is added to classpath"
- (let [topo-cp "/any/path"
+ (let [topo-cp (str file-path-separator "any" file-path-separator "path")
exp-args (exp-args-fn [] [] (add-to-classpath mock-cp [topo-cp]))
mock-supervisor {:conf {STORM-CLUSTER-MODE :distributed}}]
(stubbing [read-supervisor-storm-conf {TOPOLOGY-CLASSPATH topo-cp}
supervisor-stormdist-root nil
supervisor/jlp nil
launch-process nil
- current-classpath "/base"]
+ current-classpath (str file-path-separator "base")]
(supervisor/launch-worker mock-supervisor
mock-storm-id
mock-port
@@ -331,7 +329,7 @@
supervisor-stormdist-root nil
supervisor/jlp nil
launch-process nil
- current-classpath "/base"]
+ current-classpath (str file-path-separator "base")]
(supervisor/launch-worker mock-supervisor
mock-storm-id
mock-port
@@ -471,4 +469,4 @@
(validate-launched-once (:launched changed)
{"sup1" [3 4]}
(get-storm-id (:storm-cluster-state cluster) "topology2"))
- )))
+ )))
\ No newline at end of file
diff --git a/storm-dist/binary/LICENSE b/storm-dist/binary/LICENSE
index f6dde01..5ad9e5d 100644
--- a/storm-dist/binary/LICENSE
+++ b/storm-dist/binary/LICENSE
@@ -413,13 +413,11 @@
-----------------------------------------------------------------------
-For Purl (storm-core/src/ui/public/js/purl.js)
+For Jquery url plugin (storm-core/src/ui/public/js/url.min.js)
-Purl (A JavaScript URL parser) v2.3.1
-Developed and maintanined by Mark Perkins, mark@allmarkedup.com
-Source repository: https://github.com/allmarkedup/jQuery-URL-Parser
-Licensed under an MIT-style license. See https://github.com/allmarkedup/jQuery-URL-Parser/blob/master/LICENSE for details.
-
+Jquery Url (A Jquery plugin for URL parser) v1.8.6
+Source repository: https://github.com/websanova/js-url
+Licensed under an MIT-style license. Seehttps://github.com/websanova/js-url#license for details.
-----------------------------------------------------------------------
