Have rest request also allow to get the content as a streamable

(note, this will break plugins that implement a rest handler..., like wares, thrift, and memcached, until they are upgraded)

src/main/java/org/elasticsearch/http/netty/NettyHttpRequest.java

@@ -20,9 +20,11 @@
 package org.elasticsearch.http.netty;
 
 import com.google.common.base.Charsets;
+import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.http.HttpRequest;
 import org.elasticsearch.rest.support.AbstractRestRequest;
 import org.elasticsearch.rest.support.RestUtils;
+import org.elasticsearch.transport.netty.ChannelBufferStreamInputFactory;
 import org.jboss.netty.handler.codec.http.HttpMethod;
 
 import java.util.HashMap;
@@ -101,6 +103,11 @@ public class NettyHttpRequest extends AbstractRestRequest implements HttpRequest
         return request.getContent().readableBytes() > 0;
     }
 
+    @Override
+    public StreamInput contentStream() {
+        return ChannelBufferStreamInputFactory.create(request.getContent());
+    }
+
     @Override
     public int contentLength() {
         return request.getContent().readableBytes();

src/main/java/org/elasticsearch/rest/RestRequest.java

@@ -19,6 +19,7 @@
 
 package org.elasticsearch.rest;
 
+import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.xcontent.ToXContent;
@@ -58,6 +59,11 @@ public interface RestRequest extends ToXContent.Params {
      */
     boolean contentUnsafe();
 
+    /**
+     * The content as a stream.
+     */
+    StreamInput contentStream();
+
     byte[] contentByteArray();
 
     int contentByteArrayOffset();

src/main/java/org/elasticsearch/transport/netty/ChannelBufferStreamInputFactory.java

@@ -0,0 +1,17 @@
+package org.elasticsearch.transport.netty;
+
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.jboss.netty.buffer.ChannelBuffer;
+
+/**
+ */
+public class ChannelBufferStreamInputFactory {
+
+    public static StreamInput create(ChannelBuffer buffer) {
+        return new ChannelBufferStreamInput(buffer, buffer.readableBytes());
+    }
+
+    public static StreamInput create(ChannelBuffer buffer, int size) {
+        return new ChannelBufferStreamInput(buffer, size);
+    }
+}

src/main/java/org/elasticsearch/transport/netty/MessageChannelHandler.java

@@ -212,7 +212,7 @@ public class MessageChannelHandler extends SimpleChannelUpstreamHandler {
 
         // netty always copies a buffer, either in NioWorker in its read handler, where it copies to a fresh
         // buffer, or in the cumlation buffer, which is cleaned each time
-        StreamInput streamIn = new ChannelBufferStreamInput(buffer, size);
+        StreamInput streamIn = ChannelBufferStreamInputFactory.create(buffer, size);
 
         long requestId = buffer.readLong();
         byte status = buffer.readByte();