@@ -11,16 +11,14 @@ import Prelude
1111import Data.Maybe (Maybe(..))
1212import Data.Tuple.Nested ((/\))
1313import Effect (Effect)
14−import Effect.Aff (forkAff, killFiber)
1514import Effect.Aff as Aff
1615import Effect.Aff.AVar (AVar)
1716import Effect.Aff.AVar as AVar
1817import Effect.Aff.Class (liftAff)
1918import Effect.Class (liftEffect)
20−import Effect.Exception (error)
2119import Control.Monad.Rec.Class (Step(..))
2220import Unsafe.Coerce (unsafeCoerce)
23−import Yoga.Capnweb (RpcConnection, call0, call1, call2, callWithCallback)
21+import Yoga.Capnweb (RpcConnection, awaitCallback, call0, call1, call2, callWithCancellableCallback, cancelCallback)
2422import Yoga.Om (Om)
2523import Yoga.Om.Strom (Strom, mkStrom, bracket)
2624import Yoga.Om.Strom as Strom
@@ -31,11 +29,12 @@ subscribe method conn = bracket acquire release use
3129 acquire = liftAff do
3230 queue <- AVar.empty
3331 let cb v = Aff.launchAff_ $ AVar.put (unsafeCoerce v :: a) queue
34− fiber <- forkAff $ callWithCallback conn method cb
35− pure { queue, fiber }
32+ handle <- callWithCancellableCallback conn method cb # liftEffect
33+ pure { queue, handle }
3634
3735 release res = liftAff do
38− killFiber (error "unsubscribed") res.fiber
36+ cancelCallback res.handle # liftEffect
37+ awaitCallback res.handle
3938
4039 use res = pullLoop res.queue
4140