raboof/pekko
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions 10

=str #23953 cleanup ConcurrentAsyncCallback implementation

Browse source 
The existing implementation had lots of duplication. The Initializing state
could be removed in favor of using `Pending(Nil)` instead which simplified
the state handling logic further. Also, instead of using `OptionVal[Promise[Done]]`
a special marker value, `NoPromise`, was introduced to mark the case when
no promise was supplied.

Fixes #23953.
This commit is contained in:
Johannes Rudolph 2017-12-06 11:32:08 +01:00
parent 6743d90e52
commit a98887140c
No known key found for this signature in database
GPG key ID: 52AF1C9ABD77E6E5
3 changed files with 54 additions and 78 deletions
Download patch file Download diff file Expand all files Collapse all files

14
akka-stream/src/main/scala/akka/stream/impl/fusing/GraphInterpreter.scala
View file

@ -194,7 +194,7 @@ import scala.util.control.NonFatal
val log: LoggingAdapter,
val logics: Array[GraphStageLogic], // Array of stage logics
val connections: Array[GraphInterpreter.Connection],
val onAsyncInput: (GraphStageLogic, Any, OptionVal[Promise[Done]], (Any) ⇒ Unit) ⇒ Unit,
val onAsyncInput: (GraphStageLogic, Any, Promise[Done], (Any) ⇒ Unit) ⇒ Unit,
val fuzzingMode: Boolean,
val context: ActorRef) {
@ -435,7 +435,7 @@ import scala.util.control.NonFatal
eventsRemaining
}
def runAsyncInput(logic: GraphStageLogic, evt: Any, promise: OptionVal[Promise[Done]], handler: (Any) ⇒ Unit): Unit =
def runAsyncInput(logic: GraphStageLogic, evt: Any, promise: Promise[Done], handler: (Any) ⇒ Unit): Unit =
if (!isStageCompleted(logic)) {
if (GraphInterpreter.Debug) println(s"$Name ASYNC $evt ($handler) [$logic]")
val currentInterpreterHolder = _currentInterpreter.get()
@ -445,16 +445,14 @@ import scala.util.control.NonFatal
activeStage = logic
try {
handler(evt)
if (promise.isDefined) {
val p = promise.get
p.success(Done)
if (promise ne GraphStageLogic.NoPromise) {
promise.success(Done)
logic.onFeedbackDispatched()
}
} catch {
case NonFatal(ex) ⇒
if (promise.isDefined) {
val p = promise.get
promise.get.failure(ex)
if (promise ne GraphStageLogic.NoPromise) {
promise.failure(ex)
logic.onFeedbackDispatched()
}
logic.failStage(ex)