Merge pull request #5 from jseely/master

Add config option to specify message properties

README.md

@@ -35,6 +35,7 @@ Or install it yourself as:
   proxy_port	   <Integer>   # Proxy port. [Optional: default => 3128]
   read_timeout     <Integer>   # HTTP Read timeout in seconds[Optional: default => 60]
   open_timeout     <Integer>   # HTTP Open timeout in seconds[Optional: default => 60]
+  message_properties <Json Object> # A json object of key/value pairs to add Properties to the events being sent to EventHubs [Optional: default => nil]
 </match>
 ```
 

fluent-plugin-azureeventhubs.gemspec

@@ -4,8 +4,8 @@ $LOAD_PATH.unshift(lib) unless $LOAD_PATH.include?(lib)
 
 Gem::Specification.new do |spec|
   spec.name          = "fluent-plugin-azureeventhubs"
-  spec.version       = "0.0.4"
-  spec.authors       = ["Hidemasa Togashi"]
+  spec.version       = "0.0.5"
+  spec.authors       = ["Hidemasa Togashi", "Justin Seely"]
   spec.email         = ["[email protected]"]
   spec.summary       = "Fluentd output plugin for Azure Event Hubs"
   spec.description   = "Fluentd output plugin for Azure Event Hubs"

lib/fluent/plugin/azureeventhubs/http.rb

@@ -44,11 +44,18 @@ def generate_sas_token(uri)
   private :generate_sas_token
 
   def send(payload)
+    send_w_properties(payload, nil)
+  end
+
+  def send_w_properties(payload, properties)
     token = generate_sas_token(@uri.to_s)
     headers = {
       'Content-Type' => 'application/atom+xml;type=entry;charset=utf-8',
       'Authorization' => token
     }
+    if not properties.nil?
+      headers = headers.merge(properties)
+    end
     if (@proxy_addr.to_s.empty?)
     	https = Net::HTTP.new(@uri.host, @uri.port)
         https.open_timeout = @open_timeout

lib/fluent/plugin/out_azureeventhubs_buffered.rb

@@ -14,6 +14,7 @@ class AzureEventHubsOutputBuffered < Fluent::BufferedOutput
     config_param :proxy_port, :integer,:default => 3128
     config_param :open_timeout, :integer,:default => 60
     config_param :read_timeout, :integer,:default => 60
+    config_param :message_properties, :hash, :default => nil
 
     def configure(conf)
       super
@@ -39,7 +40,7 @@ def write(chunk)
         if @include_time
           record[@tag_time_name] = time
         end
-        @sender.send(record)
+        @sender.send_w_properties(record, @message_properties)
       }
     end
   end