flush codec after decode

lib/logstash/inputs/http.rb

@@ -92,7 +92,6 @@ class LogStash::Inputs::Http < LogStash::Inputs::Base
   # mostly due to rack compliance
   REJECTED_HEADERS = ["puma.socket", "rack.hijack?", "rack.hijack", "rack.url_scheme", "rack.after_reply", "rack.version", "rack.errors", "rack.multithread", "rack.multiprocess", "rack.run_once", "SCRIPT_NAME", "QUERY_STRING", "SERVER_PROTOCOL", "SERVER_SOFTWARE", "GATEWAY_INTERFACE"]
 
-  public
   def register
     require "logstash/util/http_compressed_requests"
     @server = ::HTTPInputWebServer.new(nil) # we'll set the rack handler later
@@ -154,12 +153,12 @@ def run(queue)
         end
         begin
           codec = local_codecs[req["content_type"]] || local_codecs[:default]
-          codec.decode(body.read) do |event|
-            event.set("host", remote_host)
-            event.set("headers", req)
-            decorate(event)
-            queue << event
-          end
+
+          codec.decode(body.read) { |event| push_decorated_event(queue, event, remote_host, req) }
+
+          # since payloads are self-contained and we don't handle multipart we should flush
+          # the codec after each request.
+          codec.flush { |event| push_decorated_event(queue, event, remote_host, req) }
         ensure
           @write_slots.put(local_codecs)
         end
@@ -188,7 +187,16 @@ def run(queue)
     @server.run.join
   end
 
+  def stop
+    return unless @server
+    @server.stop(true)
+    @server.binder.close if @server.binder
+  rescue IOError
+    # do nothing
+  end
+
   private
+
   def lowercase_keys(hash)
     new_hash = {}
     hash.each_pair do |k,v|
@@ -197,13 +205,11 @@ def lowercase_keys(hash)
     new_hash
   end
 
-  public
-  def stop
-    return unless @server
-    @server.stop(true)
-    @server.binder.close if @server.binder
-  rescue IOError
-    # do nothing
+  def push_decorated_event(queue, event, host, headers)
+    event.set("host", host)
+    event.set("headers", headers)
+    decorate(event)
+    queue << event
   end
 
 end # class LogStash::Inputs::Http

logstash-input-http.gemspec

@@ -26,5 +26,6 @@ Gem::Specification.new do |s|
 
   s.add_development_dependency 'logstash-devutils'
   s.add_development_dependency 'logstash-codec-json'
+  s.add_development_dependency 'logstash-codec-line'
   s.add_development_dependency 'ftw'
 end

spec/inputs/http_spec.rb

@@ -136,6 +136,7 @@ def do_post
         end
       end
     end
+
     context "with json codec" do
       subject { LogStash::Inputs::Http.new("port" => port, "codec" => "json") }
       it "should parse the json body" do
@@ -145,6 +146,20 @@ def do_post
       end
     end
 
+    context "with json_lines codec without final delimiter" do
+      subject { LogStash::Inputs::Http.new("port" => port, "codec" => "line") }
+      let(:line1) { "foo" }
+      let(:line2) { "bar" }
+      it "should parse all json_lines in body including last one" do
+        agent.post!("http://localhost:#{port}/meh.json", :body => "#{line1}\n#{line2}")
+        expect(queue.size).to eq(2)
+        event = queue.pop
+        expect(event.get("message")).to eq("foo")
+        event = queue.pop
+        expect(event.get("message")).to eq("bar")
+      end
+    end
+
     context "when using a custom codec mapping" do
       subject { LogStash::Inputs::Http.new("port" => port,
                                            "additional_codecs" => { "application/json" => "plain" }) }