From 30e0a8c249a2ff2fc451b33f1837b16d0fa9d419 Mon Sep 17 00:00:00 2001 From: Matthias Andreas Benkard Date: Wed, 10 Jan 2018 22:00:02 +0100 Subject: React to monix.execution.Ack.Stop signals properly. Previously, we just ignored the return value of Observer.Sync.onNext in the implementation of Fiber.run. Now we stop when we receive a Stop response. --- core/lib/Fiber.scala | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) (limited to 'core/lib') diff --git a/core/lib/Fiber.scala b/core/lib/Fiber.scala index 42b416b..bf55cf4 100644 --- a/core/lib/Fiber.scala +++ b/core/lib/Fiber.scala @@ -1,6 +1,7 @@ package eu.mulk.fibers import monix.eval.Task +import monix.execution.Ack.{Continue, Stop} import monix.execution.cancelables.MultiAssignmentCancelable import monix.execution.{Cancelable, Scheduler} import monix.reactive.{Observable, OverflowStrategy} @@ -135,10 +136,8 @@ object Fiber extends FiberConversions with FiberTypes { var more = init var done = false cancelable := Cancelable { () ⇒ - if (!done) { - done = true - out.onComplete() - } + out.onComplete() + done = true } while (!done) { try { @@ -163,8 +162,12 @@ object Fiber extends FiberConversions with FiberTypes { state.fiberVar = value more = continue(()).asInstanceOf[(Effect[Out], Any ⇒ Any)] case Effect.Emit(returnValue) ⇒ - out.onNext(returnValue.asInstanceOf[Out]) - more = continue(()).asInstanceOf[(Effect[Out], Any ⇒ Any)] + out.onNext(returnValue.asInstanceOf[Out]) match { + case _: Continue ⇒ + more = continue(()).asInstanceOf[(Effect[Out], Any ⇒ Any)] + case _: Stop ⇒ + cancelable.cancel() + } case Effect.Fail(throwable) ⇒ out.onError(throwable) done = true -- cgit v1.2.3