=str - fixing TODO about scaladsl & javadsl Source/Sink/Flow/BidiFlow.wrap not having to needlessly re-wrap

akka-stream/src/main/scala/akka/stream/javadsl/BidiFlow.scala

@@ -17,7 +17,11 @@ object BidiFlow {
    * A graph with the shape of a BidiFlow logically is a BidiFlow, this method makes
    * it so also in type.
    */
-  def wrap[I1, O1, I2, O2, M](g: Graph[BidiShape[I1, O1, I2, O2], M]): BidiFlow[I1, O1, I2, O2, M] = new BidiFlow(scaladsl.BidiFlow.wrap(g))
+  def wrap[I1, O1, I2, O2, M](g: Graph[BidiShape[I1, O1, I2, O2], M]): BidiFlow[I1, O1, I2, O2, M] =
+    g match {
+      case bidi: BidiFlow[I1, O1, I2, O2, M] ⇒ bidi
+      case other                             ⇒ new BidiFlow(scaladsl.BidiFlow.wrap(other))
+    }
 
   /**
    * Create a BidiFlow where the top and bottom flows are just one simple mapping

akka-stream/src/main/scala/akka/stream/javadsl/Flow.scala

@@ -39,7 +39,11 @@ object Flow {
    * A graph with the shape of a flow logically is a flow, this method makes
    * it so also in type.
    */
-  def wrap[I, O, M](g: Graph[FlowShape[I, O], M]): Flow[I, O, M] = new Flow(scaladsl.Flow.wrap(g))
+  def wrap[I, O, M](g: Graph[FlowShape[I, O], M]): Flow[I, O, M] =
+    g match {
+      case f: Flow[I, O, M] ⇒ f
+      case other            ⇒ new Flow(scaladsl.Flow.wrap(other))
+    }
 
   /**
    * Helper to create `Flow` from a pair of sink and source.

akka-stream/src/main/scala/akka/stream/javadsl/Sink.scala

@@ -118,7 +118,11 @@ object Sink {
    * A graph with the shape of a sink logically is a sink, this method makes
    * it so also in type.
    */
-  def wrap[T, M](g: Graph[SinkShape[T], M]): Sink[T, M] = new Sink(scaladsl.Sink.wrap(g))
+  def wrap[T, M](g: Graph[SinkShape[T], M]): Sink[T, M] =
+    g match {
+      case s: Sink[T, M] ⇒ s
+      case other         ⇒ new Sink(scaladsl.Sink.wrap(other))
+    }
 }
 
 /**

akka-stream/src/main/scala/akka/stream/javadsl/Source.scala

@@ -112,7 +112,9 @@ object Source {
     // but there is not anything we can do to prevent that from happening.
     // ConcurrentModificationException will be thrown in some cases.
     val scalaIterable = new immutable.Iterable[O] {
+
       import collection.JavaConverters._
+
       override def iterator: Iterator[O] = iterable.iterator().asScala
     }
     new Source(scaladsl.Source(scalaIterable))
@@ -215,7 +217,11 @@ object Source {
    * A graph with the shape of a source logically is a source, this method makes
    * it so also in type.
    */
-  def wrap[T, M](g: Graph[SourceShape[T], M]): Source[T, M] = new Source(scaladsl.Source.wrap(g))
+  def wrap[T, M](g: Graph[SourceShape[T], M]): Source[T, M] =
+    g match {
+      case s: Source[T, M] ⇒ s
+      case other           ⇒ new Source(scaladsl.Source.wrap(other))
+    }
 }
 
 /**

akka-stream/src/main/scala/akka/stream/scaladsl/BidiFlow.scala

@@ -130,7 +130,11 @@ object BidiFlow extends BidiFlowApply {
    * A graph with the shape of a flow logically is a flow, this method makes
    * it so also in type.
    */
-  def wrap[I1, O1, I2, O2, Mat](graph: Graph[BidiShape[I1, O1, I2, O2], Mat]): BidiFlow[I1, O1, I2, O2, Mat] = new BidiFlow(graph.module)
+  def wrap[I1, O1, I2, O2, Mat](graph: Graph[BidiShape[I1, O1, I2, O2], Mat]): BidiFlow[I1, O1, I2, O2, Mat] =
+    graph match {
+      case bidi: BidiFlow[I1, O1, I2, O2, Mat] ⇒ bidi
+      case other                               ⇒ new BidiFlow(other.module)
+    }
 
   /**
    * Create a BidiFlow where the top and bottom flows are just one simple mapping

akka-stream/src/main/scala/akka/stream/scaladsl/Flow.scala

@@ -302,7 +302,11 @@ object Flow extends FlowApply {
    * A graph with the shape of a flow logically is a flow, this method makes
    * it so also in type.
    */
-  def wrap[I, O, M](g: Graph[FlowShape[I, O], M]): Flow[I, O, M] = new Flow(g.module)
+  def wrap[I, O, M](g: Graph[FlowShape[I, O], M]): Flow[I, O, M] =
+    g match {
+      case f: Flow[I, O, M] ⇒ f
+      case other            ⇒ new Flow(other.module)
+    }
 
   /**
    * Helper to create `Flow` from a pair of sink and source.

akka-stream/src/main/scala/akka/stream/scaladsl/Sink.scala

@@ -58,7 +58,11 @@ object Sink extends SinkApply {
    * A graph with the shape of a sink logically is a sink, this method makes
    * it so also in type.
    */
-  def wrap[T, M](g: Graph[SinkShape[T], M]): Sink[T, M] = new Sink(g.module)
+  def wrap[T, M](g: Graph[SinkShape[T], M]): Sink[T, M] =
+    g match {
+      case s: Sink[T, M] ⇒ s
+      case other         ⇒ new Sink(other.module)
+    }
 
   /**
    * Helper to create [[Sink]] from `Subscriber`.

akka-stream/src/main/scala/akka/stream/scaladsl/Source.scala

@@ -202,8 +202,10 @@ object Source extends SourceApply {
    * A graph with the shape of a source logically is a source, this method makes
    * it so also in type.
    */
-  // TODO optimize if no wrapping needed
+  def wrap[T, M](g: Graph[SourceShape[T], M]): Source[T, M] = g match {
-  def wrap[T, M](g: Graph[SourceShape[T], M]): Source[T, M] = new Source(g.module)
+    case s: Source[T, M] ⇒ s
+    case other           ⇒ new Source(other.module)
+  }
 
   /**
    * Helper to create [[Source]] from `Iterable`.