Copyright | Ivan Lazar Miljenovic |
---|---|
License | MIT |
Maintainer | Ivan.Miljenovic@gmail.com |
Safe Haskell | None |
Language | Haskell2010 |
Consider a physical desk for someone that has to deal with correspondence.
A typical system is to have two baskets/trays: one for incoming papers that still needs to be processed, and another for outgoing papers that have already been processed.
We use this metaphor for dealing with Buffer
s: data is fed into
one using the InBasket
(until the buffer indicates that it has
had enough) and taken out from the OutBasket
.
- data Buffer a
- unbounded :: Buffer a
- bounded :: Int -> Buffer a
- latest :: a -> Buffer a
- newest :: Int -> Buffer a
- withBuffer :: (MonadMask m, MonadBaseControl IO m) => Buffer a -> (InBasket a -> m i) -> (OutBasket a -> m r) -> m r
- withBufferedTransform :: (MonadMask m, MonadBaseControl IO m) => Int -> (OutBasket a -> InBasket b -> m ab) -> (InBasket a -> m i) -> (OutBasket b -> m r) -> m r
- newtype InBasket a = InBasket {}
- newtype OutBasket a = OutBasket {
- receiveMsg :: STM (Maybe a)
- writeStreamBasket :: MonadBase IO m => Stream (Of a) m r -> InBasket a -> m ()
- withStreamBasket :: MonadBase IO m => OutBasket a -> (Stream (Of a) m () -> r) -> r
- withMergedStreams :: (MonadMask m, MonadBaseControl IO m, MonadBase IO n, Foldable t) => Buffer a -> t (Stream (Of a) m v) -> (Stream (Of a) n () -> m r) -> m r
- withStreamMap :: (MonadMask m, MonadBaseControl IO m, MonadBase IO n) => Int -> (a -> b) -> Stream (Of a) m i -> (Stream (Of b) n () -> m r) -> m r
- withStreamMapM :: (MonadMask m, MonadBaseControl IO m, MonadBase IO n) => Int -> (a -> m b) -> Stream (Of a) m i -> (Stream (Of b) n () -> m r) -> m r
- withStreamTransform :: (MonadMask m, MonadBaseControl IO m, MonadBase IO n) => Int -> (Stream (Of a) m () -> Stream (Of b) m t) -> Stream (Of a) m i -> (Stream (Of b) n () -> m r) -> m r
- writeByteStringBasket :: MonadBase IO m => ByteString m r -> InBasket ByteString -> m ()
- withByteStringBasket :: MonadBase IO m => OutBasket ByteString -> (ByteString m () -> r) -> r
- withMergedByteStrings :: (MonadMask m, MonadBaseControl IO m, MonadBase IO n, Foldable t) => Buffer ByteString -> t (ByteString m v) -> (ByteString n () -> m r) -> m r
Buffers
bounded :: Int -> Buffer a Source #
Store a bounded number of messages, specified by the Int
argument.
A buffer size <= 0
will result in a permanently empty buffer,
which could result in a system that hang.
latest :: a -> Buffer a Source #
Only store the "latest" message, beginning with an initial value.
This buffer is never empty nor full; as such, it is up to the
caller to ensure they only take as many values as they need
(e.g. using
as the final
parameter to print
. readStreamBasket
withBuffer
will -- after all other values are
processed -- keep printing the last value over and over again).
Using a buffer
withBuffer :: (MonadMask m, MonadBaseControl IO m) => Buffer a -> (InBasket a -> m i) -> (OutBasket a -> m r) -> m r Source #
Use a buffer to asynchronously communicate.
Two functions are taken as parameters:
- How to provide input to the buffer (the result of this is discarded)
- How to take values from the buffer
As soon as one function indicates that it is complete then the other is terminated. This is safe: trying to write data to a closed buffer will not achieve anything.
However, reading a buffer that has not indicated that it is closed (e.g. waiting on an action to complete to be able to provide the next value) but contains no values will block.
withBufferedTransform Source #
:: (MonadMask m, MonadBaseControl IO m) | |
=> Int | How many concurrent computations to run. |
-> (OutBasket a -> InBasket b -> m ab) | What to do with each individual concurrent computation; result is ignored. |
-> (InBasket a -> m i) | Provide initial data; result is ignored. |
-> (OutBasket b -> m r) | |
-> m r |
Use buffers to concurrently transform the provided data.
In essence, this is a demultiplexer -> multiplexer
transformation: the incoming data is split into n
individual
segments, the results of which are then merged back together
again.
Note: ordering of elements in the output is undeterministic.
Since: 0.2.0.0
An exhaustible source of values.
receiveMsg
returns Nothing
if the source is exhausted.
OutBasket | |
|
Stream support
writeStreamBasket :: MonadBase IO m => Stream (Of a) m r -> InBasket a -> m () Source #
Write a single stream to a buffer.
Type written to make it easier if this is the only stream being written to the buffer.
withStreamBasket :: MonadBase IO m => OutBasket a -> (Stream (Of a) m () -> r) -> r Source #
Read the output of a buffer into a stream.
Since: 0.2.0.0
withMergedStreams :: (MonadMask m, MonadBaseControl IO m, MonadBase IO n, Foldable t) => Buffer a -> t (Stream (Of a) m v) -> (Stream (Of a) n () -> m r) -> m r Source #
Concurrently merge multiple streams together.
The resulting order is unspecified.
Note that the monad of the resultant Stream can be different from the final result.
Since: 0.2.0.0
Mapping
:: (MonadMask m, MonadBaseControl IO m, MonadBase IO n) | |
=> Int | How many concurrent computations to run. |
-> (a -> b) | |
-> Stream (Of a) m i | |
-> (Stream (Of b) n () -> m r) | |
-> m r |
Concurrently map a function over all elements of a Stream
.
Note: ordering of elements in the output is undeterministic.
Since: 0.2.0.0
:: (MonadMask m, MonadBaseControl IO m, MonadBase IO n) | |
=> Int | How many concurrent computations to run. |
-> (a -> m b) | |
-> Stream (Of a) m i | |
-> (Stream (Of b) n () -> m r) | |
-> m r |
Concurrently map a monadic function over all elements of a
Stream
.
Note: ordering of elements in the output is undeterministic.
Since: 0.2.0.0
:: (MonadMask m, MonadBaseControl IO m, MonadBase IO n) | |
=> Int | How many concurrent computations to run. |
-> (Stream (Of a) m () -> Stream (Of b) m t) | |
-> Stream (Of a) m i | |
-> (Stream (Of b) n () -> m r) | |
-> m r |
Concurrently split the provided stream into n
streams and
transform them all using the provided function.
Note: ordering of elements in the output is undeterministic.
Since: 0.2.0.0
ByteString support
writeByteStringBasket :: MonadBase IO m => ByteString m r -> InBasket ByteString -> m () Source #
A streaming ByteString
variant of writeStreamBasket
.
withByteStringBasket :: MonadBase IO m => OutBasket ByteString -> (ByteString m () -> r) -> r Source #
A streaming ByteString
variant of withStreamBasket
.
Since: 0.2.0.0
withMergedByteStrings :: (MonadMask m, MonadBaseControl IO m, MonadBase IO n, Foldable t) => Buffer ByteString -> t (ByteString m v) -> (ByteString n () -> m r) -> m r Source #
A streaming ByteString
variant of withMergedStreams
.
Since: 0.2.0.0
No ByteString
equivalents of withStreamMap
, etc. are provided as
it is very rare for individual chunks of a ByteString
- the sizes of
which can vary - to be independent of their position within the
overall stream.
If you can make such guarantees (e.g. you know that each chunk is a
distinct line and the ordering of these doesn't matter) then you can
use withBufferedTransform
to write your own.