Fix: improve compatibility with MessageConsumer implementations (#51)

* avoids JMS MessageConsumer get/each extension methods

Author: kares

CHANGELOG.md

@@ -1,5 +1,7 @@
-## unreleased
- - Fix: Fix test failures due to ECS compatibility default changes in `8.x` of logstash [#53](https://github.com/logstash-plugins/logstash-input-jms/pull/53)
+## 3.2.1
+ - Fix: improve compatibility with MessageConsumer implementations [#51](https://github.com/logstash-plugins/logstash-input-jms/pull/51),
+   such as IBM MQ.
+ - Test: Fix test failures due to ECS compatibility default changes in `8.x` of logstash [#53](https://github.com/logstash-plugins/logstash-input-jms/pull/53)
 
 ## 3.2.0
  - Feat: event_factory support + targets to aid ECS [#49](https://github.com/logstash-plugins/logstash-input-jms/pull/49)

lib/logstash/inputs/jms.rb

@@ -84,7 +84,7 @@ class LogStash::Inputs::Jms < LogStash::Inputs::Threadable
   # If pub-sub (topic) style should be used.
   config :pub_sub, :validate => :boolean, :default => false
 
-  # Durable subscriber settings.
+  # Durable message_consumer settings.
   # By default the `durable_subscriber_name` will be set to the topic, and `durable_subscriber_client_id` will be set
   # to 'Logstash'
   config :durable_subscriber, :validate => :boolean, :default => false
@@ -273,9 +273,9 @@ def run(output_queue)
       params = {:timeout => @timeout * 1000, :selector => @selector}
       subscriber = subscriber(session, params)
       until stop?
-        # This will read from the queue/topic until :timeout is breached, or messages are available whichever comes
-        # first.
-        subscriber.each({:timeout => @interval * 1000}) do |message|
+        # read from the queue/topic until :timeout is reached, or a message is available
+        # (whichever comes first)
+        do_receive_message(subscriber, timeout: @interval * 1000) do |message|
           queue_event(message, output_queue)
           break if stop?
         end
@@ -336,6 +336,28 @@ def queue_event(msg, output_queue)
     end
   end
 
+  # Loop through messages received and yield them.
+  #
+  # NOTE: a simplified replacement for JMS::MessageConsumer#each and #get (extensions).
+  #
+  # @param message_consumer [javax.jms.MessageConsumer]
+  # @param timeout in milliseconds
+  def do_receive_message(message_consumer, timeout: 0)
+    # Receive messages according to timeout
+    while true
+      case timeout
+      when  0 # Return immediately if no message is available
+        message = message_consumer.receiveNoWait()
+      when -1 # Wait forever
+        message = message_consumer.receive()
+      else # Wait for x milli-seconds for a message to be received from the broker
+        message = message_consumer.receive(timeout)
+      end
+      break unless message
+      yield message
+    end
+  end
+
   def to_string_keyed_hash(hash)
     hash.inject({}) { |h, (key, val)| h[key.to_s] = val; h }
   end

logstash-input-jms.gemspec

@@ -1,7 +1,7 @@
 Gem::Specification.new do |s|
 
   s.name            = 'logstash-input-jms'
-  s.version         = '3.2.0'
+  s.version         = '3.2.1'
   s.licenses        = ['Apache License (2.0)']
   s.summary         = "Reads events from a Jms Broker"
   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/inputs/spec_helper.rb

@@ -32,7 +32,6 @@ def send_message(&block)
   end
   input.run(queue)
 
-  destination = "#{pub_sub ? 'topic' : 'queue'}://#{queue_name}"
   tt.join(3)
 end