Merge pull request #14 from robbavey/master

Add DSL support for named streams.

Author: R. Tyler Croy

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 'red_storm/loggable'
 require 'pathname'
 
@@ -16,14 +17,12 @@ class Bolt
       include Loggable
       attr_reader :collector, :context, :config
 
+      include OutputFields
+
       def self.java_proxy; "Java::RedstormStormJruby::JRubyBolt"; end
 
       # DSL class methods
-
-      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

@@ -2,6 +2,7 @@
 require 'red_storm/configurator'
 require 'red_storm/environment'
 require 'red_storm/loggable'
+require 'red_storm/dsl/output_fields'
 require 'pathname'
 
 module RedStorm
@@ -13,6 +14,8 @@ class Spout
       include Loggable
       attr_reader :config, :context, :collector
 
+      include OutputFields
+
       def self.java_proxy; "Java::RedstormStormJruby::JRubySpout"; end
 
       # DSL class methods
@@ -21,10 +24,6 @@ def self.configure(&configure_block)
         @configure_block = block_given? ? configure_block : lambda {}
       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
@@ -120,10 +119,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
@@ -148,10 +143,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
@@ -171,7 +162,7 @@ def self.reliable?
       # below non-dry see Bolt class
       def self.inherited(subclass)
         path = (caller.first.to_s =~ /^(.+):\d+.*$/) ? $1 : raise(SpoutError, "unable to extract base topology class path from #{caller.first.inspect}")
-        subclass.base_class_path = File.expand_path(path)
+        subclass.base_class_path = Pathname.new(path).relative_path_from(Pathname.new(RedStorm::BASE_PATH)).to_s
       end
 
       def self.base_class_path=(path)

lib/red_storm/dsl/topology.rb

@@ -5,6 +5,7 @@
 
 java_import 'backtype.storm.topology.TopologyBuilder'
 java_import 'backtype.storm.generated.SubmitOptions'
+java_import 'backtype.storm.utils.Utils'
 
 module RedStorm
   module DSL
@@ -27,16 +28,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
@@ -48,7 +78,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
@@ -61,29 +91,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
@@ -97,7 +131,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