diff --git a/akka-stream/src/main/scala/akka/stream/javadsl/BidiFlow.scala b/akka-stream/src/main/scala/akka/stream/javadsl/BidiFlow.scala
index 162aa2ba0b..f0082a7cec 100644
--- a/akka-stream/src/main/scala/akka/stream/javadsl/BidiFlow.scala
+++ b/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
diff --git a/akka-stream/src/main/scala/akka/stream/javadsl/Flow.scala b/akka-stream/src/main/scala/akka/stream/javadsl/Flow.scala
index aa4990fa86..d331ebfcb3 100644
--- a/akka-stream/src/main/scala/akka/stream/javadsl/Flow.scala
+++ b/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.
diff --git a/akka-stream/src/main/scala/akka/stream/javadsl/Sink.scala b/akka-stream/src/main/scala/akka/stream/javadsl/Sink.scala
index c993cf59a3..af3c6480ed 100644
--- a/akka-stream/src/main/scala/akka/stream/javadsl/Sink.scala
+++ b/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))
+    }
 }
 
 /**
diff --git a/akka-stream/src/main/scala/akka/stream/javadsl/Source.scala b/akka-stream/src/main/scala/akka/stream/javadsl/Source.scala
index c545db1963..1d4336e85e 100644
--- a/akka-stream/src/main/scala/akka/stream/javadsl/Source.scala
+++ b/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))
+    }
 }
 
 /**
diff --git a/akka-stream/src/main/scala/akka/stream/scaladsl/BidiFlow.scala b/akka-stream/src/main/scala/akka/stream/scaladsl/BidiFlow.scala
index 66d4e3f8da..65c1b77fef 100644
--- a/akka-stream/src/main/scala/akka/stream/scaladsl/BidiFlow.scala
+++ b/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
diff --git a/akka-stream/src/main/scala/akka/stream/scaladsl/Flow.scala b/akka-stream/src/main/scala/akka/stream/scaladsl/Flow.scala
index c4b7296843..6a240e63a4 100644
--- a/akka-stream/src/main/scala/akka/stream/scaladsl/Flow.scala
+++ b/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.
diff --git a/akka-stream/src/main/scala/akka/stream/scaladsl/Sink.scala b/akka-stream/src/main/scala/akka/stream/scaladsl/Sink.scala
index d1d784fc85..d3b0c5cf88 100644
--- a/akka-stream/src/main/scala/akka/stream/scaladsl/Sink.scala
+++ b/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`.
diff --git a/akka-stream/src/main/scala/akka/stream/scaladsl/Source.scala b/akka-stream/src/main/scala/akka/stream/scaladsl/Source.scala
index 99551c371a..72f5b03770 100644
--- a/akka-stream/src/main/scala/akka/stream/scaladsl/Source.scala
+++ b/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] = new Source(g.module)
+  def wrap[T, M](g: Graph[SourceShape[T], M]): Source[T, M] = g match {
+    case s: Source[T, M] ⇒ s
+    case other           ⇒ new Source(other.module)
+  }
 
   /**
    * Helper to create [[Source]] from `Iterable`.