refactor internal transport streams to be more effecient, heavily reduce stream size

.idea/dictionaries/kimchy.xml

@@ -35,6 +35,7 @@
       <w>mmap</w>
       <w>multi</w>
       <w>nanos</w>
+      <w>newcount</w>
       <w>ngram</w>
       <w>param</w>
       <w>params</w>

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

@@ -19,7 +19,7 @@
 
 package org.elasticsearch.action;
 
-import org.elasticsearch.util.io.Streamable;
+import org.elasticsearch.util.io.stream.Streamable;
 
 /**
  * @author kimchy (Shay Banon)

modules/elasticsearch/src/main/java/org/elasticsearch/action/ActionResponse.java

@@ -19,7 +19,7 @@
 
 package org.elasticsearch.action;
 
-import org.elasticsearch.util.io.Streamable;
+import org.elasticsearch.util.io.stream.Streamable;
 
 /**
  * @author kimchy (Shay Banon)

modules/elasticsearch/src/main/java/org/elasticsearch/action/ShardOperationFailedException.java

@@ -19,7 +19,7 @@
 
 package org.elasticsearch.action;
 
-import org.elasticsearch.util.io.Streamable;
+import org.elasticsearch.util.io.stream.Streamable;
 
 /**
  * An exception indicating that a failure occurred performing an operation on the shard.

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

@@ -23,9 +23,9 @@ import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.support.master.MasterNodeOperationRequest;
 import org.elasticsearch.util.Strings;
 import org.elasticsearch.util.TimeValue;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 import java.util.concurrent.TimeUnit;
 
@@ -94,9 +94,9 @@ public class ClusterHealthRequest extends MasterNodeOperationRequest {
         return null;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
-        int size = in.readInt();
+        int size = in.readVInt();
         if (size == 0) {
             indices = Strings.EMPTY_ARRAY;
         } else {
@@ -112,12 +112,12 @@ public class ClusterHealthRequest extends MasterNodeOperationRequest {
         waitForRelocatingShards = in.readInt();
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
         if (indices == null) {
-            out.writeInt(0);
+            out.writeVInt(0);
         } else {
-            out.writeInt(indices.length);
+            out.writeVInt(indices.length);
             for (String index : indices) {
                 out.writeUTF(index);
             }

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/cluster/health/ClusterHealthResponse.java

@@ -22,9 +22,9 @@ package org.elasticsearch.action.admin.cluster.health;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Maps;
 import org.elasticsearch.action.ActionResponse;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 import java.util.Iterator;
 import java.util.List;
@@ -116,19 +116,19 @@ public class ClusterHealthResponse implements ActionResponse, Iterable<ClusterIn
         return indices.values().iterator();
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         clusterName = in.readUTF();
-        activePrimaryShards = in.readInt();
+        activePrimaryShards = in.readVInt();
-        activeShards = in.readInt();
+        activeShards = in.readVInt();
-        relocatingShards = in.readInt();
+        relocatingShards = in.readVInt();
         status = ClusterHealthStatus.fromValue(in.readByte());
-        int size = in.readInt();
+        int size = in.readVInt();
         for (int i = 0; i < size; i++) {
             ClusterIndexHealth indexHealth = readClusterIndexHealth(in);
             indices.put(indexHealth.index(), indexHealth);
         }
         timedOut = in.readBoolean();
-        size = in.readInt();
+        size = in.readVInt();
         if (size == 0) {
             validationFailures = ImmutableList.of();
         } else {
@@ -138,19 +138,19 @@ public class ClusterHealthResponse implements ActionResponse, Iterable<ClusterIn
         }
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         out.writeUTF(clusterName);
-        out.writeInt(activePrimaryShards);
+        out.writeVInt(activePrimaryShards);
-        out.writeInt(activeShards);
+        out.writeVInt(activeShards);
-        out.writeInt(relocatingShards);
+        out.writeVInt(relocatingShards);
         out.writeByte(status.value());
-        out.writeInt(indices.size());
+        out.writeVInt(indices.size());
         for (ClusterIndexHealth indexHealth : this) {
             indexHealth.writeTo(out);
         }
         out.writeBoolean(timedOut);
 
-        out.writeInt(validationFailures.size());
+        out.writeVInt(validationFailures.size());
         for (String failure : validationFailures) {
             out.writeUTF(failure);
         }

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/cluster/health/ClusterIndexHealth.java

@@ -21,10 +21,10 @@ package org.elasticsearch.action.admin.cluster.health;
 
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Maps;
-import org.elasticsearch.util.io.Streamable;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
+import org.elasticsearch.util.io.stream.Streamable;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 import java.util.Iterator;
 import java.util.List;
@@ -105,27 +105,27 @@ public class ClusterIndexHealth implements Iterable<ClusterShardHealth>, Streama
         return shards.values().iterator();
     }
 
-    public static ClusterIndexHealth readClusterIndexHealth(DataInput in) throws IOException, ClassNotFoundException {
+    public static ClusterIndexHealth readClusterIndexHealth(StreamInput in) throws IOException {
         ClusterIndexHealth indexHealth = new ClusterIndexHealth();
         indexHealth.readFrom(in);
         return indexHealth;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         index = in.readUTF();
-        numberOfShards = in.readInt();
+        numberOfShards = in.readVInt();
-        numberOfReplicas = in.readInt();
+        numberOfReplicas = in.readVInt();
-        activePrimaryShards = in.readInt();
+        activePrimaryShards = in.readVInt();
-        activeShards = in.readInt();
+        activeShards = in.readVInt();
-        relocatingShards = in.readInt();
+        relocatingShards = in.readVInt();
         status = ClusterHealthStatus.fromValue(in.readByte());
 
-        int size = in.readInt();
+        int size = in.readVInt();
         for (int i = 0; i < size; i++) {
             ClusterShardHealth shardHealth = readClusterShardHealth(in);
             shards.put(shardHealth.id(), shardHealth);
         }
-        size = in.readInt();
+        size = in.readVInt();
         if (size == 0) {
             validationFailures = ImmutableList.of();
         } else {
@@ -135,21 +135,21 @@ public class ClusterIndexHealth implements Iterable<ClusterShardHealth>, Streama
         }
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         out.writeUTF(index);
-        out.writeInt(numberOfShards);
+        out.writeVInt(numberOfShards);
-        out.writeInt(numberOfReplicas);
+        out.writeVInt(numberOfReplicas);
-        out.writeInt(activePrimaryShards);
+        out.writeVInt(activePrimaryShards);
-        out.writeInt(activeShards);
+        out.writeVInt(activeShards);
-        out.writeInt(relocatingShards);
+        out.writeVInt(relocatingShards);
         out.writeByte(status.value());
 
-        out.writeInt(shards.size());
+        out.writeVInt(shards.size());
         for (ClusterShardHealth shardHealth : this) {
             shardHealth.writeTo(out);
         }
 
-        out.writeInt(validationFailures.size());
+        out.writeVInt(validationFailures.size());
         for (String failure : validationFailures) {
             out.writeUTF(failure);
         }

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/cluster/health/ClusterShardHealth.java

@@ -19,10 +19,10 @@
 
 package org.elasticsearch.action.admin.cluster.health;
 
-import org.elasticsearch.util.io.Streamable;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
+import org.elasticsearch.util.io.stream.Streamable;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
@@ -68,25 +68,25 @@ public class ClusterShardHealth implements Streamable {
         return primaryActive;
     }
 
-    static ClusterShardHealth readClusterShardHealth(DataInput in) throws IOException, ClassNotFoundException {
+    static ClusterShardHealth readClusterShardHealth(StreamInput in) throws IOException {
         ClusterShardHealth ret = new ClusterShardHealth();
         ret.readFrom(in);
         return ret;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
-        shardId = in.readInt();
+        shardId = in.readVInt();
         status = ClusterHealthStatus.fromValue(in.readByte());
-        activeShards = in.readInt();
+        activeShards = in.readVInt();
-        relocatingShards = in.readInt();
+        relocatingShards = in.readVInt();
         primaryActive = in.readBoolean();
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
-        out.writeInt(shardId);
+        out.writeVInt(shardId);
         out.writeByte(status.value());
-        out.writeInt(activeShards);
+        out.writeVInt(activeShards);
-        out.writeInt(relocatingShards);
+        out.writeVInt(relocatingShards);
         out.writeBoolean(primaryActive);
     }
 }

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

@@ -22,11 +22,11 @@ package org.elasticsearch.action.admin.cluster.node.info;
 import com.google.common.collect.ImmutableMap;
 import org.elasticsearch.action.support.nodes.NodeOperationResponse;
 import org.elasticsearch.cluster.node.Node;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 import org.elasticsearch.util.settings.ImmutableSettings;
 import org.elasticsearch.util.settings.Settings;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 import java.util.Map;
 
@@ -60,16 +60,16 @@ public class NodeInfo extends NodeOperationResponse {
         return this.settings;
     }
 
-    public static NodeInfo readNodeInfo(DataInput in) throws ClassNotFoundException, IOException {
+    public static NodeInfo readNodeInfo(StreamInput in) throws IOException {
         NodeInfo nodeInfo = new NodeInfo();
         nodeInfo.readFrom(in);
         return nodeInfo;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
         ImmutableMap.Builder<String, String> builder = ImmutableMap.builder();
-        int size = in.readInt();
+        int size = in.readVInt();
         for (int i = 0; i < size; i++) {
             builder.put(in.readUTF(), in.readUTF());
         }
@@ -77,9 +77,9 @@ public class NodeInfo extends NodeOperationResponse {
         settings = ImmutableSettings.readSettingsFromStream(in);
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
-        out.writeInt(attributes.size());
+        out.writeVInt(attributes.size());
         for (Map.Entry<String, String> entry : attributes.entrySet()) {
             out.writeUTF(entry.getKey());
             out.writeUTF(entry.getValue());

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

@@ -21,9 +21,9 @@ package org.elasticsearch.action.admin.cluster.node.info;
 
 import org.elasticsearch.action.support.nodes.NodesOperationResponse;
 import org.elasticsearch.cluster.ClusterName;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
@@ -38,17 +38,17 @@ public class NodesInfoResponse extends NodesOperationResponse<NodeInfo> {
         super(clusterName, nodes);
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
-        nodes = new NodeInfo[in.readInt()];
+        nodes = new NodeInfo[in.readVInt()];
         for (int i = 0; i < nodes.length; i++) {
             nodes[i] = NodeInfo.readNodeInfo(in);
         }
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
-        out.writeInt(nodes.length);
+        out.writeVInt(nodes.length);
         for (NodeInfo node : nodes) {
             node.writeTo(out);
         }

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/cluster/node/shutdown/NodesShutdownRequest.java

@@ -21,9 +21,9 @@ package org.elasticsearch.action.admin.cluster.node.shutdown;
 
 import org.elasticsearch.action.support.nodes.NodesOperationRequest;
 import org.elasticsearch.util.TimeValue;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 import static org.elasticsearch.util.TimeValue.*;
@@ -60,12 +60,12 @@ public class NodesShutdownRequest extends NodesOperationRequest {
         return this.delay;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
         delay = readTimeValue(in);
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
         delay.writeTo(out);
     }

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/cluster/node/shutdown/NodesShutdownResponse.java

@@ -23,13 +23,13 @@ import org.elasticsearch.action.support.nodes.NodeOperationResponse;
 import org.elasticsearch.action.support.nodes.NodesOperationResponse;
 import org.elasticsearch.cluster.ClusterName;
 import org.elasticsearch.cluster.node.Node;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
- * @author kimchy (Shay Banon)
+ * @author kimchy (shay.banon)
  */
 public class NodesShutdownResponse extends NodesOperationResponse<NodesShutdownResponse.NodeShutdownResponse> {
 
@@ -40,17 +40,17 @@ public class NodesShutdownResponse extends NodesOperationResponse<NodesShutdownR
         super(clusterName, nodes);
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
-        nodes = new NodeShutdownResponse[in.readInt()];
+        nodes = new NodeShutdownResponse[in.readVInt()];
         for (int i = 0; i < nodes.length; i++) {
             nodes[i] = NodeShutdownResponse.readNodeShutdownResponse(in);
         }
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
-        out.writeInt(nodes.length);
+        out.writeVInt(nodes.length);
         for (NodeShutdownResponse node : nodes) {
             node.writeTo(out);
         }
@@ -65,7 +65,7 @@ public class NodesShutdownResponse extends NodesOperationResponse<NodesShutdownR
             super(node);
         }
 
-        public static NodeShutdownResponse readNodeShutdownResponse(DataInput in) throws ClassNotFoundException, IOException {
+        public static NodeShutdownResponse readNodeShutdownResponse(StreamInput in) throws IOException {
             NodeShutdownResponse res = new NodeShutdownResponse();
             res.readFrom(in);
             return res;

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/cluster/node/shutdown/TransportNodesShutdown.java

@@ -31,10 +31,10 @@ import org.elasticsearch.server.Server;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.util.TimeValue;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 import org.elasticsearch.util.settings.Settings;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
@@ -124,12 +124,12 @@ public class TransportNodesShutdown extends TransportNodesOperationAction<NodesS
             this.delay = delay;
         }
 
-        @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+        @Override public void readFrom(StreamInput in) throws IOException {
             super.readFrom(in);
             delay = readTimeValue(in);
         }
 
-        @Override public void writeTo(DataOutput out) throws IOException {
+        @Override public void writeTo(StreamOutput out) throws IOException {
             super.writeTo(out);
             delay.writeTo(out);
         }

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/cluster/ping/broadcast/BroadcastPingRequest.java

@@ -21,9 +21,9 @@ package org.elasticsearch.action.admin.cluster.ping.broadcast;
 
 import org.elasticsearch.action.support.broadcast.BroadcastOperationRequest;
 import org.elasticsearch.action.support.broadcast.BroadcastOperationThreading;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
@@ -53,11 +53,11 @@ public class BroadcastPingRequest extends BroadcastOperationRequest {
         return this;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
     }
 }

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/cluster/ping/broadcast/BroadcastPingResponse.java

@@ -21,9 +21,9 @@ package org.elasticsearch.action.admin.cluster.ping.broadcast;
 
 import org.elasticsearch.action.ShardOperationFailedException;
 import org.elasticsearch.action.support.broadcast.BroadcastOperationResponse;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 import java.util.List;
 
@@ -40,11 +40,11 @@ public class BroadcastPingResponse extends BroadcastOperationResponse {
         super(successfulShards, failedShards, shardFailures);
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
     }
 }

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/cluster/ping/broadcast/BroadcastShardPingRequest.java

@@ -20,9 +20,9 @@
 package org.elasticsearch.action.admin.cluster.ping.broadcast;
 
 import org.elasticsearch.action.support.broadcast.BroadcastShardOperationRequest;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
@@ -38,11 +38,11 @@ public class BroadcastShardPingRequest extends BroadcastShardOperationRequest {
         super(index, shardId);
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
     }
 }

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/cluster/ping/broadcast/BroadcastShardPingResponse.java

@@ -20,9 +20,9 @@
 package org.elasticsearch.action.admin.cluster.ping.broadcast;
 
 import org.elasticsearch.action.support.broadcast.BroadcastShardOperationResponse;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
@@ -38,11 +38,11 @@ public class BroadcastShardPingResponse extends BroadcastShardOperationResponse
         super(index, shardId);
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
     }
 }

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/cluster/ping/replication/IndexReplicationPingRequest.java

@@ -21,9 +21,9 @@ package org.elasticsearch.action.admin.cluster.ping.replication;
 
 import org.elasticsearch.action.support.replication.IndexReplicationOperationRequest;
 import org.elasticsearch.util.TimeValue;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
@@ -48,11 +48,11 @@ public class IndexReplicationPingRequest extends IndexReplicationOperationReques
         return this;
     }
 
-    public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
     }
 
-    public void writeTo(DataOutput out) throws IOException {
+    public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
     }
 }

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/cluster/ping/replication/IndexReplicationPingResponse.java

@@ -20,10 +20,10 @@
 package org.elasticsearch.action.admin.cluster.ping.replication;
 
 import org.elasticsearch.action.ActionResponse;
-import org.elasticsearch.util.io.Streamable;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
+import org.elasticsearch.util.io.stream.Streamable;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
@@ -63,15 +63,15 @@ public class IndexReplicationPingResponse implements ActionResponse, Streamable
         return successfulShards + failedShards;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         index = in.readUTF();
-        successfulShards = in.readInt();
+        successfulShards = in.readVInt();
-        failedShards = in.readInt();
+        failedShards = in.readVInt();
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         out.writeUTF(index);
-        out.writeInt(successfulShards);
+        out.writeVInt(successfulShards);
-        out.writeInt(failedShards);
+        out.writeVInt(failedShards);
     }
 }

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/cluster/ping/replication/ReplicationPingResponse.java

@@ -20,10 +20,10 @@
 package org.elasticsearch.action.admin.cluster.ping.replication;
 
 import org.elasticsearch.action.ActionResponse;
-import org.elasticsearch.util.io.Streamable;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
+import org.elasticsearch.util.io.stream.Streamable;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 import java.util.HashMap;
 import java.util.Map;
@@ -47,8 +47,8 @@ public class ReplicationPingResponse implements ActionResponse, Streamable {
         return responses.get(index);
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
-        int size = in.readInt();
+        int size = in.readVInt();
         for (int i = 0; i < size; i++) {
             IndexReplicationPingResponse response = new IndexReplicationPingResponse();
             response.readFrom(in);
@@ -56,8 +56,8 @@ public class ReplicationPingResponse implements ActionResponse, Streamable {
         }
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
-        out.writeInt(responses.size());
+        out.writeVInt(responses.size());
         for (IndexReplicationPingResponse response : responses.values()) {
             response.writeTo(out);
         }

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/cluster/ping/replication/ShardReplicationPingRequest.java

@@ -20,9 +20,9 @@
 package org.elasticsearch.action.admin.cluster.ping.replication;
 
 import org.elasticsearch.action.support.replication.ShardReplicationOperationRequest;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
@@ -49,13 +49,13 @@ public class ShardReplicationPingRequest extends ShardReplicationOperationReques
         return this.shardId;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
-        shardId = in.readInt();
+        shardId = in.readVInt();
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
-        out.writeInt(shardId);
+        out.writeVInt(shardId);
     }
 }

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/cluster/ping/replication/ShardReplicationPingResponse.java

@@ -20,10 +20,10 @@
 package org.elasticsearch.action.admin.cluster.ping.replication;
 
 import org.elasticsearch.action.ActionResponse;
-import org.elasticsearch.util.io.Streamable;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
+import org.elasticsearch.util.io.stream.Streamable;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
@@ -35,9 +35,9 @@ public class ShardReplicationPingResponse implements ActionResponse, Streamable
 
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
     }
 }

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/cluster/ping/single/SinglePingRequest.java

@@ -20,9 +20,9 @@
 package org.elasticsearch.action.admin.cluster.ping.single;
 
 import org.elasticsearch.action.support.single.SingleOperationRequest;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
@@ -61,11 +61,11 @@ public class SinglePingRequest extends SingleOperationRequest {
         return this;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
     }
 

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/cluster/ping/single/SinglePingResponse.java

@@ -20,9 +20,9 @@
 package org.elasticsearch.action.admin.cluster.ping.single;
 
 import org.elasticsearch.action.ActionResponse;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
@@ -30,9 +30,9 @@ import java.io.IOException;
  */
 public class SinglePingResponse implements ActionResponse {
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
     }
 }

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/cluster/state/ClusterStateRequest.java

@@ -21,9 +21,9 @@ package org.elasticsearch.action.admin.cluster.state;
 
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.support.master.MasterNodeOperationRequest;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
@@ -38,9 +38,9 @@ public class ClusterStateRequest extends MasterNodeOperationRequest {
         return null;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
     }
 }

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/cluster/state/ClusterStateResponse.java

@@ -21,9 +21,9 @@ package org.elasticsearch.action.admin.cluster.state;
 
 import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
@@ -44,11 +44,11 @@ public class ClusterStateResponse implements ActionResponse {
         return this.clusterState;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         clusterState = ClusterState.Builder.readFrom(in, null, null);
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         ClusterState.Builder.writeTo(clusterState, out);
     }
 }

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/indices/create/CreateIndexRequest.java

@@ -24,11 +24,11 @@ import org.elasticsearch.ElasticSearchIllegalArgumentException;
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.support.master.MasterNodeOperationRequest;
 import org.elasticsearch.util.TimeValue;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 import org.elasticsearch.util.json.JsonBuilder;
 import org.elasticsearch.util.settings.Settings;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
@@ -153,21 +153,21 @@ public class CreateIndexRequest extends MasterNodeOperationRequest {
         return this;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         index = in.readUTF();
         settings = readSettingsFromStream(in);
         timeout = readTimeValue(in);
-        int size = in.readInt();
+        int size = in.readVInt();
         for (int i = 0; i < size; i++) {
             mappings.put(in.readUTF(), in.readUTF());
         }
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         out.writeUTF(index);
         writeSettingsToStream(settings, out);
         timeout.writeTo(out);
-        out.writeInt(mappings.size());
+        out.writeVInt(mappings.size());
         for (Map.Entry<String, String> entry : mappings.entrySet()) {
             out.writeUTF(entry.getKey());
             out.writeUTF(entry.getValue());

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/indices/create/CreateIndexResponse.java

@@ -20,10 +20,10 @@
 package org.elasticsearch.action.admin.indices.create;
 
 import org.elasticsearch.action.ActionResponse;
-import org.elasticsearch.util.io.Streamable;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
+import org.elasticsearch.util.io.stream.Streamable;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
@@ -50,11 +50,11 @@ public class CreateIndexResponse implements ActionResponse, Streamable {
         return acknowledged;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         acknowledged = in.readBoolean();
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         out.writeBoolean(acknowledged);
     }
 }

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/indices/delete/DeleteIndexRequest.java

@@ -22,9 +22,9 @@ package org.elasticsearch.action.admin.indices.delete;
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.support.master.MasterNodeOperationRequest;
 import org.elasticsearch.util.TimeValue;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 import static org.elasticsearch.action.Actions.*;
@@ -83,12 +83,12 @@ public class DeleteIndexRequest extends MasterNodeOperationRequest {
         return this;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         index = in.readUTF();
         timeout = readTimeValue(in);
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         out.writeUTF(index);
         timeout.writeTo(out);
     }

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/indices/delete/DeleteIndexResponse.java

@@ -20,10 +20,10 @@
 package org.elasticsearch.action.admin.indices.delete;
 
 import org.elasticsearch.action.ActionResponse;
-import org.elasticsearch.util.io.Streamable;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
+import org.elasticsearch.util.io.stream.Streamable;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
@@ -50,11 +50,11 @@ public class DeleteIndexResponse implements ActionResponse, Streamable {
         return acknowledged;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         acknowledged = in.readBoolean();
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         out.writeBoolean(acknowledged);
     }
 }

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/indices/flush/FlushRequest.java

@@ -21,9 +21,9 @@ package org.elasticsearch.action.admin.indices.flush;
 
 import org.elasticsearch.action.support.broadcast.BroadcastOperationRequest;
 import org.elasticsearch.action.support.broadcast.BroadcastOperationThreading;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
@@ -87,12 +87,12 @@ public class FlushRequest extends BroadcastOperationRequest {
         return this;
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
         out.writeBoolean(refresh);
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
         refresh = in.readBoolean();
     }

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/indices/flush/FlushResponse.java

@@ -21,9 +21,9 @@ package org.elasticsearch.action.admin.indices.flush;
 
 import org.elasticsearch.action.ShardOperationFailedException;
 import org.elasticsearch.action.support.broadcast.BroadcastOperationResponse;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 import java.util.List;
 
@@ -42,11 +42,11 @@ public class FlushResponse extends BroadcastOperationResponse {
         super(successfulShards, failedShards, shardFailures);
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
     }
 }

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/indices/flush/ShardFlushRequest.java

@@ -20,9 +20,9 @@
 package org.elasticsearch.action.admin.indices.flush;
 
 import org.elasticsearch.action.support.broadcast.BroadcastShardOperationRequest;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
@@ -44,12 +44,12 @@ class ShardFlushRequest extends BroadcastShardOperationRequest {
         return this.refresh;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
         refresh = in.readBoolean();
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
         out.writeBoolean(refresh);
     }

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/indices/flush/ShardFlushResponse.java

@@ -20,9 +20,9 @@
 package org.elasticsearch.action.admin.indices.flush;
 
 import org.elasticsearch.action.support.broadcast.BroadcastShardOperationResponse;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
@@ -38,11 +38,11 @@ class ShardFlushResponse extends BroadcastShardOperationResponse {
         super(index, shardId);
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
     }
 }

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/indices/gateway/snapshot/GatewaySnapshotResponse.java

@@ -20,10 +20,10 @@
 package org.elasticsearch.action.admin.indices.gateway.snapshot;
 
 import org.elasticsearch.action.ActionResponse;
-import org.elasticsearch.util.io.Streamable;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
+import org.elasticsearch.util.io.stream.Streamable;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 import java.util.HashMap;
 import java.util.Iterator;
@@ -60,8 +60,8 @@ public class GatewaySnapshotResponse implements ActionResponse, Streamable, Iter
         return indices.get(index);
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
-        int size = in.readInt();
+        int size = in.readVInt();
         for (int i = 0; i < size; i++) {
             IndexGatewaySnapshotResponse response = new IndexGatewaySnapshotResponse();
             response.readFrom(in);
@@ -69,8 +69,8 @@ public class GatewaySnapshotResponse implements ActionResponse, Streamable, Iter
         }
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
-        out.writeInt(indices.size());
+        out.writeVInt(indices.size());
         for (IndexGatewaySnapshotResponse indexGatewaySnapshotResponse : indices.values()) {
             indexGatewaySnapshotResponse.writeTo(out);
         }

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/indices/gateway/snapshot/IndexGatewaySnapshotRequest.java

@@ -21,9 +21,9 @@ package org.elasticsearch.action.admin.indices.gateway.snapshot;
 
 import org.elasticsearch.action.support.replication.IndexReplicationOperationRequest;
 import org.elasticsearch.util.TimeValue;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
@@ -48,11 +48,11 @@ public class IndexGatewaySnapshotRequest extends IndexReplicationOperationReques
         return this;
     }
 
-    public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
     }
 
-    public void writeTo(DataOutput out) throws IOException {
+    public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
     }
 }

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/indices/gateway/snapshot/IndexGatewaySnapshotResponse.java

@@ -20,10 +20,10 @@
 package org.elasticsearch.action.admin.indices.gateway.snapshot;
 
 import org.elasticsearch.action.ActionResponse;
-import org.elasticsearch.util.io.Streamable;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
+import org.elasticsearch.util.io.stream.Streamable;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
@@ -77,15 +77,15 @@ public class IndexGatewaySnapshotResponse implements ActionResponse, Streamable
         return successfulShards + failedShards;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         index = in.readUTF();
-        successfulShards = in.readInt();
+        successfulShards = in.readVInt();
-        failedShards = in.readInt();
+        failedShards = in.readVInt();
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         out.writeUTF(index);
-        out.writeInt(successfulShards);
+        out.writeVInt(successfulShards);
-        out.writeInt(failedShards);
+        out.writeVInt(failedShards);
     }
 }

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/indices/gateway/snapshot/ShardGatewaySnapshotRequest.java

@@ -20,9 +20,9 @@
 package org.elasticsearch.action.admin.indices.gateway.snapshot;
 
 import org.elasticsearch.action.support.replication.ShardReplicationOperationRequest;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
@@ -49,13 +49,13 @@ class ShardGatewaySnapshotRequest extends ShardReplicationOperationRequest {
         return this.shardId;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
-        shardId = in.readInt();
+        shardId = in.readVInt();
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
-        out.writeInt(shardId);
+        out.writeVInt(shardId);
     }
 }

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/indices/gateway/snapshot/ShardGatewaySnapshotResponse.java

@@ -20,10 +20,10 @@
 package org.elasticsearch.action.admin.indices.gateway.snapshot;
 
 import org.elasticsearch.action.ActionResponse;
-import org.elasticsearch.util.io.Streamable;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
+import org.elasticsearch.util.io.stream.Streamable;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
@@ -35,9 +35,9 @@ class ShardGatewaySnapshotResponse implements ActionResponse, Streamable {
 
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
     }
 }

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/indices/mapping/put/PutMappingRequest.java

@@ -24,10 +24,10 @@ import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.support.master.MasterNodeOperationRequest;
 import org.elasticsearch.util.Required;
 import org.elasticsearch.util.TimeValue;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 import org.elasticsearch.util.json.JsonBuilder;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 import java.util.concurrent.TimeUnit;
 
@@ -163,9 +163,9 @@ public class PutMappingRequest extends MasterNodeOperationRequest {
         return this;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
-        indices = new String[in.readInt()];
+        indices = new String[in.readVInt()];
         for (int i = 0; i < indices.length; i++) {
             indices[i] = in.readUTF();
         }
@@ -177,12 +177,12 @@ public class PutMappingRequest extends MasterNodeOperationRequest {
         ignoreConflicts = in.readBoolean();
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
         if (indices == null) {
-            out.writeInt(0);
+            out.writeVInt(0);
         } else {
-            out.writeInt(indices.length);
+            out.writeVInt(indices.length);
             for (String index : indices) {
                 out.writeUTF(index);
             }

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/indices/mapping/put/PutMappingResponse.java

@@ -20,10 +20,10 @@
 package org.elasticsearch.action.admin.indices.mapping.put;
 
 import org.elasticsearch.action.ActionResponse;
-import org.elasticsearch.util.io.Streamable;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
+import org.elasticsearch.util.io.stream.Streamable;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
@@ -51,11 +51,11 @@ public class PutMappingResponse implements ActionResponse, Streamable {
         return acknowledged;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         acknowledged = in.readBoolean();
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         out.writeBoolean(acknowledged);
     }
 }

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/indices/optimize/OptimizeRequest.java

@@ -21,9 +21,9 @@ package org.elasticsearch.action.admin.indices.optimize;
 
 import org.elasticsearch.action.support.broadcast.BroadcastOperationRequest;
 import org.elasticsearch.action.support.broadcast.BroadcastOperationThreading;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
@@ -157,7 +157,7 @@ public class OptimizeRequest extends BroadcastOperationRequest {
         return this;
     }
 
-    public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
         waitForMerge = in.readBoolean();
         maxNumSegments = in.readInt();
@@ -166,7 +166,7 @@ public class OptimizeRequest extends BroadcastOperationRequest {
         refresh = in.readBoolean();
     }
 
-    public void writeTo(DataOutput out) throws IOException {
+    public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
         out.writeBoolean(waitForMerge);
         out.writeInt(maxNumSegments);

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/indices/optimize/OptimizeResponse.java

@@ -21,9 +21,9 @@ package org.elasticsearch.action.admin.indices.optimize;
 
 import org.elasticsearch.action.ShardOperationFailedException;
 import org.elasticsearch.action.support.broadcast.BroadcastOperationResponse;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 import java.util.List;
 
@@ -42,11 +42,11 @@ public class OptimizeResponse extends BroadcastOperationResponse {
         super(successfulShards, failedShards, shardFailures);
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
     }
 }

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/indices/optimize/ShardOptimizeRequest.java

@@ -20,9 +20,9 @@
 package org.elasticsearch.action.admin.indices.optimize;
 
 import org.elasticsearch.action.support.broadcast.BroadcastShardOperationRequest;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
@@ -72,7 +72,7 @@ class ShardOptimizeRequest extends BroadcastShardOperationRequest {
         return refresh;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
         waitForMerge = in.readBoolean();
         maxNumSegments = in.readInt();
@@ -81,7 +81,7 @@ class ShardOptimizeRequest extends BroadcastShardOperationRequest {
         refresh = in.readBoolean();
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
         out.writeBoolean(waitForMerge);
         out.writeInt(maxNumSegments);

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/indices/optimize/ShardOptimizeResponse.java

@@ -20,9 +20,9 @@
 package org.elasticsearch.action.admin.indices.optimize;
 
 import org.elasticsearch.action.support.broadcast.BroadcastShardOperationResponse;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
@@ -37,11 +37,11 @@ class ShardOptimizeResponse extends BroadcastShardOperationResponse {
         super(index, shardId);
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
     }
 }

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/indices/refresh/RefreshRequest.java

@@ -21,9 +21,9 @@ package org.elasticsearch.action.admin.indices.refresh;
 
 import org.elasticsearch.action.support.broadcast.BroadcastOperationRequest;
 import org.elasticsearch.action.support.broadcast.BroadcastOperationThreading;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
@@ -74,12 +74,12 @@ public class RefreshRequest extends BroadcastOperationRequest {
         return this;
     }
 
-    public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
         waitForOperations = in.readBoolean();
     }
 
-    public void writeTo(DataOutput out) throws IOException {
+    public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
         out.writeBoolean(waitForOperations);
     }

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/indices/refresh/RefreshResponse.java

@@ -21,9 +21,9 @@ package org.elasticsearch.action.admin.indices.refresh;
 
 import org.elasticsearch.action.ShardOperationFailedException;
 import org.elasticsearch.action.support.broadcast.BroadcastOperationResponse;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 import java.util.List;
 
@@ -42,11 +42,11 @@ public class RefreshResponse extends BroadcastOperationResponse {
         super(successfulShards, failedShards, shardFailures);
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
     }
 }

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/indices/refresh/ShardRefreshRequest.java

@@ -20,9 +20,9 @@
 package org.elasticsearch.action.admin.indices.refresh;
 
 import org.elasticsearch.action.support.broadcast.BroadcastShardOperationRequest;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
@@ -49,12 +49,12 @@ class ShardRefreshRequest extends BroadcastShardOperationRequest {
         return this;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
         waitForOperations = in.readBoolean();
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
         out.writeBoolean(waitForOperations);
     }

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/indices/refresh/ShardRefreshResponse.java

@@ -20,9 +20,9 @@
 package org.elasticsearch.action.admin.indices.refresh;
 
 import org.elasticsearch.action.support.broadcast.BroadcastShardOperationResponse;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
@@ -37,11 +37,11 @@ class ShardRefreshResponse extends BroadcastShardOperationResponse {
         super(index, shardId);
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
     }
 }

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/indices/status/IndexShardStatus.java

@@ -70,26 +70,32 @@ public class IndexShardStatus implements Iterable<ShardStatus> {
     public SizeValue storeSize() {
         long bytes = -1;
         for (ShardStatus shard : shards()) {
-            if (shard.storeSize().bytes() != SizeValue.UNKNOWN.bytes()) {
+            if (shard.storeSize() != null) {
                 if (bytes == -1) {
                     bytes = 0;
                 }
                 bytes += shard.storeSize().bytes();
             }
         }
+        if (bytes == -1) {
+            return null;
+        }
         return new SizeValue(bytes);
     }
 
     public SizeValue estimatedFlushableMemorySize() {
         long bytes = -1;
         for (ShardStatus shard : shards()) {
-            if (shard.estimatedFlushableMemorySize().bytes() != SizeValue.UNKNOWN.bytes()) {
+            if (shard.estimatedFlushableMemorySize() != null) {
                 if (bytes == -1) {
                     bytes = 0;
                 }
                 bytes += shard.estimatedFlushableMemorySize().bytes();
             }
         }
+        if (bytes == -1) {
+            return null;
+        }
         return new SizeValue(bytes);
     }
 

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/indices/status/IndexStatus.java

@@ -98,26 +98,32 @@ public class IndexStatus implements Iterable<IndexShardStatus> {
     public SizeValue storeSize() {
         long bytes = -1;
         for (IndexShardStatus shard : this) {
-            if (shard.storeSize().bytes() != SizeValue.UNKNOWN.bytes()) {
+            if (shard.storeSize() != null) {
                 if (bytes == -1) {
                     bytes = 0;
                 }
                 bytes += shard.storeSize().bytes();
             }
         }
+        if (bytes == -1) {
+            return null;
+        }
         return new SizeValue(bytes);
     }
 
     public SizeValue estimatedFlushableMemorySize() {
         long bytes = -1;
         for (IndexShardStatus shard : this) {
-            if (shard.estimatedFlushableMemorySize().bytes() != SizeValue.UNKNOWN.bytes()) {
+            if (shard.estimatedFlushableMemorySize() != null) {
                 if (bytes == -1) {
                     bytes = 0;
                 }
                 bytes += shard.estimatedFlushableMemorySize().bytes();
             }
         }
+        if (bytes == -1) {
+            return null;
+        }
         return new SizeValue(bytes);
     }
 

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/indices/status/IndicesStatusResponse.java

@@ -23,10 +23,10 @@ import com.google.common.collect.ImmutableMap;
 import org.elasticsearch.action.ShardOperationFailedException;
 import org.elasticsearch.action.support.broadcast.BroadcastOperationResponse;
 import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 import org.elasticsearch.util.settings.Settings;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 import java.util.List;
 import java.util.Map;
@@ -87,27 +87,27 @@ public class IndicesStatusResponse extends BroadcastOperationResponse {
         return indicesStatus;
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
-        out.writeInt(shards().length);
+        out.writeVInt(shards().length);
         for (ShardStatus status : shards()) {
             status.writeTo(out);
         }
-        out.writeInt(indicesSettings.size());
+        out.writeVInt(indicesSettings.size());
         for (Map.Entry<String, Settings> entry : indicesSettings.entrySet()) {
             out.writeUTF(entry.getKey());
             writeSettingsToStream(entry.getValue(), out);
         }
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
-        shards = new ShardStatus[in.readInt()];
+        shards = new ShardStatus[in.readVInt()];
         for (int i = 0; i < shards.length; i++) {
             shards[i] = readIndexShardStatus(in);
         }
         indicesSettings = newHashMap();
-        int size = in.readInt();
+        int size = in.readVInt();
         for (int i = 0; i < size; i++) {
             indicesSettings.put(in.readUTF(), readSettingsFromStream(in));
         }

modules/elasticsearch/src/main/java/org/elasticsearch/action/admin/indices/status/ShardStatus.java

@@ -23,9 +23,9 @@ import org.elasticsearch.action.support.broadcast.BroadcastShardOperationRespons
 import org.elasticsearch.cluster.routing.ShardRouting;
 import org.elasticsearch.index.shard.IndexShardState;
 import org.elasticsearch.util.SizeValue;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 import static org.elasticsearch.cluster.routing.ImmutableShardRouting.*;
@@ -60,9 +60,9 @@ public class ShardStatus extends BroadcastShardOperationResponse {
 
     IndexShardState state;
 
-    SizeValue storeSize = SizeValue.UNKNOWN;
+    SizeValue storeSize;
 
-    SizeValue estimatedFlushableMemorySize = SizeValue.UNKNOWN;
+    SizeValue estimatedFlushableMemorySize;
 
     long translogId = -1;
 
@@ -106,18 +106,28 @@ public class ShardStatus extends BroadcastShardOperationResponse {
         return docs;
     }
 
-    public static ShardStatus readIndexShardStatus(DataInput in) throws ClassNotFoundException, IOException {
+    public static ShardStatus readIndexShardStatus(StreamInput in) throws IOException {
         ShardStatus shardStatus = new ShardStatus();
         shardStatus.readFrom(in);
         return shardStatus;
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
         shardRouting.writeTo(out);
         out.writeByte(state.id());
-        storeSize.writeTo(out);
+        if (storeSize == null) {
-        estimatedFlushableMemorySize.writeTo(out);
+            out.writeBoolean(false);
+        } else {
+            out.writeBoolean(true);
+            storeSize.writeTo(out);
+        }
+        if (estimatedFlushableMemorySize == null) {
+            out.writeBoolean(false);
+        } else {
+            out.writeBoolean(true);
+            estimatedFlushableMemorySize.writeTo(out);
+        }
         out.writeLong(translogId);
         out.writeLong(translogOperations);
         out.writeInt(docs.numDocs());
@@ -125,12 +135,16 @@ public class ShardStatus extends BroadcastShardOperationResponse {
         out.writeInt(docs.deletedDocs());
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
         shardRouting = readShardRoutingEntry(in);
         state = IndexShardState.fromId(in.readByte());
-        storeSize = readSizeValue(in);
+        if (in.readBoolean()) {
-        estimatedFlushableMemorySize = readSizeValue(in);
+            storeSize = readSizeValue(in);
+        }
+        if (in.readBoolean()) {
+            estimatedFlushableMemorySize = readSizeValue(in);
+        }
         translogId = in.readLong();
         translogOperations = in.readLong();
         docs = new Docs();

modules/elasticsearch/src/main/java/org/elasticsearch/action/count/CountRequest.java

@@ -26,9 +26,9 @@ import org.elasticsearch.util.Nullable;
 import org.elasticsearch.util.Required;
 import org.elasticsearch.util.Strings;
 import org.elasticsearch.util.Unicode;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 import java.util.Arrays;
 
@@ -166,15 +166,15 @@ public class CountRequest extends BroadcastOperationRequest {
         return this;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
         minScore = in.readFloat();
-        querySource = new byte[in.readInt()];
+        querySource = new byte[in.readVInt()];
         in.readFully(querySource());
         if (in.readBoolean()) {
             queryParserName = in.readUTF();
         }
-        int typesSize = in.readInt();
+        int typesSize = in.readVInt();
         if (typesSize > 0) {
             types = new String[typesSize];
             for (int i = 0; i < typesSize; i++) {
@@ -183,18 +183,18 @@ public class CountRequest extends BroadcastOperationRequest {
         }
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
         out.writeFloat(minScore);
-        out.writeInt(querySource.length);
+        out.writeVInt(querySource.length);
-        out.write(querySource);
+        out.writeBytes(querySource);
         if (queryParserName == null) {
             out.writeBoolean(false);
         } else {
             out.writeBoolean(true);
             out.writeUTF(queryParserName);
         }
-        out.writeInt(types.length);
+        out.writeVInt(types.length);
         for (String type : types) {
             out.writeUTF(type);
         }

modules/elasticsearch/src/main/java/org/elasticsearch/action/count/CountResponse.java

@@ -21,9 +21,9 @@ package org.elasticsearch.action.count;
 
 import org.elasticsearch.action.ShardOperationFailedException;
 import org.elasticsearch.action.support.broadcast.BroadcastOperationResponse;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 import java.util.List;
 
@@ -52,13 +52,13 @@ public class CountResponse extends BroadcastOperationResponse {
         return count;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
-        count = in.readLong();
+        count = in.readVLong();
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
-        out.writeLong(count);
+        out.writeVLong(count);
     }
 }

modules/elasticsearch/src/main/java/org/elasticsearch/action/count/ShardCountRequest.java

@@ -22,9 +22,9 @@ package org.elasticsearch.action.count;
 import org.elasticsearch.action.support.broadcast.BroadcastShardOperationRequest;
 import org.elasticsearch.util.Nullable;
 import org.elasticsearch.util.Strings;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
@@ -68,15 +68,15 @@ class ShardCountRequest extends BroadcastShardOperationRequest {
         return this.types;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
         minScore = in.readFloat();
-        querySource = new byte[in.readInt()];
+        querySource = new byte[in.readVInt()];
         in.readFully(querySource);
         if (in.readBoolean()) {
             queryParserName = in.readUTF();
         }
-        int typesSize = in.readInt();
+        int typesSize = in.readVInt();
         if (typesSize > 0) {
             types = new String[typesSize];
             for (int i = 0; i < typesSize; i++) {
@@ -85,18 +85,18 @@ class ShardCountRequest extends BroadcastShardOperationRequest {
         }
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
         out.writeFloat(minScore);
-        out.writeInt(querySource.length);
+        out.writeVInt(querySource.length);
-        out.write(querySource);
+        out.writeBytes(querySource);
         if (queryParserName == null) {
             out.writeBoolean(false);
         } else {
             out.writeBoolean(true);
             out.writeUTF(queryParserName);
         }
-        out.writeInt(types.length);
+        out.writeVInt(types.length);
         for (String type : types) {
             out.writeUTF(type);
         }

modules/elasticsearch/src/main/java/org/elasticsearch/action/count/ShardCountResponse.java

@@ -20,9 +20,9 @@
 package org.elasticsearch.action.count;
 
 import org.elasticsearch.action.support.broadcast.BroadcastShardOperationResponse;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
@@ -47,13 +47,13 @@ class ShardCountResponse extends BroadcastShardOperationResponse {
         return this.count;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
-        count = in.readLong();
+        count = in.readVLong();
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
-        out.writeLong(count);
+        out.writeVLong(count);
     }
 }

modules/elasticsearch/src/main/java/org/elasticsearch/action/delete/DeleteRequest.java

@@ -23,9 +23,9 @@ import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.support.replication.ShardReplicationOperationRequest;
 import org.elasticsearch.util.Required;
 import org.elasticsearch.util.TimeValue;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 import static org.elasticsearch.action.Actions.*;
@@ -136,13 +136,13 @@ public class DeleteRequest extends ShardReplicationOperationRequest {
         return this;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
         type = in.readUTF();
         id = in.readUTF();
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
         out.writeUTF(type);
         out.writeUTF(id);

modules/elasticsearch/src/main/java/org/elasticsearch/action/delete/DeleteResponse.java

@@ -20,10 +20,10 @@
 package org.elasticsearch.action.delete;
 
 import org.elasticsearch.action.ActionResponse;
-import org.elasticsearch.util.io.Streamable;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
+import org.elasticsearch.util.io.stream.Streamable;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
@@ -72,13 +72,13 @@ public class DeleteResponse implements ActionResponse, Streamable {
         return this.id;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         index = in.readUTF();
         id = in.readUTF();
         type = in.readUTF();
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         out.writeUTF(index);
         out.writeUTF(id);
         out.writeUTF(type);

modules/elasticsearch/src/main/java/org/elasticsearch/action/deletebyquery/DeleteByQueryRequest.java

@@ -26,9 +26,9 @@ import org.elasticsearch.util.Required;
 import org.elasticsearch.util.Strings;
 import org.elasticsearch.util.TimeValue;
 import org.elasticsearch.util.Unicode;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 import java.util.Arrays;
 
@@ -149,19 +149,19 @@ public class DeleteByQueryRequest extends IndicesReplicationOperationRequest {
         return this;
     }
 
-    public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
-        querySource = new byte[in.readInt()];
+        querySource = new byte[in.readVInt()];
         in.readFully(querySource);
         if (in.readBoolean()) {
             queryParserName = in.readUTF();
         }
     }
 
-    public void writeTo(DataOutput out) throws IOException {
+    public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
-        out.writeInt(querySource.length);
+        out.writeVInt(querySource.length);
-        out.write(querySource);
+        out.writeBytes(querySource);
         if (queryParserName == null) {
             out.writeBoolean(false);
         } else {

modules/elasticsearch/src/main/java/org/elasticsearch/action/deletebyquery/DeleteByQueryResponse.java

@@ -20,10 +20,10 @@
 package org.elasticsearch.action.deletebyquery;
 
 import org.elasticsearch.action.ActionResponse;
-import org.elasticsearch.util.io.Streamable;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
+import org.elasticsearch.util.io.stream.Streamable;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 import java.util.Iterator;
 import java.util.Map;
@@ -62,8 +62,8 @@ public class DeleteByQueryResponse implements ActionResponse, Streamable, Iterab
         return indices.get(index);
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
-        int size = in.readInt();
+        int size = in.readVInt();
         for (int i = 0; i < size; i++) {
             IndexDeleteByQueryResponse response = new IndexDeleteByQueryResponse();
             response.readFrom(in);
@@ -71,8 +71,8 @@ public class DeleteByQueryResponse implements ActionResponse, Streamable, Iterab
         }
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
-        out.writeInt(indices.size());
+        out.writeVInt(indices.size());
         for (IndexDeleteByQueryResponse indexResponse : indices.values()) {
             indexResponse.writeTo(out);
         }

modules/elasticsearch/src/main/java/org/elasticsearch/action/deletebyquery/IndexDeleteByQueryRequest.java

@@ -25,9 +25,9 @@ import org.elasticsearch.index.query.QueryBuilder;
 import org.elasticsearch.util.Required;
 import org.elasticsearch.util.Strings;
 import org.elasticsearch.util.TimeValue;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 import static org.elasticsearch.action.Actions.*;
@@ -99,14 +99,14 @@ public class IndexDeleteByQueryRequest extends IndexReplicationOperationRequest
         return this;
     }
 
-    public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
-        querySource = new byte[in.readInt()];
+        querySource = new byte[in.readVInt()];
         in.readFully(querySource);
         if (in.readBoolean()) {
             queryParserName = in.readUTF();
         }
-        int typesSize = in.readInt();
+        int typesSize = in.readVInt();
         if (typesSize > 0) {
             types = new String[typesSize];
             for (int i = 0; i < typesSize; i++) {
@@ -115,17 +115,17 @@ public class IndexDeleteByQueryRequest extends IndexReplicationOperationRequest
         }
     }
 
-    public void writeTo(DataOutput out) throws IOException {
+    public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
-        out.writeInt(querySource.length);
+        out.writeVInt(querySource.length);
-        out.write(querySource);
+        out.writeBytes(querySource);
         if (queryParserName == null) {
             out.writeBoolean(false);
         } else {
             out.writeBoolean(true);
             out.writeUTF(queryParserName);
         }
-        out.writeInt(types.length);
+        out.writeVInt(types.length);
         for (String type : types) {
             out.writeUTF(type);
         }

modules/elasticsearch/src/main/java/org/elasticsearch/action/deletebyquery/IndexDeleteByQueryResponse.java

@@ -20,10 +20,10 @@
 package org.elasticsearch.action.deletebyquery;
 
 import org.elasticsearch.action.ActionResponse;
-import org.elasticsearch.util.io.Streamable;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
+import org.elasticsearch.util.io.stream.Streamable;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
@@ -77,15 +77,15 @@ public class IndexDeleteByQueryResponse implements ActionResponse, Streamable {
         return failedShards;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         index = in.readUTF();
-        successfulShards = in.readInt();
+        successfulShards = in.readVInt();
-        failedShards = in.readInt();
+        failedShards = in.readVInt();
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         out.writeUTF(index);
-        out.writeInt(successfulShards);
+        out.writeVInt(successfulShards);
-        out.writeInt(failedShards);
+        out.writeVInt(failedShards);
     }
 }

modules/elasticsearch/src/main/java/org/elasticsearch/action/deletebyquery/ShardDeleteByQueryRequest.java

@@ -23,9 +23,9 @@ import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.support.replication.ShardReplicationOperationRequest;
 import org.elasticsearch.util.Nullable;
 import org.elasticsearch.util.Strings;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 import static org.elasticsearch.action.Actions.*;
@@ -82,15 +82,15 @@ public class ShardDeleteByQueryRequest extends ShardReplicationOperationRequest
         return this.types;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
-        querySource = new byte[in.readInt()];
+        querySource = new byte[in.readVInt()];
         in.readFully(querySource);
         if (in.readBoolean()) {
             queryParserName = in.readUTF();
         }
-        shardId = in.readInt();
+        shardId = in.readVInt();
-        int typesSize = in.readInt();
+        int typesSize = in.readVInt();
         if (typesSize > 0) {
             types = new String[typesSize];
             for (int i = 0; i < typesSize; i++) {
@@ -99,18 +99,18 @@ public class ShardDeleteByQueryRequest extends ShardReplicationOperationRequest
         }
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
-        out.writeInt(querySource.length);
+        out.writeVInt(querySource.length);
-        out.write(querySource);
+        out.writeBytes(querySource);
         if (queryParserName == null) {
             out.writeBoolean(false);
         } else {
             out.writeBoolean(true);
             out.writeUTF(queryParserName);
         }
-        out.writeInt(shardId);
+        out.writeVInt(shardId);
-        out.writeInt(types.length);
+        out.writeVInt(types.length);
         for (String type : types) {
             out.writeUTF(type);
         }

modules/elasticsearch/src/main/java/org/elasticsearch/action/deletebyquery/ShardDeleteByQueryResponse.java

@@ -20,10 +20,10 @@
 package org.elasticsearch.action.deletebyquery;
 
 import org.elasticsearch.action.ActionResponse;
-import org.elasticsearch.util.io.Streamable;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
+import org.elasticsearch.util.io.stream.Streamable;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
@@ -33,9 +33,9 @@ import java.io.IOException;
  */
 public class ShardDeleteByQueryResponse implements ActionResponse, Streamable {
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
     }
 }

modules/elasticsearch/src/main/java/org/elasticsearch/action/get/GetField.java

@@ -19,10 +19,10 @@
 
 package org.elasticsearch.action.get;
 
-import org.elasticsearch.util.io.Streamable;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
+import org.elasticsearch.util.io.stream.Streamable;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Iterator;
@@ -58,15 +58,15 @@ public class GetField implements Streamable, Iterable<Object> {
         return values.iterator();
     }
 
-    public static GetField readGetField(DataInput in) throws IOException, ClassNotFoundException {
+    public static GetField readGetField(StreamInput in) throws IOException {
         GetField result = new GetField();
         result.readFrom(in);
         return result;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         name = in.readUTF();
-        int size = in.readInt();
+        int size = in.readVInt();
         values = new ArrayList<Object>(size);
         for (int i = 0; i < size; i++) {
             Object value;
@@ -84,7 +84,7 @@ public class GetField implements Streamable, Iterable<Object> {
             } else if (type == 5) {
                 value = in.readBoolean();
             } else if (type == 6) {
-                int bytesSize = in.readInt();
+                int bytesSize = in.readVInt();
                 value = new byte[bytesSize];
                 in.readFully(((byte[]) value));
             } else {
@@ -94,33 +94,33 @@ public class GetField implements Streamable, Iterable<Object> {
         }
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         out.writeUTF(name);
-        out.writeInt(values.size());
+        out.writeVInt(values.size());
         for (Object obj : values) {
             Class type = obj.getClass();
             if (type == String.class) {
-                out.write(0);
+                out.writeByte((byte) 0);
                 out.writeUTF((String) obj);
             } else if (type == Integer.class) {
-                out.write(1);
+                out.writeByte((byte) 1);
                 out.writeInt((Integer) obj);
             } else if (type == Long.class) {
-                out.write(2);
+                out.writeByte((byte) 2);
                 out.writeLong((Long) obj);
             } else if (type == Float.class) {
-                out.write(3);
+                out.writeByte((byte) 3);
                 out.writeFloat((Float) obj);
             } else if (type == Double.class) {
-                out.write(4);
+                out.writeByte((byte) 4);
                 out.writeDouble((Double) obj);
             } else if (type == Boolean.class) {
-                out.write(5);
+                out.writeByte((byte) 5);
                 out.writeBoolean((Boolean) obj);
             } else if (type == byte[].class) {
-                out.write(6);
+                out.writeByte((byte) 6);
-                out.writeInt(((byte[]) obj).length);
+                out.writeVInt(((byte[]) obj).length);
-                out.write(((byte[]) obj));
+                out.writeBytes(((byte[]) obj));
             } else {
                 throw new IOException("Can't write type [" + type + "]");
             }

modules/elasticsearch/src/main/java/org/elasticsearch/action/get/GetRequest.java

@@ -21,9 +21,9 @@ package org.elasticsearch.action.get;
 
 import org.elasticsearch.action.support.single.SingleOperationRequest;
 import org.elasticsearch.util.Required;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
@@ -113,7 +113,7 @@ public class GetRequest extends SingleOperationRequest {
         return this;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
         int size = in.readInt();
         if (size >= 0) {
@@ -124,7 +124,7 @@ public class GetRequest extends SingleOperationRequest {
         }
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
         if (fields == null) {
             out.writeInt(-1);

modules/elasticsearch/src/main/java/org/elasticsearch/action/get/GetResponse.java

@@ -22,10 +22,10 @@ package org.elasticsearch.action.get;
 import org.elasticsearch.ElasticSearchParseException;
 import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.util.Unicode;
-import org.elasticsearch.util.io.Streamable;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
+import org.elasticsearch.util.io.stream.Streamable;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 import java.util.Iterator;
 import java.util.Map;
@@ -143,18 +143,18 @@ public class GetResponse implements ActionResponse, Streamable, Iterable<GetFiel
         return fields.values().iterator();
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         index = in.readUTF();
         type = in.readUTF();
         id = in.readUTF();
         exists = in.readBoolean();
         if (exists) {
-            int size = in.readInt();
+            int size = in.readVInt();
             if (size > 0) {
                 source = new byte[size];
                 in.readFully(source);
             }
-            size = in.readInt();
+            size = in.readVInt();
             if (size > 0) {
                 fields = newHashMapWithExpectedSize(size);
                 for (int i = 0; i < size; i++) {
@@ -165,22 +165,22 @@ public class GetResponse implements ActionResponse, Streamable, Iterable<GetFiel
         }
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         out.writeUTF(index);
         out.writeUTF(type);
         out.writeUTF(id);
         out.writeBoolean(exists);
         if (exists) {
             if (source == null) {
-                out.writeInt(0);
+                out.writeVInt(0);
             } else {
-                out.writeInt(source.length);
+                out.writeVInt(source.length);
-                out.write(source);
+                out.writeBytes(source);
             }
             if (fields == null) {
-                out.writeInt(0);
+                out.writeVInt(0);
             } else {
-                out.writeInt(fields.size());
+                out.writeVInt(fields.size());
                 for (GetField field : fields.values()) {
                     field.writeTo(out);
                 }

modules/elasticsearch/src/main/java/org/elasticsearch/action/index/IndexRequest.java

@@ -27,10 +27,10 @@ import org.elasticsearch.util.Required;
 import org.elasticsearch.util.TimeValue;
 import org.elasticsearch.util.Unicode;
 import org.elasticsearch.util.io.FastByteArrayOutputStream;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 import org.elasticsearch.util.json.JsonBuilder;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 import java.util.Map;
 
@@ -263,18 +263,18 @@ public class IndexRequest extends ShardReplicationOperationRequest {
         return this.opType;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
         type = in.readUTF();
         if (in.readBoolean()) {
             id = in.readUTF();
         }
-        source = new byte[in.readInt()];
+        source = new byte[in.readVInt()];
-        in.readFully(source, 0, source.length);
+        in.readFully(source);
         opType = OpType.fromId(in.readByte());
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
         out.writeUTF(type);
         if (id == null) {
@@ -283,8 +283,8 @@ public class IndexRequest extends ShardReplicationOperationRequest {
             out.writeBoolean(true);
             out.writeUTF(id);
         }
-        out.writeInt(source.length);
+        out.writeVInt(source.length);
-        out.write(source);
+        out.writeBytes(source);
         out.writeByte(opType.id());
     }
 

modules/elasticsearch/src/main/java/org/elasticsearch/action/index/IndexResponse.java

@@ -20,10 +20,10 @@
 package org.elasticsearch.action.index;
 
 import org.elasticsearch.action.ActionResponse;
-import org.elasticsearch.util.io.Streamable;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
+import org.elasticsearch.util.io.stream.Streamable;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
@@ -72,13 +72,13 @@ public class IndexResponse implements ActionResponse, Streamable {
         return this.id;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         index = in.readUTF();
         id = in.readUTF();
         type = in.readUTF();
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         out.writeUTF(index);
         out.writeUTF(id);
         out.writeUTF(type);

modules/elasticsearch/src/main/java/org/elasticsearch/action/mlt/MoreLikeThisRequest.java

@@ -29,9 +29,9 @@ import org.elasticsearch.util.Bytes;
 import org.elasticsearch.util.Required;
 import org.elasticsearch.util.Strings;
 import org.elasticsearch.util.Unicode;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 import static org.elasticsearch.search.Scroll.*;
@@ -450,35 +450,35 @@ public class MoreLikeThisRequest implements ActionRequest {
         return this;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         index = in.readUTF();
         type = in.readUTF();
         id = in.readUTF();
         // no need to pass threading over the network, they are always false when coming throw a thread pool
-        int size = in.readInt();
+        int size = in.readVInt();
         if (size == 0) {
             fields = Strings.EMPTY_ARRAY;
         } else {
-            fields = new String[in.readInt()];
+            fields = new String[size];
             for (int i = 0; i < size; i++) {
                 fields[i] = in.readUTF();
             }
         }
 
         percentTermsToMatch = in.readFloat();
-        minTermFrequency = in.readInt();
+        minTermFrequency = in.readVInt();
-        maxQueryTerms = in.readInt();
+        maxQueryTerms = in.readVInt();
-        size = in.readInt();
+        size = in.readVInt();
         if (size > 0) {
             stopWords = new String[size];
             for (int i = 0; i < size; i++) {
                 stopWords[i] = in.readUTF();
             }
         }
-        minDocFreq = in.readInt();
+        minDocFreq = in.readVInt();
-        maxDocFreq = in.readInt();
+        maxDocFreq = in.readVInt();
-        minWordLen = in.readInt();
+        minWordLen = in.readVInt();
-        maxWordLen = in.readInt();
+        maxWordLen = in.readVInt();
         if (in.readBoolean()) {
             boostTerms = in.readBoolean();
         }
@@ -487,68 +487,68 @@ public class MoreLikeThisRequest implements ActionRequest {
         if (in.readBoolean()) {
             searchQueryHint = in.readUTF();
         }
-        size = in.readInt();
+        size = in.readVInt();
-        if (size == -1) {
+        if (size == 0) {
             searchIndices = null;
-        } else if (size == 0) {
+        } else if (size == 1) {
             searchIndices = Strings.EMPTY_ARRAY;
         } else {
-            searchIndices = new String[size];
+            searchIndices = new String[size - 1];
-            for (int i = 0; i < size; i++) {
+            for (int i = 0; i < searchIndices.length; i++) {
                 searchIndices[i] = in.readUTF();
             }
         }
-        size = in.readInt();
+        size = in.readVInt();
-        if (size == -1) {
+        if (size == 0) {
             searchTypes = null;
-        } else if (size == 0) {
+        } else if (size == 1) {
             searchTypes = Strings.EMPTY_ARRAY;
         } else {
-            searchTypes = new String[size];
+            searchTypes = new String[size - 1];
-            for (int i = 0; i < size; i++) {
+            for (int i = 0; i < searchTypes.length; i++) {
                 searchTypes[i] = in.readUTF();
             }
         }
         if (in.readBoolean()) {
             searchScroll = readScroll(in);
         }
-        size = in.readInt();
+        size = in.readVInt();
         if (size == 0) {
             searchSource = Bytes.EMPTY_ARRAY;
         } else {
-            searchSource = new byte[in.readInt()];
+            searchSource = new byte[in.readVInt()];
             in.readFully(searchSource);
         }
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         out.writeUTF(index);
         out.writeUTF(type);
         out.writeUTF(id);
         if (fields == null) {
-            out.writeInt(0);
+            out.writeVInt(0);
         } else {
-            out.writeInt(fields.length);
+            out.writeVInt(fields.length);
             for (String field : fields) {
                 out.writeUTF(field);
             }
         }
 
         out.writeFloat(percentTermsToMatch);
-        out.writeInt(minTermFrequency);
+        out.writeVInt(minTermFrequency);
-        out.writeInt(maxQueryTerms);
+        out.writeVInt(maxQueryTerms);
         if (stopWords == null) {
-            out.writeInt(0);
+            out.writeVInt(0);
         } else {
-            out.writeInt(stopWords.length);
+            out.writeVInt(stopWords.length);
             for (String stopWord : stopWords) {
                 out.writeUTF(stopWord);
             }
         }
-        out.writeInt(minDocFreq);
+        out.writeVInt(minDocFreq);
-        out.writeInt(maxDocFreq);
+        out.writeVInt(maxDocFreq);
-        out.writeInt(minWordLen);
+        out.writeVInt(minWordLen);
-        out.writeInt(maxWordLen);
+        out.writeVInt(maxWordLen);
         if (boostTerms == null) {
             out.writeBoolean(false);
         } else {
@@ -565,17 +565,17 @@ public class MoreLikeThisRequest implements ActionRequest {
             out.writeUTF(searchQueryHint);
         }
         if (searchIndices == null) {
-            out.writeInt(-1);
+            out.writeVInt(0);
         } else {
-            out.writeInt(searchIndices.length);
+            out.writeVInt(searchIndices.length + 1);
             for (String index : searchIndices) {
                 out.writeUTF(index);
             }
         }
         if (searchTypes == null) {
-            out.writeInt(-1);
+            out.writeVInt(0);
         } else {
-            out.writeInt(searchTypes.length);
+            out.writeVInt(searchTypes.length + 1);
             for (String type : searchTypes) {
                 out.writeUTF(type);
             }
@@ -587,10 +587,10 @@ public class MoreLikeThisRequest implements ActionRequest {
             searchScroll.writeTo(out);
         }
         if (searchSource == null) {
-            out.writeInt(0);
+            out.writeVInt(0);
         } else {
-            out.writeInt(searchSource.length);
+            out.writeVInt(searchSource.length);
-            out.write(searchSource);
+            out.writeBytes(searchSource);
         }
     }
 }

modules/elasticsearch/src/main/java/org/elasticsearch/action/search/SearchRequest.java

@@ -27,9 +27,9 @@ import org.elasticsearch.util.Bytes;
 import org.elasticsearch.util.Strings;
 import org.elasticsearch.util.TimeValue;
 import org.elasticsearch.util.Unicode;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 import static org.elasticsearch.action.Actions.*;
@@ -278,11 +278,11 @@ public class SearchRequest implements ActionRequest {
         return this;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         operationThreading = SearchOperationThreading.fromId(in.readByte());
         searchType = SearchType.fromId(in.readByte());
 
-        indices = new String[in.readInt()];
+        indices = new String[in.readVInt()];
         for (int i = 0; i < indices.length; i++) {
             indices[i] = in.readUTF();
         }
@@ -297,14 +297,14 @@ public class SearchRequest implements ActionRequest {
         if (in.readBoolean()) {
             timeout = readTimeValue(in);
         }
-        int size = in.readInt();
+        int size = in.readVInt();
         if (size == 0) {
             source = Bytes.EMPTY_ARRAY;
         } else {
             source = new byte[size];
             in.readFully(source);
         }
-        size = in.readInt();
+        size = in.readVInt();
         if (size == 0) {
             extraSource = Bytes.EMPTY_ARRAY;
         } else {
@@ -312,7 +312,7 @@ public class SearchRequest implements ActionRequest {
             in.readFully(extraSource);
         }
 
-        int typesSize = in.readInt();
+        int typesSize = in.readVInt();
         if (typesSize > 0) {
             types = new String[typesSize];
             for (int i = 0; i < typesSize; i++) {
@@ -321,11 +321,11 @@ public class SearchRequest implements ActionRequest {
         }
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         out.writeByte(operationThreading.id());
         out.writeByte(searchType.id());
 
-        out.writeInt(indices.length);
+        out.writeVInt(indices.length);
         for (String index : indices) {
             out.writeUTF(index);
         }
@@ -350,18 +350,18 @@ public class SearchRequest implements ActionRequest {
             timeout.writeTo(out);
         }
         if (source == null) {
-            out.writeInt(0);
+            out.writeVInt(0);
         } else {
-            out.writeInt(source.length);
+            out.writeVInt(source.length);
-            out.write(source);
+            out.writeBytes(source);
         }
         if (extraSource == null) {
-            out.writeInt(0);
+            out.writeVInt(0);
         } else {
-            out.writeInt(extraSource.length);
+            out.writeVInt(extraSource.length);
-            out.write(extraSource);
+            out.writeBytes(extraSource);
         }
-        out.writeInt(types.length);
+        out.writeVInt(types.length);
         for (String type : types) {
             out.writeUTF(type);
         }

modules/elasticsearch/src/main/java/org/elasticsearch/action/search/SearchResponse.java

@@ -23,11 +23,11 @@ import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.search.SearchHits;
 import org.elasticsearch.search.facets.Facets;
 import org.elasticsearch.search.internal.InternalSearchResponse;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 import org.elasticsearch.util.json.JsonBuilder;
 import org.elasticsearch.util.json.ToJson;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 import static org.elasticsearch.action.search.ShardSearchFailure.*;
@@ -111,12 +111,6 @@ public class SearchResponse implements ActionResponse, ToJson {
         return scrollId;
     }
 
-    public static SearchResponse readSearchResponse(DataInput in) throws IOException, ClassNotFoundException {
-        SearchResponse response = new SearchResponse();
-        response.readFrom(in);
-        return response;
-    }
-
     @Override public void toJson(JsonBuilder builder, Params params) throws IOException {
         if (scrollId != null) {
             builder.field("_scrollId", scrollId);
@@ -144,11 +138,17 @@ public class SearchResponse implements ActionResponse, ToJson {
         internalResponse.toJson(builder, params);
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    public static SearchResponse readSearchResponse(StreamInput in) throws IOException {
+        SearchResponse response = new SearchResponse();
+        response.readFrom(in);
+        return response;
+    }
+
+    @Override public void readFrom(StreamInput in) throws IOException {
         internalResponse = readInternalSearchResponse(in);
-        totalShards = in.readInt();
+        totalShards = in.readVInt();
-        successfulShards = in.readInt();
+        successfulShards = in.readVInt();
-        int size = in.readInt();
+        int size = in.readVInt();
         if (size == 0) {
             shardFailures = ShardSearchFailure.EMPTY_ARRAY;
         } else {
@@ -162,12 +162,12 @@ public class SearchResponse implements ActionResponse, ToJson {
         }
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         internalResponse.writeTo(out);
-        out.writeInt(totalShards);
+        out.writeVInt(totalShards);
-        out.writeInt(successfulShards);
+        out.writeVInt(successfulShards);
 
-        out.writeInt(shardFailures.length);
+        out.writeVInt(shardFailures.length);
         for (ShardSearchFailure shardSearchFailure : shardFailures) {
             shardSearchFailure.writeTo(out);
         }

modules/elasticsearch/src/main/java/org/elasticsearch/action/search/SearchScrollRequest.java

@@ -22,9 +22,9 @@ package org.elasticsearch.action.search;
 import org.elasticsearch.action.ActionRequest;
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.search.Scroll;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 import static org.elasticsearch.action.Actions.*;
@@ -76,14 +76,14 @@ public class SearchScrollRequest implements ActionRequest {
         this.scroll = scroll;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         scrollId = in.readUTF();
         if (in.readBoolean()) {
             scroll = readScroll(in);
         }
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         out.writeUTF(scrollId);
         if (scroll == null) {
             out.writeBoolean(false);

modules/elasticsearch/src/main/java/org/elasticsearch/action/search/ShardSearchFailure.java

@@ -24,9 +24,9 @@ import org.elasticsearch.action.ShardOperationFailedException;
 import org.elasticsearch.search.SearchException;
 import org.elasticsearch.search.SearchShardTarget;
 import org.elasticsearch.util.Nullable;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 import static org.elasticsearch.search.SearchShardTarget.*;
@@ -99,20 +99,20 @@ public class ShardSearchFailure implements ShardOperationFailedException {
         return "Search Failure Shard " + shardTarget + ", reason [" + reason + "]";
     }
 
-    public static ShardSearchFailure readShardSearchFailure(DataInput in) throws IOException, ClassNotFoundException {
+    public static ShardSearchFailure readShardSearchFailure(StreamInput in) throws IOException {
         ShardSearchFailure shardSearchFailure = new ShardSearchFailure();
         shardSearchFailure.readFrom(in);
         return shardSearchFailure;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         if (in.readBoolean()) {
             shardTarget = readSearchShardTarget(in);
         }
         reason = in.readUTF();
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         if (shardTarget == null) {
             out.writeBoolean(false);
         } else {

modules/elasticsearch/src/main/java/org/elasticsearch/action/support/DefaultShardOperationFailedException.java

@@ -21,9 +21,9 @@ package org.elasticsearch.action.support;
 
 import org.elasticsearch.action.ShardOperationFailedException;
 import org.elasticsearch.index.shard.IndexShardException;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 import static org.elasticsearch.ExceptionsHelper.*;
@@ -67,28 +67,28 @@ public class DefaultShardOperationFailedException implements ShardOperationFaile
         return this.reason;
     }
 
-    public static DefaultShardOperationFailedException readShardOperationFailed(DataInput in) throws IOException, ClassNotFoundException {
+    public static DefaultShardOperationFailedException readShardOperationFailed(StreamInput in) throws IOException {
         DefaultShardOperationFailedException exp = new DefaultShardOperationFailedException();
         exp.readFrom(in);
         return exp;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         if (in.readBoolean()) {
             index = in.readUTF();
         }
-        shardId = in.readInt();
+        shardId = in.readVInt();
         reason = in.readUTF();
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         if (index == null) {
             out.writeBoolean(false);
         } else {
             out.writeBoolean(true);
             out.writeUTF(index);
         }
-        out.writeInt(shardId);
+        out.writeVInt(shardId);
         out.writeUTF(reason);
     }
 }

modules/elasticsearch/src/main/java/org/elasticsearch/action/support/broadcast/BroadcastOperationRequest.java

@@ -23,13 +23,13 @@ import org.elasticsearch.action.ActionRequest;
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.util.Nullable;
 import org.elasticsearch.util.Strings;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
- * @author kimchy (Shay Banon)
+ * @author kimchy (shay.banon)
  */
 public abstract class BroadcastOperationRequest implements ActionRequest {
 
@@ -90,11 +90,11 @@ public abstract class BroadcastOperationRequest implements ActionRequest {
         return this;
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         if (indices == null) {
-            out.writeInt(0);
+            out.writeVInt(0);
         } else {
-            out.writeInt(indices.length);
+            out.writeVInt(indices.length);
             for (String index : indices) {
                 out.writeUTF(index);
             }
@@ -108,8 +108,8 @@ public abstract class BroadcastOperationRequest implements ActionRequest {
         out.writeByte(operationThreading.id());
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
-        int size = in.readInt();
+        int size = in.readVInt();
         if (size == 0) {
             indices = Strings.EMPTY_ARRAY;
         } else {

modules/elasticsearch/src/main/java/org/elasticsearch/action/support/broadcast/BroadcastOperationResponse.java

@@ -22,14 +22,15 @@ package org.elasticsearch.action.support.broadcast;
 import com.google.common.collect.ImmutableList;
 import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.action.ShardOperationFailedException;
-import org.elasticsearch.action.support.DefaultShardOperationFailedException;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
 
+import static org.elasticsearch.action.support.DefaultShardOperationFailedException.*;
+
 /**
  * Base class for all broadcast operation based responses.
  *
@@ -86,22 +87,22 @@ public abstract class BroadcastOperationResponse implements ActionResponse {
         return shardFailures;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
-        successfulShards = in.readInt();
+        successfulShards = in.readVInt();
-        failedShards = in.readInt();
+        failedShards = in.readVInt();
-        int size = in.readInt();
+        int size = in.readVInt();
         if (size > 0) {
             shardFailures = new ArrayList<ShardOperationFailedException>(size);
             for (int i = 0; i < size; i++) {
-                shardFailures.add(DefaultShardOperationFailedException.readShardOperationFailed(in));
+                shardFailures.add(readShardOperationFailed(in));
             }
         }
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
-        out.writeInt(successfulShards);
+        out.writeVInt(successfulShards);
-        out.writeInt(failedShards);
+        out.writeVInt(failedShards);
-        out.writeInt(shardFailures.size());
+        out.writeVInt(shardFailures.size());
         for (ShardOperationFailedException exp : shardFailures) {
             exp.writeTo(out);
         }

modules/elasticsearch/src/main/java/org/elasticsearch/action/support/broadcast/BroadcastShardOperationRequest.java

@@ -19,14 +19,14 @@
 
 package org.elasticsearch.action.support.broadcast;
 
-import org.elasticsearch.util.io.Streamable;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
+import org.elasticsearch.util.io.stream.Streamable;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
- * @author kimchy (Shay Banon)
+ * @author kimchy (shay.banon)
  */
 public abstract class BroadcastShardOperationRequest implements Streamable {
 
@@ -50,13 +50,13 @@ public abstract class BroadcastShardOperationRequest implements Streamable {
         return this.shardId;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         index = in.readUTF();
-        shardId = in.readInt();
+        shardId = in.readVInt();
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         out.writeUTF(index);
-        out.writeInt(shardId);
+        out.writeVInt(shardId);
     }
 }

modules/elasticsearch/src/main/java/org/elasticsearch/action/support/broadcast/BroadcastShardOperationResponse.java

@@ -19,10 +19,10 @@
 
 package org.elasticsearch.action.support.broadcast;
 
-import org.elasticsearch.util.io.Streamable;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
+import org.elasticsearch.util.io.stream.Streamable;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
@@ -51,13 +51,13 @@ public abstract class BroadcastShardOperationResponse implements Streamable {
         return this.shardId;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         index = in.readUTF();
-        shardId = in.readInt();
+        shardId = in.readVInt();
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         out.writeUTF(index);
-        out.writeInt(shardId);
+        out.writeVInt(shardId);
     }
 }

modules/elasticsearch/src/main/java/org/elasticsearch/action/support/master/MasterNodeOperationRequest.java

@@ -20,9 +20,9 @@
 package org.elasticsearch.action.support.master;
 
 import org.elasticsearch.action.ActionRequest;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
@@ -42,9 +42,9 @@ public abstract class MasterNodeOperationRequest implements ActionRequest {
         return this;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
     }
 }

modules/elasticsearch/src/main/java/org/elasticsearch/action/support/nodes/NodeOperationRequest.java

@@ -19,10 +19,10 @@
 
 package org.elasticsearch.action.support.nodes;
 
-import org.elasticsearch.util.io.Streamable;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
+import org.elasticsearch.util.io.stream.Streamable;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
@@ -40,11 +40,11 @@ public abstract class NodeOperationRequest implements Streamable {
         this.nodeId = nodeId;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         nodeId = in.readUTF();
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         out.writeUTF(nodeId);
     }
 }

modules/elasticsearch/src/main/java/org/elasticsearch/action/support/nodes/NodeOperationResponse.java

@@ -20,14 +20,14 @@
 package org.elasticsearch.action.support.nodes;
 
 import org.elasticsearch.cluster.node.Node;
-import org.elasticsearch.util.io.Streamable;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
+import org.elasticsearch.util.io.stream.Streamable;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
- * @author kimchy (Shay Banon)
+ * @author kimchy (shay.banon)
  */
 public abstract class NodeOperationResponse implements Streamable {
 
@@ -44,11 +44,11 @@ public abstract class NodeOperationResponse implements Streamable {
         return node;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         node = Node.readNode(in);
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         node.writeTo(out);
     }
 }

modules/elasticsearch/src/main/java/org/elasticsearch/action/support/nodes/NodesOperationRequest.java

@@ -22,9 +22,9 @@ package org.elasticsearch.action.support.nodes;
 import org.elasticsearch.action.ActionRequest;
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.util.Strings;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
@@ -63,18 +63,18 @@ public abstract class NodesOperationRequest implements ActionRequest {
         return null;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
-        nodesIds = new String[in.readInt()];
+        nodesIds = new String[in.readVInt()];
         for (int i = 0; i < nodesIds.length; i++) {
             nodesIds[i] = in.readUTF();
         }
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         if (nodesIds == null) {
-            out.writeInt(0);
+            out.writeVInt(0);
         } else {
-            out.writeInt(nodesIds.length);
+            out.writeVInt(nodesIds.length);
             for (String nodeId : nodesIds) {
                 out.writeUTF(nodeId);
             }

modules/elasticsearch/src/main/java/org/elasticsearch/action/support/nodes/NodesOperationResponse.java

@@ -22,9 +22,9 @@ package org.elasticsearch.action.support.nodes;
 import com.google.common.collect.Maps;
 import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.cluster.ClusterName;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 import java.util.Iterator;
 import java.util.Map;
@@ -70,11 +70,11 @@ public abstract class NodesOperationResponse<NodeResponse extends NodeOperationR
         return nodesMap;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         clusterName = ClusterName.readClusterName(in);
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         clusterName.writeTo(out);
     }
 }

modules/elasticsearch/src/main/java/org/elasticsearch/action/support/replication/IndexReplicationOperationRequest.java

@@ -22,9 +22,9 @@ package org.elasticsearch.action.support.replication;
 import org.elasticsearch.action.ActionRequest;
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.util.TimeValue;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 import static org.elasticsearch.action.Actions.*;
@@ -65,12 +65,12 @@ public class IndexReplicationOperationRequest implements ActionRequest {
         return validationException;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         timeout = TimeValue.readTimeValue(in);
         index = in.readUTF();
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         timeout.writeTo(out);
         out.writeUTF(index);
     }

modules/elasticsearch/src/main/java/org/elasticsearch/action/support/replication/IndicesReplicationOperationRequest.java

@@ -22,9 +22,9 @@ package org.elasticsearch.action.support.replication;
 import org.elasticsearch.action.ActionRequest;
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.util.TimeValue;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
@@ -65,17 +65,17 @@ public class IndicesReplicationOperationRequest implements ActionRequest {
         return this;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         timeout = TimeValue.readTimeValue(in);
-        indices = new String[in.readInt()];
+        indices = new String[in.readVInt()];
         for (int i = 0; i < indices.length; i++) {
             indices[i] = in.readUTF();
         }
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         timeout.writeTo(out);
-        out.writeInt(indices.length);
+        out.writeVInt(indices.length);
         for (String index : indices) {
             out.writeUTF(index);
         }

modules/elasticsearch/src/main/java/org/elasticsearch/action/support/replication/ShardReplicationOperationRequest.java

@@ -22,9 +22,9 @@ package org.elasticsearch.action.support.replication;
 import org.elasticsearch.action.ActionRequest;
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.util.TimeValue;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 import java.util.concurrent.TimeUnit;
 
@@ -91,13 +91,13 @@ public abstract class ShardReplicationOperationRequest implements ActionRequest
         return validationException;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         timeout = TimeValue.readTimeValue(in);
         index = in.readUTF();
         // no need to serialize threaded* parameters, since they only matter locally
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         timeout.writeTo(out);
         out.writeUTF(index);
     }

modules/elasticsearch/src/main/java/org/elasticsearch/action/support/replication/TransportShardReplicationOperationAction.java

@@ -42,12 +42,12 @@ import org.elasticsearch.indices.IndicesService;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.*;
 import org.elasticsearch.util.TimeValue;
-import org.elasticsearch.util.io.Streamable;
 import org.elasticsearch.util.io.VoidStreamable;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
+import org.elasticsearch.util.io.stream.Streamable;
 import org.elasticsearch.util.settings.Settings;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -176,14 +176,14 @@ public abstract class TransportShardReplicationOperationAction<Request extends S
             this.request = request;
         }
 
-        @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+        @Override public void readFrom(StreamInput in) throws IOException {
-            shardId = in.readInt();
+            shardId = in.readVInt();
             request = newRequestInstance();
             request.readFrom(in);
         }
 
-        @Override public void writeTo(DataOutput out) throws IOException {
+        @Override public void writeTo(StreamOutput out) throws IOException {
-            out.writeInt(shardId);
+            out.writeVInt(shardId);
             request.writeTo(out);
         }
     }

modules/elasticsearch/src/main/java/org/elasticsearch/action/support/single/SingleOperationRequest.java

@@ -22,13 +22,13 @@ package org.elasticsearch.action.support.single;
 import org.elasticsearch.action.ActionRequest;
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.Actions;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
- * @author kimchy (Shay Banon)
+ * @author kimchy (shay.banon)
  */
 public abstract class SingleOperationRequest implements ActionRequest {
 
@@ -101,14 +101,14 @@ public abstract class SingleOperationRequest implements ActionRequest {
         return this;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         index = in.readUTF();
         type = in.readUTF();
         id = in.readUTF();
         // no need to pass threading over the network, they are always false when coming throw a thread pool
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         out.writeUTF(index);
         out.writeUTF(type);
         out.writeUTF(id);

modules/elasticsearch/src/main/java/org/elasticsearch/action/support/single/TransportSingleOperationAction.java

@@ -33,11 +33,11 @@ import org.elasticsearch.cluster.routing.ShardsIterator;
 import org.elasticsearch.indices.IndicesService;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.*;
-import org.elasticsearch.util.io.Streamable;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
+import org.elasticsearch.util.io.stream.Streamable;
 import org.elasticsearch.util.settings.Settings;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 import java.util.Iterator;
 
@@ -285,15 +285,15 @@ public abstract class TransportSingleOperationAction<Request extends SingleOpera
             return shardId;
         }
 
-        @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+        @Override public void readFrom(StreamInput in) throws IOException {
             request = newRequest();
             request.readFrom(in);
-            shardId = in.readInt();
+            shardId = in.readVInt();
         }
 
-        @Override public void writeTo(DataOutput out) throws IOException {
+        @Override public void writeTo(StreamOutput out) throws IOException {
             request.writeTo(out);
-            out.writeInt(shardId);
+            out.writeVInt(shardId);
         }
     }
 }

modules/elasticsearch/src/main/java/org/elasticsearch/action/terms/FieldTermsFreq.java

@@ -20,11 +20,11 @@
 package org.elasticsearch.action.terms;
 
 import com.google.common.collect.Iterators;
-import org.elasticsearch.util.io.Streamable;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
+import org.elasticsearch.util.io.stream.Streamable;
 import org.elasticsearch.util.trove.ExtTObjectIntHasMap;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 import java.util.Iterator;
 
@@ -84,23 +84,23 @@ public class FieldTermsFreq implements Streamable, Iterable<TermFreq> {
         return Iterators.forArray(termsFreqs);
     }
 
-    public static FieldTermsFreq readFieldTermsFreq(DataInput in) throws IOException, ClassNotFoundException {
+    public static FieldTermsFreq readFieldTermsFreq(StreamInput in) throws IOException {
         FieldTermsFreq fieldTermsFreq = new FieldTermsFreq();
         fieldTermsFreq.readFrom(in);
         return fieldTermsFreq;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         fieldName = in.readUTF();
-        termsFreqs = new TermFreq[in.readInt()];
+        termsFreqs = new TermFreq[in.readVInt()];
         for (int i = 0; i < termsFreqs.length; i++) {
             termsFreqs[i] = readTermFreq(in);
         }
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         out.writeUTF(fieldName);
-        out.writeInt(termsFreqs.length);
+        out.writeVInt(termsFreqs.length);
         for (TermFreq termFreq : termsFreqs) {
             termFreq.writeTo(out);
         }

modules/elasticsearch/src/main/java/org/elasticsearch/action/terms/ShardTermsRequest.java

@@ -20,9 +20,9 @@
 package org.elasticsearch.action.terms;
 
 import org.elasticsearch.action.support.broadcast.BroadcastShardOperationRequest;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
@@ -114,9 +114,9 @@ class ShardTermsRequest extends BroadcastShardOperationRequest {
         return this.exact;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
-        fields = new String[in.readInt()];
+        fields = new String[in.readVInt()];
         for (int i = 0; i < fields.length; i++) {
             fields[i] = in.readUTF();
         }
@@ -134,15 +134,15 @@ class ShardTermsRequest extends BroadcastShardOperationRequest {
         if (in.readBoolean()) {
             regexp = in.readUTF();
         }
-        size = in.readInt();
+        size = in.readVInt();
         convert = in.readBoolean();
         sortType = TermsRequest.SortType.fromValue(in.readByte());
         exact = in.readBoolean();
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
-        out.writeInt(fields.length);
+        out.writeVInt(fields.length);
         for (String field : fields) {
             out.writeUTF(field);
         }
@@ -172,7 +172,7 @@ class ShardTermsRequest extends BroadcastShardOperationRequest {
             out.writeBoolean(true);
             out.writeUTF(regexp);
         }
-        out.writeInt(size);
+        out.writeVInt(size);
         out.writeBoolean(convert);
         out.writeByte(sortType.value());
         out.writeBoolean(exact);

modules/elasticsearch/src/main/java/org/elasticsearch/action/terms/ShardTermsResponse.java

@@ -22,9 +22,9 @@ package org.elasticsearch.action.terms;
 import org.elasticsearch.action.support.broadcast.BroadcastShardOperationResponse;
 import org.elasticsearch.util.gnu.trove.TObjectIntHashMap;
 import org.elasticsearch.util.gnu.trove.TObjectIntIterator;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 import java.util.HashMap;
 import java.util.Map;
@@ -72,38 +72,38 @@ class ShardTermsResponse extends BroadcastShardOperationResponse {
         return fieldsTermsFreqs;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
-        numDocs = in.readInt();
+        numDocs = in.readVInt();
-        maxDoc = in.readInt();
+        maxDoc = in.readVInt();
-        numDeletedDocs = in.readInt();
+        numDeletedDocs = in.readVInt();
-        int size = in.readInt();
+        int size = in.readVInt();
         for (int i = 0; i < size; i++) {
             String fieldName = in.readUTF();
 
             TObjectIntHashMap<String> termsFreq = new TObjectIntHashMap<String>();
-            int size1 = in.readInt();
+            int size1 = in.readVInt();
             for (int j = 0; j < size1; j++) {
-                termsFreq.put(in.readUTF(), in.readInt());
+                termsFreq.put(in.readUTF(), in.readVInt());
             }
 
             fieldsTermsFreqs.put(fieldName, termsFreq);
         }
     }
 
-    @Override public void writeTo(final DataOutput out) throws IOException {
+    @Override public void writeTo(final StreamOutput out) throws IOException {
         super.writeTo(out);
-        out.writeInt(numDocs);
+        out.writeVInt(numDocs);
-        out.writeInt(maxDoc);
+        out.writeVInt(maxDoc);
-        out.writeInt(numDeletedDocs);
+        out.writeVInt(numDeletedDocs);
-        out.writeInt(fieldsTermsFreqs.size());
+        out.writeVInt(fieldsTermsFreqs.size());
         for (Map.Entry<String, TObjectIntHashMap<String>> entry : fieldsTermsFreqs.entrySet()) {
             out.writeUTF(entry.getKey());
-            out.writeInt(entry.getValue().size());
+            out.writeVInt(entry.getValue().size());
             for (TObjectIntIterator<String> it = entry.getValue().iterator(); it.hasNext();) {
                 it.advance();
                 out.writeUTF(it.key());
-                out.writeInt(it.value());
+                out.writeVInt(it.value());
             }
         }
     }

modules/elasticsearch/src/main/java/org/elasticsearch/action/terms/TermFreq.java

@@ -19,10 +19,10 @@
 
 package org.elasticsearch.action.terms;
 
-import org.elasticsearch.util.io.Streamable;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
+import org.elasticsearch.util.io.stream.Streamable;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 import java.util.Comparator;
 
@@ -106,19 +106,19 @@ public class TermFreq implements Streamable {
         return docFreq;
     }
 
-    public static TermFreq readTermFreq(DataInput in) throws IOException, ClassNotFoundException {
+    public static TermFreq readTermFreq(StreamInput in) throws IOException {
         TermFreq termFreq = new TermFreq();
         termFreq.readFrom(in);
         return termFreq;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         term = in.readUTF();
-        docFreq = in.readInt();
+        docFreq = in.readVInt();
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         out.writeUTF(term);
-        out.writeInt(docFreq);
+        out.writeVInt(docFreq);
     }
 }

modules/elasticsearch/src/main/java/org/elasticsearch/action/terms/TermsRequest.java

@@ -22,9 +22,9 @@ package org.elasticsearch.action.terms;
 import org.elasticsearch.ElasticSearchIllegalArgumentException;
 import org.elasticsearch.action.support.broadcast.BroadcastOperationRequest;
 import org.elasticsearch.index.mapper.AllFieldMapper;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
@@ -353,9 +353,9 @@ public class TermsRequest extends BroadcastOperationRequest {
         return this;
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
-        out.writeInt(fields.length);
+        out.writeVInt(fields.length);
         for (String field : fields) {
             out.writeUTF(field);
         }
@@ -385,17 +385,17 @@ public class TermsRequest extends BroadcastOperationRequest {
             out.writeBoolean(true);
             out.writeUTF(regexp);
         }
-        out.writeInt(size);
+        out.writeVInt(size);
         out.writeBoolean(convert);
         out.writeByte(sortType.value());
-        out.writeInt(minFreq);
+        out.writeVInt(minFreq);
-        out.writeInt(maxFreq);
+        out.writeVInt(maxFreq);
         out.writeBoolean(exact);
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
-        fields = new String[in.readInt()];
+        fields = new String[in.readVInt()];
         for (int i = 0; i < fields.length; i++) {
             fields[i] = in.readUTF();
         }
@@ -413,11 +413,11 @@ public class TermsRequest extends BroadcastOperationRequest {
         if (in.readBoolean()) {
             regexp = in.readUTF();
         }
-        size = in.readInt();
+        size = in.readVInt();
         convert = in.readBoolean();
         sortType = TermsRequest.SortType.fromValue(in.readByte());
-        minFreq = in.readInt();
+        minFreq = in.readVInt();
-        maxFreq = in.readInt();
+        maxFreq = in.readVInt();
         exact = in.readBoolean();
     }
 }

modules/elasticsearch/src/main/java/org/elasticsearch/action/terms/TermsResponse.java

@@ -22,9 +22,9 @@ package org.elasticsearch.action.terms;
 import com.google.common.collect.Iterators;
 import org.elasticsearch.action.ShardOperationFailedException;
 import org.elasticsearch.action.support.broadcast.BroadcastOperationResponse;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 import java.util.HashMap;
 import java.util.Iterator;
@@ -124,23 +124,23 @@ public class TermsResponse extends BroadcastOperationResponse implements Iterabl
         return fieldsTermsFreqMap;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
-        numDocs = in.readLong();
+        numDocs = in.readVLong();
-        maxDoc = in.readLong();
+        maxDoc = in.readVLong();
-        numDeletedDocs = in.readLong();
+        numDeletedDocs = in.readVLong();
-        fieldsTermsFreq = new FieldTermsFreq[in.readInt()];
+        fieldsTermsFreq = new FieldTermsFreq[in.readVInt()];
         for (int i = 0; i < fieldsTermsFreq.length; i++) {
             fieldsTermsFreq[i] = readFieldTermsFreq(in);
         }
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
-        out.writeLong(numDocs);
+        out.writeVLong(numDocs);
-        out.writeLong(maxDoc);
+        out.writeVLong(maxDoc);
-        out.writeLong(numDeletedDocs);
+        out.writeVLong(numDeletedDocs);
-        out.writeInt(fieldsTermsFreq.length);
+        out.writeVInt(fieldsTermsFreq.length);
         for (FieldTermsFreq fieldTermsFreq : fieldsTermsFreq) {
             fieldTermsFreq.writeTo(out);
         }

modules/elasticsearch/src/main/java/org/elasticsearch/cluster/ClusterName.java

@@ -19,11 +19,11 @@
 
 package org.elasticsearch.cluster;
 
-import org.elasticsearch.util.io.Streamable;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
+import org.elasticsearch.util.io.stream.Streamable;
 import org.elasticsearch.util.settings.Settings;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
@@ -53,17 +53,17 @@ public class ClusterName implements Streamable {
         return this.value;
     }
 
-    public static ClusterName readClusterName(DataInput in) throws ClassNotFoundException, IOException {
+    public static ClusterName readClusterName(StreamInput in) throws IOException {
         ClusterName clusterName = new ClusterName();
         clusterName.readFrom(in);
         return clusterName;
     }
 
-    @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+    @Override public void readFrom(StreamInput in) throws IOException {
         value = in.readUTF();
     }
 
-    @Override public void writeTo(DataOutput out) throws IOException {
+    @Override public void writeTo(StreamOutput out) throws IOException {
         out.writeUTF(value);
     }
 

modules/elasticsearch/src/main/java/org/elasticsearch/cluster/ClusterState.java

@@ -25,12 +25,12 @@ import org.elasticsearch.cluster.node.Nodes;
 import org.elasticsearch.cluster.routing.RoutingNodes;
 import org.elasticsearch.cluster.routing.RoutingTable;
 import org.elasticsearch.util.Nullable;
-import org.elasticsearch.util.io.ByteArrayDataInputStream;
+import org.elasticsearch.util.io.stream.BytesStreamInput;
-import org.elasticsearch.util.io.ByteArrayDataOutputStream;
+import org.elasticsearch.util.io.stream.BytesStreamOutput;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 import org.elasticsearch.util.settings.Settings;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
@@ -142,23 +142,23 @@ public class ClusterState {
         }
 
         public static byte[] toBytes(ClusterState state) throws IOException {
-            ByteArrayDataOutputStream os = ByteArrayDataOutputStream.Cached.cached();
+            BytesStreamOutput os = BytesStreamOutput.Cached.cached();
             writeTo(state, os);
             return os.copiedByteArray();
         }
 
-        public static ClusterState fromBytes(byte[] data, Settings globalSettings, Node localNode) throws IOException, ClassNotFoundException {
+        public static ClusterState fromBytes(byte[] data, Settings globalSettings, Node localNode) throws IOException {
-            return readFrom(new ByteArrayDataInputStream(data), globalSettings, localNode);
+            return readFrom(new BytesStreamInput(data), globalSettings, localNode);
         }
 
-        public static void writeTo(ClusterState state, DataOutput out) throws IOException {
+        public static void writeTo(ClusterState state, StreamOutput out) throws IOException {
             out.writeLong(state.version());
             MetaData.Builder.writeTo(state.metaData(), out);
             RoutingTable.Builder.writeTo(state.routingTable(), out);
             Nodes.Builder.writeTo(state.nodes(), out);
         }
 
-        public static ClusterState readFrom(DataInput in, @Nullable Settings globalSettings, @Nullable Node localNode) throws ClassNotFoundException, IOException {
+        public static ClusterState readFrom(StreamInput in, @Nullable Settings globalSettings, @Nullable Node localNode) throws IOException {
             Builder builder = new Builder();
             builder.version = in.readLong();
             builder.metaData = MetaData.Builder.readFrom(in, globalSettings);

modules/elasticsearch/src/main/java/org/elasticsearch/cluster/action/index/MappingUpdatedAction.java

@@ -29,10 +29,10 @@ import org.elasticsearch.cluster.ClusterService;
 import org.elasticsearch.cluster.metadata.MetaDataService;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
+import org.elasticsearch.util.io.stream.StreamInput;
+import org.elasticsearch.util.io.stream.StreamOutput;
 import org.elasticsearch.util.settings.Settings;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 /**
@@ -69,10 +69,10 @@ public class MappingUpdatedAction extends TransportMasterNodeOperationAction<Map
     }
 
     public static class MappingUpdatedResponse implements ActionResponse {
-        @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+        @Override public void readFrom(StreamInput in) throws IOException {
         }
 
-        @Override public void writeTo(DataOutput out) throws IOException {
+        @Override public void writeTo(StreamOutput out) throws IOException {
         }
     }
 
@@ -109,14 +109,14 @@ public class MappingUpdatedAction extends TransportMasterNodeOperationAction<Map
             return null;
         }
 
-        @Override public void readFrom(DataInput in) throws IOException, ClassNotFoundException {
+        @Override public void readFrom(StreamInput in) throws IOException {
             super.readFrom(in);
             index = in.readUTF();
             type = in.readUTF();
             mappingSource = in.readUTF();
         }
 
-        @Override public void writeTo(DataOutput out) throws IOException {
+        @Override public void writeTo(StreamOutput out) throws IOException {
             super.writeTo(out);
             out.writeUTF(index);
             out.writeUTF(type);