Testkit sink write

core/build.sbt

@@ -295,32 +295,33 @@ lazy val cloudflowStreamlets =
       crossScalaVersions := Vector(Dependencies.Scala212, Dependencies.Scala213),
       scalafmtOnCompile := true)
 
-lazy val cloudflowAkkastream =
+lazy val cloudflowAkka =
   Project(id = "cloudflow-akka", base = file("cloudflow-akka"))
     .enablePlugins(GenJavadocPlugin, JavaFormatterPlugin, ScalafmtPlugin)
     .dependsOn(cloudflowStreamlets)
-    .settings(Dependencies.cloudflowAkkastream)
+    .settings(Dependencies.cloudflowAkka)
     .settings(
       scalaVersion := Dependencies.Scala212,
       crossScalaVersions := Vector(Dependencies.Scala212, Dependencies.Scala213),
       javacOptions += "-Xlint:deprecation",
       scalafmtOnCompile := true)
 
-lazy val cloudflowAkkastreamTestkit =
+lazy val cloudflowAkkaTestkit =
   Project(id = "cloudflow-akka-testkit", base = file("cloudflow-akka-testkit"))
     .enablePlugins(GenJavadocPlugin, JavaFormatterPlugin, ScalafmtPlugin)
-    .dependsOn(cloudflowAkkastream)
-    .settings(Dependencies.cloudflowAkkastreamTestkit)
+    .dependsOn(cloudflowAkka)
+    .settings(Dependencies.cloudflowAkkaTestkit)
     .settings(
       scalaVersion := Dependencies.Scala212,
       crossScalaVersions := Vector(Dependencies.Scala212, Dependencies.Scala213),
       scalafmtOnCompile := true,
-      javacOptions ++= Seq("-Xlint:deprecation", "-Xlint:unchecked"))
+      javacOptions ++= Seq("-Xlint:deprecation", "-Xlint:unchecked"),
+      (sourceGenerators in Test) += (avroScalaGenerateSpecific in Test).taskValue)
 
-lazy val cloudflowAkkastreamUtil =
+lazy val cloudflowAkkaUtil =
   Project(id = "cloudflow-akka-util", base = file("cloudflow-akka-util"))
     .enablePlugins(GenJavadocPlugin, JavaFormatterPlugin, ScalafmtPlugin)
-    .dependsOn(cloudflowAkkastream, (cloudflowAkkastreamTestkit % "test->test").classpathDependency)
+    .dependsOn(cloudflowAkka, (cloudflowAkkaTestkit % "test->test").classpathDependency)
     .settings(Dependencies.cloudflowAkkaUtil)
     .settings(
       scalaVersion := Dependencies.Scala212,
@@ -329,11 +330,11 @@ lazy val cloudflowAkkastreamUtil =
       javacOptions += "-Xlint:deprecation",
       (Test / sourceGenerators) += (Test / avroScalaGenerateSpecific).taskValue)
 
-lazy val cloudflowAkkastreamTests =
+lazy val cloudflowAkkaTests =
   Project(id = "cloudflow-akka-tests", base = file("cloudflow-akka-tests"))
     .enablePlugins(JavaFormatterPlugin, ScalafmtPlugin)
-    .dependsOn(cloudflowAkkastream, (cloudflowAkkastreamTestkit % "test->test").classpathDependency)
-    .settings(Dependencies.cloudflowAkkastreamTests)
+    .dependsOn(cloudflowAkka, (cloudflowAkkaTestkit % "test->test").classpathDependency)
+    .settings(Dependencies.cloudflowAkkaTests)
     .settings(
       scalaVersion := Dependencies.Scala212,
       crossScalaVersions := Vector(Dependencies.Scala212, Dependencies.Scala213),
@@ -495,9 +496,9 @@ lazy val root = Project(id = "root", base = file("."))
     },
     ScalaUnidoc / unidoc / unidocProjectFilter := inProjects(
         cloudflowStreamlets,
-        cloudflowAkkastream,
-        cloudflowAkkastreamUtil,
-        cloudflowAkkastreamTestkit,
+        cloudflowAkka,
+        cloudflowAkkaUtil,
+        cloudflowAkkaTestkit,
         cloudflowSpark,
         cloudflowSparkTestkit),
     JavaUnidoc / unidoc / unidocProjectFilter := (ScalaUnidoc / unidoc / unidocProjectFilter).value)
@@ -514,10 +515,10 @@ lazy val root = Project(id = "root", base = file("."))
     cloudflowSbtPlugin,
     cloudflowRunnerConfig,
     cloudflowStreamlets,
-    cloudflowAkkastream,
-    cloudflowAkkastreamTestkit,
-    cloudflowAkkastreamUtil,
-    cloudflowAkkastreamTests,
+    cloudflowAkka,
+    cloudflowAkkaTestkit,
+    cloudflowAkkaUtil,
+    cloudflowAkkaTests,
     cloudflowFlink,
     cloudflowFlinkTestkit,
     cloudflowFlinkTests,

core/cloudflow-akka-testkit/src/main/scala/cloudflow/akkastream/testkit/TestContext.scala

@@ -173,29 +173,48 @@ private[testkit] case class TestContext(
   }
   def sinkRef[T](outlet: CodecOutlet[T]): WritableSinkRef[T] =
     new WritableSinkRef[T] {
-      def sink = {
-        val flow = Flow[(T, Committable)]
+      lazy val sink = writeSink.contramap[(T, Committable)] {
+        case (t, c) => (t, Promise.successful(t), TestCommittableOffset())
+      }
+      val writeSink: Sink[(T, Promise[T], Committable), NotUsed] = {
+        val flow = Flow[(T, Promise[T], Committable)]
         outletTaps
           .find(_.portName == outlet.name)
           .map { tap =>
             val outletTap = tap.asInstanceOf[OutletTap[T]]
             flow
-              .map { case (t, _) => outletTap.toPartitionedValue(t) }
+              .map { case (t, p, _) => outletTap.toPartitionedValue(t, p) }
               .via(killSwitch.flow)
               .mapError {
                 case cause: Throwable =>
                   execution.complete(Failure(cause))
                   cause
               }
-              .to(outletTap.sink)
+              .via(outletTap.flow)
+              .map { pv =>
+                pv.promise.trySuccess(pv.getValue)
+                pv
+              }
+              .to(Sink.ignore)
           }
           .getOrElse(
             throw TestContextException(outlet.name, s"Bad test context, could not find sink for outlet ${outlet.name}"))
       }
 
+      val valueSink: Sink[(T, Promise[T]), NotUsed] = writeSink.contramap[(T, Promise[T])] {
+        case (t, p) => (t, p, TestCommittableOffset())
+      }
+
+      val runnableGraph: RunnableGraph[Sink[(T, Promise[T]), NotUsed]] =
+        MergeHub.source[(T, Promise[T])].to(valueSink)
+      val hubSink = runnableGraph.run()
+      implicit val ec = system.dispatcher
       def write(value: T): Future[T] = {
-        Source.single(value).runWith(sink.contramap[T](t => (t, TestCommittableOffset())))
-        Future.successful(value)
+        val p = Promise[T]
+        Source
+          .single((value, p))
+          .runWith(hubSink)
+        p.future
       }
     }
 

core/cloudflow-akka-testkit/src/main/scala/cloudflow/akkastream/testkit/javadsl/AkkaStreamletTestKit.scala

@@ -36,7 +36,7 @@ object AkkaStreamletTestKit {
 }
 
 /**
- * Java testkit for testing akkastreams streamlets.
+ * Java testkit for testing akka streamlets.
  *
  * API:
  *

core/cloudflow-akka-testkit/src/main/scala/cloudflow/akkastream/testkit/javadsl/OutletTap.scala

@@ -31,25 +31,28 @@ case class Failed(e: Throwable)
 
 case class SinkOutletTap[T](outlet: CodecOutlet[T], val snk: akka.stream.javadsl.Sink[Pair[String, T], NotUsed])
     extends OutletTap[T] {
-  private[testkit] val sink: Sink[PartitionedValue[T], Future[Done]] =
+  private[testkit] val flow: Flow[PartitionedValue[T], PartitionedValue[T], NotUsed] =
     Flow[PartitionedValue[T]]
       .alsoTo(
         Flow[PartitionedValue[T]]
           .map(pv => Pair(pv.key, pv.value))
           .to(snk))
-      .toMat(Sink.ignore)(Keep.right)
+
+  private[testkit] val sink: Sink[PartitionedValue[T], Future[Done]] =
+    flow.toMat(Sink.ignore)(Keep.right)
 }
 
 case class ProbeOutletTap[T](outlet: CodecOutlet[T])(implicit system: ActorSystem) extends OutletTap[T] {
   val probe = new JTestKit(system)
-
-  // This will emit akka.japi.Pair elements to the test actor (partitioning key -> data)
-  // for easy usage in Java-based tests
-  private[cloudflow] val sink: Sink[PartitionedValue[T], Future[Done]] =
+  private[testkit] val flow: Flow[PartitionedValue[T], PartitionedValue[T], NotUsed] =
     Flow[PartitionedValue[T]]
       .alsoTo(
         Flow[PartitionedValue[T]]
           .map(pv => Pair(pv.key, pv.value))
           .to(Sink.actorRef[Pair[String, T]](probe.getTestActor, Completed, Failed)))
-      .toMat(Sink.ignore)(Keep.right)
+
+  // This will emit akka.japi.Pair elements to the test actor (partitioning key -> data)
+  // for easy usage in Java-based tests
+  private[cloudflow] val sink: Sink[PartitionedValue[T], Future[Done]] =
+    flow.toMat(Sink.ignore)(Keep.right)
 }

core/cloudflow-akka-testkit/src/main/scala/cloudflow/akkastream/testkit/package.scala

@@ -17,7 +17,8 @@
 package cloudflow.akkastream
 
 package testkit {
-  import scala.concurrent.Future
+  import scala.util.Try
+  import scala.concurrent.{ Future, Promise }
   import akka.{ Done, NotUsed }
   import akka.stream.scaladsl._
   import cloudflow.streamlets.CodecOutlet
@@ -33,18 +34,22 @@ package testkit {
   trait OutletTap[T] {
     def outlet: CodecOutlet[T]
     def portName: String = outlet.name
-
+    private[testkit] def flow: Flow[PartitionedValue[T], PartitionedValue[T], NotUsed]
     // This is for internal usage so using a scaladsl Source is no problem
     private[testkit] def sink: Sink[PartitionedValue[T], Future[Done]]
 
-    private[testkit] def toPartitionedValue(element: T): PartitionedValue[T] =
-      PartitionedValue(outlet.partitioner(element), element)
+    private[testkit] def toPartitionedValue(element: T): PartitionedValue[T] = {
+      PartitionedValue(outlet.partitioner(element), element, Promise.successful(element))
+    }
+
+    private[testkit] def toPartitionedValue(element: T, promise: Promise[T]): PartitionedValue[T] =
+      PartitionedValue(outlet.partitioner(element), element, promise)
   }
 
   /**
    * A representation of a key-value pair that is not bound to the Scala or Java DSLs
    */
-  case class PartitionedValue[T](key: String, value: T) {
+  private[testkit] case class PartitionedValue[T](key: String, value: T, promise: Promise[T]) {
     def getKey(): String = key
     def getValue(): T = value
   }

core/cloudflow-akka-testkit/src/main/scala/cloudflow/akkastream/testkit/scaladsl/OutletTap.scala

@@ -29,25 +29,29 @@ import cloudflow.akkastream.testkit.PartitionedValue
 case class Failed(e: Throwable)
 
 case class SinkOutletTap[T](outlet: CodecOutlet[T], val snk: Sink[(String, T), NotUsed]) extends OutletTap[T] {
-  private[testkit] val sink: Sink[PartitionedValue[T], Future[Done]] =
+  private[testkit] val flow: Flow[PartitionedValue[T], PartitionedValue[T], NotUsed] =
     Flow[PartitionedValue[T]]
       .alsoTo(
         Flow[PartitionedValue[T]]
           .map(pv => (pv.key, pv.value))
           .to(snk))
-      .toMat(Sink.ignore)(Keep.right)
+
+  private[testkit] val sink: Sink[PartitionedValue[T], Future[Done]] =
+    flow.toMat(Sink.ignore)(Keep.right)
 }
 
 case class ProbeOutletTap[T](outlet: CodecOutlet[T])(implicit system: ActorSystem) extends OutletTap[T] {
   val probe = new TestKit(system)
 
-  // This will emit Tuple2 elements to the test actor (partitioning key -> data)
-  // for easy usage in Scala-based tests
-  private[testkit] val sink: Sink[PartitionedValue[T], Future[Done]] =
+  private[testkit] val flow: Flow[PartitionedValue[T], PartitionedValue[T], NotUsed] =
     Flow[PartitionedValue[T]]
       .alsoTo(
         Flow[PartitionedValue[T]]
           .map(pv => (pv.key, pv.value))
           .to(Sink.actorRef[Tuple2[String, T]](probe.testActor, Completed, Failed)))
-      .toMat(Sink.ignore)(Keep.right)
+
+  // This will emit Tuple2 elements to the test actor (partitioning key -> data)
+  // for easy usage in Scala-based tests
+  private[testkit] val sink: Sink[PartitionedValue[T], Future[Done]] =
+    flow.toMat(Sink.ignore)(Keep.right)
 }

core/cloudflow-akka-testkit/src/test/avro/TestData.avsc

@@ -0,0 +1,13 @@
+{
+    "namespace": "cloudflow.akkastream.testdata",
+    "type": "record",
+    "name": "TestData",
+    "fields":[
+         {
+            "name": "id", "type": "int"
+         },
+         {
+            "name": "name", "type": "string"
+         }
+    ]
+}

core/cloudflow-akka-testkit/src/test/scala/cloudflow/akkastream/testkit/TestSinkRefSpec.scala

@@ -0,0 +1,97 @@
+/*
+ * Copyright (C) 2016-2020 Lightbend Inc. <https://www.lightbend.com>
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package cloudflow.akkastream.testkit
+
+import akka.actor.ActorSystem
+import akka.stream.scaladsl.{ RunnableGraph, Source }
+import akka.testkit.TestKit
+import cloudflow.akkastream.{ AkkaStreamlet, AkkaStreamletLogic }
+import cloudflow.akkastream.scaladsl.RunnableGraphStreamletLogic
+import cloudflow.streamlets.StreamletShape
+import cloudflow.akkastream.testdata.TestData
+import cloudflow.akkastream.testkit.scaladsl._
+import cloudflow.streamlets.avro.{ AvroInlet, AvroOutlet }
+import org.scalatest._
+import org.scalatest.wordspec.AnyWordSpec
+import org.scalatest.matchers.must.Matchers
+import scala.concurrent.Future
+
+class TestSinkRefSpec extends AnyWordSpec with Matchers with BeforeAndAfterAll {
+  private implicit val system = ActorSystem("CloudflowAkkaTestkitErrorReproducerSpec")
+
+  override def afterAll: Unit =
+    TestKit.shutdownActorSystem(system)
+
+  object TestFixture {
+    val msgs = List.tabulate(10)(i => TestData(i, i.toString))
+
+    class TestStreamlet extends AkkaStreamlet {
+      val in = AvroInlet[TestData]("in")
+      val out = AvroOutlet[TestData]("out")
+
+      override val shape: StreamletShape = StreamletShape(in).withOutlets(out)
+
+      override protected def createLogic(): AkkaStreamletLogic = new RunnableGraphStreamletLogic() {
+        override def runnableGraph(): RunnableGraph[_] = {
+          val snk = sinkRef(out)
+          sourceWithCommittableContext(in)
+            .mapAsync(parallelism = 1) { element =>
+              snk.write(element)
+            }
+            .to(committableSink)
+        }
+      }
+    }
+  }
+
+  import TestFixture._
+
+  "Cloudflow Akka TestKit" should {
+    "emit a dedicated completed messages after each message emitted via sinkRef.write, but should not" in {
+      val testkit = AkkaStreamletTestKit(system)
+      val s = new TestStreamlet()
+      val in = testkit.inletFromSource(s.in, Source(msgs))
+      val out = testkit.outletAsTap(s.out)
+
+      testkit.run(s, List(in), List(out), () => {
+        val results = out.probe.receiveN(msgs.size)
+
+        val resultWithoutIndex = results.asInstanceOf[Seq[(_, TestData)]].map(_._2)
+        resultWithoutIndex must contain allElementsOf msgs
+      })
+    }
+
+    (0 until 300).foreach { i =>
+      s"maintain the order in which messages are emitted via sinkRef.write (run #$i), but should not" in {
+        val testkit = AkkaStreamletTestKit(system)
+        val s = new TestStreamlet()
+        val in = testkit.inletFromSource(s.in, Source(msgs))
+        val out = testkit.outletAsTap(s.out)
+
+        testkit.run(s, List(in), List(out), () => {
+          val got = out.probe
+            .receiveN(msgs.size)
+            .map {
+              case (_, m: TestData) => m
+            }
+          got mustEqual msgs
+        })
+      }
+    }
+  }
+
+}

core/cloudflow-akka-util/src/test/scala/cloudflow/akkastream/util/scaladsl/HttpServerSpec.scala

@@ -83,7 +83,6 @@ class HttpServerSpec
       response.status mustEqual StatusCodes.Accepted
       response.entity.discardBytes()
       out.probe.expectMsg(("1", data))
-      out.probe.expectMsg(Completed)
     }
 
     "reject POST requests without an entity when using the default route" in {
@@ -102,7 +101,6 @@ class HttpServerSpec
       response.entity.discardBytes()
       response.status mustEqual StatusCodes.OK
       out.probe.expectMsg(("42", data))
-      out.probe.expectMsg(Completed)
 
       val badData = Data(1, "a")
       val badRequest = Put(s"http://127.0.0.1:$port", badData)