Wrongly assumed topo bolts inherit from redstorm

Not all bolts defined in a topology inherit from the RedStorm::DSL. This
ensures the output fields in the ComponentDefinition are only copied if
the class uses the OutputFields module. Also, redefining field names for
the default stream will replace any previously defined default field
names.

lib/red_storm/dsl/topology.rb

@@ -27,18 +27,22 @@ module RedStorm
           @constructor_args = constructor_args
           @id = id.to_s
           @parallelism = parallelism
-          @output_fields = @clazz.fields.clone
+          @output_fields = Hash.new([])
+
+          initialize_output_fields
         end
 
         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
-              @output_fields[Utils::DEFAULT_STREAM_ID] |= field.kind_of?(Array) ? field.map(&:to_s) : [field.to_s]
+              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
@@ -49,6 +53,12 @@ module RedStorm
 
         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|

spec/red_storm/dsl/topology_spec.rb

@@ -118,6 +118,28 @@ describe RedStorm::SimpleTopology do
         Topology1.spouts.last.output_fields.should == { "default" => ["f3"] }
       end
 
+      it "should default output_fields to the class defined fields" do
+        class SpoutClass1
+          output_fields :f1, :f2
+        end
+        class Topology1 < RedStorm::SimpleTopology
+          spout SpoutClass1
+        end
+        Topology1.spouts.first.output_fields.should == { "default" => ["f1", "f2"] }
+      end
+
+      it "should override class defined fields with topology output fields" do
+        class SpoutClass1
+          output_fields :f1, :f2
+        end
+        class Topology1 < RedStorm::SimpleTopology
+          spout SpoutClass1 do
+            output_fields :f3, :f4
+          end
+        end
+        Topology1.spouts.first.output_fields.should == { "default" => ["f3", "f4"] }
+      end
+
     end
 
     describe "bolt statement" do
@@ -201,6 +223,29 @@ describe RedStorm::SimpleTopology do
         Topology1.bolts.last.output_fields.should == { "default" => ["f3"] }
       end
 
+      it "should default output_fields to the class defined fields" do
+        class BoltClass1
+          output_fields :f1, :f2
+        end
+        class Topology1 < RedStorm::SimpleTopology
+          bolt BoltClass1 do
+          end
+        end
+        Topology1.bolts.first.output_fields.should == { "default" => ["f1", "f2"] }
+      end
+
+      it "should override class defined fields with topology output fields" do
+        class BoltClass1
+          output_fields :f1, :f2
+        end
+        class Topology1 < RedStorm::SimpleTopology
+          bolt BoltClass1 do
+            output_fields :f3, :f4
+          end
+        end
+        Topology1.bolts.first.output_fields.should == { "default" => ["f3", "f4"] }
+      end
+
     end
 
     describe "configure statement" do