Box.Connectors

Various ways to connect things up.

Documentation

qList :: [a] -> CoEmitter IO a

Queue a list Unbounded.

>>> pushList <$|> qList [1,2,3]
[1,2,3]

qListWith :: Queue a -> [a] -> CoEmitter IO a

Queue a list with an explicit Queue.

>>> pushList <$|> qListWith Single [1,2,3]
[1,2,3]

popList :: Monad m => [a] -> Committer m a -> m ()

Directly supply a list to a committer action, via pop.

>>> popList [1..3] showStdout
1
2
3

pushList :: Monad m => Emitter m a -> m [a]

Push an Emitter into a list, via push.

>>> pushList <$|> qList [1..3]
[1,2,3]

pushListN :: Monad m => Int -> Emitter m a -> m [a]

Push an Emitter into a list, finitely.

>>> pushListN 2 <$|> qList [1..3]
[1,2]

sink :: Int -> (a -> IO ()) -> CoCommitter IO a

Create a finite Committer Unbounded Queue.

glue <$> sink 2 print <*|> qList [1..3]

1 2

sinkWith :: Queue a -> Int -> (a -> IO ()) -> CoCommitter IO a

Create a finite Committer Queue.

source :: Int -> IO a -> CoEmitter IO a

Create a finite (Co)Emitter Unbounded Queue.

>>> glue toStdout <$|> source 2 (pure "hi")
hi
hi

sourceWith :: Queue a -> Int -> IO a -> CoEmitter IO a

Create a finite (Co)Emitter Unbounded Queue.

>>> glue toStdout <$|> sourceWith Single 2 (pure "hi")
hi
hi

forkEmit :: Monad m => Emitter m a -> Committer m a -> Emitter m a

Glues an emitter to a committer, then resupplies the emitter.

>>> (c1,l1) <- refCommitter :: IO (Committer IO Int, IO [Int])
>>> close $ toListM <$> (forkEmit <$> (qList [1..3]) <*> pure c1)
[1,2,3]

>>> l1
[1,2,3]

bufferCommitter :: Committer IO a -> CoCommitter IO a

Buffer a committer.

bufferEmitter :: Emitter IO a -> CoEmitter IO a

Buffer an emitter.

concurrentE :: Queue a -> Emitter IO a -> Emitter IO a -> CoEmitter IO a

Concurrently run two emitters.

This differs to (<>), which is left-biased.

Note that functions such as toListM, which complete on the first Nothing emitted, will not work as expected.

>>> close $ (fmap toListM) (join $ concurrentE Single <$> qList [1..3] <*> qList [5..9])
[1,2,3]

In the code below, the ordering is non-deterministic.

(c,l) <- refCommitter :: IO (Committer IO Int, IO [Int])
close $ glue c <$> (join $ concurrentE Single <$> qList [1..30] <*> qList [40..60])

concurrentC :: Queue a -> Committer IO a -> Committer IO a -> CoCommitter IO a

Concurrently run two committers.

>>> import Data.Functor.Contravariant
>>> import Data.Text (pack)
>>> cFast = witherC (\b -> pure (Just b)) . contramap ("fast: " <>) $ toStdout
>>> cSlow = witherC (\b -> sleep 0.1 >> pure (Just b)) . contramap ("slow: " <>) $ toStdout
>>> close $ (popList ((pack . show) <$> [1..3]) <$> (concurrentC Unbounded cFast cSlow)) <> pure (sleep 1)
fast: 1
fast: 2
fast: 3
slow: 1
slow: 2
slow: 3

takeQ :: Queue a -> Int -> Emitter IO a -> CoEmitter IO a

Take and queue n emits.

>>> import Control.Monad.State.Lazy
>>> toListM <$|> (takeQ Single 4 =<< qList [0..])
[0,1,2,3]

evalEmitter :: s -> Emitter (StateT s IO) a -> CoEmitter IO a

queue a stateful emitter, supplying initial state

>>> import Control.Monad.State.Lazy
>>> toListM <$|> (evalEmitter 0 <$> takeE 4 =<< qList [0..])
[0,1,2,3]

evalEmitterWith :: Queue a -> s -> Emitter (StateT s IO) a -> CoEmitter IO a

queue a stateful emitter, supplying initial state