Control.Eff.Concurrent.Api.Observer.Queue

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

Documentation

data ObservationQueue a

Contains a TBQueue capturing observations. See spawnLinkObservationQueueWriter, readObservationQueue.

type ObservationQueueReader a = Reader (ObservationQueue a)

A Reader for an ObservationQueue.

readObservationQueue :: forall o r. (Member (ObservationQueueReader o) r, HasCallStack, MonadIO (Eff r), Typeable o, Member Logs r) => Eff r o

Read queued observations captured and enqueued in the shared TBQueue by spawnLinkObservationQueueWriter. This blocks until something was captured or an interrupt or exceptions was thrown. For a non-blocking variant use tryReadObservationQueue or flushObservationQueue.

tryReadObservationQueue :: forall o r. (Member (ObservationQueueReader o) r, HasCallStack, MonadIO (Eff r), Typeable o, Member Logs r) => Eff r (Maybe o)

Read queued observations captured and enqueued in the shared TBQueue by spawnLinkObservationQueueWriter. Return the oldest enqueued observation immediately or Nothing if the queue is empty. Use readObservationQueue to block until an observation is observed.

flushObservationQueue :: forall o r. (Member (ObservationQueueReader o) r, HasCallStack, MonadIO (Eff r), Typeable o, Member Logs r) => Eff r [o]

Read at once all currently queued observations captured and enqueued in the shared TBQueue by spawnLinkObservationQueueWriter. This returns immediately all currently enqueued observations. For a blocking variant use readObservationQueue.

withObservationQueue :: forall o b e len. (HasCallStack, Typeable o, Show o, Member Logs e, Lifted IO e, Integral len, Member Interrupts e) => len -> Eff (ObservationQueueReader o ': e) b -> Eff e b

Create a mutable queue for observations. Use spawnLinkObservationQueueWriter for a simple way to get a process that enqueues all observations.

Example

withObservationQueue 100 $ do
  q  <- ask @(ObservationQueueReader TestEvent)
  wq <- spawnLinkObservationQueueWriter q
  registerObserver wq testServer
  ...
  cast testServer DoSomething
  evt <- readObservationQueue @TestEvent
  ...

Since: 0.18.0

spawnLinkObservationQueueWriter :: forall o q. (Typeable o, Show o, Member Logs q, Lifted IO q, HasCallStack) => ObservationQueue o -> Eff (InterruptableProcess q) (Observer o)

Spawn a process that can be used as an Observer that enqueues the observations into an ObservationQueue. See withObservationQueue for an example.

The observations can be obtained by readObservationQueue. All observations are captured up to the queue size limit, such that the first message received will be first message returned by readObservationQueue.

Since: 0.18.0