FS.Concurrent

This module provides combinators for running streams concurrently,
merging the output they produce nondeterministically, or interrupting
a stream after a timeout.

Unlike the combinators in `FS.Stream`, we need a concurrent runtime
to use the combinators provided here, which means that they run in
the `Async` monad.

Reexports

Definitions

0 AsyncPull : Type -> Type -> List Type -> Type -> Type

  Convenience alias for `Pull . Async`

Totality: total
Visibility: public export

0 AsyncStream : Type -> List Type -> Type -> Type

  Convenience alias for `Stream . Async`

Totality: total
Visibility: public export

sleep : TimerH e => Clock Duration -> AsyncStream e es o

  An empty stream that terminates after the given delay.

Totality: total
Visibility: export

waitTill : TimerH e => Clock Monotonic -> AsyncStream e es o

  An empty stream that terminates at the given clock time.

Totality: total
Visibility: export

delayed : TimerH e => Clock Duration -> o -> AsyncStream e es o

  Emits the given value after a delay of the given duration.

Totality: total
Visibility: export

atClock : TimerH e => Clock Monotonic -> o -> AsyncStream e es o

  Emits the given value after at the given clock time

Totality: total
Visibility: export

timed : TimerH e => Clock Duration -> o -> AsyncStream e es o

  Infinitely emits the given value at regular intervals.

Totality: total
Visibility: export

every : TimerH e => Clock Duration -> AsyncStream e es a -> AsyncStream e es a

  Emits the values from the given stream, each with a delay of the
  given duration.
  
  The first value will be emitted `after` the given delay.
  
  Note: If the given stream emits some values more slowly than specified
        by the delays, irregular emission of several values at a time
        might be observed.

Totality: total
Visibility: export

every0 : TimerH e => Clock Duration -> AsyncStream e es a -> AsyncStream e es a

  Like `every` but the first value will be emitted as soon as the
  original stream emits it.

Totality: total
Visibility: export

dequeue : BQueue o -> AsyncStream e es o

  Converts a bounded queue of values into an infinite stream
  of values.

Totality: total
Visibility: export

receive : Channel o -> AsyncStream e es o

  Converts a channel of chunks into an infinite stream of values.

Totality: total
Visibility: export

timeout : TimerH e => Clock Duration -> AsyncStream e es o -> AsyncStream e es o

  Runs the given stream until the given duration expires.

Totality: total
Visibility: export

merge : List (AsyncStream e es o) -> AsyncStream e es o

  Runs the given streams in parallel and nondeterministically
  (but chunkc-wise) interleaves their output.
  
  The resulting stream will emit chunks as long as one of the input
  streams is still alive, or until one of the input streams terminates
  with an error, in which case the output stream will terminate with
  the same error.

Totality: total
Visibility: export

mergeHaltL : AsyncStream e es o -> AsyncStream e es o -> AsyncStream e es o

  Runs the given streams in parallel and nondeterministically interleaves
  their output.
  
  This will terminate as soon as the first string is exhausted.

Totality: total
Visibility: export

mergeHaltBoth : AsyncStream e es o -> AsyncStream e es o -> AsyncStream e es o

  Runs the given streams in parallel and nondeterministically interleaves
  their output.
  
  This will terminate as soon as either stream is exhausted.

Totality: total
Visibility: export

haltOn : AsyncStream e es o -> AsyncStream e es p -> AsyncStream e es p

  Runs the second stream until the first emits a value

Totality: total
Visibility: export

parJoin : (maxOpen : Nat) -> {auto 0 _ : IsSucc maxOpen} -> AsyncStream e es (AsyncStream e es o) -> AsyncStream e es o

  Nondeterministically merges a stream of streams (`outer`) in to a single stream,
  opening at most `maxOpen` streams at any point in time.
  
  The outer stream is evaluated and each resulting inner stream is run concurrently,
  up to `maxOpen` stream. Once this limit is reached, evaluation of the outer stream
  is paused until one of the inner streams finishes evaluating.
  
  When the outer stream stops gracefully, all inner streams continue to run,
  resulting in a stream that will stop when all inner streams finish
  their evaluation.
  
  Finalizers on each inner stream are run at the end of the inner stream,
  concurrently with other stream computations.
  
  Finalizers on the outer stream are run after all inner streams have been pulled
  from the outer stream but not before all inner streams terminate
  -- hence finalizers on the outer stream will run
  AFTER the LAST finalizer on the very last inner stream.
  
  Finalizers on the returned stream are run after the outer stream has finished
  and all open inner streams have finished.

Totality: total
Visibility: export

parBind : (maxOpen : Nat) -> {auto 0 _ : IsSucc maxOpen} -> (o -> AsyncStream e es p) -> AsyncStream e es o -> AsyncStream e es p

  Convenience alias for `P.mapOutput inner outer |> parJoin maxOpen`

Totality: total
Visibility: export

broadcast : AsyncStream e es o -> (pipes : List (o -> AsyncStream e es p)) -> {auto 0 _ : NonEmpty pipes} -> AsyncStream e es p

Totality: total
Visibility: export

parMapE : (maxOpen : Nat) -> {auto 0 _ : IsSucc maxOpen} -> (o -> Result es p) -> AsyncStream e es o -> AsyncStream e es p

  Uses `parJoin` to map the given function over each emitted output
  in parallel.

Totality: total
Visibility: export

parMapI : Has x es => (maxOpen : Nat) -> {auto 0 _ : IsSucc maxOpen} -> (o -> Either x p) -> AsyncStream e es o -> AsyncStream e es p

  Like `parMapE`, but injects the error first.

Totality: total
Visibility: export

parMap : (maxOpen : Nat) -> {auto 0 _ : IsSucc maxOpen} -> (o -> p) -> AsyncStream e es o -> AsyncStream e es p

  Like `parMapE`, but for a function that cannot fail.

Totality: total
Visibility: export

foreachPar : (maxOpen : Nat) -> {auto 0 _ : IsSucc maxOpen} -> (o -> Async e [] ()) -> AsyncPull e o es r -> AsyncPull e q es r

Totality: total
Visibility: export

switchMap : (o -> AsyncStream e es p) -> AsyncStream e es o -> AsyncStream e es p

  Like `flatMap` but interrupts the inner stream when
  new elements arrive in the outer stream.
  
  Finializers of each inner stream are guaranteed to run
  before the next inner stream starts.
  When the outer stream stops gracefully, the currently running
  inner stream will continue to run.
  
  When an inner stream terminates/interrupts, nothing
  happens until the next element arrives
  in the outer stream.
  
  When either the inner or outer stream fails, the entire
  stream fails and the finalizer of the
  inner stream runs before the outer one.
  

Totality: total
Visibility: export

record Hold : Type -> List Type -> Type -> Type

Totality: total
Visibility: public export
Constructor: 

H : Async e [] () -> AsyncStream e es o -> Hold e es o

Projections:

.release : Hold e es o -> Async e [] ()

.stream : Hold e es o -> AsyncStream e es o

Hint: 

Resource (Async e) (Hold e es o)

.release : Hold e es o -> Async e [] ()

Totality: total
Visibility: public export

release : Hold e es o -> Async e [] ()

Totality: total
Visibility: public export

.stream : Hold e es o -> AsyncStream e es o

Totality: total
Visibility: public export

stream : Hold e es o -> AsyncStream e es o

Totality: total
Visibility: public export

hold : o -> AsyncStream e es o -> Async e fs (Hold e es o)

  Converts a discrete stream of values into a continuous one that will
  emit the last value emitted by the original stream on every pull starting
  with the given initial value.
  
  The original stream is immediately started and
  processed in the background.
  
  This should be used in combination with a call to `bracket` or
  `resource`, so that the stream running in the background is
  properly terminated and its resources released
  once the resulting stream is exhausted.
  
  ```idris example
  signalOn :
      o
   -> AsyncStream e es ()
   -> AsyncStream e es o
   -> AsyncStream e es o
  signalOn ini tick sig =
    resource (hold ini sig) (zipRight tick . stream)
  ```

Totality: total
Visibility: export

hold1 : AsyncStream e es o -> Async e fs (Hold e es o)

  Like `hold` but the resulting stream will not emit a value
  until after the original stream first emitted a value.

Totality: total
Visibility: export

signalOn : o -> AsyncStream e es () -> AsyncStream e es o -> AsyncStream e es o

  Runs the second stream in the background, emitting its latest
  output whenever the first stream emits.

Totality: total
Visibility: export

signalOn1 : AsyncStream e es () -> AsyncStream e es o -> AsyncStream e es o

  Like `signalOn` but only starts emitting values *after* the
  second stream emitted its first value.

Totality: total
Visibility: export

logExec : All (Loggable e) es => t -> Async e es t -> Pull (Async e) o [] t

Totality: total
Visibility: export

tryExec : All (Loggable e) es => Async e es t -> Pull (Async e) o [] (Maybe t)

Totality: total
Visibility: export

tryPull : All (Loggable e) es => r -> Pull (Async e) o es r -> Pull (Async e) o [] r

Totality: total
Visibility: export

tryStream : All (Loggable e) es => Stream (Async e) es o -> Stream (Async e) [] o

Totality: total
Visibility: export

record Signal : Type -> Type

  An observe-only wrapper around a `SignalRef`.
  
  Use this if you still want to observe a mutable value by means of
  `discrete` or `continuous` but you want to prevent it to be further used
  as a data sink.

Totality: total
Visibility: export
Constructor: 

MkSignal : IO1 a -> Stream (Async e) es a -> Signal a

Projections:

.sigdisc : Signal a -> Stream (Async e) es a

.signow : Signal a -> IO1 a

Hints:

Applicative Signal

Discrete Signal

Functor Signal

Reference Signal

SignalRef a => Signal a

Zippable Signal

sig : SignalRef a -> Signal a

Totality: total
Visibility: export

ref2sig : SignalRef a => Signal a

Totality: total
Visibility: export

hobserveSig : LIO (f es) => All Signal ts -> HZipFun (o :: ts) (f es ()) -> Pull f o es r -> Pull f o es r

  Generalization of `observeSig`: Acts on the output of a pull by combining
  with the values from a heterogeneous list of signals.

Totality: total
Visibility: export

hforeachSig : LIO (f es) => All Signal ts -> HZipFun (o :: ts) (f es ()) -> Pull f o es r -> Pull f q es r

  Like `hobserveSig` but drains the stream in the process.

Totality: total
Visibility: export