Use Supplier for StreamInput#readOptionalStreamable

This commit changes the signature of StreamInput#readOptionalStreamable
to accept a Supplier to create new streamables rather than requiring
callers to construct new instances. This has the advantage of avoiding
an allocation in cases when the stream indicates the resulting
streamable is null

core/src/main/java/org/elasticsearch/action/admin/cluster/node/stats/NodeStats.java

@@ -223,8 +223,8 @@ public class NodeStats extends BaseNodeResponse implements ToXContent {
             http = HttpStats.readHttpStats(in);
         }
         breaker = AllCircuitBreakerStats.readOptionalAllCircuitBreakerStats(in);
-        scriptStats = in.readOptionalStreamable(new ScriptStats());
-        discoveryStats = in.readOptionalStreamable(new DiscoveryStats(null));
+        scriptStats = in.readOptionalStreamable(ScriptStats::new);
+        discoveryStats = in.readOptionalStreamable(() -> new DiscoveryStats(null));
 
     }
 

core/src/main/java/org/elasticsearch/action/admin/indices/stats/CommonStats.java

@@ -553,10 +553,10 @@ public class CommonStats implements Streamable, ToXContent {
         if (in.readBoolean()) {
             segments = SegmentsStats.readSegmentsStats(in);
         }
-        translog = in.readOptionalStreamable(new TranslogStats());
-        suggest = in.readOptionalStreamable(new SuggestStats());
-        requestCache = in.readOptionalStreamable(new RequestCacheStats());
-        recoveryStats = in.readOptionalStreamable(new RecoveryStats());
+        translog = in.readOptionalStreamable(TranslogStats::new);
+        suggest = in.readOptionalStreamable(SuggestStats::new);
+        requestCache = in.readOptionalStreamable(RequestCacheStats::new);
+        recoveryStats = in.readOptionalStreamable(RecoveryStats::new);
     }
 
     @Override

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

@@ -346,7 +346,7 @@ public class SearchRequest extends ActionRequest<SearchRequest> implements Indic
         indicesOptions = IndicesOptions.readIndicesOptions(in);
 
         requestCache = in.readOptionalBoolean();
-        template = in.readOptionalStreamable(new Template());
+        template = in.readOptionalStreamable(Template::new);
     }
 
     @Override

core/src/main/java/org/elasticsearch/cluster/routing/RestoreSource.java

@@ -68,7 +68,7 @@ public class RestoreSource implements Streamable, ToXContent {
     }
 
     public static RestoreSource readOptionalRestoreSource(StreamInput in) throws IOException {
-        return in.readOptionalStreamable(new RestoreSource());
+        return in.readOptionalStreamable(RestoreSource::new);
     }
 
     @Override

core/src/main/java/org/elasticsearch/cluster/routing/RoutingValidationException.java

@@ -39,7 +39,7 @@ public class RoutingValidationException extends RoutingException {
 
     public RoutingValidationException(StreamInput in) throws IOException {
         super(in);
-        validation = in.readOptionalStreamable(new RoutingTableValidation());
+        validation = in.readOptionalStreamable(RoutingTableValidation::new);
     }
 
     @Override

core/src/main/java/org/elasticsearch/common/io/stream/StreamInput.java

@@ -52,6 +52,7 @@ import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.function.Supplier;
 
 import static org.elasticsearch.ElasticsearchException.readException;
 import static org.elasticsearch.ElasticsearchException.readStackTrace;
@@ -517,8 +518,9 @@ public abstract class StreamInput extends InputStream {
     /**
      * Serializes a potential null value.
      */
-    public <T extends Streamable> T readOptionalStreamable(T streamable) throws IOException {
+    public <T extends Streamable> T readOptionalStreamable(Supplier<T> supplier) throws IOException {
         if (readBoolean()) {
+            T streamable = supplier.get();
             streamable.readFrom(this);
             return streamable;
         } else {

core/src/main/java/org/elasticsearch/index/engine/CommitStats.java

@@ -62,7 +62,7 @@ public final class CommitStats implements Streamable, ToXContent {
     }
 
     public static CommitStats readOptionalCommitStatsFrom(StreamInput in) throws IOException {
-        return in.readOptionalStreamable(new CommitStats());
+        return in.readOptionalStreamable(CommitStats::new);
     }
 
 

core/src/main/java/org/elasticsearch/indices/breaker/AllCircuitBreakerStats.java

@@ -57,7 +57,7 @@ public class AllCircuitBreakerStats implements Streamable, ToXContent {
     }
 
     public static AllCircuitBreakerStats readOptionalAllCircuitBreakerStats(StreamInput in) throws IOException {
-        AllCircuitBreakerStats stats = in.readOptionalStreamable(new AllCircuitBreakerStats());
+        AllCircuitBreakerStats stats = in.readOptionalStreamable(AllCircuitBreakerStats::new);
         return stats;
     }
 

core/src/main/java/org/elasticsearch/indices/breaker/CircuitBreakerStats.java

@@ -74,7 +74,7 @@ public class CircuitBreakerStats implements Streamable, ToXContent {
     }
 
     public static CircuitBreakerStats readOptionalCircuitBreakerStats(StreamInput in) throws IOException {
-        CircuitBreakerStats stats = in.readOptionalStreamable(new CircuitBreakerStats());
+        CircuitBreakerStats stats = in.readOptionalStreamable(CircuitBreakerStats::new);
         return stats;
     }
 

core/src/main/java/org/elasticsearch/search/aggregations/InternalAggregations.java

@@ -194,7 +194,7 @@ public class InternalAggregations implements Aggregations, ToXContent, Streamabl
     }
 
     public static InternalAggregations readOptionalAggregations(StreamInput in) throws IOException {
-        return in.readOptionalStreamable(new InternalAggregations());
+        return in.readOptionalStreamable(InternalAggregations::new);
     }
 
     @Override

core/src/main/java/org/elasticsearch/search/internal/InternalSearchHit.java

@@ -559,7 +559,7 @@ public class InternalSearchHit implements SearchHit {
         score = in.readFloat();
         id = in.readText();
         type = in.readText();
-        nestedIdentity = in.readOptionalStreamable(new InternalNestedIdentity());
+        nestedIdentity = in.readOptionalStreamable(InternalNestedIdentity::new);
         version = in.readLong();
         source = in.readBytesReference();
         if (source.length() == 0) {
@@ -810,7 +810,7 @@ public class InternalSearchHit implements SearchHit {
         public void readFrom(StreamInput in) throws IOException {
             field = in.readOptionalText();
             offset = in.readInt();
-            child = in.readOptionalStreamable(new InternalNestedIdentity());
+            child = in.readOptionalStreamable(InternalNestedIdentity::new);
         }
 
         @Override

core/src/main/java/org/elasticsearch/search/internal/ShardSearchLocalRequest.java

@@ -180,7 +180,7 @@ public class ShardSearchLocalRequest extends ContextAndHeaderHolder implements S
         types = in.readStringArray();
         filteringAliases = in.readStringArray();
         nowInMillis = in.readVLong();
-        template = in.readOptionalStreamable(new Template());
+        template = in.readOptionalStreamable(Template::new);
         requestCache = in.readOptionalBoolean();
     }
 

core/src/main/java/org/elasticsearch/snapshots/RestoreInfo.java

@@ -190,7 +190,7 @@ public class RestoreInfo implements ToXContent, Streamable {
      * @return restore info
      */
     public static RestoreInfo readOptionalRestoreInfo(StreamInput in) throws IOException {
-        return in.readOptionalStreamable(new RestoreInfo());
+        return in.readOptionalStreamable(RestoreInfo::new);
     }
 
 }

core/src/main/java/org/elasticsearch/snapshots/SnapshotInfo.java

@@ -324,7 +324,7 @@ public class SnapshotInfo implements ToXContent, Streamable {
      * @return deserialized snapshot info or null
      */
     public static SnapshotInfo readOptionalSnapshotInfo(StreamInput in) throws IOException {
-        return in.readOptionalStreamable(new SnapshotInfo());
+        return in.readOptionalStreamable(SnapshotInfo::new);
     }
 
 }