- toList :: Topic IO a -> IO [a]
- fromList :: Monad m => [a] -> Topic m a
- tee :: Topic IO a -> IO (Topic IO a, Topic IO a)
- teeEager :: Topic IO a -> IO (Topic IO a, Topic IO a)
- fan :: Int -> Topic IO a -> IO [Topic IO a]
- share :: Topic IO a -> IO (Topic IO a)
- topicRate :: (Functor m, MonadIO m) => Double -> Topic m a -> Topic m a
- partition :: (a -> Bool) -> Topic IO a -> IO (Topic IO a, Topic IO a)
- consecutive :: Monad m => Topic m a -> Topic m (a, a)
- (<+>) :: Topic IO a -> Topic IO b -> Topic IO (Either a b)
- everyNew :: Topic IO a -> Topic IO b -> Topic IO (a, b)
- bothNew :: Topic IO a -> Topic IO b -> Topic IO (a, b)
- merge :: Topic IO a -> Topic IO a -> Topic IO a
- finiteDifference :: (Functor m, Monad m) => (a -> a -> b) -> Topic m a -> Topic m b
- weightedMeanNormalized :: Monad m => n -> n -> (b -> b -> c) -> (n -> a -> b) -> (c -> a) -> Topic m a -> Topic m a
- simpsonsRule :: (Monad m, Fractional n) => (a -> a -> a) -> (n -> a -> a) -> Topic m a -> Topic m a
- weightedMean :: (Monad m, Num n) => n -> (a -> a -> a) -> (n -> a -> a) -> Topic m a -> Topic m a
- weightedMean2 :: Monad m => n -> n -> (a -> a -> a) -> (n -> a -> a) -> Topic m a -> Topic m a
- filterBy :: Monad m => Topic m (a -> Bool) -> Topic m a -> Topic m a
- gate :: (Applicative m, Monad m) => Topic m a -> Topic m b -> Topic m a
- concats :: (Monad m, Foldable f) => Topic m (f a) -> Topic m a
- interruptible :: Foldable t => Topic IO (t a) -> Topic IO a
- forkTopic :: Topic IO a -> IO (Topic IO a)
- slidingWindow :: (Monad m, Monoid a) => Int -> Topic m a -> Topic m a
- slidingWindowG :: (Monad m, AdditiveGroup a) => Int -> Topic m a -> Topic m a
- topicOn :: (Applicative m, Monad m) => (a -> b) -> (a -> c -> d) -> m (b -> m c) -> Topic m a -> Topic m d
- subsample :: Monad m => Int -> Topic m b -> Topic m b

# Documentation

tee :: Topic IO a -> IO (Topic IO a, Topic IO a)

Tee a `Topic`

into two duplicate `Topic`

s. Each returned `Topic`

will receive all the values of the original `Topic`

while any
side-effect produced by each step of the original `Topic`

will
occur only once.

This version of `tee`

lazily pulls data from the original `Topic`

when it is first required by a consumer of either of the returned
`Topic`

s. This behavior is crucial when lazily consuming the data
stream is preferred. For instance, using `interruptible`

with `tee`

will allow for a chunk of data to be abandoned before being fully
consumed as long as neither consumer has forced its way too far
down the stream.

This function is useful when two consumers must see all the same
elements from a `Topic`

. If the `Topic`

was instead `share`

d, then
one consumer might get the first value from the `Topic`

before the
second consumer's buffer is created since buffer creation is lazy.

teeEager :: Topic IO a -> IO (Topic IO a, Topic IO a)

This version of `tee`

eagerly pulls data from the
original `Topic`

as soon as it is available. This behavior is
undesirable when lazily consuming the data stream is preferred. For
instance, using `interruptible`

with `teeEager`

will likely not
work well. However, `teeEager`

may have slightly better performance
than `tee`

.

share :: Topic IO a -> IO (Topic IO a)

Make a `Topic`

shareable among multiple consumers. Each consumer
of a Topic gets its own read buffer automatically as soon as it
starts pulling items from the Topic. Without calling one of
`share`

, `tee`

, or `fan`

on a Topic, the Topic's values will be
split among all consumers (e.g. consumer *A* gets half the values
produced by the `Topic`

, while consumer *B* gets the other half
with some unpredictable interleaving). Note that Topics returned by
the `Ros.Node.subscribe`

are already shared.

topicRate :: (Functor m, MonadIO m) => Double -> Topic m a -> Topic m a

The application `topicRate rate t`

runs `Topic`

`t`

no faster than
`rate`

Hz.

consecutive :: Monad m => Topic m a -> Topic m (a, a)

everyNew :: Topic IO a -> Topic IO b -> Topic IO (a, b)

Returns a `Topic`

that produces a new pair every time either of
the component `Topic`

s produces a new value. The value of the
other element of the pair will be the newest available value. The
resulting `Topic`

will produce a new value at the rate of the
faster component `Topic`

, and may contain duplicate consecutive
elements.

finiteDifference :: (Functor m, Monad m) => (a -> a -> b) -> Topic m a -> Topic m b

Apply a function to each consecutive pair of elements from a `Topic`

.

weightedMeanNormalized :: Monad m => n -> n -> (b -> b -> c) -> (n -> a -> b) -> (c -> a) -> Topic m a -> Topic m a

Compute a running "average" of a `Topic`

using a user-provided
normalization function applied to the sum of products. The
arguments are a constat `alpha`

that is used to scale the current
average, a constant `invAlpha`

used to scale the newest value, a
function for adding two scaled values, a function for scaling
input values, a function for normalizing the sum of scaled values,
and finally the stream to average. Parameterizing over all the
arithmetic to this extent allows for the use of denormalizing
scaling factors, as might be used to keep all arithmetic
integral. An example would be scaling the average by the integer
7, the new value by the integer 1, then normalizing by dividing
the sum of scaled values by 8.

simpsonsRule :: (Monad m, Fractional n) => (a -> a -> a) -> (n -> a -> a) -> Topic m a -> Topic m a

weightedMean :: (Monad m, Num n) => n -> (a -> a -> a) -> (n -> a -> a) -> Topic m a -> Topic m a

Compute a running "average" of a `Topic`

. The application
`weightedMean alpha plus scale t`

sums the product of `alpha`

and
the current average with the product of `1 - alpha`

and the newest
value produced by `Topic`

`t`

. The addition and scaling operations
are performed using the supplied `plus`

and `scale`

functions.

weightedMean2 :: Monad m => n -> n -> (a -> a -> a) -> (n -> a -> a) -> Topic m a -> Topic m a

Compute a running "average" of a `Topic`

. The application
`weightedMean2 alpha invAlpha plus scale t`

sums the product of
`alpha`

and the current average with the product of `invAlpha`

and
the newest value produced by `Topic`

`t`

. The addition and scaling
operations are performed using the supplied `plus`

and `scale`

functions.

filterBy :: Monad m => Topic m (a -> Bool) -> Topic m a -> Topic m a

Use a `Topic`

of functions to filter a `Topic`

of values. The
application `filterBy t1 t2`

causes each function from `Topic`

`t1`

to be applied to values produced by `t2`

until it returns
`True`

. At that point, the `filterBy`

application produces the
accepted value of the `t2`

and moves on to the next function from
`t1`

which is applied to the rest of `t2`

in the same manner.

slidingWindowG :: (Monad m, AdditiveGroup a) => Int -> Topic m a -> Topic m a

Sliding window over an `AdditiveGroup`

. `slidingWindowG n t`

slides a window of width `n`

along `Topic`

`t`

. As soon as at least
`n`

elements have been produced by `t`

, the output `Topic`

starts
producing the total sum of the elements of the window. This
function is more efficient than `slidingWindow`

because the group
inverse operation is used to remove elements falling behind the
window from the running sum.

topicOn :: (Applicative m, Monad m) => (a -> b) -> (a -> c -> d) -> m (b -> m c) -> Topic m a -> Topic m d

A way of pushing a monadic action into and along a `Topic`

. The
application `topicOn proj inj trans t`

extracts a function from
`trans`

that is then applied to the result of applying `proj`

to
each value of `Topic`

`t`

. The result of that application is
supplied to the result of applying `inj`

to the same values from
`t`

to produce a value for the output `Topic`

. A typical use case
is projecting out a field from the original `Topic`

`t`

using
`proj`

so that it may be modified by `trans`

and then injected back
into the original structure using `inj`

.