--- a/src/share/classes/java/util/streams/Sink.java	Fri Sep 21 16:24:40 2012 -0700
+++ b/src/share/classes/java/util/streams/Sink.java	Fri Sep 21 16:30:57 2012 -0700
@@ -115,4 +115,24 @@
             downstream.end();
         }
     }
+
+    public static class ChainedMapAdapter<K, V> extends ChainedMap<K, V> {
+        public ChainedMapAdapter(ChainedValue<? extends Mapping<K, V>> downstream) {
+            super(downstream);
+        }
+
+        @Override
+        public void accept(Mapping<K, V> e) {
+            downstream.accept(e);
+        }
+
+        @Override
+        public void accept(K k, V v) {
+            downstream.accept(new Mapping.MappingValue<>(k, v));
+        }
+
+        public static <K, V> ChainedMap adapt(ChainedValue<? extends Mapping<K, V>> sink) {
+            return new ChainedMapAdapter<K, V>(sink);
+        }
+    }
 }

--- a/src/share/classes/java/util/streams/ops/MapSortedOp.java	Fri Sep 21 16:24:40 2012 -0700
+++ b/src/share/classes/java/util/streams/ops/MapSortedOp.java	Fri Sep 21 16:30:57 2012 -0700
@@ -37,15 +37,10 @@
  * @param <V> Values associated with keys to be sorted
  *
  */
-public class MapSortedOp<K extends Comparable<? super K>,V> implements StatefulOp<Mapping<K,V>, Mapping<K,V>> {
+public class MapSortedOp<K extends Comparable<? super K>, V> extends SortedOp<Mapping<K, V>> {
     private static final int DEFAULT_PRIORITY_QUEUE_SIZE = 16;
 
     /**
-     * Comparator used for sorting.
-     */
-    private final Comparator<? super Mapping<K,V>> comparator;
-
-    /**
      * Sort using natural order of {@literal <K>} which must be
      * {@code Comparable}.
      */
@@ -60,56 +55,19 @@
      * @param comparator The comparator to be used to evaluate ordering.
      */
     public MapSortedOp(Comparator<? super Mapping<K,V>> comparator) {
-        this.comparator = Objects.requireNonNull(comparator);
-    }
-
-    @Override
-    public int getStreamFlags(int upstreamFlags) {
-        return upstreamFlags | Stream.FLAG_SORTED;
+        super(Objects.requireNonNull(comparator));
     }
 
     @Override
     public Sink<Mapping<K, V>, K, V> wrapSink(Sink sink) {
-        return new Sink.ChainedMap<K, V>(sink) {
-            PriorityQueue<Mapping<K,V>> pq;
-            @Override
-            public void begin(int size) {
-                pq = new PriorityQueue<>(size > 0 ? size : DEFAULT_PRIORITY_QUEUE_SIZE, comparator);
-            }
-
-            @Override
-            public void end() {
-                while (!pq.isEmpty()) {
-                    downstream.accept(pq.remove());
-                }
-            }
-
-            @Override
-            public void accept(K k, V v) {
-                // @@@ Do something here!
-            }
-
-            @Override
-            public void accept(Mapping<K,V> m) {
-                pq.add(m);
-            }
-        };
+        return new Sink.ChainedMapAdapter<K, V>(
+            (Sink.ChainedValue<Mapping<K, V>>) super.wrapSink(sink));
     }
 
     @Override
     public MapIterator<K,V> wrapIterator(Iterator<Mapping<K, V>> iterator) {
         Objects.requireNonNull(iterator);
-        return iterator(iterator, comparator);
-    }
-
-    private static <K,V> MapIterator<K,V> iterator(Iterator<Mapping<K, V>> iterator, Comparator<? super Mapping<K,V>> comparator) {
-        Objects.requireNonNull(iterator);
-        Objects.requireNonNull(comparator);
-        final PriorityQueue<Mapping<K,V>> pq = new PriorityQueue<>(DEFAULT_PRIORITY_QUEUE_SIZE, comparator);
-        while (iterator.hasNext()) {
-            pq.add(iterator.next());
-        }
-        return MapIterator.IteratorAdapter.adapt(pq.iterator());
+        return MapIterator.IteratorAdapter.adapt(super.wrapIterator(iterator));
     }
 
     @Override

--- a/src/share/classes/java/util/streams/ops/SortedOp.java	Fri Sep 21 16:24:40 2012 -0700
+++ b/src/share/classes/java/util/streams/ops/SortedOp.java	Fri Sep 21 16:30:57 2012 -0700
@@ -36,7 +36,7 @@
  *
  * @author Brian Goetz
  */
-public class SortedOp<T> implements StatefulOp<T,T> {
+public class SortedOp<T> implements StatefulOp<T, T> {
     private static final int DEFAULT_PRIORITY_QUEUE_SIZE = 16;
 
     /**