1 files changed, 46 insertions, 18 deletions

diff --git a/core/src/main/java/org/elasticsearch/search/aggregations/pipeline/PipelineAggregator.java b/core/src/main/java/org/elasticsearch/search/aggregations/pipeline/PipelineAggregator.java
index 8f7ea955e9..8c43f7460d 100644
--- a/core/src/main/java/org/elasticsearch/search/aggregations/pipeline/PipelineAggregator.java
+++ b/core/src/main/java/org/elasticsearch/search/aggregations/pipeline/PipelineAggregator.java
@@ -21,6 +21,7 @@ package org.elasticsearch.search.aggregations.pipeline;
 
 
 import org.elasticsearch.common.ParseField;
+import org.elasticsearch.common.io.stream.NamedWriteable;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Streamable;
@@ -33,7 +34,8 @@ import org.elasticsearch.search.aggregations.PipelineAggregationBuilder;
 import java.io.IOException;
 import java.util.Map;
 
-public abstract class PipelineAggregator implements Streamable {
+public abstract class PipelineAggregator implements Streamable, NamedWriteable {
+    // NORELEASE remove Streamable
 
     /**
      * Parse the {@link PipelineAggregationBuilder} from a {@link QueryParseContext}.
@@ -73,25 +75,36 @@ public abstract class PipelineAggregator implements Streamable {
         this.metaData = metaData;
     }
 
-    public String name() {
-        return name;
+    /**
+     * Read from a stream.
+     */
+    protected PipelineAggregator(StreamInput in) throws IOException {
+        name = in.readString();
+        bucketsPaths = in.readStringArray();
+        metaData = in.readMap();
     }
 
-    public String[] bucketsPaths() {
-        return bucketsPaths;
+    @Override
+    public final void readFrom(StreamInput in) throws IOException {
+        try {
+            getWriteableName(); // Throws UnsupportedOperationException if this aggregation should be read using old style Streams
+            assert false : "Used reading constructor instead";
+        } catch (UnsupportedOperationException e) {
+            // OK
+        }
+        name = in.readString();
+        bucketsPaths = in.readStringArray();
+        metaData = in.readMap();
+        doReadFrom(in);
     }
 
-    public Map<String, Object> metaData() {
-        return metaData;
+    protected void doReadFrom(StreamInput in) throws IOException {
+        throw new UnsupportedOperationException("Use reading constructor instead"); // NORELEASE remove when we remove Streamable
     }
 
-    public abstract Type type();
-
-    public abstract InternalAggregation reduce(InternalAggregation aggregation, ReduceContext reduceContext);
-
     @Override
     public final void writeTo(StreamOutput out) throws IOException {
-        out.writeString(name);
+        out.writeString(name);  // NORELEASE remote writing the name - it is automatically handled with writeNamedWriteable
         out.writeStringArray(bucketsPaths);
         out.writeMap(metaData);
         doWriteTo(out);
@@ -100,12 +113,27 @@ public abstract class PipelineAggregator implements Streamable {
     protected abstract void doWriteTo(StreamOutput out) throws IOException;
 
     @Override
-    public final void readFrom(StreamInput in) throws IOException {
-        name = in.readString();
-        bucketsPaths = in.readStringArray();
-        metaData = in.readMap();
-        doReadFrom(in);
+    public String getWriteableName() {
+        // NORELEASE remove me when all InternalAggregations override it
+        throw new UnsupportedOperationException("Override on every class");
+    }
+
+    public String name() {
+        return name;
+    }
+
+    public String[] bucketsPaths() {
+        return bucketsPaths;
+    }
+
+    public Map<String, Object> metaData() {
+        return metaData;
     }
 
-    protected abstract void doReadFrom(StreamInput in) throws IOException;
+    public Type type() {
+        // NORELEASE remove this method
+        throw new UnsupportedOperationException(getClass().getName() + " used type but should Use getWriteableName instead");
+    }
+
+    public abstract InternalAggregation reduce(InternalAggregation aggregation, ReduceContext reduceContext);
 }