Merge pull request #60 from smeyfroi/master

Tidy up CSV destination

Author: thbar

lib/etl/control/destination.rb

@@ -94,16 +94,7 @@ def generators
 
       # Get the order of elements from the source order
       def order_from_source
-        order = []
-        control.sources.first.definition.each do |item|
-          case item
-          when Hash
-            order << item[:name]
-          else
-            order << item
-          end
-        end
-        order
+        control.sources.first.order
       end
 
       # Return true if the row is allowed. The row will not be allowed if the

lib/etl/control/destination/csv_destination.rb

@@ -48,9 +48,13 @@ def initialize(control, configuration, mapping={})
         @enclose = true & configuration[:enclose]
         @unique = configuration[:unique] ? configuration[:unique] + scd_required_fields : configuration[:unique]
         @unique.uniq! unless @unique.nil?
-        @order = mapping[:order] ? mapping[:order] + scd_required_fields : order_from_source
+        @write_header = configuration[:write_header]
+        @order = mapping[:order] + scd_required_fields if mapping[:order]
         @order.uniq! unless @order.nil?
-        raise ControlError, "Order required in mapping" unless @order
+      end
+
+      def order
+        @order ||= order_from_source
       end
 
       # Close the destination. This will flush the buffer and close the underlying stream or connection.
@@ -63,6 +67,11 @@ def close
       # Flush the destination buffer
       def flush
         #puts "Flushing buffer (#{file}) with #{buffer.length} rows"
+        if @write_header && !@header_written
+          f << order
+          @header_written = true
+        end
+
         buffer.flatten.each do |row|
           #puts "row change type: #{row.change_type}"
           # check to see if this row's compound key constraint already exists

lib/etl/control/source.rb

@@ -125,6 +125,19 @@ def read_locally
         Engine.read_locally
       end
 
+      # Get the order of fields that this source will present to the pipeline
+      def order
+        order = []
+        definition.each do |item|
+          case item
+          when Hash
+            order << item[:name]
+          else
+            order << item
+          end
+        end
+        order
+      end
     end
   end
 end

lib/etl/control/source/file_source.rb

@@ -53,6 +53,10 @@ def each
         end
       end
 
+      def order
+        @parser.fields.collect {|field| field.name}
+      end
+
       private
       # Copy source data to a local directory structure
       def copy_sources