New test/specs for bolt stream support

lib/red_storm/dsl/bolt.rb

@@ -26,7 +26,9 @@ module RedStorm
         @fields ||= []
         fields.each do |field|
           if field.kind_of? Hash
-            @fields << Hash[ field.map { |k, v| [k.to_s, v.to_s] } ]
+            @fields << Hash[
+              field.map { |k, v| [k.to_s, v.kind_of?(Array) ? v.map(&:to_s) : v.to_s] }
+            ]
           else
             @fields << field.to_s
           end
@@ -142,7 +144,7 @@ module RedStorm
           end
         end
 
-        declarer.declare(Fields.new(default_fields.flatten))
+        declarer.declare(Fields.new(default_fields.flatten)) unless default_fields.empty?
       end
 
       def get_component_configuration

spec/red_storm/dsl/bolt_spec.rb

@@ -62,6 +62,27 @@ describe RedStorm::SimpleBolt do
         Bolt1.send(:fields).should == ["f1", "f2"]
       end
 
+      it "should parse single hash argument" do
+        class Bolt1 < RedStorm::SimpleBolt
+          output_fields :stream => :f1
+        end
+        Bolt1.send(:fields).should == [{"stream" => "f1"}]
+      end
+
+      it "should parse hash of string and symbols" do
+        class Bolt1 < RedStorm::SimpleBolt
+          output_fields "stream" => [:f1, :f2]
+        end
+        Bolt1.send(:fields).should == [{"stream" => ["f1", "f2"]}]
+      end
+
+      it "should parse string and hash arguments" do
+        class Bolt1 < RedStorm::SimpleBolt
+          output_fields :f1, :stream => :f2
+        end
+        Bolt1.send(:fields).should == ["f1", {"stream" => "f2"}]
+      end
+
       it "should not share state over mutiple classes" do
         class Bolt1 < RedStorm::SimpleBolt
           output_fields :f1
@@ -115,6 +136,7 @@ describe RedStorm::SimpleBolt do
           Bolt1.send(:emit?).should be_true
           Bolt1.send(:ack?).should be_false
           Bolt1.send(:anchor?).should be_false
+          Bolt1.send(:stream?).should be_false
         end
 
         it "should parse :emit option" do
@@ -147,16 +169,27 @@ describe RedStorm::SimpleBolt do
           Bolt1.send(:anchor?).should be_true
         end
 
-        it "should parse multiple option" do
+        it "should parse :stream option" do
           class Bolt1 < RedStorm::SimpleBolt
-            on_receive :emit => false, :ack =>true, :anchor => true do
+            on_receive :stream => "test" do
             end
           end
 
-          Bolt1.receive_options.should == DEFAULT_RECEIVE_OPTIONS.merge(:emit =>false, :ack => true, :anchor => true)
+          Bolt1.receive_options.should == DEFAULT_RECEIVE_OPTIONS.merge(:stream => "test")
+          Bolt1.send(:stream?).should be_true
+        end
+
+        it "should parse multiple option" do
+          class Bolt1 < RedStorm::SimpleBolt
+            on_receive :emit => false, :ack =>true, :anchor => true, :stream => "test" do
+            end
+          end
+
+          Bolt1.receive_options.should == DEFAULT_RECEIVE_OPTIONS.merge(:emit =>false, :ack => true, :anchor => true, :stream => "test")
           Bolt1.send(:emit?).should be_false
           Bolt1.send(:ack?).should be_true
           Bolt1.send(:anchor?).should be_true
+          Bolt1.send(:stream?).should be_true
         end
       end
 
@@ -166,13 +199,13 @@ describe RedStorm::SimpleBolt do
           class Bolt1 < RedStorm::SimpleBolt
             def test_method; end
             on_receive :test_method
-
           end
 
           Bolt1.receive_options.should == DEFAULT_RECEIVE_OPTIONS
           Bolt1.send(:emit?).should be_true
           Bolt1.send(:ack?).should be_false
           Bolt1.send(:anchor?).should be_false
+          Bolt1.send(:stream?).should be_false
         end
 
         it "should parse :emit option" do
@@ -186,8 +219,7 @@ describe RedStorm::SimpleBolt do
 
         it "should parse :ack option" do
           class Bolt1 < RedStorm::SimpleBolt
-            on_receive :ack => true do
-            end
+            on_receive :test_method, :ack => true
           end
 
           Bolt1.receive_options.should == DEFAULT_RECEIVE_OPTIONS.merge(:ack => true)
@@ -196,24 +228,32 @@ describe RedStorm::SimpleBolt do
 
         it "should parse :anchor option" do
           class Bolt1 < RedStorm::SimpleBolt
-            on_receive :anchor => true do
-            end
+            on_receive :test_method, :anchor => true
           end
 
           Bolt1.receive_options.should == DEFAULT_RECEIVE_OPTIONS.merge(:anchor => true)
           Bolt1.send(:anchor?).should be_true
         end
 
-        it "should parse multiple option" do
+        it "should parse :stream option" do
           class Bolt1 < RedStorm::SimpleBolt
-            on_receive :emit => false, :ack =>true, :anchor => true do
-            end
+            on_receive :test_method, :stream => "test"
           end
 
-          Bolt1.receive_options.should == DEFAULT_RECEIVE_OPTIONS.merge(:emit =>false, :ack => true, :anchor => true)
+          Bolt1.receive_options.should == DEFAULT_RECEIVE_OPTIONS.merge(:stream => "test")
+          Bolt1.send(:stream?).should be_true
+        end
+
+        it "should parse multiple option" do
+          class Bolt1 < RedStorm::SimpleBolt
+            on_receive :test_method, :emit => false, :ack =>true, :anchor => true, :stream => "test"
+          end
+
+          Bolt1.receive_options.should == DEFAULT_RECEIVE_OPTIONS.merge(:emit =>false, :ack => true, :anchor => true, :stream => "test")
           Bolt1.send(:emit?).should be_false
           Bolt1.send(:ack?).should be_true
           Bolt1.send(:anchor?).should be_true
+          Bolt1.send(:stream?).should be_true
         end
       end
 
@@ -227,6 +267,7 @@ describe RedStorm::SimpleBolt do
           Bolt1.send(:emit?).should be_true
           Bolt1.send(:ack?).should be_false
           Bolt1.send(:anchor?).should be_false
+          Bolt1.send(:stream?).should be_false
         end
 
         it "should parse :emit option" do
@@ -256,15 +297,25 @@ describe RedStorm::SimpleBolt do
           Bolt1.send(:anchor?).should be_true
         end
 
-        it "should parse multiple option" do
+        it "should parse :stream option" do
           class Bolt1 < RedStorm::SimpleBolt
-            on_receive :emit => false, :ack =>true, :anchor => true
+            on_receive :stream => "test"
           end
 
-          Bolt1.receive_options.should == DEFAULT_RECEIVE_OPTIONS.merge(:emit =>false, :ack => true, :anchor => true)
+          Bolt1.receive_options.should == DEFAULT_RECEIVE_OPTIONS.merge(:stream => "test")
+          Bolt1.send(:stream?).should be_true
+        end
+
+        it "should parse multiple option" do
+          class Bolt1 < RedStorm::SimpleBolt
+            on_receive :emit => false, :ack =>true, :anchor => true, :stream => "test"
+          end
+
+          Bolt1.receive_options.should == DEFAULT_RECEIVE_OPTIONS.merge(:emit =>false, :ack => true, :anchor => true, :stream => "test")
           Bolt1.send(:emit?).should be_false
           Bolt1.send(:ack?).should be_true
           Bolt1.send(:anchor?).should be_true
+          Bolt1.send(:stream?).should be_true
         end
       end
     end
@@ -757,6 +808,32 @@ describe RedStorm::SimpleBolt do
         RedStorm::Fields.should_receive(:new).with(["f1", "f2"]).and_return("fields")
         bolt.declare_output_fields(declarer)
       end
+
+      it "should declare stream with fields" do
+        class Bolt1 < RedStorm::SimpleBolt
+          output_fields :stream => [:f1, :f2]
+        end
+        bolt = Bolt1.new
+        class RedStorm::Fields; end
+        declarer = mock("Declarer")
+        declarer.should_receive(:declareStream).with("stream", "fields")
+        RedStorm::Fields.should_receive(:new).with(["f1", "f2"]).and_return("fields")
+        bolt.declare_output_fields(declarer)
+      end
+
+      it "should declare default stream fields and custom stream fields" do
+        class Bolt1 < RedStorm::SimpleBolt
+          output_fields :f1, :f2, :stream => [:f3, :f4]
+        end
+        bolt = Bolt1.new
+        class RedStorm::Fields; end
+        declarer = mock("Declarer")
+        declarer.should_receive(:declareStream).with("stream", "stream_fields")
+        declarer.should_receive(:declare).with("default_fields")
+        RedStorm::Fields.should_receive(:new).with(["f3", "f4"]).and_return("stream_fields")
+        RedStorm::Fields.should_receive(:new).with(["f1", "f2"]).and_return("default_fields")
+        bolt.declare_output_fields(declarer)
+      end
     end
 
     describe "get_component_configuration" do