changeset 6395:bac0249a3f18

StreamShape consistently uses AbstractPipeline. Methods on AbstractPipeline to chain to return a pipeline or a stream.
author psandoz
date Tue, 13 Nov 2012 10:30:17 +0100
parents ba5cff68c181
children 38dc6745cd9b
files src/share/classes/java/util/streams/AbstractPipeline.java src/share/classes/java/util/streams/StreamShape.java src/share/classes/java/util/streams/StreamShapeFactory.java test-ng/tests/org/openjdk/tests/java/util/streams/OpTestCase.java
diffstat 4 files changed, 26 insertions(+), 22 deletions(-) [+]
line wrap: on
line diff
--- a/src/share/classes/java/util/streams/AbstractPipeline.java	Tue Nov 13 10:10:14 2012 +0100
+++ b/src/share/classes/java/util/streams/AbstractPipeline.java	Tue Nov 13 10:30:17 2012 +0100
@@ -447,6 +447,17 @@
     // Chaining and result methods
 
     /**
+     * Chain an operation to the tail of this pipeline to create a new pipeline.
+     *
+     * @param op the operation to chain.
+     * @param <E_NEXT> the type of elements output from the new pipeline.
+     * @return the new pipeline.
+     */
+    public <E_NEXT> AbstractPipeline<E_OUT, E_NEXT> chain(IntermediateOp<E_OUT, E_NEXT> op) {
+        return op.outputShape().chain(this, op);
+    }
+
+    /**
      * Chain an operation to the tail of this pipeline to create a new stream.
      *
      * @param op the operation to chain.
@@ -456,7 +467,7 @@
      */
     @SuppressWarnings("unchecked")
     public <E_NEXT, S extends BaseStream<E_NEXT>> S pipeline(IntermediateOp<E_OUT, E_NEXT> op) {
-        return (S) op.outputShape().chain(this, op);
+        return (S) chain(op);
     }
 
     /**
--- a/src/share/classes/java/util/streams/StreamShape.java	Tue Nov 13 10:10:14 2012 +0100
+++ b/src/share/classes/java/util/streams/StreamShape.java	Tue Nov 13 10:30:17 2012 +0100
@@ -41,7 +41,7 @@
     Class<S> getStreamType();
 
     /**
-     * Create a new stream by chaining an intermediate operation to an upstream pipeline.
+     * Create a new pipeline by chaining an intermediate operation to an upstream pipeline.
      * <p>
      * The output shape if the upstream pipeline must be the same as the input shape of
      * the intermediate operation.
@@ -49,10 +49,10 @@
      * @param upstream the upstream pipeline.
      * @param op the intermediate operation.
      * @param <U> the type of elements output from the upstream pipeline and input to the new stream.
-     * @param <V> the type of elements output from the new stream.
-     * @return a the new stream.
+     * @param <V> the type of elements output from the new pipeline.
+     * @return a the new pipeline.
      */
-    <U, V> S chain(AbstractPipeline<?, U> upstream, IntermediateOp<U, V> op);
+    <U, V> AbstractPipeline<U, V> chain(AbstractPipeline<?, U> upstream, IntermediateOp<U, V> op);
 
 
     // @@@ Should this be the place where the base mechanisms to create a stream reside?
--- a/src/share/classes/java/util/streams/StreamShapeFactory.java	Tue Nov 13 10:10:14 2012 +0100
+++ b/src/share/classes/java/util/streams/StreamShapeFactory.java	Tue Nov 13 10:30:17 2012 +0100
@@ -45,7 +45,7 @@
             }
 
             @Override
-            public <U, V> Stream<V> chain(AbstractPipeline<?, U> upstream, IntermediateOp<U, V> op) {
+            public <U, V> AbstractPipeline<U, V> chain(AbstractPipeline<?, U> upstream, IntermediateOp<U, V> op) {
                 return new ReferencePipeline<>(upstream, op);
             }
 
--- a/test-ng/tests/org/openjdk/tests/java/util/streams/OpTestCase.java	Tue Nov 13 10:10:14 2012 +0100
+++ b/test-ng/tests/org/openjdk/tests/java/util/streams/OpTestCase.java	Tue Nov 13 10:30:17 2012 +0100
@@ -316,33 +316,26 @@
 
     //
 
-    @SuppressWarnings({ "rawtypes", "unchecked" })
-    static<T> AbstractPipeline<?, T> chain(AbstractPipeline upstream, IntermediateOp<?, T> op) {
-        BaseStream pipe = upstream.pipeline(op);
-        if (pipe instanceof AbstractPipeline) {
-            return (AbstractPipeline<?, T>) pipe;
-        } else {
-            throw new IllegalStateException("Chained pipeline is not an instance of AbstractPipeline: " + pipe);
-        }
+    @SuppressWarnings({"rawtypes", "unchecked"})
+    static <T> AbstractPipeline<?, T> chain(AbstractPipeline upstream, IntermediateOp<?, T> op) {
+        return upstream.chain(op);
     }
 
-    @SuppressWarnings({ "rawtypes", "unchecked" })
-    static<U> U chain(AbstractPipeline pipe, TerminalOp<?, U> op) {
+    @SuppressWarnings({"rawtypes", "unchecked"})
+    static <U> U chain(AbstractPipeline pipe, TerminalOp<?, U> op) {
         return (U) pipe.pipeline(op);
     }
 
-    @SuppressWarnings({ "rawtypes", "unchecked" })
+    @SuppressWarnings({"rawtypes", "unchecked"})
     static AbstractPipeline<?, ?> chain(AbstractPipeline pipe, IntermediateOp... ops) {
         for (IntermediateOp op : ops)
             pipe = chain(pipe, op);
         return pipe;
     }
 
-    @SuppressWarnings({ "rawtypes", "unchecked" })
-    static<U> U chain(AbstractPipeline pipe, TerminalOp<?, U> terminal, IntermediateOp... ops) {
-        for (IntermediateOp op : ops)
-            pipe = chain(pipe, op);
-        return chain(pipe, terminal);
+    @SuppressWarnings("rawtypes")
+    static <U> U chain(AbstractPipeline pipe, TerminalOp<?, U> terminal, IntermediateOp... ops) {
+        return chain(chain(pipe, ops), terminal);
     }
 
     // Test data