Rewrite to use async queue consumers (#2)

* Moved from Effect to an internal Aff queue

Author: robertdp

package.json

@@ -4,7 +4,8 @@
   "description": "",
   "main": "index.js",
   "scripts": {
-    "build": "spago build"
+    "build": "spago build",
+    "test": "spago test"
   },
   "author": "",
   "license": "ISC",

spago.dhall

@@ -5,11 +5,12 @@ You can edit this file as you like.
 { name = "wire"
 , dependencies =
   [ "arrays"
-  , "effect"
+  , "avar"
+  , "console"
   , "filterable"
-  , "js-timers"
   , "profunctor"
   , "refs"
+  , "strings"
   , "unsafe-reference"
   ]
 , packages = ./packages.dhall

src/Wire/Event.purs

@@ -1,31 +1,37 @@
 module Wire.Event where
 
 import Prelude
-import Control.Alt (class Alt)
-import Control.Alternative (class Alternative, class Plus)
+import Control.Alt (class Alt, alt)
+import Control.Alternative (class Alternative, class Plus, empty)
 import Control.Apply (lift2)
+import Control.Monad.Rec.Class (Step(..), tailRecM)
 import Data.Array as Array
-import Data.Either (either, hush)
+import Data.Either (Either(..), either, hush)
 import Data.Filterable (class Compactable, class Filterable, filterMap, partitionMap)
-import Data.Foldable (class Foldable, sequence_, traverse_)
+import Data.Foldable (class Foldable, for_, sequence_, traverse_)
 import Data.Maybe (Maybe(..), fromJust, isJust)
 import Effect (Effect)
+import Effect.Aff (Milliseconds(..))
+import Effect.Aff as Aff
+import Effect.Class (liftEffect)
 import Effect.Ref as Ref
 import Partial.Unsafe (unsafePartial)
 import Unsafe.Reference (unsafeRefEq)
+import Wire.Event.Queue as Queue
 
 newtype Event a
-  = Event (Subscribe a)
+  = Event (Subscriber a -> Effect Canceller)
 
-type Subscribe a
-  = (a -> Effect Unit) -> Effect Canceler
+type Subscriber a
+  = a -> Effect Unit
 
-type Canceler
+type Canceller
   = Effect Unit
 
-create :: forall a. Effect { event :: Event a, push :: a -> Effect Unit }
+create :: forall a. Effect { event :: Event a, push :: a -> Effect Unit, cancel :: Effect Unit }
 create = do
   subscribers <- Ref.new []
+  queue <- Queue.create \a -> Ref.read subscribers >>= traverse_ \k -> k a
   let
     event =
       Event \emit -> do
@@ -36,14 +42,12 @@ create = do
         pure do
           Ref.write true unsubscribing
           Ref.modify_ (Array.deleteBy unsafeRefEq subscriber) subscribers
+  pure { event, push: (queue.push <<< pure), cancel: queue.kill }
 
-    push a = Ref.read subscribers >>= traverse_ \emit -> emit a
-  pure { event, push }
-
-makeEvent :: forall a. Subscribe a -> Event a
+makeEvent :: forall a. (Subscriber a -> Effect Canceller) -> Event a
 makeEvent = Event
 
-subscribe :: forall a. Event a -> Subscribe a
+subscribe :: forall a. Event a -> Subscriber a -> Effect Canceller
 subscribe (Event event) = event
 
 filter :: forall a. (a -> Boolean) -> Event a -> Event a
@@ -62,16 +66,16 @@ share source = do
   shared <- create
   let
     incrementCount = do
-      count <- Ref.modify (_ + 1) subscriberCount
+      count <- liftEffect do Ref.modify (_ + 1) subscriberCount
       when (count == 1) do
-        cancel <- subscribe source shared.push
-        Ref.write (Just cancel) cancelSource
+        cancel <- subscribe source do liftEffect <<< shared.push
+        liftEffect do Ref.write (Just cancel) cancelSource
 
     decrementCount = do
-      count <- Ref.modify (_ - 1) subscriberCount
+      count <- liftEffect do Ref.modify (_ - 1) subscriberCount
       when (count == 0) do
-        Ref.read cancelSource >>= sequence_
-        Ref.write Nothing cancelSource
+        liftEffect (Ref.read cancelSource) >>= sequence_
+        liftEffect do Ref.write Nothing cancelSource
 
     event =
       Event \emit -> do
@@ -90,34 +94,72 @@ distinct (Event event) =
         Ref.write (pure a) latest
         emit a
 
-bufferUntil :: forall a b. Event b -> Event a -> Event (Array a)
-bufferUntil (Event flush) (Event event) =
-  Event \emit -> do
-    buffer <- Ref.new []
-    cancelEvent <- event \a -> Ref.modify_ (flip Array.snoc a) buffer
-    cancelFlush <- flush \_ -> Ref.modify' { state: [], value: _ } buffer >>= emit
-    pure do cancelEvent *> cancelFlush
+bufferUntil :: forall b a. Event b -> Event a -> Event (Array a)
+bufferUntil flush source =
+  alt (Nothing <$ flush) (Just <$> source)
+    # fold
+        ( \{ buffer } -> case _ of
+            Nothing -> { buffer: [], output: Just buffer }
+            Just a -> { buffer: Array.snoc buffer a, output: Nothing }
+        )
+        { buffer: [], output: Nothing }
+    # filterMap _.output
 
 fromFoldable :: forall a f. Foldable f => f a -> Event a
-fromFoldable xs = Event \emit -> traverse_ emit xs *> mempty
+fromFoldable xs =
+  Event \emit -> do
+    fiber <-
+      Aff.launchAff do
+        for_ xs \x -> do
+          liftEffect do emit x
+          Aff.delay (Milliseconds 0.0)
+    pure do
+      Aff.launchAff_ do Aff.killFiber (Aff.error "cancelled") fiber
+
+range :: Int -> Int -> Event Int
+range start end =
+  Event \emit -> do
+    let
+      go pos
+        | pos /= end = do
+          liftEffect do emit pos
+          Aff.delay (Milliseconds 0.0)
+          pure (Loop (pos + step))
+
+      go _ = do
+        liftEffect do emit end
+        pure (Done unit)
+    fiber <- Aff.launchAff do tailRecM go start
+    pure do
+      Aff.launchAff_ do Aff.killFiber (Aff.error "cancelled") fiber
+  where
+  step = if start < end then 1 else -1
+
+times :: Int -> Event Int
+times n
+  | n > 0 = range 1 n
+
+times _ = empty
 
 instance functorEvent :: Functor Event where
-  map f (Event event) = Event \emit -> event \a -> emit (f a)
+  map f (Event event) =
+    Event \emit -> do
+      queue <- Queue.create (emit <<< f)
+      cancel <- event (queue.push <<< pure)
+      pure do
+        cancel
+        queue.kill
 
 instance applyEvent :: Apply Event where
-  apply (Event eventF) (Event eventA) =
-    Event \emitB -> do
-      latestF <- Ref.new Nothing
-      latestA <- Ref.new Nothing
-      cancelF <-
-        eventF \f -> do
-          Ref.write (Just f) latestF
-          Ref.read latestA >>= traverse_ \a -> emitB (f a)
-      cancelA <-
-        eventA \a -> do
-          Ref.write (Just a) latestA
-          Ref.read latestF >>= traverse_ \f -> emitB (f a)
-      pure do cancelF *> cancelA
+  apply eventF eventA =
+    alt (Left <$> eventF) (Right <$> eventA)
+      # fold
+          ( \{ left, right } -> case _ of
+              Left l -> { left: Just l, right }
+              Right r -> { left, right: Just r }
+          )
+          { left: Nothing, right: Nothing }
+      # filterMap (\{ left, right } -> apply left right)
 
 instance applicativeEvent :: Applicative Event where
   pure a = Event \emit -> emit a *> mempty
@@ -127,10 +169,11 @@ instance bindEvent :: Bind Event where
     Event \emit -> do
       cancelInner <- Ref.new Nothing
       cancelOuter <-
-        outer \a -> do
-          Ref.read cancelInner >>= sequence_
-          cancel <- subscribe (f a) emit
-          Ref.write (Just cancel) cancelInner
+        outer \a ->
+          liftEffect do
+            Ref.read cancelInner >>= sequence_
+            cancel <- subscribe (f a) emit
+            Ref.write (Just cancel) cancelInner
       pure do
         Ref.read cancelInner >>= sequence_
         cancelOuter

src/Wire/Event/Queue.purs

@@ -0,0 +1,27 @@
+module Wire.Event.Queue where
+
+import Prelude
+import Control.Monad.Rec.Class (forever)
+import Effect (Effect)
+import Effect.AVar as AVar
+import Effect.Aff (Aff)
+import Effect.Aff as Aff
+import Effect.Aff.AVar as AffVar
+import Effect.Class (liftEffect)
+
+create :: forall a b. (a -> Effect b) -> Effect { push :: Aff a -> Effect Unit, kill :: Effect Unit }
+create consumer = do
+  queue <- AVar.empty
+  fiber <-
+    (Aff.launchAff <<< forever) do
+      aff <- AffVar.take queue
+      a <- aff
+      liftEffect do consumer a
+  let
+    killFiber = Aff.launchAff_ do Aff.killFiber (Aff.error "killing queue consumer") fiber
+
+    killQueue = AVar.kill (Aff.error "killing queue") queue
+  pure
+    { push: \a -> Aff.launchAff_ do AffVar.put a queue
+    , kill: killFiber *> killQueue
+    }

src/Wire/Event/Time.purs

@@ -2,28 +2,36 @@ module Wire.Event.Time where
 
 import Prelude
 import Control.Alt ((<|>))
-import Effect.Ref as Ref
-import Effect.Timer as Timer
+import Control.Monad.Rec.Class (forever)
+import Data.Time.Duration (class Duration, fromDuration)
+import Effect.Aff as Aff
+import Effect.Class (liftEffect)
 import Wire.Event (Event)
 import Wire.Event as Event
+import Wire.Event.Queue as Queue
 
-delay :: forall a. Int -> Event a -> Event a
-delay ms event =
+delay :: forall offset a. Duration offset => offset -> Event a -> Event a
+delay offset event = do
+  let
+    ms = fromDuration offset
   Event.makeEvent \emit -> do
-    canceled <- Ref.new false
-    cancel <-
-      Event.subscribe event \a -> do
-        _ <- Timer.setTimeout ms do unlessM (Ref.read canceled) do emit a
-        pure unit
+    queue <- Queue.create emit
+    cancel <- Event.subscribe event \a -> queue.push do Aff.delay ms *> pure a
     pure do
-      Ref.write true canceled
       cancel
+      queue.kill
 
-interval :: Int -> Event Unit
-interval ms =
+interval :: forall spacing. Duration spacing => spacing -> Event Unit
+interval spacing = do
+  let
+    ms = fromDuration spacing
   Event.makeEvent \emit -> do
-    intervalId <- Timer.setInterval ms do emit unit
-    pure do Timer.clearInterval intervalId
+    fiber <-
+      (Aff.launchAff <<< forever) do
+        Aff.delay ms
+        liftEffect do emit unit
+    pure do
+      Aff.launchAff_ do Aff.killFiber (Aff.error "cancelling") fiber
 
-timer :: Int -> Int -> Event Unit
-timer after ms = delay after do pure unit <|> interval ms
+timer :: forall offset spacing. Duration offset => Duration spacing => offset -> spacing -> Event Unit
+timer offset spacing = delay offset do pure unit <|> interval spacing

src/Wire/Signal.purs

@@ -8,9 +8,10 @@ import Wire.Event as Event
 
 type Signal a
   = { event :: Event a
+    , modify :: (a -> a) -> Effect Unit
     , read :: Effect a
     , write :: a -> Effect Unit
-    , modify :: (a -> a) -> Effect Unit
+    , cancel :: Effect Unit
     }
 
 create :: forall a. a -> Effect (Signal a)
@@ -28,4 +29,4 @@ create init = do
       Event.makeEvent \emit -> do
         Ref.read value >>= emit
         Event.subscribe inner.event emit
-  pure { event, read, write, modify }
+  pure { event, read, write, modify, cancel: inner.cancel }

test/Main.purs

@@ -0,0 +1,33 @@
+module Test.Main where
+
+import Prelude
+import Control.Alt ((<|>))
+import Data.Array as Array
+import Data.FoldableWithIndex (foldlWithIndex)
+import Data.Int as Int
+import Data.List.Lazy (range)
+import Data.String.CodeUnits as CodeUnits
+import Effect (Effect)
+import Effect.Class.Console as Console
+import Wire.Event (Event)
+import Wire.Event as Event
+
+main :: Effect Unit
+main = do
+  void $ Event.subscribe (Event.distinct (sumFromOneToOneHundred <|> sumFromOneToOneHundred) >>= pure <<< formatNumber <<< show) do Console.log
+
+sumFromOneToOneHundred :: Event Number
+sumFromOneToOneHundred =
+  range 1 100
+    # Event.fromFoldable
+    # map Int.toNumber
+    # Event.fold (+) 0.0
+
+formatNumber :: String -> String
+formatNumber =
+  CodeUnits.dropRight 2
+    >>> CodeUnits.toCharArray
+    >>> Array.reverse
+    >>> foldlWithIndex (\i o c -> if i /= 0 && i `mod` 3 == 0 then o <> [ ',', c ] else o <> [ c ]) []
+    >>> Array.reverse
+    >>> CodeUnits.fromCharArray