+ *
+ * | Returned Flow |
+ * Sink (in) |
+ * Source (out) |
+ *
+ *
+ * | cause: upstream (sink-side) receives completion |
+ * effect: receives completion |
+ * effect: receives cancel |
+ *
+ *
+ * | cause: upstream (sink-side) receives error |
+ * effect: receives error |
+ * effect: receives cancel |
+ *
+ *
+ * | cause: downstream (source-side) receives cancel |
+ * effect: completes |
+ * effect: receives cancel |
+ *
+ *
+ * | effect: cancels upstream, completes downstream |
+ * effect: completes |
+ * cause: signals complete |
+ *
+ *
+ * | effect: cancels upstream, errors downstream |
+ * effect: receives error |
+ * cause: signals error or throws |
+ *
+ *
+ * | effect: cancels upstream, completes downstream |
+ * cause: cancels |
+ * effect: receives cancel |
+ *
+ *
+ *
+ */
+ def fromSinkAndSourceCoupled[I, O](sink: Graph[SinkShape[I], _], source: Graph[SourceShape[O], _]): Flow[I, O, NotUsed] =
+ new Flow(scaladsl.Flow.fromSinkAndSourceCoupled(sink, source))
+
+ /**
+ * Allows coupling termination (cancellation, completion, erroring) of Sinks and Sources while creating a Flow from them.
+ * Similar to [[Flow.fromSinkAndSource]] however couples the termination of these two stages.
+ *
+ * E.g. if the emitted [[Flow]] gets a cancellation, the [[Source]] of course is cancelled,
+ * however the Sink will also be completed. The table on [[Flow.fromSinkAndSourceCoupled]]
+ * illustrates the effects in detail.
+ *
+ * The `combine` function is used to compose the materialized values of the `sink` and `source`
+ * into the materialized value of the resulting [[Flow]].
+ */
+ def fromSinkAndSourceCoupledMat[I, O, M1, M2, M](
+ sink: Graph[SinkShape[I], M1], source: Graph[SourceShape[O], M2],
+ combine: function.Function2[M1, M2, M]): Flow[I, O, M] =
+ new Flow(scaladsl.Flow.fromSinkAndSourceCoupledMat(sink, source)(combinerToScala(combine)))
}
/** Create a `Flow` which can process elements of type `T`. */
diff --git a/akka-stream/src/main/scala/akka/stream/scaladsl/CoupledTerminationFlow.scala b/akka-stream/src/main/scala/akka/stream/scaladsl/CoupledTerminationFlow.scala
index 60236ad3ac..d4fe7cf6c9 100644
--- a/akka-stream/src/main/scala/akka/stream/scaladsl/CoupledTerminationFlow.scala
+++ b/akka-stream/src/main/scala/akka/stream/scaladsl/CoupledTerminationFlow.scala
@@ -59,15 +59,9 @@ object CoupledTerminationFlow {
*
* The order in which the `in` and `out` sides receive their respective completion signals is not defined, do not rely on its ordering.
*/
+ @deprecated("Use `Flow.fromSinkAndSourceCoupledMat(..., ...)(Keep.both)` instead", "2.5.2")
def fromSinkAndSource[I, O, M1, M2](in: Sink[I, M1], out: Source[O, M2]): Flow[I, O, (M1, M2)] =
- // format: OFF
- Flow.fromGraph(GraphDSL.create(in, out)(Keep.both) { implicit b => (i, o) =>
- import GraphDSL.Implicits._
- val bidi = b.add(new CoupledTerminationBidi[I, O])
- /* bidi.in1 ~> */ bidi.out1 ~> i; o ~> bidi.in2 /* ~> bidi.out2 */
- FlowShape(bidi.in1, bidi.out2)
- })
- // format: ON
+ Flow.fromSinkAndSourceCoupledMat(in, out)(Keep.both)
}
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 56225be88a..7e17fef307 100644
--- a/akka-stream/src/main/scala/akka/stream/scaladsl/Flow.scala
+++ b/akka-stream/src/main/scala/akka/stream/scaladsl/Flow.scala
@@ -315,6 +315,11 @@ object Flow {
/**
* Creates a `Flow` from a `Sink` and a `Source` where the Flow's input
* will be sent to the Sink and the Flow's output will come from the Source.
+ *
+ * The completion of the Sink and Source sides of a Flow constructed using
+ * this method are independent. So if the Sink receives a completion signal,
+ * the Source side will remain unaware of that. If you are looking to couple
+ * the termination signals of the two sides use `Flow.fromSinkAndSourceCoupled` instead.
*/
def fromSinkAndSource[I, O](sink: Graph[SinkShape[I], _], source: Graph[SourceShape[O], _]): Flow[I, O, NotUsed] =
fromSinkAndSourceMat(sink, source)(Keep.none)
@@ -323,11 +328,86 @@ object Flow {
* Creates a `Flow` from a `Sink` and a `Source` where the Flow's input
* will be sent to the Sink and the Flow's output will come from the Source.
*
+ * The completion of the Sink and Source sides of a Flow constructed using
+ * this method are independent. So if the Sink receives a completion signal,
+ * the Source side will remain unaware of that. If you are looking to couple
+ * the termination signals of the two sides use `Flow.fromSinkAndSourceCoupledMat` instead.
+ *
* The `combine` function is used to compose the materialized values of the `sink` and `source`
* into the materialized value of the resulting [[Flow]].
*/
def fromSinkAndSourceMat[I, O, M1, M2, M](sink: Graph[SinkShape[I], M1], source: Graph[SourceShape[O], M2])(combine: (M1, M2) ⇒ M): Flow[I, O, M] =
fromGraph(GraphDSL.create(sink, source)(combine) { implicit b ⇒ (in, out) ⇒ FlowShape(in.in, out.out) })
+
+ /**
+ * Allows coupling termination (cancellation, completion, erroring) of Sinks and Sources while creating a Flow from them.
+ * Similar to [[Flow.fromSinkAndSource]] however couples the termination of these two stages.
+ *
+ * E.g. if the emitted [[Flow]] gets a cancellation, the [[Source]] of course is cancelled,
+ * however the Sink will also be completed. The table below illustrates the effects in detail:
+ *
+ *
+ *
+ * | Returned Flow |
+ * Sink (in) |
+ * Source (out) |
+ *
+ *
+ * | cause: upstream (sink-side) receives completion |
+ * effect: receives completion |
+ * effect: receives cancel |
+ *
+ *
+ * | cause: upstream (sink-side) receives error |
+ * effect: receives error |
+ * effect: receives cancel |
+ *
+ *
+ * | cause: downstream (source-side) receives cancel |
+ * effect: completes |
+ * effect: receives cancel |
+ *
+ *
+ * | effect: cancels upstream, completes downstream |
+ * effect: completes |
+ * cause: signals complete |
+ *
+ *
+ * | effect: cancels upstream, errors downstream |
+ * effect: receives error |
+ * cause: signals error or throws |
+ *
+ *
+ * | effect: cancels upstream, completes downstream |
+ * cause: cancels |
+ * effect: receives cancel |
+ *
+ *
+ *
+ */
+ def fromSinkAndSourceCoupled[I, O](sink: Graph[SinkShape[I], _], source: Graph[SourceShape[O], _]): Flow[I, O, NotUsed] =
+ fromSinkAndSourceCoupledMat(sink, source)(Keep.none)
+
+ /**
+ * Allows coupling termination (cancellation, completion, erroring) of Sinks and Sources while creating a Flow from them.
+ * Similar to [[Flow.fromSinkAndSource]] however couples the termination of these two stages.
+ *
+ * E.g. if the emitted [[Flow]] gets a cancellation, the [[Source]] of course is cancelled,
+ * however the Sink will also be completed. The table on [[Flow.fromSinkAndSourceCoupled]]
+ * illustrates the effects in detail.
+ *
+ * The `combine` function is used to compose the materialized values of the `sink` and `source`
+ * into the materialized value of the resulting [[Flow]].
+ */
+ def fromSinkAndSourceCoupledMat[I, O, M1, M2, M](sink: Graph[SinkShape[I], M1], source: Graph[SourceShape[O], M2])(combine: (M1, M2) ⇒ M): Flow[I, O, M] =
+ // format: OFF
+ Flow.fromGraph(GraphDSL.create(sink, source)(combine) { implicit b => (i, o) =>
+ import GraphDSL.Implicits._
+ val bidi = b.add(new CoupledTerminationBidi[I, O])
+ /* bidi.in1 ~> */ bidi.out1 ~> i; o ~> bidi.in2 /* ~> bidi.out2 */
+ FlowShape(bidi.in1, bidi.out2)
+ })
+ // format: ON
}
object RunnableGraph {