document FlowOps extension

akka-docs/rst/scala/code/docs/stream/GraphStageDocSpec.scala

@@ -519,4 +519,32 @@ class GraphStageDocSpec extends AkkaSpec {
     sub.cancel()
   }
 
+  "Demonstrate stream extension" when {
+
+    "targeting a Source" in {
+      //#extending-source
+      implicit class SourceDuplicator[Out, Mat](s: Source[Out, Mat]) {
+        def duplicateElements: Source[Out, Mat] = s.via(new Duplicator)
+      }
+
+      val s = Source(1 to 3).duplicateElements
+
+      s.runWith(Sink.seq).futureValue should ===(Seq(1, 1, 2, 2, 3, 3))
+      //#extending-source
+    }
+
+    "targeting a Flow" in {
+      //#extending-flow
+      implicit class FlowDuplicator[In, Out, Mat](s: Flow[In, Out, Mat]) {
+        def duplicateElements: Flow[In, Out, Mat] = s.via(new Duplicator)
+      }
+
+      val f = Flow[Int].duplicateElements
+
+      Source(1 to 3).via(f).runWith(Sink.seq).futureValue should ===(Seq(1, 1, 2, 2, 3, 3))
+      //#extending-flow
+    }
+
+  }
+
 }

akka-docs/rst/scala/code/docs/stream/TwitterStreamQuickstartDocSpec.scala

@@ -32,7 +32,8 @@ object TwitterStreamQuickstartDocSpec {
   //#model
 
   //#tweet-source
-  val tweets: Source[Tweet, NotUsed] //#tweet-source
+  val tweets: Source[Tweet, NotUsed]
+  //#tweet-source
   = Source(
     Tweet(Author("rolandkuhn"), System.currentTimeMillis, "#akka rocks!") ::
       Tweet(Author("patriknw"), System.currentTimeMillis, "#akka !") ::