And just for kicks, here's my custom version of fromSinkAndSource that
propagates stream termination. It's cobbled together from Akka code:
def fromSinkAndSource[I, O](sink: Graph[SinkShape[I], _], source:
Graph[SourceShape[O], _]): Flow[I, O, Unit] = {
val p1 = Promise[Unit]()
val p2 = Promise[Unit]()
val tw1 = tw[I](p1.complete, p2)
val tw2 = tw[O](p2.complete, p1)
tw1.via(Flow.fromSinkAndSource(sink, source)).via(tw2)
}
abstract class KillableGraphStageLogic(val terminationSignal: Future[Unit],
_shape: Shape) extends GraphStageLogic(_shape) {
override def preStart(): Unit = {
terminationSignal.value match {
case Some(status) ⇒ onSwitch(status)
case _ ⇒
terminationSignal.onComplete(getAsyncCallback[Try[Unit]](onSwitch).invoke)
}
}
private def onSwitch(mode: Try[Unit]): Unit = mode match {
case Success(_) ⇒ completeStage()
case Failure(ex) ⇒ failStage(ex)
}
}
case class TerminationWatcher(callback: Try[Unit] ⇒ Any, promise:
Promise[Unit]) extends GraphStageWithMaterializedValue[FlowShape[Any, Any],
Unit] {
val in = Inlet[Any]("terminationWatcher.in")
val out = Outlet[Any]("terminationWatcher.out")
override val shape = FlowShape(in, out)
override def createLogicAndMaterializedValue(attr: Attributes) = {
val logic = new KillableGraphStageLogic(promise.future, shape) {
setHandler(in, new InHandler {
override def onPush(): Unit = push(out, grab(in))
override def onUpstreamFinish(): Unit = {
callback(Success[Unit](()))
completeStage()
}
override def onUpstreamFailure(ex: Throwable): Unit = {
callback(Failure(ex))
failStage(ex)
}
})
setHandler(out, new OutHandler {
override def onPull(): Unit = pull(in)
override def onDownstreamFinish(): Unit = {
callback(Success[Unit](()))
completeStage()
}
})
}
(logic, ())
}
override def toString = "TerminationWatcher"
}
def terminationWatcher[T](callback: Try[Unit] => Any, promise: Promise[Unit]):
GraphStageWithMaterializedValue[FlowShape[T, T], Unit] =
TerminationWatcher(callback,
promise).asInstanceOf[GraphStageWithMaterializedValue[FlowShape[T, T], Unit]]
def tw[T](callback: Try[Unit] => Any, promise: Promise[Unit]) =
Flow.fromGraph(terminationWatcher[T](callback, promise))
--
>>>>>>>>>> Read the docs: http://akka.io/docs/
>>>>>>>>>> Check the FAQ:
>>>>>>>>>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user
---
You received this message because you are subscribed to the Google Groups "Akka
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
