Add support for per-message properties

Fixes #66

Author: robbavey

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 4.0.9
+  -  Support per-message properties
+
 ## 4.0.7
   -  Require v4.3.0 of rabbitmq-connection_mixin which bumps the major version of the java rabbitmq lib
 

docs/index.asciidoc

@@ -46,6 +46,7 @@ This plugin supports the following configuration options plus the <<plugins-{typ
 | <<plugins-{type}s-{plugin}-heartbeat>> |<<number,number>>|No
 | <<plugins-{type}s-{plugin}-host>> |<<string,string>>|Yes
 | <<plugins-{type}s-{plugin}-key>> |<<string,string>>|No
+| <<plugins-{type}s-{plugin}-message_properties>> |<<Hash,Hash>>|No
 | <<plugins-{type}s-{plugin}-passive>> |<<boolean,boolean>>|No
 | <<plugins-{type}s-{plugin}-password>> |<<password,password>>|No
 | <<plugins-{type}s-{plugin}-persistent>> |<<boolean,boolean>>|No
@@ -169,6 +170,19 @@ Key to route to by default. Defaults to 'logstash'
 
 * Routing keys are ignored on fanout exchanges.
 
+[id="plugins-{type}s-{plugin}-message_properties"]
+===== `message_properties`
+
+  * Value type is <<hash,hash>>
+  * Default value is `{}`
+
+Add properties to be set per-message here, such as 'Content-Type', 'Priority'
+
+Example:
+[source,ruby]
+    message_properties => { "priority" => "1" }
+
+
 [id="plugins-{type}s-{plugin}-passive"]
 ===== `passive` 
 

lib/logstash/outputs/rabbitmq.rb

@@ -38,12 +38,18 @@ class RabbitMQ < LogStash::Outputs::Base
       # Should RabbitMQ persist messages to disk?
       config :persistent, :validate => :boolean, :default => true
 
+      # Properties to be passed along with the message
+      config :message_properties, :validate => :hash, :default => {}
+
       def register
         connect!
         @hare_info.exchange = declare_exchange!(@hare_info.channel, @exchange, @exchange_type, @durable)
         @codec.on_event(&method(:publish))
       end
 
+      def symbolize(myhash)
+        Hash[myhash.map{|(k,v)| [k.to_sym,v]}]
+      end
 
       def receive(event)
         @codec.encode(event)
@@ -53,7 +59,7 @@ def receive(event)
 
       def publish(event, message)
         raise ArgumentError, "No exchange set in HareInfo!!!" unless @hare_info.exchange
-        @hare_info.exchange.publish(message, :routing_key => event.sprintf(@key), :properties => { :persistent => @persistent })
+        @hare_info.exchange.publish(message, :routing_key => event.sprintf(@key), :properties => symbolize(@message_properties.merge(:persistent => @persistent)))
       rescue MarchHare::Exception, IOError, AlreadyClosedException, TimeoutException => e
         @logger.error("Error while publishing. Will retry.",
                       :message => e.message,

logstash-output-rabbitmq.gemspec

@@ -1,6 +1,6 @@
 Gem::Specification.new do |s|
   s.name            = 'logstash-output-rabbitmq'
-  s.version         = '4.0.8'
+  s.version         = '4.0.9'
   s.licenses        = ['Apache License (2.0)']
   s.summary         = "Push events to a RabbitMQ exchange"
   s.description     = "This gem is a Logstash plugin required to be installed on top of the Logstash core pipeline using $LS_HOME/bin/logstash-plugin install gemname. This gem is not a stand-alone program"

spec/outputs/rabbitmq_spec.rb

@@ -135,12 +135,16 @@
   let(:klass) { LogStash::Outputs::RabbitMQ }
   let(:exchange) { "myexchange" }
   let(:exchange_type) { "topic" }
+  let(:priority) { 34 }
   let(:default_plugin_config) {
     {
       "host" => "127.0.0.1",
       "exchange" => exchange,
       "exchange_type" => exchange_type,
-      "key" => "foo"
+      "key" => "foo",
+      "message_properties" => {
+          "priority" => priority
+      }
     }
   }
   let(:config) { default_plugin_config }
@@ -158,7 +162,7 @@ def spawn_and_wait(instance)
     # Extra time to make sure the output can attach
     sleep 1
   end
-
+  let(:message) { LogStash::Event.new("message" => "Foo Message", "extra_field" => "Blah") }
   let(:test_connection) { MarchHare.connect(instance.send(:rabbitmq_settings)) }
   let(:test_channel) { test_connection.create_channel }
   let(:test_queue) {
@@ -190,6 +194,14 @@ def spawn_and_wait(instance)
       instance.close
       expect(instance.connected?).to be_falsey
     end
+
+    it 'applies per message settings' do
+      instance.receive(message)
+      sleep 1.0
+
+      message, payload = test_queue.pop
+      expect(message.properties.to_s).to include("priority=#{priority}")
+    end
   end
 
   describe "sending a message with an exchange specified" do