introduce TransportRequest (with optional headers)

introduce a new class, TransportRequest, which includes headers. This class can be used when sending requests over the transport layer, and ActionRequest also extends it now.
This is the first phase of the refactoring part in the transport layer and action layer to allow for simpler implementations of those as well as simpler "filtering" capabilities in the future

Author: kimchy

src/main/java/org/elasticsearch/action/Action.java

@@ -24,7 +24,7 @@
 /**
  * Main action (used with {@link Client} API.
  */
-public abstract class Action<Request extends ActionRequest, Response extends ActionResponse, RequestBuilder extends ActionRequestBuilder<Request, Response>>
+public abstract class Action<Request extends ActionRequest, Response extends ActionResponse, RequestBuilder extends ActionRequestBuilder<Request, Response, RequestBuilder>>
         extends GenericAction<Request, Response> {
 
     protected Action(String name) {

src/main/java/org/elasticsearch/action/ActionRequest.java

@@ -19,25 +19,58 @@
 
 package org.elasticsearch.action;
 
-import org.elasticsearch.common.io.stream.Streamable;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.transport.TransportRequest;
+
+import java.io.IOException;
 
 /**
  *
  */
-public interface ActionRequest extends Streamable {
+public abstract class ActionRequest<T extends ActionRequest> extends TransportRequest {
+
+    private boolean listenerThreaded = false;
+
+    protected ActionRequest() {
+        super();
+    }
 
-    ActionRequestValidationException validate();
+    protected ActionRequest(ActionRequest request) {
+        super(request);
+        // this does not set the listenerThreaded API, if needed, its up to the caller to set it
+        // since most times, we actually want it to not be threaded...
+        //this.listenerThreaded = request.listenerThreaded();
+    }
 
     /**
      * Should the response listener be executed on a thread or not.
      * <p/>
      * <p>When not executing on a thread, it will either be executed on the calling thread, or
      * on an expensive, IO based, thread.
      */
-    boolean listenerThreaded();
+    public final boolean listenerThreaded() {
+        return this.listenerThreaded;
+    }
 
     /**
      * Sets if the response listener be executed on a thread or not.
      */
-    ActionRequest listenerThreaded(boolean listenerThreaded);
+    @SuppressWarnings("unchecked")
+    public final T listenerThreaded(boolean listenerThreaded) {
+        this.listenerThreaded = listenerThreaded;
+        return (T) this;
+    }
+
+    public abstract ActionRequestValidationException validate();
+
+    @Override
+    public void readFrom(StreamInput in) throws IOException {
+        super.readFrom(in);
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        super.writeTo(out);
+    }
 }

src/main/java/org/elasticsearch/action/ActionRequestBuilder.java

@@ -19,14 +19,48 @@
 
 package org.elasticsearch.action;
 
+import org.elasticsearch.action.support.PlainListenableActionFuture;
+import org.elasticsearch.client.internal.InternalGenericClient;
+
 /**
  *
  */
-public interface ActionRequestBuilder<Request extends ActionRequest, Response extends ActionResponse> {
+public abstract class ActionRequestBuilder<Request extends ActionRequest, Response extends ActionResponse, RequestBuilder extends ActionRequestBuilder> {
+
+    protected final Request request;
+
+    protected final InternalGenericClient client;
+
+    protected ActionRequestBuilder(InternalGenericClient client, Request request) {
+        this.client = client;
+        this.request = request;
+    }
+
+    public Request request() {
+        return this.request;
+    }
+
+    @SuppressWarnings("unchecked")
+    public final RequestBuilder setListenerThreaded(boolean listenerThreaded) {
+        request.listenerThreaded(listenerThreaded);
+        return (RequestBuilder) this;
+    }
+
+    @SuppressWarnings("unchecked")
+    public final RequestBuilder putHeader(String key, Object value) {
+        request.putHeader(key, value);
+        return (RequestBuilder) this;
+    }
 
-    Request request();
+    public ListenableActionFuture<Response> execute() {
+        PlainListenableActionFuture<Response> future = new PlainListenableActionFuture<Response>(request.listenerThreaded(), client.threadPool());
+        execute(future);
+        return future;
+    }
 
-    ListenableActionFuture<Response> execute();
+    public void execute(ActionListener<Response> listener) {
+        doExecute(listener);
+    }
 
-    void execute(ActionListener<Response> listener);
+    protected abstract void doExecute(ActionListener<Response> listener);
 }

src/main/java/org/elasticsearch/action/admin/cluster/ClusterAction.java

@@ -28,7 +28,7 @@
 /**
  * Cluster action (used with {@link ClusterAdminClient} API.
  */
-public abstract class ClusterAction<Request extends ActionRequest, Response extends ActionResponse, RequestBuilder extends ActionRequestBuilder<Request, Response>>
+public abstract class ClusterAction<Request extends ActionRequest, Response extends ActionResponse, RequestBuilder extends ActionRequestBuilder<Request, Response, RequestBuilder>>
         extends GenericAction<Request, Response> {
 
     protected ClusterAction(String name) {

src/main/java/org/elasticsearch/action/admin/cluster/health/ClusterHealthRequest.java

@@ -34,7 +34,7 @@
 /**
  *
  */
-public class ClusterHealthRequest extends MasterNodeOperationRequest {
+public class ClusterHealthRequest extends MasterNodeOperationRequest<ClusterHealthRequest> {
 
     private String[] indices;
 
@@ -141,7 +141,7 @@ public void readFrom(StreamInput in) throws IOException {
         } else {
             indices = new String[size];
             for (int i = 0; i < indices.length; i++) {
-                indices[i] = in.readUTF();
+                indices[i] = in.readString();
             }
         }
         timeout = readTimeValue(in);
@@ -150,7 +150,7 @@ public void readFrom(StreamInput in) throws IOException {
         }
         waitForRelocatingShards = in.readInt();
         waitForActiveShards = in.readInt();
-        waitForNodes = in.readUTF();
+        waitForNodes = in.readString();
     }
 
     @Override
@@ -161,7 +161,7 @@ public void writeTo(StreamOutput out) throws IOException {
         } else {
             out.writeVInt(indices.length);
             for (String index : indices) {
-                out.writeUTF(index);
+                out.writeString(index);
             }
         }
         timeout.writeTo(out);
@@ -173,6 +173,6 @@ public void writeTo(StreamOutput out) throws IOException {
         }
         out.writeInt(waitForRelocatingShards);
         out.writeInt(waitForActiveShards);
-        out.writeUTF(waitForNodes);
+        out.writeString(waitForNodes);
     }
 }

src/main/java/org/elasticsearch/action/admin/cluster/health/ClusterHealthRequestBuilder.java

@@ -20,32 +20,25 @@
 package org.elasticsearch.action.admin.cluster.health;
 
 import org.elasticsearch.action.ActionListener;
-import org.elasticsearch.action.admin.cluster.support.BaseClusterRequestBuilder;
+import org.elasticsearch.action.support.master.MasterNodeOperationRequestBuilder;
 import org.elasticsearch.client.ClusterAdminClient;
+import org.elasticsearch.client.internal.InternalClusterAdminClient;
 import org.elasticsearch.common.unit.TimeValue;
 
 /**
  *
  */
-public class ClusterHealthRequestBuilder extends BaseClusterRequestBuilder<ClusterHealthRequest, ClusterHealthResponse> {
+public class ClusterHealthRequestBuilder extends MasterNodeOperationRequestBuilder<ClusterHealthRequest, ClusterHealthResponse, ClusterHealthRequestBuilder> {
 
     public ClusterHealthRequestBuilder(ClusterAdminClient clusterClient) {
-        super(clusterClient, new ClusterHealthRequest());
+        super((InternalClusterAdminClient) clusterClient, new ClusterHealthRequest());
     }
 
     public ClusterHealthRequestBuilder setIndices(String... indices) {
         request.indices(indices);
         return this;
     }
 
-    /**
-     * Sets the master node timeout in case the master has not yet been discovered.
-     */
-    public ClusterHealthRequestBuilder setMasterNodeTimeout(TimeValue timeout) {
-        request.masterNodeTimeout(timeout);
-        return this;
-    }
-
     public ClusterHealthRequestBuilder setTimeout(TimeValue timeout) {
         request.timeout(timeout);
         return this;
@@ -91,6 +84,6 @@ public ClusterHealthRequestBuilder setWaitForNodes(String waitForNodes) {
 
     @Override
     protected void doExecute(ActionListener<ClusterHealthResponse> listener) {
-        client.health(request, listener);
+        ((ClusterAdminClient) client).health(request, listener);
     }
 }

src/main/java/org/elasticsearch/action/admin/cluster/node/hotthreads/NodesHotThreadsRequest.java

@@ -29,7 +29,7 @@
 
 /**
  */
-public class NodesHotThreadsRequest extends NodesOperationRequest {
+public class NodesHotThreadsRequest extends NodesOperationRequest<NodesHotThreadsRequest> {
 
     int threads = 3;
     String type = "cpu";

src/main/java/org/elasticsearch/action/admin/cluster/node/hotthreads/NodesHotThreadsRequestBuilder.java

@@ -20,21 +20,17 @@
 package org.elasticsearch.action.admin.cluster.node.hotthreads;
 
 import org.elasticsearch.action.ActionListener;
-import org.elasticsearch.action.admin.cluster.support.BaseClusterRequestBuilder;
+import org.elasticsearch.action.support.nodes.NodesOperationRequestBuilder;
 import org.elasticsearch.client.ClusterAdminClient;
+import org.elasticsearch.client.internal.InternalClusterAdminClient;
 import org.elasticsearch.common.unit.TimeValue;
 
 /**
  */
-public class NodesHotThreadsRequestBuilder extends BaseClusterRequestBuilder<NodesHotThreadsRequest, NodesHotThreadsResponse> {
+public class NodesHotThreadsRequestBuilder extends NodesOperationRequestBuilder<NodesHotThreadsRequest, NodesHotThreadsResponse, NodesHotThreadsRequestBuilder> {
 
     public NodesHotThreadsRequestBuilder(ClusterAdminClient clusterClient) {
-        super(clusterClient, new NodesHotThreadsRequest());
-    }
-
-    public NodesHotThreadsRequestBuilder setNodesIds(String... nodesIds) {
-        request.nodesIds(nodesIds);
-        return this;
+        super((InternalClusterAdminClient) clusterClient, new NodesHotThreadsRequest());
     }
 
     public NodesHotThreadsRequestBuilder setThreads(int threads) {
@@ -54,6 +50,6 @@ public NodesHotThreadsRequestBuilder setInterval(TimeValue interval) {
 
     @Override
     protected void doExecute(ActionListener<NodesHotThreadsResponse> listener) {
-        client.nodesHotThreads(request, listener);
+        ((ClusterAdminClient) client).nodesHotThreads(request, listener);
     }
 }

src/main/java/org/elasticsearch/action/admin/cluster/node/hotthreads/TransportNodesHotThreadsAction.java

@@ -117,7 +117,7 @@ static class NodeRequest extends NodeOperationRequest {
         }
 
         NodeRequest(String nodeId, NodesHotThreadsRequest request) {
-            super(nodeId);
+            super(request, nodeId);
             this.request = request;
         }
 

src/main/java/org/elasticsearch/action/admin/cluster/node/info/NodesInfoRequest.java

@@ -28,7 +28,7 @@
 /**
  * A request to get node (cluster) level information.
  */
-public class NodesInfoRequest extends NodesOperationRequest {
+public class NodesInfoRequest extends NodesOperationRequest<NodesInfoRequest> {
 
     private boolean settings = false;
     private boolean os = false;