Support codecs generating multiple events (#56)

* tests: fix build

 - use https for fixture resource retrieval
 - use "latest" 5.15.x activemq
 - declare module encapsulation exports

* feature: support codec emitting multiple events

Resolves: #22

Author: yaauie

.ci/run.sh

@@ -5,15 +5,23 @@ env
 
 set -ex
 
-export ACTIVEMQ_VERSION=5.15.15
+export ACTIVEMQ_VERSION=5.15.16
 ./setup_broker.sh
 bundle install
+
+java_major_version="$(JRUBY_OPTS="" jruby -e 'puts ENV_JAVA["java.version"]&.slice(/(?!1[.])[0-9]+/)')"
+if (( "${java_major_version}" >= "9" )); then
+  export JRUBY_OPTS="-J--add-exports=java.base/sun.security.ssl=ALL-UNNAMED${JRUBY_OPTS:+ }${JRUBY_OPTS}"
+fi
+
 bundle exec rspec
+
 ./start_ssl_broker.sh
 bundle exec rspec -fd --tag integration --tag tls --tag ~plaintext
 ./stop_broker.sh
 
 ./start_broker.sh
 bundle exec rspec -fd --tag integration --tag plaintext --tag ~tls
 ./stop_broker.sh
+
 ./teardown_broker.sh

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 3.3.0
+ - Added support for decoding multiple events from text or binary messages when using a codec that produces multiple events
+
 ## 3.2.2
  - Fix: Remove usage of `java_kind_of?` to allow this plugin to be supported for versions of Logstash using jruby-9.3.x
  [#54](https://github.com/logstash-plugins/logstash-input-jms/pull/54)

lib/logstash/inputs/jms.rb

@@ -276,7 +276,7 @@ def run(output_queue)
         # 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)
+          queue_events(message, output_queue)
           break if stop?
         end
       end
@@ -296,20 +296,27 @@ def run(output_queue)
     end
   end # def run_consumer
 
-  def queue_event(msg, output_queue)
-    begin
-      if @include_body
-        if msg.kind_of?(JMS::MapMessage)
-          event = process_map_message(msg)
-        elsif msg.kind_of?(JMS::TextMessage) || msg.kind_of?(JMS::BytesMessage)
-          event = decode_message(msg)
-        else
-          @logger.error( "Unsupported message type #{msg.data.class.to_s}" )
-        end
+  def events(msg, &event_handler)
+    return enum_for(:events, msg).to_a unless block_given?
+
+    if @include_body
+      if msg.kind_of?(JMS::MapMessage)
+        return process_map_message(msg, &event_handler)
+      elsif msg.kind_of?(JMS::TextMessage) || msg.kind_of?(JMS::BytesMessage)
+        return process_string_message(msg, &event_handler)
+      else
+        @logger.error( "Unsupported message type #{msg.data.class.to_s}" )
       end
+    end
 
-      event ||= event_factory.new_event
+    yield event_factory.new_event
+  rescue => e
+    @logger.error("Failed to create event", :message => msg, :exception => e,
+                  :backtrace => e.backtrace)
+  end
 
+  def queue_events(msg, output_queue)
+    events(msg).map do |event|
       # Here, we can use the JMS Enqueue timestamp as the @timestamp
       if @use_jms_timestamp && msg.jms_timestamp
         event.set("@timestamp", LogStash::Timestamp.at(msg.jms_timestamp / 1000, (msg.jms_timestamp % 1000) * 1000))
@@ -327,12 +334,9 @@ def queue_event(msg, output_queue)
         @properties_setter.call(event, properties)
       end
 
-      decorate(event)
-      output_queue << event
-
-    rescue => e # parse or event creation error
-      @logger.error("Failed to create event", :message => msg, :exception => e,
-                    :backtrace => e.backtrace)
+      event
+    end.each do |ready_event|
+      output_queue << ready_event
     end
   end
 
@@ -365,9 +369,17 @@ def to_string_keyed_hash(hash)
 
   # @param msg [JMS::MapMessage]
   # @return [LogStash::Event]
-  def process_map_message(msg)
+  def process_map_message(msg, &event_handler)
     data = to_string_keyed_hash(msg.data)
-    do_target_check_once_and_get_event_factory.new_event(data)
+    yield do_target_check_once_and_get_event_factory.new_event(data)
+  end
+
+  def process_string_message(msg, &event_handler)
+    text = msg.to_s # javax.jms.TextMessage#getText (e.g. JSON payload)
+    return event_factory.new_event if text.nil?
+
+    @codec.decode(text, &event_handler)
+    @codec.flush(&event_handler)
   end
 
   def do_target_check_once_and_get_event_factory

logstash-input-jms.gemspec

@@ -1,7 +1,7 @@
 Gem::Specification.new do |s|
 
   s.name            = 'logstash-input-jms'
-  s.version         = '3.2.2'
+  s.version         = '3.3.0'
   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"
@@ -32,5 +32,6 @@ Gem::Specification.new do |s|
   s.add_runtime_dependency "jruby-jms", ">= 1.2.0" #(Apache 2.0 license)
   s.add_runtime_dependency 'semantic_logger', '< 4.0.0'
 
+  s.add_development_dependency 'logstash-codec-line', '~> 3.0'
   s.add_development_dependency 'logstash-devutils'
 end

setup_broker.sh

@@ -9,6 +9,6 @@ fi
 
 curl -s -o activemq-all.jar https://repo1.maven.org/maven2/org/apache/activemq/activemq-all/$ACTIVEMQ_VERSION/activemq-all-$ACTIVEMQ_VERSION.jar
 mv activemq-all.jar ./spec/inputs/fixtures/
-curl -s -o activemq-bin.tar.gz http://archive.apache.org/dist/activemq/$ACTIVEMQ_VERSION/apache-activemq-$ACTIVEMQ_VERSION-bin.tar.gz
+curl -s -o activemq-bin.tar.gz https://archive.apache.org/dist/activemq/$ACTIVEMQ_VERSION/apache-activemq-$ACTIVEMQ_VERSION-bin.tar.gz
 tar xvf activemq-bin.tar.gz
 mv apache-activemq-$ACTIVEMQ_VERSION activemq

spec/inputs/integration/jms_spec.rb

@@ -241,6 +241,19 @@
 
         end
       end
+
+      context 'when using a multi-event codec' do
+        let(:config) { super().merge('codec' => 'line') }
+        let(:message) { 'one' + "\n" + 'two' + "\n" + 'three' }
+        it 'emits multiple events' do
+          send_message do |session|
+            session.message(message)
+          end
+          expect(queue.size).to eql 3
+          expect(queue.map { |e| e.get('message') }).to contain_exactly("one", "two", "three")
+          expect(queue).to all(have_header_value("jms_destination", destination))
+        end
+      end
     end
 
     context 'when the message is map message', :ecs_compatibility_support do

spec/inputs/unit/jms_spec.rb

@@ -291,7 +291,7 @@
         allow_any_instance_of(described_class).to receive(:ecs_compatibility).and_return(ecs_compatibility)
 
         plugin.register
-        plugin.queue_event(jms_message_double, queue = [])
+        plugin.queue_events(jms_message_double, queue = [])
         expect(queue.size).to eql 1
         @event = queue.first
       end