<NME> hot_bunnies.rb
<BEF> ADDFILE
<MSG> Initial version of the new RabbitMQ plugin version

This version uses different clients on different Ruby
implementations:

 * Bunny [1] 0.9 on MRI (CRuby)
 * HotBunnies [2] on JRuby

This eliminates a few limitations of the existing plugin,
based on Bunny 0.8:

 * Payload of large size [< than client machine's available RAM] now
   can be published without issues
 * Both are fully capable, well maintained clients that open the door
   to using more recent RabbitMQ features
 * Both Bunny 0.9 and especially HotBunnies offer double or triple (!) digits
   throughput improvements for event publishing
 * Bunny 0.9 offers automatic connection recovery as an option

The choice of Hot Bunnies on JRuby is primarily drive by Bunny triggering
a JRuby-specific issue with high CPU consumption by `IO#select` implementation.
In addition, it is based on an officially supported RabbitMQ Java client
and offers the best possible throughput of all Ruby clients/engines,
matching the Java client.

This version fully fleshed out yet, needs more work, automated
tests and network failure recovery QA.

1. http://rubybunny.info
2. http://hotbunnies.info

<DFF> @@ -0,0 +1,91 @@
+class LogStash::Inputs::RabbitMQ
+    # HotBunnies-based implementation for JRuby
+    module HotBunniesImpl
+      def register
+        require "hot_bunnies"
+
+        @vhost       ||= Bunny::DEFAULT_HOST
+        # 5672. Will be switched to 5671 by Bunny if TLS is enabled.
+        @port        ||= AMQ::Protocol::DEFAULT_PORT
+        @routing_key ||= "#"
+
+        @settings = {
+          :vhost => @vhost,
+          :host  => @host,
+          :port  => @port
+        }
+        @settings[:user]      = @user || Bunny::DEFAULT_USER
+        @settings[:pass]      = if @password
+                                  @password.value
+                                else
+                                  Bunny::DEFAULT_PASSWORD
+                                end
+
+        @settings[:log_level] = if @debug
+                                  :debug
+                                else
+                                  :error
+                                end
+
+        @settings[:tls]        = @ssl if @ssl
+
+        proto                  = if @ssl
+                                   "amqp"
+                                 else
+                                   "amqps"
+                                 end
+        @connection_url        = "#{proto}://#{@user}@#{@host}:#{@port}#{vhost}/#{@queue}"
+
+        @logger.info("Registering input #{@connection_url}")
+      end
+
+      def run(queue)
+        begin
+          @connection = HotBunnies.connect(@settings)
+
+          @logger.debug("Connecting to RabbitMQ. Settings: #{@settings.inspect}, queue: #{@queue.inspect}")
+          return if terminating?
+          @conn.start
+
+          @ch = @conn.create_channel
+
+          @ch.prefetch = @prefetch_count
+
+          @arguments_hash = Hash[*@arguments]
+
+          @q = @ch.queue(@queue,
+                         :durable     => @durable,
+                         :auto_delete => @auto_delete,
+                         :exclusive   => @exclusive,
+                         :arguments   => @arguments_hash)
+          @q.bind(@exchange, :routing_key => @key)
+
+          # we manually build a consumer here to be able to keep a reference to it
+          # in an @ivar even though we use a blocking version of HB::Queue#subscribe
+          @consumer = @q.build_consumer(:block => true)
+          @q.subscribe_with(@consumer, :manual_ack => @ack, :block => true) do |_, data|
+            @codec.decode(data) do |event|
+              event["source"] = @connection_url
+              queue << event
+            end
+          end
+        rescue HotBunnies::Exception, java.lang.Throwable => e
+          @logger.error("RabbitMQ connection error, will reconnect: #{e}")
+
+          sleep 3
+          retry
+        end
+      end
+
+      def teardown
+        @consumer.cancel
+        @consumer.gracefully_shut_down
+        @q.delete unless @durable
+
+        @ch.close         if @ch && @ch.open?
+        @connection.close if @connection && @connection.open?
+
+        finished
+      end
+    end # HotBunniesImpl
+end

Initial version of the new RabbitMQ plugin version