AVarSM.purs

type Callback eff a = Either Error a -> Eff eff Unit

type Producer eff a = Tuple a (Callback eff Unit)

data Consumer eff a
  = Reader (Callback eff a)
  | Taker (Callback eff a)

type Queue = NonEmpty Array

data AVarStatus eff a
  = Empty
  | Killed Error
  | Filled a
  | Surplus a (Queue (Producer a))
  | Deficit (Queue (Consumer a))

data AVarAction eff a
  = Take (Callback eff a)
  | Read (Callback eff a)
  | Put a (Callback eff a)
  | Kill Error

type Scheduler eff = Eff eff Unit -> Eff eff Unit

runAVar
  :: forall eff a
   . Scheduler eff
  -> AVarStatus eff a
  -> AVarAction eff a
  -> Eff eff (AVarStatus eff a)
runAVar schedule = case _, _ of
  Take cb, Empty -> do
    pure (Deficit (Taker cb :| [])
  Take cb, Filled a -> do
    schedule (cb (Right a))
    pure Empty
  Take cb, Killed e -> do
    schedule (cb (Left e))
    pure (Killed e)
  Take cb, Surplus a (Tuple b cb' :| tail) -> do
    schedule (cb (Right a))
    schedule (cb' (Right unit))
    case uncons tail of
      Just as -> pure (Surplus b (as.head :| as.tail))
      Nothing -> pure (Filled b)
  Take cb, Deficit (head :| tail) -> do
    pure (Deficit (Taker cb :| cons head tail))
  Read cb, Empty -> do
    pure (Deficit (Reader cb :| []))
  Read cb, Filled a -> do
    schedule (cb (Right a))
    pure (Filled a)
  Read cb, Killed e -> do
    schedule (cb (Left e))
    pure (Killed e)
  Read cb, Surplus a ps -> do
    schedule (cb (Right a))
    pure (Surplus a ps))
  Read cb, Deficit (c :| cs) -> do
    pure (Deficit (Reader cb :| cons c cs))
  Put a cb, Empty -> do
    schedule (cb (Right Unit))
    pure (Filled a)
  Put a cb, Killed e -> do
    schedule (cb (Left e))
    pure (Killed e)
  Put a' cb, Filled a -> do
    pure (Surplus a (Tuple a' cb))
  Put a' cb, Surplus a (p :| ps) -> do
    pure (Surplus a (Tuple a' cb :| cons p ps))
  Put a cb, Deficit (c :| cs) -> do
    Tuple cbs cs' <- filterConsumers (cons c cs)
    for_ cbs \cb -> schedule (cb (Right a))
    schedule (cb (Right unit))
    case uncons cs' of
      Just cs'' -> pure (Deficit (cs''.head :| cs''.tail))
      Nothing -> pure Empty
  Kill e, Empty -> do
    pure (Killed e)
  Kill e, Filled a -> do
    pure (Killed e)
  Kill _, Killed e -> do
    pure (Killed e)
  Kill e, Surplus (p :| ps) -> do
    for_ (cons p ps) \(Tuple _ cb) -> schedule (cb (Left e))
    pure (Killed e)
  Kill e, Deficit (c :| cs) -> do
    for_ (cons c cs) case _ of
      Reader cb -> schedule (cb (Left e))
      Taker cb -> schedule (cb (Left e))
    pure (Killed e)

filterConsumers
  :: forall eff a
   . Array (Consumer a)
  -> Tuple (Array (Callback eff a)) (Array (Consumer a))
filterConsumers = done <<< foldl go { cbs: [], taken: false, rest: [] }
  where
  go { cbs, taken, rest } = case _ of
    Reader cb -> { cbs: cons cb cbs, taken, rest }
    Taker cb | taken -> { cbs, taken, rest: snoc cb rest }
    Taker cb -> { cbs: cons cb cbs, taken: true, rest }

  done { cbs, rest } =
    Tuple cbs rest