extensible-effects-concurrent 0.17.0 → 0.18.0

raw patch · 4 files changed

+58/−61 lines, 4 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• ChangeLog.md +4/−0
• extensible-effects-concurrent.cabal +1/−1
• src/Control/Eff/Concurrent.hs +2/−1
• src/Control/Eff/Concurrent/Api/Observer/Queue.hs +51/−59

@@ -1,5 +1,9 @@ # Changelog for extensible-effects-concurrent +## 0.18.0++- Split-up and replace `spawnLinkObservationQueue` with a simpler (but more verbose) alternative+ ## 0.17.0  - Rename misspelled `spawnLinkObserverationQueue` to `spawnLinkObservationQueue`

@@ -1,5 +1,5 @@ name:           extensible-effects-concurrent-version:        0.17.0+version:        0.18.0 description:    Please see the README on GitHub at <https://github.com/sheyll/extensible-effects-concurrent#readme> synopsis:       Message passing concurrency as extensible-effect homepage:       https://github.com/sheyll/extensible-effects-concurrent#readme

@@ -208,7 +208,8 @@                                                 , readObservationQueue                                                 , tryReadObservationQueue                                                 , flushObservationQueue-                                                , spawnLinkObservationQueue+                                                , withObservationQueue+                                                , spawnLinkObservationQueueWriter                                                 ) import           Control.Eff.Concurrent.Process.ForkIOScheduler                                                 ( schedule

@@ -1,23 +1,21 @@--- | A small process to capture and _share_ observation's by enqueueing them into an STM 'TBQeueu'.+-- | A small process to capture and _share_ observation's by enqueueing them into an STM 'TBQueue'. module Control.Eff.Concurrent.Api.Observer.Queue   ( ObservationQueue()   , ObservationQueueReader   , readObservationQueue   , tryReadObservationQueue   , flushObservationQueue-  , spawnLinkObservationQueue+  , withObservationQueue+  , spawnLinkObservationQueueWriter   ) where  import           Control.Concurrent.STM import           Control.Eff-import           Control.Eff.Extend import           Control.Eff.ExceptionExtra     ( ) import           Control.Eff.Lift import           Control.Eff.Concurrent.Process import           Control.Eff.Log-import           Control.Eff.Concurrent.Api-import           Control.Eff.Concurrent.Api.Client import           Control.Eff.Concurrent.Api.Observer import           Control.Eff.Concurrent.Api.Server import           Control.Eff.Reader.Strict@@ -25,11 +23,10 @@ import           Control.Monad.IO.Class import           Control.Monad                  ( unless ) import           Data.Typeable-import           Text.Printf import           GHC.Stack  -- | Contains a 'TBQueue' capturing observations received by 'enqueueObservationsRegistered'--- or 'spawnLinkObservationQueue'.+-- or 'spawnLinkObservationQueueWriter'. newtype ObservationQueue a = ObservationQueue (TBQueue a)  -- | A 'Reader' for an 'ObservationQueue'.@@ -38,7 +35,7 @@ logPrefix :: forall o proxy . (HasCallStack, Typeable o) => proxy o -> String logPrefix px = "observation queue: " ++ show (typeRep px) --- | Read queued observations captured and enqueued in the shared 'TBQueue' by 'spawnLinkObservationQueue'.+-- | 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'. readObservationQueue@@ -54,7 +51,7 @@   ObservationQueue q <- ask @(ObservationQueue o)   liftIO (atomically (readTBQueue q)) --- | Read queued observations captured and enqueued in the shared 'TBQueue' by 'spawnLinkObservationQueue'.+-- | 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. tryReadObservationQueue@@ -70,7 +67,7 @@   ObservationQueue q <- ask @(ObservationQueue o)   liftIO (atomically (tryReadTBQueue q)) --- | Read at once all currently queued observations captured and enqueued in the shared 'TBQueue' by 'spawnLinkObservationQueue'.+-- | Read at once all currently queued observations captured and enqueued in the shared 'TBQueue' by 'spawnLinkObservationQueueWriter'. -- This returns immediately all currently enqueued 'Observation's. For a blocking -- variant use 'readObservationQueue'. flushObservationQueue@@ -86,54 +83,24 @@   ObservationQueue q <- ask @(ObservationQueue o)   liftIO (atomically (flushTBQueue q)) ---- | Capture an observation.-data instance Api (ObservationQueue a) r where-  EnqueueObservation :: a -> Api (ObservationQueue a) 'Asynchronous-  StopObservationQueue :: Api (ObservationQueue a) ('Synchronous ())---- | Observe a 'Server' that implements an 'Observable' 'Api'. 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'.-spawnLinkObservationQueue-  :: forall o q a-   . (Typeable o, Show o, HasLogging IO q, Lifted IO q, HasCallStack)-  => Server (ObserverRegistry o)-  -> Int-  -> Eff (ObservationQueueReader o ': InterruptableProcess q) a-  -> Eff (InterruptableProcess q) a-spawnLinkObservationQueue oSvr queueLimit k = withQueue-  queueLimit-  (do-    ObservationQueue q <- ask @(ObservationQueue o)-    do-      logDebug-        (printf "%s starting with queue limit: %d"-                (logPrefix (Proxy @o))-                queueLimit-        )-      cbo <- raise-        (spawnLinkApiServer-          (handleCasts-            (\case-              EnqueueObservation o -> do-                lift (atomically (writeTBQueue q o))-                pure AwaitNext-            )-          )-          stopServerOnInterrupt-        )-      let thisObserver = toObserverFor EnqueueObservation cbo-      registerObserver thisObserver oSvr-      res <- k-      forgetObserver thisObserver oSvr-      call cbo StopObservationQueue-      logDebug (printf "%s stopped observer process" (logPrefix (Proxy @o)))-      return res-  )--withQueue+-- | 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+withObservationQueue   :: forall o b e len    . ( HasCallStack      , Typeable o@@ -145,7 +112,7 @@   => len   -> Eff (ObservationQueueReader o ': e) b   -> Eff e b-withQueue queueLimit e = do+withObservationQueue queueLimit e = do   q   <- liftIO (newTBQueueIO (fromIntegral queueLimit))   res <- handleInterrupts (return . Left)                           (Right <$> runReader (ObservationQueue q) e)@@ -154,3 +121,28 @@     (null rest)     (logNotice (logPrefix (Proxy @o) ++ " unread observations: " ++ show rest))   either (\em -> logError (show em) >> liftIO (throwIO em)) return res++-- | 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+spawnLinkObservationQueueWriter+  :: forall o q+   . (Typeable o, Show o, HasLogging IO q, Lifted IO q, HasCallStack)+  => ObservationQueue o+  -> Eff (InterruptableProcess q) (Observer o)+spawnLinkObservationQueueWriter (ObservationQueue q) = do+  cbo <- spawnLinkApiServer+    (handleObservations+      (\case+        o -> do+          lift (atomically (writeTBQueue q o))+          pure AwaitNext+      )+    )+    stopServerOnInterrupt+  pure (toObserver cbo)