Add DSL support for declaring named streams for Bolts

Gemfile.lock

@@ -7,6 +7,7 @@ PATH
 GEM
   remote: https://rubygems.org/
   specs:
+    coderay (1.1.0)
     colorize (0.5.8)
     coveralls (0.6.7)
       colorize
@@ -15,8 +16,15 @@ GEM
       simplecov (>= 0.7)
       thor
     diff-lcs (1.2.4)
+    ffi (1.9.8-java)
+    method_source (0.8.2)
     mime-types (1.23)
     multi_json (1.7.7)
+    pry (0.10.1-java)
+      coderay (~> 1.1.0)
+      method_source (~> 0.8.1)
+      slop (~> 3.4)
+      spoon (~> 0.0)
     rake (10.0.4)
     redis (3.0.4)
     rest-client (1.6.7)
@@ -33,13 +41,17 @@ GEM
       multi_json (~> 1.0)
       simplecov-html (~> 0.7.1)
     simplecov-html (0.7.1)
+    slop (3.6.0)
+    spoon (0.0.4)
+      ffi
     thor (0.18.1)
 
 PLATFORMS
   java
 
 DEPENDENCIES
   coveralls
+  pry
   redis
   redstorm!
   rspec (~> 2.13)

lib/red_storm/dsl/bolt.rb

@@ -1,6 +1,7 @@
 require 'java'
 require 'red_storm/configurator'
 require 'red_storm/environment'
+require 'red_storm/dsl/output_fields'
 require 'pathname'
 
 java_import 'backtype.storm.tuple.Fields'
@@ -14,6 +15,8 @@ class BoltError < StandardError; end
     class Bolt
       attr_reader :collector, :context, :config
 
+      include OutputFields
+
       def self.java_proxy; "Java::RedstormStormJruby::JRubyBolt"; end
 
       # DSL class methods
@@ -22,10 +25,6 @@ def self.log
         @log ||= Java::OrgApacheLog4j::Logger.getLogger(self.name)
       end
 
-      def self.output_fields(*fields)
-        @fields = fields.map(&:to_s)
-      end
-
       def self.configure(&configure_block)
         @configure_block = block_given? ? configure_block : lambda {}
       end
@@ -62,10 +61,18 @@ def unanchored_emit(*values)
         @collector.emit_tuple(Values.new(*values))
       end
 
+      def unanchored_stream_emit(stream, *values)
+        @collector.emit_tuple_stream(stream, Values.new(*values))
+      end
+
       def anchored_emit(tuple, *values)
         @collector.emit_anchor_tuple(tuple, Values.new(*values))
       end
 
+      def anchored_stream_emit(stream, tuple, *values)
+        @collector.emit_anchor_tuple_stream(stream, tuple, Values.new(*values))
+      end
+
       def ack(tuple)
         @collector.ack(tuple)
       end
@@ -80,7 +87,21 @@ def execute(tuple)
         output = on_receive(tuple)
         if output && self.class.emit?
           values_list = !output.is_a?(Array) ? [[output]] : !output.first.is_a?(Array) ? [output] : output
-          values_list.each{|values| self.class.anchor? ? anchored_emit(tuple, *values) : unanchored_emit(*values)}
+          values_list.each do |values|
+            if self.class.anchor?
+              if self.class.stream?
+                anchored_stream_emit(self.stream, tuple, *values)
+              else
+                anchored_emit(tuple, *values)
+              end
+            else
+              if self.class.stream?
+                unanchored_stream_emit(self.stream, *values)
+              else
+                unanchored_emit(*values)
+              end
+            end
+          end
           @collector.ack(tuple) if self.class.ack?
         end
       end
@@ -97,10 +118,6 @@ def cleanup
         on_close
       end
 
-      def declare_output_fields(declarer)
-        declarer.declare(Fields.new(self.class.fields))
-      end
-
       def get_component_configuration
         configurator = Configurator.new
         configurator.instance_exec(&self.class.configure_block)
@@ -113,10 +130,6 @@ def get_component_configuration
       def on_init; end
       def on_close; end
 
-      def self.fields
-        @fields ||= []
-      end
-
       def self.configure_block
         @configure_block ||= lambda {}
       end

lib/red_storm/dsl/output_collector.rb

@@ -6,4 +6,13 @@
 class OutputCollector
   java_alias :emit_tuple, :emit, [java.lang.Class.for_name("java.util.List")]
   java_alias :emit_anchor_tuple, :emit, [Tuple.java_class, java.lang.Class.for_name("java.util.List")]
+  java_alias :emit_tuple_stream, :emit, [
+    java.lang.String,
+    java.lang.Class.for_name("java.util.List")
+  ]
+  java_alias :emit_anchor_tuple_stream, :emit, [
+    java.lang.String,
+    Tuple.java_class,
+    java.lang.Class.for_name("java.util.List")
+  ]
 end

lib/red_storm/dsl/output_fields.rb

@@ -0,0 +1,48 @@
+module RedStorm
+  module DSL
+    module OutputFields
+
+      def self.included(base)
+        base.extend ClassMethods
+      end
+
+      def declare_output_fields(declarer)
+        self.class.fields.each do |stream, fields|
+          declarer.declareStream(stream, Fields.new(fields))
+        end
+      end
+
+      def stream
+        self.class.stream
+      end
+
+      module ClassMethods
+
+        def output_fields(*fields)
+          @output_fields ||= Hash.new([])
+          fields.each do |field|
+            case field
+            when Hash
+              field.each { |k, v| @output_fields[k.to_s] = v.kind_of?(Array) ? v.map(&:to_s) : [v.to_s] }
+            else
+              @output_fields['default'] |= field.kind_of?(Array) ? field.map(&:to_s) : [field.to_s]
+            end
+          end
+        end
+
+        def fields
+          @output_fields ||= Hash.new([])
+        end
+
+        def stream?
+          self.receive_options[:stream] && !self.receive_options[:stream].empty?
+        end
+
+        def stream
+          self.receive_options[:stream]
+        end
+      end
+    end
+  end
+end
+

lib/red_storm/dsl/spout.rb

@@ -1,6 +1,7 @@
 require 'java'
 require 'red_storm/configurator'
 require 'red_storm/environment'
+require 'red_storm/dsl/output_fields'
 require 'pathname'
 
 module RedStorm
@@ -11,6 +12,8 @@ class SpoutError < StandardError; end
     class Spout
       attr_reader :config, :context, :collector
 
+      include OutputFields
+
       def self.java_proxy; "Java::RedstormStormJruby::JRubySpout"; end
 
       # DSL class methods
@@ -23,10 +26,6 @@ def self.log
         @log ||= Java::OrgApacheLog4j::Logger.getLogger(self.name)
       end
 
-      def self.output_fields(*fields)
-        @fields = fields.map(&:to_s)
-      end
-
       def self.on_send(*args, &on_send_block)
         options = args.last.is_a?(Hash) ? args.pop : {}
         method_name = args.first
@@ -126,10 +125,6 @@ def deactivate
         on_deactivate
       end
 
-      def declare_output_fields(declarer)
-        declarer.declare(Fields.new(self.class.fields))
-      end
-
       def ack(msg_id)
         on_ack(msg_id)
       end
@@ -154,10 +149,6 @@ def on_deactivate; end
       def on_ack(msg_id); end
       def on_fail(msg_id); end
 
-      def self.fields
-        @fields ||= []
-      end
-
       def self.configure_block
         @configure_block ||= lambda {}
       end

lib/red_storm/dsl/topology.rb

@@ -4,6 +4,7 @@
 
 java_import 'backtype.storm.topology.TopologyBuilder'
 java_import 'backtype.storm.generated.SubmitOptions'
+java_import 'backtype.storm.utils.Utils'
 
 module RedStorm
   module DSL
@@ -26,16 +27,45 @@ def initialize(component_class, constructor_args, id, parallelism)
           @constructor_args = constructor_args
           @id = id.to_s
           @parallelism = parallelism
-          @output_fields = []
+          @output_fields = Hash.new([])
+
+          initialize_output_fields
         end
 
-        def output_fields(*args)
-          args.empty? ? @output_fields : @output_fields = args.map(&:to_s)
+        def output_fields(*fields)
+          default_fields = []
+          fields.each do |field|
+            case field
+            when Hash
+              field.each { |k, v| @output_fields[k.to_s] = v.kind_of?(Array) ? v.map(&:to_s) : [v.to_s] }
+            else
+              default_fields |= field.kind_of?(Array) ? field.map(&:to_s) : [field.to_s]
+            end
+          end
+          @output_fields[Utils::DEFAULT_STREAM_ID] = default_fields unless default_fields.empty?
+
+          @output_fields
         end
 
         def is_java?
           @clazz.name.split('::').first.downcase == 'java'
         end
+
+        private
+
+        def initialize_output_fields
+          if @clazz.ancestors.include?(RedStorm::DSL::OutputFields)
+            @output_fields = @clazz.fields.clone
+          end
+        end
+
+        def java_safe_fields
+          java_hash = java.util.HashMap.new()
+          @output_fields.each do |k, v|
+            java_hash.put(k, v.to_java('java.lang.String')) unless v.empty?
+          end
+          java_hash
+        end
       end
 
       class SpoutDefinition < ComponentDefinition
@@ -47,7 +77,7 @@ def new_instance
           elsif is_java?
             @clazz.new(*constructor_args)
           else
-            Object.module_eval(@clazz.java_proxy).new(@clazz.base_class_path, @clazz.name, @output_fields)
+            Object.module_eval(@clazz.java_proxy).new(@clazz.base_class_path, @clazz.name, java_safe_fields)
           end
         end
       end
@@ -60,29 +90,33 @@ def initialize(*args)
           @sources = []
         end
 
-        def source(source_id, grouping)
-          @sources << [source_id.is_a?(Class) ? Topology.underscore(source_id) : source_id.to_s, grouping.is_a?(Hash) ? grouping : {grouping => nil}]
+        def source(source_id, grouping, stream = Utils::DEFAULT_STREAM_ID)
+          @sources << [
+            source_id.is_a?(Class) ? Topology.underscore(source_id) : source_id.to_s,
+            grouping.is_a?(Hash) ? grouping : {grouping => nil},
+            stream.to_s
+          ]
         end
 
         def define_grouping(declarer)
-          @sources.each do |source_id, grouping|
+          @sources.each do |source_id, grouping, stream|
             grouper, params = grouping.first
               # declarer.fieldsGrouping(source_id, Fields.new())
             case grouper
             when :fields
-              declarer.fieldsGrouping(source_id, Fields.new(*([params].flatten.map(&:to_s))))
+              declarer.fieldsGrouping(source_id, stream, Fields.new(*([params].flatten.map(&:to_s))))
             when :global
-              declarer.globalGrouping(source_id)
+              declarer.globalGrouping(source_id, stream)
             when :shuffle
-              declarer.shuffleGrouping(source_id)
+              declarer.shuffleGrouping(source_id, stream)
             when :local_or_shuffle
-              declarer.localOrShuffleGrouping(source_id)
+              declarer.localOrShuffleGrouping(source_id, stream)
             when :none
-              declarer.noneGrouping(source_id)
+              declarer.noneGrouping(source_id, stream)
             when :all
-              declarer.allGrouping(source_id)
+              declarer.allGrouping(source_id, stream)
             when :direct
-              declarer.directGrouping(source_id)
+              declarer.directGrouping(source_id, stream)
             else
               raise("unknown grouper=#{grouper.inspect}")
             end
@@ -96,7 +130,7 @@ def new_instance
           elsif is_java?
             @clazz.new(*constructor_args)
           else
-            Object.module_eval(@clazz.java_proxy).new(@clazz.base_class_path, @clazz.name, @output_fields)
+            Object.module_eval(@clazz.java_proxy).new(@clazz.base_class_path, @clazz.name, java_safe_fields)
           end
         end
       end

redstorm.gemspec

@@ -21,5 +21,6 @@ Gem::Specification.new do |s|
   s.executables   = ['redstorm']
 
   s.add_development_dependency 'rspec', '~> 2.13'
+  s.add_development_dependency 'pry'
   s.add_runtime_dependency 'rake'
 end