Control.Eff.Concurrent.Protocol.Observer.Queue

A small process to capture and _share_ observation's by enqueueing them into an STM TBQueue.

Documentation

newtype ObservationQueue a

Contains a TBQueue capturing observations. See observe.

Constructors

ObservationQueue (TBQueue a)

Instances

Default (Model (ObservationQueue event))

Defined in Control.Eff.Concurrent.Protocol.Observer.Queue

Methods

def :: Model (ObservationQueue event)

(Typeable event, Lifted IO q, Member Logs q) => Server (ObservationQueue event) (Processes q)

Defined in Control.Eff.Concurrent.Protocol.Observer.Queue

Associated Types

data StartArgument (ObservationQueue event) :: Type

type Protocol (ObservationQueue event) :: Type

data Model (ObservationQueue event) :: Type

type Settings (ObservationQueue event) :: Type

Methods

title :: StartArgument (ObservationQueue event) -> ProcessTitle

setup :: Endpoint (Protocol (ObservationQueue event)) -> StartArgument (ObservationQueue event) -> Eff (Processes q) (Model (ObservationQueue event), Settings (ObservationQueue event))

update :: Endpoint (Protocol (ObservationQueue event)) -> StartArgument (ObservationQueue event) -> Event (Protocol (ObservationQueue event)) -> Eff (ModelState (ObservationQueue event) ': (SettingsReader (ObservationQueue event) ': Processes q)) ()

data StartArgument (ObservationQueue event)

Defined in Control.Eff.Concurrent.Protocol.Observer.Queue

data StartArgument (ObservationQueue event) = MkObservationQueue (ObservationQueue event)

type Protocol (ObservationQueue event)

Defined in Control.Eff.Concurrent.Protocol.Observer.Queue

type Protocol (ObservationQueue event) = Observer event

newtype Model (ObservationQueue event)

Defined in Control.Eff.Concurrent.Protocol.Observer.Queue

newtype Model (ObservationQueue event) = MkObservationQueueModel ()

type Settings (ObservationQueue event)

Defined in Control.Eff.Concurrent.Protocol.Observer.Queue

type Settings (ObservationQueue event) = ()

type Reader a = Reader (ObservationQueue a)

A Reader for an ObservationQueue.

observe :: forall event eventSource e q len b. (HasCallStack, HasProcesses e q, FilteredLogging e, FilteredLogging q, FilteredLogging (Processes q), Lifted IO e, Lifted IO q, IsObservable eventSource event, Integral len, Server (ObservationQueue event) (Processes q), Tangible (Pdu eventSource Asynchronous)) => len -> Endpoint eventSource -> Eff (Reader event ': e) b -> Eff e b

Listen to, and capture observations in an ObservationQueue.

Fork an Observer process that runs only while the body expression is executed. Register the observer to the observable process passed to this function.

The captured observations can be obtained by await, tryRead and flush.

The queue size is limited to the given number.

Example

import qualified Control.Eff.Concurrent.Observer.Queue as OQ

foo =
  do
    observed <- startLink SomeObservable
    OQ.observe 100 observed $ do
      ...
      cast observed DoSomething
      evt <- OQ.await @TestEvent
      ...

Since: 0.28.0

await :: forall event r. (Member (Reader event) r, HasCallStack, MonadIO (Eff r), Typeable event, Member Logs r) => Eff r event

Read queued observations captured and enqueued in the shared ObservationQueue by observe.

This blocks until something was captured or an interrupt or exceptions was thrown. For a non-blocking variant use tryRead or flush.

Since: 0.28.0

tryRead :: forall event r. (Member (Reader event) r, HasCallStack, MonadIO (Eff r), Typeable event, Member Logs r) => Eff r (Maybe event)

Read queued observations captured and enqueued in the shared ObservationQueue by observe.

Return the oldest enqueued observation immediately or Nothing if the queue is empty. Use await to block until an observation is observerRegistryNotify.

Since: 0.28.0

flush :: forall event r. (Member (Reader event) r, HasCallStack, MonadIO (Eff r), Typeable event, Member Logs r) => Eff r [event]

Read at once all currently queued observations captured and enqueued in the shared ObservationQueue by observe.

This returns immediately all currently enqueued observations. For a blocking variant use await.

Since: 0.28.0