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

=str 20843 Scan failure should flow downstream ASAP

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

3
akka-stream-tests/src/test/scala/akka/stream/scaladsl/FlowScanSpec.scala
View file

@ -71,10 +71,9 @@ class FlowScanSpec extends AkkaSpec {
.expectNext(0) .expectNext(0)
.expectComplete() .expectComplete()
} }
"fail after emitting first element when upsrteam failed" in { "fail when upstream failed" in {
Source.failed[Int](TE("")).scan(0) { case (a, b) ⇒ a + b }.runWith(TestSink.probe[Int]) Source.failed[Int](TE("")).scan(0) { case (a, b) ⇒ a + b }.runWith(TestSink.probe[Int])
.request(2) .request(2)
.expectNext(0)
.expectError(TE("")) .expectError(TE(""))
} }
} }

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

@ -324,12 +324,6 @@ private[akka] final case class Scan[In, Out](zero: Out, f: (Out, In) ⇒ Out) ex
completeStage() 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 onPull(): Unit = pull(in)