Adds response_body to configuration options

docs/index.asciidoc

@@ -25,7 +25,7 @@ Using this input you can receive single or multiline events over http(s).
 Applications can send an HTTP request to the endpoint started by this input and
 Logstash will convert it into an event for subsequent processing. Users 
 can pass plain text, JSON, or any formatted data and use a corresponding codec with this
-input. For Content-Type `application/json` the `json` codec is used, but for all other
+input. For content-type `application/json` the `json` codec is used, but for all other
 data formats, `plain` codec is used.
 
 This input can also be used to receive webhook requests to integrate with other services
@@ -101,6 +101,7 @@ This plugin supports the following configuration options plus the <<plugins-{typ
 | <<plugins-{type}s-{plugin}-max_pending_requests>> |<<number,number>>|No
 | <<plugins-{type}s-{plugin}-response_headers>> |<<hash,hash>>|No
 | <<plugins-{type}s-{plugin}-response_code>> |<<number,number>>, one of `[200, 201, 202, 204]`|No
+| <<plugins-{type}s-{plugin}-response_body>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-ssl>> |<<boolean,boolean>>|No
 | <<plugins-{type}s-{plugin}-ssl_certificate>> |a valid filesystem path|No
 | <<plugins-{type}s-{plugin}-ssl_certificate_authorities>> |<<array,array>>|No
@@ -280,11 +281,22 @@ 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 text body of the response returned to clients on successful POSTs.
+The default `ok` can be overriden with any string content such as JSON.
+
+If 204 (No Content) is set, the `response_body` option is ignored.
+
 [id="plugins-{type}s-{plugin}-response_headers"]
 ===== `response_headers` 
 
   * Value type is <<hash,hash>>
-  * Default value is `{"Content-Type"=>"text/plain"}`
+  * Default value is `{"content-type"=>"text/plain"}`
 
 specify a custom set of response headers
 

lib/logstash/inputs/http.rb

@@ -98,8 +98,14 @@ class LogStash::Inputs::Http < LogStash::Inputs::Base
   # and no codec for the request's content-type is found
   config :additional_codecs, :validate => :hash, :default => { "application/json" => "json" }
 
-  # specify a custom set of response headers
-  config :response_headers, :validate => :hash, :default => { 'Content-Type' => 'text/plain' }
+  # Send reponses with this HTTP code. 204 is `no content`, and forces an empty response body
+  config :response_code, :validate => [200, 201, 202, 204], :default => 200
+
+  # Send this as the body to each HTTP POST. A JSON example: `'{"ok": true}'`.
+  config :response_body, :default => "ok"
+
+  # specify a custom set of response headers (use lowercase keys, per netty.io)
+  config :response_headers, :validate => :hash, :default => { 'content-type' => 'text/plain' }
 
   # target field for the client host of the http request
   config :remote_host_target_field, :validate => :string
@@ -113,8 +119,6 @@ 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
-
   # Deprecated options
 
   # The JKS keystore to validate the client's certificates
@@ -282,7 +286,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

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

@@ -20,14 +20,16 @@ public class HttpInitializer extends ChannelInitializer<SocketChannel> {
     private SslHandlerProvider sslHandlerProvider;
     private final int maxContentLength;
     private final HttpResponseStatus responseStatus;
+    private final String responseBody;
     private final ThreadPoolExecutor executorGroup;
 
     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 +42,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

@@ -25,17 +25,19 @@ 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 UTF8_CHARSET = Charset.forName("UTF-8");
     private final static Logger LOGGER = LogManager.getLogger(MessageHandler.class);
 
     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() {
@@ -100,15 +102,19 @@ private FullHttpResponse generateResponse(Map<String, String> stringHeaders) {
                 req.protocolVersion(),
                 responseStatus);
         final DefaultHttpHeaders headers = new DefaultHttpHeaders();
+        boolean hasContentTypeHeader = false;
         for(String key : stringHeaders.keySet()) {
             headers.set(key, stringHeaders.get(key));
+            hasContentTypeHeader = (HttpHeaderNames.CONTENT_TYPE.contentEqualsIgnoreCase(key));
+        }
+        if (!hasContentTypeHeader) {
+            headers.set(HttpHeaderNames.CONTENT_TYPE, "text/plain");
         }
         response.headers().set(headers);
 
         if (responseStatus != HttpResponseStatus.NO_CONTENT) {
-            final ByteBuf payload = Unpooled.wrappedBuffer("ok".getBytes(UTF8_CHARSET));
+            final ByteBuf payload = Unpooled.wrappedBuffer(responseBody.getBytes(UTF8_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,10 +32,12 @@ 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;
+
         this.responseStatus = HttpResponseStatus.valueOf(responseCode);
         processorGroup = new NioEventLoopGroup(threads, daemonThreadFactory("http-input-processor"));
 
@@ -44,7 +46,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);