You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by kn...@apache.org on 2016/04/04 18:18:41 UTC
[4/6] storm git commit: Clean up macros in storm starter
Clean up macros in storm starter
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/1ed04f14
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/1ed04f14
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/1ed04f14
Branch: refs/heads/master
Commit: 1ed04f14e54efd4b9fbaac4f9e4b73869961ed20
Parents: 3b7c4a8
Author: Bryant <br...@gmail.com>
Authored: Thu Mar 31 18:24:48 2016 -0700
Committer: Bryant <br...@gmail.com>
Committed: Thu Mar 31 20:40:08 2016 -0700
----------------------------------------------------------------------
.../clj/org/apache/storm/starter/clj/bolts.clj | 20 ++++---
.../org/apache/storm/starter/clj/bolts_test.clj | 59 ++++++++++----------
2 files changed, 41 insertions(+), 38 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/1ed04f14/examples/storm-starter/src/clj/org/apache/storm/starter/clj/bolts.clj
----------------------------------------------------------------------
diff --git a/examples/storm-starter/src/clj/org/apache/storm/starter/clj/bolts.clj b/examples/storm-starter/src/clj/org/apache/storm/starter/clj/bolts.clj
index 95aa8bf..270952f 100644
--- a/examples/storm-starter/src/clj/org/apache/storm/starter/clj/bolts.clj
+++ b/examples/storm-starter/src/clj/org/apache/storm/starter/clj/bolts.clj
@@ -44,12 +44,12 @@
(.incrementCount counter word)
(ack! collector tuple)))))))
-(defmacro update-rankings [& body]
- `(if (TupleUtils/isTick ~'tuple)
+(defmacro update-rankings [tuple collector rankings & body]
+ `(if (TupleUtils/isTick ~tuple)
(do
(log-debug "Received tick tuple, triggering emit of current rankings")
- (emit-bolt! ~'collector [(.copy ~'rankings)])
- (log-debug "Rankings: " ~'rankings))
+ (emit-bolt! ~collector [(.copy ~rankings)])
+ (log-debug "Rankings: " ~rankings))
~@body))
(defbolt intermediate-rankings-bolt ["rankings"]
@@ -60,7 +60,9 @@
(let [rankings (Rankings. top-n)]
(bolt
(execute [tuple]
- (update-rankings (.updateWith rankings (RankableObjectWithFields/from tuple)))))))
+ (update-rankings
+ tuple collector rankings
+ (.updateWith rankings (RankableObjectWithFields/from tuple)))))))
(defbolt total-rankings-bolt ["rankings"]
{:prepare true
@@ -70,6 +72,8 @@
(let [rankings (Rankings. top-n)]
(bolt
(execute [{rankings-to-merge "rankings" :as tuple}]
- (update-rankings (doto rankings
- (.updateWith rankings-to-merge)
- (.pruneZeroCounts)))))))
+ (update-rankings
+ tuple collector rankings
+ (doto rankings
+ (.updateWith rankings-to-merge)
+ (.pruneZeroCounts)))))))
http://git-wip-us.apache.org/repos/asf/storm/blob/1ed04f14/examples/storm-starter/test/clj/org/apache/storm/starter/clj/bolts_test.clj
----------------------------------------------------------------------
diff --git a/examples/storm-starter/test/clj/org/apache/storm/starter/clj/bolts_test.clj b/examples/storm-starter/test/clj/org/apache/storm/starter/clj/bolts_test.clj
index d18f19b..164e7ac 100644
--- a/examples/storm-starter/test/clj/org/apache/storm/starter/clj/bolts_test.clj
+++ b/examples/storm-starter/test/clj/org/apache/storm/starter/clj/bolts_test.clj
@@ -25,20 +25,18 @@
[org.apache.storm.starter.tools Rankable]
[org.apache.storm.tuple Tuple]))
-(defmacro test-with-tuple [[bolt tuples] & body]
- `(let [bolt# ~bolt
- tuples# (atom [])]
- (.prepare bolt# {} nil (OutputCollector.
- (reify IOutputCollector
- (emit [_ _ _ tuple#]
- (swap! tuples# conj tuple#))
- (ack [_ input]))))
- (if (vector? ~tuples)
- (doseq [t# ~tuples]
- (.execute bolt# t#))
- (.execute bolt# ~tuples))
- (let [~'tuples @tuples#]
- ~@body)))
+(defn execute-tuples [bolt tuples]
+ (let [out (atom [])]
+ (.prepare bolt {} nil (OutputCollector.
+ (reify IOutputCollector
+ (emit [_ _ _ tuple]
+ (swap! out conj tuple))
+ (ack [_ input]))))
+ (if (vector? tuples)
+ (doseq [t tuples]
+ (.execute bolt t))
+ (.execute bolt tuples))
+ @out))
(defn- mock-tuple [m & {component :component stream-id :stream-id
:or {component "1" stream-id "1"}}]
@@ -65,38 +63,39 @@
(deftest test-split-sentence
(testing "Bolt emits word per sentence"
- (test-with-tuple [split-sentence (mock-tuple
- {"sentence" "the cat jumped over the door"})]
+ (let [tuples (execute-tuples
+ split-sentence
+ (mock-tuple {"sentence" "the cat jumped over the door"}))]
(is (= [["the"] ["cat"] ["jumped"] ["over"] ["the"] ["door"]] tuples)))))
(deftest test-word-count
(testing "Bolt emits new count"
- (test-with-tuple [word-count [(mock-tuple {"word" "the"})
- (mock-tuple {"word" "the"})
- (mock-tuple {"word" "cat"})]]
+ (let [tuples (execute-tuples word-count [(mock-tuple {"word" "the"})
+ (mock-tuple {"word" "the"})
+ (mock-tuple {"word" "cat"})])]
(is (ms= [["the" 1] ["the" 2] ["cat" 1]] tuples)))))
(deftest test-exclamation-bolt
(testing "Bolt emits word with exclamation marks"
- (test-with-tuple [exclamation-bolt (mock-tuple {"word" "nathan"})]
+ (let [tuples (execute-tuples exclamation-bolt (mock-tuple {"word" "nathan"}))]
(is (= [["nathan!!!"]] tuples)))))
(deftest test-rolling-bolt
(testing "Emits nothing if no object has been counted"
- (test-with-tuple [(rolling-count-bolt 9 3) tick-tuple]
+ (let [tuples (execute-tuples (rolling-count-bolt 9 3) tick-tuple)]
(is (empty? tuples))))
(testing "Emits something if object was counted"
- (test-with-tuple [(rolling-count-bolt 9 3)
- [(mock-tuple {"word" "nathan"}) tick-tuple]]
+ (let [tuples (execute-tuples (rolling-count-bolt 9 3)
+ [(mock-tuple {"word" "nathan"}) tick-tuple])]
(is (= [["nathan" 1 0]] tuples)))))
(deftest test-intermediate-rankings-bolt
(testing "Emits rankings for tick tuple"
- (test-with-tuple [(intermediate-rankings-bolt 5 2) tick-tuple]
+ (let [tuples (execute-tuples (intermediate-rankings-bolt 5 2) tick-tuple)]
(is (seq tuples))))
(testing "Emits nothing for normal tuple"
- (test-with-tuple [(intermediate-rankings-bolt 5 2)
- (mock-tuple {"obj" "nathan" "count" 1})]
+ (let [tuples (execute-tuples (intermediate-rankings-bolt 5 2)
+ (mock-tuple {"obj" "nathan" "count" 1}))]
(is (empty? tuples)))))
(defn- mock-rankable [object count]
@@ -107,9 +106,9 @@
(deftest test-total-rankings-bolt
(testing "Emits rankings for tick tuple"
- (test-with-tuple [(total-rankings-bolt 5 2) tick-tuple]
+ (let [tuples (execute-tuples (total-rankings-bolt 5 2) tick-tuple)]
(is (seq tuples))))
(testing "Emits nothing for normal tuple"
- (test-with-tuple [(total-rankings-bolt 5 2)
- (mock-tuple {"rankings" (mock-rankable "nathan" 2)})]
- (is (empty? tuples)))))
+ (let [tuples (execute-tuples (total-rankings-bolt 5 2)
+ (mock-tuple {"rankings" (mock-rankable "nathan" 2)}))]
+ (is (empty? tuples)))))