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

=str 20843 Scan.in cannot pull closed port

Browse source
This commit is contained in:
Alexander Golubev 2016-07-05 00:00:52 -04:00 • committed by Konrad Malawski
parent 8d0c163eaf
commit f679e14168
2 changed files with 29 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

17
akka-stream/src/main/scala/akka/stream/impl/fusing/Ops.scala
View file

@ -315,7 +315,22 @@ private[akka] final case class Scan[In, Out](zero: Out, f: (Out, In) ⇒ Out) ex
setHandlers(in, out, self)
}
})
setHandler(in, totallyIgnorantInput)
setHandler(in, new InHandler {
override def onPush(): Unit = ()
override def onUpstreamFinish(): Unit = setHandler(out, new OutHandler {
override def onPull(): Unit = {
push(out, aggregator)
completeStage()
}
})
override def onUpstreamFailure(ex: Throwable): Unit = setHandler(out, new OutHandler {
override def onPull(): Unit = {
push(out, aggregator)
failStage(ex)
}
})
})
override def onPull(): Unit = pull(in)
override def onPush(): Unit = {