From 30e0a8c249a2ff2fc451b33f1837b16d0fa9d419 Mon Sep 17 00:00:00 2001
From: Matthias Andreas Benkard <code@mail.matthias.benkard.de>
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