changeset 5866:a0aade7ea2c6 it2-bootstrap

Redefine Streamable to include both serial and parallel stream accessors; fallback implementation for parallel stream is serial
author briangoetz
date Mon, 27 Aug 2012 15:17:08 -0400
parents 0b7ec96b2ec9
children fb20a07d5b76
files src/share/classes/java/util/Arrays.java src/share/classes/java/util/streams/Streamable.java src/share/classes/java/util/streams/ops/TreeUtils.java test-ng/tests/org/openjdk/tests/java/util/streams/ops/StreamOpTestCase.java
diffstat 4 files changed, 52 insertions(+), 4 deletions(-) [+]
line wrap: on
line diff
--- a/src/share/classes/java/util/Arrays.java	Mon Aug 27 00:01:19 2012 -0400
+++ b/src/share/classes/java/util/Arrays.java	Mon Aug 27 15:17:08 2012 -0400
@@ -3888,6 +3888,24 @@
         return Streams.parallel(new ArrayParallelStreamAccessor<>(array, offset, length));
     }
 
+    public static<T> Streamable<T> asStreamable(final T[] array, final int offset, final int length) {
+        return new Streamable<T>() {
+            @Override
+            public Stream<T> stream() {
+                return Arrays.stream(array, offset, length);
+            }
+
+            @Override
+            public Stream<T> parallel() {
+                return Arrays.parallel(array, offset, length);
+            }
+        };
+    }
+
+    public static<T> Streamable<T> asStreamable(T[] array) {
+        return asStreamable(array, 0, array.length);
+    }
+
     public static<T> Sink<T> sink(T[] array) {
         return sink(array, 0, array.length);
     }
--- a/src/share/classes/java/util/streams/Streamable.java	Mon Aug 27 00:01:19 2012 -0400
+++ b/src/share/classes/java/util/streams/Streamable.java	Mon Aug 27 15:17:08 2012 -0400
@@ -37,4 +37,5 @@
  */
 public interface Streamable<T> {
     Stream<T> stream();
+    Stream<T> parallel() default { return stream(); }
 }
--- a/src/share/classes/java/util/streams/ops/TreeUtils.java	Mon Aug 27 00:01:19 2012 -0400
+++ b/src/share/classes/java/util/streams/ops/TreeUtils.java	Mon Aug 27 15:17:08 2012 -0400
@@ -204,7 +204,7 @@
         }
     }
 
-    public static interface Node<T> extends Traversable<T>, Splittable<T>, Sized {
+    public static interface Node<T> extends Traversable<T>, Streamable<T>, Splittable<T>, Sized {
         void copyTo(T[] array, int offset);
     }
 
@@ -234,8 +234,18 @@
             }
 
             @Override
+            public Stream<T> stream() {
+                return Arrays.stream(data);
+            }
+
+            @Override
+            public Stream<T> parallel() {
+                return Arrays.parallel(data);
+            }
+
+            @Override
             public Spliterator<T> spliterator() {
-                return Arrays.spliterator(data, 0, data.length);
+                return Arrays.spliterator(data);
             }
 
             @Override
@@ -265,6 +275,16 @@
             }
 
             @Override
+            public Stream<T> stream() {
+                return Streams.stream(data.iterator(), data.size());
+            }
+
+            @Override
+            public Stream<T> parallel() {
+                return Arrays.parallel((T[]) data.toArray());
+            }
+
+            @Override
             public Iterator<T> iterator() {
                 return data.iterator();
             }
@@ -330,6 +350,16 @@
         }
 
         @Override
+        public Stream<T> stream() {
+            return Streams.stream(iterator(), size());
+        }
+
+        @Override
+        public Stream<T> parallel() {
+            return Streams.parallel(spliterator(), size());
+        }
+
+        @Override
         public Spliterator<T> spliterator() {
             return new InternalNodeSpliterator<>(this);
         }
--- a/test-ng/tests/org/openjdk/tests/java/util/streams/ops/StreamOpTestCase.java	Mon Aug 27 00:01:19 2012 -0400
+++ b/test-ng/tests/org/openjdk/tests/java/util/streams/ops/StreamOpTestCase.java	Mon Aug 27 15:17:08 2012 -0400
@@ -63,7 +63,7 @@
         return ((LinearPipeline) Streams.stream(Arrays.iterator(data), data.length)).pipeline(op);
     }
 
-    protected<T, U> Stream<T> par(T[] data, IntermediateOp<T,U> op) {
+    protected<T, U> Stream<U> par(T[] data, IntermediateOp<T,U> op) {
         return ((LinearPipeline) Streams.parallel(Arrays.spliterator(data), data.length)).pipeline(op);
     }
 
@@ -87,7 +87,6 @@
     protected <T, U> void assertConsistentOpBehavior(T[] data,
                                                      IntermediateOp<T, U> op,
                                                      Mapper<Integer, Integer> lengthMapper) {
-
         int length = (lengthMapper == null) ? 100 : lengthMapper.map(data.length);
 
         ArraySink<U> refResult = new ArraySink<>(length);