Add reponse_body to be able to provide a custom body (issue 117)

docs/index.asciidoc

@@ -197,6 +197,17 @@ invalid credentials (401), internal errors (503) or backpressure (429).
 
 If 204 (No Content) is set, the response body will not be sent in the response.
 
+[id="plugins-{type}s-{plugin}-response_body"]
+===== `response_body`
+
+  * Value type is: <<string,string>>
+  * Default value is `ok`
+
+The response body if the request is processed successfully.
+
+The response body is not validated and the `Content-type` is not adjusted to match its actual type.
+No body will be sent if `response_code` is configured to 204 (No Content).
+
 [id="plugins-{type}s-{plugin}-response_headers"]
 ===== `response_headers` 
 

lib/logstash/inputs/http.rb

@@ -116,6 +116,8 @@ class LogStash::Inputs::Http < LogStash::Inputs::Base
   config :max_content_length, :validate => :number, :required => false, :default => 100 * 1024 * 1024
 
   config :response_code, :validate => [200, 201, 202, 204], :default => 200
+
+  config :response_body, :validate => :string, :default => "ok"
   # Deprecated options
 
   # The JKS keystore to validate the client's certificates
@@ -211,7 +213,7 @@ def validate_ssl_settings!
 
   def create_http_server(message_handler)
     org.logstash.plugins.inputs.http.NettyHttpServer.new(
-      @host, @port, message_handler, build_ssl_params(), @threads, @max_pending_requests, @max_content_length, @response_code)
+      @host, @port, message_handler, build_ssl_params(), @threads, @max_pending_requests, @max_content_length, @response_code, @response_body)
   end
 
   def build_ssl_params

spec/inputs/http_spec.rb

@@ -356,6 +356,55 @@
           expect(response.code).to eq(202)
         end
       end
+      context "when response_code is set to 204" do
+        let(:code) { 204 }
+        subject { LogStash::Inputs::Http.new("port" => port, "response_code" => code) }
+        it "responds with the configured code and no body even if forced" do
+          response = client.post("http://127.0.0.1:#{port}", :body => "hello")
+          response.call
+          expect(response.code).to eq(204)
+          expect(response.body).to eq(nil)
+        end
+      end
+    end
+    describe "return body" do
+      context "when response_body is not configured" do
+        subject { LogStash::Inputs::Http.new("port" => port) }
+        it "responds with the default body" do
+          response = client.post("http://127.0.0.1:#{port}", :body => "hello")
+          response.call
+          expect(response.body).to eq("ok")
+        end
+      end
+      context "when response_body is configured" do
+        let(:body) { "world!" }
+        subject { LogStash::Inputs::Http.new("port" => port, "response_body" => body) }
+        it "responds with the configured body" do
+          response = client.post("http://127.0.0.1:#{port}", :body => "hello")
+          response.call
+          expect(response.body).to eq(body)
+        end
+      end
+      context "when response_body is configured to an empty string" do
+        let(:body) { "" }
+        subject { LogStash::Inputs::Http.new("port" => port, "response_body" => body) }
+        it "responds with the configured body" do
+          response = client.post("http://127.0.0.1:#{port}", :body => "hello")
+          response.call
+          expect(response.body).to eq(body)
+        end
+      end
+      context "when response_body is configured and content-type is specified" do
+        let(:body) { "{\"test\": \"body\"}" }
+        let(:custom_headers) { { 'content-type' => "application/json" } }
+        subject { LogStash::Inputs::Http.new("port" => port, "response_body" => body, "response_headers" => custom_headers) }
+        it "responds with the configured body and headers" do
+          response = client.post("http://127.0.0.1:#{port}", :body => "Plain-text")
+          response.call
+          expect(response.body).to eq(body)
+          expect(response.headers.to_hash).to include({ "content-type" => "application/json" })
+        end
+      end
     end
   end
 

src/main/java/org/logstash/plugins/inputs/http/HttpInitializer.java

@@ -21,13 +21,16 @@ public class HttpInitializer extends ChannelInitializer<SocketChannel> {
     private final int maxContentLength;
     private final HttpResponseStatus responseStatus;
     private final ThreadPoolExecutor executorGroup;
+    private final String responseBody;
 
     public HttpInitializer(IMessageHandler messageHandler, ThreadPoolExecutor executorGroup,
-                           int maxContentLength, HttpResponseStatus responseStatus) {
+                           int maxContentLength, HttpResponseStatus responseStatus,
+                           String responseBody) {
         this.messageHandler = messageHandler;
         this.executorGroup = executorGroup;
         this.maxContentLength = maxContentLength;
         this.responseStatus = responseStatus;
+        this.responseBody = responseBody;
     }
 
     protected void initChannel(SocketChannel socketChannel) throws Exception {
@@ -40,7 +43,7 @@ protected void initChannel(SocketChannel socketChannel) throws Exception {
         pipeline.addLast(new HttpServerCodec());
         pipeline.addLast(new HttpContentDecompressor());
         pipeline.addLast(new HttpObjectAggregator(maxContentLength));
-        pipeline.addLast(new HttpServerHandler(messageHandler.copy(), executorGroup, responseStatus));
+        pipeline.addLast(new HttpServerHandler(messageHandler.copy(), executorGroup, responseStatus, responseBody));
     }
 
     public void enableSSL(SslHandlerProvider sslHandlerProvider) {

src/main/java/org/logstash/plugins/inputs/http/HttpServerHandler.java

@@ -23,19 +23,21 @@ public class HttpServerHandler extends SimpleChannelInboundHandler<FullHttpReque
     private final IMessageHandler messageHandler;
     private final ThreadPoolExecutor executorGroup;
     private final HttpResponseStatus responseStatus;
+    private final String responseBody;
 
     public HttpServerHandler(IMessageHandler messageHandler, ThreadPoolExecutor executorGroup,
-                             HttpResponseStatus responseStatus) {
+                             HttpResponseStatus responseStatus, String responseBody) {
         this.messageHandler = messageHandler;
         this.executorGroup = executorGroup;
         this.responseStatus = responseStatus;
+        this.responseBody = responseBody;
     }
 
     @Override
     public void channelRead0(ChannelHandlerContext ctx, FullHttpRequest msg) {
         final String remoteAddress = ((InetSocketAddress) ctx.channel().remoteAddress()).getAddress().getHostAddress();
         msg.retain();
-        final MessageProcessor messageProcessor = new MessageProcessor(ctx, msg, remoteAddress, messageHandler, responseStatus);
+        final MessageProcessor messageProcessor = new MessageProcessor(ctx, msg, remoteAddress, messageHandler, responseStatus, responseBody);
         executorGroup.execute(messageProcessor);
     }
 

src/main/java/org/logstash/plugins/inputs/http/MessageProcessor.java

@@ -23,15 +23,18 @@ public class MessageProcessor implements RejectableRunnable {
     private final String remoteAddress;
     private final IMessageHandler messageHandler;
     private final HttpResponseStatus responseStatus;
+    private final String responseBody;
     private static final Charset charset = Charset.forName("UTF-8");
 
     MessageProcessor(ChannelHandlerContext ctx, FullHttpRequest req, String remoteAddress,
-                            IMessageHandler messageHandler, HttpResponseStatus responseStatus) {
+                            IMessageHandler messageHandler, HttpResponseStatus responseStatus,
+                            String responseBody) {
         this.ctx = ctx;
         this.req = req;
         this.remoteAddress = remoteAddress;
         this.messageHandler = messageHandler;
         this.responseStatus = responseStatus;
+        this.responseBody = responseBody;
     }
 
     public void onRejection() {
@@ -88,9 +91,8 @@ private FullHttpResponse generateResponse(Map<String, String> stringHeaders) {
         response.headers().set(headers);
 
         if (responseStatus != HttpResponseStatus.NO_CONTENT) {
-            final ByteBuf payload = Unpooled.wrappedBuffer("ok".getBytes(charset));
+            final ByteBuf payload = Unpooled.wrappedBuffer(this.responseBody.getBytes(charset));
             response.headers().set(HttpHeaderNames.CONTENT_LENGTH, payload.readableBytes());
-            response.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/plain");
             response.content().writeBytes(payload);
         }
 

src/main/java/org/logstash/plugins/inputs/http/NettyHttpServer.java

@@ -32,7 +32,8 @@ public class NettyHttpServer implements Runnable, Closeable {
 
     public NettyHttpServer(String host, int port, IMessageHandler messageHandler,
                            SslHandlerProvider sslHandlerProvider, int threads,
-                           int maxPendingRequests, int maxContentLength, int responseCode)
+                           int maxPendingRequests, int maxContentLength, int responseCode,
+                           String responseBody)
     {
         this.host = host;
         this.port = port;
@@ -44,7 +45,7 @@ public NettyHttpServer(String host, int port, IMessageHandler messageHandler,
                 new CustomRejectedExecutionHandler());
 
         final HttpInitializer httpInitializer = new HttpInitializer(messageHandler, executorGroup,
-                                                                      maxContentLength, responseStatus);
+                                                                      maxContentLength, responseStatus, responseBody);
 
         if (sslHandlerProvider != null) {
             httpInitializer.enableSSL(sslHandlerProvider);