extensible-effects-concurrent 0.29.2 → 0.30.0

raw patch · 35 files changed

+4110/−1536 lines, 35 filesdep +hashabledep −bytestringdep −paralleldep −processdep ~data-defaultdep ~deepseqdep ~filepathPVP ok

version bump matches the API change (PVP)

Dependencies added: hashable

Dependencies removed: bytestring, parallel, process, tagged

Dependency ranges changed: data-default, deepseq, filepath, lens, text

API changes (from Hackage documentation)

Files

• ChangeLog.md +33/−0
• README.md +200/−48
• examples/example-1/Main.hs +1/−1
• examples/example-2/Main.hs +32/−12
• examples/example-embedded-protocols/Main.hs +28/−26
• extensible-effects-concurrent.cabal +21/−27
• src/Control/Eff/Concurrent.hs +8/−2
• src/Control/Eff/Concurrent/Misc.hs +14/−11
• src/Control/Eff/Concurrent/Process.hs +187/−85
• src/Control/Eff/Concurrent/Process/ForkIOScheduler.hs +123/−52
• src/Control/Eff/Concurrent/Process/SingleThreadedScheduler.hs +30/−10
• src/Control/Eff/Concurrent/Process/Timer.hs +152/−46
• src/Control/Eff/Concurrent/Protocol/Broker.hs +645/−0
• src/Control/Eff/Concurrent/Protocol/Broker/InternalState.hs +117/−0
• src/Control/Eff/Concurrent/Protocol/Client.hs +5/−3
• src/Control/Eff/Concurrent/Protocol/EffectfulServer.hs +18/−14
• src/Control/Eff/Concurrent/Protocol/Observer.hs +2/−2
• src/Control/Eff/Concurrent/Protocol/Observer/Queue.hs +31/−22
• src/Control/Eff/Concurrent/Protocol/StatefulServer.hs +103/−12
• src/Control/Eff/Concurrent/Protocol/Supervisor.hs +0/−457
• src/Control/Eff/Concurrent/Protocol/Supervisor/InternalState.hs +0/−105
• src/Control/Eff/Concurrent/Protocol/Watchdog.hs +616/−0
• src/Control/Eff/Log.hs +3/−0
• src/Control/Eff/Log/Handler.hs +74/−3
• src/Control/Eff/LogWriter/Console.hs +7/−2
• src/Control/Eff/LogWriter/File.hs +2/−2
• stack.yaml +1/−0
• test/BrokerTests.hs +367/−0
• test/Common.hs +143/−35
• test/GenServerTests.hs +15/−15
• test/LoopTests.hs +0/−6
• test/ObserverTests.hs +44/−35
• test/ProcessBehaviourTestCases.hs +351/−212
• test/SupervisorTests.hs +0/−291
• test/WatchdogTests.hs +737/−0

@@ -1,5 +1,38 @@ # Changelog for extensible-effects-concurrent +## 0.30.0+- Improve inline code documentation+- **Supervisor:** +    - Rename to **[Broker](./src/Control/Eff/Concurrent/Protocol/Broker.hs)**  +    - Add the `ChildEvent` needed by the new **watchdog**  +    - Add `callById` and `castById`+- [Process](./src/Control/Eff/Concurrent/Process.hs)+    - Introduce a new `Interrupt NoRecovery` clause: `ExitOtherProcessNotRunning`+    - Change the second parameter of `ProcessDown` from `SomeExitReason` to `Interrupt NoRecovery`+    - Introduce new `Interrupt` reasons+      for all categories with an existential+      parameter, that must have `NFData`, `Show`+      and `Typeable` constraints.+    - Introduce a new timing primitive: `Delay`   +- [StatefulServer](./src/Control/Eff/Concurrent/Protocol/StatefulServer.hs)+    - _Upgrade_ the associated **type alias** `Model`  to an associated **type**.+    - Add `mapEffects` +    - Add `coerceEffects`        +- Export the `TimeoutMicros` constructor+- Add **[Watchdog](./src/Control/Eff/Concurrent/Protocol/Watchdog.hs)** a server that watches +  a **[Broker](./src/Control/Eff/Concurrent/Protocol/Broker.hs)** and restarts crashed+  processes registered at the broker.+        +- **[Logging](./src/Control/Eff/Log.hs)**+    - Add `logCallStack`+    - Add `logMultiLine`      ++-  **[Timer](./src/Control/Eff/Concurrent/Process/Timer.hs)**+    - Allow timers to have custom titles via:+        - `sendAfterWithTitle`+        - `startTimerWithTitle`+    - Switch to use the new `Delay` primitive    +             ## 0.29.2 - Improve `Supervisor` API: Use `Init` from the effectful server as start   argument type.

@@ -20,6 +20,56 @@ The fundamental approach to modelling applications in Erlang is based on the concept of concurrent, communicating processes. ++### Example Code++```haskell+module Main where++import           Control.Eff+import           Control.Eff.Concurrent++main :: IO ()+main = defaultMain example++example :: Eff Effects ()+example = do+  person <- spawn "alice" alice+  replyToMe <- self+  sendMessage person replyToMe+  personName <- receiveMessage+  logInfo' ("I just met " ++ personName)++alice :: Eff Effects ()+alice = do+  logInfo "I am waiting for someone to ask me..."+  sender <- receiveMessage+  sendMessage sender ("Alice" :: String)+  logInfo' (show sender ++ " message received.")++```+This is taken from [example-4](./examples/example-4/Main.hs).+++**Running** this example causes this output:++```text+DEBUG      no proc  scheduler loop entered                                       at ForkIOScheduler.hs:209+DEBUG        init!1 enter process                                                at ForkIOScheduler.hs:691+NOTICE       init!1 ++++++++ main process started ++++++++                       at ForkIOScheduler.hs:579+DEBUG       alice!2 enter process                                                at ForkIOScheduler.hs:691+INFO        alice!2 I am waiting for someone to ask me...                        at Main.hs:19+INFO        alice!2 !1 message received.                                         at Main.hs:22+DEBUG       alice!2 exit: Process finished successfully                          at ForkIOScheduler.hs:729+INFO         init!1 I just met Alice                                             at Main.hs:15+NOTICE       init!1 ++++++++ main process returned ++++++++                      at ForkIOScheduler.hs:581+DEBUG        init!1 exit: Process finished successfully                          at ForkIOScheduler.hs:729+DEBUG      no proc  scheduler loop returned                                      at ForkIOScheduler.hs:211+DEBUG      no proc  scheduler cleanup begin                                      at ForkIOScheduler.hs:205+NOTICE     no proc  cancelling processes: []                                     at ForkIOScheduler.hs:222+NOTICE     no proc  all processes cancelled                                      at ForkIOScheduler.hs:239+```+ The mental model of the programming framework regards objects as **processes** with an isolated internal state.  @@ -65,72 +115,174 @@ a process is gone, when a process exits; hence restarting a process will not be bothered by left-over, possibly inconsistent, state.  -### Higher Level Abstractions+### Timers  -Processes can receive only message of type `Dynamic`.+[The Timer module](src/Control/Eff/Concurrent/Process/Timer.hs) contains functions to send messages after a time+has passed, and reiceive messages with timeouts. -In order to leverage Haskells type-safety, a bit of support code is available.+### More Type-Safety: [The Protocol Metaphor](./src/Control/Eff/Concurrent/Protocol.hs) -There is a **data family** called **`Api`** allowing to model **calls** and -**casts**, as well as event management and process supervision.+As the library carefully leaves the realm of untyped messages, it uses the +concept of a **protocol** that governs the communication+between concurrent processes, which are either **protocol servers** or+**clients** as a metaphor. -These facilities are very important to build **non-defensive**, **let-it-crash**-applications, resilient to runtime errors.   +The communication is initiated by the client. -### Additional services+The idea is to indicate such a _protocol_ using a **custom data type**,+e.g. `data TemperatureSensorReader` or `data SqlServer`.+ +The library consists some tricks to restrict the kinds of messages that+are acceptable when communicating with processes _adhering to the protocol_. -Currently a custom **logging effect** is also part of the code base.+This _protocol_ is not encoded in the users code, but rather something that+the programmer keeps in his head.  -## Usage and Implementation+In order to be appreciated by authors of real world applications, the +protocol can be defined by giving an abstract message sum-type and +code for spawning server processes. -### Example Code+It focusses on these questions: -```haskell-module Main where+1. What messages does a process accept?+2. When sending a certain message, should the sender wait for an answer? -import           Control.Eff-import           Control.Eff.Concurrent+#### Protocol Phantom Type -main :: IO ()-main = defaultMain example+In this library, the key to a _protocol_ is a single type,+that could even be a so called __phantom type__, i.e. a+type without any runtime values: -example :: Eff Effects ()-example = do-  person <- spawn "alice" alice-  replyToMe <- self-  sendMessage person replyToMe-  personName <- receiveMessage-  logInfo' ("I just met " ++ personName)+```haskell  -alice :: Eff Effects ()-alice = do-  logInfo "I am waiting for someone to ask me..."-  sender <- receiveMessage-  sendMessage sender ("Alice" :: String)-  logInfo' (show sender ++ " message received.")+data UserRegistry -- look mom, no constructors!!  -```-This is taken from [example-4](./examples/example-4/Main.hs).+```  +Such a type exists only for the type system. -**Running** this example causes this output:+It can only be used as a parameter to certain type constructors,+and for defining type class and type family instances, e.g. -```text-DEBUG     scheduler loop entered                                                   ForkIOScheduler.hs line 157-DEBUG            !1 enter process                                                            ForkIOScheduler.hs line 549-NOTICE           !1 ++++++++ main process started ++++++++                                   ForkIOScheduler.hs line 461-DEBUG            !2 enter process                                                            ForkIOScheduler.hs line 549-INFO             !2 I am waiting for someone to ask me...                                               Main.hs line 26-INFO             !2 !1 just needed to know it.                                                          Main.hs line 29-DEBUG            !2 exit normally                                                            ForkIOScheduler.hs line 568-INFO             !1 I just met Alice                                                                    Main.hs line 34-NOTICE           !1 ++++++++ main process returned ++++++++                                  ForkIOScheduler.hs line 463-DEBUG            !1 exit normally                                                            ForkIOScheduler.hs line 568-DEBUG     scheduler loop returned                                                  ForkIOScheduler.hs line 159-DEBUG     scheduler cleanup begin                                                  ForkIOScheduler.hs line 154-NOTICE    cancelling processes: []                                                 ForkIOScheduler.hs line 168-NOTICE    all processes cancelled                                                  ForkIOScheduler.hs line 179++```haskell ++newtype Endpoint protocol = MkServer { _processId :: ProcessId }++data UserRegistry++startUserRegistry :: Eff e (Endpoint UserRegistry)+startUserRegistry = +  error "just an example"+ ```++Here the `Endpoint` has a type parameter `protocol` but the type is not+used by the constructor to hold any values, hence we can use `UserRegistry`,+as a parameter, since `UserRegistry` has no value constructors.++#### Protocol Data Units++Messages that belong to a protocol are called __protocol data units (PDU)__.++#### Protocol Servers Endpoints++The `ProcessId` of a process identifies the messages box that+`receiveMessage` will use, when waiting for an incoming message.++While it defines _where_ the messages are collected, it does +not restrict or inform about _what_ data is handled by a process.++An [Endpoint](https://hackage.haskell.org/package/extensible-effects-concurrent/docs/Control-Eff-Concurrent-Protocol.html#t:Endpoint) is a wrapper +around the `ProcessId` that takes a _type parameter_.++The type does not have to have any values, it can be a phantom type.++This type serves only the tag the process as a server accepting messages identified by the [HasPdu](https://hackage.haskell.org/package/extensible-effects-concurrent/docs/Control-Eff-Concurrent-Protocol.html#t:HasPdu)   type class.++#### Server/Protocol Composability++Usually a protocol consists of some really custom PDUs and+some PDUs that more or less are found in many protocols,+like event listener registration and event notification.++It is therefore helpful to be able to compose protocols.++The machinery in this library allows to list several +PDU instances understood by endpoints of a given protocol+phantom type.++#### Protocol Clients   ++_Clients_ use a protocol by sending `Pdu`s indexed by +some protocol phantom type to a server process.++Clients use `Endpoint`s to address these servers, and the+functions defined in [the corresponding module](https://hackage.haskell.org/package/extensible-effects-concurrent/docs/Control-Eff-Concurrent-Protocol-Client.html).++Most important are these two functions:  ++* [cast](https://hackage.haskell.org/package/extensible-effects-concurrent/docs/Control-Eff-Concurrent-Protocol-Client.html#v:cast)+  to send fire-and-forget messages+   +* [call](https://hackage.haskell.org/package/extensible-effects-concurrent/docs/Control-Eff-Concurrent-Protocol-Client.html#v:cast)+  to send RPC-style requests and wait (block) for the responses.+  Also, when the server is not running, or crashes in +  while waiting, the calling process is interrupted++#### Protocol Servers   ++This library offers an API for defining practically safe to+ use __protocol servers__:++* [EffectfulServer](./src/Control/Eff/Concurrent/Protocol/EffectfulServer.hs) This modules defines the framework+  of a process, that has a callback function that is repeatedly+  called when ever a message was received.+  +  The callback may rely on any extra effects (extensible effects).    ++* [StatefulServer](./src/Control/Eff/Concurrent/Protocol/StatefulServer.hs) A server based on the `EffectfulServer` that includes the definition of+  in internal state called __Model__, and some nice +  helper functions to access the model. These functions+  allow the use of lenses. +  Unlike the effectful server, the effects that the +  callback functions can use are defined in this module.++* [CallbackServer](./src/Control/Eff/Concurrent/Protocol/CallbackServer.hs) A server based on the `EffectfulServer` that does not require a type class+   instance like the stateful and effectful servers do.+   It can be used to define _inline_ servers.+  +#### Events and Observers++A parameterized protocol for event handling is provided in +the module:++* [Observer](./src/Control/Eff/Concurrent/Protocol/Observer.hs) + ++#### Brokers and Watchdogs++A key part of a robust system is monitoring and possibly restarting +stuff that crashes, this is done in conjunction by two modules:++* [Broker](./src/Control/Eff/Concurrent/Protocol/Broker.hs)+* [Watchdog](./src/Control/Eff/Concurrent/Protocol/Watchdog.hs)++A client of a process that might be restarted cannot use the `ProcessId` +directly, but has to use an abstract ID and lookup the `ProcessId` from a +process **[broker](./src/Control/Eff/Concurrent/Protocol/Broker.hs)**, that manages the current `ProcessId` of protocol server+processes.++That way, when ever the server process registered at a broker crashes,+**a [watchdog](./src/Control/Eff/Concurrent/Protocol/Watchdog.hs) process** can (re-)start the crashed server.  ++### Additional services++Currently a **logging effect** is also part of the code base.++## Usage and Implementation++Should work with `stack`, `cabal` and `nix`.   ### Required GHC Extensions 

@@ -76,7 +76,7 @@   go  testServerLoop :: Eff Effects (Endpoint TestProtocol)-testServerLoop = Callback.start (Callback.callbacks handleReq "test-server-1")+testServerLoop = Callback.startLink (Callback.callbacks handleReq "test-server-1")  where   handleReq :: Endpoint TestProtocol -> Event TestProtocol -> Eff Effects ()   handleReq me (OnCall rt cm) =

@@ -13,6 +13,7 @@ import           Control.DeepSeq import qualified Data.Text as T import           Data.Type.Pretty+import           Data.Default  main :: IO () main = defaultMain (void counterExample)@@ -90,7 +91,7 @@  instance (LogIo q) => Server SupiCounter (Processes q) where -  type instance Model SupiCounter =+  newtype instance Model SupiCounter = SupiCounterModel     ( Integer     , ObserverRegistry CounterChanged     , Maybe (ReplyTarget SupiCounter (Maybe ()))@@ -98,39 +99,57 @@    data instance StartArgument SupiCounter (Processes q) = MkEmptySupiCounter -  setup _ _ = return ((0, emptyObserverRegistry, Nothing), ())+  setup _ _ = return (SupiCounterModel (0, emptyObserverRegistry, Nothing), ())    update _ _ = \case     OnCall rt callReq ->       case callReq of         ToPdu1 Cnt ->-          sendReply rt =<< useModel @SupiCounter _1+          sendReply rt =<< useModel supiCounter         ToPdu2 _ -> error "unreachable"         ToPdu3 (Whoopediedoo c) ->-          modifyModel @SupiCounter (_3 .~ if c then Just rt else Nothing)+          modifyModel @SupiCounter (supiTarget .~ if c then Just rt else Nothing)      OnCast castReq ->       case castReq of         ToPdu1 Inc -> do-          val' <- view _1 <$> modifyAndGetModel @SupiCounter (_1 %~ (+ 1))-          zoomModel @SupiCounter _2 (observerRegistryNotify (CounterChanged val'))+          val' <- view supiCounter <$> modifyAndGetModel (supiCounter %~ (+ 1))+          zoomModel supiRegistry (observerRegistryNotify (CounterChanged val'))           when (val' > 5) $-            getAndModifyModel @SupiCounter (_3 .~ Nothing)-            >>= traverse_ (\rt' -> sendReply rt' (Just ())) . view _3+            getAndModifyModel  (supiTarget .~ Nothing)+            >>= traverse_ (\rt' -> sendReply rt' (Just ())) . view supiTarget         ToPdu2 x ->-          zoomModel @SupiCounter _2 (observerRegistryHandlePdu x)+          zoomModel supiRegistry (observerRegistryHandlePdu x)         ToPdu3 _ -> error "unreachable"      OnDown pd -> do-      wasRemoved <- zoomModel @SupiCounter _2 (observerRegistryRemoveProcess @CounterChanged (downProcess pd))+      wasRemoved <- zoomModel supiRegistry (observerRegistryRemoveProcess @CounterChanged (downProcess pd))       if wasRemoved         then logDebug ("removed: "    <> T.pack (show pd))         else logError ("unexpected: " <> T.pack (show pd))      other -> logWarning (T.pack (show other)) +supiCounter :: Lens' (Model SupiCounter) Integer+supiCounter =+  lens+    (\(SupiCounterModel (x,_,_)) -> x)+    (\(SupiCounterModel (_,y,z)) x -> SupiCounterModel (x,y,z))++supiRegistry :: Lens' (Model SupiCounter) (ObserverRegistry CounterChanged)+supiRegistry =+  lens+    (\(SupiCounterModel (_,y,_)) -> y)+    (\(SupiCounterModel (x,_,z)) y -> SupiCounterModel (x,y,z))++supiTarget :: Lens' (Model SupiCounter) (Maybe (ReplyTarget SupiCounter (Maybe ())))+supiTarget =+  lens+    (\(SupiCounterModel (_,_,z)) -> z)+    (\(SupiCounterModel (x,y,_)) z -> SupiCounterModel (x,y,z))+ spawnCounter :: (LogIo q) => Eff (Processes q) ( Endpoint SupiCounter )-spawnCounter = start MkEmptySupiCounter+spawnCounter = startLink MkEmptySupiCounter   deriving instance Show (Pdu Counter x)@@ -138,10 +157,11 @@ logCounterObservations   :: (LogIo q, Typeable q)   => Eff (Processes q) (Endpoint (Observer CounterChanged))-logCounterObservations = start OCCStart+logCounterObservations = startLink OCCStart  instance Member Logs q => Server (Observer CounterChanged) (Processes q) where   data instance StartArgument (Observer CounterChanged) (Processes q) = OCCStart+  newtype instance Model (Observer CounterChanged) = CounterChangedModel () deriving Default   update _ _ e =     case e of       OnCast (Observed msg) -> logInfo' ("observerRegistryNotify: " ++ show msg)

@@ -1,8 +1,7 @@ {-# LANGUAGE QuantifiedConstraints #-} {-# LANGUAGE UndecidableInstances #-}-{-# LANGUAGE NumericUnderscores #-} -- | Another  example for the library that uses embedded protocols with multiple server back--- ends and a polymorphic client, as well as the 'Supervisor.Sup' module to start multiple+-- ends and a polymorphic client, as well as the 'Broker' module to start multiple -- back-ends. -- -- @since 0.29.0@@ -12,10 +11,9 @@ import           Control.Eff import           Control.Eff.State.Lazy as State import           Control.Eff.Concurrent-import           Control.Eff.Concurrent.Process.Timer import           Control.Eff.Concurrent.Protocol.EffectfulServer as Effectful import           Control.Eff.Concurrent.Protocol.StatefulServer as Stateful-import           Control.Eff.Concurrent.Protocol.Supervisor as Supervisor+import           Control.Eff.Concurrent.Protocol.Broker as Broker import           Control.Lens import           Control.Monad import           Data.Dynamic@@ -24,6 +22,7 @@ import           Data.String import qualified Data.Text as T import GHC.Stack (HasCallStack)+import Data.Default  main :: IO () main =@@ -31,13 +30,13 @@  embeddedExample :: HasCallStack => Eff Effects () embeddedExample = do-  b1 <- Stateful.start InitBackend1-  b2Sup <- startBackend2Sup-  b2 <- Supervisor.spawnOrLookup @Backend2 b2Sup 1-  _ <- Supervisor.spawnOrLookup b2Sup 2-  b2_2 <- Supervisor.spawnOrLookup b2Sup 2-  b2_3 <- Supervisor.spawnOrLookup b2Sup 3-  app <- Stateful.start InitApp+  b1 <- Stateful.startLink InitBackend1+  b2Broker <- startBackend2Broker+  b2 <- Broker.spawnOrLookup @Backend2 b2Broker 1+  _ <- Broker.spawnOrLookup b2Broker 2+  b2_2 <- Broker.spawnOrLookup b2Broker 2+  b2_3 <- Broker.spawnOrLookup b2Broker 3+  app <- Stateful.startLink InitApp   cast app DoThis   cast app DoThis   cast app DoThis@@ -65,11 +64,11 @@   cast app DoThis   call app (SetBackend (Just (SomeBackend b2_2)))   cast app DoThis-  _ <- Supervisor.stopChild b2Sup 2+  _ <- Broker.stopChild b2Broker 2   cast app DoThis   call app (SetBackend (Just (SomeBackend b2_3)))   cast app DoThis-  Supervisor.stopSupervisor b2Sup+  Broker.stopBroker b2Broker   cast app DoThis  @@ -78,20 +77,20 @@ -- Application layer  instance Stateful.Server App Effects where-  type instance Model App = Maybe SomeBackend+  newtype instance Model App = MkApp (Maybe SomeBackend) deriving Default   data instance StartArgument App Effects = InitApp   update me _x e =     case e of       OnCall rt (SetBackend b) -> do         logInfo "setting backend"-        oldB <- getAndPutModel @App b+        MkApp oldB <- getAndPutModel @App (MkApp b)         traverse_ (`backendForgetObserver` me) oldB         traverse_ (`backendRegisterObserver` me) b         sendReply rt ()       OnCast (AppBackendEvent be) ->         logInfo ("got backend event: " <> T.pack (show be))       OnCast DoThis ->-        do m <- getModel @App+        do MkApp m <- getModel @App            case m of             Nothing -> logInfo "doing this without backend"             Just b -> handleInterrupts (logWarning . T.pack . show) $ do@@ -203,31 +202,34 @@  instance Stateful.Server Backend1 Effects where   type instance Protocol Backend1 = (Backend, ObserverRegistry BackendEvent)-  type instance Model Backend1 = (Int, ObserverRegistry BackendEvent)+  newtype instance Model Backend1 = MkBackend1 (Int, ObserverRegistry BackendEvent)   data instance StartArgument Backend1 Effects = InitBackend1-  setup _ _ = pure ( (0, emptyObserverRegistry), () )+  setup _ _ = pure ( MkBackend1 (0, emptyObserverRegistry), () )   update me _ e = do     model <- getModel @Backend1     case e of       OnCall rt (ToPduLeft GetBackendInfo) ->         sendReply           (toEmbeddedReplyTarget @(Stateful.Protocol Backend1) @Backend rt)-          ("Backend1 " <> show me <> " " <> show (model ^. _1))+          ("Backend1 " <> show me <> " " <> show (model ^. modelBackend1 . _1))       OnCast (ToPduLeft BackendWork) -> do         logInfo "working..."-        modifyModel @Backend1 (over _1 (+ 1))+        modifyModel @Backend1 (over (modelBackend1 . _1) (+ 1))       OnCast (ToPduRight x) -> do         logInfo "event registration stuff ..."-        zoomModel @Backend1 _2 (observerRegistryHandlePdu x)+        zoomModel @Backend1 (modelBackend1 . _2) (observerRegistryHandlePdu x)       OnDown pd -> do         logWarning (T.pack (show pd))-        wasObserver <- zoomModel @Backend1 _2 (observerRegistryRemoveProcess @BackendEvent (downProcess pd))+        wasObserver <- zoomModel @Backend1 (modelBackend1 . _2) (observerRegistryRemoveProcess @BackendEvent (downProcess pd))         when wasObserver $           logNotice "observer removed"       _ -> logWarning ("unexpected: " <> T.pack (show e)) --- Backend 2 is behind a supervisor+modelBackend1 :: Iso' (Model Backend1)  (Int, ObserverRegistry BackendEvent)+modelBackend1 = iso (\(MkBackend1 x) -> x) MkBackend1 +-- Backend 2 is behind a broker+ data Backend2 deriving Typeable  instance HasPdu Backend2 where@@ -286,10 +288,10 @@           logNotice "observer removed"       _ -> logWarning ("unexpected: " <> T.pack (show e)) -type instance Supervisor.ChildId Backend2 = Int+type instance Broker.ChildId Backend2 = Int -startBackend2Sup :: Eff Effects (Endpoint (Supervisor.Sup Backend2))-startBackend2Sup = Supervisor.startSupervisor (Supervisor.MkSupConfig (TimeoutMicros 1_000_000) InitBackend2)+startBackend2Broker :: Eff Effects (Endpoint (Broker.Broker Backend2))+startBackend2Broker = Broker.startLink (Broker.MkBrokerConfig (TimeoutMicros 1_000_000) InitBackend2)  -- EXPERIMENTING data EP a where

@@ -1,6 +1,6 @@ cabal-version:  2.0 name:           extensible-effects-concurrent-version:        0.29.2+version:        0.30.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@@ -40,10 +40,10 @@   build-depends:       async >= 2.2 && <3,       base >= 4.12 && <5,-      bytestring >= 0.10 && < 0.11,       data-default >= 0.7 && < 0.8,       deepseq >= 1.4 && < 1.5,       directory,+      hashable >= 1.2,       hostname,       exceptions >= 0.10 && < 0.11,       safe-exceptions >= 0.1 && < 0.2,@@ -51,14 +51,10 @@       time >= 1.8 && < 1.9,       mtl >= 2.2 && < 2.3,       containers >=0.5.8 && <0.7,-      QuickCheck,       lens >= 4.14 && < 4.18,-      parallel >= 3.2 && < 3.3,-      process >= 1.6 && < 1.7,       monad-control >= 1.0 && < 1.1,       extensible-effects >= 5 && < 6,       stm >= 2.4.5 && <2.6,-      tagged >= 0.8 && < 0.9,       transformers-base >= 0.4 && < 0.5,       text >= 1.2 && < 1.3,       network >= 2 && < 4,@@ -82,25 +78,26 @@                   Control.Eff.LogWriter.UDP,                   Control.Eff.LogWriter.UnixSocket,                   Control.Eff.Concurrent,-                  Control.Eff.Concurrent.Pure,-                  Control.Eff.Concurrent.SingleThreaded,+                  Control.Eff.Concurrent.Misc,                   Control.Eff.Concurrent.Protocol,                   Control.Eff.Concurrent.Protocol.CallbackServer,                   Control.Eff.Concurrent.Protocol.Client,                   Control.Eff.Concurrent.Protocol.EffectfulServer,                   Control.Eff.Concurrent.Protocol.StatefulServer,-                  Control.Eff.Concurrent.Protocol.Supervisor,+                  Control.Eff.Concurrent.Protocol.Broker,+                  Control.Eff.Concurrent.Protocol.Watchdog,                   Control.Eff.Concurrent.Protocol.Wrapper,                   Control.Eff.Concurrent.Process,                   Control.Eff.Concurrent.Process.Timer,                   Control.Eff.Concurrent.Process.ForkIOScheduler,                   Control.Eff.Concurrent.Process.Interactive,                   Control.Eff.Concurrent.Process.SingleThreadedScheduler,-                  Control.Eff.Concurrent.Protocol.Observer-                  Control.Eff.Concurrent.Protocol.Observer.Queue+                  Control.Eff.Concurrent.Protocol.Observer,+                  Control.Eff.Concurrent.Protocol.Observer.Queue,+                  Control.Eff.Concurrent.Pure,+                  Control.Eff.Concurrent.SingleThreaded   other-modules:-                Control.Eff.Concurrent.Misc,-                Control.Eff.Concurrent.Protocol.Supervisor.InternalState,+                Control.Eff.Concurrent.Protocol.Broker.InternalState,                 Paths_extensible_effects_concurrent   default-extensions:                      AllowAmbiguousTypes,@@ -120,6 +117,7 @@                      LambdaCase,                      OverloadedStrings,                      MultiParamTypeClasses,+                     NumericUnderscores,                      RankNTypes,                      ScopedTypeVariables,                      StandaloneDeriving,@@ -140,10 +138,8 @@   hs-source-dirs: examples/example-1   build-depends:                 base-              , data-default               , extensible-effects-concurrent               , extensible-effects-              , lens               , text               , deepseq               , pretty-types >= 0.2.3.1 && < 0.4@@ -159,6 +155,7 @@                       , GADTs                       , GeneralizedNewtypeDeriving                       , LambdaCase+                      , NumericUnderscores                       , OverloadedStrings                       , RankNTypes                       , ScopedTypeVariables@@ -190,6 +187,7 @@                       , FunctionalDependencies                       , GADTs                       , GeneralizedNewtypeDeriving+                      , NumericUnderscores                       , LambdaCase                       , RankNTypes                       , ScopedTypeVariables@@ -205,14 +203,9 @@   hs-source-dirs: examples/example-3   build-depends:                 base-              , data-default-              , directory               , extensible-effects-concurrent               , extensible-effects-              , filepath               , lens-              , text-              , pretty-types >= 0.2.3.1 && < 0.4   ghc-options: -Wall -threaded -fno-full-laziness   default-language: Haskell2010   default-extensions:   AllowAmbiguousTypes@@ -224,6 +217,7 @@                       , GADTs                       , GeneralizedNewtypeDeriving                       , LambdaCase+                      , NumericUnderscores                       , OverloadedStrings                       , RankNTypes                       , ScopedTypeVariables@@ -240,10 +234,6 @@                 base               , extensible-effects-concurrent               , extensible-effects-              , deepseq-              , text-              , deepseq-              , pretty-types >= 0.2.3.1 && < 0.4   ghc-options: -Wall -threaded -fno-full-laziness   default-language: Haskell2010   default-extensions:   AllowAmbiguousTypes@@ -255,6 +245,7 @@                       , GADTs                       , GeneralizedNewtypeDeriving                       , LambdaCase+                      , NumericUnderscores                       , OverloadedStrings                       , RankNTypes                       , ScopedTypeVariables@@ -271,10 +262,10 @@   build-depends:                 base               , deepseq+              , data-default               , extensible-effects-concurrent               , extensible-effects               , lens-              , pretty-types >= 0.2.3.1 && < 0.4               , text   ghc-options: -Wall -threaded -fno-full-laziness   default-language: Haskell2010@@ -288,6 +279,7 @@                       , GADTs                       , GeneralizedNewtypeDeriving                       , LambdaCase+                      , NumericUnderscores                       , OverloadedStrings                       , RankNTypes                       , ScopedTypeVariables@@ -312,9 +304,10 @@               , LoopTests               , ObserverTests               , ProcessBehaviourTestCases-              , SupervisorTests+              , BrokerTests               , SingleThreadedScheduler-  ghc-options: -Wall -threaded -fno-full-laziness+              , WatchdogTests+  ghc-options: -Wall -threaded -fno-full-laziness -Wno-missing-signatures   build-depends:                 async               , base@@ -347,6 +340,7 @@                       , GADTs                       , GeneralizedNewtypeDeriving                       , LambdaCase+                      , NumericUnderscores                       , OverloadedStrings                       , RankNTypes                       , ScopedTypeVariables

@@ -81,6 +81,7 @@                                                 , fromProcessTitle                                                 , ProcessDetails(..)                                                 , fromProcessDetails+                                                , Timeout(TimeoutMicros, fromTimeoutMicros)                                                 , StrictDynamic()                                                 , toStrictDynamic                                                 , fromStrictDynamic@@ -89,6 +90,7 @@                                                 , ProcessId(..)                                                 , fromProcessId                                                 , yieldProcess+                                                , delay                                                 , sendMessage                                                 , sendAnyMessage                                                 , makeReference@@ -162,16 +164,20 @@                                                 , receiverPid                                                 ) import           Control.Eff.Concurrent.Process.Timer-                                                ( Timeout(fromTimeoutMicros)-                                                , TimerReference()+                                                ( TimerReference()                                                 , TimerElapsed(fromTimerElapsed)                                                 , sendAfter                                                 , startTimer+                                                , sendAfterWithTitle+                                                , startTimerWithTitle                                                 , selectTimerElapsed                                                 , cancelTimer                                                 , receiveAfter                                                 , receiveSelectedAfter                                                 , receiveSelectedWithMonitorAfter+                                                , receiveAfterWithTitle+                                                , receiveSelectedAfterWithTitle+                                                , receiveSelectedWithMonitorAfterWithTitle                                                 )  import           Control.Eff.Concurrent.Protocol

@@ -36,14 +36,17 @@ -- -- @since 0.24.0 showSTypeRepPrec :: Int -> SomeTypeRep -> ShowS-showSTypeRepPrec d (SomeTypeRep tr) sIn =-  let (con, conArgs) = splitApps tr-   in case conArgs of-        [] -> showString (tyConName con) sIn-        _ ->-          showParen-            (d >= 10)-            (showString (tyConName con) . showChar ':' .-              foldr1 (\f acc -> showChar '-' . f . acc)-                     (showSTypeRepPrec 10 <$> conArgs))-            sIn+showSTypeRepPrec _d (SomeTypeRep tr) =+  let+      (con, conArgs) = splitApps tr++      renderArgs =+        foldr1 (\f acc -> showString ", " . f . acc)+               (showSTypeRepPrec 10 <$> conArgs)++   in  showString (tyConName con) .+          if not (null conArgs) then+                showChar '<'+              . renderArgs+              . showChar '>'+          else id

@@ -17,7 +17,7 @@     -- ** Process Effect Aliases   , SafeProcesses   , Processes-    -- ** Process Effect Contraints+    -- ** Process Effect Constraints   , HasProcesses   , HasSafeProcesses       -- ** Process Info@@ -25,6 +25,10 @@   , fromProcessTitle   , ProcessDetails(..)   , fromProcessDetails++      -- ** Process Timer+  , Timeout(TimeoutMicros, fromTimeoutMicros)+   , -- ** Message Data     StrictDynamic()   , toStrictDynamic@@ -39,6 +43,8 @@   , ProcessState(..)   -- ** Yielding   , yieldProcess+  -- ** Waiting/Sleeping+  , delay   -- ** Sending Messages   , sendMessage   , sendAnyMessage@@ -79,7 +85,7 @@   -- *** Interrupting Processes   , interrupt   , sendInterrupt-  -- *** Exitting Processes+  -- *** Exiting Processes   , exitBecause   , exitNormally   , exitWithError@@ -188,6 +194,11 @@   --   -- @since 0.12.0   YieldProcess :: Process r (ResumeProcess ())+  -- | Simply wait until the time in the given 'Timeout' has elapsed and return.+  --+  -- @since 0.30.0+  Delay :: Timeout -> Process r (ResumeProcess ())+   -- | Return the current 'ProcessId'   SelfPid :: Process r (ResumeProcess ProcessId)   -- | Start a new process, the new process will execute an effect, the function@@ -236,10 +247,19 @@   -- if one of the processes crashes (i.e. 'isCrash' returns 'True'), the other   -- is shutdown with the 'Interrupt' 'LinkedProcessCrashed'.   --+  -- You might wonder: Why not tearing down the linked process when exiting+  -- normally?+  -- I thought about this. If a process exits normally, it should have the+  -- opportunity to shutdown stuff explicitly.+  -- And if you want to make sure that there are no dangling child processes+  -- after e.g. a broker crash, you can always use 'monitor'.+  --   -- @since 0.12.0   Link :: ProcessId -> Process r (ResumeProcess ())   -- | Unlink the calling process from the other process.   --+  -- See 'Link' for a discussion on linking.+  --   -- @since 0.12.0   Unlink :: ProcessId -> Process r (ResumeProcess ()) @@ -254,6 +274,7 @@   showsPrec d = \case     FlushMessages -> showString "flush messages"     YieldProcess  -> showString "yield process"+    Delay t       -> showString "delay process until: " . shows t     SelfPid       -> showString "lookup the current process id"     Spawn   t _   -> showString "spawn a new process: " . shows t     SpawnLink t _ -> showString "spawn a new process and link to it" . shows t@@ -322,7 +343,16 @@ instance Show ProcessDetails where   showsPrec _ (MkProcessDetails t) = showString (T.unpack t) +-- | A number of micro seconds.+--+-- @since 0.12.0+newtype Timeout = TimeoutMicros {fromTimeoutMicros :: Int}+  deriving (NFData, Ord,Eq, Num, Integral, Real, Enum, Typeable) +instance Show Timeout where+  showsPrec d (TimeoutMicros t) =+    showParen (d >= 10) (showString "timeout: " . shows t . showString " µs")+ -- | Data flows between 'Process'es via these messages. -- -- This is just a newtype wrapper around 'Dynamic'.@@ -468,7 +498,8 @@     ProcessBooting              -- ^ The process has just been started but not                                 --   scheduled yet.   | ProcessIdle                 -- ^ The process yielded it's time slice-  | ProcessBusy                 -- ^ The process is busy with non-blocking+  | ProcessBusy                 -- ^ The process is busy with a non-blocking operation+  | ProcessBusySleeping         -- ^ The process is sleeping until the 'Timeout' given to 'Delay'   | ProcessBusyUpdatingDetails  -- ^ The process is busy with 'UpdateProcessDetails'   | ProcessBusySending          -- ^ The process is busy with sending a message   | ProcessBusySendingShutdown  -- ^ The process is busy with killing@@ -504,15 +535,20 @@  -- | Get the 'ExitRecovery' toExitRecovery :: Interrupt r -> ExitRecovery-toExitRecovery = \case-  NormalExitRequested           -> Recoverable-  (OtherProcessNotRunning    _)  -> Recoverable-  (TimeoutInterrupt       _)  -> Recoverable-  (LinkedProcessCrashed _)  -> Recoverable-  (ErrorInterrupt         _)  -> Recoverable-  ExitNormally              -> NoRecovery-  (ExitUnhandledError _) -> NoRecovery-  ExitProcessCancelled                    -> NoRecovery+toExitRecovery =+  \case+    NormalExitRequested -> Recoverable+    (NormalExitRequestedWith _) -> Recoverable+    (OtherProcessNotRunning _) -> Recoverable+    (TimeoutInterrupt _) -> Recoverable+    (LinkedProcessCrashed _) -> Recoverable+    (InterruptedBy _) -> Recoverable+    (ErrorInterrupt _) -> Recoverable+    ExitNormally -> NoRecovery+    (ExitNormallyWith _) -> NoRecovery+    (ExitUnhandledError _) -> NoRecovery+    (ExitProcessCancelled _) -> NoRecovery+    (ExitOtherProcessNotRunning _) -> NoRecovery  -- | This value indicates whether a process exited in way consistent with -- the planned behaviour or not.@@ -551,21 +587,36 @@     -- @since 0.13.2     NormalExitRequested       :: Interrupt 'Recoverable+    -- | Extension of 'ExitNormally' with a custom reason+    --+    -- @since 0.30.0+    NormalExitRequestedWith+      :: forall a . (Typeable a, Show a, NFData a) => a -> Interrupt 'Recoverable     -- | A process that should be running was not running.     OtherProcessNotRunning       :: ProcessId -> Interrupt 'Recoverable     -- | A 'Recoverable' timeout has occurred.     TimeoutInterrupt       :: String -> Interrupt 'Recoverable-    -- | A linked process is down+    -- | A linked process is down, see 'Link' for a discussion on linking.     LinkedProcessCrashed       :: ProcessId -> Interrupt 'Recoverable     -- | An exit reason that has an error message and is 'Recoverable'.     ErrorInterrupt       :: String -> Interrupt 'Recoverable+    -- | An interrupt with a custom message.+    --+    -- @since 0.30.0+    InterruptedBy+      :: forall a . (Typeable a, Show a, NFData a) => a -> Interrupt 'Recoverable     -- | A process function returned or exited without any error.     ExitNormally       :: Interrupt 'NoRecovery+    -- | A process function returned or exited without any error, and with a custom message+    --+    -- @since 0.30.0+    ExitNormallyWith+      :: forall a . (Typeable a, Show a, NFData a) => a -> Interrupt 'NoRecovery     -- | An error causes the process to exit immediately.     -- For example an unexpected runtime exception was thrown, i.e. an exception     -- derived from 'Control.Exception.Safe.SomeException'@@ -574,7 +625,10 @@       :: Text -> Interrupt 'NoRecovery     -- | A process shall exit immediately, without any cleanup was cancelled (e.g. killed, in 'Async.cancel')     ExitProcessCancelled-      :: Interrupt 'NoRecovery+      :: Maybe ProcessId -> Interrupt 'NoRecovery+    -- | A process that is vital to the crashed process was not running+    ExitOtherProcessNotRunning+      :: ProcessId -> Interrupt 'NoRecovery   deriving Typeable  -- | Return either 'ExitNormally' or 'interruptToExit' from a 'Recoverable' 'Interrupt';@@ -582,81 +636,106 @@ -- If the 'Interrupt' is 'NormalExitRequested' then return 'ExitNormally' interruptToExit :: Interrupt 'Recoverable -> Interrupt 'NoRecovery interruptToExit NormalExitRequested = ExitNormally+interruptToExit (NormalExitRequestedWith x) = (ExitNormallyWith x) interruptToExit x = ExitUnhandledError (pack (show x))  instance Show (Interrupt x) where   showsPrec d =-    showParen (d >= 9)-      . (\case-          NormalExitRequested        -> showString "interrupt: A normal exit was requested"-          OtherProcessNotRunning p    -> showString "interrupt: Another process is not running: " . showsPrec 10 p-          TimeoutInterrupt reason -> showString "interrupt: A timeout occured: " . showString reason-          LinkedProcessCrashed m ->-            showString "interrupt: A linked process " . showsPrec 10 m . showString " crashed"-          ErrorInterrupt reason   -> showString "interrupt: An error occured: " . showString reason-          ExitNormally          -> showString "exit: Process finished successfully"-          ExitUnhandledError w ->-            showString "exit: Unhandled " . showString (unpack w)-          ExitProcessCancelled -> showString "exit: The process was cancelled"-        )+    showParen (d >= 10) .+    (\case+       NormalExitRequested -> showString "interrupt: A normal exit was requested"+       NormalExitRequestedWith p -> showString "interrupt: A normal exit was requested: " . showsPrec 10 p+       OtherProcessNotRunning p -> showString "interrupt: Another process is not running: " . showsPrec 10 p+       TimeoutInterrupt reason -> showString "interrupt: A timeout occured: " . showString reason+       LinkedProcessCrashed m -> showString "interrupt: A linked process " . showsPrec 10 m . showString " crashed"+       InterruptedBy reason -> showString "interrupt: " . showsPrec 10 reason+       ErrorInterrupt reason -> showString "interrupt: An error occured: " . showString reason+       ExitNormally -> showString "exit: Process finished successfully"+       ExitNormallyWith reason -> showString "exit: Process finished successfully: " . showsPrec 10 reason+       ExitUnhandledError w -> showString "exit: Unhandled " . showString (unpack w)+       ExitProcessCancelled Nothing -> showString "exit: The process was cancelled by a runtime exception"+       ExitProcessCancelled (Just origin) -> showString "exit: The process was cancelled by: " . shows origin+       ExitOtherProcessNotRunning p -> showString "exit: Another process is not running: " . showsPrec 10 p)  instance Exc.Exception (Interrupt 'Recoverable) instance Exc.Exception (Interrupt 'NoRecovery )  instance NFData (Interrupt x) where-  rnf NormalExitRequested               = rnf ()-  rnf (OtherProcessNotRunning    !l)     = rnf l+  rnf NormalExitRequested             = rnf ()+  rnf (NormalExitRequestedWith   !l)  = rnf l+  rnf (OtherProcessNotRunning    !l)  = rnf l   rnf (TimeoutInterrupt       !l)     = rnf l-  rnf (LinkedProcessCrashed !l)     = rnf l+  rnf (LinkedProcessCrashed !l)       = rnf l   rnf (ErrorInterrupt         !l)     = rnf l-  rnf ExitNormally                  = rnf ()-  rnf (ExitUnhandledError !l) = rnf l-  rnf ExitProcessCancelled                        = rnf ()+  rnf (InterruptedBy         !l)    = rnf l+  rnf ExitNormally                    = rnf ()+  rnf (ExitNormallyWith !l)           = rnf l+  rnf (ExitUnhandledError !l)         = rnf l+  rnf (ExitProcessCancelled  !o)      = rnf o+  rnf (ExitOtherProcessNotRunning !l) = rnf l  instance Ord (Interrupt x) where-  compare NormalExitRequested          NormalExitRequested         = EQ-  compare NormalExitRequested          _                           = LT-  compare _                        NormalExitRequested             = GT-  compare (OtherProcessNotRunning l)    (OtherProcessNotRunning r) = compare l r-  compare (OtherProcessNotRunning _)    _                          = LT-  compare _                        (OtherProcessNotRunning    _)   = GT-  compare (TimeoutInterrupt l)       (TimeoutInterrupt r)          = compare l r-  compare (TimeoutInterrupt _) _                                   = LT-  compare _                        (TimeoutInterrupt _)            = GT-  compare (LinkedProcessCrashed l) (LinkedProcessCrashed r)        = compare l r-  compare (LinkedProcessCrashed _) _                               = LT-  compare _                        (LinkedProcessCrashed _)        = GT-  compare (ErrorInterrupt l)         (ErrorInterrupt         r)    = compare l r-  compare ExitNormally             ExitNormally                    = EQ-  compare ExitNormally             _                               = LT-  compare _                        ExitNormally                    = GT-  compare (ExitUnhandledError l) (ExitUnhandledError r)            = compare l r-  compare (ExitUnhandledError _ ) _                                = LT-  compare _                         (ExitUnhandledError _)         = GT-  compare ExitProcessCancelled  ExitProcessCancelled               = EQ+  compare NormalExitRequested          NormalExitRequested               = EQ+  compare NormalExitRequested          _                                 = LT+  compare _                        NormalExitRequested                   = GT+  compare (NormalExitRequestedWith _)    (NormalExitRequestedWith _)     = EQ+  compare (NormalExitRequestedWith _)    _                               = LT+  compare _                        (NormalExitRequestedWith    _)        = GT+  compare (OtherProcessNotRunning l)    (OtherProcessNotRunning r)       = compare l r+  compare (OtherProcessNotRunning _)    _                                = LT+  compare _                        (OtherProcessNotRunning    _)         = GT+  compare (TimeoutInterrupt l)       (TimeoutInterrupt r)                = compare l r+  compare (TimeoutInterrupt _) _                                         = LT+  compare _                        (TimeoutInterrupt _)                  = GT+  compare (LinkedProcessCrashed l) (LinkedProcessCrashed r)              = compare l r+  compare (LinkedProcessCrashed _) _                                     = LT+  compare _                        (LinkedProcessCrashed _)              = GT+  compare (InterruptedBy _) (InterruptedBy _)                        = EQ+  compare (InterruptedBy _) _                                          = LT+  compare _                        (InterruptedBy _)                   = GT+  compare (ErrorInterrupt l)         (ErrorInterrupt         r)          = compare l r+  compare ExitNormally             ExitNormally                          = EQ+  compare ExitNormally             _                                     = LT+  compare _                        ExitNormally                          = GT+  compare (ExitNormallyWith _) (ExitNormallyWith _)                      = EQ+  compare (ExitNormallyWith _ ) _                                        = LT+  compare _                         (ExitNormallyWith _)                 = GT+  compare (ExitUnhandledError l) (ExitUnhandledError r)                  = compare l r+  compare (ExitUnhandledError _ ) _                                      = LT+  compare _                         (ExitUnhandledError _)               = GT+  compare (ExitOtherProcessNotRunning l)  (ExitOtherProcessNotRunning r) = compare l r+  compare (ExitOtherProcessNotRunning _ ) _                              = LT+  compare _                         (ExitOtherProcessNotRunning _)       = GT+  compare (ExitProcessCancelled l)  (ExitProcessCancelled r)             = compare l r  instance Eq (Interrupt x) where-  (==) NormalExitRequested          NormalExitRequested          = True-  (==) (OtherProcessNotRunning l)    (OtherProcessNotRunning r)    = (==) l r-  (==) ExitNormally             ExitNormally             = True-  (==) (TimeoutInterrupt l)       (TimeoutInterrupt r)       = l == r+  (==) NormalExitRequested NormalExitRequested = True+  (==) (OtherProcessNotRunning l) (OtherProcessNotRunning r) = (==) l r+  (==) ExitNormally ExitNormally = True+  (==) (TimeoutInterrupt l) (TimeoutInterrupt r) = l == r   (==) (LinkedProcessCrashed l) (LinkedProcessCrashed r) = l == r-  (==) (ErrorInterrupt         l) (ErrorInterrupt         r) = (==) l r+  (==) (ErrorInterrupt l) (ErrorInterrupt r) = (==) l r   (==) (ExitUnhandledError l) (ExitUnhandledError r) = (==) l r-  (==) ExitProcessCancelled ExitProcessCancelled = True-  (==) _      _      = False+  (==) (ExitOtherProcessNotRunning l) (ExitOtherProcessNotRunning r) = (==) l r+  (==) (ExitProcessCancelled l) (ExitProcessCancelled r) = l == r+  (==) _ _ = False  -- | A predicate for linked process __crashes__. isProcessDownInterrupt :: Maybe ProcessId -> Interrupt r -> Bool-isProcessDownInterrupt mOtherProcess = \case-  NormalExitRequested         -> False-  OtherProcessNotRunning    _  -> False-  TimeoutInterrupt       _  -> False-  LinkedProcessCrashed p  -> maybe True (== p) mOtherProcess-  ErrorInterrupt         _  -> False-  ExitNormally            -> False-  ExitUnhandledError _ -> False-  ExitProcessCancelled -> False+isProcessDownInterrupt mOtherProcess =+  \case+    NormalExitRequested -> False+    NormalExitRequestedWith _ -> False+    OtherProcessNotRunning _ -> False+    TimeoutInterrupt _ -> False+    LinkedProcessCrashed p -> maybe True (== p) mOtherProcess+    InterruptedBy _ -> False+    ErrorInterrupt _ -> False+    ExitNormally -> False+    ExitNormallyWith _ -> False+    ExitUnhandledError _ -> False+    ExitProcessCancelled _ -> False+    ExitOtherProcessNotRunning _ -> False  -- | 'Interrupt's which are 'Recoverable'. type RecoverableInterrupt = Interrupt 'Recoverable@@ -669,7 +748,7 @@ -- This constrains the effect list to look like this: -- @[e1 ... eN, 'Interrupts', 'Process' [e(N+1) .. e(N+k)], e(N+1) .. e(N+k)]@ ----- It contrains @e@ beyond 'HasSafeProcesses' to encompass 'Interrupts'.+-- It constrains @e@ beyond 'HasSafeProcesses' to encompass 'Interrupts'. -- -- @since 0.27.1 type HasProcesses e inner = (HasSafeProcesses e inner, Member Interrupts e)@@ -683,7 +762,7 @@ -- This constrains the effect list to look like this: -- @[e1 ... eN, 'Process' [e(N+1) .. e(N+k)], e(N+1) .. e(N+k)]@ ----- It contrains @e@ to support the (only) 'Process' effect.+-- It constrains @e@ to support the (only) 'Process' effect. -- -- This is more relaxed that 'HasProcesses' since it does not require 'Interrupts'. --@@ -792,17 +871,21 @@  -- | Partition a 'SomeExitReason' back into either a 'NoRecovery' -- or a 'Recoverable' 'Interrupt'-fromSomeExitReason-  :: SomeExitReason -> Either (Interrupt 'NoRecovery) (Interrupt 'Recoverable)-fromSomeExitReason (SomeExitReason e) = case e of-  recoverable@NormalExitRequested          -> Right recoverable-  recoverable@(OtherProcessNotRunning    _) -> Right recoverable-  recoverable@(TimeoutInterrupt       _) -> Right recoverable-  recoverable@(LinkedProcessCrashed _) -> Right recoverable-  recoverable@(ErrorInterrupt         _) -> Right recoverable-  noRecovery@ExitNormally              -> Left noRecovery-  noRecovery@(ExitUnhandledError _) -> Left noRecovery-  noRecovery@ExitProcessCancelled -> Left noRecovery+fromSomeExitReason :: SomeExitReason -> Either (Interrupt 'NoRecovery) (Interrupt 'Recoverable)+fromSomeExitReason (SomeExitReason e) =+  case e of+    recoverable@NormalExitRequested -> Right recoverable+    recoverable@(NormalExitRequestedWith _) -> Right recoverable+    recoverable@(OtherProcessNotRunning _) -> Right recoverable+    recoverable@(TimeoutInterrupt _) -> Right recoverable+    recoverable@(LinkedProcessCrashed _) -> Right recoverable+    recoverable@(InterruptedBy _) -> Right recoverable+    recoverable@(ErrorInterrupt _) -> Right recoverable+    noRecovery@ExitNormally -> Left noRecovery+    noRecovery@(ExitNormallyWith _) -> Left noRecovery+    noRecovery@(ExitUnhandledError _) -> Left noRecovery+    noRecovery@(ExitProcessCancelled _) -> Left noRecovery+    noRecovery@(ExitOtherProcessNotRunning _) -> Left noRecovery  -- | Print a 'Interrupt' to 'Just' a formatted 'String' when 'isCrash' -- is 'True'.@@ -823,7 +906,7 @@ -- | Log the 'Interrupt's logProcessExit   :: forall e x . (Member Logs e, HasCallStack) => Interrupt x -> Eff e ()-logProcessExit (toCrashReason -> Just ex) = withFrozenCallStack (logError ex)+logProcessExit (toCrashReason -> Just ex) = withFrozenCallStack (logWarning ex) logProcessExit ex = withFrozenCallStack (logDebug (fromString (show ex)))  @@ -880,12 +963,25 @@   => Eff r () yieldProcess = executeAndResumeOrThrow YieldProcess ++-- | Simply block until the time in the 'Timeout' has passed.+--+-- @since 0.30.0+delay+  :: forall r q+   . ( HasProcesses r q+     , HasCallStack+     )+  => Timeout+  -> Eff r ()+delay = executeAndResumeOrThrow . Delay+ -- | Send a message to a process addressed by the 'ProcessId'. -- See 'SendMessage'. -- -- The message will be reduced to normal form ('rnf') by/in the caller process. sendMessage-  :: forall r q o+  :: forall o r q    . ( HasProcesses r q      , HasCallStack      , Typeable o@@ -958,6 +1054,8 @@  -- | Start a new process, and immediately link to it. --+-- See 'Link' for a discussion on linking.+-- -- @since 0.12.0 spawnLink   :: forall r q@@ -1199,7 +1297,7 @@ data ProcessDown =   ProcessDown     { downReference :: !MonitorReference-    , downReason    :: !SomeExitReason+    , downReason    :: !(Interrupt 'NoRecovery)     , downProcess   :: !ProcessId     }   deriving (Typeable, Generic, Eq, Ord)@@ -1251,6 +1349,8 @@ -- if one of the processes crashes (i.e. 'isCrash' returns 'True'), the other -- is shutdown with the 'Interrupt' 'LinkedProcessCrashed'. --+-- See 'Link' for a discussion on linking.+-- -- @since 0.12.0 linkProcess   :: forall r q@@ -1260,6 +1360,8 @@ linkProcess = executeAndResumeOrThrow . Link . force  -- | Unlink the calling process from the other process.+--+-- See 'Link' for a discussion on linking. -- -- @since 0.12.0 unlinkProcess

@@ -21,7 +21,7 @@   ) where -import           Control.Concurrent             ( yield )+import           Control.Concurrent             ( yield, threadDelay, forkIO, killThread ) import qualified Control.Concurrent.Async      as Async import           Control.Concurrent.Async       ( Async(..) ) import           Control.Concurrent.STM        as STM@@ -38,7 +38,7 @@ import           Control.Exception.Safe        as Safe import           Control.Lens import           Control.Monad                  ( void-                                                , when+                                                , when, unless                                                 ) import           Control.Monad.Trans.Control    ( MonadBaseControl(..)                                                 , control@@ -57,7 +57,6 @@ import qualified Data.Text                     as T import           GHC.Stack import           System.Timeout-import           Text.Printf  -- * Process Types @@ -128,46 +127,74 @@   traverse_ logDebug pm   logDebug $ "nextMonitorIndex: " <> nm +-- | Allocate a new 'MonitorReference'+nextMonitorReference :: ProcessId -> SchedulerState -> STM MonitorReference+nextMonitorReference target schedulerState = do+  aNewMonitorIndex <- readTVar (schedulerState ^. nextMonitorIndex)+  modifyTVar' (schedulerState ^. nextMonitorIndex) (+ 1)+  return (MonitorReference aNewMonitorIndex target)+ -- | Add monitor: If the process is dead, enqueue a 'ProcessDown' message into the -- owners message queue addMonitoring-  :: HasCallStack => ProcessId -> ProcessId -> SchedulerState -> STM MonitorReference-addMonitoring target owner schedulerState = do-  aNewMonitorIndex <- readTVar (schedulerState ^. nextMonitorIndex)-  modifyTVar' (schedulerState ^. nextMonitorIndex) (+ 1)-  let monitorRef = MonitorReference aNewMonitorIndex target-  when (target /= owner) $ do-    pt <- readTVar (schedulerState ^. processTable)-    if Map.member target pt-      then modifyTVar' (schedulerState ^. processMonitors)-                       (Set.insert (monitorRef, owner))-      else-        let processDownMessage =-              ProcessDown monitorRef (SomeExitReason (OtherProcessNotRunning target)) target-        in  enqueueMessageOtherProcess owner-                                       (toStrictDynamic processDownMessage)-                                       schedulerState-  return monitorRef+  :: HasCallStack => MonitorReference -> ProcessId -> SchedulerState -> STM Int+addMonitoring monitorRef@(MonitorReference _ target) owner schedulerState =+  if target == owner then pure 0+    else+      do+        pt <- readTVar (schedulerState ^. processTable)+        case Map.lookup target pt of+          Just targetProcessInfo ->+            do (_, targetState) <- readTVar (targetProcessInfo ^. processState)+               check (   targetState == ProcessShuttingDown+                      || targetState == ProcessBusyReceiving+                      || targetState == ProcessIdle)+               if targetState /= ProcessShuttingDown then do+                insertMonitoringReference >> pure 1+               else+                processAlreadyDead >> pure 2+          Nothing ->+            processAlreadyDead >> pure 3 -removeMonitoring :: HasCallStack => MonitorReference -> SchedulerState -> STM ()-removeMonitoring monitorRef schedulerState = modifyTVar'-  (schedulerState ^. processMonitors)-  (Set.filter (\(ref, _) -> ref /= monitorRef))+  where+    insertMonitoringReference =+      modifyTVar' (schedulerState ^. processMonitors)+                  (Set.insert (monitorRef, owner)) +    processAlreadyDead = do+      let processDownMessage =+            ProcessDown monitorRef (ExitOtherProcessNotRunning target) target+      wasEnqueued <- enqueueMessageOtherProcess owner+                                     (toStrictDynamic processDownMessage)+                                     schedulerState+      check wasEnqueued+ triggerAndRemoveMonitor-  :: ProcessId -> SomeExitReason -> SchedulerState -> STM ()+  :: ProcessId -> Interrupt 'NoRecovery -> SchedulerState -> STM [ProcessId] triggerAndRemoveMonitor downPid reason schedulerState = do   monRefs <- readTVar (schedulerState ^. processMonitors)-  traverse_ go monRefs+  catMaybes <$> traverse go (toList monRefs)  where-  go (mr, owner) = when-    (monitoredProcess mr == downPid)-    (do-      let processDownMessage = ProcessDown mr reason downPid-      enqueueMessageOtherProcess owner (toStrictDynamic processDownMessage) schedulerState-      removeMonitoring mr schedulerState-    )+  go (mr, owner) = do+    if monitoredProcess mr == downPid+     then do+        let processDownMessage = ProcessDown mr reason downPid+        wasEnqueued <-+          enqueueMessageOtherProcess+            owner+            (toStrictDynamic processDownMessage)+            schedulerState+        removeMonitoring mr schedulerState+        pure $ if wasEnqueued then Nothing else Just owner+     else+      pure Nothing +removeMonitoring :: MonitorReference -> SchedulerState -> STM ()+removeMonitoring monitorRef schedulerState = modifyTVar'+  (schedulerState ^. processMonitors)+  (Set.filter (\(ref, _) -> ref /= monitorRef))++ -- * Process Implementation instance Show ProcessInfo where   show p = "process info: " ++ show (p ^. processId)@@ -283,7 +310,7 @@   -> Eff SafeEffects (Interrupt 'NoRecovery)   -> Eff BaseEffects (Interrupt 'NoRecovery) handleProcess myProcessInfo actionToRun = fix-  (handle_relay' singleStep (\er _nextRef -> return er))+  (handle_relay' singleStep (\er _nextRef -> setMyProcessState ProcessShuttingDown >> return er))   actionToRun   0  where@@ -304,9 +331,9 @@ -- setMyProcessState st = do --  oldSt <- lift (atomically (readTVar myProcessStateVar <* setMyProcessStateSTM st)) --  logDebug ("state change: "<> show oldSt <> " -> " <> show st)-  setMyProcessStateSTM = modifyTVar myProcessStateVar . set _2+  setMyProcessStateSTM   = modifyTVar myProcessStateVar . set _2   setMyProcessDetailsSTM = modifyTVar myProcessStateVar . set _1-  myMessageQVar        = myProcessInfo ^. messageQ+  myMessageQVar          = myProcessInfo ^. messageQ   kontinueWith     :: forall s v a      . HasCallStack@@ -384,6 +411,7 @@     SelfPid                  -> diskontinue shutdownRequest     MakeReference            -> diskontinue shutdownRequest     YieldProcess             -> diskontinue shutdownRequest+    Delay           _        -> diskontinue shutdownRequest     Shutdown        r        -> diskontinue r     GetProcessState _        -> diskontinue shutdownRequest     UpdateProcessDetails _   -> diskontinue shutdownRequest@@ -413,6 +441,7 @@       SelfPid                  -> kontinue (Interrupted interruptRequest)       MakeReference            -> kontinue (Interrupted interruptRequest)       YieldProcess             -> kontinue (Interrupted interruptRequest)+      Delay           _        -> kontinue (Interrupted interruptRequest)       Shutdown        r        -> diskontinue r       GetProcessState _        -> kontinue (Interrupted interruptRequest)       UpdateProcessDetails _   -> kontinue (Interrupted interruptRequest)@@ -430,7 +459,7 @@     -> Eff BaseEffects a   interpretRequest kontinue diskontinue nextRef = \case     SendMessage toPid msg ->-      interpretSend toPid msg >>= kontinue nextRef . ResumeWith+      void (interpretSend toPid msg) >>= kontinue nextRef . ResumeWith     SendInterrupt toPid msg -> if toPid == myPid       then kontinue nextRef (Interrupted msg)       else@@ -453,11 +482,13 @@     ReceiveSelectedMessage f -> do       recvRes <- interpretReceive f       either diskontinue (kontinue nextRef) recvRes+    Shutdown r    -> diskontinue r     FlushMessages -> interpretFlush >>= kontinue nextRef     SelfPid       -> kontinue nextRef (ResumeWith myPid)     MakeReference -> kontinue (nextRef + 1) (ResumeWith nextRef)     YieldProcess  -> kontinue nextRef (ResumeWith ())-    Shutdown r    -> diskontinue r+    Delay t ->+      interpretDelay t >>= either diskontinue (kontinue nextRef)     GetProcessState toPid ->       interpretGetProcessState toPid >>= kontinue nextRef . ResumeWith     UpdateProcessDetails d ->@@ -472,7 +503,9 @@     interpretMonitor !target = do       setMyProcessState ProcessBusyMonitoring       schedulerState <- getSchedulerState-      lift (atomically (addMonitoring target myPid schedulerState))+      monitoringReference <- lift (atomically (nextMonitorReference target schedulerState))+      void $ lift (atomically (addMonitoring monitoringReference myPid schedulerState))+      return monitoringReference     interpretDemonitor !ref = do       setMyProcessState ProcessBusyMonitoring       schedulerState <- getSchedulerState@@ -539,6 +572,31 @@         myMessageQ <- readTVar myMessageQVar         modifyTVar' myMessageQVar (incomingMessages .~ Seq.Empty)         return (ResumeWith (toList (myMessageQ ^. incomingMessages)))+    interpretDelay+      :: Timeout+      -> Eff BaseEffects (Either (Interrupt 'NoRecovery) (ResumeProcess ()))+    interpretDelay (TimeoutMicros t) = do+      setMyProcessState ProcessBusySleeping+      lift $ do+        timeoutTVar <- newTVarIO False+        newDelayThreadId <- forkIO $ do+          atomically $ writeTVar timeoutTVar False+          threadDelay t+          atomically $ writeTVar timeoutTVar True+        (elapsed, res) <- atomically $ do+          myMessageQ <- readTVar myMessageQVar+          case myMessageQ ^. shutdownRequests of+            Nothing -> do+              done <- readTVar timeoutTVar+              unless done retry+              return (True, Right (ResumeWith ()))+            Just shutdownRequest -> do+              modifyTVar' myMessageQVar (shutdownRequests .~ Nothing)+              case fromSomeExitReason shutdownRequest of+                Left  sr -> return (False, Left sr)+                Right ir -> return (False, Right (Interrupted ir))+        unless elapsed (killThread newDelayThreadId)+        return res     interpretReceive       :: MessageSelector b       -> Eff BaseEffects (Either (Interrupt 'NoRecovery) (ResumeProcess b))@@ -623,15 +681,12 @@   logAppendProcInfo pid =     let addProcessId = over           lmProcessId-          (maybe (Just (T.pack (printf "% 9s" (show title ++ show pid)))) Just)+          (maybe (Just (T.pack (show title ++ show pid))) Just)     in  censorLogs @(Lift IO) addProcessId   triggerProcessLinksAndMonitors-    :: ProcessId -> Interrupt e -> TVar (Set ProcessId) -> Eff BaseEffects ()+    :: ProcessId -> Interrupt 'NoRecovery -> TVar (Set ProcessId) -> Eff BaseEffects ()   triggerProcessLinksAndMonitors !pid !reason !linkSetVar = do     schedulerState <- getSchedulerState-    lift $ atomically $ triggerAndRemoveMonitor pid-                                                (SomeExitReason reason)-                                                schedulerState     let exitSeverity = toExitSeverity reason         sendIt !linkedPid = do           let msg = SomeExitReason (LinkedProcessCrashed pid)@@ -656,6 +711,7 @@                           else                              return (Right (Right linkedPid))                         else return (Left linkedPid)+    downMessageSendResults <- lift . atomically $ triggerAndRemoveMonitor pid reason schedulerState     linkedPids <- lift       (atomically         (do@@ -666,14 +722,19 @@       )     res <- traverse sendIt (toList linkedPids)     traverse_-      (logDebug . either-        (("linked process no found: " <>) . T.pack . show)+      (either+        (logNotice . ("linked process not found: " <>) . T.pack . show)         (either-              (("linked process crashed, interrupting linked process: " <>) . T.pack . show)-              (("linked process exited peacefully, not sending shutdown to linked process: " <>) . T.pack . show)+              (logWarning . ("process crashed, interrupting linked process: " <>) . T.pack . show)+              (logDebug . ("linked process exited peacefully, not sending shutdown to linked process: " <>) . T.pack . show)         )       )       res+    when (not (null downMessageSendResults)) $+      logWarning+        (  "failed to enqueue monitor down messages for: "+        <> T.pack(show downMessageSendResults)+        )   doForkProc     :: ProcessInfo     -> SchedulerState@@ -718,13 +779,23 @@     )    where     exitReasonFromException exc = case Safe.fromException exc of-      Just Async.AsyncCancelled -> ExitProcessCancelled+      Just Async.AsyncCancelled -> ExitProcessCancelled Nothing       Nothing -> ExitUnhandledError (  "runtime exception: "                                     <> T.pack (prettyCallStack callStack)                                     <> " "                                     <> T.pack (Safe.displayException exc)                                     )     logExitAndTriggerLinksAndMonitors reason pid = do+      (_, currentState) <-+        lift (atomically+          (readTVar (procInfo ^. processState)+            <* modifyTVar' (procInfo ^. processState) (_2 .~ ProcessShuttingDown)))+      when (currentState /= ProcessShuttingDown)+        (logNotice ("aborted in state: "+          <> T.pack (show currentState)+          <> " by: "+          <> T.pack (show reason)+          ))       triggerProcessLinksAndMonitors pid reason (procInfo ^. processLinks)       logProcessExit reason       return reason@@ -734,13 +805,13 @@ getSchedulerState = ask  enqueueMessageOtherProcess-  :: HasCallStack => ProcessId -> StrictDynamic -> SchedulerState -> STM ()+  :: HasCallStack => ProcessId -> StrictDynamic -> SchedulerState -> STM Bool enqueueMessageOtherProcess toPid msg schedulerState =   view (at toPid) <$> readTVar (schedulerState ^. processTable) >>= maybe-    (return ())+    (return False)     (\toProcessTable -> do       modifyTVar' (toProcessTable ^. messageQ) (incomingMessages %~ (:|> msg))-      return ()+      return True     )  enqueueShutdownRequest

@@ -21,6 +21,7 @@ import           Control.Concurrent             ( yield ) import           Control.Eff import           Control.Eff.Extend+import           Control.Eff.Concurrent.Misc import           Control.Eff.Concurrent.Process import           Control.Eff.Log import           Control.Eff.LogWriter.IO@@ -42,8 +43,8 @@ import           Data.Foldable import           Data.Monoid import qualified Control.Monad.State.Strict    as State-import Data.Function (fix)-import Data.Dynamic (dynTypeRep)+import           Data.Function                  ( fix)+import           Data.Dynamic                   ( dynTypeRep )  -- ----------------------------------------------------------------------------- --  STS and ProcessInfo@@ -62,8 +63,9 @@       (d >= 10)       (appEndo          (Endo (showChar ' ' . shows pTitle . showString ": ") <>-          foldMap (Endo . shows . dynTypeRep . unwrapStrictDynamic) (toList pQ) <>-          Endo (shows pDetails)))+          foldMap (Endo . showSTypeRep . dynTypeRep . unwrapStrictDynamic) (toList pQ) <>+          Endo (shows pDetails))+      )  makeLenses ''ProcessInfo @@ -166,14 +168,14 @@       pt <- use msgQs       if Map.member target pt         then monitors %= Set.insert (mref, owner)-        else let pdown = ProcessDown mref (SomeExitReason (OtherProcessNotRunning target)) target+        else let pdown = ProcessDown mref (ExitOtherProcessNotRunning target) target               in State.modify' (enqueueMsg owner (toStrictDynamic pdown))     return mref  removeMonitoring :: MonitorReference -> STS m r -> STS m r removeMonitoring mref = monitors %~ Set.filter (\(ref, _) -> ref /= mref) -triggerAndRemoveMonitor :: ProcessId -> SomeExitReason -> STS m r -> STS m r+triggerAndRemoveMonitor :: ProcessId -> Interrupt 'NoRecovery -> STS m r -> STS m r triggerAndRemoveMonitor downPid reason = State.execState $ do   monRefs <- use monitors   traverse_ go monRefs@@ -294,7 +296,10 @@   OnFlushMessages :: (ResumeProcess [StrictDynamic] -> Eff r (OnYield r a))                   -> OnYield r a   OnYield :: (ResumeProcess () -> Eff r (OnYield r a))-         -> OnYield r a+          -> OnYield r a+  OnDelay :: Timeout+          -> (ResumeProcess () -> Eff r (OnYield r a))+          -> OnYield r a   OnSelf :: (ResumeProcess ProcessId -> Eff r (OnYield r a))          -> OnYield r a   OnSpawn :: Bool@@ -346,9 +351,10 @@   show = \case     OnFlushMessages _          -> "OnFlushMessages"     OnYield         _          -> "OnYield"+    OnDelay t _                -> "OnDelay " ++ show t     OnSelf          _          -> "OnSelf"-    OnSpawn False t _ _        -> "OnSpawn" ++ show t-    OnSpawn True  t _ _        -> "OnSpawn (link)" ++ show t+    OnSpawn False t _ _        -> "OnSpawn " ++ show t+    OnSpawn True  t _ _        -> "OnSpawn (link) " ++ show t     OnDone     _               -> "OnDone"     OnShutdown e               -> "OnShutdown " ++ show e     OnInterrupt e _            -> "OnInterrupt " ++ show e@@ -378,6 +384,7 @@        -> Eff r (OnYield r v)   cont k q FlushMessages                = return (OnFlushMessages (k . qApp q))   cont k q YieldProcess                 = return (OnYield (k . qApp q))+  cont k q (Delay t)                    = return (OnDelay t (k . qApp q))   cont k q SelfPid                      = return (OnSelf (k . qApp q))   cont k q (Spawn     t e             ) = return (OnSpawn False t e (k . qApp q))   cont k q (SpawnLink t e             ) = return (OnSpawn True t e (k . qApp q))@@ -426,7 +433,12 @@             handleProcess               (dropMsgQ                  pid-                 (triggerAndRemoveMonitor pid (either SomeExitReason (const (SomeExitReason ExitNormally)) res) stsNew))+                 (triggerAndRemoveMonitor+                  pid+                  (either+                    id+                    (const ExitNormally) res)+                    stsNew))               nextTargets    in case processState of         OnDone r -> handleExit (Right r)@@ -449,6 +461,7 @@                         OnSendShutdown _ _ _tk -> return (OnShutdown sr)                         OnFlushMessages _tk -> return (OnShutdown sr)                         OnYield _tk -> return (OnShutdown sr)+                        OnDelay _t _tk -> return (OnShutdown sr)                         OnSelf _tk -> return (OnShutdown sr)                         OnSend _ _ _tk -> return (OnShutdown sr)                         OnRecv _ _tk -> return (OnShutdown sr)@@ -503,6 +516,12 @@           let (msgs, newSts) = flushMsgs pid sts           nextK <- kontinue newSts k msgs           handleProcess newSts (rest :|> (nextK, pid))+        OnDelay t k ->+          if t <= 0 then do+               nextK <- kontinue sts k ()+               handleProcess sts (rest :|> (nextK, pid))+          else+               handleProcess sts (rest :|> (OnDelay (t - 1) k, pid))         recv@(OnRecv messageSelector k) ->           case receiveMsg pid messageSelector sts of             Nothing -> do@@ -556,6 +575,7 @@                 OnSendInterrupt _ _ tk -> tk (Interrupted sr)                 OnSendShutdown _ _ tk -> tk (Interrupted sr)                 OnFlushMessages tk -> tk (Interrupted sr)+                OnDelay _ tk -> tk (Interrupted sr)                 OnYield tk -> tk (Interrupted sr)                 OnSelf tk -> tk (Interrupted sr)                 OnSend _ _ tk -> tk (Interrupted sr)

@@ -1,35 +1,36 @@--- | Functions for timeouts when receiving messages.+-- | Functions for receive timeouts and delayed messages sending. ----- NOTE: If you use a single threaded scheduler, these functions will not work--- as expected. (This is an open TODO)+-- Based on the 'delay' function. -- -- @since 0.12.0 module Control.Eff.Concurrent.Process.Timer-  ( Timeout(TimeoutMicros, fromTimeoutMicros)-  , TimerReference()+  ( TimerReference()   , TimerElapsed(fromTimerElapsed)   , sendAfter   , startTimer+  , sendAfterWithTitle+  , startTimerWithTitle   , cancelTimer   , selectTimerElapsed   , receiveAfter   , receiveSelectedAfter   , receiveSelectedWithMonitorAfter+  , receiveAfterWithTitle+  , receiveSelectedAfterWithTitle+  , receiveSelectedWithMonitorAfterWithTitle   )-   -- , receiveSelectedAfter, receiveAnyAfter, sendMessageAfter)+ where  import           Control.Eff.Concurrent.Process-import           Control.Concurrent+import           Control.Eff.Concurrent.Misc import           Control.Eff import           Control.DeepSeq-import           Control.Monad.IO.Class import           Data.Typeable import           Data.Text as T import           Control.Applicative import           GHC.Stack - -- | Wait for a message of the given type for the given time. When no message -- arrives in time, return 'Nothing'. This is based on -- 'receiveSelectedAfter'.@@ -37,8 +38,7 @@ -- @since 0.12.0 receiveAfter   :: forall a r q-   . ( Lifted IO q-     , HasCallStack+   . ( HasCallStack      , HasProcesses r q      , Typeable a      , NFData a@@ -56,16 +56,23 @@ -- @since 0.12.0 receiveSelectedAfter   :: forall a r q-   . ( Lifted IO q-     , HasCallStack+   . ( HasCallStack      , HasProcesses r q      , Show a+     , Typeable a      )   => MessageSelector a   -> Timeout   -> Eff r (Either TimerElapsed a) receiveSelectedAfter sel t = do-  timerRef <- startTimer t+  let timerTitle =+        MkProcessTitle+          ("receive-timer-"+          <> pack (showSTypeable @a "")+          <> "-"+          <> pack (show t)+          )+  timerRef <- startTimerWithTitle timerTitle t   res      <- receiveSelectedMessage     (Left <$> selectTimerElapsed timerRef <|> Right <$> sel)   cancelTimer timerRef@@ -76,17 +83,87 @@ -- @since 0.22.0 receiveSelectedWithMonitorAfter   :: forall a r q-   . ( Lifted IO q-     , HasCallStack+   . ( HasCallStack      , HasProcesses r q      , Show a+     , Typeable a      )   => ProcessId   -> MessageSelector a   -> Timeout   -> Eff r (Either (Either ProcessDown TimerElapsed) a)-receiveSelectedWithMonitorAfter pid sel t = do-  timerRef <- startTimer t+receiveSelectedWithMonitorAfter pid sel t =+  let timerTitle =+        MkProcessTitle+          ("receive-timer-"+          <> pack (showSTypeable @a "")+          <> "-monitoring-"+          <> pack (show pid)+          <> "-"+          <> pack (show t)+          )+  in receiveSelectedWithMonitorAfterWithTitle pid sel t timerTitle+++-- | Wait for a message of the given type for the given time. When no message+-- arrives in time, return 'Nothing'. This is based on+-- 'receiveSelectedAfterWithTitle'.+--+-- @since 0.12.0+receiveAfterWithTitle+  :: forall a r q+   . ( HasCallStack+     , HasProcesses r q+     , Typeable a+     , NFData a+     , Show a+     )+  => Timeout+  -> ProcessTitle+  -> Eff r (Maybe a)+receiveAfterWithTitle t timerTitle =+  either (const Nothing) Just <$> receiveSelectedAfterWithTitle (selectMessage @a) t timerTitle++-- | Wait for a message of the given type for the given time. When no message+-- arrives in time, return 'Left' 'TimerElapsed'. This is based on+-- 'selectTimerElapsed' and 'startTimerWithTitle'.+--+-- @since 0.12.0+receiveSelectedAfterWithTitle+  :: forall a r q+   . ( HasCallStack+     , HasProcesses r q+     , Show a+     , Typeable a+     )+  => MessageSelector a+  -> Timeout+  -> ProcessTitle+  -> Eff r (Either TimerElapsed a)+receiveSelectedAfterWithTitle sel t timerTitle = do+  timerRef <- startTimerWithTitle timerTitle t+  res      <- receiveSelectedMessage+    (Left <$> selectTimerElapsed timerRef <|> Right <$> sel)+  cancelTimer timerRef+  return res++-- | Like 'receiveWithMonitorWithTitle' combined with 'receiveSelectedAfterWithTitle'.+--+-- @since 0.30.0+receiveSelectedWithMonitorAfterWithTitle+  :: forall a r q+   . ( HasCallStack+     , HasProcesses r q+     , Show a+     , Typeable a+     )+  => ProcessId+  -> MessageSelector a+  -> Timeout+  -> ProcessTitle+  -> Eff r (Either (Either ProcessDown TimerElapsed) a)+receiveSelectedWithMonitorAfterWithTitle pid sel t timerTitle = do+  timerRef <- startTimerWithTitle timerTitle t   res      <- withMonitor pid $ \pidMon -> do                 receiveSelectedMessage                   (   Left . Left  <$> selectProcessDown pidMon@@ -104,17 +181,6 @@ selectTimerElapsed timerRef =   filterMessage (\(TimerElapsed timerRefIn) -> timerRef == timerRefIn) ---- | A number of micro seconds.------ @since 0.12.0-newtype Timeout = TimeoutMicros {fromTimeoutMicros :: Int}-  deriving (NFData, Ord,Eq, Num, Integral, Real, Enum, Typeable)--instance Show Timeout where-  showsPrec d (TimeoutMicros t) =-    showParen (d >= 10) (showString "timeout: " . shows t . showString " µs")- -- | The reference to a timer started by 'startTimer', required to stop -- a timer via 'cancelTimer'. --@@ -135,9 +201,7 @@ instance Show TimerElapsed where   showsPrec d (TimerElapsed t) =     showParen (d >= 10) (shows t . showString " elapsed")----- @since 0.12.0 - -- | Send a message to a given process after waiting. The message is created by -- applying the function parameter to the 'TimerReference', such that the -- message can directly refer to the timer.@@ -145,8 +209,7 @@ -- @since 0.12.0 sendAfter   :: forall r q message-   . ( Lifted IO q-     , HasCallStack+   . ( HasCallStack      , HasProcesses r q      , Typeable message      , NFData message@@ -155,30 +218,74 @@   -> Timeout   -> (TimerReference -> message)   -> Eff r TimerReference-sendAfter pid (TimeoutMicros us) mkMsg =+sendAfter pid t mkMsg =+  sendAfterWithTitle+    (MkProcessTitle ("send-after-timer-" <> T.pack (show t) <> "-" <> T.pack (showSTypeable @message "") <> "-" <> T.pack (show pid)))+    pid+    t+    mkMsg++-- | Like 'sendAfter' but with a user provided name for the timer process.+--+-- @since 0.30.0+sendAfterWithTitle+  :: forall r q message+   . ( HasCallStack+     , HasProcesses r q+     , Typeable message+     , NFData message+     )+  => ProcessTitle+  -> ProcessId+  -> Timeout+  -> (TimerReference -> message)+  -> Eff r TimerReference+sendAfterWithTitle title pid t mkMsg =   TimerReference <$>-  spawn-    (MkProcessTitle ("after_" <> T.pack (show us) <> "us_to_" <> T.pack (show pid)))-    ((if us == 0-        then yieldProcess-        else liftIO (threadDelay us)) >>-     self >>=-     (sendMessage pid . force . mkMsg . TimerReference))+  (spawn+    title+    (delay t+    >>  self+    >>= (sendMessage pid . force . mkMsg . TimerReference)))+ -- | Start a new timer, after the time has elapsed, 'TimerElapsed' is sent to -- calling process. The message also contains the 'TimerReference' returned by -- this function. Use 'cancelTimer' to cancel the timer. Use -- 'selectTimerElapsed' to receive the message using 'receiveSelectedMessage'. -- To receive messages with guarded with a timeout see 'receiveAfter'. --+-- This calls 'sendAfterWithTitle' under the hood with 'TimerElapsed' as+-- message.+--+-- @since 0.30.0+startTimerWithTitle+  :: forall r q+   . ( HasCallStack+     , HasProcesses r q+     )+  => ProcessTitle+  -> Timeout+  -> Eff r TimerReference -- TODO add a parameter to the TimerReference+startTimerWithTitle title t = do+  p <- self+  sendAfterWithTitle title p t TimerElapsed++-- | Start a new timer, after the time has elapsed, 'TimerElapsed' is sent to+-- calling process. The message also contains the 'TimerReference' returned by+-- this function. Use 'cancelTimer' to cancel the timer. Use+-- 'selectTimerElapsed' to receive the message using 'receiveSelectedMessage'.+-- To receive messages with guarded with a timeout see 'receiveAfter'.+--+-- Calls 'sendAfter' under the hood.+-- -- @since 0.12.0 startTimer   :: forall r q-   . ( Lifted IO q-     , HasCallStack+   . ( HasCallStack      , HasProcesses r q      )   => Timeout-  -> Eff r TimerReference+  -> Eff r TimerReference -- TODO add a parameter to the TimerReference startTimer t = do   p <- self   sendAfter p t TimerElapsed@@ -188,8 +295,7 @@ -- @since 0.12.0 cancelTimer   :: forall r q-   . ( Lifted IO q-     , HasCallStack+   . ( HasCallStack      , HasProcesses r q      )   => TimerReference

@@ -0,0 +1,645 @@+{-# LANGUAGE UndecidableInstances #-}+-- | A process broker spawns and monitors child processes.+--+-- The child processes are mapped to symbolic identifier values: Child-IDs.+--+-- This is the barest, most minimal version of a broker. Children+-- can be started, but not restarted.+--+-- Children can efficiently be looked-up by an id-value, and when+-- the broker is shutdown, all children will be shutdown these+-- are actually all the features of this broker implementation.+--+-- Also, this minimalist broker only knows how to spawn a+-- single kind of child process.+--+-- When a broker spawns a new child process, it expects the+-- child process to return a 'ProcessId'. The broker will+-- 'monitor' the child process.+--+-- This is in stark contrast to how Erlang/OTP handles things;+-- In the OTP Supervisor, the child has to link to the parent.+-- This allows the child spec to be more flexible in that no+-- @pid@ has to be passed from the child start function to the+-- broker process, and also, a child may break free from+-- the broker by unlinking.+--+-- Now while this seems nice at first, this might actually+-- cause surprising results, since it is usually expected that+-- stopping a broker also stops the children, or that a child+-- exit shows up in the logging originating from the former+-- broker.+--+-- The approach here is to allow any child to link to the+-- broker to realize when the broker was violently killed,+-- and otherwise give the child no chance to unlink itself from+-- its broker.+--+-- This module is far simpler than the Erlang/OTP counter part,+-- of a @simple_one_for_one@ supervisor.+--+-- The future of this broker might not be a-lot more than+-- it currently is. The ability to restart processes might be+-- implemented outside of this broker module.+--+-- One way to do that is to implement the restart logic in+-- a separate module, since the child-id can be reused when a child+-- exits.+--+-- @since 0.23.0+module Control.Eff.Concurrent.Protocol.Broker+  ( startLink+  , statefulChild+  , stopBroker+  , isBrokerAlive+  , monitorBroker+  , getDiagnosticInfo+  , spawnChild+  , spawnOrLookup+  , lookupChild+  , callById+  , castById+  , stopChild+  , ChildNotFound(..)+  , Broker()+  , Pdu(StartC, StopC, LookupC, GetDiagnosticInfo)+  , ChildId+  , Stateful.StartArgument(MkBrokerConfig)+  , brokerConfigChildStopTimeout+  , SpawnErr(AlreadyStarted)+  , ChildEvent(OnChildSpawned, OnChildDown, OnBrokerShuttingDown)+  ) where++import Control.DeepSeq (NFData(rnf))+import Control.Eff as Eff+import Control.Eff.Concurrent.Protocol+import Control.Eff.Concurrent.Protocol.Client+import Control.Eff.Concurrent.Protocol.Wrapper+import Control.Eff.Concurrent.Protocol.Observer+import qualified Control.Eff.Concurrent.Protocol.EffectfulServer as Effectful+import qualified Control.Eff.Concurrent.Protocol.StatefulServer as Stateful+import Control.Eff.Concurrent.Protocol.Broker.InternalState+import Control.Eff.Concurrent.Process+import Control.Eff.Concurrent.Process.Timer+import Control.Eff.Extend (raise)+import Control.Eff.Log+import Control.Eff.State.Strict as Eff+import Control.Lens hiding ((.=), use)+import Data.Default+import Data.Dynamic+import Data.Foldable+import Data.Kind+import qualified Data.Map as Map+import Data.Text (Text, pack)+import Data.Type.Pretty+import GHC.Generics (Generic)+import GHC.Stack+import Control.Applicative ((<|>))++-- * Broker Server API++-- ** Functions++-- | Start and link a new broker process with the given 'SpawnFun'unction.+--+-- To spawn new child processes use 'spawnChild'.+--+-- @since 0.23.0+startLink+  :: forall p e+  . ( HasCallStack+    , LogIo (Processes e)+    , TangibleBroker p+    , Stateful.Server (Broker p) (Processes e)+    )+  => Stateful.StartArgument (Broker p) (Processes e)+  -> Eff (Processes e) (Endpoint (Broker p))+startLink = Stateful.startLink++-- | A smart constructor for 'MkBrokerConfig' that makes it easy to start a 'Stateful.Server' instance.+--+-- The user needs to instantiate @'ChildId' p@.+--+-- @since 0.30.0+statefulChild+  :: forall p e+  . ( HasCallStack+    , LogIo (Processes e)+    , TangibleBroker (Stateful.Stateful p)+    , Stateful.Server (Broker (Stateful.Stateful p)) (Processes e)+    )+  => Timeout+  -> (ChildId p -> Stateful.StartArgument p (Processes e))+  -> Stateful.StartArgument (Broker (Stateful.Stateful p)) (Processes e)+statefulChild t f = MkBrokerConfig t (Stateful.Init . f)++-- | Stop the broker and shutdown all processes.+--+-- Block until the broker has finished.+--+-- @since 0.23.0+stopBroker+  :: ( HasCallStack+     , HasProcesses e q+     , Member Logs e+     , Lifted IO e+     , TangibleBroker p+     )+  => Endpoint (Broker p)+  -> Eff e ()+stopBroker ep = do+  logInfo ("stopping broker: " <> pack (show ep))+  mr <- monitor (_fromEndpoint ep)+  sendInterrupt (_fromEndpoint ep) NormalExitRequested+  r <- receiveSelectedMessage (selectProcessDown mr)+  logInfo ("broker stopped: " <> pack (show ep) <> " " <> pack (show r))++-- | Check if a broker process is still alive.+--+-- @since 0.23.0+isBrokerAlive+  :: forall p q0 e .+     ( HasCallStack+     , Member Logs e+     , Typeable p+     , HasProcesses e q0+     )+  => Endpoint (Broker p)+  -> Eff e Bool+isBrokerAlive x = isProcessAlive (_fromEndpoint x)++-- | Monitor a broker process.+--+-- @since 0.23.0+monitorBroker+  :: forall p q0 e .+     ( HasCallStack+     , Member Logs e+     , HasProcesses e q0+     , TangibleBroker p+     )+  => Endpoint (Broker p)+  -> Eff e MonitorReference+monitorBroker x = monitor (_fromEndpoint x)++-- | Start and monitor a new child process using the 'SpawnFun' passed+-- to 'startLink'.+--+-- @since 0.23.0+spawnChild+  :: forall p q0 e .+     ( HasCallStack+     , Member Logs e+     , HasProcesses e q0+     , TangibleBroker p+     , Typeable (Effectful.ServerPdu p)+     )+  => Endpoint (Broker p)+  -> ChildId p+  -> Eff e (Either (SpawnErr p) (Endpoint (Effectful.ServerPdu p)))+spawnChild ep cId = call ep (StartC cId)++-- | Start and monitor a new child process using the 'SpawnFun' passed+-- to 'startLink'.+--+-- Call 'spawnChild' and unpack the 'Either' result,+-- ignoring the 'AlreadyStarted' error.+--+-- @since 0.29.2+spawnOrLookup+  :: forall p q0 e .+     ( HasCallStack+     , Member Logs e+     , HasProcesses e q0+     , TangibleBroker p+     , Typeable (Effectful.ServerPdu p)+     )+  => Endpoint (Broker p)+  -> ChildId p+  -> Eff e (Endpoint (Effectful.ServerPdu p))+spawnOrLookup supEp cId =+  do res <- spawnChild supEp cId+     pure $+      case res of+        Left (AlreadyStarted _ ep) -> ep+        Right ep -> ep++-- | Lookup the given child-id and return the output value of the 'SpawnFun'+--   if the client process exists.+--+-- @since 0.23.0+lookupChild ::+    forall p e q0 .+     ( HasCallStack+     , Member Logs e+     , HasProcesses e q0+     , TangibleBroker p+     , Typeable (Effectful.ServerPdu p)+     )+  => Endpoint (Broker p)+  -> ChildId p+  -> Eff e (Maybe (Endpoint (Effectful.ServerPdu p)))+lookupChild ep cId = call ep (LookupC @p cId)++-- | Stop a child process, and block until the child has exited.+--+-- Return 'True' if a process with that ID was found, 'False' if no process+-- with the given ID was running.+--+-- @since 0.23.0+stopChild ::+    forall p e q0 .+     ( HasCallStack+     , Member Logs e+     , HasProcesses e q0+     , TangibleBroker p+     )+  => Endpoint (Broker p)+  -> ChildId p+  -> Eff e Bool+stopChild ep cId = call ep (StopC @p cId (TimeoutMicros 4000000))++callById ::+    forall destination protocol result e q0 .+     ( HasCallStack+     , Member Logs e+     , HasProcesses e q0+     , Lifted IO e+     , Lifted IO q0+     , TangibleBroker protocol+     , TangiblePdu destination ( 'Synchronous result)+     , TangiblePdu protocol ( 'Synchronous result)+     , Embeds (Effectful.ServerPdu destination) protocol+     , Ord (ChildId destination)+     , Tangible (ChildId destination)+     , Typeable (Effectful.ServerPdu destination)+     , Tangible result+     , NFData (Pdu protocol ( 'Synchronous result))+     , NFData (Pdu (Effectful.ServerPdu destination) ( 'Synchronous result))+     , Show (Pdu (Effectful.ServerPdu destination) ( 'Synchronous result))+     )+  => Endpoint (Broker destination)+  -> ChildId destination+  -> Pdu protocol ( 'Synchronous result)+  -> Timeout+  -> Eff e result+callById broker cId pdu tMax =+  lookupChild broker cId+   >>=+    maybe+      (do logError ("callById failed for: " <> pack (show pdu))+          interrupt (InterruptedBy (ChildNotFound cId broker))+      )+      (\cEp -> callWithTimeout cEp pdu tMax)++data ChildNotFound child where+  ChildNotFound :: ChildId child -> Endpoint (Broker child) -> ChildNotFound child+  deriving (Typeable)++instance (Show (ChildId child), Typeable child) => Show (ChildNotFound child) where+  showsPrec d (ChildNotFound cId broker)+    = showParen (d >= 10) ( showString "child not found: "+                           . showsPrec 10 cId+                           . showString " at: "+                           . showsPrec 10 broker+                           )++instance NFData (ChildId c) => NFData (ChildNotFound c) where+  rnf (ChildNotFound cId broker) = rnf cId `seq` rnf broker `seq` ()++castById ::+    forall destination protocol e q0 .+     ( HasCallStack+     , Member Logs e+     , HasProcesses e q0+     , TangibleBroker protocol+     , TangiblePdu destination 'Asynchronous+     , TangiblePdu protocol 'Asynchronous+     )+  => Endpoint (Broker destination)+  -> ChildId destination+  -> Pdu protocol 'Asynchronous+  -> Eff e ()+castById = error "TODO"+++-- | Return a 'Text' describing the current state of the broker.+--+-- @since 0.23.0+getDiagnosticInfo+  :: forall p e q0 .+     ( HasCallStack+     , HasProcesses e q0+     , TangibleBroker p+     )+  => Endpoint (Broker p)+  -> Eff e Text+getDiagnosticInfo s = call s (GetDiagnosticInfo @p)++-- ** Types++-- | The index type of 'Server' supervisors.+--+-- A @'Broker' p@ manages the life cycle of the processes, running the @'Server' p@+-- methods of that specific type.+--+-- The broker maps an identifier value of type @'ChildId' p@ to an @'Endpoint' p@.+--+-- @since 0.24.0+data Broker (p :: Type) deriving Typeable++instance Typeable p => HasPdu (Broker p) where+  type instance EmbeddedPduList (Broker p) = '[ObserverRegistry (ChildEvent p)]+  -- | The 'Pdu' instance contains methods to start, stop and lookup a child+  -- process, as well as a diagnostic callback.+  --+  -- @since 0.23.0+  data instance  Pdu (Broker p) r where+          StartC :: ChildId p -> Pdu (Broker p) ('Synchronous (Either (SpawnErr p) (Endpoint (Effectful.ServerPdu p))))+          StopC :: ChildId p -> Timeout -> Pdu (Broker p) ('Synchronous Bool)+          LookupC :: ChildId p -> Pdu (Broker p) ('Synchronous (Maybe (Endpoint (Effectful.ServerPdu p))))+          GetDiagnosticInfo :: Pdu (Broker p) ('Synchronous Text)+          ChildEventObserverRegistry :: Pdu (ObserverRegistry (ChildEvent p)) r -> Pdu (Broker p) r+      deriving Typeable++instance (Typeable p, Show (ChildId p)) => Show (Pdu (Broker p) r) where+  showsPrec d (StartC c) = showParen (d >= 10) (showString "StartC " . showsPrec 10 c)+  showsPrec d (StopC c t) = showParen (d >= 10) (showString "StopC " . showsPrec 10 c . showChar ' ' . showsPrec 10 t)+  showsPrec d (LookupC c) = showParen (d >= 10) (showString "LookupC " . showsPrec 10 c)+  showsPrec _ GetDiagnosticInfo = showString "GetDiagnosticInfo"+  showsPrec d (ChildEventObserverRegistry c) = showParen (d >= 10) (showString "ChildEventObserverRegistry " . showsPrec 10 c)++instance (NFData (ChildId p)) => NFData (Pdu (Broker p) r) where+  rnf (StartC ci) = rnf ci+  rnf (StopC ci t) = rnf ci `seq` rnf t+  rnf (LookupC ci) = rnf ci+  rnf GetDiagnosticInfo = ()+  rnf (ChildEventObserverRegistry x) = rnf x++instance Typeable p => HasPduPrism (Broker p) (ObserverRegistry (ChildEvent p)) where+ embedPdu = ChildEventObserverRegistry+ fromPdu (ChildEventObserverRegistry x) = Just x+ fromPdu _ = Nothing++type instance ToPretty (Broker p) = "broker" <:> ToPretty p++-- | The event type to indicate that a child was started or stopped.+--+-- The need for this type originated for the watchdog functionality introduced+-- in 0.30.0.+-- The watch dog shall restart a crashed child, and in order to do so, it+-- must somehow monitor the child.+-- Since no order is specified in which processes get the 'ProcessDown' events,+-- a watchdog cannot monitor a child and restart it immediately, because it might+-- have received the process down event before the broker.+-- So instead the watchdog can simply use the broker events, and monitor only the+-- broker process.+--+-- @since 0.30.0+data ChildEvent p where+  OnChildSpawned :: Endpoint (Broker p) -> ChildId p -> Endpoint (Effectful.ServerPdu p) -> ChildEvent p+  OnChildDown :: Endpoint (Broker p) -> ChildId p -> Endpoint (Effectful.ServerPdu p) -> Interrupt 'NoRecovery -> ChildEvent p+  OnBrokerShuttingDown :: Endpoint (Broker p) -> ChildEvent p -- ^ The broker is shutting down and will soon begin stopping/killing its children+  deriving (Typeable, Generic)++instance (NFData (ChildId p)) => NFData (ChildEvent p)++instance (Typeable p, Typeable (Effectful.ServerPdu p), Show (ChildId p)) => Show (ChildEvent p) where+  showsPrec d x =+    case x of+     OnChildSpawned s i e ->+        showParen (d >= 10)+          (shows s . showString ": child-spawned: " . shows i . showChar ' ' . shows e)+     OnChildDown s i e r ->+        showParen (d >= 10)+          (shows s . showString ": child-down: " . shows i . showChar ' ' . shows e . showChar ' ' . showsPrec 10 r)+     OnBrokerShuttingDown s ->+        shows s . showString ": shutting down"++-- | The type of value used to index running 'Server' processes managed by a 'Broker'.+--+-- Note, that the type you provide must be 'Tangible'.+--+-- @since 0.24.0+type family ChildId p++type instance ChildId (Stateful.Stateful p) = ChildId p++-- | Constraints on the parameters to 'Broker'.+--+-- @since 0.24.0+type TangibleBroker p =+  ( Tangible (ChildId p)+  , Ord (ChildId p)+  , Typeable p+  )+++instance+  ( LogIo q+  , TangibleBroker p+  , Tangible (ChildId p)+  , Typeable (Effectful.ServerPdu p)+  , Effectful.Server p (Processes q)+  , HasProcesses (Effectful.ServerEffects p (Processes q)) q+  ) => Stateful.Server (Broker p) (Processes q) where++  -- | Options that control the 'Broker p' process.+  --+  -- This contains:+  --+  -- * a 'SpawnFun'+  -- * the 'Timeout' after requesting a normal child exit before brutally killing the child.+  --+  -- @since 0.24.0+  data instance StartArgument (Broker p) (Processes q) = MkBrokerConfig+    {+      brokerConfigChildStopTimeout :: Timeout+    , brokerConfigStartFun :: ChildId p -> Effectful.Init p (Processes q)+    }++  data instance Model (Broker p) =+    BrokerModel { _children :: Children (ChildId p) p+                , _childEventObserver :: ObserverRegistry (ChildEvent p)+                }+      deriving Typeable++  setup _ _cfg = pure (BrokerModel def emptyObserverRegistry, ())+  update _ _brokerConfig (Stateful.OnCast req) =+    case  req of+      ChildEventObserverRegistry x ->+        Stateful.zoomModel @(Broker p) childEventObserverLens (observerRegistryHandlePdu x)++  update me brokerConfig (Stateful.OnCall rt req) =+    case req of+      ChildEventObserverRegistry x ->+        logEmergency ("unreachable: " <> pack (show x))++      GetDiagnosticInfo -> zoomToChildren @p $ do+        p <- (pack . show <$> getChildren @(ChildId p) @p)+        sendReply rt p++      LookupC i -> zoomToChildren @p $ do+        p <- fmap (view childEndpoint) <$> lookupChildById @(ChildId p) @p i+        sendReply rt p++      StopC i t -> zoomToChildren @p $ do+        mExisting <- lookupAndRemoveChildById @(ChildId p) @p i+        case mExisting of+          Nothing -> sendReply rt False+          Just existingChild -> do+            reason <- stopOrKillChild i existingChild t+            sendReply rt True+            Stateful.zoomModel @(Broker p)+              childEventObserverLens+              (observerRegistryNotify @(ChildEvent p) (OnChildDown me i (existingChild^.childEndpoint) reason))++      StartC i -> do+        mExisting <- zoomToChildren @p $ lookupChildById @(ChildId p) @p i+        case mExisting of+          Nothing -> do+            childEp <- raise (raise (Effectful.startLink (brokerConfigStartFun brokerConfig i)))+            let childPid = _fromEndpoint childEp+            cMon <- monitor childPid+            zoomToChildren @p $ putChild i (MkChild @p childEp cMon)+            sendReply rt (Right childEp)+            Stateful.zoomModel @(Broker p)+              childEventObserverLens+              (observerRegistryNotify @(ChildEvent p) (OnChildSpawned me i childEp))+          Just existingChild ->+            sendReply rt (Left (AlreadyStarted i (existingChild ^. childEndpoint)))++  update me _brokerConfig (Stateful.OnDown pd) = do+      wasObserver <- Stateful.zoomModel @(Broker p)+        childEventObserverLens+        (observerRegistryRemoveProcess @(ChildEvent p) (downProcess pd))+      if wasObserver+       then logInfo ("observer process died: " <> pack (show pd))+       else do+        oldEntry <- zoomToChildren @p $ lookupAndRemoveChildByMonitor @(ChildId p) @p (downReference pd)+        case oldEntry of+          Nothing -> logWarning ("unexpected: " <> pack (show pd))+          Just (i, c) -> do+            logInfo (  pack (show pd)+                    <> " for child "+                    <> pack (show i)+                    <> " => "+                    <> pack (show (c^.childEndpoint))+                    )+            Stateful.zoomModel @(Broker p)+                childEventObserverLens+                (observerRegistryNotify @(ChildEvent p) (OnChildDown me i (c^.childEndpoint) (downReason pd)))+  update me brokerConfig (Stateful.OnInterrupt e) =+    case e of+      NormalExitRequested -> do+        logDebug ("broker stopping: " <> pack (show e))+        Stateful.zoomModel @(Broker p)+            childEventObserverLens+            (observerRegistryNotify @(ChildEvent p) (OnBrokerShuttingDown me))+        stopAllChildren @p me (brokerConfigChildStopTimeout brokerConfig)+        exitNormally+      LinkedProcessCrashed linked ->+        logNotice (pack (show linked))+      _ -> do+        logWarning ("broker interrupted: " <> pack (show e))+        Stateful.zoomModel @(Broker p)+            childEventObserverLens+            (observerRegistryNotify @(ChildEvent p) (OnBrokerShuttingDown me))+        stopAllChildren @p me (brokerConfigChildStopTimeout brokerConfig)+        exitBecause (interruptToExit e)++  update _ _brokerConfig o = logWarning ("unexpected: " <> pack (show o))+++zoomToChildren :: forall p c e+  . Member (Stateful.ModelState (Broker p)) e+  => Eff (State (Children (ChildId p) p) ': e) c+  -> Eff e c+zoomToChildren = Stateful.zoomModel @(Broker p) childrenLens++childrenLens :: Lens' (Stateful.Model (Broker p)) (Children (ChildId p) p)+childrenLens = lens _children (\(BrokerModel _ o) c -> BrokerModel c o)++childEventObserverLens :: Lens' (Stateful.Model (Broker p)) (ObserverRegistry (ChildEvent p))+childEventObserverLens = lens _childEventObserver (\(BrokerModel o _) c -> BrokerModel o c)++-- | Runtime-Errors occurring when spawning child-processes.+--+-- @since 0.23.0+data SpawnErr p = AlreadyStarted (ChildId p) (Endpoint (Effectful.ServerPdu p))+  deriving (Typeable, Generic)++deriving instance Eq (ChildId p) => Eq (SpawnErr p)+deriving instance Ord (ChildId p) => Ord (SpawnErr p)+deriving instance (Typeable (Effectful.ServerPdu p), Show (ChildId p)) => Show (SpawnErr p)++instance NFData (ChildId p) => NFData (SpawnErr p)++-- Internal Functions++stopOrKillChild+  :: forall p e q0 .+     ( HasCallStack+     , HasProcesses e q0+     , Lifted IO e+     , Lifted IO q0+     , Member Logs e+     , Member (Stateful.ModelState (Broker p)) e+     , TangibleBroker p+     , Typeable (Effectful.ServerPdu p)+     )+  => ChildId p+  -> Child p+  -> Timeout+  -> Eff e (Interrupt 'NoRecovery)+stopOrKillChild cId c stopTimeout =+      do+        broker <- asEndpoint @(Broker p) <$> self+        t <- startTimerWithTitle+                (MkProcessTitle ("child-exit-timer-" <> pack (show broker) <> "-" <> pack (show cId)))+                stopTimeout+        sendInterrupt (_fromEndpoint (c^.childEndpoint)) NormalExitRequested+        r1 <- receiveSelectedMessage (   Right <$> selectProcessDown (c^.childMonitoring)+                                     <|> Left  <$> selectTimerElapsed t )+        demonitor (c^.childMonitoring)+        unlinkProcess (_fromEndpoint (c^.childEndpoint))+        case r1 of+          Left timerElapsed -> do+            logWarning (pack (show timerElapsed) <> ": child "<> pack (show cId) <>" => " <> pack(show (c^.childEndpoint)) <>" did not shutdown in time")+            let reason = interruptToExit (TimeoutInterrupt+                                           ("child did not shut down in time and was terminated by the "+                                             ++ show broker))+            sendShutdown (_fromEndpoint (c^.childEndpoint)) reason+            return reason+          Right downMsg -> do+            logInfo ("child "<> pack (show cId) <>" => " <> pack(show (c^.childEndpoint)) <>" terminated: " <> pack (show (downReason downMsg)))+            return (downReason downMsg)++stopAllChildren+  :: forall p e q0 .+     ( HasCallStack+     , HasProcesses e q0+     , Lifted IO e+     , Lifted IO q0+     , Member Logs e+     , Member (Stateful.ModelState (Broker p)) e+     , TangibleBroker p+     , Typeable (Effectful.ServerPdu p)+     )+  => Endpoint (Broker p) -> Timeout -> Eff e ()+stopAllChildren me stopTimeout =+  zoomToChildren @p+    (removeAllChildren @(ChildId p) @p)+  >>= pure . Map.assocs+  >>= traverse_ killAndNotify+  where+    killAndNotify (cId, c) = do+      reason <- provideInterrupts (stopOrKillChild cId c stopTimeout) >>= either crash return+      Stateful.zoomModel @(Broker p)+          childEventObserverLens+          (observerRegistryNotify @(ChildEvent p) (OnChildDown me cId (c^.childEndpoint) reason))++      where+        crash e = do+          logError (pack (show e) <> " while stopping child: " <> pack (show cId) <> " " <> pack (show c))+          return (interruptToExit e)

@@ -0,0 +1,117 @@+module Control.Eff.Concurrent.Protocol.Broker.InternalState+ ( Child(MkChild)+ , childMonitoring+ , childEndpoint+ , putChild+ , Children()+ , removeAllChildren+ , getChildren+ , lookupChildById+ , lookupAndRemoveChildById+ , lookupAndRemoveChildByMonitor+ )+ where++import Control.DeepSeq+import Control.Eff as Eff+import Control.Eff.Concurrent.Process+import Control.Eff.Concurrent.Protocol+import Control.Eff.Concurrent.Protocol.EffectfulServer+import Control.Eff.State.Strict as Eff+import Control.Lens hiding ((.=), use)+import Data.Default+import Data.Dynamic+import Data.Map (Map)+import GHC.Generics (Generic)+++data Child p = MkChild+  { _childEndpoint :: Endpoint (ServerPdu p)+  , _childMonitoring :: MonitorReference+  }+  deriving (Generic, Typeable, Eq, Ord)++instance NFData (Child o)++instance Typeable (ServerPdu p) => Show (Child p) where+  showsPrec d c = showParen (d>=10)+    (showString "process broker entry: " . shows (_childEndpoint c)  . showChar ' ' . shows (_childMonitoring c) )++makeLenses ''Child++-- | Internal state.+data Children i p = MkChildren+  { _childrenById :: Map i (Child p)+  , _childrenByMonitor :: Map MonitorReference (i, Child p)+  } deriving (Show, Generic, Typeable)++instance Default (Children i p) where+  def = MkChildren def def++instance (NFData i) => NFData (Children i p)++makeLenses ''Children++-- | State accessor+getChildren+  ::  (Ord i, Member (State (Children i o)) e)+  => Eff e (Children i o)+getChildren = Eff.get++putChild+  :: (Ord i, Member (State (Children i o)) e)+  => i+  -> Child o+  -> Eff e ()+putChild cId c = modify ( (childrenById . at cId .~ Just c)+                        . (childrenByMonitor . at (_childMonitoring c) .~ Just (cId, c))+                        )++lookupChildById+  :: (Ord i, Member (State (Children i o)) e)+  => i+  -> Eff e (Maybe (Child o))+lookupChildById i = view (childrenById . at i) <$> get++lookupChildByMonitor+  :: (Ord i, Member (State (Children i o)) e)+  => MonitorReference+  -> Eff e (Maybe (i, Child o))+lookupChildByMonitor m = view (childrenByMonitor . at m) <$> get++lookupAndRemoveChildById+  :: forall i o e. (Ord i, Member (State (Children i o)) e)+  => i+  -> Eff e (Maybe (Child o))+lookupAndRemoveChildById i =+  traverse go =<< lookupChildById i+  where+    go c = pure c <* removeChild i c++removeChild+  :: forall i o e. (Ord i, Member (State (Children i o)) e)+  => i+  -> Child o+  -> Eff e ()+removeChild i c = do+  modify @(Children i o) ( (childrenById . at i .~ Nothing)+                         . (childrenByMonitor . at (_childMonitoring c) .~ Nothing)+                         )++lookupAndRemoveChildByMonitor+  :: forall i o e. (Ord i, Member (State (Children i o)) e)+  => MonitorReference+  -> Eff e (Maybe (i, Child o))+lookupAndRemoveChildByMonitor r = do+  traverse go =<< lookupChildByMonitor r+  where+    go (i, c) = pure (i, c) <* removeChild i c++removeAllChildren+  :: forall i o e. (Ord i, Member (State (Children i o)) e)+  => Eff e (Map i (Child o))+removeAllChildren = do+  cm <- view childrenById <$> getChildren @i+  modify @(Children i o) (childrenById .~ mempty)+  modify @(Children i o) (childrenByMonitor .~ mempty)+  return cm

@@ -26,6 +26,7 @@ import           Control.Eff.Concurrent.Process.Timer import           Control.Eff.Log import           Data.Typeable                  ( Typeable )+import           Data.Text (pack) import           GHC.Stack  @@ -104,8 +105,6 @@      , TangiblePdu protocol ( 'Synchronous result)      , Tangible result      , Member Logs r-     , Lifted IO q-     , Lifted IO r      , HasCallStack      , Embeds destination protocol      )@@ -125,7 +124,10 @@             extractResult (Reply origin' result) =               if origin == origin' then Just result else Nothing         in selectMessageWith extractResult-  resultOrError <- receiveSelectedWithMonitorAfter pidInternal selectResult timeOut+  let timerTitle = MkProcessTitle ( "call-timer-" <> pack (show serverP)+                                  <> "-" <> pack (show origin)+                                  <> "-" <> pack (show timeOut))+  resultOrError <- receiveSelectedWithMonitorAfterWithTitle pidInternal selectResult timeOut timerTitle   let onTimeout timerRef = do         let msg = "call timed out after "                   ++ show timeOut ++ " to server: "

@@ -63,8 +63,8 @@   -- Usually you should rely on the default implementation   serverTitle :: Init a e -> ProcessTitle -  default serverTitle :: Typeable (ServerPdu a) => Init a e -> ProcessTitle-  serverTitle _ = fromString $ showSTypeable @(ServerPdu a) "-server"+  default serverTitle :: Typeable a => Init a e -> ProcessTitle+  serverTitle _ = fromString $ showSTypeable @a ""    -- | Process the effects of the implementation   runEffects :: Endpoint (ServerPdu a) -> Init a e -> Eff (ServerEffects a e) x -> Eff e x@@ -125,13 +125,10 @@   => Init a (Processes q) -> Eff (Processes q) () protocolServerLoop a = do   myEp <- asEndpoint @(ServerPdu a) <$> self-  let myEpTxt = T.pack . show $ myEp-  censorLogs (lmAddEp myEpTxt) $ do-    logDebug ("starting")-    runEffects  myEp a (receiveSelectedLoop sel (mainLoop myEp))-    return ()+  logDebug ("starting")+  runEffects  myEp a (receiveSelectedLoop sel (mainLoop myEp))+  return ()   where-    lmAddEp myEp = lmProcessId ?~ myEp     sel :: MessageSelector (Event (ServerPdu a))     sel =  onRequest <$> selectMessage @(Request (ServerPdu a))        <|> OnDown    <$> selectMessage @ProcessDown@@ -158,12 +155,19 @@   -- use 'toEmbeddedReplyTarget' to convert a 'ReplyTarget' safely to the embedded protocol.   --   -- @since 0.24.1-  OnCall :: forall a r. (Tangible r, TangiblePdu a ('Synchronous r)) => ReplyTarget a r -> Pdu a ('Synchronous r) -> Event a-  OnCast :: forall a. TangiblePdu a 'Asynchronous => Pdu a 'Asynchronous -> Event a-  OnInterrupt  :: (Interrupt 'Recoverable) -> Event a-  OnDown  :: ProcessDown -> Event a-  OnTimeOut  :: TimerElapsed -> Event a-  OnMessage  :: StrictDynamic -> Event a+  OnCall+    :: forall a r. (Tangible r, TangiblePdu a ('Synchronous r))+    => ReplyTarget a r+    -> Pdu a ('Synchronous r)+    -> Event a+  OnCast+    :: forall a. TangiblePdu a 'Asynchronous+    => Pdu a 'Asynchronous+    -> Event a+  OnInterrupt :: Interrupt 'Recoverable -> Event a+  OnDown :: ProcessDown -> Event a+  OnTimeOut :: TimerElapsed -> Event a+  OnMessage :: StrictDynamic -> Event a   deriving Typeable  instance Show (Event a) where

@@ -75,7 +75,7 @@  instance Typeable event => Show (Observer event) where   showsPrec d (MkObserver (Arg x (MkObservationSink _ m))) =-    showParen (d>=10) (showSTypeable @event . showString "-observer: " . shows x . showChar ' ' . shows m )+    showParen (d>=10) (showString "observer: " . showSTypeable @event . showString " ". showsPrec 10 x . showChar ' ' . showsPrec 10 m )  type instance ToPretty (Observer event) =   PrettyParens ("observing" <:> ToPretty event)@@ -90,7 +90,7 @@  instance Show event => Show (Pdu (Observer event) r) where   showsPrec d (Observed event) =-    showParen (d >= 10) (showString "observered: " . shows event)+    showParen (d >= 10) (showString "observered: " . showsPrec 10 event)  -- | The Information necessary to wrap an 'Observed' event to a process specific -- message, e.g. the embedded 'Observer' 'Pdu' instance, and the 'MonitorReference' of

@@ -2,7 +2,7 @@ -- | A small process to capture and _share_ observation's by enqueueing them into an STM 'TBQueue'. module Control.Eff.Concurrent.Protocol.Observer.Queue   ( ObservationQueue(..)-  , ObservationQueueReader+  , Reader   , observe   , await   , tryRead@@ -12,13 +12,14 @@  import           Control.Concurrent.STM import           Control.Eff+import           Control.Eff.Concurrent.Misc import           Control.Eff.Concurrent.Protocol import           Control.Eff.Concurrent.Protocol.Observer import           Control.Eff.Concurrent.Protocol.StatefulServer import           Control.Eff.Concurrent.Process import           Control.Eff.ExceptionExtra     ( ) import           Control.Eff.Log-import           Control.Eff.Reader.Strict+import qualified Control.Eff.Reader.Strict     as Eff import           Control.Exception.Safe        as Safe import           Control.Lens import           Control.Monad.IO.Class@@ -26,16 +27,17 @@ import qualified Data.Text                     as T import           Data.Typeable import           GHC.Stack+import Data.Default (Default)  -- | Contains a 'TBQueue' capturing observations. -- See 'observe'. newtype ObservationQueue a = ObservationQueue (TBQueue a)  -- | A 'Reader' for an 'ObservationQueue'.-type ObservationQueueReader a = Reader (ObservationQueue a)+type Reader a = Eff.Reader (ObservationQueue a)  logPrefix :: forall event proxy . (HasCallStack, Typeable event) => proxy event -> T.Text-logPrefix px = "observation queue: " <> T.pack (show (typeRep px))+logPrefix _px = "observation queue: " <> T.pack (showSTypeable @event "")  -- | Read queued observations captured and enqueued in the shared 'ObservationQueue' by 'observe'. --@@ -45,7 +47,7 @@ -- @since 0.28.0 await   :: forall event r-   . ( Member (ObservationQueueReader event) r+   . ( Member (Reader event) r      , HasCallStack      , MonadIO (Eff r)      , Typeable event@@ -53,7 +55,7 @@      )   => Eff r event await = do-  ObservationQueue q <- ask @(ObservationQueue event)+  ObservationQueue q <- Eff.ask @(ObservationQueue event)   liftIO (atomically (readTBQueue q))  -- | Read queued observations captured and enqueued in the shared 'ObservationQueue' by 'observe'.@@ -64,7 +66,7 @@ -- @since 0.28.0 tryRead   :: forall event r-   . ( Member (ObservationQueueReader event) r+   . ( Member (Reader event) r      , HasCallStack      , MonadIO (Eff r)      , Typeable event@@ -72,7 +74,7 @@      )   => Eff r (Maybe event) tryRead = do-  ObservationQueue q <- ask @(ObservationQueue event)+  ObservationQueue q <- Eff.ask @(ObservationQueue event)   liftIO (atomically (tryReadTBQueue q))  -- | Read at once all currently queued observations captured and enqueued@@ -84,7 +86,7 @@ -- @since 0.28.0 flush   :: forall event r-   . ( Member (ObservationQueueReader event) r+   . ( Member (Reader event) r      , HasCallStack      , MonadIO (Eff r)      , Typeable event@@ -92,7 +94,7 @@      )   => Eff r [event] flush = do-  ObservationQueue q <- ask @(ObservationQueue event)+  ObservationQueue q <- Eff.ask @(ObservationQueue event)   liftIO (atomically (flushTBQueue q))  -- | Listen to, and capture observations in an 'ObservationQueue'.@@ -113,7 +115,7 @@ -- -- foo = --   do---     observed <- start SomeObservable+--     observed <- startLink SomeObservable --     OQ.observe 100 observed $ do --       ... --       cast observed DoSomething@@ -123,11 +125,14 @@ -- -- @since 0.28.0 observe-  :: forall event eventSource e q len b+  :: forall event eventSource e q len b h   . ( HasCallStack     , HasProcesses e q-    , LogIo q-    , LogIo e+    , LogsTo h e+    , LogsTo h q+    , LogsTo h (Processes q)+    , Lifted IO e+    , Lifted IO q     , IsObservable eventSource event     , Integral len     , Server (ObservationQueue event) (Processes q)@@ -135,7 +140,7 @@     )   => len   -> Endpoint eventSource-  -> Eff (ObservationQueueReader event ': e) b+  -> Eff (Reader event ': e) b   -> Eff e b observe queueLimit eventSource e =   withObservationQueue queueLimit (withWriter @event eventSource e)@@ -152,12 +157,12 @@      , Member Interrupts e      )   => len-  -> Eff (ObservationQueueReader event ': e) b+  -> Eff (Reader event ': e) b   -> Eff e b withObservationQueue queueLimit e = do   q   <- lift (newTBQueueIO (fromIntegral queueLimit))   res <- handleInterrupts (return . Left)-                          (Right <$> runReader (ObservationQueue q) e)+                          (Right <$> Eff.runReader (ObservationQueue q) e)   rest <- lift (atomically (flushTBQueue q))   unless     (null rest)@@ -185,7 +190,7 @@   => ObservationQueue event   -> Eff r (Endpoint (Observer event)) spawnWriter q =-  start @_ @r @q @h (MkObservationQueue q)+  startLink @_ @r @q @h (MkObservationQueue q)  -- | Spawn a process that can be used as an 'Observer' that enqueues the observations into an --   'ObservationQueue'. See 'withObservationQueue'.@@ -196,19 +201,21 @@ -- -- @since 0.28.0 withWriter-  :: forall event eventSource e q b+  :: forall event eventSource e q b h   . ( HasCallStack     , HasProcesses e q-    , LogIo q+    , Lifted IO q+    , LogsTo h (Processes q)+    , Member Logs q     , IsObservable eventSource event-    , Member (ObservationQueueReader event) e+    , Member (Reader event) e     , Tangible (Pdu eventSource 'Asynchronous)     )   => Endpoint eventSource   -> Eff e b   -> Eff e b withWriter eventSource e = do-  q <- ask @(ObservationQueue event)+  q <- Eff.ask @(ObservationQueue event)   w <- spawnWriter @event q   registerObserver @event eventSource w   res <- e@@ -222,6 +229,8 @@    data instance StartArgument (ObservationQueue event) (Processes q) =      MkObservationQueue (ObservationQueue event)++  newtype instance Model (ObservationQueue event) = MkObservationQueueModel () deriving Default    update _ (MkObservationQueue (ObservationQueue q)) =     \case

@@ -5,8 +5,8 @@   ( Server(..)   , Stateful   , Effectful.Init(..)-  , start   , startLink+  , start   , ModelState   , modifyModel   , getAndModifyModel@@ -15,29 +15,38 @@   , putModel   , getAndPutModel   , useModel+  , preuseModel   , zoomModel+  , logModel   , SettingsReader   , askSettings   , viewSettings+  , mapEffects+  , coerceEffects   -- * Re-exports   , Effectful.Event(..)   )   where  import Control.Eff-import Control.Eff.Extend ()+import Control.Eff.Concurrent.Misc import Control.Eff.Concurrent.Process import Control.Eff.Concurrent.Protocol import qualified Control.Eff.Concurrent.Protocol.EffectfulServer as Effectful+import Control.Eff.Extend () import Control.Eff.Log import Control.Eff.Reader.Strict import Control.Eff.State.Strict import Control.Lens+import Data.Coerce import Data.Default import Data.Kind+import Data.String (fromString)+import Data.Text (Text, pack) import Data.Typeable import GHC.Stack (HasCallStack)-+import Control.Eff.Extend (raise)+import Data.Monoid (First) -- | A type class for server loops. -- -- This class serves as interface for other mechanisms, for example /process supervision/@@ -68,12 +77,21 @@   -- via the 'ModelState' effect.   -- The /model/ of a server loop is changed through incoming 'Event's.   -- It is initially calculated by 'setup'.-  type Model a :: Type-  type Model a = ()+  data family Model a :: Type+   -- | Type of read-only state.   type Settings a :: Type   type Settings a = () +  -- | Return a new 'ProcessTitle' for the stateful process,+  -- while it is running.+  --+  -- @since 0.30.0+  title :: StartArgument a q -> ProcessTitle++  default title :: Typeable a => StartArgument a q -> ProcessTitle+  title _ = fromString $ showSTypeable @a ""+   -- | Return an initial 'Model' and 'Settings'   setup ::        Endpoint (Protocol a)@@ -113,11 +131,12 @@    onEvent selfEndpoint (Init sa) = update selfEndpoint sa +  serverTitle (Init startArg) = title startArg  -- | Execute the server loop. -- -- @since 0.24.0-start+startLink   :: forall a r q h   . ( HasCallStack     , Typeable a@@ -127,23 +146,23 @@     , HasProcesses r q     )   => StartArgument a (Processes q) -> Eff r (Endpoint (Protocol a))-start = Effectful.start . Init+startLink = Effectful.startLink . Init --- | Execute the server loop.+-- | Execute the server loop. Please use 'startLink' if you can. -- -- @since 0.24.0-startLink+start   :: forall a r q h   . ( HasCallStack     , Typeable a-    , LogsTo h (Processes q)     , Effectful.Server (Stateful a) (Processes q)     , Server a (Processes q)+    , HandleLogWriter h+    , LogsTo h (Processes q)     , HasProcesses r q     )   => StartArgument a (Processes q) -> Eff r (Endpoint (Protocol a))-startLink = Effectful.startLink . Init-+start = Effectful.start . Init  -- | The 'Eff'ect type of mutable 'Model' in a 'Server' instance. --@@ -180,6 +199,12 @@ useModel :: forall a b e . Member (ModelState a) e => Getting b (Model a) b -> Eff e b useModel l = view l <$> getModel @a +-- | Return a element selected by a 'Lens' of the 'Model' of a 'Server'.+--+-- @since 0.30.0+preuseModel :: forall a b e . Member (ModelState a) e => Getting (First b) (Model a) b -> Eff e (Maybe b)+preuseModel l = preview l <$> getModel @a+ -- | Overwrite the 'Model' of a 'Server'. -- -- @since 0.24.0@@ -202,6 +227,18 @@   modifyModel @a (l .~ m1)   return c +-- | Log the 'Model' of a 'Server' using 'logDebug'.+--+-- @since 0.30.0+logModel+  :: forall m e q. ( Show (Model m)+                   , Member Logs e+                   , HasProcesses e q+                   , Member (ModelState m) e)+  => Text -> Eff e ()+logModel x =+  getModel @m >>= logDebug . (x <>) . pack . show+ -- | The 'Eff'ect type of readonly 'Settings' in a 'Server' instance. -- -- @since 0.24.0@@ -218,3 +255,57 @@ -- @since 0.24.0 viewSettings :: forall a b e . Member (SettingsReader a) e =>  Getting b (Settings a) b -> Eff e b viewSettings l = view l <$> askSettings @a++-- | Map 'ModelState' and 'SettingsReader' effects.+-- Use this to embed 'update' from another 'Server' instance.+--+-- @since 0.30.0+mapEffects+  :: forall inner outer a e.+     (Settings outer -> Settings inner) -- ^ A function to get the /inner/ settings out of the /outer/ settings+  -> Lens' (Model outer) (Model inner)  -- ^ A 'Lens' to get and set the /inner/ model inside the /outer/ model+  -> Eff (ModelState inner : SettingsReader inner : e) a+  -> Eff (ModelState outer : SettingsReader outer : e) a+mapEffects innerSettings innerStateLens innerEff =+  do st0 <- getModel @outer+     s0 <- askSettings @outer+     (res, st1) <-+      raise+        (raise+          (runReader+            @(Settings inner)+            (innerSettings s0)+            (runState+              @(Model inner)+              (st0 ^. innerStateLens)+              innerEff)))+     modifyModel @outer (innerStateLens .~ st1)+     return res+++-- | Coerce 'Coercible' 'ModelState' and 'SettingsReader' effects.+-- Use this to embed 'update' from a /similar/ 'Server' instance.+--+-- @since 0.30.0+coerceEffects+  :: forall inner outer a e.+     ( Coercible (Model inner) (Model outer)+     , Coercible (Model outer) (Model inner)+     , Coercible (Settings outer) (Settings inner)+     )+  => Eff (ModelState inner : SettingsReader inner : e) a+  -> Eff (ModelState outer : SettingsReader outer : e) a+coerceEffects innerEff =+  do st0 <- getModel @outer+     s0 <- askSettings @outer+     (res, st1) <-+      raise+        (raise+          (runReader+            @(Settings inner)+            (coerce s0)+            (runState+              (coerce @(Model outer) st0)+              innerEff)))+     putModel @outer (coerce st1)+     return res

@@ -1,457 +0,0 @@-{-# LANGUAGE UndecidableInstances #-}--- | A process supervisor spawns and monitors child processes.------ The child processes are mapped to symbolic identifier values: Child-IDs.------ This is the barest, most minimal version of a supervisor. Children--- can be started, but not restarted.------ Children can efficiently be looked-up by an id-value, and when--- the supervisor is shutdown, all children will be shutdown these--- are actually all the features of this supervisor implementation.------ Also, this minimalist supervisor only knows how to spawn a--- single kind of child process.------ When a supervisor spawns a new child process, it expects the--- child process to return a 'ProcessId'. The supervisor will--- 'monitor' the child process.------ This is in stark contrast to how Erlang/OTP handles things;--- In the OTP Supervisor, the child has to link to the parent.--- This allows the child spec to be more flexible in that no--- @pid@ has to be passed from the child start function to the--- supervisor process, and also, a child may break free from--- the supervisor by unlinking.------ Now while this seems nice at first, this might actually--- cause surprising results, since it is usually expected that--- stopping a supervisor also stops the children, or that a child--- exit shows up in the logging originating from the former--- supervisor.------ The approach here is to allow any child to link to the--- supervisor to realize when the supervisor was violently killed,--- and otherwise give the child no chance to unlink itself from--- its supervisor.------ This module is far simpler than the Erlang/OTP counter part,--- of a @simple_one_for_one@ supervisor.------ The future of this supervisor might not be a-lot more than--- it currently is. The ability to restart processes might be--- implemented outside of this supervisor module.------ One way to do that is to implement the restart logic in--- a separate module, since the child-id can be reused when a child--- exits.------ @since 0.23.0-module Control.Eff.Concurrent.Protocol.Supervisor-  ( startSupervisor-  , stopSupervisor-  , isSupervisorAlive-  , monitorSupervisor-  , getDiagnosticInfo-  , spawnChild-  , spawnOrLookup-  , lookupChild-  , stopChild-  , Sup()-  , ChildId-  , Stateful.StartArgument(MkSupConfig)-  , supConfigChildStopTimeout-  , SpawnErr(AlreadyStarted)-  ) where--import Control.DeepSeq (NFData(rnf))-import Control.Eff as Eff-import Control.Eff.Concurrent.Protocol-import Control.Eff.Concurrent.Protocol.Client-import Control.Eff.Concurrent.Protocol.Wrapper-import qualified Control.Eff.Concurrent.Protocol.EffectfulServer as Effectful-import qualified Control.Eff.Concurrent.Protocol.StatefulServer as Stateful-import Control.Eff.Concurrent.Protocol.Supervisor.InternalState-import Control.Eff.Concurrent.Process-import Control.Eff.Concurrent.Process.Timer-import Control.Eff.Extend (raise)-import Control.Eff.Log-import Control.Eff.State.Strict as Eff-import Control.Lens hiding ((.=), use)-import Data.Default-import Data.Dynamic-import Data.Foldable-import Data.Kind-import qualified Data.Map as Map-import Data.Text (Text, pack)-import Data.Type.Pretty-import GHC.Generics (Generic)-import GHC.Stack-import Control.Applicative ((<|>))---- * Supervisor Server API----- ** Functions---- | Start and link a new supervisor process with the given 'SpawnFun'unction.------ To spawn new child processes use 'spawnChild'.------ @since 0.23.0-startSupervisor-  :: forall p e-  . ( HasCallStack-    , LogIo (Processes e)-    , TangibleSup p-    , Stateful.Server (Sup p) (Processes e)-    )-  => Stateful.StartArgument (Sup p) (Processes e)-  -> Eff (Processes e) (Endpoint (Sup p))-startSupervisor = Stateful.start---- | Stop the supervisor and shutdown all processes.------ Block until the supervisor has finished.------ @since 0.23.0-stopSupervisor-  :: ( HasCallStack-     , HasProcesses e q-     , Member Logs e-     , Lifted IO e-     , TangibleSup p-     )-  => Endpoint (Sup p)-  -> Eff e ()-stopSupervisor ep = do-  logInfo ("stopping supervisor: " <> pack (show ep))-  mr <- monitor (_fromEndpoint ep)-  sendInterrupt (_fromEndpoint ep) NormalExitRequested-  r <- receiveSelectedMessage (selectProcessDown mr)-  logInfo ("supervisor stopped: " <> pack (show ep) <> " " <> pack (show r))---- | Check if a supervisor process is still alive.------ @since 0.23.0-isSupervisorAlive-  :: forall p q0 e .-     ( HasCallStack-     , Member Logs e-     , Typeable p-     , HasProcesses e q0-     )-  => Endpoint (Sup p)-  -> Eff e Bool-isSupervisorAlive x = isProcessAlive (_fromEndpoint x)---- | Monitor a supervisor process.------ @since 0.23.0-monitorSupervisor-  :: forall p q0 e .-     ( HasCallStack-     , Member Logs e-     , HasProcesses e q0-     , TangibleSup p-     )-  => Endpoint (Sup p)-  -> Eff e MonitorReference-monitorSupervisor x = monitor (_fromEndpoint x)---- | Start and monitor a new child process using the 'SpawnFun' passed--- to 'startSupervisor'.------ @since 0.23.0-spawnChild-  :: forall p q0 e .-     ( HasCallStack-     , Member Logs e-     , HasProcesses e q0-     , TangibleSup p-     , Typeable (Effectful.ServerPdu p)-     )-  => Endpoint (Sup p)-  -> ChildId p-  -> Eff e (Either (SpawnErr p) (Endpoint (Effectful.ServerPdu p)))-spawnChild ep cId = call ep (StartC cId)---- | Start and monitor a new child process using the 'SpawnFun' passed--- to 'startSupervisor'.------ Call 'spawnChild' and unpack the 'Either' result,--- ignoring the 'AlreadyStarted' error.------ @since 0.29.2-spawnOrLookup-  :: forall p q0 e .-     ( HasCallStack-     , Member Logs e-     , HasProcesses e q0-     , TangibleSup p-     , Typeable (Effectful.ServerPdu p)-     )-  => Endpoint (Sup p)-  -> ChildId p-  -> Eff e (Endpoint (Effectful.ServerPdu p))-spawnOrLookup supEp cId =-  do res <- spawnChild supEp cId-     pure $-      case res of-        Left (AlreadyStarted _ ep) -> ep-        Right ep -> ep---- | Lookup the given child-id and return the output value of the 'SpawnFun'---   if the client process exists.------ @since 0.23.0-lookupChild ::-    forall p e q0 .-     ( HasCallStack-     , Member Logs e-     , HasProcesses e q0-     , TangibleSup p-     , Typeable (Effectful.ServerPdu p)-     )-  => Endpoint (Sup p)-  -> ChildId p-  -> Eff e (Maybe (Endpoint (Effectful.ServerPdu p)))-lookupChild ep cId = call ep (LookupC @p cId)---- | Stop a child process, and block until the child has exited.------ Return 'True' if a process with that ID was found, 'False' if no process--- with the given ID was running.------ @since 0.23.0-stopChild ::-    forall p e q0 .-     ( HasCallStack-     , Member Logs e-     , HasProcesses e q0-     , TangibleSup p-     )-  => Endpoint (Sup p)-  -> ChildId p-  -> Eff e Bool-stopChild ep cId = call ep (StopC @p cId (TimeoutMicros 4000000))---- | Return a 'Text' describing the current state of the supervisor.------ @since 0.23.0-getDiagnosticInfo-  :: forall p e q0 .-     ( HasCallStack-     , HasProcesses e q0-     , TangibleSup p-     )-  => Endpoint (Sup p)-  -> Eff e Text-getDiagnosticInfo s = call s (GetDiagnosticInfo @p)---- ** Types---- | The index type of 'Server' supervisors.------ A @'Sup' p@ manages the life cycle of the processes, running the @'Server' p@--- methods of that specific type.------ The supervisor maps an identifier value of type @'ChildId' p@ to an @'Endpoint' p@.------ @since 0.24.0-data Sup (p :: Type) deriving Typeable--instance Typeable p => HasPdu (Sup p) where-  -- | The 'Pdu' instance contains methods to start, stop and lookup a child-  -- process, as well as a diagnostic callback.-  ---  -- @since 0.23.0-  data instance  Pdu (Sup p) r where-          StartC :: ChildId p -> Pdu (Sup p) ('Synchronous (Either (SpawnErr p) (Endpoint (Effectful.ServerPdu p))))-          StopC :: ChildId p -> Timeout -> Pdu (Sup p) ('Synchronous Bool)-          LookupC :: ChildId p -> Pdu (Sup p) ('Synchronous (Maybe (Endpoint (Effectful.ServerPdu p))))-          GetDiagnosticInfo :: Pdu (Sup p) ('Synchronous Text)-      deriving Typeable--instance (Show (ChildId p)) => Show (Pdu (Sup p) ('Synchronous r)) where-  showsPrec d (StartC c) = showParen (d >= 10) (showString "StartC " . showsPrec 10 c)-  showsPrec d (StopC c t) = showParen (d >= 10) (showString "StopC " . showsPrec 10 c . showChar ' ' . showsPrec 10 t)-  showsPrec d (LookupC c) = showParen (d >= 10) (showString "LookupC " . showsPrec 10 c)-  showsPrec _ GetDiagnosticInfo = showString "GetDiagnosticInfo"--instance (NFData (ChildId p)) => NFData (Pdu (Sup p) ('Synchronous r)) where-  rnf (StartC ci) = rnf ci-  rnf (StopC ci t) = rnf ci `seq` rnf t-  rnf (LookupC ci) = rnf ci-  rnf GetDiagnosticInfo = ()--type instance ToPretty (Sup p) = "supervisor" <:> ToPretty p---- | The type of value used to index running 'Server' processes managed by a 'Sup'.------ Note, that the type you provide must be 'Tangible'.------ @since 0.24.0-type family ChildId p---- | Constraints on the parameters to 'Sup'.------ @since 0.24.0-type TangibleSup p =-  ( Tangible (ChildId p)-  , Ord (ChildId p)-  , Typeable p-  )---instance-  ( LogIo q-  , TangibleSup p-  , Tangible (ChildId p)-  , Typeable (Effectful.ServerPdu p)-  , Effectful.Server p (Processes q)-  , HasProcesses (Effectful.ServerEffects p (Processes q)) q-  ) => Stateful.Server (Sup p) (Processes q) where--  -- | Options that control the 'Sup p' process.-  ---  -- This contains:-  ---  -- * a 'SpawnFun'-  -- * the 'Timeout' after requesting a normal child exit before brutally killing the child.-  ---  -- @since 0.24.0-  data StartArgument (Sup p) (Processes q) = MkSupConfig-    {-      -- , supConfigChildRestartPolicy :: ChildRestartPolicy-      -- , supConfigResilience :: Resilience-      supConfigChildStopTimeout :: Timeout-    , supConfigStartFun :: ChildId p -> Effectful.Init p (Processes q)-    }--  type Model (Sup p) = Children (ChildId p) p--  setup _ _cfg = pure (def, ())-  update _ supConfig (Stateful.OnCall rt req) =-    case req of-      GetDiagnosticInfo ->  do-        p <- (pack . show <$> getChildren @(ChildId p) @p)-        sendReply rt p--      LookupC i -> do-        p <- fmap _childEndpoint <$> lookupChildById @(ChildId p) @p i-        sendReply rt p--      StopC i t -> do-        mExisting <- lookupAndRemoveChildById @(ChildId p) @p i-        case mExisting of-          Nothing -> sendReply rt False-          Just existingChild -> do-            stopOrKillChild i existingChild t-            sendReply rt True--      StartC i -> do-        mExisting <- lookupChildById @(ChildId p) @p i-        case mExisting of-          Nothing -> do-            childEp <- raise (raise (Effectful.start (supConfigStartFun supConfig i)))-            let childPid = _fromEndpoint childEp-            cMon <- monitor childPid-            putChild i (MkChild @p childEp cMon)-            sendReply rt (Right childEp)-          Just existingChild ->-            sendReply rt (Left (AlreadyStarted i (existingChild ^. childEndpoint)))--  update _ _supConfig (Stateful.OnDown pd) = do-      oldEntry <- lookupAndRemoveChildByMonitor @(ChildId p) @p (downReference pd)-      case oldEntry of-        Nothing -> logWarning ("unexpected: " <> pack (show pd))-        Just (i, c) -> do-          logInfo (  pack (show pd)-                  <> " for child "-                  <> pack (show i)-                  <> " => "-                  <> pack (show (_childEndpoint c))-                  )-  update _ supConfig (Stateful.OnInterrupt e) = do-      let (logSev, exitReason) =-            case e of-              NormalExitRequested ->-                (debugSeverity, ExitNormally)-              _ ->-                (warningSeverity, interruptToExit (ErrorInterrupt ("supervisor interrupted: " <> show e)))-      stopAllChildren @p (supConfigChildStopTimeout supConfig)-      logWithSeverity logSev ("supervisor stopping: " <> pack (show e))-      exitBecause exitReason--  update _ _supConfig o = logWarning ("unexpected: " <> pack (show o))----- | Runtime-Errors occurring when spawning child-processes.------ @since 0.23.0-data SpawnErr p = AlreadyStarted (ChildId p) (Endpoint (Effectful.ServerPdu p))-  deriving (Typeable, Generic)--deriving instance Eq (ChildId p) => Eq (SpawnErr p)-deriving instance Ord (ChildId p) => Ord (SpawnErr p)-deriving instance (Typeable (Effectful.ServerPdu p), Show (ChildId p)) => Show (SpawnErr p)--instance NFData (ChildId p) => NFData (SpawnErr p)---- Internal Functions--stopOrKillChild-  :: forall p e q0 .-     ( HasCallStack-     , HasProcesses e q0-     , Lifted IO e-     , Lifted IO q0-     , Member Logs e-     , Member (State (Children (ChildId p) p)) e-     , TangibleSup p-     , Typeable (Effectful.ServerPdu p)-     )-  => ChildId p-  -> Child p-  -> Timeout-  -> Eff e ()-stopOrKillChild cId c stopTimeout =-      do-        sup <- asEndpoint @(Sup p) <$> self-        t <- startTimer stopTimeout-        sendInterrupt (_fromEndpoint (c^.childEndpoint)) NormalExitRequested-        r1 <- receiveSelectedMessage (   Right <$> selectProcessDown (c^.childMonitoring)-                                     <|> Left  <$> selectTimerElapsed t )-        demonitor (c^.childMonitoring)-        unlinkProcess (_fromEndpoint (c^.childEndpoint))-        case r1 of-          Left timerElapsed -> do-            logWarning (pack (show timerElapsed) <> ": child "<> pack (show cId) <>" => " <> pack(show (c^.childEndpoint)) <>" did not shutdown in time")-            sendShutdown-              (_fromEndpoint (c^.childEndpoint))-              (interruptToExit-                (TimeoutInterrupt-                  ("child did not shut down in time and was terminated by the "-                    ++ show sup)))-          Right downMsg ->-            logInfo ("child "<> pack (show cId) <>" => " <> pack(show (c^.childEndpoint)) <>" terminated: " <> pack (show (downReason downMsg)))--stopAllChildren-  :: forall p e q0 .-     ( HasCallStack-     , HasProcesses e q0-     , Lifted IO e-     , Lifted IO q0-     , Member Logs e-     , Member (State (Children (ChildId p) p)) e-     , TangibleSup p-     , Typeable (Effectful.ServerPdu p)-     )-  => Timeout -> Eff e ()-stopAllChildren stopTimeout = removeAllChildren @(ChildId p) @p >>= pure . Map.assocs >>= traverse_ xxx-  where-    xxx (cId, c) = provideInterrupts (stopOrKillChild cId c stopTimeout) >>= either crash return-      where-        crash e = do-          logError (pack (show e) <> " while stopping child: " <> pack (show cId) <> " " <> pack (show c))

@@ -1,105 +0,0 @@-module Control.Eff.Concurrent.Protocol.Supervisor.InternalState where--import Control.DeepSeq-import Control.Eff as Eff-import Control.Eff.Concurrent.Process-import Control.Eff.Concurrent.Protocol-import Control.Eff.Concurrent.Protocol.EffectfulServer-import Control.Eff.State.Strict as Eff-import Control.Lens hiding ((.=), use)-import Data.Default-import Data.Dynamic-import Data.Map (Map)-import GHC.Generics (Generic)---data Child p = MkChild-  { _childEndpoint :: Endpoint (ServerPdu p)-  , _childMonitoring :: MonitorReference-  }-  deriving (Generic, Typeable, Eq, Ord)--instance NFData (Child o)--instance Typeable (ServerPdu p) => Show (Child p) where-  showsPrec d c = showParen (d>=10)-    (showString "supervised process: " . shows (_childEndpoint c)  . showChar ' ' . shows (_childMonitoring c) )--makeLenses ''Child---- | Internal state.-data Children i p = MkChildren-  { _childrenById :: Map i (Child p)-  , _childrenByMonitor :: Map MonitorReference (i, Child p)-  } deriving (Show, Generic, Typeable)--instance Default (Children i p) where-  def = MkChildren def def--instance (NFData i) => NFData (Children i p)--makeLenses ''Children---- | State accessor-getChildren-  ::  (Ord i, Member (State (Children i o)) e)-  => Eff e (Children i o)-getChildren = Eff.get--putChild-  :: (Ord i, Member (State (Children i o)) e)-  => i-  -> Child o-  -> Eff e ()-putChild cId c = modify ( (childrenById . at cId .~ Just c)-                        . (childrenByMonitor . at (_childMonitoring c) .~ Just (cId, c))-                        )--lookupChildById-  :: (Ord i, Member (State (Children i o)) e)-  => i-  -> Eff e (Maybe (Child o))-lookupChildById i = view (childrenById . at i) <$> get--lookupChildByMonitor-  :: (Ord i, Member (State (Children i o)) e)-  => MonitorReference-  -> Eff e (Maybe (i, Child o))-lookupChildByMonitor m = view (childrenByMonitor . at m) <$> get--lookupAndRemoveChildById-  :: forall i o e. (Ord i, Member (State (Children i o)) e)-  => i-  -> Eff e (Maybe (Child o))-lookupAndRemoveChildById i =-  traverse go =<< lookupChildById i-  where-    go c = pure c <* removeChild i c--removeChild-  :: forall i o e. (Ord i, Member (State (Children i o)) e)-  => i-  -> Child o-  -> Eff e ()-removeChild i c = do-  modify @(Children i o) ( (childrenById . at i .~ Nothing)-                         . (childrenByMonitor . at (_childMonitoring c) .~ Nothing)-                         )--lookupAndRemoveChildByMonitor-  :: forall i o e. (Ord i, Member (State (Children i o)) e)-  => MonitorReference-  -> Eff e (Maybe (i, Child o))-lookupAndRemoveChildByMonitor r = do-  traverse go =<< lookupChildByMonitor r-  where-    go (i, c) = pure (i, c) <* removeChild i c--removeAllChildren-  :: forall i o e. (Ord i, Member (State (Children i o)) e)-  => Eff e (Map i (Child o))-removeAllChildren = do-  cm <- view childrenById <$> getChildren @i-  modify @(Children i o) (childrenById .~ mempty)-  modify @(Children i o) (childrenByMonitor .~ mempty)-  return cm

@@ -0,0 +1,616 @@+{-# LANGUAGE UndecidableInstances #-}+-- | Monitor a process and act when it is unresponsive.+--+-- Behaviour of the watchdog:+--+-- When a child crashes:+-- * if the allowed maximum number crashes per time span has been reached for the process,+-- ** cancel all other timers+-- ** don't start the child again+-- ** if this is a /permanent/ watchdog crash the watchdog+-- * otherwise+-- ** tell the broker to start the child+-- ** record a crash and start a timer to remove the record later+-- ** monitor the child+--+-- When a child crash timer elapses:+-- * remove the crash record+--+-- @since 0.30.0+module Control.Eff.Concurrent.Protocol.Watchdog+  ( startLink+  , Watchdog+  , attachTemporary+  , attachPermanent+  , getCrashReports+  , CrashRate(..)+  , crashCount+  , crashTimeSpan+  , crashesPerSeconds+  , CrashCount+  , CrashTimeSpan+  , ChildWatch(..)+  , parent+  , crashes+  , ExonerationTimer(..)+  , CrashReport(..)+  , crashTime+  , crashReason+  , exonerationTimerReference+  ) where++import Control.DeepSeq+import Control.Eff (Eff, Member, lift, Lifted)+import Control.Eff.Concurrent.Misc+import Control.Eff.Concurrent.Process+import Control.Eff.Concurrent.Process.Timer+import Control.Eff.Concurrent.Protocol+import Control.Eff.Concurrent.Protocol.Client+import Control.Eff.Concurrent.Protocol.Wrapper+import qualified Control.Eff.Concurrent.Protocol.Observer as Observer+import Control.Eff.Concurrent.Protocol.Observer (Observer)+import qualified Control.Eff.Concurrent.Protocol.Broker as Broker+import Control.Eff.Concurrent.Protocol.Broker (Broker)+import qualified Control.Eff.Concurrent.Protocol.StatefulServer as Stateful+import qualified Control.Eff.Concurrent.Protocol.EffectfulServer as Effectful+import Control.Lens+import Control.Eff.Log+import Data.Set (Set)+import Data.Typeable+import qualified Data.Set as Set+import Data.Map (Map)+import qualified Data.Map as Map+import Data.Time.Clock+import Data.Kind (Type)+import Data.Default+import Data.Text (pack)+import GHC.Stack (HasCallStack)+import Data.Maybe (isJust)+import Data.Foldable (traverse_, forM_)+import Control.Monad (when)+++-- | The phantom for watchdog processes, that watch the given type of servers+--+-- This type is used for the 'Effectful.Server' and 'HasPdu' instances.+--+-- @since 0.30.0+data Watchdog (child :: Type) deriving Typeable++-- | Start and link a new watchdog process.+--+-- The watchdog process will register itself to the 'Broker.ChildEvent's and+-- restart crashed children.+--+-- @since 0.30.0+startLink+  :: forall child e q h+  . ( HasCallStack+    , Typeable child+    , LogsTo h (Processes q)+    , Member Logs q+    , HasProcesses e q+    , Tangible (Broker.ChildId child)+    , Ord (Broker.ChildId child)+    , HasPdu (Effectful.ServerPdu child)+    , Lifted IO q+    )+  => CrashRate -> Eff e (Endpoint (Watchdog child))+startLink = Stateful.startLink @(Watchdog child) . StartWatchDog++-- | Restart children of the given broker.+--+-- When the broker exits, ignore the children of that broker.+--+-- @since 0.30.0+attachTemporary+  :: forall child q e h+  . ( HasCallStack+    , LogsTo h e+    , Typeable child+    , HasPdu (Effectful.ServerPdu child)+    , Tangible (Broker.ChildId child)+    , Ord (Broker.ChildId child)+    , HasProcesses e q+    )+  => Endpoint (Watchdog child) -> Endpoint (Broker child) -> Eff e ()+attachTemporary wd broker =+  callWithTimeout wd (Attach broker False) (TimeoutMicros 1_000_000)++-- | Restart children of the given broker.+--+-- When the broker exits, the watchdog process will exit, too.+--+-- @since 0.30.0+attachPermanent+  :: forall child q e h+  . ( HasCallStack+    , LogsTo h e+    , Typeable child+    , HasPdu (Effectful.ServerPdu child)+    , Tangible (Broker.ChildId child)+    , Ord (Broker.ChildId child)+    , HasProcesses e q+    )+  => Endpoint (Watchdog child) -> Endpoint (Broker child) -> Eff e ()+attachPermanent wd broker =+  callWithTimeout wd (Attach broker True) (TimeoutMicros 1_000_000)++-- | Return a list of 'CrashReport's.+--+-- Useful for diagnostics+--+-- @since 0.30.0+getCrashReports+  :: forall child q e h+  . ( HasCallStack+    , LogsTo h e+    , Typeable child+    , HasPdu (Effectful.ServerPdu child)+    , Tangible (Broker.ChildId child)+    , Ord (Broker.ChildId child)+    , HasProcesses e q+    , Lifted IO q+    , Lifted IO e+    , Member Logs e+    )+  => Endpoint (Watchdog child) -> Eff e (Map (Broker.ChildId child) (ChildWatch child))+getCrashReports wd = callWithTimeout wd GetCrashReports (TimeoutMicros 5_000_000)++instance Typeable child => HasPdu (Watchdog child) where+  type instance EmbeddedPduList (Watchdog child) = '[Observer (Broker.ChildEvent child)]+  data Pdu (Watchdog child) r where+    Attach :: Endpoint (Broker child) -> Bool -> Pdu (Watchdog child) ('Synchronous ())+    GetCrashReports :: Pdu (Watchdog child) ('Synchronous (Map (Broker.ChildId child) (ChildWatch child)))+    OnChildEvent :: Broker.ChildEvent child -> Pdu (Watchdog child) 'Asynchronous+      deriving Typeable++instance Typeable child => HasPduPrism (Watchdog child) (Observer (Broker.ChildEvent child)) where+  embedPdu (Observer.Observed e) = OnChildEvent e+  fromPdu (OnChildEvent x) = Just (Observer.Observed x)+  fromPdu _ = Nothing++instance (NFData (Broker.ChildId child)) => NFData (Pdu (Watchdog child) r) where+  rnf (Attach e b) = rnf e `seq` rnf b `seq` ()+  rnf GetCrashReports = ()+  rnf (OnChildEvent o) = rnf o++instance+  ( Show (Broker.ChildId child)+  , Typeable child+  , Typeable (Effectful.ServerPdu child)+  )+  => Show (Pdu (Watchdog child) r) where+  showsPrec d (Attach e False) = showParen (d>=10) (showString "attach-temporary: " . shows e)+  showsPrec d (Attach e True) = showParen (d>=10) (showString "attach-permanent: " . shows e)+  showsPrec _ GetCrashReports = showString "get-crash-reports"+  showsPrec d (OnChildEvent o) = showParen (d>=10) (showString "on-child-event: " . showsPrec 10 o)++-- ------------------ Broker Watches++data BrokerWatch =+  MkBrokerWatch { _brokerMonitor ::  MonitorReference, _isPermanent :: Bool }+  deriving (Typeable)++instance Show BrokerWatch where+  showsPrec d (MkBrokerWatch mon False) = showParen (d>=10) (showString "temporary-broker: " . showsPrec 10 mon)+  showsPrec d (MkBrokerWatch mon True) = showParen (d>=10) (showString "permanent-broker: " . showsPrec 10 mon)++brokerMonitor :: Lens' BrokerWatch MonitorReference+brokerMonitor = lens _brokerMonitor (\(MkBrokerWatch _ x) m -> MkBrokerWatch m x)++isPermanent :: Lens' BrokerWatch Bool+isPermanent = lens _isPermanent (\(MkBrokerWatch x _) m -> MkBrokerWatch x m)++-- --- Server Definition++instance+  ( Typeable child+  , HasPdu (Effectful.ServerPdu child)+  , Tangible (Broker.ChildId child)+  , Ord (Broker.ChildId child)+  , Eq (Broker.ChildId child)+  , Lifted IO e+  , Member Logs e+  ) => Stateful.Server (Watchdog child) (Processes e) where++  data instance StartArgument (Watchdog child) (Processes e) =+    StartWatchDog { _crashRate :: CrashRate+                  }+      deriving Typeable++  data instance Model (Watchdog child) =+    WatchdogModel { _brokers :: Map (Endpoint (Broker child)) BrokerWatch+                  , _watched :: Map (Broker.ChildId child) (ChildWatch child)+                  }++  update me startArg =+    \case+      Effectful.OnCall rt (Attach broker permanent) -> do+        logDebug (  "attaching "+                 <> if permanent then "permanently" else "temporarily"+                 <> " to: " <> pack (show broker)+                 )+        oldMonitor <- Stateful.preuseModel @(Watchdog child) (brokers . at broker . _Just . brokerMonitor)+        newMonitor <- maybe (monitor (broker^.fromEndpoint)) return oldMonitor+        case oldMonitor of+          Nothing -> do+            logDebug ("start observing: " <> pack (show broker))+            Observer.registerObserver @(Broker.ChildEvent child) broker me+          Just _ ->+            logDebug ("already observing " <> pack (show broker))+        let newBrokerWatch = MkBrokerWatch newMonitor permanent+        Stateful.modifyModel (brokers . at broker ?~ newBrokerWatch)+        Observer.registerObserver @(Broker.ChildEvent child) broker me+        sendReply rt ()++      Effectful.OnCall rt GetCrashReports ->+        Stateful.useModel @(Watchdog child) watched >>= sendReply rt++      Effectful.OnCast (OnChildEvent e) ->+        case e of+          down@(Broker.OnBrokerShuttingDown broker) -> do+            logInfo ("received: " <> pack (show down))+            removeBroker me broker++          spawned@(Broker.OnChildSpawned broker _ _) -> do+            logInfo ("received: " <> pack (show spawned))+            currentModel <- Stateful.getModel @(Watchdog child)+            when (not (Set.member broker (currentModel ^. brokers . to Map.keysSet)))+             (logWarning ("received child event for unknown broker: " <> pack (show spawned)))++          down@(Broker.OnChildDown broker cId _ ExitNormally) -> do+            logInfo ("received: " <> pack (show down))+            currentModel <- Stateful.getModel @(Watchdog child)+            if not (Set.member broker (currentModel ^. brokers . to Map.keysSet))+             then logWarning ("received child event for unknown broker: " <> pack (show down))+             else removeAndCleanChild @child cId++          down@(Broker.OnChildDown broker cId _ reason) -> do+            logInfo ("received: " <> pack (show down))+            currentModel <- Stateful.getModel @(Watchdog child)+            if not (Set.member broker (currentModel ^. brokers . to Map.keysSet))+             then+              logWarning ("received child event for unknown broker: " <> pack (show down))+             else do+              let recentCrashes = countRecentCrashes broker cId currentModel+                  rate = startArg ^. crashRate+                  maxCrashCount = rate ^. crashCount+              if recentCrashes < maxCrashCount then do+                logNotice ("restarting (" <> pack (show recentCrashes) <> "/" <> pack (show maxCrashCount) <> "): "+                            <> pack (show cId) <> " of " <> pack (show broker))+                res <- Broker.spawnChild broker cId+                logNotice ("restarted: " <> pack (show cId) <> " of "+                            <> pack (show broker) <> ": " <> pack (show res))+                crash <- startExonerationTimer @child cId reason (rate ^. crashTimeSpan)+                if isJust (currentModel ^? childWatchesById cId)+                  then do+                    logDebug ("recording crash for child: " <> pack (show cId) <> " of " <> pack (show broker))+                    Stateful.modifyModel (watched @child . at cId . _Just . crashes %~ Set.insert crash)+                  else do+                    logDebug ("recording crash for new child: " <> pack (show cId) <> " of " <> pack (show broker))+                    Stateful.modifyModel (watched @child . at cId .~ Just (MkChildWatch broker (Set.singleton crash)))+              else do+                logWarning ("restart rate exceeded: " <> pack (show rate)+                            <> ", for child: " <> pack (show cId)+                            <> " of " <> pack (show broker))+                removeAndCleanChild @child cId+                forM_ (currentModel ^? brokers . at broker . _Just) $ \bw ->+                  if  bw ^. isPermanent then do+                    logError ("a child of a permanent broker crashed too often, interrupting: " <> pack (show broker))+                    let r =  ExitUnhandledError "restart frequency exceeded"+                    demonitor (bw ^. brokerMonitor)+                    sendShutdown (broker ^. fromEndpoint) r+                    exitBecause r+                    -- TODO shutdown all other permanent brokers!+                  else+                    logError ("a child of a temporary broker crashed too often: " <> pack (show broker))++      Effectful.OnDown pd@(ProcessDown _mref _ pid) -> do+        logDebug ("received " <> pack (show pd))+        let broker = asEndpoint pid+        removeBroker @child me broker++      Effectful.OnTimeOut t -> do+        logError ("received: " <> pack (show t))++      Effectful.OnMessage (fromStrictDynamic -> Just (MkExonerationTimer cId ref :: ExonerationTimer (Broker.ChildId child))) -> do+        logInfo ("exonerating: " <> pack (show cId))+        Stateful.modifyModel+          (watched @child . at cId . _Just . crashes %~ Set.filter (\c -> c^.exonerationTimerReference /= ref))++      Effectful.OnMessage t -> do+        logError ("received: " <> pack (show t))++      Effectful.OnInterrupt reason -> do+        logError ("received: " <> pack (show reason))++-- ------------------ Start Argument++crashRate :: Lens' (Stateful.StartArgument (Watchdog child) (Processes e)) CrashRate+crashRate = lens _crashRate (\m x -> m {_crashRate = x})++-- ----------------- Crash Rate++-- | The limit of crashes (see 'CrashCount') per time span (see 'CrashTimeSpan') that justifies restarting+-- child processes.+--+-- Used as parameter for 'startLink'.+--+-- Use 'crashesPerSeconds' to construct a value.+--+-- This governs how long the 'ExonerationTimer' runs before cleaning up a 'CrashReport' in a 'ChildWatch'.+--+-- @since 0.30.0+data CrashRate =+  CrashesPerSeconds { _crashCount :: CrashCount+                    , _crashTimeSpan :: CrashTimeSpan+                    }+  deriving (Typeable, Eq, Ord)++-- | The default is three crashes in 30 seconds.+--+-- @since 0.30.0+instance Default CrashRate where+  def = 3 `crashesPerSeconds` 30++instance Show CrashRate where+  showsPrec d (CrashesPerSeconds count time) =+    showParen (d>=7) (shows count . showString " crashes/" . shows time . showString " seconds")++instance NFData CrashRate where+  rnf (CrashesPerSeconds c t) = c `seq` t `seq` ()++-- | Number of crashes in 'CrashRate'.+--+-- @since 0.30.0+type CrashCount = Int++-- | Time span in which crashes are counted in 'CrashRate'.+--+-- @since 0.30.0+type CrashTimeSpan = Int++-- | A smart constructor for 'CrashRate'.+--+-- The first parameter is the number of crashes allowed per number of seconds (second parameter)+-- until the watchdog should give up restarting a child.+--+-- @since 0.30.0+crashesPerSeconds :: CrashCount -> CrashTimeSpan -> CrashRate+crashesPerSeconds = CrashesPerSeconds++-- | A lens for '_crashCount'.+--+-- @since 0.30.0+crashCount :: Lens' CrashRate CrashCount+crashCount = lens _crashCount (\(CrashesPerSeconds _ time) count -> CrashesPerSeconds count time )++-- | A lens for '_crashTimeSpan'.+--+-- @since 0.30.0+crashTimeSpan :: Lens' CrashRate CrashTimeSpan+crashTimeSpan = lens _crashTimeSpan (\(CrashesPerSeconds count _) time -> CrashesPerSeconds count time)++-- ------------------ Crash++-- | An internal data structure that records a single crash of a child of an attached 'Broker'.+--+-- See 'attachPermanent' and 'attachTemporary'.+--+-- @since 0.30.0+data CrashReport a =+  MkCrashReport { _exonerationTimerReference :: TimerReference+                  -- ^ After a crash, an 'ExonerationTimer' according to the 'CrashRate' of the 'Watchdog'+                  -- is started, this is the reference+                , _crashTime :: UTCTime+                  -- ^ Recorded time of the crash+                , _crashReason :: Interrupt 'NoRecovery+                  -- ^ Recorded crash reason+                }+  deriving (Eq, Ord, Typeable)++instance (Show a, Typeable a) => Show (CrashReport a) where+  showsPrec d c =+    showParen (d>=10)+      ( showString "crash report: "+      . showString " time: " . showsPrec 10 (c^.crashTime)+      . showString " reason: " . showsPrec 10 (c^.crashReason)+      . showString " " . showsPrec 10 (c^.exonerationTimerReference)+      )++instance NFData (CrashReport a) where+  rnf (MkCrashReport !a !b !c) = rnf a `seq` rnf b `seq` rnf c `seq` ()++-- | Lens for '_crashTime'+--+-- @since 0.30.0+crashTime :: Lens' (CrashReport a) UTCTime+crashTime = lens _crashTime (\c t -> c { _crashTime = t})++-- | Lens for '_crashReason'+--+-- @since 0.30.0+crashReason :: Lens' (CrashReport a) (Interrupt 'NoRecovery)+crashReason = lens _crashReason (\c t -> c { _crashReason = t})++-- | Lens for '_exonerationTimerReference'+--+-- @since 0.30.0+exonerationTimerReference :: Lens' (CrashReport a) TimerReference+exonerationTimerReference = lens _exonerationTimerReference (\c t -> c { _exonerationTimerReference = t})++startExonerationTimer :: forall child a q e .+     (HasProcesses e q, Lifted IO q, Lifted IO e, Show a, NFData a, Typeable a, Typeable child)+     => a -> Interrupt 'NoRecovery -> CrashTimeSpan -> Eff e (CrashReport a)+startExonerationTimer cId r t = do+  let title = MkProcessTitle ("ExonerationTimer<" <> pack (showSTypeable @child ">") <> pack (show cId))+  me <- self+  ref <- sendAfterWithTitle title me (TimeoutMicros (t * 1_000_000)) (MkExonerationTimer cId)+  now <- lift getCurrentTime+  return (MkCrashReport ref now r)++-- | The timer started based on the 'CrashRate' '_crashTimeSpan' when a 'CrashReport' is recorded.+--+-- After this timer elapses, the 'Watchdog' server will remove the 'CrashReport' from the 'ChildWatch' of+-- that child.+--+-- @since 0.30.0+data ExonerationTimer a =  MkExonerationTimer !a !TimerReference+  deriving (Eq, Ord, Typeable)++instance NFData a => NFData (ExonerationTimer a) where+  rnf (MkExonerationTimer !x !r) = rnf r `seq` rnf x `seq` ()++instance Show a => Show (ExonerationTimer a) where+  showsPrec d (MkExonerationTimer x r) =+    showParen (d >= 10)+      ( showString "exonerate: " . showsPrec 10 x+      . showString " after: " .  showsPrec 10 r+      )++-- --------------------------- Child Watches++-- | An internal data structure that keeps the 'CrashReport's of a child of an attached 'Broker' monitored by a 'Watchdog'.+--+-- See 'attachPermanent' and 'attachTemporary', 'ExonerationTimer', 'CrashRate'.+--+-- @since 0.30.0+data ChildWatch child =+  MkChildWatch+    { _parent :: Endpoint (Broker child)+      -- ^ The attached 'Broker' that started the child+    , _crashes :: Set (CrashReport (Broker.ChildId child))+      -- ^ The crashes of the child. If the number of crashes+      -- surpasses the allowed number of crashes before the+      -- 'ExonerationTimer's clean them, the child is finally crashed.+    }+   deriving Typeable++instance NFData (ChildWatch child) where+  rnf (MkChildWatch p c) =+    rnf p `seq` rnf c `seq` ()++instance (Typeable child, Typeable (Broker.ChildId child), Show (Broker.ChildId child)) => Show (ChildWatch child) where+  showsPrec d (MkChildWatch p c) =+    showParen (d>=10) ( showString "child-watch: parent: "+                      . showsPrec 10 p+                      . showString " crashes: "+                      . foldr (.) id (showsPrec 10 <$> Set.toList c)+                      )++-- | A lens for '_parent'.+--+-- @since 0.30.0+parent :: Lens' (ChildWatch child) (Endpoint (Broker child))+parent = lens _parent (\m x -> m {_parent = x})++-- | A lens for '_crashes'+--+-- @since 0.30.0+crashes :: Lens' (ChildWatch child) (Set (CrashReport (Broker.ChildId child)))+crashes = lens _crashes (\m x -> m {_crashes = x})++-- ------------------ Model++instance Default (Stateful.Model (Watchdog child)) where+  def = WatchdogModel def Map.empty++instance ( Show (Broker.ChildId child)+         , Typeable (Broker.ChildId child)+         , Typeable child+         )+          => Show (Stateful.Model (Watchdog child))+  where+  showsPrec d  (WatchdogModel bs cs) =+    showParen (d>=10)+      (showString "watchdog model broker watches: "+      . showsPrec 10 bs+      . showString " watchdog model child watches: "+      . showsPrec 10 cs+      )++-- -------------------------- Model -> Child Watches++watched :: Lens' (Stateful.Model (Watchdog child)) (Map (Broker.ChildId child) (ChildWatch child))+watched = lens _watched (\m x -> m {_watched = x})++childWatches :: IndexedTraversal' (Broker.ChildId child) (Stateful.Model (Watchdog child)) (ChildWatch child)+childWatches = watched . itraversed++childWatchesById ::+     Eq (Broker.ChildId child)+  => Broker.ChildId child+  -> Traversal' (Stateful.Model (Watchdog child)) (ChildWatch child)+childWatchesById theCId = childWatches . ifiltered (\cId _ -> cId == theCId)++childWatchesByParenAndId ::+     Eq (Broker.ChildId child)+  => Endpoint (Broker child)+  -> Broker.ChildId child+  -> Traversal' (Stateful.Model (Watchdog child)) (ChildWatch child)+childWatchesByParenAndId theParent theCId =+  childWatches . ifiltered (\cId cw -> cw ^. parent == theParent && cId == theCId)++countRecentCrashes ::+     Eq (Broker.ChildId child)+  => Endpoint (Broker child)+  -> Broker.ChildId child+  -> Stateful.Model (Watchdog child)+  -> CrashCount+countRecentCrashes theParent theCId theModel =+ length (theModel ^.. childWatchesByParenAndId theParent theCId . crashes . folded)++-- --------------------------- Model -> Broker Watches++brokers :: Lens' (Stateful.Model (Watchdog child)) (Map (Endpoint (Broker child)) BrokerWatch)+brokers = lens _brokers (\m x -> m {_brokers = x})++-- -------------------------- Server Implementation Helpers++removeAndCleanChild ::+  forall child q e.+  ( HasProcesses e q+  , Typeable child+  , Typeable (Broker.ChildId child)+  , Ord (Broker.ChildId child)+  , Show (Broker.ChildId child)+  , Member (Stateful.ModelState (Watchdog child)) e+  , Member Logs e+  )+  => Broker.ChildId child+  -> Eff e ()+removeAndCleanChild cId = do+    oldModel <- Stateful.modifyAndGetModel (watched @child . at cId .~ Nothing)+    forMOf_ (childWatchesById cId) oldModel $ \w -> do+      logDebug ("removing client entry: " <> pack (show cId))+      forMOf_ (crashes . folded . exonerationTimerReference) w cancelTimer+      logDebug (pack (show w))++removeBroker ::+  forall child q e.+  ( HasProcesses e q+  , Typeable child+  , Tangible (Broker.ChildId child)+  , Typeable (Effectful.ServerPdu child)+  , Ord (Broker.ChildId child)+  , Show (Broker.ChildId child)+  , Member (Stateful.ModelState (Watchdog child)) e+  , Member Logs e+  )+  => Endpoint (Watchdog child)+  -> Endpoint (Broker child)+  -> Eff e ()+removeBroker me broker = do+    oldModel <- Stateful.getAndModifyModel @(Watchdog child)+                  ( (brokers . at broker .~ Nothing)+                  . (watched %~ Map.filter (\cw -> cw^.parent /= broker))+                  )+    forM_ (oldModel ^? brokers . at broker . _Just) $ \deadBroker ->  do+      logNotice ("dettaching: " <> pack (show deadBroker) <> " " <> pack (show broker))+      let forgottenChildren = oldModel ^.. watched . itraversed . filtered (\cw -> cw^.parent == broker)+      traverse_ (logNotice . ("forgetting: " <>) . pack . show) forgottenChildren+      Observer.forgetObserver @(Broker.ChildEvent child) broker me+      when (view isPermanent deadBroker) $ do+        logError ("permanent broker exited: " <> pack (show broker))+        exitBecause (ExitOtherProcessNotRunning (broker ^. fromEndpoint))

@@ -88,6 +88,9 @@   , logInfo'   , logDebug   , logDebug'+  , logCallStack+  , logMultiLine+  , logMultiLine'      -- ** Log Message Pre-Filtering #LogPredicate#     -- $LogPredicate

@@ -26,6 +26,9 @@   , logInfo'   , logDebug   , logDebug'+  , logCallStack+  , logMultiLine+  , logMultiLine'    -- ** Log Message Pre-Filtering #LogPredicate#   , includeLogMessages@@ -79,13 +82,15 @@                                                  ) import           Data.Default import           Data.Function                  ( fix )+import           Data.Hashable import           Data.Text                     as T import           GHC.Stack                      ( HasCallStack                                                 , callStack                                                 , withFrozenCallStack+                                                , prettyCallStack                                                 )--+import Data.Foldable                           ( traverse_ )+import Text.Printf                             ( printf )  -- | This effect sends 'LogMessage's and is a reader for a 'LogPredicate'. --@@ -458,6 +463,72 @@   -> Eff e () logDebug' = withFrozenCallStack (logWithSeverity' debugSeverity) +-- | Log the current 'callStack' using the given 'Severity'.+--+-- @since 0.30.0+logCallStack :: forall e . (HasCallStack, Member Logs e) => Severity -> Eff e ()+logCallStack =+  withFrozenCallStack $ \s ->+    let stackTraceLines = T.lines (pack (prettyCallStack callStack))+    in logMultiLine s stackTraceLines+++-- | Issue a log statement for each item in the list prefixed with a line number and a message hash.+--+-- When several concurrent processes issue log statements, multiline log statements are often+-- interleaved.+--+-- In order to make the logs easier to read, this function will count the items and calculate a unique+-- hash and prefix each message, so a user can grep to get all the lines of an interleaved,+-- multi-line log message.+--+-- @since 0.30.0+logMultiLine+  :: forall e+    . ( HasCallStack+    , Member Logs e+    )+    => Severity+    -> [Text]+    -> Eff e ()+logMultiLine =+  withFrozenCallStack $ \s messageLines -> do+    let msgHash = T.pack $ printf "multi-line message %06X" $ hash messageLines `mod` 0x1000000+        messageLinesWithLineNum =+          let messageLineCount = Prelude.length messageLines+              messageLineCountString = T.pack (show messageLineCount)+              messageLineCountStringLen = T.length messageLineCountString+              printLineNum i =+                let i' = T.pack (show i)+                    padding = messageLineCountStringLen - T.length i'+                in msgHash <> " line " <> T.replicate padding " " <> i' <> " of " <> messageLineCountString <> ":    "+          in Prelude.zipWith (<>) (printLineNum <$> [1 :: Int ..])  messageLines+    traverse_ (logWithSeverity s) messageLinesWithLineNum+++-- | Issue a log statement for each item in the list prefixed with a line number and a message hash.+--+-- When several concurrent processes issue log statements, multiline log statements are often+-- interleaved.+--+-- In order to make the logs easier to read, this function will count the items and calculate a unique+-- hash and prefix each message, so a user can grep to get all the lines of an interleaved,+-- multi-line log message.+--+-- This function takes a list of 'String's as opposed to 'logMultiLine'.+--+-- @since 0.30.0+logMultiLine'+  :: forall e+    . ( HasCallStack+    , Member Logs e+    )+    => Severity+    -> [String]+    -> Eff e ()+logMultiLine' s = logMultiLine s . fmap pack++ -- | Get the current 'Logs' filter/transformer function. -- -- See "Control.Eff.Log#LogPredicate"@@ -522,7 +593,7 @@  -- | Include 'LogMessage's that match a 'LogPredicate'. ----- @excludeLogMessages p@ allows log message to be logged if @p m@+-- @includeLogMessages p@ allows log message to be logged if @p m@ -- -- Although it is enough if the previous predicate holds. -- See 'excludeLogMessages' and 'modifyLogPredicate'.

@@ -11,6 +11,7 @@ import Control.Eff.LogWriter.IO import Data.Text import qualified Data.Text.IO                  as T+import qualified System.IO                     as IO  -- | Enable logging to @standard output@ using the 'consoleLogWriter', with some 'LogMessage' fields preset -- as in 'withIoLogging'.@@ -33,7 +34,9 @@   -> LogPredicate -- ^ The inital predicate for log messages, there are some pre-defined in "Control.Eff.Log.Message#PredefinedPredicates"   -> Eff (Logs : LogWriterReader (Lift IO) : e) a   -> Eff e a-withConsoleLogging = withIoLogging consoleLogWriter+withConsoleLogging a b c d = do+  lift (IO.hSetBuffering IO.stdout IO.LineBuffering)+  withIoLogging consoleLogWriter a b c d   -- | Enable logging to @standard output@ using the 'consoleLogWriter'.@@ -51,7 +54,9 @@ withConsoleLogWriter   :: (LogIo e)   => Eff e a -> Eff e a-withConsoleLogWriter = addLogWriter consoleLogWriter+withConsoleLogWriter e = do+  lift (IO.hSetBuffering IO.stdout IO.LineBuffering)+  addLogWriter consoleLogWriter e  -- | Write 'LogMessage's to standard output, formatted with 'printLogMessage'. --

@@ -43,7 +43,7 @@   -> LogPredicate -- ^ The inital predicate for log messages, there are some pre-defined in "Control.Eff.Log.Message#PredefinedPredicates"   -> Eff (Logs : LogWriterReader (Lift IO) : e) a   -> Eff e a-withFileLogging fnIn a f p e =+withFileLogging fnIn a f p e = do   liftBaseOp (withOpenedLogFile fnIn) (\lw -> withIoLogging lw a f p e)  @@ -72,7 +72,7 @@     fnCanon <- canonicalizePath fnIn     createDirectoryIfMissing True (takeDirectory fnCanon)     h <- IO.openFile fnCanon IO.AppendMode-    IO.hSetBuffering h (IO.BlockBuffering (Just 1024))+    IO.hSetBuffering h IO.LineBuffering     return h   )   (\h -> Safe.try @IO @Catch.SomeException (IO.hFlush h) >> IO.hClose h)

@@ -63,3 +63,4 @@ # # Allow a newer minor version of GHC than the snapshot specifies # compiler-check: newer-minor+ghc-options: {"$locals": -ddump-to-file -ddump-hi}

@@ -0,0 +1,367 @@+module BrokerTests+  ( test_Broker+  ) where++import Common+import Control.Eff.Concurrent.Protocol.EffectfulServer (Event(..))+import qualified Control.Eff.Concurrent.Protocol.StatefulServer as Stateful+import Control.Eff.Concurrent.Protocol.Broker as Broker+import qualified Control.Eff.Concurrent.Protocol.Observer.Queue as OQ+import Control.Lens++test_Broker :: HasCallStack => TestTree+test_Broker =+  setTravisTestOptions $+  testGroup+    "Broker"+    (let startTestBroker = startTestBrokerWith ExitWhenRequested (TimeoutMicros 500000)+         startTestBrokerWith e t = Broker.startLink (MkBrokerConfig t (Stateful.Init . TestServerArgs e))+         spawnTestChild broker i = Broker.spawnChild broker i >>= either (lift . assertFailure . show) pure+      in [ runTestCase "The broker starts and is shut down" $ do+             outerSelf <- self+             testWorker <-+               spawn "test-worker" $ do+                 broker <- startTestBroker+                 sendMessage outerSelf broker+                 () <- receiveMessage+                 sendMessage outerSelf ()+                 () <- receiveMessage+                 Broker.stopBroker broker+             unlinkProcess testWorker+             broker <- receiveMessage :: Eff Effects  (Endpoint (Broker.Broker (Stateful.Stateful TestProtocol)))+             brokerAliveAfter1 <- isBrokerAlive broker+             logInfo ("still alive 1: " <> pack (show brokerAliveAfter1))+             lift (brokerAliveAfter1 @=? True)+             sendMessage testWorker ()+             () <- receiveMessage+             brokerAliveAfter2 <- isBrokerAlive broker+             logInfo ("still alive 2: " <> pack (show brokerAliveAfter2))+             lift (brokerAliveAfter2 @=? True)+             sendMessage testWorker ()+             testWorkerMonitorRef <- monitor testWorker+             d1 <- receiveSelectedMessage (selectProcessDown testWorkerMonitorRef)+             logInfo ("got test worker down: " <> pack (show d1))+             testBrokerMonitorRef <- monitorBroker broker+             d2 <- receiveSelectedMessage (selectProcessDown testBrokerMonitorRef)+             logInfo ("got broker down: " <> pack (show d2))+             brokerAliveAfterOwnerExited <- isBrokerAlive broker+             logInfo ("still alive after owner exited: " <> pack (show brokerAliveAfterOwnerExited))+             lift (brokerAliveAfterOwnerExited @=? False)+         , testGroup+             "Diagnostics"+             [ runTestCase "When only time passes the diagnostics do not change" $ do+                 broker <- startTestBroker+                 info1 <- Broker.getDiagnosticInfo broker+                 lift (threadDelay 10000)+                 info2 <- Broker.getDiagnosticInfo broker+                 lift (assertEqual "diagnostics should not differ: " info1 info2)+             , runTestCase "When a child is started the diagnostics change" $ do+                 broker <- startTestBroker+                 info1 <- Broker.getDiagnosticInfo broker+                 logInfo ("got diagnostics: " <> info1)+                 let childId = 1+                 _child <- fromRight (error "failed to spawn child") <$> Broker.spawnChild broker childId+                 info2 <- Broker.getDiagnosticInfo broker+                 logInfo ("got diagnostics: " <> info2)+                 lift $ assertBool ("diagnostics should differ: " ++ show (info1, info2)) (info1 /= info2)+             ]+         , let childId = 1+            in testGroup+                 "Startup and shutdown"+                 [ runTestCase "When a broker is shut down, all children are shutdown" $ do+                     broker <- startTestBroker+                     child <- spawnTestChild broker childId+                     let childPid = _fromEndpoint child+                     brokerMon <- monitorBroker broker+                     childMon <- monitor childPid+                     isProcessAlive childPid >>= lift . assertBool "child process not running"+                     isBrokerAlive broker >>= lift . assertBool "broker process not running"+                     call child (TestGetStringLength "123") >>= lift . assertEqual "child not working" 3+                     stopBroker broker+                     d1@(ProcessDown mon1 er1 _) <-+                       fromMaybe (error "receive timeout 1") <$> receiveAfter (TimeoutMicros 1000000)+                     logInfo ("got process down: " <> pack (show d1))+                     d2@(ProcessDown mon2 er2 _) <-+                       fromMaybe (error "receive timeout 2") <$> receiveAfter (TimeoutMicros 1000000)+                     logInfo ("got process down: " <> pack (show d2))+                     case if mon1 == brokerMon && mon2 == childMon+                            then Right (er1, er2)+                            else if mon1 == childMon && mon2 == brokerMon+                                   then Right (er2, er1)+                                   else Left+                                          ("unexpected monitor down: first: " <> show (mon1, er1) <> ", and then: " <>+                                           show (mon2, er2) <>+                                           ", brokerMon: " <>+                                           show brokerMon <>+                                           ", childMon: " <>+                                           show childMon) of+                       Right (brokerER, childER) -> do+                         lift (assertEqual "bad broker exit reason" ExitNormally brokerER)+                         lift (assertEqual "bad child exit reason" ExitNormally childER)+                       Left x -> lift (assertFailure x)+                 , runTestCase+                     "When a broker is shut down, children that won't shutdown, are killed after some time" $ do+                     broker <- startTestBrokerWith IgnoreNormalExitRequest (TimeoutMicros 10000)+                     child <- spawnTestChild broker childId+                     let childPid = _fromEndpoint child+                     brokerMon <- monitorBroker broker+                     childMon <- monitor childPid+                     isProcessAlive childPid >>= lift . assertBool "child process not running"+                     isBrokerAlive broker >>= lift . assertBool "broker process not running"+                     call child (TestGetStringLength "123") >>= lift . assertEqual "child not working" 3+                     stopBroker broker+                     d1@(ProcessDown mon1 er1 _) <-+                       fromMaybe (error "receive timeout 1") <$> receiveAfter (TimeoutMicros 1000000)+                     logInfo ("got process down: " <> pack (show d1))+                     d2@(ProcessDown mon2 er2 _) <-+                       fromMaybe (error "receive timeout 2") <$> receiveAfter (TimeoutMicros 1000000)+                     logInfo ("got process down: " <> pack (show d2))+                     case if mon1 == brokerMon && mon2 == childMon+                            then Right er1+                            else if mon1 == childMon && mon2 == brokerMon+                                   then Right er2+                                   else Left+                                          ("unexpected monitor down: first: " <> show (mon1, er1) <> ", and then: " <>+                                           show (mon2, er2) <>+                                           ", brokerMon: " <>+                                           show brokerMon <>+                                           ", childMon: " <>+                                           show childMon) of+                       Right brokerER -> do+                         lift (assertEqual "bad broker exit reason" ExitNormally brokerER)+                       Left x -> lift (assertFailure x)+                 ]+         , let i = 123+            in testGroup+                 "Spawning and Using Children"+                 [ runTestCase+                     "When a broker is requested to start two children with the same id, an already started error is returned" $ do+                     broker <- startTestBroker+                     c <- spawnTestChild broker i+                     x <- Broker.spawnChild broker i+                     case x of+                       Left (AlreadyStarted i' c') ->+                         lift $ do+                           assertEqual "bad pid returned" c c'+                           assertEqual "bad child id returned" 123 i'+                       _ -> lift (assertFailure "AlreadyStarted expected!")+                 , runTestCase "When a child is started it can be lookup up" $ do+                     broker <- startTestBroker+                     c <- spawnTestChild broker i+                     c' <- Broker.lookupChild broker i >>= maybe (lift (assertFailure "child not found")) pure+                     lift (assertEqual "lookupChild returned wrong child" c c')+                 , runTestCase "When several children are started they can be lookup up and don't crash" $ do+                     broker <- startTestBroker+                     c <- spawnTestChild broker i+                     someOtherChild <- spawnTestChild broker (i + 1)+                     c' <- Broker.lookupChild broker i >>= maybe (lift (assertFailure "child not found")) pure+                     lift (assertEqual "lookupChild returned wrong child" c c')+                     childStillRunning <- isProcessAlive (_fromEndpoint c)+                     lift (assertBool "child not running" childStillRunning)+                     someOtherChildStillRunning <- isProcessAlive (_fromEndpoint someOtherChild)+                     lift (assertBool "someOtherChild not running" someOtherChildStillRunning)+                 , let startTestBrokerAndChild = do+                         broker <- startTestBroker+                         c <- spawnTestChild broker i+                         cm <- monitor (_fromEndpoint c)+                         return (broker, cm)+                    in testGroup+                         "Stopping children"+                         [ runTestCase "When a child is started it can be stopped" $ do+                             (broker, cm) <- startTestBrokerAndChild+                             Broker.stopChild broker i >>= lift . assertBool "child not found"+                             (ProcessDown _ r _) <- receiveSelectedMessage (selectProcessDown cm)+                             lift (assertEqual "bad exit reason" ExitNormally r)+                         , runTestCase+                             "When a child is stopped but doesn't exit voluntarily, it is kill after some time" $ do+                             broker <- startTestBrokerWith IgnoreNormalExitRequest (TimeoutMicros 5000)+                             c <- spawnTestChild broker i+                             cm <- monitor (_fromEndpoint c)+                             Broker.stopChild broker i >>= lift . assertBool "child not found"+                             (ProcessDown _ r _) <- receiveSelectedMessage (selectProcessDown cm)+                             case r of+                               ExitUnhandledError _ -> return ()+                               _ -> lift (assertFailure ("bad exit reason: " ++ show r))+                         , runTestCase "When a stopChild is called with an unallocated ID, False is returned" $ do+                             (broker, _) <- startTestBrokerAndChild+                             Broker.stopChild broker (i + 1) >>= lift . assertBool "child not found" . not+                         , runTestCase "When a child is stopped, lookup won't find it" $ do+                             (broker, _) <- startTestBrokerAndChild+                             Broker.stopChild broker i >>= lift . assertBool "child not found"+                             x <- Broker.lookupChild broker i+                             lift (assertEqual "lookup should not find a child" Nothing x)+                         , runTestCase "When a child is stopped, the child id can be reused" $ do+                             (broker, _) <- startTestBrokerAndChild+                             Broker.stopChild broker i >>= lift . assertBool "child not found"+                             Broker.spawnChild broker i >>= lift . assertBool "id could not be reused" . isRight+                         ]+                 , let startTestBrokerAndChild = do+                         broker <- startTestBroker+                         c <- spawnTestChild broker i+                         cm <- monitor (_fromEndpoint c)+                         return (broker, c, cm)+                    in testGroup+                         "Child exit handling"+                         [ runTestCase "When a child exits normally, lookupChild will not find it" $ do+                             (broker, c, cm) <- startTestBrokerAndChild+                             cast c (TestInterruptWith NormalExitRequested)+                             (ProcessDown _ r _) <- receiveSelectedMessage (selectProcessDown cm)+                             lift (assertEqual "bad exit reason" ExitNormally r)+                             x <- Broker.lookupChild broker i+                             lift (assertEqual "lookup should not find a child" Nothing x)+                         , runTestCase "When a child exits with an error, lookupChild will not find it" $ do+                             (broker, c, cm) <- startTestBrokerAndChild+                             cast c (TestInterruptWith (ErrorInterrupt "test error reason"))+                             (ProcessDown _ _ _) <- receiveSelectedMessage (selectProcessDown cm)+                             x <- Broker.lookupChild broker i+                             lift (assertEqual "lookup should not find a child" Nothing x)+                         , runTestCase "When a child is interrupted from another process and dies, lookupChild will not find it" $ do+                             (broker, c, cm) <- startTestBrokerAndChild+                             sendInterrupt (_fromEndpoint c) NormalExitRequested+                             (ProcessDown _ _ _) <- receiveSelectedMessage (selectProcessDown cm)+                             x <- Broker.lookupChild broker i+                             lift (assertEqual "lookup should not find a child" Nothing x)+                         , runTestCase "When a child is shutdown from another process and dies, lookupChild will not find it" $ do+                             (broker, c, cm) <- startTestBrokerAndChild+                             self >>= sendShutdown (_fromEndpoint c) . ExitProcessCancelled . Just+                             (ProcessDown _ _ _) <- receiveSelectedMessage (selectProcessDown cm)+                             x <- Broker.lookupChild broker i+                             lift (assertEqual "lookup should not find a child" Nothing x)+                         ]+                 , testGroup "broker events"+                     [ runTestCase "when a child starts the observer is notified" $ do+                         broker <- startTestBroker+                         OQ.observe @(Broker.ChildEvent (Stateful.Stateful TestProtocol)) (100 :: Int) broker $ do+                           c <- Broker.spawnChild @(Stateful.Stateful TestProtocol) broker i >>= either (lift . assertFailure . show) pure+                           e <- OQ.await @(Broker.ChildEvent (Stateful.Stateful TestProtocol))+                           case e of+                            Broker.OnChildSpawned broker' i' c' -> do+                              lift (assertEqual "wrong endpoint" broker broker')+                              lift (assertEqual "wrong child" c c')+                              lift (assertEqual "wrong child-id" i i')+                            _ ->+                              lift (assertFailure ("unexpected event: " ++ show e))+                    , runTestCase "when a child stops the observer is notified" $ do+                         broker <- startTestBroker+                         c <- Broker.spawnChild @(Stateful.Stateful TestProtocol) broker i >>= either (lift . assertFailure . show) pure+                         OQ.observe @(Broker.ChildEvent (Stateful.Stateful TestProtocol)) (100 :: Int) broker $ do+                           Broker.stopChild broker i >>= lift . assertBool "child not found"+                           OQ.await @(Broker.ChildEvent (Stateful.Stateful TestProtocol))+                            >>= \case+                                    Broker.OnChildDown broker' i' c' e -> do+                                      lift (assertEqual "wrong endpoint" broker broker')+                                      lift (assertEqual "wrong child" c c')+                                      lift (assertEqual "wrong child-id" i i')+                                      lift (assertEqual "wrong exit reason" ExitNormally e)+                                    e ->+                                      lift (assertFailure ("unexpected event: " ++ show e))++                    , runTestCase "when a child crashes the observer is notified" $ do+                         broker <- startTestBroker+                         OQ.observe @(Broker.ChildEvent (Stateful.Stateful TestProtocol)) (100 :: Int) broker $ do+                           c <- Broker.spawnChild @(Stateful.Stateful TestProtocol) broker i >>= either (lift . assertFailure . show) pure+                           let expectedError = ExitUnhandledError "test error"+                           sendShutdown (_fromEndpoint c) expectedError+                           OQ.await @(Broker.ChildEvent (Stateful.Stateful TestProtocol))+                            >>= \case+                                    Broker.OnChildSpawned broker' i' c' -> do+                                      lift (assertEqual "wrong endpoint" broker broker')+                                      lift (assertEqual "wrong child" c c')+                                      lift (assertEqual "wrong child-id" i i')+                                    e ->+                                      lift (assertFailure ("unexpected event: " ++ show e))+                           OQ.await @(Broker.ChildEvent (Stateful.Stateful TestProtocol))+                            >>= \case+                                    Broker.OnChildDown broker' i' c' e -> do+                                      lift (assertEqual "wrong endpoint" broker broker')+                                      lift (assertEqual "wrong child" c c')+                                      lift (assertEqual "wrong child-id" i i')+                                      lift (assertEqual "wrong exit reason" expectedError e)+                                    e ->+                                      lift (assertFailure ("unexpected event: " ++ show e))+                    , runTestCase "when a child does not stop when requested and is killed, the oberser is notified correspondingly" $ do+                         broker <- startTestBrokerWith IgnoreNormalExitRequest (TimeoutMicros 5000)+                         c <- spawnTestChild broker i+                         OQ.observe @(Broker.ChildEvent (Stateful.Stateful TestProtocol)) (100 :: Int) broker $ do+                           cm <- monitor (_fromEndpoint c)+                           Broker.stopChild broker i >>= lift . assertBool "child not found"+                           (ProcessDown _ expectedError _) <- receiveSelectedMessage (selectProcessDown cm)+                           OQ.await @(Broker.ChildEvent (Stateful.Stateful TestProtocol))+                            >>= \case+                                    Broker.OnChildDown broker' i' c' e -> do+                                      lift (assertEqual "wrong endpoint" broker broker')+                                      lift (assertEqual "wrong child" c c')+                                      lift (assertEqual "wrong child-id" i i')+                                      lift (assertEqual "wrong exit reason" expectedError e)+                                    e ->+                                      lift (assertFailure ("unexpected event: " ++ show e))++                    , runTestCase "when the broker stops, an OnBrokerShuttingDown is emitted before any child is stopped" $ do+                         broker <- startTestBroker+                         unlinkProcess (broker ^. fromEndpoint)+                         void (Broker.spawnChild @(Stateful.Stateful TestProtocol) broker i+                                >>= either (lift . assertFailure . show) pure)+                         OQ.observe @(Broker.ChildEvent (Stateful.Stateful TestProtocol)) (1000 :: Int) broker $ do+                           Broker.stopBroker broker+                           OQ.await @(Broker.ChildEvent (Stateful.Stateful TestProtocol))+                            >>= \case+                                    Broker.OnBrokerShuttingDown _ -> logNotice "received OnBrokerShuttingDown"+                                    e ->+                                      lift (assertFailure ("unexpected event: " ++ show e))+                           OQ.await @(Broker.ChildEvent (Stateful.Stateful TestProtocol))+                            >>= \case+                                  Broker.OnChildDown _ _ _ e -> lift (assertEqual "wrong exit reason" ExitNormally e)+                                  e ->+                                    lift (assertFailure ("unexpected event: " ++ show e))+                    ]+                 ]+         ])++data TestProtocol+  deriving (Typeable)++type instance ToPretty TestProtocol = PutStr "test"++instance HasPdu TestProtocol where+  data instance Pdu TestProtocol x where+    TestGetStringLength :: String -> Pdu TestProtocol ('Synchronous Int)+    TestInterruptWith :: Interrupt 'Recoverable -> Pdu TestProtocol 'Asynchronous+      deriving Typeable++instance NFData (Pdu TestProtocol x) where+  rnf (TestGetStringLength x) = rnf x+  rnf (TestInterruptWith x) = rnf x++instance Show (Pdu TestProtocol r) where+  show (TestGetStringLength s) = "TestGetStringLength " ++ show s+  show (TestInterruptWith s) = "TestInterruptWith " ++ show s++data TestProtocolServerMode+  = IgnoreNormalExitRequest+  | ExitWhenRequested+  deriving Eq++instance Stateful.Server TestProtocol Effects where+  newtype instance Model TestProtocol = TestProtocolModel () deriving Default+  update _me (TestServerArgs testMode tId) evt =+    case evt of+      OnCast (TestInterruptWith i) -> do+        logInfo (pack (show tId) <> ": stopping with: " <> pack (show i))+        interrupt i+      OnCall rt (TestGetStringLength str) -> do+        logInfo (pack (show tId) <> ": calculating length of: " <> pack str)+        sendReply rt (length str)+      OnInterrupt x -> do+        logNotice (pack (show tId) <> ": " <> pack (show x))+        if testMode == IgnoreNormalExitRequest+          then+            logNotice $ pack (show tId) <> ": ignoring normal exit request"+          else do+            logNotice $ pack (show tId) <> ": exitting normally"+            exitBecause (interruptToExit x)+      _ ->+        logDebug (pack (show tId) <> ": got some info: " <> pack (show evt))+  data instance StartArgument TestProtocol Effects = TestServerArgs TestProtocolServerMode Int++type instance ChildId TestProtocol = Int+

@@ -1,30 +1,95 @@-module Common where+module Common+  ( module Common+  , module Test.Tasty+  , module Test.Tasty.HUnit+  , module Test.Tasty.Runners+  , module Control.Eff.Extend+  , module Control.Monad+  , module GHC.Stack+  , module Control.Concurrent+  , module Control.Concurrent.STM+  , module Control.DeepSeq+  , module Control.Eff+  , module Control.Eff.Concurrent+  , module Control.Eff.Concurrent.Misc+  , module Data.Default+  , module Data.Foldable+  , module Data.Typeable+  , module Data.Text+  , module Data.Either+  , module Data.Maybe+  , module Data.Type.Pretty+  )+where -import Control.Concurrent-import Control.Concurrent.STM-import Control.Eff-import Control.Eff.Concurrent hiding (Timeout)-import Control.Eff.Concurrent.Process.ForkIOScheduler as Scheduler-import Control.Eff.Extend-import Control.Monad (void)-import GHC.Stack-import Test.Tasty-import Test.Tasty.HUnit-import Test.Tasty.Runners+import           Control.Concurrent+import           Control.Concurrent.STM+import           Control.DeepSeq+import           Control.Eff+--import           Control.Eff.Log+import           Control.Eff.Concurrent+import           Control.Eff.Concurrent.Misc+import           Control.Eff.Concurrent.Process.ForkIOScheduler+                                               as Scheduler+import           Control.Eff.Extend+import           Control.Monad+import           Control.Lens+import           Data.Default+import           Data.Foldable+import           Data.Text                      ( Text+                                                , pack+                                                )+import qualified Data.Text                     as T+import           Data.Typeable           hiding ( cast )+import           GHC.Stack+import           Test.Tasty              hiding ( Timeout+                                                , defaultMain+                                                )+import qualified Test.Tasty                    as Tasty+import           Test.Tasty.HUnit+import           Test.Tasty.Runners+import           Data.Either                    ( fromRight+                                                , isLeft+                                                , isRight+                                                )+import           Data.Maybe                     ( fromMaybe )+import           Data.Type.Pretty+import qualified System.IO                     as IO  setTravisTestOptions :: TestTree -> TestTree-setTravisTestOptions = localOption (timeoutSeconds 60) . localOption (NumThreads 1)+setTravisTestOptions =+  localOption (timeoutSeconds 60) . localOption (NumThreads 1) -timeoutSeconds :: Integer -> Timeout-timeoutSeconds seconds = Timeout (seconds * 1000000) (show seconds ++ "s")+timeoutSeconds :: Integer -> Tasty.Timeout+timeoutSeconds seconds =+  Tasty.Timeout (seconds * 1000000) (show seconds ++ "s")  runTestCase :: TestName -> Eff Effects () -> TestTree-runTestCase msg =-  testCase msg .-  runLift . withTraceLogging "unit-tests" local0 allLogMessages . Scheduler.schedule . handleInterrupts onInt-  where-    onInt = lift . assertFailure . show+runTestCase msg et =+  testCase msg $ do+    IO.hSetBuffering IO.stdout IO.LineBuffering+    runLift+      $ withIoLogging (stdoutLogWriter renderMinimalisticWide) "unit-tests" local0 allLogMessages+      $ Scheduler.schedule+      $ handleInterrupts onInt+        et+  where onInt = lift . assertFailure . show +-- | Render a 'LogMessage' human readable, for console logging+renderMinimalisticWide :: LogMessageRenderer T.Text+renderMinimalisticWide l =+  T.unwords $ filter+    (not . T.null)+    [ let s = l ^. lmSeverity . to (T.pack . show)+      in s <> T.replicate (max 0 (15 - T.length s)) " "+    , let p = fromMaybe " no proc " (l ^. lmProcessId)+      in p <> T.replicate (max 0 (55 - T.length p)) " "+    , let msg = l^.lmMessage+      in  msg <> T.replicate (max 0 (100 - T.length msg)) " "+    -- , fromMaybe "" (renderLogMessageSrcLoc l)+    ]++ withTestLogC :: (e -> IO ()) -> (IO (e -> IO ()) -> TestTree) -> TestTree withTestLogC doSchedule k = k (return doSchedule) @@ -33,25 +98,30 @@   r <- send pa   case r of     Interrupted _ -> return ()-    _ -> untilInterrupted pa+    _             -> untilInterrupted pa -scheduleAndAssert ::-     forall r. (LogIo r)+scheduleAndAssert+  :: forall r+   . (LogIo r)   => IO (Eff (Processes r) () -> IO ())   -> ((String -> Bool -> Eff (Processes r) ()) -> Eff (Processes r) ())   -> IO ()-scheduleAndAssert schedulerFactory testCaseAction =-  withFrozenCallStack $ do-    resultVar <- newEmptyTMVarIO-    void-      (applySchedulerFactory-         schedulerFactory-         (testCaseAction (\title cond -> lift (atomically (putTMVar resultVar (title, cond))))))-    (title, result) <- atomically (takeTMVar resultVar)-    assertBool title result+scheduleAndAssert schedulerFactory testCaseAction = withFrozenCallStack $ do+  IO.hSetBuffering IO.stdout IO.LineBuffering+  resultVar <- newEmptyTMVarIO+  void+    (applySchedulerFactory+      schedulerFactory+      (testCaseAction+        (\title cond -> lift (atomically (putTMVar resultVar (title, cond))))+      )+    )+  (title, result) <- atomically (takeTMVar resultVar)+  assertBool title result -applySchedulerFactory ::-     forall r. (LogIo r)+applySchedulerFactory+  :: forall r+   . (LogIo r)   => IO (Eff (Processes r) () -> IO ())   -> Eff (Processes r) ()   -> IO ()@@ -59,8 +129,46 @@   scheduler <- factory   scheduler (procAction >> lift (threadDelay 20000)) -awaitProcessDown :: (HasCallStack, HasProcesses r q) => ProcessId -> Eff r ProcessDown+assertShutdown+  :: (Member Logs r, HasCallStack, HasProcesses r q, Lifted IO r)+  => ProcessId+  -> Interrupt 'NoRecovery+  -> Eff r ()+assertShutdown p r = do+  unlinkProcess p+  m <- monitor p+  sendShutdown p r+  logInfo+    (  "awaitProcessDown: "+    <> pack (show p)+    <> " "+    <> pack (show m)+    )+  logCallStack debugSeverity+  receiveSelectedMessage (selectProcessDown m)+    >>= lift . assertEqual "bad exit reason" r . downReason++awaitProcessDown+  :: (Member Logs r, HasCallStack, HasProcesses r q)+  => ProcessId+  -> Eff r ProcessDown awaitProcessDown p = do   m <- monitor p+  logInfo+    (  "awaitProcessDown: "+    <> pack (show p)+    <> " "+    <> pack (show m)+    )+  logCallStack debugSeverity   receiveSelectedMessage (selectProcessDown m)++awaitProcessDownAny+  :: (Member Logs r, HasCallStack, HasProcesses r q)+  => Eff r ProcessDown+awaitProcessDownAny = do+  logInfo "awaitProcessDownAny"+  logCallStack debugSeverity+  receiveMessage+ 

@@ -4,17 +4,10 @@   ) where  import Common-import Control.DeepSeq-import Control.Eff-import Control.Eff.Concurrent-import Control.Eff.Concurrent.Protocol.Supervisor as Sup+import Control.Eff.Concurrent.Protocol.Broker as Broker import qualified Control.Eff.Concurrent.Protocol.EffectfulServer as E import qualified Control.Eff.Concurrent.Protocol.StatefulServer as S-import Data.Text as T-import Data.Type.Pretty import Data.Typeable (Typeable)-import Test.Tasty-import Test.Tasty.HUnit  -- ------------------------------ @@ -41,11 +34,12 @@  instance LogIo e => S.Server Small (Processes e) where   data StartArgument Small (Processes e) = MkSmall-  type Model Small = String+  newtype instance Model Small = SmallModel String deriving Default   update _me MkSmall x =     case x of       E.OnCall rt (SmallCall f) ->-       sendReply rt f+       do S.modifyModel (\(SmallModel y) -> SmallModel (y ++ ", " ++ show f))+          sendReply rt f       E.OnCast msg ->        logInfo' (show msg)       other ->@@ -84,26 +78,32 @@  instance LogIo e => S.Server Big (Processes e) where   data instance StartArgument Big (Processes e) = MkBig-  type Model Big = String+  newtype Model Big = BigModel String deriving Default   update me MkBig = \case     E.OnCall rt req ->           case req of             BigCall o -> do               logNotice ("BigCall " <> pack (show o))               sendReply rt o-            BigSmall x -> S.update (toEmbeddedEndpoint me) MkSmall (S.OnCall (toEmbeddedReplyTarget rt) x)+            BigSmall x ->+               S.coerceEffects+                  (S.update+                    (toEmbeddedEndpoint me)+                    MkSmall+                    (S.OnCall (toEmbeddedReplyTarget rt) x))     E.OnCast req ->         case req of-          BigCast o -> S.putModel @Big o-          BigSmall x -> S.update (toEmbeddedEndpoint me) MkSmall (S.OnCast x)+          BigCast o -> S.putModel (BigModel o)+          BigSmall x -> S.coerceEffects (S.update (toEmbeddedEndpoint me) MkSmall (S.OnCast x))     other ->       interrupt (ErrorInterrupt (show other))+ -- ----------------------------------------------------------------------------  test_genServer :: HasCallStack => TestTree test_genServer = setTravisTestOptions $ testGroup "Server" [   runTestCase "When a server is started it handles call Pdus without dieing" $ do-    big <- S.start MkBig+    big <- S.startLink MkBig     call big (BigCall True) >>=  lift . assertBool "invalid result 1"     isProcessAlive (_fromEndpoint big) >>= lift . assertBool "process dead"     call big (BigCall False) >>=  lift . assertBool "invalid result 2" . not

@@ -4,13 +4,7 @@     ) where -import           Control.DeepSeq-import           Control.Eff-import           Control.Eff.Concurrent import           Control.Eff.State.Strict-import           Control.Monad-import           Test.Tasty-import           Test.Tasty.HUnit import           Common import           Control.Eff.Concurrent.Process.SingleThreadedScheduler                                                as Scheduler

@@ -4,18 +4,11 @@   ) where  import Common-import Control.DeepSeq-import Control.Eff-import Control.Eff.Concurrent import qualified Control.Eff.Concurrent.Protocol.EffectfulServer as S import qualified Control.Eff.Concurrent.Protocol.Observer.Queue as OQ import qualified Control.Eff.Concurrent.Protocol.StatefulServer as M import Control.Lens-import Control.Monad-import Data.Text as T import Data.Typeable (Typeable)-import Test.Tasty-import Test.Tasty.HUnit   test_observer :: HasCallStack => TestTree@@ -26,7 +19,7 @@ basicTests =   [runTestCase "when no observer is present, nothing crashes"       $ do-            testObservable <- S.start TestObservableServerInit+            testObservable <- S.startLink TestObservableServerInit             call testObservable (SendTestEvent "1")             cast testObservable StopTestObservable @@ -34,11 +27,11 @@    , runTestCase "observers receive only messages sent after registration"       $ do-            testObservable <- S.start TestObservableServerInit+            testObservable <- S.startLink TestObservableServerInit             call testObservable (SendTestEvent "1")             call testObservable (SendTestEvent "2")             call testObservable (SendTestEvent "3")-            testObserver <- M.start MkTestObserver+            testObserver <- M.startLink MkTestObserver             registerObserver @String testObservable testObserver             call testObservable (SendTestEvent "4")             call testObservable (SendTestEvent "5")@@ -48,11 +41,11 @@             void $ awaitProcessDown (testObservable ^. fromEndpoint)   , runTestCase "observers receive only messages sent before de-registration"       $ do-            testObservable <- S.start TestObservableServerInit+            testObservable <- S.startLink TestObservableServerInit             call testObservable (SendTestEvent "1")             call testObservable (SendTestEvent "2")             call testObservable (SendTestEvent "3")-            testObserver <- M.start MkTestObserver+            testObserver <- M.startLink MkTestObserver             registerObserver @String testObservable testObserver             call testObservable (SendTestEvent "4")             call testObservable (SendTestEvent "5")@@ -65,11 +58,11 @@             void $ awaitProcessDown (testObservable ^. fromEndpoint)   , runTestCase "observers receive only messages sent between registration and deregistration"       $ do-            testObservable <- S.start TestObservableServerInit+            testObservable <- S.startLink TestObservableServerInit             call testObservable (SendTestEvent "1")             call testObservable (SendTestEvent "2")             call testObservable (SendTestEvent "3")-            testObserver <- M.start MkTestObserver+            testObserver <- M.startLink MkTestObserver             registerObserver @String testObservable testObserver             call testObservable (SendTestEvent "4")             call testObservable (SendTestEvent "5")@@ -83,13 +76,13 @@             void $ awaitProcessDown (testObservable ^. fromEndpoint)   , runTestCase "all observers receive all messages sent between registration and deregistration"       $ do-            testObservable <- S.start TestObservableServerInit+            testObservable <- S.startLink TestObservableServerInit             call testObservable (SendTestEvent "1")             call testObservable (SendTestEvent "2")             call testObservable (SendTestEvent "3")-            testObserver1 <- M.start MkTestObserver-            testObserver2 <- M.start MkTestObserver-            testObserver3 <- M.start MkTestObserver+            testObserver1 <- M.startLink MkTestObserver+            testObserver2 <- M.startLink MkTestObserver+            testObserver3 <- M.startLink MkTestObserver             registerObserver @String testObservable testObserver1             call testObservable (SendTestEvent "4")             registerObserver @String testObservable testObserver2@@ -112,12 +105,12 @@             void $ awaitProcessDown (testObservable ^. fromEndpoint)   , runTestCase "when an observer exits, the messages are still deliviered to the others"       $ do-            testObservable <- S.start TestObservableServerInit+            testObservable <- S.startLink TestObservableServerInit             call testObservable (SendTestEvent "1")             call testObservable (SendTestEvent "2")             call testObservable (SendTestEvent "3")-            testObserver1 <- M.start MkTestObserver-            testObserver2 <- M.start MkTestObserver+            testObserver1 <- M.startLink MkTestObserver+            testObserver2 <- M.startLink MkTestObserver             registerObserver @String testObservable testObserver1             registerObserver @String testObservable testObserver2             call testObservable (SendTestEvent "4")@@ -129,13 +122,30 @@             lift (["4", "5", "6"] @=? es1)             cast testObservable StopTestObservable             void $ awaitProcessDown (testObservable ^. fromEndpoint)+  , runTestCase "evil observer monitoring"+      $ do+            testObservable <- S.startLink TestObservableServerInit+            call testObservable (SendTestEvent "1")+            call testObservable (SendTestEvent "2")+            call testObservable (SendTestEvent "3")+            testObserver1 <- M.startLink MkTestObserver+            testObserver2 <- M.startLink MkTestObserver+            registerObserver @String testObservable testObserver1+            sendShutdown (testObserver2^.fromEndpoint) ExitNormally+            void $ monitor (testObserver2^.fromEndpoint)+            void $ awaitProcessDownAny+            call testObservable (SendTestEvent "6")+            es1 <- call testObserver1 GetCapturedEvents+            lift (["6"] @=? es1)+            cast testObservable StopTestObservable+            void $ awaitProcessDown (testObservable ^. fromEndpoint)   , runTestCase "when an observer registers multiple times, it still gets the messages only once"       $ do-            testObservable <- S.start TestObservableServerInit+            testObservable <- S.startLink TestObservableServerInit             call testObservable (SendTestEvent "1")             call testObservable (SendTestEvent "2")             call testObservable (SendTestEvent "3")-            testObserver1 <- M.start MkTestObserver+            testObserver1 <- M.startLink MkTestObserver             registerObserver @String testObservable testObserver1             registerObserver @String testObservable testObserver1             call testObservable (SendTestEvent "4")@@ -147,11 +157,11 @@             void $ awaitProcessDown (testObservable ^. fromEndpoint)   , runTestCase "when an observer is forgotton multiple times, nothing bad happens"       $ do-            testObservable <- S.start TestObservableServerInit+            testObservable <- S.startLink TestObservableServerInit             call testObservable (SendTestEvent "1")             call testObservable (SendTestEvent "2")             call testObservable (SendTestEvent "3")-            testObserver1 <- M.start MkTestObserver+            testObserver1 <- M.startLink MkTestObserver             registerObserver @String testObservable testObserver1             call testObservable (SendTestEvent "4")             call testObservable (SendTestEvent "5")@@ -170,7 +180,7 @@   testGroup "observer-queue"   [ runTestCase "tryRead"       $ do-          testObservable <- S.start TestObservableServerInit+          testObservable <- S.startLink TestObservableServerInit           let len :: Int               len = 1           OQ.observe  @String len testObservable $ do@@ -182,7 +192,7 @@           void $ awaitProcessDown (testObservable ^. fromEndpoint)   , runTestCase "observe then read"       $ do-          testObservable <- S.start TestObservableServerInit+          testObservable <- S.startLink TestObservableServerInit           let len :: Int               len = 1           OQ.observe  @String len testObservable $ do@@ -197,7 +207,7 @@           void $ awaitProcessDown (testObservable ^. fromEndpoint)   , runTestCase "FIFO"       $ do-          testObservable <- S.start TestObservableServerInit+          testObservable <- S.startLink TestObservableServerInit           let len :: Int               len = 3           OQ.observe  @String len testObservable $ do@@ -212,7 +222,7 @@           void $ awaitProcessDown (testObservable ^. fromEndpoint)   , runTestCase "flush"       $ do-          testObservable <- S.start TestObservableServerInit+          testObservable <- S.startLink TestObservableServerInit           let len :: Int               len = 3           OQ.observe  @String len testObservable $ do@@ -226,7 +236,7 @@           void $ awaitProcessDown (testObservable ^. fromEndpoint)   , runTestCase "when the queue is full, new observations are dropped"       $ do-            testObservable <- S.start TestObservableServerInit+            testObservable <- S.startLink TestObservableServerInit             let len :: Int                 len = 2             OQ.observe  @String len testObservable $ do@@ -242,7 +252,7 @@             void $ awaitProcessDown (testObservable ^. fromEndpoint)   , runTestCase "flush after queue full"       $ do-          testObservable <- S.start TestObservableServerInit+          testObservable <- S.startLink TestObservableServerInit           let len :: Int               len = 3           OQ.observe  @String len testObservable $ do@@ -333,13 +343,12 @@  instance (LogIo r, HasProcesses r q) => M.Server TestObserver r where   data StartArgument TestObserver r = MkTestObserver-  type Model TestObserver = [String]-  setup _ MkTestObserver = pure ([], ())+  newtype instance Model TestObserver = TestObserverModel {fromTestObserverModel :: [String]} deriving Default   update _ MkTestObserver e =     case e of       M.OnCall rt GetCapturedEvents ->-        M.getAndPutModel @TestObserver [] >>= sendReply rt+        M.getAndPutModel (TestObserverModel []) >>= sendReply rt . fromTestObserverModel       M.OnCast (OnTestEvent (Observed x)) ->-        M.modifyModel @TestObserver (++ [x])+        M.modifyModel (\ (TestObserverModel o) -> TestObserverModel (o ++ [x]))       _ ->         logError ("unexpected: " <> pack (show e))

@@ -2,39 +2,21 @@  import           Common import           Control.Exception-import           Control.Concurrent-import           Control.Concurrent.STM-import           Control.Eff.Concurrent.Process-import           Control.Eff.Concurrent.Process.Timer-import           Control.Eff.Concurrent.Protocol-import           Control.Eff.Concurrent.Protocol.Wrapper-import           Control.Eff.Concurrent.Protocol.Client-import qualified Control.Eff.Concurrent.Protocol.CallbackServer as Callback+import qualified Control.Eff.Concurrent.Protocol.CallbackServer+                                               as Callback import           Control.Eff.Concurrent.Protocol.EffectfulServer import qualified Control.Eff.Concurrent.Process.ForkIOScheduler                                                as ForkIO import qualified Control.Eff.Concurrent.Process.SingleThreadedScheduler                                                as SingleThreaded-import           Control.Eff-import           Control.Eff.Extend-import           Control.Eff.Log-import           Control.Eff.LogWriter.Async-import           Control.Eff.LogWriter.Console-import           Control.Eff.LogWriter.IO-import           Control.Eff.Loop-import           Control.Monad import           Control.Applicative-import           Control.Lens (view)-import           Control.DeepSeq+import           Control.Lens                   ( view ) import           Data.List                      ( sort ) import           Data.Foldable                  ( traverse_ ) import           Data.Maybe-import           Data.Typeable-import           Data.Type.Pretty import           Data.Void-import           Test.Tasty-import           Test.Tasty.HUnit-import GHC.Generics (Generic)+import           GHC.Generics                   ( Generic )+import           Data.String                    ( fromString )   testInterruptReason :: Interrupt 'Recoverable@@ -43,23 +25,25 @@ test_forkIo :: TestTree test_forkIo = setTravisTestOptions $ withTestLogC   (\c ->-      runLift-    $ withLogging (filteringLogWriter (lmSeverityIsAtLeast errorSeverity) consoleLogWriter)-    $ withAsyncLogWriter (100 :: Int)-    $ ForkIO.schedule c)+    runLift+      $ withLogging+          (filteringLogWriter (lmSeverityIsAtLeast errorSeverity)+                              consoleLogWriter+          )+      $ withAsyncLogWriter (100 :: Int)+      $ ForkIO.schedule c+  )   (\factory -> testGroup "ForkIOScheduler" [allTests factory])   test_singleThreaded :: TestTree test_singleThreaded = setTravisTestOptions $ withTestLogC   (\e ->-    let runEff-          :: Eff LoggingAndIo a-          -> IO a-        runEff =-            runLift-          . withLogging-              (mkLogWriterIO (\m -> when (view lmSeverity m < errorSeverity) (printLogMessage m)))+    let runEff :: Eff LoggingAndIo a -> IO a+        runEff = runLift . withLogging+          (mkLogWriterIO+            (\m -> when (view lmSeverity m < errorSeverity) (printLogMessage m))+          )     in  void $ SingleThreaded.scheduleM runEff yield e   )   (\factory -> testGroup "SingleThreadedScheduler" [allTests factory])@@ -79,6 +63,7 @@     , exitTests schedulerFactory     , pingPongTests schedulerFactory     , yieldLoopTests schedulerFactory+    , delayTests schedulerFactory     , selectiveReceiveTests schedulerFactory     , linkingTests schedulerFactory     , monitoringTests schedulerFactory@@ -128,23 +113,18 @@   :: forall q    . (HasCallStack, LogIo q, Typeable q)   => Eff (Processes q) (Endpoint ReturnToSender)-returnToSenderServer =-  Callback.start @ReturnToSender-    $ Callback.onEvent-        (\evt ->-          case evt of-            OnCall rt msg ->-              case msg of-                StopReturnToSender -> interrupt testInterruptReason-                ReturnToSender fromP echoMsg -> do-                  sendMessage fromP echoMsg-                  yieldProcess-                  sendReply rt True-            OnInterrupt i ->-              interrupt i-            other -> interrupt (ErrorInterrupt (show other))-        )-        "return-to-sender"+returnToSenderServer = Callback.startLink @ReturnToSender $ Callback.onEvent+  (\evt -> case evt of+    OnCall rt msg -> case msg of+      StopReturnToSender           -> interrupt testInterruptReason+      ReturnToSender fromP echoMsg -> do+        sendMessage fromP echoMsg+        yieldProcess+        sendReply rt True+    OnInterrupt i -> interrupt i+    other         -> interrupt (ErrorInterrupt (show other))+  )+  "return-to-sender"  selectiveReceiveTests   :: forall r@@ -170,7 +150,7 @@           senderLoop destination =             traverse_ (sendMessage destination) [1 .. nMax] -        me          <- self+        me           <- self         receiverPid2 <- spawn "reciever loop" (receiverLoop me)         spawn_ "sender loop" (senderLoop receiverPid2)         ok <- receiveMessage @Bool@@ -182,16 +162,44 @@         ok  <- returnToSender srv "test"         ()  <- stopReturnToSender srv         lift (ok @? "selective receive failed")+    , testCase "when sending multiple messages, it is possible to receive them selectively in any order"+    $ applySchedulerFactory schedulerFactory+    $ do+        me <- self+        let messages :: [Int]+            messages = [1 .. 100]+        mRefs <- traverse+                  (\i ->+                    spawn (fromString ("sender-" ++ show i))+                          (yieldProcess >> sendMessage me i >> logInfo ("sent: " <> pack (show i)))+                    >>= monitor)+                  messages+        traverse_+                  (\i ->+                    receiveSelectedMessage (filterMessage (== i))+                    >>= logInfo . ("received: " <> ) . pack . show)+                  messages+        traverse_+                  (\(mref, i) ->+                    logInfo ("waiting for " <> pack (show mref) <> " of " <> pack (show i))+                      >> receiveSelectedMessage (selectProcessDown mref)+                      >>= logInfo . (("down: " <> pack (show i) <> " ") <> ) . pack . show)+                  (mRefs `zip` messages)+     , testCase "flush messages" $ applySchedulerFactory schedulerFactory $ do       me <- self-      spawn_ "sender-bool" $ replicateM_ 10 (sendMessage me True) >> sendMessage me ()+      spawn_ "sender-bool"+        $  replicateM_ 10 (sendMessage me True)+        >> sendMessage me ()       spawn_ "sender-float"         $  replicateM_ 10 (sendMessage me (123.23411 :: Float))         >> sendMessage me ()-      spawn_ "sender-string" $ replicateM_ 10 (sendMessage me ("123"::String)) >> sendMessage me ()-      ()   <- receiveMessage-      ()   <- receiveMessage-      ()   <- receiveMessage+      spawn_ "sender-string"+        $  replicateM_ 10 (sendMessage me ("123" :: String))+        >> sendMessage me ()+      ()       <- receiveMessage+      ()       <- receiveMessage+      ()       <- receiveMessage       -- replicateCheapM_ 40 yieldProcess       messages <- flushMessages       lift (length messages @?= 30)@@ -199,11 +207,62 @@   )  +delayTests+  :: forall r . (LogIo r) => IO (Eff (Processes r) () -> IO ()) -> TestTree+delayTests schedulerFactory =+  let maxN = 100000+  in  setTravisTestOptions+        (testGroup+          "delay tests"+          [ testCase+            "delay many times (forM_)"+            (applySchedulerFactory+              schedulerFactory+              (forM_ [1 :: Int .. maxN] (\_ -> delay (TimeoutMicros 1)))+            )+          , testCase+            "process can be interrupted during a delay"+            (applySchedulerFactory+              schedulerFactory+              (do+                me <- self+                sleeper <- spawn "sleeper"+                              (tryUninterrupted (delay (TimeoutMicros 10_000_000)) >>= sendMessage me)+                delay (TimeoutMicros 1_000)+                let expected = TimeoutInterrupt "test timeout interrupt"+                sendInterrupt sleeper expected+                receiveMessage @(Either (Interrupt 'Recoverable) ()) >>= lift . assertEqual "wrong message" (Left expected)+              )+            )+          , testCase+            "messages are enqueued for a process that is currently delaying"+            (applySchedulerFactory+              schedulerFactory+              (do+                me <- self+                sleeper <- spawn "sleeper"+                              (do sendMessage me True+                                  delay (TimeoutMicros 500_000)+                                  receiveMessage @Int >>= sendMessage me+                                  receiveMessage @Int >>= sendMessage me+                                  receiveMessage @Int >>= sendMessage me+                              )+                let x1, x2, x3 :: Int+                    [x1,x2,x3] = [1..3]+                receiveMessage >>= lift . assertBool "sleeper not started"+                sendMessage sleeper x1+                sendMessage sleeper x2+                sendMessage sleeper x3+                receiveMessage @Int >>= lift . assertEqual "wrong message 1" x1+                receiveMessage @Int >>= lift . assertEqual "wrong message 2" x2+                receiveMessage @Int >>= lift . assertEqual "wrong message 3" x3+              )+            )+          ]+        )+ yieldLoopTests-  :: forall r-   . (LogIo r)-  => IO (Eff (Processes r) () -> IO ())-  -> TestTree+  :: forall r . (LogIo r) => IO (Eff (Processes r) () -> IO ()) -> TestTree yieldLoopTests schedulerFactory =   let maxN = 100000   in  setTravisTestOptions@@ -242,10 +301,7 @@ instance NFData Pong  pingPongTests-  :: forall r-   . (LogIo r)-  => IO (Eff (Processes r) () -> IO ())-  -> TestTree+  :: forall r . (LogIo r) => IO (Eff (Processes r) () -> IO ()) -> TestTree pingPongTests schedulerFactory = testGroup   "Yield Tests"   [ testCase "ping pong a message between two processes, both don't yield"@@ -317,10 +373,7 @@   ]  errorTests-  :: forall r-   . (LogIo r)-  => IO (Eff (Processes r) () -> IO ())-  -> TestTree+  :: forall r . (LogIo r) => IO (Eff (Processes r) () -> IO ()) -> TestTree errorTests schedulerFactory = testGroup   "causing and handling errors"   [ testGroup@@ -343,7 +396,8 @@           traverse_             (\(i :: Int) -> spawn "test" $ foreverCheap               (void-                (  sendMessage (888888 + fromIntegral i) ("test message" :: String)+                (  sendMessage (888888 + fromIntegral i)+                               ("test message" :: String)                 >> yieldProcess                 )               )@@ -362,10 +416,7 @@   ]  concurrencyTests-  :: forall r-   . (LogIo r)-  => IO (Eff (Processes r) () -> IO ())-  -> TestTree+  :: forall r . (LogIo r) => IO (Eff (Processes r) () -> IO ()) -> TestTree concurrencyTests schedulerFactory =   let n = 100   in@@ -378,7 +429,8 @@           me <- self           traverse_             (const-              (spawn "reciever"+              (spawn+                "reciever"                 (do                   m <- receiveAnyMessage                   void (sendAnyMessage me m)@@ -398,24 +450,28 @@       $ scheduleAndAssert schedulerFactory       $ \assertEff -> do           me     <- self-          child1 <- spawn "reciever"+          child1 <- spawn+            "reciever"             (do               m <- receiveAnyMessage               void (sendAnyMessage me m)             )-          child2 <- spawn "sender" (foreverCheap (void (sendMessage 888888 (""::String))))+          child2 <- spawn+            "sender"+            (foreverCheap (void (sendMessage 888888 ("" :: String))))           sendMessage child1 ("test" :: String)           i <- receiveMessage           sendInterrupt child2 testInterruptReason-          assertEff "" (i == ("test"::String))+          assertEff "" (i == ("test" :: String))       , testCase "most processes send foreverCheap"       $ scheduleAndAssert schedulerFactory       $ \assertEff -> do           me <- self           traverse_-            (\(i :: Int) -> spawn "sender"$ do+            (\(i :: Int) -> spawn "sender" $ do               when (i `rem` 5 == 0) $ void $ sendMessage me i-              foreverCheap $ void (sendMessage 888 ("test message to 888" :: String))+              foreverCheap+                $ void (sendMessage 888 ("test message to 888" :: String))             )             [0 .. n]           oks <- replicateM (length [0, 5 .. n]) receiveMessage@@ -425,7 +481,7 @@       $ \assertEff -> do           me <- self           traverse_-            (\(i :: Int) -> spawn "sender"$ do+            (\(i :: Int) -> spawn "sender" $ do               when (i `rem` 5 == 0) $ void $ sendMessage me i               foreverCheap $ void self             )@@ -453,7 +509,11 @@               when (i `rem` 5 == 0) $ void $ sendMessage me i               parent <- self               foreverCheap $ void-                (spawn  "sender" (void (sendMessage parent ("test msg from child"::String))))+                (spawn+                  "sender"+                  (void (sendMessage parent ("test msg from child" :: String))+                  )+                )             )             [0 .. n]           oks <- replicateM (length [0, 5 .. n]) receiveMessage@@ -463,7 +523,7 @@       $ \assertEff -> do           me <- self           traverse_-            (\(i :: Int) -> spawn  "sender" $ do+            (\(i :: Int) -> spawn "sender" $ do               when (i `rem` 5 == 0) $ void $ sendMessage me i               foreverCheap $ void receiveAnyMessage             )@@ -473,10 +533,7 @@       ]  exitTests-  :: forall r-   . (LogIo r)-  => IO (Eff (Processes r) () -> IO ())-  -> TestTree+  :: forall r . (LogIo r) => IO (Eff (Processes r) () -> IO ()) -> TestTree exitTests schedulerFactory =   testGroup "process exit tests"     $ [ testGroup@@ -509,11 +566,18 @@           [ ( "receiving"             , void               (send-                (ReceiveSelectedMessage @r (filterMessage (== ("test message" :: String))))+                (ReceiveSelectedMessage @r+                  (filterMessage (== ("test message" :: String)))+                )               )             )           , ( "sending"-            , void (send (SendMessage @r 44444 (toStrictDynamic ("test message" :: String))))+            , void+              (send+                (SendMessage @r 44444+                                (toStrictDynamic ("test message" :: String))+                )+              )             )           , ( "sending shutdown"             , void (send (SendShutdown @r 44444 ExitNormally))@@ -522,7 +586,8 @@           , ( "spawn-ing"             , void               (send-                (Spawn @r  "reciever"+                (Spawn @r+                  "reciever"                   (void (send (ReceiveSelectedMessage @r selectAnyMessage)))                 )               )@@ -541,11 +606,18 @@           [ ( "receiving"             , void               (send-                (ReceiveSelectedMessage @r (filterMessage (== ("test message"::String))))+                (ReceiveSelectedMessage @r+                  (filterMessage (== ("test message" :: String)))+                )               )             )           , ( "sending"-            , void (send (SendMessage @r 44444 (toStrictDynamic ("test message"::String))))+            , void+              (send+                (SendMessage @r 44444+                                (toStrictDynamic ("test message" :: String))+                )+              )             )           , ( "sending shutdown"             , void (send (SendShutdown @r 44444 ExitNormally))@@ -555,7 +627,8 @@             , void               (send                 (Spawn @r-                  "reciever"  (void (send (ReceiveSelectedMessage @r selectAnyMessage)))+                  "reciever"+                  (void (send (ReceiveSelectedMessage @r selectAnyMessage)))                 )               )             )@@ -573,7 +646,7 @@           $ \assertEff -> do               p1 <- spawn "busyloop" $ foreverCheap busyEffect               lift (threadDelay 1000)-              void $ spawn "sleep loop"  $ do+              void $ spawn "sleep loop" $ do                 lift (threadDelay 1000)                 doExit               lift (threadDelay 100000)@@ -587,11 +660,18 @@           [ ( "receiving"             , void               (send-                (ReceiveSelectedMessage @r (filterMessage (== ("test message"::String))))+                (ReceiveSelectedMessage @r+                  (filterMessage (== ("test message" :: String)))+                )               )             )           , ( "sending"-            , void (send (SendMessage @r 44444 (toStrictDynamic ("test message"::String))))+            , void+              (send+                (SendMessage @r 44444+                                (toStrictDynamic ("test message" :: String))+                )+              )             )           , ( "sending shutdown"             , void (send (SendShutdown @r 44444 ExitNormally))@@ -610,8 +690,10 @@         , (howToExit, doExit    ) <-           [ ("normally"        , void exitNormally)           , ("simply returning", return ())-          , ("raiseError", void (interrupt (ErrorInterrupt "test error raised")))-          , ("exitWithError"   , void (exitWithError "test error exit"))+          , ( "raiseError"+            , void (interrupt (ErrorInterrupt "test error raised"))+            )+          , ("exitWithError", void (exitWithError "test error exit"))           , ( "sendShutdown to self"             , do               me <- self@@ -623,10 +705,7 @@   sendShutdownTests-  :: forall r-   . (LogIo r)-  => IO (Eff (Processes r) () -> IO ())-  -> TestTree+  :: forall r . (LogIo r) => IO (Eff (Processes r) () -> IO ()) -> TestTree sendShutdownTests schedulerFactory = testGroup   "sendShutdown"   [ testCase "... self" $ applySchedulerFactory schedulerFactory $ do@@ -650,19 +729,22 @@     $ scheduleAndAssert schedulerFactory     $ \assertEff -> do         me    <- self-        other <- spawn "interrupted"+        other <- spawn+          "interrupted"           (do-            untilInterrupted (SendMessage @r 666666 (toStrictDynamic ("test"::String)))-            void (sendMessage me ("OK"::String))+            untilInterrupted+              (SendMessage @r 666666 (toStrictDynamic ("test" :: String)))+            void (sendMessage me ("OK" :: String))           )         void (sendInterrupt other testInterruptReason)         a <- receiveMessage-        assertEff "" (a == ("OK"::String))+        assertEff "" (a == ("OK" :: String))     , testCase "while it is receiving"     $ scheduleAndAssert schedulerFactory     $ \assertEff -> do         me    <- self-        other <- spawn "interrupted"+        other <- spawn+          "interrupted"           (do             untilInterrupted (ReceiveSelectedMessage @r selectAnyMessage)             void (sendMessage me ("OK" :: String))@@ -674,7 +756,8 @@     $ scheduleAndAssert schedulerFactory     $ \assertEff -> do         me    <- self-        other <- spawn "interrupted"+        other <- spawn+          "interrupted"           (do             untilInterrupted (SelfPid @r)             void (sendMessage me ("OK" :: String))@@ -686,26 +769,28 @@     $ scheduleAndAssert schedulerFactory     $ \assertEff -> do         me    <- self-        other <- spawn "interrupted"+        other <- spawn+          "interrupted"           (do             untilInterrupted (Spawn @r "returner" (return ()))-            void (sendMessage me ("OK"::String))+            void (sendMessage me ("OK" :: String))           )         void (sendInterrupt other testInterruptReason)         a <- receiveMessage-        assertEff "" (a == ("OK"::String))+        assertEff "" (a == ("OK" :: String))     , testCase "while it is sending shutdown messages"     $ scheduleAndAssert schedulerFactory     $ \assertEff -> do         me    <- self-        other <- spawn "interrupted"+        other <- spawn+          "interrupted"           (do             untilInterrupted (SendShutdown @r 777 ExitNormally)-            void (sendMessage me ("OK"::String))+            void (sendMessage me ("OK" :: String))           )         void (sendInterrupt other testInterruptReason)         a <- receiveMessage-        assertEff "" (a == ("OK"::String))+        assertEff "" (a == ("OK" :: String))     , testCase "handleInterrupt handles my own interrupts"     $ scheduleAndAssert schedulerFactory     $ \assertEff ->@@ -716,10 +801,7 @@   ]  linkingTests-  :: forall r-   . (LogIo r)-  => IO (Eff (Processes r) () -> IO ())-  -> TestTree+  :: forall r . (LogIo r) => IO (Eff (Processes r) () -> IO ()) -> TestTree linkingTests schedulerFactory = setTravisTestOptions   (testGroup     "process linking tests"@@ -762,7 +844,7 @@           (lift . (@?= LinkedProcessCrashed foo))           (do             linkProcess foo-            sendShutdown foo ExitProcessCancelled+            self >>= sendShutdown foo . ExitProcessCancelled . Just             void (receiveMessage @Void)           )     , testCase "link multiple times"@@ -779,7 +861,7 @@             linkProcess foo             linkProcess foo             linkProcess foo-            sendShutdown foo ExitProcessCancelled+            self >>= sendShutdown foo . ExitProcessCancelled . Just             void (receiveMessage @Void)           )     , testCase "unlink multiple times"@@ -798,18 +880,22 @@             unlinkProcess foo             unlinkProcess foo             withMonitor foo $ \ref -> do-              sendShutdown foo ExitProcessCancelled+              self >>= sendShutdown foo . ExitProcessCancelled . Just               void (receiveSelectedMessage (selectProcessDown ref))           )     , testCase "spawnLink" $ applySchedulerFactory schedulerFactory $ do       let foo = void (receiveMessage @Void)-      handleInterrupts (\er -> lift (isProcessDownInterrupt Nothing er @? show er)) $ do-        x <- spawnLink "foo" foo-        sendShutdown x ExitProcessCancelled-        void (receiveMessage @Void)-    , testCase "spawnLink and child exits by returning from spawn" $ applySchedulerFactory schedulerFactory $ do+      handleInterrupts+          (\er -> lift (isProcessDownInterrupt Nothing er @? show er))+        $ do+            x <- spawnLink "foo" foo+            self >>= sendShutdown x . ExitProcessCancelled . Just+            void (receiveMessage @Void)+    , testCase "spawnLink and child exits by returning from spawn"+    $ applySchedulerFactory schedulerFactory+    $ do         me <- self-        u <- spawn "unlinker" $ do+        u  <- spawn "unlinker" $ do           logCritical "unlinked child started"           l <- spawnLink "linked" $ do             logCritical "linked child started"@@ -823,16 +909,17 @@         sendMessage l ()         pL@(ProcessDown _ _ _) <- receiveMessage         logCritical' ("linked process down: " <> show pL)-        _ <- monitor u+        _   <- monitor u         mpU <- receiveAfter (TimeoutMicros 1000)         case mpU of           Just (pU@(ProcessDown _ _ _)) ->             error ("unlinked process down: " <> show pU)-          Nothing ->  logInfo "passed"--    , testCase "spawnLink and child exits via exitWithError" $ applySchedulerFactory schedulerFactory $ do+          Nothing -> logInfo "passed"+    , testCase "spawnLink and child exits via exitWithError"+    $ applySchedulerFactory schedulerFactory+    $ do         me <- self-        u <- spawn "unlinker" $ do+        u  <- spawn "unlinker" $ do           logCritical "unlinked child started"           l <- spawnLink "linker" $ do             logCritical "linked child started"@@ -847,13 +934,12 @@         sendMessage l ()         pL@(ProcessDown _ _ _) <- receiveMessage         logCritical' ("linked process down: " <> show pL)-        _ <- monitor u+        _   <- monitor u         mpU <- receiveAfter (TimeoutMicros 1000)         case mpU of           Just (pU@(ProcessDown _ _ _)) ->             logInfo' ("unlinked process down: " <> show pU)-          Nothing ->  error "linked process not exited!"-+          Nothing -> error "linked process not exited!"     , testCase "ignore normal exit"     $ applySchedulerFactory schedulerFactory     $ do@@ -862,26 +948,31 @@               logNotice "linker"               foreverCheap $ do                 x <- receiveMessage-                linkProcess x-                sendMessage mainProc True+                case x of+                  Right p -> do+                    linkProcess p+                    sendMessage mainProc True+                  Left e ->+                    exitBecause e         linker <- spawnLink "link-server" linkingServer         logNotice "mainProc"         do           x <- spawnLink "x1" (logNotice "x 1" >> void (receiveMessage @Void))           withMonitor x $ \xRef -> do-            sendMessage linker x+            sendMessage linker (Right x :: Either (Interrupt 'NoRecovery) ProcessId)             void $ receiveSelectedMessage (filterMessage id)             sendShutdown x ExitNormally             void (receiveSelectedMessage (selectProcessDown xRef))         do           x <- spawnLink "x2" (logNotice "x 2" >> void (receiveMessage @Void))           withMonitor x $ \xRef -> do-            sendMessage linker x+            sendMessage linker (Right x :: Either (Interrupt 'NoRecovery) ProcessId)             void $ receiveSelectedMessage (filterMessage id)             sendShutdown x ExitNormally             void (receiveSelectedMessage (selectProcessDown xRef))         handleInterrupts (lift . (LinkedProcessCrashed linker @=?)) $ do-          sendShutdown linker ExitProcessCancelled+          me <- self+          sendMessage linker (Left (ExitProcessCancelled (Just me)) :: Either (Interrupt 'NoRecovery) ProcessId)           void (receiveMessage @Void)     , testCase "unlink" $ applySchedulerFactory schedulerFactory $ do       let@@ -892,7 +983,7 @@           lift (("unlink foo1" :: String) @=? r1)           unlinkProcess foo1Pid           sendMessage barPid ("unlinked foo1" :: String, foo1Pid)-          receiveMessage >>= lift . (@?= ("the end":: String))+          receiveMessage >>= lift . (@?= ("the end" :: String))           exitWithError "foo two"         bar foo2Pid parentPid = do           linkProcess foo2Pid@@ -904,7 +995,7 @@             (const (return ()))             (do               linkProcess foo1Pid-              sendShutdown foo1Pid ExitProcessCancelled+              self >>= sendShutdown foo1Pid . ExitProcessCancelled . Just               void (receiveMessage @Void)             )           handleInterrupts@@ -930,10 +1021,7 @@   )  monitoringTests-  :: forall r-   . (LogIo r)-  => IO (Eff (Processes r) () -> IO ())-  -> TestTree+  :: forall r . (LogIo r) => IO (Eff (Processes r) () -> IO ()) -> TestTree monitoringTests schedulerFactory = setTravisTestOptions   (testGroup     "process monitoring tests"@@ -943,8 +1031,42 @@         let badPid = 132123         ref <- monitor badPid         pd  <- receiveSelectedMessage (selectProcessDown ref)-        lift (downReason pd @?= SomeExitReason (OtherProcessNotRunning badPid))+        lift (downReason pd @?= ExitOtherProcessNotRunning badPid)         lift (threadDelay 10000)+    , testCase+      "monitor twice, once when it is running and one, when the monitored process is not running (variant 1)"+    $ applySchedulerFactory schedulerFactory+    $ do+        target <- spawn "reciever-exit-value" (receiveMessage >>= exitBecause)+        me     <- self+        spawn_ "monitor 1" $ do+          ref1 <- monitor target+          receiveSelectedMessage (selectProcessDown ref1)+            >>= sendMessage me+            .   Just+        lift (threadDelay 10000)+        sendMessage target ExitNormally+        pd1 <- receiveMessage+        lift (downReason <$> pd1 @?= Just ExitNormally)+        ref <- monitor target+        pd2 <- receiveSelectedMessage (selectProcessDown ref)+        lift (downReason pd2 @?= ExitOtherProcessNotRunning target)+        lift (threadDelay 10000)+    , testCase+      "spawn, shutdown and monitor many times in a tight loop"+    $ applySchedulerFactory schedulerFactory+    $ do+        tests <- replicateM 60 $ spawn "spawn, shutdown and monitor test" $ do+          target <- spawn "target" (receiveMessage >>= exitBecause)+          replicateM_ 102 $ spawn_ "monitor" $ do+            ref <- monitor target+            logInfo ("monitoring now" <> pack (show ref))+            void $ receiveSelectedMessage (selectProcessDown ref)+          sendMessage target ExitNormally+          ref <- monitor target+          logInfo ("monitoring now" <> pack (show ref))+          void (receiveSelectedMessage (selectProcessDown ref))+        traverse_ awaitProcessDown tests     , testCase "monitored process exit normally"     $ applySchedulerFactory schedulerFactory     $ do@@ -952,7 +1074,7 @@         ref    <- monitor target         sendMessage target ExitNormally         pd <- receiveSelectedMessage (selectProcessDown ref)-        lift (downReason pd @?= SomeExitReason ExitNormally)+        lift (downReason pd @?= ExitNormally)         lift (threadDelay 10000)     , testCase "multiple monitors some demonitored"     $ applySchedulerFactory schedulerFactory@@ -967,20 +1089,21 @@         demonitor ref5         sendMessage target ExitNormally         pd1 <- receiveSelectedMessage (selectProcessDown ref1)-        lift (downReason pd1 @?= SomeExitReason ExitNormally)+        lift (downReason pd1 @?= ExitNormally)         pd2 <- receiveSelectedMessage (selectProcessDown ref2)-        lift (downReason pd2 @?= SomeExitReason ExitNormally)+        lift (downReason pd2 @?= ExitNormally)         pd4 <- receiveSelectedMessage (selectProcessDown ref4)-        lift (downReason pd4 @?= SomeExitReason ExitNormally)+        lift (downReason pd4 @?= ExitNormally)         lift (threadDelay 10000)     , testCase "monitored process killed"     $ applySchedulerFactory schedulerFactory     $ do         target <- spawn "reciever-exit-value" (receiveMessage >>= exitBecause)         ref    <- monitor target-        sendMessage target ExitProcessCancelled+        self >>= sendMessage target . ExitProcessCancelled . Just         pd <- receiveSelectedMessage (selectProcessDown ref)-        lift (downReason pd @?= SomeExitReason ExitProcessCancelled)+        me <- self+        lift (downReason pd @?= ExitProcessCancelled (Just me))         lift (threadDelay 10000)     , testCase "demonitored process killed"     $ applySchedulerFactory schedulerFactory@@ -988,7 +1111,7 @@         target <- spawn "reciever-exit-value" (receiveMessage >>= exitBecause)         ref    <- monitor target         demonitor ref-        sendMessage target ExitProcessCancelled+        self >>= sendMessage target . ExitProcessCancelled . Just         me <- self         spawn_ "wait-and-send" (lift (threadDelay 10000) >> sendMessage me ())         pd <- receiveSelectedMessage@@ -1000,10 +1123,7 @@   )  timerTests-  :: forall r-   . (LogIo r)-  => IO (Eff (Processes r) () -> IO ())-  -> TestTree+  :: forall r . (LogIo r) => IO (Eff (Processes r) () -> IO ()) -> TestTree timerTests schedulerFactory = setTravisTestOptions   (testGroup     "process timer tests"@@ -1017,7 +1137,8 @@     $ applySchedulerFactory schedulerFactory     $ do         me    <- self-        other <- spawn "reciever"+        other <- spawn+          "reciever"           (do             r <- receiveMessage @()             lift (r @?= ())@@ -1036,7 +1157,8 @@             testMsg :: Float             testMsg = 123         me    <- self-        other <- spawn "test"+        other <- spawn+          "test"           (do             replicateM_ n $ sendMessage me ("bad message" :: String)             r <- receiveMessage @()@@ -1056,55 +1178,72 @@     ]   ) -processDetailsTests ::-     forall r. (LogIo r)-  => IO (Eff (Processes r) () -> IO ())-  -> TestTree-processDetailsTests schedulerFactory =-  setTravisTestOptions-    (testGroup-       "process info tests"-       [ testCase-           "no infos for a non-existant process"-           (applySchedulerFactory-              schedulerFactory-              (do let nonExistentPid = ProcessId 123-                  me <- self-                  pInfo <- getProcessState nonExistentPid-                  lift (assertEqual "" (nonExistentPid /= me) (isNothing pInfo))))-       , testCase-           "no infos for a dead process"-           (applySchedulerFactory-              schedulerFactory-              (do deadProc <- spawn "dead" (return ())-                  mr <- monitor deadProc-                  void $ receiveSelectedMessage (selectProcessDown mr)-                  pInfo <- getProcessState deadProc-                  lift (assertBool "no process state of a dead process expected" (isNothing pInfo))))-       , testGroup-           "process title tests"-           [ testCase-               "getProcessState returns the title passed to spawn"-               (applySchedulerFactory-                  schedulerFactory-                  (do let expectedTitle = "expected title"-                      p <- spawn expectedTitle (void receiveAnyMessage)-                      (actualTitle, _, _) <- fromJust <$> getProcessState p-                      lift (assertEqual "unexpected process title" expectedTitle actualTitle)))-           ]-       , testGroup "process details tests" [-            testCase-                       "update"-                       (applySchedulerFactory-                          schedulerFactory-                          (do let expected1 = "test details 1"-                                  expected2 = "test details 2"-                              updateProcessDetails expected1-                              (_, actual1 , _) <- fromJust <$> (self >>= getProcessState)-                              lift (assertEqual "1" expected1 actual1)-                              updateProcessDetails expected2-                              (_, actual2 , _) <- fromJust <$> (self >>= getProcessState)-                              lift (assertEqual "2" expected2 actual2)-                              ))-       ]-       ])+processDetailsTests+  :: forall r . (LogIo r) => IO (Eff (Processes r) () -> IO ()) -> TestTree+processDetailsTests schedulerFactory = setTravisTestOptions+  (testGroup+    "process info tests"+    [ testCase+      "no infos for a non-existant process"+      (applySchedulerFactory+        schedulerFactory+        (do+          let nonExistentPid = ProcessId 123+          me    <- self+          pInfo <- getProcessState nonExistentPid+          lift (assertEqual "" (nonExistentPid /= me) (isNothing pInfo))+        )+      )+    , testCase+      "no infos for a dead process"+      (applySchedulerFactory+        schedulerFactory+        (do+          deadProc <- spawn "dead" (return ())+          mr       <- monitor deadProc+          void $ receiveSelectedMessage (selectProcessDown mr)+          pInfo <- getProcessState deadProc+          lift+            (assertBool "no process state of a dead process expected"+                        (isNothing pInfo)+            )+        )+      )+    , testGroup+      "process title tests"+      [ testCase+          "getProcessState returns the title passed to spawn"+          (applySchedulerFactory+            schedulerFactory+            (do+              let expectedTitle = "expected title"+              p <- spawn expectedTitle (void receiveAnyMessage)+              (actualTitle, _, _) <- fromJust <$> getProcessState p+              lift+                (assertEqual "unexpected process title"+                             expectedTitle+                             actualTitle+                )+            )+          )+      ]+    , testGroup+      "process details tests"+      [ testCase+          "update"+          (applySchedulerFactory+            schedulerFactory+            (do+              let expected1 = "test details 1"+                  expected2 = "test details 2"+              updateProcessDetails expected1+              (_, actual1, _) <- fromJust <$> (self >>= getProcessState)+              lift (assertEqual "1" expected1 actual1)+              updateProcessDetails expected2+              (_, actual2, _) <- fromJust <$> (self >>= getProcessState)+              lift (assertEqual "2" expected2 actual2)+            )+          )+      ]+    ]+  )

@@ -1,291 +0,0 @@-module SupervisorTests-  ( test_Supervisor-  ) where--import Common-import Control.Concurrent (threadDelay)-import Control.DeepSeq-import Control.Eff-import Control.Eff.Concurrent-import Control.Eff.Concurrent.Protocol.EffectfulServer (Event(..))-import qualified Control.Eff.Concurrent.Protocol.StatefulServer as Stateful-import Control.Eff.Concurrent.Protocol.Supervisor as Sup-import Control.Eff.Concurrent.Process.Timer-import Data.Either (fromRight, isLeft, isRight)-import Data.Maybe (fromMaybe)-import Data.Text (pack)-import Data.Type.Pretty-import Data.Typeable (Typeable)-import Test.Tasty-import Test.Tasty.HUnit--test_Supervisor :: HasCallStack => TestTree-test_Supervisor =-  setTravisTestOptions $-  testGroup-    "Supervisor"-    (let startTestSup = startTestSupWith ExitWhenRequested (TimeoutMicros 500000)-         startTestSupWith e t = Sup.startSupervisor (MkSupConfig t (Stateful.Init . TestServerArgs e))-         spawnTestChild sup i = Sup.spawnChild sup i >>= either (lift . assertFailure . show) pure-      in [ runTestCase "The supervisor starts and is shut down" $ do-             outerSelf <- self-             testWorker <--               spawn "test-worker" $ do-                 sup <- startTestSup-                 sendMessage outerSelf sup-                 () <- receiveMessage-                 sendMessage outerSelf ()-                 () <- receiveMessage-                 Sup.stopSupervisor sup-             unlinkProcess testWorker-             sup <- receiveMessage :: Eff Effects  (Endpoint (Sup.Sup (Stateful.Stateful TestProtocol)))-             supAliveAfter1 <- isSupervisorAlive sup-             logInfo ("still alive 1: " <> pack (show supAliveAfter1))-             lift (supAliveAfter1 @=? True)-             sendMessage testWorker ()-             () <- receiveMessage-             supAliveAfter2 <- isSupervisorAlive sup-             logInfo ("still alive 2: " <> pack (show supAliveAfter2))-             lift (supAliveAfter2 @=? True)-             sendMessage testWorker ()-             testWorkerMonitorRef <- monitor testWorker-             d1 <- receiveSelectedMessage (selectProcessDown testWorkerMonitorRef)-             logInfo ("got test worker down: " <> pack (show d1))-             testSupervisorMonitorRef <- monitorSupervisor sup-             d2 <- receiveSelectedMessage (selectProcessDown testSupervisorMonitorRef)-             logInfo ("got supervisor down: " <> pack (show d2))-             supAliveAfterOwnerExited <- isSupervisorAlive sup-             logInfo ("still alive after owner exited: " <> pack (show supAliveAfterOwnerExited))-             lift (supAliveAfterOwnerExited @=? False)-         , testGroup-             "Diagnostics"-             [ runTestCase "When only time passes the diagnostics do not change" $ do-                 sup <- startTestSup-                 info1 <- Sup.getDiagnosticInfo sup-                 lift (threadDelay 10000)-                 info2 <- Sup.getDiagnosticInfo sup-                 lift (assertEqual "diagnostics should not differ: " info1 info2)-             , runTestCase "When a child is started the diagnostics change" $ do-                 sup <- startTestSup-                 info1 <- Sup.getDiagnosticInfo sup-                 logInfo ("got diagnostics: " <> info1)-                 let childId = 1-                 _child <- fromRight (error "failed to spawn child") <$> Sup.spawnChild sup childId-                 info2 <- Sup.getDiagnosticInfo sup-                 logInfo ("got diagnostics: " <> info2)-                 lift $ assertBool ("diagnostics should differ: " ++ show (info1, info2)) (info1 /= info2)-             ]-         , let childId = 1-            in testGroup-                 "Startup and shutdown"-                 [ runTestCase "When a supervisor is shut down, all children are shutdown" $ do-                     sup <- startTestSup-                     child <- spawnTestChild sup childId-                     let childPid = _fromEndpoint child-                     supMon <- monitorSupervisor sup-                     childMon <- monitor childPid-                     isProcessAlive childPid >>= lift . assertBool "child process not running"-                     isSupervisorAlive sup >>= lift . assertBool "supervisor process not running"-                     call child (TestGetStringLength "123") >>= lift . assertEqual "child not working" 3-                     stopSupervisor sup-                     d1@(ProcessDown mon1 er1 _) <--                       fromMaybe (error "receive timeout 1") <$> receiveAfter (TimeoutMicros 1000000)-                     logInfo ("got process down: " <> pack (show d1))-                     d2@(ProcessDown mon2 er2 _) <--                       fromMaybe (error "receive timeout 2") <$> receiveAfter (TimeoutMicros 1000000)-                     logInfo ("got process down: " <> pack (show d2))-                     case if mon1 == supMon && mon2 == childMon-                            then Right (er1, er2)-                            else if mon1 == childMon && mon2 == supMon-                                   then Right (er2, er1)-                                   else Left-                                          ("unexpected monitor down: first: " <> show (mon1, er1) <> ", and then: " <>-                                           show (mon2, er2) <>-                                           ", supMon: " <>-                                           show supMon <>-                                           ", childMon: " <>-                                           show childMon) of-                       Right (supER, childER) -> do-                         lift (assertEqual "bad supervisor exit reason" (SomeExitReason ExitNormally) supER)-                         lift (assertEqual "bad child exit reason" (SomeExitReason ExitNormally) childER)-                       Left x -> lift (assertFailure x)-                 , runTestCase-                     "When a supervisor is shut down, children that won't shutdown, are killed after some time" $ do-                     sup <- startTestSupWith IgnoreNormalExitRequest (TimeoutMicros 10000)-                     child <- spawnTestChild sup childId-                     let childPid = _fromEndpoint child-                     supMon <- monitorSupervisor sup-                     childMon <- monitor childPid-                     isProcessAlive childPid >>= lift . assertBool "child process not running"-                     isSupervisorAlive sup >>= lift . assertBool "supervisor process not running"-                     call child (TestGetStringLength "123") >>= lift . assertEqual "child not working" 3-                     stopSupervisor sup-                     d1@(ProcessDown mon1 er1 _) <--                       fromMaybe (error "receive timeout 1") <$> receiveAfter (TimeoutMicros 1000000)-                     logInfo ("got process down: " <> pack (show d1))-                     d2@(ProcessDown mon2 er2 _) <--                       fromMaybe (error "receive timeout 2") <$> receiveAfter (TimeoutMicros 1000000)-                     logInfo ("got process down: " <> pack (show d2))-                     case if mon1 == supMon && mon2 == childMon-                            then Right (er1, er2)-                            else if mon1 == childMon && mon2 == supMon-                                   then Right (er2, er1)-                                   else Left-                                          ("unexpected monitor down: first: " <> show (mon1, er1) <> ", and then: " <>-                                           show (mon2, er2) <>-                                           ", supMon: " <>-                                           show supMon <>-                                           ", childMon: " <>-                                           show childMon) of-                       Right (supER, childER) -> do-                         lift (assertEqual "bad supervisor exit reason" (SomeExitReason ExitNormally) supER)-                         lift (assertBool "bad child exit reason" (isLeft (fromSomeExitReason childER)))-                       Left x -> lift (assertFailure x)-                 ]-         , let i = 123-            in testGroup-                 "Spawning and Using Children"-                 [ runTestCase-                     "When a supervisor is requested to start two children with the same id, an already started error is returned" $ do-                     sup <- startTestSup-                     c <- spawnTestChild sup i-                     x <- Sup.spawnChild sup i-                     case x of-                       Left (AlreadyStarted i' c') ->-                         lift $ do-                           assertEqual "bad pid returned" c c'-                           assertEqual "bad child id returned" 123 i'-                       _ -> lift (assertFailure "AlreadyStarted expected!")-                 , runTestCase "When a child is started it can be lookup up" $ do-                     sup <- startTestSup-                     c <- spawnTestChild sup i-                     c' <- Sup.lookupChild sup i >>= maybe (lift (assertFailure "child not found")) pure-                     lift (assertEqual "lookupChild returned wrong child" c c')-                 , runTestCase "When several children are started they can be lookup up and don't crash" $ do-                     sup <- startTestSup-                     c <- spawnTestChild sup i-                     someOtherChild <- spawnTestChild sup (i + 1)-                     c' <- Sup.lookupChild sup i >>= maybe (lift (assertFailure "child not found")) pure-                     lift (assertEqual "lookupChild returned wrong child" c c')-                     childStillRunning <- isProcessAlive (_fromEndpoint c)-                     lift (assertBool "child not running" childStillRunning)-                     someOtherChildStillRunning <- isProcessAlive (_fromEndpoint someOtherChild)-                     lift (assertBool "someOtherChild not running" someOtherChildStillRunning)-                 , let startTestSupAndChild = do-                         sup <- startTestSup-                         c <- spawnTestChild sup i-                         cm <- monitor (_fromEndpoint c)-                         return (sup, cm)-                    in testGroup-                         "Stopping children"-                         [ runTestCase "When a child is started it can be stopped" $ do-                             (sup, cm) <- startTestSupAndChild-                             Sup.stopChild sup i >>= lift . assertBool "child not found"-                             (ProcessDown _ r _) <- receiveSelectedMessage (selectProcessDown cm)-                             lift (assertEqual "bad exit reason" (SomeExitReason ExitNormally) r)-                         , runTestCase-                             "When a child is stopped but doesn't exit voluntarily, it is kill after some time" $ do-                             sup <- startTestSupWith IgnoreNormalExitRequest (TimeoutMicros 5000)-                             c <- spawnTestChild sup i-                             cm <- monitor (_fromEndpoint c)-                             Sup.stopChild sup i >>= lift . assertBool "child not found"-                             (ProcessDown _ r _) <- receiveSelectedMessage (selectProcessDown cm)-                             case r of-                               SomeExitReason (ExitUnhandledError _) -> return ()-                               _ -> lift (assertFailure ("bad exit reason: " ++ show r))-                         , runTestCase "When a stopChild is called with an unallocated ID, False is returned" $ do-                             (sup, _) <- startTestSupAndChild-                             Sup.stopChild sup (i + 1) >>= lift . assertBool "child not found" . not-                         , runTestCase "When a child is stopped, lookup won't find it" $ do-                             (sup, _) <- startTestSupAndChild-                             Sup.stopChild sup i >>= lift . assertBool "child not found"-                             x <- Sup.lookupChild sup i-                             lift (assertEqual "lookup should not find a child" Nothing x)-                         , runTestCase "When a child is stopped, the child id can be reused" $ do-                             (sup, _) <- startTestSupAndChild-                             Sup.stopChild sup i >>= lift . assertBool "child not found"-                             Sup.spawnChild sup i >>= lift . assertBool "id could not be reused" . isRight-                         ]-                 , let startTestSupAndChild = do-                         sup <- startTestSup-                         c <- spawnTestChild sup i-                         cm <- monitor (_fromEndpoint c)-                         return (sup, c, cm)-                    in testGroup-                         "Child exit handling"-                         [ runTestCase "When a child exits normally, lookupChild will not find it" $ do-                             (sup, c, cm) <- startTestSupAndChild-                             cast c (TestInterruptWith NormalExitRequested)-                             (ProcessDown _ r _) <- receiveSelectedMessage (selectProcessDown cm)-                             lift (assertEqual "bad exit reason" (SomeExitReason ExitNormally) r)-                             x <- Sup.lookupChild sup i-                             lift (assertEqual "lookup should not find a child" Nothing x)-                         , runTestCase "When a child exits with an error, lookupChild will not find it" $ do-                             (sup, c, cm) <- startTestSupAndChild-                             cast c (TestInterruptWith (ErrorInterrupt "test error reason"))-                             (ProcessDown _ _ _) <- receiveSelectedMessage (selectProcessDown cm)-                             x <- Sup.lookupChild sup i-                             lift (assertEqual "lookup should not find a child" Nothing x)-                         , runTestCase "When a child is interrupted from another process and dies, lookupChild will not find it" $ do-                             (sup, c, cm) <- startTestSupAndChild-                             sendInterrupt (_fromEndpoint c) NormalExitRequested-                             (ProcessDown _ _ _) <- receiveSelectedMessage (selectProcessDown cm)-                             x <- Sup.lookupChild sup i-                             lift (assertEqual "lookup should not find a child" Nothing x)-                         , runTestCase "When a child is shutdown from another process and dies, lookupChild will not find it" $ do-                             (sup, c, cm) <- startTestSupAndChild-                             sendShutdown (_fromEndpoint c) ExitProcessCancelled-                             (ProcessDown _ _ _) <- receiveSelectedMessage (selectProcessDown cm)-                             x <- Sup.lookupChild sup i-                             lift (assertEqual "lookup should not find a child" Nothing x)-                         ]-                 ]-         ])--data TestProtocol-  deriving (Typeable)--type instance ToPretty TestProtocol = PutStr "test"--instance HasPdu TestProtocol where-  data instance Pdu TestProtocol x where-    TestGetStringLength :: String -> Pdu TestProtocol ('Synchronous Int)-    TestInterruptWith :: Interrupt 'Recoverable -> Pdu TestProtocol 'Asynchronous-      deriving Typeable--instance NFData (Pdu TestProtocol x) where-  rnf (TestGetStringLength x) = rnf x-  rnf (TestInterruptWith x) = rnf x--instance Show (Pdu TestProtocol r) where-  show (TestGetStringLength s) = "TestGetStringLength " ++ show s-  show (TestInterruptWith s) = "TestInterruptWith " ++ show s--data TestProtocolServerMode-  = IgnoreNormalExitRequest-  | ExitWhenRequested-  deriving Eq--instance Stateful.Server TestProtocol Effects where-  update _me (TestServerArgs testMode tId) evt =-    case evt of-      OnCast (TestInterruptWith i) -> do-        logInfo (pack (show tId) <> ": stopping with: " <> pack (show i))-        interrupt i-      OnCall rt (TestGetStringLength str) -> do-        logInfo (pack (show tId) <> ": calculating length of: " <> pack str)-        sendReply rt (length str)-      OnInterrupt x -> do-        logNotice (pack (show tId) <> ": " <> pack (show x))-        if testMode == IgnoreNormalExitRequest-          then-            logNotice $ pack (show tId) <> ": ignoring normal exit request"-          else do-            logNotice $ pack (show tId) <> ": exitting normally"-            exitBecause (interruptToExit x)-      _ ->-        logDebug (pack (show tId) <> ": got some info: " <> pack (show evt))-  data instance StartArgument TestProtocol Effects = TestServerArgs TestProtocolServerMode Int--type instance ChildId (Stateful.Stateful TestProtocol) = Int-

@@ -0,0 +1,737 @@+{-# LANGUAGE UndecidableInstances #-}+module WatchdogTests(test_watchdogTests) where++import Common  hiding (runTestCase)+import qualified Common+import qualified Control.Eff.Concurrent.Protocol.StatefulServer as Stateful+import Control.Eff.Concurrent.Protocol.StatefulServer (Stateful)+import qualified Control.Eff.Concurrent.Protocol.Broker as Broker+import Control.Eff.Concurrent.Protocol.Broker (Broker)+import qualified Control.Eff.Concurrent.Protocol.Observer.Queue as OQ+import qualified Control.Eff.Concurrent.Protocol.Watchdog as Watchdog+import Control.Lens+import Data.Set (Set)+import qualified Data.Set as Set+import qualified Data.Map as Map+import Data.Maybe (isNothing, fromJust)+++runTestCase :: TestName -> Eff Effects () -> TestTree+runTestCase = Common.runTestCase+++test_watchdogTests :: HasCallStack => TestTree+test_watchdogTests =+  testGroup "watchdog"+    [ runTestCase "demonstrate Bookshelf" bookshelfDemo++    , testGroup "watchdog broker interaction"+      [++        runTestCase "test 1: when the broker exits, the watchdog does not care, it can be re-attached to a new broker" $ do+          wd <- Watchdog.startLink def+          unlinkProcess (wd ^. fromEndpoint)+          do+            broker <- Broker.startLink (Broker.statefulChild @BookShelf (TimeoutMicros 1_000_000) id)+            unlinkProcess (broker ^. fromEndpoint)+            logNotice "started broker"+            Watchdog.attachTemporary wd broker+            logNotice "attached watchdog"+            assertShutdown (broker ^. fromEndpoint) ExitNormally+            logNotice "stopped broker"++          do+            broker <- Broker.startLink (Broker.statefulChild @BookShelf (TimeoutMicros 1_000_000) id)+            unlinkProcess (broker ^. fromEndpoint)+            logNotice "started new broker"+            Watchdog.attachTemporary wd broker+            logNotice "attached broker"+            let expected = ExitUnhandledError "test-broker-kill"+            logNotice "crashing broker"+            assertShutdown (broker ^. fromEndpoint) expected+            logNotice "crashed broker"++          logNotice "shutting down watchdog"+          assertShutdown (wd^.fromEndpoint) ExitNormally+          logNotice "watchdog down"+++      , runTestCase "test 2: when the broker exits, and the watchdog is linked to it, it will exit" $ do+          wd <- Watchdog.startLink def+          unlinkProcess (wd ^. fromEndpoint)+          logNotice "started watchdog"+          do+            broker <- Broker.startLink (Broker.statefulChild @BookShelf (TimeoutMicros 1_000_000) id)+            unlinkProcess (broker ^. fromEndpoint)+            logNotice "started broker"+            Watchdog.attachPermanent wd broker+            logNotice "attached and linked broker"+            logNotice "crashing broker"+            let expected = ExitUnhandledError "test-broker-kill"+            sendShutdown (broker ^. fromEndpoint) expected+            logNotice "crashed broker"+            logNotice "waiting for watchdog to crash"+            awaitProcessDown (wd^.fromEndpoint)+              >>= lift+              . assertEqual "bad exit reason" (ExitOtherProcessNotRunning (broker^.fromEndpoint))+              . downReason+            logNotice "watchdog crashed"++      , runTestCase "test 3: when the same broker is attached twice, the second attachment is ignored" $ do+          wd <- Watchdog.startLink def+          unlinkProcess (wd ^. fromEndpoint)+          logNotice "started watchdog"+          do+            broker <- Broker.startLink (Broker.statefulChild @BookShelf (TimeoutMicros 1_000_000) id)+            unlinkProcess (broker ^. fromEndpoint)+            logNotice "started broker"+            Watchdog.attachTemporary wd broker+            logNotice "attached broker"+            Watchdog.attachTemporary wd broker+            logNotice "attached broker"+            logNotice "restart child test begin"+            restartChildTest broker+            logNotice "restart child test finished"++      , runTestCase "test 4: when the same broker is attached twice, the\+                     \ first linked then not linked, the watchdog is not linked anymore" $ do+          wd <- Watchdog.startLink def+          unlinkProcess (wd ^. fromEndpoint)+          logNotice "started watchdog"+          do+            broker <- Broker.startLink (Broker.statefulChild @BookShelf (TimeoutMicros 1_000_000) id)+            unlinkProcess (broker ^. fromEndpoint)+            logNotice "started broker"+            Watchdog.attachPermanent wd broker+            logNotice "attached and linked broker"+            Watchdog.attachTemporary wd broker+            logNotice "attached broker"+            logNotice "run restart child test"+            restartChildTest broker+            logNotice "restart child test finished"+            logNotice "crashing broker"+            let expected = ExitUnhandledError "test-broker-kill"+            assertShutdown (broker ^. fromEndpoint) expected+            logNotice "crashed broker"+          do+            broker <- Broker.startLink (Broker.statefulChild @BookShelf (TimeoutMicros 1_000_000) id)+            unlinkProcess (broker ^. fromEndpoint)+            logNotice "started new broker"+            Watchdog.attachTemporary wd broker+            logNotice "attached new broker"+            logNotice "run restart child test again to show everything is ok"+            restartChildTest broker+            logNotice "restart child test finished"+++      , runTestCase "test 5: when the same broker is attached twice,\+                    \ the first time not linked but then linked, the watchdog is linked \+                    \ to that broker" $ do+          wd <- Watchdog.startLink def+          unlinkProcess (wd ^. fromEndpoint)+          mwd <- monitor (wd ^. fromEndpoint)+          logNotice "started watchdog"+          do+            broker <- Broker.startLink (Broker.statefulChild @BookShelf (TimeoutMicros 1_000_000) id)+            unlinkProcess (broker ^. fromEndpoint)+            logNotice "started broker"+            Watchdog.attachTemporary wd broker+            logNotice "attached broker"+            Watchdog.attachPermanent wd broker+            logNotice "attached and linked broker"+            logNotice "run restart child test"+            restartChildTest broker+            logNotice "restart child test finished"+            logNotice "crashing broker"+            let expected = ExitUnhandledError "test-broker-kill"+            assertShutdown (broker ^. fromEndpoint) expected+            logNotice "crashed broker"+            logNotice "waiting for watchdog to crash"+            receiveSelectedMessage (selectProcessDown mwd)+              >>= lift+              . assertEqual "bad exit reason" (ExitOtherProcessNotRunning (broker^.fromEndpoint))+              . downReason+            logNotice "watchdog down"+++      , runTestCase "test 6: when multiple brokers are attached to a watchdog,\+                     \ and a linked broker exits, the watchdog should exit" $ do+          wd <- Watchdog.startLink def+          unlinkProcess (wd ^. fromEndpoint)+          wdMon <- monitor (wd ^. fromEndpoint)+          logNotice "started watchdog"+          do+            broker1 <- Broker.startLink (Broker.statefulChild @BookShelf (TimeoutMicros 1_000_000) id)+            unlinkProcess (broker1^.fromEndpoint)+            logNotice "started broker 1"+            Watchdog.attachPermanent wd broker1+            logNotice "attached + linked broker 1"++            broker2 <- Broker.startLink (Broker.statefulChild @BookShelf (TimeoutMicros 1_000_000) id)+            logNotice "started broker 2"+            Watchdog.attachTemporary wd broker2+            logNotice "attached broker 2"++            broker3 <- Broker.startLink (Broker.statefulChild @BookShelf (TimeoutMicros 1_000_000) id)+            logNotice "started broker 3"+            Watchdog.attachTemporary wd broker3+            logNotice "attached broker 3"+            unlinkProcess (broker3 ^. fromEndpoint)++            let expected = ExitUnhandledError "test-broker-kill 1"+            logNotice "crashing broker 1"+            assertShutdown (broker1 ^. fromEndpoint) expected+            logNotice "crashed broker 1"++            isProcessAlive (broker2 ^. fromEndpoint)+              >>= lift . assertBool "broker2 should be running"+            isProcessAlive (broker3 ^. fromEndpoint)+              >>= lift . assertBool "broker3 should be running"+            logNotice "other brokers are alive"++            logNotice "waiting for watchdog to crash"+            receiveSelectedMessage (selectProcessDown wdMon)+                >>= lift+                . assertEqual "bad exit reason" (ExitOtherProcessNotRunning (broker1^.fromEndpoint))+                . downReason+            logNotice "watchdog crashed"++      ]+    , testGroup "restarting children"+      [ runTestCase "test 7: when a child exits it is restarted" $ do+          broker <- Broker.startLink (Broker.statefulChild @BookShelf (TimeoutMicros 1_000_000) id)+          logNotice "started broker"+          wd <- Watchdog.startLink def+          logNotice "started watchdog"+          Watchdog.attachTemporary wd broker+          logNotice "attached broker"+          logNotice "running child tests"+          restartChildTest broker+          logNotice "finished child tests"++      , runTestCase "test 8: when the broker emits the shutting down\+                    \ event the watchdog does not restart children" $ do+          broker <- Broker.startLink (Broker.statefulChild @BookShelf (TimeoutMicros 1_000_000) id)+          unlinkProcess (broker ^. fromEndpoint)+          logNotice "started broker"+          wd <- Watchdog.startLink def+          unlinkProcess (wd ^. fromEndpoint)+          logNotice "started watchdog"+          Watchdog.attachTemporary wd broker+          OQ.observe @(Broker.ChildEvent (Stateful BookShelf)) (100 :: Int) broker $ do+            let c0 = BookShelfId 0+            do+              void $ Broker.spawnOrLookup broker c0+              awaitChildStartedEvent c0+              logNotice "bookshelf started"++              Broker.stopBroker broker+              awaitBrokershuttingDownEvent broker+              lift (threadDelay 10_000)+              logNotice "checking state"+              Watchdog.getCrashReports wd >>= lift . assertBool "no crash reports expected" . null++          assertShutdown (wd ^. fromEndpoint) ExitNormally+          logNotice "watchdog stopped"++      , runTestCase "test 9: a child is only restarted if it crashes" $ do+          broker <- Broker.startLink (Broker.statefulChild @BookShelf (TimeoutMicros 1_000_000) id)+          logNotice "started broker"+          wd <- Watchdog.startLink def+          logNotice "started watchdog"+          Watchdog.attachTemporary wd broker+          OQ.observe @(Broker.ChildEvent (Stateful BookShelf)) (100 :: Int) broker $ do+            let shelfId = BookShelfId 0+            do+              shelf <- Broker.spawnOrLookup broker shelfId+              awaitChildStartedEvent shelfId+              logNotice "bookshelf started"++              cast shelf StopShelf+              awaitChildDownEvent shelfId+              void $ awaitProcessDown (shelf ^. fromEndpoint)+              logNotice "shelf stopped"+              lift (threadDelay 10_000)+              Broker.lookupChild broker shelfId >>= lift . assertBool "child must not be restarted" . isNothing++          assertShutdown (wd ^. fromEndpoint) ExitNormally+          logNotice "watchdog stopped"++      , testGroup "child restart"+        $ let threeTimesASecond = 3 `Watchdog.crashesPerSeconds` 1+          in  [ runTestCase "test 10: if a child crashes 3 times in 300ms, waits 1.1 seconds\+                            \ and crashes again 3 times, and is restarted three times" $ do++                  broker <- Broker.startLink (Broker.statefulChild @BookShelf (TimeoutMicros 1_000_000) id)+                  logNotice "started broker"+                  wd <- Watchdog.startLink threeTimesASecond+                  logNotice "started watchdog"+                  Watchdog.attachTemporary wd broker++                  OQ.observe @(Broker.ChildEvent (Stateful BookShelf)) (100 :: Int) broker $ do+                    let crash3Times = do+                          logNotice "crashing 3 times"+                          replicateM_ 3 $ do+                            spawnAndCrashBookShelf broker shelfId+                            logNotice "waiting for last restart"+                            awaitChildStartedEvent shelfId+                            lift (threadDelay 100_000)+                        shelfId = BookShelfId 0++                    crash3Times+                    logNotice "sleeping"+                    lift (threadDelay 1_100_000)+                    logNotice "crashing again"+                    crash3Times++                  assertShutdown (wd ^. fromEndpoint) ExitNormally+                  logNotice "watchdog stopped"++              , runTestCase "test 11: if a child crashes 4 times within 1s it is not restarted\+                            \ and the watchdog exits with an error" $ do+                  broker <- Broker.startLink (Broker.statefulChild @BookShelf (TimeoutMicros 1_000_000) id)+                  logNotice "started broker"+                  wd <- Watchdog.startLink threeTimesASecond+                  logNotice "started watchdog"+                  unlinkProcess (wd ^. fromEndpoint)+                  Watchdog.attachTemporary wd broker++                  OQ.observe @(Broker.ChildEvent (Stateful BookShelf)) (100 :: Int) broker $ do+                    let shelfId = BookShelfId 0+                        crash3Times = do+                          logNotice "crashing 3 times"+                          replicateM_ 3 $ do+                            spawnAndCrashBookShelf broker shelfId+                            logNotice "waiting for restart"+                            awaitChildStartedEvent shelfId+                            cr <- Watchdog.getCrashReports @(Stateful BookShelf) wd+                            logNotice (pack (show cr))+                            lift (threadDelay 100_000)++                    crash3Times++                    logNotice "crashing 4. time"+                    spawnAndCrashBookShelf broker shelfId+                    lift (threadDelay 10_000)+                    Broker.lookupChild broker shelfId+                      >>= lift . assertBool "child must not be reststarted" . isNothing++                  assertShutdown (wd ^. fromEndpoint) ExitNormally+                  logNotice "watchdog stopped"++              , runTestCase "test 12: if a child of a linked broker crashes too often,\+                            \ the watchdog exits with an error and interrupts the broker" $ do++                  broker <- Broker.startLink (Broker.statefulChild @BookShelf (TimeoutMicros 1_000_000) id)+                  logNotice "started broker"+                  unlinkProcess (broker ^. fromEndpoint)+                  mBroker <- monitor (broker ^. fromEndpoint)+                  wd <- Watchdog.startLink threeTimesASecond+                  logNotice "started watchdog"+                  mwd <- monitor (wd ^. fromEndpoint)+                  unlinkProcess (wd ^. fromEndpoint)+                  Watchdog.attachPermanent wd broker++                  OQ.observe @(Broker.ChildEvent (Stateful BookShelf)) (100 :: Int) broker $ do+                    let shelfId = BookShelfId 0+                        crash3Times = do+                          logNotice "crashing 3 times"+                          replicateM_ 3 $ do+                            spawnAndCrashBookShelf broker shelfId+                            logNotice "waiting for restart"+                            awaitChildStartedEvent shelfId+                            lift (threadDelay 100_000)++                    crash3Times++                    logNotice "crashing 4. time"+                    spawnAndCrashBookShelf broker shelfId++                  logNotice "await broker down"+                  receiveSelectedMessage (selectProcessDown mBroker)+                    >>= lift+                        . assertEqual+                              "wrong exit reason"+                              (ExitUnhandledError "restart frequency exceeded")+                        . downReason+                  logNotice "broker down"++                  logNotice "await watchdog down"+                  receiveSelectedMessage (selectProcessDown mwd)+                    >>= lift+                        . assertEqual+                              "wrong exit reason"+                              (ExitUnhandledError "restart frequency exceeded")+                        . downReason+                  logNotice "watchdog down"++              , runTestCase "test 13: if the watchdog is killed, all watched children of all\+                            \ brokers are still alive" $ do+                  wd <- Watchdog.startLink (1 `Watchdog.crashesPerSeconds` 1)+                  unlinkProcess (wd ^. fromEndpoint)+                  logNotice "started watchdog"++                  brokerT <- Broker.startLink (Broker.statefulChild @BookShelf (TimeoutMicros 1_000_000) id)+                  unlinkProcess (brokerT^.fromEndpoint)+                  logNotice "started temporary broker"++                  brokerP <- Broker.startLink (Broker.statefulChild @BookShelf (TimeoutMicros 1_000_000) id)+                  unlinkProcess (brokerP^.fromEndpoint)+                  logNotice "started permanent broker"++                  Watchdog.attachTemporary wd brokerT+                  logNotice "attached temporary broker"++                  Watchdog.attachPermanent wd brokerP+                  logNotice "attached permanent broker"++                  let b0 = BookShelfId 0+                      b1 = BookShelfId 1+                      b2 = BookShelfId 2+                      b3 = BookShelfId 3++                  (e0, e1) <- OQ.observe @(Broker.ChildEvent (Stateful BookShelf)) (100 :: Int) brokerT $+                    (,) <$> spawnBookShelf brokerT b0 <*> spawnBookShelf brokerT b1+                  m0 <- monitor (e0^.fromEndpoint)+                  m1 <- monitor (e1^.fromEndpoint)+                  logNotice ("started bookshelfs of temporary broker: " <> pack (show (m0, m1)))++                  (e2, e3) <- OQ.observe @(Broker.ChildEvent (Stateful BookShelf)) (100 :: Int) brokerP $+                    (,) <$> spawnBookShelf brokerP b2 <*> spawnBookShelf brokerP b3+                  m2 <- monitor (e2^.fromEndpoint)+                  m3 <- monitor (e3^.fromEndpoint)+                  logNotice ("started bookshelfs of permanent broker: " <> pack (show (m2, m3)))++                  logNotice "crashing the watchdog"+                  assertShutdown (wd ^. fromEndpoint) (ExitUnhandledError "watchdog test crash")+                  logNotice "watchdog stopped"++                  call e0 (AddBook "Solaris")+                  logNotice "e0 still alive"+                  call e1 (AddBook "Solaris")+                  logNotice "e1 still alive"+                  call e2 (AddBook "Solaris")+                  logNotice "e2 still alive"+                  call e3 (AddBook "Solaris")+                  logNotice "e3 still alive"+                  assertShutdown (e0 ^. fromEndpoint) ExitNormally+                  assertShutdown (e1 ^. fromEndpoint) ExitNormally+                  assertShutdown (e2 ^. fromEndpoint) ExitNormally+                  assertShutdown (e3 ^. fromEndpoint) ExitNormally+                  assertShutdown (brokerT ^. fromEndpoint) ExitNormally+                  assertShutdown (brokerP ^. fromEndpoint) ExitNormally++              , runTestCase "test 14: if the watchdog receives OnChildDown\+                            \ for a known broker it behaves as if there\+                            \ was an OnChildSpawned for that child" $ do+                  wd <- Watchdog.startLink (1 `Watchdog.crashesPerSeconds` 1)+                  logNotice "started watchdog"++                  brokerT <- Broker.startLink (Broker.statefulChild @BookShelf (TimeoutMicros 1_000_000) id)+                  logNotice "started temporary broker"++                  OQ.observe @(Broker.ChildEvent (Stateful BookShelf)) (100 :: Int) brokerT $ do+                    let b0 = BookShelfId 0+                    e0 <- spawnBookShelf brokerT b0+                    logNotice ("started bookshelf of temporary broker: " <> pack (show e0))+                    Watchdog.attachTemporary wd brokerT+                    logNotice "attached temporary broker"+                    logNotice "crash the bookshelf"+                    assertShutdown (e0 ^. fromEndpoint) (ExitUnhandledError "bookshelf test crash")+                    awaitChildDownEvent b0+                    logNotice "bookshelf down"+                    awaitChildStartedEvent b0+                    logNotice "bookshelf restarted"+                    assertShutdown (wd ^. fromEndpoint) ExitNormally+                    assertShutdown (brokerT ^. fromEndpoint) ExitNormally++              , runTestCase "test 15: when a child exits normally,\+                            \ and a new child with the same ChildId\+                            \ crashes, the watchdog behaves as if\+                            \ the first child never existed" $ do+                  wd <- Watchdog.startLink (1 `Watchdog.crashesPerSeconds` 10)+                  logNotice "started watchdog"++                  brokerT <- Broker.startLink (Broker.statefulChild @BookShelf (TimeoutMicros 1_000_000) id)+                  logNotice "started temporary broker"++                  OQ.observe @(Broker.ChildEvent (Stateful BookShelf)) (100 :: Int) brokerT $ do+                    let b0 = BookShelfId 0++                    e0 <- spawnBookShelf brokerT b0+                    logNotice ("started bookshelf of temporary broker: " <> pack (show e0))++                    Watchdog.attachTemporary wd brokerT+                    logNotice "attached temporary broker"+                    logNotice "crash the bookshelf"+                    assertShutdown (e0 ^. fromEndpoint) (ExitUnhandledError "bookshelf test crash")+                    awaitChildDownEvent b0+                    logNotice "bookshelf down"+                    awaitChildStartedEvent b0+                    e1 <- fromJust <$> Broker.lookupChild brokerT b0+                    logNotice "bookshelf restarted"++                    logNotice "exit the bookshelf normally"+                    assertShutdown (e1 ^. fromEndpoint) ExitNormally+                    awaitChildDownEvent b0+                    logNotice "bookshelf stopped, starting a new one"+                    e2 <- spawnBookShelf brokerT b0+                    logNotice "bookshelf restarted"++                    assertShutdown (e2 ^. fromEndpoint) (ExitUnhandledError "bookshelf test crash")+                    awaitChildDownEvent b0+                    logNotice "bookshelf down"+                    awaitChildStartedEvent b0+                    logNotice "bookshelf restarted"++                    assertShutdown (wd ^. fromEndpoint) ExitNormally+                    assertShutdown (brokerT ^. fromEndpoint) ExitNormally++              , testGroup "test 16: the amount of state is kept to a minimum" $+                let+                  setup k = do+                    wd <- Watchdog.startLink (4 `Watchdog.crashesPerSeconds` 30)+                    logNotice "started watchdog"+                    brokerT <- Broker.startLink (Broker.statefulChild @BookShelf (TimeoutMicros 1_000_000) id)+                    logNotice "started temporary broker"+                    Watchdog.attachTemporary wd brokerT+                    logNotice "attached temporary broker"+                    brokerP <- Broker.startLink (Broker.statefulChild @BookShelf (TimeoutMicros 1_000_000) id)+                    logNotice "started permanent broker"+                    Watchdog.attachTemporary wd brokerP+                    logNotice "attached permanent broker"+                    do+                      p <- self+                      t <- spawn "test-proc" $ do+                              void $ k wd brokerT brokerP+                              sendMessage p ()+                      res <- receiveSelectedWithMonitorAfter t selectMessage (TimeoutMicros 10_000_000)+                      logNotice ("test-proc finish message: " <> pack (show res))+                      case res of+                        Left x ->+                          do logError  ("test-proc crashed: " <> pack (show x))+                             lift (assertFailure ("test-proc crashed: " <> show x))+                        Right () ->+                          logNotice "test-proc succeeded"++                  setupThenShutdown k =+                    setup $ \wd brokerT brokerP -> do+                      void $ k wd brokerT brokerP+                      assertShutdown (wd ^. fromEndpoint) ExitNormally+                      assertShutdown (brokerT ^. fromEndpoint) ExitNormally+                      assertShutdown (brokerP ^. fromEndpoint) ExitNormally+                in+                 [+                  runTestCase "test 17: child events of unknown broker don't add state" $+                    setupThenShutdown $ \wd _brokerT _brokerP -> do+                      someBroker <- asEndpoint @(Broker (Stateful BookShelf)) <$> spawn "someBroker" (pure ())+                      someChild <- asEndpoint @BookShelf <$> spawn "someChild" (pure ())+                      let someChildId = BookShelfId 2321+                      cast wd (Observed (Broker.OnChildDown someBroker someChildId someChild (ExitUnhandledError "test error")))+                      lift (threadDelay 1_000)+                      cr <- Watchdog.getCrashReports @(Stateful BookShelf) wd+                      logNotice (pack (show cr))+                      lift (assertBool "no crash reports expected" (Map.null cr))++                  , runTestCase "test 18: when a broker exits, the children of that broker are forgotten and ignored" $ do+                      setup $ \wd brokerT brokerP -> do+                        let+                          someChildId1 = BookShelfId 2321+                          someChildId2 = BookShelfId 2322+                          someChildId3 = BookShelfId 2323++                        (someChild1, someChild2)+                          <- OQ.observe @(Broker.ChildEvent (Stateful BookShelf))+                              (100 :: Int)+                              brokerT+                                ((,) <$> spawnBookShelf brokerT someChildId1 <*> spawnBookShelf brokerT someChildId2)++                        someChild3+                          <- OQ.observe @(Broker.ChildEvent (Stateful BookShelf))+                              (100 :: Int) brokerP (spawnBookShelf brokerP someChildId3)+                        do+                          cr <- Watchdog.getCrashReports @(Stateful BookShelf) wd+                          logNotice ("crash-reports: " <> pack (show cr))+                          lift (assertBool "no crash reports expected" (Map.null cr))++                        OQ.observe @(Broker.ChildEvent (Stateful BookShelf)) (100 :: Int) brokerT $ do+                            crashBookShelf someChild1 someChildId1+                            awaitChildStartedEvent someChildId1+                            crashBookShelf someChild2 someChildId2+                            awaitChildStartedEvent someChildId2++                        OQ.observe @(Broker.ChildEvent (Stateful BookShelf)) (100 :: Int) brokerP $ do+                            crashBookShelf someChild3 someChildId3+                            awaitChildStartedEvent someChildId3++                        do+                          cr <- Watchdog.getCrashReports @(Stateful BookShelf) wd+                          logNotice ("crash-reports: " <> pack (show cr))+                          lift (assertEqual "wrong number of crash reports" 3 (Map.size cr))++                        assertShutdown (brokerT ^. fromEndpoint) ExitNormally+                        do+                          cr <- Watchdog.getCrashReports @(Stateful BookShelf) wd+                          logNotice ("crash-reports: " <> pack (show cr))+                          lift (assertEqual "wrong number of crash reports" 1 (Map.size cr))++                        assertShutdown (wd ^. fromEndpoint) ExitNormally+                        assertShutdown (brokerP ^. fromEndpoint) ExitNormally+                 ]+              ]+      ]+    ]+++bookshelfDemo :: HasCallStack => Eff Effects ()+bookshelfDemo = do+  logNotice "Bookshelf Demo Begin"+  broker <- Broker.startLink (Broker.statefulChild @BookShelf (TimeoutMicros 1_000_000) id)+  shelf1 <- Broker.spawnOrLookup broker (BookShelfId 1)+  call shelf1 (AddBook "Solaris")+  call shelf1 GetBookList >>= logDebug . pack . show+  cast shelf1 (TakeBook "Solaris")+  call shelf1 GetBookList >>= logDebug . pack . show+  logNotice "Bookshelf Demo End"++restartChildTest :: HasCallStack => Endpoint (Broker (Stateful BookShelf)) -> Eff Effects ()+restartChildTest broker =+  OQ.observe @(Broker.ChildEvent (Stateful BookShelf)) (100 :: Int) broker $ do+    let c0 = BookShelfId 123+    spawnAndCrashBookShelf broker c0+    c01m <- Broker.lookupChild broker c0+    case c01m of+      Nothing ->+        lift (assertFailure "failed to lookup child, seems it wasn't restarted!")+      Just c01 -> do+        call c01 (AddBook "Solaris")+        logNotice "part 3 passed"++spawnBookShelf+  :: HasCallStack+  => Endpoint (Broker (Stateful BookShelf))+  -> Broker.ChildId (Stateful BookShelf)+  -> Eff (OQ.Reader (Broker.ChildEvent (Stateful BookShelf)) : Effects) (Endpoint BookShelf)+spawnBookShelf broker c0 = do+   logNotice "spawn or lookup book shelf"+   res <- Broker.spawnChild broker c0+   c00 <- case res of+     Left (Broker.AlreadyStarted _ ep) -> pure ep+     Right ep -> awaitChildStartedEvent c0 >> pure ep+   logNotice ("got book shelf: " <> pack (show c00))+   pure c00++spawnAndCrashBookShelf+  :: HasCallStack+  => Endpoint (Broker (Stateful BookShelf))+  -> Broker.ChildId (Stateful BookShelf)+  -> Eff (OQ.Reader (Broker.ChildEvent (Stateful BookShelf)) : Effects) ()+spawnAndCrashBookShelf broker c0 = do+   c00 <- spawnBookShelf broker c0+   crashBookShelf c00 c0++crashBookShelf+  :: HasCallStack+  => Endpoint BookShelf+  -> Broker.ChildId (Stateful BookShelf)+  -> Eff (OQ.Reader (Broker.ChildEvent (Stateful BookShelf)) : Effects) ()+crashBookShelf c00 c0 = do+   logNotice "adding Solaris"+   call c00 (AddBook "Solaris")+   logNotice "added Solaris"+   logNotice "adding Solaris again to crash the book shelf"+   handleInterrupts (const (pure ())) (call c00 (AddBook "Solaris")) -- adding twice the same book causes a crash+   logNotice "added Solaris again waiting for crash"+   void $ awaitProcessDown (c00 ^. fromEndpoint)+   logNotice "shelf crashed"+   awaitChildDownEvent c0++awaitBrokershuttingDownEvent broker = do+   logNotice ("waiting for broker shutting down event of " <> pack (show broker))+   evt <- OQ.await @(Broker.ChildEvent (Stateful BookShelf))+   case evt of+    (Broker.OnBrokerShuttingDown broker') | broker == broker' ->+      logNotice ("broker shutting down: " <> pack (show broker))+    otherEvent ->+      lift (assertFailure ("wrong event received: " <> show otherEvent))++awaitChildDownEvent c0 = do+   logNotice ("waiting for down event of " <> pack (show c0))+   evt <- OQ.await @(Broker.ChildEvent (Stateful BookShelf))+   case evt of+    (Broker.OnChildDown _ c' _ _) | c0 == c' ->+      logNotice ("child down: " <> pack (show c0))+    otherEvent ->+      lift (assertFailure ("wrong broker down event received: " <> show otherEvent))++awaitChildStartedEvent c0 = do+   logNotice ("waiting for start event of " <> pack (show c0))+   evt <- OQ.await @(Broker.ChildEvent (Stateful BookShelf))+   case evt of+    (Broker.OnChildSpawned _ c' _) | c0 == c' ->+      logNotice ("child started: " <> pack (show c0))+    otherEvent ->+      lift (assertFailure ("wrong broker start event received: " <> show otherEvent))+++data BookShelf deriving Typeable++type Book = String++instance HasPdu BookShelf where+  data Pdu BookShelf r where+    GetBookList :: Pdu BookShelf ('Synchronous [Book])+    AddBook :: Book -> Pdu BookShelf ('Synchronous ())+    TakeBook :: Book -> Pdu BookShelf 'Asynchronous+    StopShelf :: Pdu BookShelf 'Asynchronous+      deriving Typeable++instance Stateful.Server BookShelf Effects where+  newtype StartArgument BookShelf Effects = BookShelfId Int+    deriving (Show, Eq, Ord, Num, NFData, Typeable)++  newtype instance Model BookShelf = BookShelfModel (Set String) deriving (Eq, Show, Default)++  update _ shelfId event = do+    logDebug (pack (show shelfId ++ " " ++ show event))+    case event of++      Stateful.OnCast StopShelf -> do+        logNotice "stop shelf"+        exitNormally++      Stateful.OnCast (TakeBook book) -> do+        booksBefore <- Stateful.getModel @BookShelf+        booksAfter <- Stateful.modifyAndGetModel (over bookShelfModel (Set.delete book))+        when (booksBefore == booksAfter) (exitWithError "book not taken")++      Stateful.OnCall rt callEvent ->+        case callEvent of+          GetBookList -> do+            books <- Stateful.useModel bookShelfModel+            sendReply rt (toList books)++          AddBook book -> do+            booksBefore <- Stateful.getModel @BookShelf+            booksAfter <- Stateful.modifyAndGetModel (over bookShelfModel (Set.insert book))+            when (booksBefore == booksAfter) (exitWithError "book not added")+            sendReply rt ()++      other -> logWarning (pack (show other))++bookShelfModel :: Iso' (Stateful.Model BookShelf) (Set String)+bookShelfModel = iso (\(BookShelfModel s) -> s) BookShelfModel++type instance Broker.ChildId BookShelf = Stateful.StartArgument BookShelf Effects++instance NFData (Pdu BookShelf r) where+  rnf GetBookList = ()+  rnf StopShelf = ()+  rnf (AddBook b) = rnf b+  rnf (TakeBook b) = rnf b++instance Show (Pdu BookShelf r) where+  show GetBookList = "get-book-list"+  show StopShelf = "stop-shelf"+  show (AddBook b) = "add-book: " ++ b+  show (TakeBook b) = "take-book: " ++ b