extensible-effects-concurrent 0.23.0 → 0.24.0

raw patch · 33 files changed

+2379/−2441 lines, 33 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• ChangeLog.md +21/−12
• examples/example-1/Main.hs +72/−60
• examples/example-2/Main.hs +74/−81
• extensible-effects-concurrent.cabal +13/−12
• src/Control/Eff/Concurrent.hs +32/−80
• src/Control/Eff/Concurrent/Api.hs +0/−109
• src/Control/Eff/Concurrent/Api/Client.hs +0/−198
• src/Control/Eff/Concurrent/Api/GenServer.hs +0/−49
• src/Control/Eff/Concurrent/Api/Observer.hs +0/−267
• src/Control/Eff/Concurrent/Api/Observer/Queue.hs +0/−156
• src/Control/Eff/Concurrent/Api/Request.hs +0/−102
• src/Control/Eff/Concurrent/Api/Server.hs +0/−517
• src/Control/Eff/Concurrent/Api/Supervisor.hs +0/−460
• src/Control/Eff/Concurrent/Api/Supervisor/InternalState.hs +0/−101
• src/Control/Eff/Concurrent/Process.hs +5/−5
• src/Control/Eff/Concurrent/Process/Interactive.hs +9/−18
• src/Control/Eff/Concurrent/Protocol.hs +238/−0
• src/Control/Eff/Concurrent/Protocol/Client.hs +195/−0
• src/Control/Eff/Concurrent/Protocol/Observer.hs +286/−0
• src/Control/Eff/Concurrent/Protocol/Observer/Queue.hs +161/−0
• src/Control/Eff/Concurrent/Protocol/Request.hs +93/−0
• src/Control/Eff/Concurrent/Protocol/Server.hs +416/−0
• src/Control/Eff/Concurrent/Protocol/Supervisor.hs +439/−0
• src/Control/Eff/Concurrent/Protocol/Supervisor/InternalState.hs +105/−0
• src/Control/Eff/Log.hs +1/−0
• src/Control/Eff/Log/Handler.hs +7/−1
• src/Control/Eff/Log/Writer.hs +5/−11
• src/Control/Eff/LogWriter/Capture.hs +2/−2
• test/Common.hs +10/−3
• test/GenServerTests.hs +111/−0
• test/ProcessBehaviourTestCases.hs +33/−25
• test/ServerForkIO.hs +0/−112
• test/SupervisorTests.hs +51/−60

@@ -1,23 +1,32 @@ # Changelog for extensible-effects-concurrent -## Plan for future 0.24.0+## Plan for future Versions -- Add `gen_server` behaviour clone:    -    - Disallow non-`Api`-messages-    - Add `Api AnyMsg 'Asynchronous` -    - Add `Api GetInfo ('Synchronous Text)`  -         -    - Every `Api` type instance now **must** be an `NFData` -      and a `ToLogText` instance-    - `call` will always require a `Timeout`-    - `call` now monitors the caller-    - `call` now monitors the called process-    - `call` now returns `Either TimeoutError a`   +- Every `Api` type instance now **must** be an `NFData` +  and a `ToLogText` instance+- `call` will always require a `Timeout`+- `call` now monitors the caller+- `call` now monitors the called process+- `call` now returns `Either TimeoutError a`     - Logging improvements:     - Introduce `ToLogText`     - Remove `ToLogMessage`     - Remove `logXXX'` users have to use `logXXX` and `ToLogText` ++## 0.24.0++- Add `GenServer` module for `Api` handling: +       +    - Reduce the server to handle a single `Api` instance+        - Instances of `GenServer` must provide `lenses` and `prisms`+          for accessing the *other* Api instances like `Observer`+          and `ServerCallback` +    - Add `ServerCallback`, an internal API for interacting with these+      `GenServer` instances+        - `Api AnyMsg 'Asynchronous` +        - `Api GetInfo ('Synchronous Text)`  +           ## 0.23.0 

@@ -6,24 +6,25 @@ import           Control.Monad import           Data.Dynamic import           Control.Eff.Concurrent+import           Control.Eff.Concurrent.Protocol.Server import qualified Control.Exception             as Exc import qualified Data.Text as T import           Control.DeepSeq import           Data.Type.Pretty -data TestApi+data TestProtocol   deriving Typeable -type instance ToPretty TestApi = PutStr "test"+type instance ToPretty TestProtocol = PutStr "test" -data instance Api TestApi x where-  SayHello :: String -> Api TestApi ('Synchronous Bool)-  Shout :: String -> Api TestApi 'Asynchronous-  Terminate :: Api TestApi ('Synchronous ())-  TerminateError :: String -> Api TestApi ('Synchronous ())+data instance Pdu TestProtocol x where+  SayHello :: String -> Pdu TestProtocol ('Synchronous Bool)+  Shout :: String -> Pdu TestProtocol 'Asynchronous+  Terminate :: Pdu TestProtocol ('Synchronous ())+  TerminateError :: String -> Pdu TestProtocol ('Synchronous ())   deriving (Typeable) -instance NFData (Api TestApi x) where+instance NFData (Pdu TestProtocol x) where   rnf (SayHello s) = rnf s   rnf (Shout s) = rnf s   rnf Terminate = ()@@ -35,15 +36,15 @@  instance Exc.Exception MyException -deriving instance Show (Api TestApi x)+deriving instance Show (Pdu TestProtocol x)  main :: IO () main = defaultMain example -mainProcessSpawnsAChildAndReturns :: HasCallStack => Eff (InterruptableProcess q) ()+mainProcessSpawnsAChildAndReturns :: HasCallStack => Eff InterruptableProcEff () mainProcessSpawnsAChildAndReturns = void (spawn (void receiveAnyMessage)) -example:: ( HasCallStack, Member Logs q, Lifted IO q) => Eff (InterruptableProcess q) ()+example:: HasCallStack => Eff InterruptableProcEff () example = do   me <- self   logInfo (T.pack ("I am " ++ show me))@@ -54,71 +55,82 @@         x <- lift getLine         case x of           ('K' : rest) -> do-            callRegistered (TerminateError rest)+            call server (TerminateError rest)             go           ('S' : _) -> do-            callRegistered Terminate+            call server Terminate             go           ('C' : _) -> do-            castRegistered (Shout x)+            cast server (Shout x)             go           ('R' : rest) -> do-            replicateM_ (read rest) (castRegistered (Shout x))+            replicateM_ (read rest) (cast server (Shout x))             go           ('q' : _) -> logInfo "Done."           _         -> do-            res <- callRegistered (SayHello x)+            res <- call server (SayHello x)             logInfo (T.pack ("Result: " ++ show res))             go-  registerServer server go+  go -testServerLoop-  :: forall q-   . ( HasCallStack-     , Member Logs q-     , Lifted IO q-     )-  => Eff (InterruptableProcess q) (Server TestApi)-testServerLoop = spawnApiServer-  (handleCastTest <> handleCalls handleCallTest) handleTerminateTest+type instance GenServerProtocol TestProtocol = TestProtocol++testServerLoop :: Eff InterruptableProcEff (Endpoint TestProtocol)+testServerLoop = start (statelessGenServer handleReq "test-server-1")  where-  handleCastTest = handleCasts $ \(Shout x) -> do-    me <- self-    logInfo (T.pack (show me ++ " Shouting: " ++ x))-    return AwaitNext-  handleCallTest :: Api TestApi ('Synchronous r) -> (Eff (InterruptableProcess q) (Maybe r, CallbackResult 'Recoverable) -> xxx) -> xxx-  handleCallTest (SayHello "e1") k = k $ do-    me <- self-    logInfo (T.pack (show me ++ " raising an error"))-    interrupt (ErrorInterrupt "No body loves me... :,(")-  handleCallTest (SayHello "e2") k = k $ do-    me <- self-    logInfo (T.pack (show me ++ " throwing a MyException "))-    void (lift (Exc.throw MyException))-    pure (Nothing, AwaitNext)-  handleCallTest (SayHello "self") k = k $ do-    me <- self-    logInfo (T.pack (show me ++ " casting to self"))-    cast (asServer @TestApi me) (Shout "from me")-    return (Just False, AwaitNext)-  handleCallTest (SayHello "stop") k = k $ do-    me <- self-    logInfo (T.pack (show me ++ " stopping me"))-    return (Just False, StopServer (ErrorInterrupt "test error"))-  handleCallTest (SayHello x) k = k $ do-    me <- self-    logInfo (T.pack (show me ++ " Got Hello: " ++ x))-    return (Just (length x > 3), AwaitNext)-  handleCallTest Terminate k = k $ do+  handleReq :: GenServerId TestProtocol -> Event TestProtocol -> Eff (ToStatelessEffects InterruptableProcEff) ()+  handleReq _myId (OnRequest (Call orig Terminate)) = do     me <- self     logInfo (T.pack (show me ++ " exiting"))-    pure (Just (), StopServer NormalExitRequested)-  handleCallTest (TerminateError msg) k = k $ do+    sendReply orig ()+    interrupt NormalExitRequested++  handleReq _myId (OnRequest (Call orig (TerminateError e))) = do     me <- self-    logInfo (T.pack (show me ++ " exiting with error: " ++ msg))-    pure (Just (), StopServer (ErrorInterrupt msg))-  handleTerminateTest = InterruptCallback $ \msg -> do+    logInfo (T.pack (show me ++ " exiting with error: " ++ e))+    sendReply orig ()+    interrupt (ErrorInterrupt e)++  handleReq _myId (OnRequest (Call orig (SayHello mx))) =+    case mx of+      "e1" -> do+        me <- self+        logInfo (T.pack (show me ++ " raising an error"))+        interrupt (ErrorInterrupt "No body loves me... :,(")++      "e2" -> do+        me <- self+        logInfo (T.pack (show me ++ " throwing a MyException "))+        void (lift (Exc.throw MyException))++      "self" -> do+        me <- self+        logInfo (T.pack (show me ++ " casting to self"))+        cast (asEndpoint @TestProtocol me) (Shout "from me")+        sendReply orig False++      "stop" -> do+        me <- self+        logInfo (T.pack (show me ++ " stopping me"))+        sendReply orig False+        interrupt (ErrorInterrupt "test error")++      x -> do+        me <- self+        logInfo (T.pack (show me ++ " Got Hello: " ++ x))+        sendReply orig (length x > 3)++  handleReq _myId (OnRequest (Cast (Shout x))) = do     me <- self+    logInfo (T.pack (show me ++ " Shouting: " ++ x))++  handleReq _myId (OnInterrupt msg) = do+    me <- self     logInfo (T.pack (show me ++ " is exiting: " ++ show msg))     logProcessExit msg-    pure (StopServer (interruptToExit msg))+    interrupt msg++  handleReq _myId wtf = do+    me <- self+    logCritical (T.pack (show me ++ " WTF: " ++ show wtf))+

@@ -1,14 +1,17 @@+ -- | Another complete example for the library module Main where  import           Data.Dynamic import           Control.Eff import           Control.Eff.Concurrent-import           Control.Eff.State.Strict+import           Control.Eff.Concurrent.Protocol.Server import           Control.Monad import           Data.Foldable+import           Control.Lens import           Control.Concurrent import           Control.DeepSeq+import qualified Data.Text as T import           Data.Type.Pretty  main :: IO ()@@ -20,24 +23,25 @@  type instance ToPretty Counter = PutStr "counter" -data instance Api Counter x where-  Inc :: Api Counter 'Asynchronous-  Cnt :: Api Counter ('Synchronous Integer)+data instance Pdu Counter x where+  Inc :: Pdu Counter 'Asynchronous+  Cnt :: Pdu Counter ('Synchronous Integer)   deriving Typeable -instance NFData (Api Counter x) where+instance NFData (Pdu Counter x) where   rnf Inc = ()   rnf Cnt = ()  counterExample-  :: (Member Logs q, Lifted IO q)+  :: (Typeable q, LogsTo IO q, Lifted IO q)   => Eff (InterruptableProcess q) () counterExample = do-  (c, (co, (_sdp, cp))) <- spawnCounter+  c <- spawnCounter+  let cp = _fromEndpoint c   lift (threadDelay 500000)   o <- logCounterObservations   lift (threadDelay 500000)-  registerObserver o co+  registerObserver o c   lift (threadDelay 500000)   cast c Inc   lift (threadDelay 500000)@@ -63,11 +67,15 @@  type instance ToPretty SupiDupi = PutStr "supi dupi" -data instance Api SupiDupi r where-  Whoopediedoo :: Bool -> Api SupiDupi ('Synchronous (Maybe ()))+data instance Pdu SupiDupi r where+  Whoopediedoo :: Bool -> Pdu SupiDupi ('Synchronous (Maybe ()))   deriving Typeable -instance NFData (Api SupiDupi r) where+instance Show (Pdu SupiDupi r) where+  show (Whoopediedoo True) = "woopediedooo"+  show (Whoopediedoo False) = "no woopy doopy"++instance NFData (Pdu SupiDupi r) where   rnf (Whoopediedoo b) = rnf b  newtype CounterChanged = CounterChanged Integer@@ -75,81 +83,66 @@  type instance ToPretty CounterChanged = PutStr "counter changed" -spawnCounter-  :: (Member Logs q)-  => Eff-       (InterruptableProcess q)-       ( Server Counter-       , ( Server (ObserverRegistry CounterChanged)-         , (Server SupiDupi, ProcessId)-         )-       )-spawnCounter = spawnApiServerEffectful-  (manageObservers @CounterChanged . evalState (0 :: Integer) . evalState-    (Nothing :: Maybe (RequestOrigin (Api SupiDupi ( 'Synchronous (Maybe ())))))-  )-  (  handleCalls-      (\case-        Cnt ->-          ($ do-            val <- get-            return (Just val, AwaitNext)-          )-      )-  <> handleCasts-       (\case-         Inc -> do-           val <- get @Integer-           let val' = val + 1-           observed (CounterChanged val')-           put val'-           when (val' > 5) $ do-             get-               @( Maybe-                   (RequestOrigin (Api SupiDupi ( 'Synchronous (Maybe ()))))-               )-               >>= traverse_ (flip sendReply (Just ()))-             put-               (Nothing :: Maybe-                   (RequestOrigin (Api SupiDupi ( 'Synchronous (Maybe ()))))-               )+type SupiCounter = (Counter, ObserverRegistry CounterChanged, SupiDupi) -           return AwaitNext-       )-  ^: handleObserverRegistration-  ^: handleCallsDeferred+type instance ToPretty (Counter, ObserverRegistry CounterChanged, SupiDupi) = PutStr "supi-counter" -       (\origin ->-         (\case-           Whoopediedoo c -> do-             if c-               then put (Just origin)-               else-                 put-                   (Nothing :: Maybe-                       ( RequestOrigin-                           (Api SupiDupi ( 'Synchronous (Maybe ())))-                       )-                   )-             pure AwaitNext-         )-       )-  ^: logUnhandledMessages-  )-  stopServerOnInterrupt+instance (LogIo q) => Server SupiCounter (InterruptableProcess q) where -deriving instance Show (Api Counter x)+  type instance Model SupiCounter = (Integer, Observers CounterChanged, Maybe (RequestOrigin SupiCounter (Maybe ()))) +  data instance StartArgument SupiCounter (InterruptableProcess q) = MkEmptySupiCounter++  setup _ = return ((0, emptyObservers, Nothing), ())++  update _ = \case+    OnRequest req ->+      case req of+        Call orig callReq ->+          case callReq of+            ToPdu1 Cnt ->+              sendReply orig =<< useModel @SupiCounter _1+            ToPdu2 _ -> error "unreachable"+            ToPdu3 (Whoopediedoo c) ->+              modifyModel @SupiCounter (_3 .~ if c then Just orig else Nothing)++        Cast castReq ->+          case castReq of+            ToPdu1 Inc -> do+              val' <- view _1 <$> modifyAndGetModel @SupiCounter (_1 %~ (+ 1))+              zoomModel @SupiCounter _2 (observed (CounterChanged val'))+              when (val' > 5) $+                getAndModifyModel @SupiCounter (_3 .~ Nothing)+                >>= traverse_ (flip sendReply (Just ())) . view _3+            ToPdu2 x ->+              zoomModel @SupiCounter _2 (handleObserverRegistration x)+            ToPdu3 _ -> error "unreachable"++    other -> logWarning (T.pack (show other))++spawnCounter :: (LogsTo IO q, Lifted IO q) => Eff (InterruptableProcess q) ( Endpoint SupiCounter )+spawnCounter = start MkEmptySupiCounter+++deriving instance Show (Pdu Counter x)+ logCounterObservations-  :: (Member Logs q)+  :: (LogsTo IO q, Lifted IO q, Typeable q)   => Eff (InterruptableProcess q) (Observer CounterChanged) logCounterObservations = do-  svr <- spawnApiServer-    (handleObservations-      (\msg -> do-        logInfo' ("observed: " ++ show msg)-        return AwaitNext-      )-    )-    stopServerOnInterrupt+  svr <- start+          $ genServer @(Observer CounterChanged)+            (\_me -> pure (emptyObservers, ()))+            (\_me ->+                \case+                  OnRequest (Cast r) ->+                    handleObservations (\msg -> logInfo' ("observed: " ++ show msg)) r+                  wtf -> logNotice (T.pack (show wtf))+            )+            "counter logger"+   pure (toObserver svr)++type instance GenServerModel (Observer CounterChanged) = Observers CounterChanged+type instance GenServerSettings (Observer CounterChanged) = ()+type instance GenServerProtocol (Observer CounterChanged) = Observer CounterChanged

@@ -1,6 +1,6 @@ cabal-version:  2.0 name:           extensible-effects-concurrent-version:        0.23.0+version:        0.24.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@@ -81,26 +81,26 @@                   Control.Eff.LogWriter.UDP,                   Control.Eff.LogWriter.UnixSocket,                   Control.Eff.Concurrent,-                  Control.Eff.Concurrent.Api,-                  Control.Eff.Concurrent.Api.Client,-                  Control.Eff.Concurrent.Api.GenServer,-                  Control.Eff.Concurrent.Api.Server,-                  Control.Eff.Concurrent.Api.Supervisor,+                  Control.Eff.Concurrent.Protocol,+                  Control.Eff.Concurrent.Protocol.Client,+                  Control.Eff.Concurrent.Protocol.Server,+                  Control.Eff.Concurrent.Protocol.Request,+                  Control.Eff.Concurrent.Protocol.Supervisor,                   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.Api.Observer-                  Control.Eff.Concurrent.Api.Observer.Queue+                  Control.Eff.Concurrent.Protocol.Observer+                  Control.Eff.Concurrent.Protocol.Observer.Queue   other-modules:-                Control.Eff.Concurrent.Api.Request,-                Control.Eff.Concurrent.Api.Supervisor.InternalState,+                Control.Eff.Concurrent.Protocol.Supervisor.InternalState,                 Paths_extensible_effects_concurrent   default-extensions:                      AllowAmbiguousTypes,                      BangPatterns,                      ConstraintKinds,+                     DefaultSignatures,                      DeriveFoldable,                      DeriveFunctor,                      DeriveFunctor,@@ -124,7 +124,8 @@                      TypeInType,                      TypeOperators,                      ViewPatterns-  other-extensions:  ImplicitParams+  other-extensions:  ImplicitParams,+                     UndecidableInstances   default-language: Haskell2010   ghc-options: -Wall -fno-full-laziness @@ -265,13 +266,13 @@                 Common               , Debug               , ForkIOScheduler+              , GenServerTests               , Interactive               , LoggingTests               , LogMessageIdeaTest               , LoopTests               , ProcessBehaviourTestCases               , SupervisorTests-              , ServerForkIO               , SingleThreadedScheduler   ghc-options: -Wall -threaded -fno-full-laziness   build-depends:

@@ -5,33 +5,27 @@     -- * Concurrent Processes with Message Passing Concurrency     module Control.Eff.Concurrent.Process   ,+    -- * /Scheduler/ Process Effect Handler+    -- ** Concurrent Scheduler+    module Control.Eff.Concurrent.Process.ForkIOScheduler+  ,+    -- ** Single Threaded Scheduler+    module Control.Eff.Concurrent.Process.SingleThreadedScheduler+  ,     -- * Timers and Timeouts     module Control.Eff.Concurrent.Process.Timer   ,     -- * Data Types and Functions for APIs (aka Protocols)-    module Control.Eff.Concurrent.Api+    module Control.Eff.Concurrent.Protocol   ,     -- ** /Client/ Functions for Consuming APIs-    module Control.Eff.Concurrent.Api.Client-  ,-    -- ** /Server/ Functions for Providing APIs-    module Control.Eff.Concurrent.Api.Server-  ,-    -- ** Encapsulate 'Api's 'Cast's as well as 'Call's and their 'Reply's-    module Control.Eff.Concurrent.Api.Request+    module Control.Eff.Concurrent.Protocol.Client   ,     -- ** /Observer/ Functions for Events and Event Listener-    module Control.Eff.Concurrent.Api.Observer+    module Control.Eff.Concurrent.Protocol.Observer   ,     -- *** Capture /Observation/ in a FIFO Queue-    module Control.Eff.Concurrent.Api.Observer.Queue-  ,-    -- * /Scheduler/ Process Effect Handler-    -- ** Concurrent Scheduler-    module Control.Eff.Concurrent.Process.ForkIOScheduler-  ,-    -- ** Single Threaded Scheduler-    module Control.Eff.Concurrent.Process.SingleThreadedScheduler+    module Control.Eff.Concurrent.Protocol.Observer.Queue   ,     -- * Utilities     -- ** Logging Effect@@ -163,35 +157,32 @@                                                 , receiveSelectedWithMonitorAfter                                                 ) -import           Control.Eff.Concurrent.Api     ( Api+import           Control.Eff.Concurrent.Protocol+                                                ( Pdu(..)                                                 , Synchronicity(..)-                                                , Server(..)+                                                , ProtocolReply                                                 , Tangible-                                                , fromServer-                                                , proxyAsServer-                                                , asServer+                                                , TangiblePdu+                                                , Endpoint(..)+                                                , fromEndpoint+                                                , proxyAsEndpoint+                                                , asEndpoint+                                                , EmbedProtocol(..)                                                 )-import           Control.Eff.Concurrent.Api.Client+import           Control.Eff.Concurrent.Protocol.Client                                                 ( cast                                                 , call                                                 , callWithTimeout-                                                , castRegistered-                                                , callRegistered-                                                , ServesApi-                                                , registerServer-                                                , whereIsServer-                                                , ServerReader-                                                )-import           Control.Eff.Concurrent.Api.Request-                                                ( Request(..)-                                                , Reply(..)-                                                , mkRequestOrigin-                                                , RequestOrigin(..)-                                                , sendReply+                                                , castEndpointReader+                                                , callEndpointReader+                                                , ServesProtocol+                                                , runEndpointReader+                                                , askEndpoint+                                                , EndpointReader                                                 )-import           Control.Eff.Concurrent.Api.Observer+import           Control.Eff.Concurrent.Protocol.Observer                                                 ( Observer(..)-                                                , Api+                                                , Pdu                                                   ( RegisterObserver                                                   , ForgetObserver                                                   , Observed@@ -203,11 +194,13 @@                                                 , toObserverFor                                                 , ObserverRegistry                                                 , ObserverState+                                                , Observers()+                                                , emptyObservers                                                 , handleObserverRegistration                                                 , manageObservers                                                 , observed                                                 )-import           Control.Eff.Concurrent.Api.Observer.Queue+import           Control.Eff.Concurrent.Protocol.Observer.Queue                                                 ( ObservationQueue()                                                 , ObservationQueueReader                                                 , readObservationQueue@@ -215,47 +208,6 @@                                                 , flushObservationQueue                                                 , withObservationQueue                                                 , spawnLinkObservationQueueWriter-                                                )-import           Control.Eff.Concurrent.Api.Server-                                                ( spawnApiServer-                                                , spawnLinkApiServer-                                                , spawnApiServerStateful-                                                , spawnApiServerEffectful-                                                , spawnLinkApiServerEffectful-                                                , CallbackResult(..)-                                                , MessageCallback(..)-                                                , handleCasts-                                                , handleCalls-                                                , handleCastsAndCalls-                                                , handleCallsDeferred-                                                , handleMessages-                                                , handleSelectedMessages-                                                , handleAnyMessages-                                                , handleProcessDowns-                                                , dropUnhandledMessages-                                                , exitOnUnhandled-                                                , logUnhandledMessages-                                                , (^:)-                                                , fallbackHandler-                                                , ToServerPids(..)-                                                , InterruptCallback(..)-                                                , stopServerOnInterrupt-                                                )-import           Control.Eff.Concurrent.Api.Supervisor-                                                ( Sup()-                                                , SpawnFun-                                                , SupConfig(MkSupConfig)-                                                , supConfigChildStopTimeout-                                                , supConfigSpawnFun-                                                , SpawnErr(AlreadyStarted)-                                                , startSupervisor-                                                , stopSupervisor-                                                , isSupervisorAlive-                                                , monitorSupervisor-                                                , getDiagnosticInfo-                                                , spawnChild-                                                , lookupChild-                                                , stopChild                                                 ) import           Control.Eff.Concurrent.Process.ForkIOScheduler                                                 ( schedule

@@ -1,109 +0,0 @@-{-# LANGUAGE UndecidableInstances #-}--- | This module contains a mechanism to specify what kind of messages (aka--- /requests/) a 'Server' ('Process') can handle, and if the caller blocks and--- waits for an answer, which the server process provides.------ The type magic in the 'Api' type family allows to define a related set of /requests/ along--- with the corresponding responses.------ Request handling can be either blocking, if a response is required, or--- non-blocking.------ A process can /serve/ a specific 'Api' instance by using the functions provided by--- the "Control.Eff.Concurrent.Api.Server" module.------ To enable a process to use such a /service/, the functions provided by--- the "Control.Eff.Concurrent.Api.Client" should be used.----module Control.Eff.Concurrent.Api-  ( Api-  , Synchronicity(..)-  , Tangible-  , Server(..)-  , fromServer-  , proxyAsServer-  , asServer-  )-where--import           Control.DeepSeq                ( NFData )-import           Control.Eff.Concurrent.Process-import           Control.Lens-import           Data.Kind-import           Data.Typeable                  ( Typeable )-import           Data.Type.Pretty---- | This data family defines an API, a communication interface description--- between at least two processes. The processes act as __servers__ or--- __client(s)__ regarding a specific instance of this type.------ The first parameter is usually a user defined phantom type that identifies--- the 'Api' instance.------ The second parameter specifies if a specific constructor of an (GADT-like)--- @Api@ instance is 'Synchronous', i.e. returns a result and blocks the caller--- or if it is 'Asynchronous'------ Also, for better logging, the an instance of 'ToPretty' for the 'Api' index--- type must be given.------ Example:------ >--- > data BookShop deriving Typeable--- >--- > data instance Api BookShop r where--- >   RentBook  :: BookId   -> Api BookShop ('Synchronous (Either RentalError RentalId))--- >   BringBack :: RentalId -> Api BookShop 'Asynchronous--- >--- > type instance ToPretty BookShop = PutStr "book shop"--- >--- > type BookId = Int--- > type RentalId = Int--- > type RentalError = String--- >-data family Api (api :: Type) (reply :: Synchronicity)--type instance ToPretty (Api x y) =-  PrettySurrounded (PutStr "<") (PutStr ">") ("API" <:> ToPretty x <+> ToPretty y)---- | A set of constraints for types that can evaluated via 'NFData', compared via 'Ord' and presented--- dynamically via 'Typeable', and represented both as values--- via 'Show', as well as on the type level via 'ToPretty'.-type Tangible i =-  ( NFData i-  , Typeable i-  , Show i-  , PrettyTypeShow (ToPretty i)-  )---- | The (promoted) constructors of this type specify (at the type level) the--- reply behavior of a specific constructor of an @Api@ instance.-data Synchronicity =-  Synchronous Type -- ^ Specify that handling a request is a blocking operation-                   -- with a specific return type, e.g. @('Synchronous (Either-                   -- RentalError RentalId))@-  | Asynchronous -- ^ Non-blocking, asynchronous, request handling-    deriving (Typeable)---- | This is a tag-type that wraps around a 'ProcessId' and holds an 'Api' index--- type.-newtype Server api = Server { _fromServer :: ProcessId }-  deriving (Eq,Ord,Typeable, NFData)--instance (PrettyTypeShow (ToPretty api)) => Show (Server api) where-  showsPrec _ s@(Server c) = showString (showPretty s) . showsPrec 10 c--type instance ToPretty (Server a) = ToPretty a <+> PutStr "server"--makeLenses ''Server---- | Tag a 'ProcessId' with an 'Api' type index to mark it a 'Server' process--- handling that API-proxyAsServer :: proxy api -> ProcessId -> Server api-proxyAsServer = const Server---- | Tag a 'ProcessId' with an 'Api' type index to mark it a 'Server' process--- handling that API-asServer :: forall api . ProcessId -> Server api-asServer = Server

@@ -1,198 +0,0 @@--- | Functions for 'Api' clients.------ This modules is required to write clients that consume an 'Api'.-module Control.Eff.Concurrent.Api.Client-  ( -- * Calling APIs directly-    cast-  , call-  , callWithTimeout-  -- * Server Process Registration-  , castRegistered-  , callRegistered-  , ServesApi-  , ServerReader-  , whereIsServer-  , registerServer-  )-where--import           Control.Eff-import           Control.Eff.Reader.Strict-import           Control.Eff.Concurrent.Api-import           Control.Eff.Concurrent.Api.Request-import           Control.Eff.Concurrent.Process-import           Control.Eff.Concurrent.Process.Timer-import           Control.Eff.Log-import           Data.Typeable                  ( Typeable )-import           Data.Type.Pretty-import           Control.DeepSeq-import           GHC.Stack---- | Send an 'Api' request that has no return value and return as fast as--- possible. The type signature enforces that the corresponding 'Api' clause is--- 'Asynchronous'. The operation never fails, if it is important to know if the--- message was delivered, use 'call' instead.------ The message will be reduced to normal form ('rnf') in the caller process.-cast-  :: forall r q o-   . ( HasCallStack-     , SetMember Process (Process q) r-     , PrettyTypeShow (ToPretty o)-     , Member Interrupts r-     , Typeable o-     , Typeable (Api o 'Asynchronous)-     , NFData (Api o 'Asynchronous)-     )-  => Server o-  -> Api o 'Asynchronous-  -> Eff r ()-cast (Server pid) castMsg = sendMessage pid (Cast castMsg)---- | Send an 'Api' request and wait for the server to return a result value.------ The type signature enforces that the corresponding 'Api' clause is--- 'Synchronous'.------ __Always prefer 'callWithTimeout' over 'call'__-call-  :: forall result api r q-   . ( SetMember Process (Process q) r-     , Member Interrupts r-     , Typeable api-     , PrettyTypeShow (ToPretty api)-     , Typeable (Api api ( 'Synchronous result))-     , NFData (Api api ( 'Synchronous result))-     , Typeable result-     , NFData result-     , Show result-     , HasCallStack-     )-  => Server api-  -> Api api ( 'Synchronous result)-  -> Eff r result-call (Server pidInternal) req = do-  callRef <- makeReference-  fromPid <- self-  let requestMessage = Call callRef fromPid $! req-  sendMessage pidInternal requestMessage-  let selectResult :: MessageSelector result-      selectResult =-        let extractResult-              :: Reply (Api api ( 'Synchronous result)) -> Maybe result-            extractResult (Reply _pxResult callRefMsg result) =-              if callRefMsg == callRef then Just result else Nothing-        in  selectMessageWith extractResult-  resultOrError <- receiveWithMonitor pidInternal selectResult-  either (interrupt . becauseProcessIsDown) return resultOrError----- | Send an 'Api' request and wait for the server to return a result value.------ The type signature enforces that the corresponding 'Api' clause is--- 'Synchronous'.------ If the server that was called dies, this function interrupts the--- process with 'ProcessDown'.--- If the server takes longer to reply than the given timeout, this--- function interrupts the process with 'TimeoutInterrupt'.------ __Always prefer this function over 'call'__------ @since 0.22.0-callWithTimeout-  :: forall result api r q-   . ( SetMember Process (Process q) r-     , Member Interrupts r-     , Typeable api-     , Typeable (Api api ( 'Synchronous result))-     , NFData (Api api ( 'Synchronous result))-     , Typeable result-     , NFData result-     , Show result-     , Member Logs r-     , Lifted IO q-     , Lifted IO r-     , HasCallStack-     , PrettyTypeShow (ToPretty api)-     )-  => Server api-  -> Api api ( 'Synchronous result)-  -> Timeout-  -> Eff r result-callWithTimeout serverP@(Server pidInternal) req timeOut = do-  fromPid <- self-  callRef <- makeReference-  let requestMessage = Call callRef fromPid $! req-  sendMessage pidInternal requestMessage-  let selectResult =-        let extractResult-              :: Reply (Api api ( 'Synchronous result)) -> Maybe result-            extractResult (Reply _pxResult callRefMsg result) =-              if callRefMsg == callRef then Just result else Nothing-        in selectMessageWith extractResult-  resultOrError <- receiveSelectedWithMonitorAfter pidInternal selectResult timeOut-  let onTimeout timerRef = do-        let msg = "call timed out after "-                  ++ show timeOut ++ " to server: "-                  ++ show serverP ++ " from "-                  ++ show fromPid ++ " "-                  ++ show timerRef-        logWarning' msg-        interrupt (TimeoutInterrupt msg)-      onProcDown p = do-        logWarning' ("call to dead server: "++ show serverP ++ " from " ++ show fromPid)-        interrupt (becauseProcessIsDown p)-  either (either onProcDown onTimeout) return resultOrError---- | Instead of passing around a 'Server' value and passing to functions like--- 'cast' or 'call', a 'Server' can provided by a 'Reader' effect, if there is--- only a __single server__ for a given 'Api' instance. This type alias is--- convenience to express that an effect has 'Process' and a reader for a--- 'Server'.-type ServesApi o r q =-  ( Typeable o-  , PrettyTypeShow (ToPretty o)-  , SetMember Process (Process q) r-  , Member (ServerReader o) r-  )---- | The reader effect for 'ProcessId's for 'Api's, see 'registerServer'-type ServerReader o = Reader (Server o)---- | Run a reader effect that contains __the one__ server handling a specific--- 'Api' instance.-registerServer-  :: HasCallStack => Server o -> Eff (ServerReader o ': r) a -> Eff r a-registerServer = runReader---- | Get the 'Server' registered with 'registerServer'.-whereIsServer :: Member (ServerReader o) e => Eff e (Server o)-whereIsServer = ask---- | Like 'call' but take the 'Server' from the reader provided by--- 'registerServer'.-callRegistered-  :: ( Typeable reply-     , ServesApi o r q-     , HasCallStack-     , NFData reply-     , Show reply-     , NFData (Api o ( 'Synchronous reply))-     , Member Interrupts r-     )-  => Api o ( 'Synchronous reply)-  -> Eff r reply-callRegistered method = do-  serverPid <- whereIsServer-  call serverPid method---- | Like 'cast' but take the 'Server' from the reader provided by--- 'registerServer'.-castRegistered-  :: (Typeable o, ServesApi o r q, HasCallStack, Member Interrupts r, NFData (Api o 'Asynchronous))-  => Api o 'Asynchronous-  -> Eff r ()-castRegistered method = do-  serverPid <- whereIsServer-  cast serverPid method

@@ -1,49 +0,0 @@--- | A better, more safe implementation of the Erlang/OTP gen_server behaviour.--- **PLANNED TODO**--- @since 0.24.0-module Control.Eff.Concurrent.Api.GenServer--   where--import Control.DeepSeq-import Control.Eff-import Control.Eff.Concurrent.Api-import Control.Eff.Concurrent.Process-import Control.Eff.Log-import Control.Eff.State.Lazy-import Data.Dynamic-import Data.Kind-import Data.Text as T---- | A type class for 'Api' values that have an implementation--- which handles the 'Api'.-class (Typeable (GenServerState a), NFData (GenServerState a)) =>-      GenServer a eff-  where-  type GenServerState a :: Type-  genServerInit :: ('[Interrupts, Logs] <:: eff, SetMember Process (Process q) eff) => a -> Eff eff (GenServerState a)-  genServerHandle ::-       ('[Interrupts, Logs] <:: eff, SetMember Process (Process q) eff)-    => Api a v-    -> Eff (State (GenServerState a) ': eff) (ApiReply v)-  genServerInterrupt ::-       ('[Interrupts, Logs] <:: eff, SetMember Process (Process q) eff)-    => Interrupt 'Recoverable-    -> Eff (State (GenServerState a) ': eff) ()-  genServerInfoCommand :: Api a ('Synchronous Text)--type family ApiReply (s :: Synchronicity) where-  ApiReply ('Synchronous t) = Maybe t-  ApiReply 'Asynchronous = ()--data SomeMessage a =-  MkSomeMessage--data instance  Api (SomeMessage a) s where-        SomeMessage :: a -> Api (SomeMessage a) 'Asynchronous--runGenServer ::-     forall q e h a. (GenServer a e, SetMember Process (Process q) e, Member Interrupts e, LogsTo h e)-  => a-  -> Eff e (Server a)-runGenServer initArg = error "TODO"

@@ -1,267 +0,0 @@--- | Observer Effects------ This module supports the implementation of observers and observables. Expected use--- case is event propagation.------ @since 0.16.0-module Control.Eff.Concurrent.Api.Observer-  ( Observer(..)-  , Api(RegisterObserver, ForgetObserver, Observed)-  , registerObserver-  , forgetObserver-  , handleObservations-  , toObserver-  , toObserverFor-  , ObserverRegistry-  , ObserverState-  , handleObserverRegistration-  , manageObservers-  , observed-  )-where--import           Control.DeepSeq               (NFData(rnf))-import           Control.Eff-import           Control.Eff.Concurrent.Process-import           Control.Eff.Concurrent.Api-import           Control.Eff.Concurrent.Api.Client-import           Control.Eff.Concurrent.Api.Server-import           Control.Eff.State.Strict-import           Control.Eff.Log-import           Control.Lens-import           Data.Data                     (typeOf)-import           Data.Dynamic-import           Data.Foldable-import           Data.Proxy-import           Data.Set                       ( Set )-import qualified Data.Set                      as Set-import           Data.Text                      ( pack )-import           Data.Typeable                  ( typeRep )-import           Data.Type.Pretty-import           GHC.Stack---- * Observers---- | Describes a process that observes another via 'Asynchronous' 'Api' messages.------ An observer consists of a filter and a process id. The filter converts an observation to--- a message understood by the observer process, and the 'ProcessId' is used to send the message.------ @since 0.16.0-data Observer o where-  Observer-    :: (PrettyTypeShow (ToPretty p), Show (Server p), Typeable p, Typeable o, NFData o, NFData (Api p 'Asynchronous))-    => (o -> Maybe (Api p 'Asynchronous)) -> Server p -> Observer o--type instance ToPretty (Observer o) =-  PrettyParens ("observing" <:> ToPretty o)--instance (NFData o) => NFData (Observer o) where-  rnf (Observer k s) = rnf k `seq` rnf s--instance Show (Observer o) where-  showsPrec d (Observer _ p) = showParen-    (d >= 10)-    (shows (typeRep (Proxy :: Proxy o)) . showString " observer: " . shows p)--instance Ord (Observer o) where-  compare (Observer _ s1) (Observer _ s2) =-    compare (s1 ^. fromServer) (s2 ^. fromServer)--instance Eq (Observer o) where-  (==) (Observer _ s1) (Observer _ s2) =-    (==) (s1 ^. fromServer) (s2 ^. fromServer)---- | And an 'Observer' to the set of recipients for all observations reported by 'observed'.---   Note that the observers are keyed by the observing process, i.e. a previous entry for the process---   contained in the 'Observer' is overwritten. If you want multiple entries for a single process, just---   combine several filter functions.------ @since 0.16.0-registerObserver-  :: ( SetMember Process (Process q) r-     , HasCallStack-     , Member Interrupts r-     , Typeable o-     , NFData o-     , PrettyTypeShow (ToPretty o)-     )-  => Observer o-  -> Server (ObserverRegistry o)-  -> Eff r ()-registerObserver observer observerRegistry =-  cast observerRegistry (RegisterObserver observer)---- | Send the 'ForgetObserver' message------ @since 0.16.0-forgetObserver-  :: ( SetMember Process (Process q) r-     , HasCallStack-     , Member Interrupts r-     , Typeable o-     , NFData o-     , PrettyTypeShow (ToPretty o)-     )-  => Observer o-  -> Server (ObserverRegistry o)-  -> Eff r ()-forgetObserver observer observerRegistry =-  cast observerRegistry (ForgetObserver observer)---- ** Observer Support Functions---- | A minimal Api for handling observations.--- This is one simple way of receiving observations - of course users can use--- any other 'Asynchronous' 'Api' message type for receiving observations.------ @since 0.16.0-data instance Api (Observer o) r where-  -- | This message denotes that the given value was 'observed'.-  ---  -- @since 0.16.1-  Observed :: o -> Api (Observer o) 'Asynchronous-  deriving Typeable--instance NFData o => NFData (Api (Observer o) 'Asynchronous) where-  rnf (Observed o) = rnf o---- | Based on the 'Api' instance for 'Observer' this simplified writing--- a callback handler for observations. In order to register to--- and 'ObserverRegistry' use 'toObserver'.------ @since 0.16.0-handleObservations-  :: (HasCallStack, Typeable o, SetMember Process (Process q) r, NFData (Observer o))-  => (o -> Eff r (CallbackResult 'Recoverable))-  -> MessageCallback (Observer o) r-handleObservations k = handleCasts-  (\case-    Observed o -> k o-  )---- | Use a 'Server' as an 'Observer' for 'handleObservations'.------ @since 0.16.0-toObserver-  :: (NFData o, Typeable o, NFData (Api (Observer o) 'Asynchronous), PrettyTypeShow (ToPretty o))-  => Server (Observer o) -> Observer o-toObserver = toObserverFor Observed---- | Create an 'Observer' that conditionally accepts all observations of the--- given type and applies the given function to them; the function takes an observation and returns an 'Api'--- cast that the observer server is compatible to.------ @since 0.16.0-toObserverFor-  :: (Typeable a, PrettyTypeShow (ToPretty a), NFData (Api a 'Asynchronous), Typeable o, NFData o)-  => (o -> Api a 'Asynchronous)-  -> Server a-  -> Observer o-toObserverFor wrapper = Observer (Just . wrapper)---- * Managing Observers---- | An 'Api' for managing 'Observer's, encompassing  registration and de-registration of--- 'Observer's.------ @since 0.16.0-data ObserverRegistry o--type instance ToPretty (ObserverRegistry o) =-  PrettyParens ("observer registry" <:> ToPretty o)---- | Api for managing observers. This can be added to any server for any number of different observation types.--- The functions 'manageObservers' and 'handleObserverRegistration' are used to include observer handling;------ @since 0.16.0-data instance Api (ObserverRegistry o) r where-  -- | This message denotes that the given 'Observer' should receive observations until 'ForgetObserver' is-  --   received.-  ---  -- @since 0.16.1-  RegisterObserver :: NFData o => Observer o -> Api (ObserverRegistry o) 'Asynchronous-  -- | This message denotes that the given 'Observer' should not receive observations anymore.-  ---  -- @since 0.16.1-  ForgetObserver :: NFData o => Observer o -> Api (ObserverRegistry o) 'Asynchronous-  deriving Typeable--instance NFData (Api (ObserverRegistry o) r) where-  rnf (RegisterObserver o) = rnf o-  rnf (ForgetObserver o) = rnf o---- ** Api for integrating 'ObserverRegistry' into processes.---- | Provide the implementation for the 'ObserverRegistry' Api, this handled 'RegisterObserver' and 'ForgetObserver'--- messages. It also adds the 'ObserverState' constraint to the effect list.------ @since 0.16.0-handleObserverRegistration-  :: forall o q r-   . ( HasCallStack-     , Typeable o-     , SetMember Process (Process q) r-     , Member (ObserverState o) r-     , Member Logs r-     )-  => MessageCallback (ObserverRegistry o) r-handleObserverRegistration = handleCasts-  (\case-    RegisterObserver ob -> do-      os <- get @(Observers o)-      logDebug ("registering "-               <> pack (show (typeOf ob))-               <> " current number of observers: "-               <> pack (show (Set.size (view observers os))))-      put (over observers (Set.insert ob)os)-      pure AwaitNext-    ForgetObserver ob -> do-      os <- get @(Observers o)-      logDebug ("forgetting "-               <> pack (show (typeOf ob))-               <> " current number of observers: "-               <> pack (show (Set.size (view observers os))))-      put (over observers (Set.delete ob) os)-      pure AwaitNext-  )----- | Keep track of registered 'Observer's.------ Handle the 'ObserverState' introduced by 'handleObserverRegistration'.------ @since 0.16.0-manageObservers :: Eff (ObserverState o ': r) a -> Eff r a-manageObservers = evalState (Observers Set.empty)---- | Internal state for 'manageObservers'-newtype Observers o =-  Observers { _observers :: Set (Observer o) }---- | Alias for the effect that contains the observers managed by 'manageObservers'-type ObserverState o = State (Observers o)--observers :: Iso' (Observers o) (Set (Observer o))-observers = iso _observers Observers---- | Report an observation to all observers.--- The process needs to 'manageObservers' and to 'handleObserverRegistration'.------ @since 0.16.0-observed-  :: forall o r q-   . ( SetMember Process (Process q) r-     , PrettyTypeShow (ToPretty o)-     , Member (ObserverState o) r-     , Member Interrupts r-   --  , NFData (Api o 'Asynchronous)-     )-  => o-  -> Eff r ()-observed observation = do-  os <- view observers <$> get-  mapM_ notifySomeObserver os- where-  notifySomeObserver (Observer messageFilter receiver) =-    traverse_ (cast receiver) (messageFilter observation)

@@ -1,156 +0,0 @@--- | A small process to capture and _share_ observation's by enqueueing them into an STM 'TBQueue'.-module Control.Eff.Concurrent.Api.Observer.Queue-  ( ObservationQueue()-  , ObservationQueueReader-  , readObservationQueue-  , tryReadObservationQueue-  , flushObservationQueue-  , withObservationQueue-  , spawnLinkObservationQueueWriter-  )-where--import           Control.Concurrent.STM-import           Control.DeepSeq (NFData)-import           Control.Eff-import           Control.Eff.Concurrent.Api-import           Control.Eff.Concurrent.Api.Observer-import           Control.Eff.Concurrent.Api.Server-import           Control.Eff.Concurrent.Process-import           Control.Eff.ExceptionExtra     ( )-import           Control.Eff.Log-import           Control.Eff.Reader.Strict-import           Control.Exception.Safe        as Safe-import           Control.Monad.IO.Class-import           Control.Monad                  ( unless )-import qualified Data.Text                     as T-import           Data.Typeable-import           GHC.Stack---- | Contains a 'TBQueue' capturing observations.--- See 'spawnLinkObservationQueueWriter', 'readObservationQueue'.-newtype ObservationQueue a = ObservationQueue (TBQueue a)---- | A 'Reader' for an 'ObservationQueue'.-type ObservationQueueReader a = Reader (ObservationQueue a)--logPrefix :: forall o proxy . (HasCallStack, Typeable o) => proxy o -> T.Text-logPrefix px = "observation queue: " <> T.pack (show (typeRep px))---- | Read queued observations captured and enqueued in the shared 'TBQueue' by 'spawnLinkObservationQueueWriter'.--- This blocks until something was captured or an interrupt or exceptions was thrown. For a non-blocking--- variant use 'tryReadObservationQueue' or 'flushObservationQueue'.-readObservationQueue-  :: forall o r-   . ( Member (ObservationQueueReader o) r-     , HasCallStack-     , MonadIO (Eff r)-     , Typeable o-     , Member Logs r-     )-  => Eff r o-readObservationQueue = do-  ObservationQueue q <- ask @(ObservationQueue o)-  liftIO (atomically (readTBQueue q))---- | Read queued observations captured and enqueued in the shared 'TBQueue' by 'spawnLinkObservationQueueWriter'.--- Return the oldest enqueued observation immediately or 'Nothing' if the queue is empty.--- Use 'readObservationQueue' to block until an observation is observed.-tryReadObservationQueue-  :: forall o r-   . ( Member (ObservationQueueReader o) r-     , HasCallStack-     , MonadIO (Eff r)-     , Typeable o-     , Member Logs r-     )-  => Eff r (Maybe o)-tryReadObservationQueue = do-  ObservationQueue q <- ask @(ObservationQueue o)-  liftIO (atomically (tryReadTBQueue q))---- | Read at once all currently queued observations captured and enqueued--- in the shared 'TBQueue' by 'spawnLinkObservationQueueWriter'.--- This returns immediately all currently enqueued observations.--- For a blocking variant use 'readObservationQueue'.-flushObservationQueue-  :: forall o r-   . ( Member (ObservationQueueReader o) r-     , HasCallStack-     , MonadIO (Eff r)-     , Typeable o-     , Member Logs r-     )-  => Eff r [o]-flushObservationQueue = do-  ObservationQueue q <- ask @(ObservationQueue o)-  liftIO (atomically (flushTBQueue q))---- | Create a mutable queue for observations. Use 'spawnLinkObservationQueueWriter' for a simple way to get--- a process that enqueues all observations.------ ==== __Example__------ @--- withObservationQueue 100 $ do---   q  <- ask \@(ObservationQueueReader TestEvent)---   wq <- spawnLinkObservationQueueWriter q---   registerObserver wq testServer---   ...---   cast testServer DoSomething---   evt <- readObservationQueue \@TestEvent---   ...--- @------ @since 0.18.0-withObservationQueue-  :: forall o b e len-   . ( HasCallStack-     , Typeable o-     , Show o-     , Member Logs e-     , Lifted IO e-     , Integral len-     , Member Interrupts e-     )-  => len-  -> Eff (ObservationQueueReader o ': e) b-  -> Eff e b-withObservationQueue queueLimit e = do-  q   <- lift (newTBQueueIO (fromIntegral queueLimit))-  res <- handleInterrupts (return . Left)-                          (Right <$> runReader (ObservationQueue q) e)-  rest <- lift (atomically (flushTBQueue q))-  unless-    (null rest)-    (logNotice (logPrefix (Proxy @o) <> " unread observations: " <> T.pack (show rest)))-  either (\em -> logError (T.pack (show em)) >> lift (throwIO em)) return res---- | Spawn a process that can be used as an 'Observer' that enqueues the observations into an---   'ObservationQueue'. See 'withObservationQueue' for an example.------ The observations can be obtained by 'readObservationQueue'. All observations are captured up to--- the queue size limit, such that the first message received will be first message--- returned by 'readObservationQueue'.------ @since 0.18.0-spawnLinkObservationQueueWriter-  :: forall o q-   . ( Tangible o-     , NFData (Api (Observer o) 'Asynchronous)-     , Member Logs q-     , Lifted IO q-     , HasCallStack)-  => ObservationQueue o-  -> Eff (InterruptableProcess q) (Observer o)-spawnLinkObservationQueueWriter (ObservationQueue q) = do-  cbo <- spawnLinkApiServer-    (handleObservations-      (\case-        o -> do-          lift (atomically (writeTBQueue q o))-          pure AwaitNext-      )-    )-    stopServerOnInterrupt-  pure (toObserver cbo)

@@ -1,102 +0,0 @@-{-# LANGUAGE UndecidableInstances #-}--- | Proxies and containers for casts and calls.------ @since 0.15.0-module Control.Eff.Concurrent.Api.Request-  ( Request(..)-  , Reply(..)-  , mkRequestOrigin-  , RequestOrigin(..)-  , sendReply-  )-where--import           Data.Typeable                  ( Typeable )-import           Data.Proxy-import           Control.Eff-import           Control.Eff.Concurrent.Api-import           Control.Eff.Concurrent.Process-import           GHC.TypeLits-import           Control.DeepSeq-import           GHC.Generics---- | A wrapper sum type for calls and casts for the methods of an 'Api' subtype------ @since 0.15.0-data Request api where-  Call :: forall api reply . (Typeable api, Typeable reply, NFData reply, Typeable (Api api ('Synchronous reply)), NFData (Api api ('Synchronous reply)))-         => Int -> ProcessId -> Api api ('Synchronous reply) -> Request api--  Cast :: forall api . (Typeable api, Typeable (Api api 'Asynchronous), NFData (Api api 'Asynchronous ))-         => Api api 'Asynchronous -> Request api-  deriving Typeable--instance NFData (Request api) where-  rnf (Call i p req) = rnf i `seq` rnf p `seq` rnf req-  rnf (Cast req)     = rnf req---- | The wrapper around replies to 'Call's.------ @since 0.15.0-data Reply request where-  Reply :: (Typeable api, Typeable reply, NFData reply)-        => Proxy (Api api ('Synchronous reply)) -> Int -> reply -> Reply (Api api ('Synchronous reply))-  deriving Typeable--instance NFData (Reply request) where-  rnf (Reply _ i r) = rnf i `seq` rnf r----- | Get the @reply@ of an @Api foo ('Synchronous reply)@.------ @since 0.15.0-type family ReplyType request where-  ReplyType (Api api ('Synchronous reply)) = reply-  ReplyType (Api api 'Asynchronous) = TypeError ('Text "Asynchronous requests (aka casts) have no reply type." )---- | Get the @reply@ of an @Api foo ('Synchronous reply)@.------ @since 0.15.0-type family ApiType request where-  ApiType (Api api 'Asynchronous) = api-  ApiType (Api api ('Synchronous reply)) = api---- | TODO remove-mkRequestOrigin :: request -> ProcessId -> Int -> RequestOrigin request-mkRequestOrigin _ = RequestOrigin---- | Wraps the source 'ProcessId' and a unique identifier for a 'Call'.------ @since 0.15.0-data RequestOrigin request =-  RequestOrigin { _requestOriginPid :: !ProcessId, _requestOriginCallRef :: !Int}-  deriving (Eq, Ord, Typeable, Generic)--instance Show (RequestOrigin r) where-  showsPrec d (RequestOrigin o r) =-    showParen (d >= 10) (showString "caller: " . shows o . showChar ' ' . shows r)--instance NFData (RequestOrigin request) where---- | Send a 'Reply' to a 'Call'.------ The reply will be deeply evaluated to 'rnf'.------ @since 0.15.0-sendReply-  :: forall request reply api eff q-   . ( SetMember Process (Process q) eff-     , Member Interrupts eff-     , Typeable api-     , ApiType request ~ api-     , ReplyType request ~ reply-     , request ~ Api api ( 'Synchronous reply)-     , Typeable reply-     , NFData reply-     )-  => RequestOrigin request-  -> reply-  -> Eff eff ()-sendReply origin reply = sendMessage-  (_requestOriginPid origin)-  (Reply (Proxy @request) (_requestOriginCallRef origin) $! force reply)

@@ -1,517 +0,0 @@-{-# LANGUAGE UndecidableInstances #-}--- | Support code to implement 'Api' _server_ processes.------ @since 0.16.0-module Control.Eff.Concurrent.Api.Server-  ( -- * Starting Api Servers-    spawnApiServer-  , spawnLinkApiServer-  , spawnApiServerStateful-  , spawnApiServerEffectful-  , spawnLinkApiServerEffectful-  , apiServerLoop-  -- ** Api Server Callbacks-  , CallbackResult(..)-  , MessageCallback(..)-  -- ** Callback Smart Constructors-  -- *** Calls and Casts (for 'Api's)-  , handleCasts-  , handleCalls-  , handleCastsAndCalls-  , handleCallsDeferred-  -- *** Generic Message Handler-  , handleMessages-  , handleSelectedMessages-  , handleAnyMessages-  , handleProcessDowns-  -- *** Fallback Handler-  , dropUnhandledMessages-  , exitOnUnhandled-  , logUnhandledMessages-  -- ** Api Composition-  , (^:)-  , fallbackHandler-  , ToServerPids(..)-  -- ** RecoverableInterrupt handler-  , InterruptCallback(..)-  , stopServerOnInterrupt-  )-where--import           Control.Applicative-import           Control.Eff-import           Control.Eff.Extend-import           Control.Eff.Log-import           Control.Eff.State.Strict-import           Control.Eff.Concurrent.Api-import           Control.Eff.Concurrent.Api.Request-import           Control.Eff.Concurrent.Process-import           Control.DeepSeq-import           Control.Monad                  ( (>=>) )-import           Data.Default-import           Data.Dynamic-import           Data.Foldable-import           Data.Kind-import           Data.Proxy-import           Data.Text as T-import           Data.Type.Pretty-import           GHC.Stack----- | /Serve/ an 'Api' in a newly spawned process.------ @since 0.13.2-spawnApiServer-  :: forall api eff-   . ( ToServerPids api-     , HasCallStack-     , Member Logs eff-     , PrettyTypeShow (PrettyServerPids api)-     )-  => MessageCallback api (InterruptableProcess eff)-  -> InterruptCallback (ConsProcess eff)-  -> Eff (InterruptableProcess eff) (ServerPids api)-spawnApiServer scb (InterruptCallback icb) = toServerPids (Proxy @api)-  <$> spawn (apiServerLoop scb (InterruptCallback (raise . icb)))---- | /Serve/ an 'Api' in a newly spawned -and linked - process.------ @since 0.14.2-spawnLinkApiServer-  :: forall api eff-   . ( ToServerPids api-     , HasCallStack-     , Member Logs eff-     , Typeable api-     , PrettyTypeShow (PrettyServerPids api)-     )-  => MessageCallback api (InterruptableProcess eff)-  -> InterruptCallback (ConsProcess eff)-  -> Eff (InterruptableProcess eff) (ServerPids api)-spawnLinkApiServer scb (InterruptCallback icb) = toServerPids (Proxy @api)-  <$> spawnLink (apiServerLoop scb (InterruptCallback (raise . icb)))---- | /Server/ an 'Api' in a newly spawned process; the callbacks have access--- to a state value.------ The initial state value is returned by the action given as parameter, from--- within the new process.------ @since 0.13.2-spawnApiServerStateful-  :: forall api eff state-   . ( HasCallStack-     , ToServerPids api-     , NFData state-     , PrettyTypeShow (PrettyServerPids api)-     )-  => Eff (InterruptableProcess eff) state-  -> MessageCallback api (State state ': InterruptableProcess eff)-  -> InterruptCallback (State state ': ConsProcess eff)-  -> Eff (InterruptableProcess eff) (ServerPids api)-spawnApiServerStateful initEffect (MessageCallback sel cb) (InterruptCallback intCb)-  = fmap (toServerPids (Proxy @api)) $ spawnRaw $ do-    state <- provideInterruptsShutdown initEffect-    evalState state $ receiveSelectedLoop sel $ \msg -> case msg of-      Left  m -> invokeIntCb m-      Right m -> do-        r <- do s <- force <$> get-                raise (provideInterrupts (runState s (cb m)))-        case r of-          Left  i                  -> invokeIntCb i-          Right (x , s')           -> do-            put (force s')-            case x of-              (StopServer i) -> invokeIntCb i-              AwaitNext      -> return Nothing- where-  invokeIntCb j = do-    l <- intCb j-    case l of-      AwaitNext               -> return Nothing-      StopServer ExitNormally -> return (Just ())-      StopServer k            -> exitBecause k---- | /Server/ an 'Api' in a newly spawned process; The caller provides an--- effect handler for arbitrary effects used by the server callbacks.------ @since 0.13.2-spawnApiServerEffectful-  :: forall api eff serverEff-   . ( HasCallStack-     , ToServerPids api-     , Member Interrupts serverEff-     , SetMember Process (Process eff) serverEff-     , Member Logs serverEff-     , PrettyTypeShow (PrettyServerPids api)-     )-  => (forall b . Eff serverEff b -> Eff (InterruptableProcess eff) b)-  -> MessageCallback api serverEff-  -> InterruptCallback serverEff-  -> Eff (InterruptableProcess eff) (ServerPids api)-spawnApiServerEffectful handleServerInternalEffects scb icb =-  toServerPids (Proxy @api)-    <$> spawn (handleServerInternalEffects (apiServerLoop scb icb))----- | /Server/ an 'Api' in a newly spawned process; The caller provides an--- effect handler for arbitrary effects used by the server callbacks.--- Links to the calling process like 'linkProcess' would.------ @since 0.14.2-spawnLinkApiServerEffectful-  :: forall api eff serverEff-   . ( HasCallStack-     , ToServerPids api-     , Member Interrupts serverEff-     , SetMember Process (Process eff) serverEff-     , Member Logs serverEff-     , PrettyTypeShow (PrettyServerPids api)-     )-  => (forall b . Eff serverEff b -> Eff (InterruptableProcess eff) b)-  -> MessageCallback api serverEff-  -> InterruptCallback serverEff-  -> Eff (InterruptableProcess eff) (ServerPids api)-spawnLinkApiServerEffectful handleServerInternalEffects scb icb =-  toServerPids (Proxy @api)-    <$> spawnLink (handleServerInternalEffects (apiServerLoop scb icb))---- | Receive loop for 'Api' 'Control.Eff.Concurrent.Api.Client.call's. This starts a receive loop for--- a 'MessageCallback'. It is used behind the scenes by 'spawnLinkApiServerEffectful'--- and 'spawnApiServerEffectful'.------ @since 0.14.2-apiServerLoop-  :: forall api eff serverEff-   . ( HasCallStack-     , ToServerPids api-     , Member Interrupts serverEff-     , SetMember Process (Process eff) serverEff-     , Member Logs serverEff-     , PrettyTypeShow (PrettyServerPids api)-     )-  => MessageCallback api serverEff-  -> InterruptCallback serverEff-  -> Eff serverEff ()-apiServerLoop (MessageCallback sel cb) (InterruptCallback intCb) = do-  logInfo ( "enter server loop: "-          <> pack (showPretty (Proxy @(PrettySurrounded (PutStr "[") (PutStr "]")-                                (PrettyServerPids api))))-          )-  receiveSelectedLoop-    sel-    ( ( either-              (fmap Left  . intCb)-              (fmap Right . tryUninterrupted . cb))-      >=> handleCallbackResult-    )- where-  handleCallbackResult-    :: Either (CallbackResult 'NoRecovery) (Either (Interrupt 'Recoverable) (CallbackResult 'Recoverable))-    -> Eff serverEff (Maybe ())-  handleCallbackResult (Left AwaitNext) = return Nothing-  handleCallbackResult (Left (StopServer r)) = exitBecause r-  handleCallbackResult (Right (Right AwaitNext)) = return Nothing-  handleCallbackResult (Right (Right (StopServer r))) =-    intCb r >>= handleCallbackResult . Left-  handleCallbackResult (Right (Left r)) =-    intCb r >>= handleCallbackResult . Left---- | A command to the server loop started by 'apiServerLoop'.--- Typically returned by a 'MessageCallback' to indicate if the server--- should continue or stop.------ @since 0.13.2-data CallbackResult (r :: ExitRecovery) where-  -- | Tell the server to keep the server loop running-  AwaitNext :: CallbackResult r-  -- | Tell the server to exit, this will cause 'apiServerLoop' to stop handling requests without-  -- exiting the process.-  StopServer :: Interrupt r -> CallbackResult r-  --  SendReply :: reply -> CallbackResult () -> CallbackResult (reply -> Eff eff ())-  deriving ( Typeable )----- | An existential wrapper around  a 'MessageSelector' and a function that--- handles the selected message. The @api@ type parameter is a phantom type.------ The return value of the handler function is a 'CallbackResult'.------ @since 0.13.2-data MessageCallback api eff where-   MessageCallback :: MessageSelector a -> (a -> Eff eff (CallbackResult 'Recoverable)) -> MessageCallback api eff--instance Semigroup (MessageCallback api eff) where-  (MessageCallback selL runL) <> (MessageCallback selR runR) =-    MessageCallback (Left <$> selL <|> Right <$> selR) (either runL runR)--instance Monoid (MessageCallback api eff) where-  mappend = (<>)-  mempty  = MessageCallback selectAnyMessage (const (pure AwaitNext))--instance Default (MessageCallback api eff) where-  def = mempty---- | A smart constructor for 'MessageCallback's------ @since 0.13.2-handleMessages-  :: forall eff a-   . (HasCallStack, NFData a, Typeable a)-  => (a -> Eff eff (CallbackResult 'Recoverable))-  -> MessageCallback '[] eff-handleMessages = MessageCallback selectMessage---- | A smart constructor for 'MessageCallback's------ @since 0.13.2-handleSelectedMessages-  :: forall eff a-   . HasCallStack-  => MessageSelector a-  -> (a -> Eff eff (CallbackResult 'Recoverable))-  -> MessageCallback '[] eff-handleSelectedMessages = MessageCallback---- | A smart constructor for 'MessageCallback's------ @since 0.13.2-handleAnyMessages-  :: forall eff-   . HasCallStack-  => (StrictDynamic -> Eff eff (CallbackResult 'Recoverable))-  -> MessageCallback '[] eff-handleAnyMessages = MessageCallback selectAnyMessage---- | A smart constructor for 'MessageCallback's------ @since 0.13.2-handleCasts-  :: forall api eff-   . ( HasCallStack, Typeable api, Typeable (Api api 'Asynchronous)-     , NFData (Request api))-  => (Api api 'Asynchronous -> Eff eff (CallbackResult 'Recoverable))-  -> MessageCallback api eff-handleCasts h = MessageCallback-  (selectMessageWith-    (\case-      cr@(Cast _ :: Request api) -> Just cr-      _callReq                   -> Nothing-    )-  )-  (\(Cast req :: Request api) -> h req)---- | A smart constructor for 'MessageCallback's------ ==== __Example__------ @--- handleCalls---   (\ (RentBook bookId customerId) runCall ->---      runCall $ do---          rentalIdE <- rentBook bookId customerId---          case rentalIdE of---            -- on fail we just don't send a reply, let the caller run into---            -- timeout---            Left err -> return (Nothing, AwaitNext)---            Right rentalId -> return (Just rentalId, AwaitNext))--- @------ @since 0.13.2-handleCalls-  :: forall api eff effScheduler-   . ( HasCallStack-     , Typeable api-     , SetMember Process (Process effScheduler) eff-     , Member Interrupts eff-     )-  => (  forall secret reply-      . (NFData reply, Typeable reply, Typeable (Api api ( 'Synchronous reply)))-     => Api api ( 'Synchronous reply)-     -> (Eff eff (Maybe reply, CallbackResult 'Recoverable) -> secret)-     -> secret-     )-  -> MessageCallback api eff-handleCalls h = MessageCallback-  (selectMessageWith-    (\case-      (Cast _ :: Request api) -> Nothing-      callReq                 -> Just callReq-    )-  )-  (\(Call callRef fromPid req :: Request api) -> h-    req-    (\resAction -> do-      (mReply, cbResult) <- resAction-      traverse_ (sendReply (mkRequestOrigin req fromPid callRef)) mReply-      return cbResult-    )-  )----- | A smart constructor for 'MessageCallback's------ @since 0.13.2-handleCastsAndCalls-  :: forall api eff effScheduler-   . ( HasCallStack-     , Typeable api-     , Typeable (Api api 'Asynchronous)-     , NFData (Request api)-     , SetMember Process (Process effScheduler) eff-     , Member Interrupts eff-     )-  => (Api api 'Asynchronous -> Eff eff (CallbackResult 'Recoverable))-  -> (  forall secret reply-      . (Typeable reply, Typeable (Api api ( 'Synchronous reply)), NFData reply)-     => Api api ( 'Synchronous reply)-     -> (Eff eff (Maybe reply, CallbackResult 'Recoverable) -> secret)-     -> secret-     )-  -> MessageCallback api eff-handleCastsAndCalls onCast onCall = handleCalls onCall <> handleCasts onCast----- | A variation of 'handleCalls' that allows to defer a reply to a call.------ @since 0.14.2-handleCallsDeferred-  :: forall api eff effScheduler-   . ( HasCallStack-     , Typeable api-     , SetMember Process (Process effScheduler) eff-     , Member Interrupts eff-     )-  => (  forall reply-      . (Typeable reply, NFData reply, Typeable (Api api ( 'Synchronous reply)))-     => RequestOrigin (Api api ( 'Synchronous reply))-     -> Api api ( 'Synchronous reply)-     -> Eff eff (CallbackResult 'Recoverable)-     )-  -> MessageCallback api eff-handleCallsDeferred h = MessageCallback-  (selectMessageWith-    (\case-      (Cast _ :: Request api) -> Nothing-      callReq                 -> Just callReq-    )-  )-  (\(Call callRef fromPid req :: Request api) ->-    h (RequestOrigin fromPid callRef) req-  )--type family ResponseType request where-  ResponseType (Api a ('Synchronous r)) = r----- | A smart constructor for 'MessageCallback's------ @since 0.13.2-handleProcessDowns-  :: forall eff-   . HasCallStack-  => (MonitorReference -> Eff eff (CallbackResult 'Recoverable))-  -> MessageCallback '[] eff-handleProcessDowns k = MessageCallback selectMessage (k . downReference)---- | Compose two 'Api's to a type-level pair of them.------ > handleCalls api1calls ^: handleCalls api2calls ^:------ @since 0.13.2-(^:)-  :: forall (api1 :: Type) (apis2 :: [Type]) eff-   . HasCallStack-  => MessageCallback api1 eff-  -> MessageCallback apis2 eff-  -> MessageCallback (api1 ': apis2) eff-(MessageCallback selL runL) ^: (MessageCallback selR runR) =-  MessageCallback (Left <$> selL <|> Right <$> selR) (either runL runR)--infixr 5 ^:---- | Make a fallback handler, i.e. a handler to which no other can be composed--- to from the right.------ @since 0.13.2-fallbackHandler-  :: forall api eff-   . HasCallStack-  => MessageCallback api eff-  -> MessageCallback '[] eff-fallbackHandler (MessageCallback s r) = MessageCallback s r---- | A 'fallbackHandler' that drops the left-over messages.------ @since 0.13.2-dropUnhandledMessages :: forall eff . HasCallStack => MessageCallback '[] eff-dropUnhandledMessages =-  MessageCallback selectAnyMessage (const (return AwaitNext))---- | A 'fallbackHandler' that terminates if there are unhandled messages.------ @since 0.13.2-exitOnUnhandled :: forall eff . HasCallStack => MessageCallback '[] eff-exitOnUnhandled = MessageCallback selectAnyMessage $ \msg ->-  return (StopServer (ErrorInterrupt ("unhandled message " <> show msg)))---- | A 'fallbackHandler' that drops the left-over messages.------ @since 0.13.2-logUnhandledMessages-  :: forall eff-   . (Member Logs eff, HasCallStack)-  => MessageCallback '[] eff-logUnhandledMessages = MessageCallback selectAnyMessage $ \msg -> do-  logWarning ("ignoring unhandled message " <> T.pack (show msg))-  return AwaitNext----- | Helper type class for the return values of 'spawnApiServer' et al.------ @since 0.13.2-class ToServerPids (t :: k) where-  type PrettyServerPids t :: PrettyType-  type ServerPids t-  toServerPids :: proxy t -> ProcessId -> ServerPids t--instance ToServerPids '[] where-  type PrettyServerPids '[] = PutStr "any message"-  type ServerPids '[] = ProcessId-  toServerPids _ = id--instance-  forall (api1 :: Type) (api2 :: [Type])-  . (ToServerPids api1, ToServerPids api2)-  => ToServerPids (api1 ': api2) where-  type PrettyServerPids (api1 ': api2) = PrettyServerPids api1 <++> PutStr ", " <++> PrettyServerPids api2-  type ServerPids (api1 ': api2) = (ServerPids api1, ServerPids api2)-  toServerPids _ p =-    (toServerPids (Proxy @api1) p, toServerPids (Proxy @api2) p)--instance-  forall (api1 :: Type)-  . (ToServerPids api1)-  => ToServerPids api1 where-  type PrettyServerPids api1 = ToPretty api1-  type ServerPids api1 = Server api1-  toServerPids _ = asServer----- | Just a wrapper around a function that will be applied to the result of--- a 'MessageCallback's 'StopServer' clause, or an 'Interrupt' caught during--- the execution of @receive@ or a 'MessageCallback'------ @since 0.13.2-data InterruptCallback eff where-   InterruptCallback ::-     (Interrupt 'Recoverable -> Eff eff (CallbackResult 'NoRecovery)) -> InterruptCallback eff--instance Default (InterruptCallback eff) where-  def = stopServerOnInterrupt---- | A smart constructor for 'InterruptCallback's------ @since 0.13.2-stopServerOnInterrupt :: forall eff . HasCallStack => InterruptCallback eff-stopServerOnInterrupt = InterruptCallback (pure . StopServer . interruptToExit)

@@ -1,460 +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 seperate module, since the child-id can be reused when a child--- exits.------ @since 0.23.0-module Control.Eff.Concurrent.Api.Supervisor-  ( Sup()-  , SpawnFun-  , SupConfig(MkSupConfig)-  , supConfigChildStopTimeout-  , supConfigSpawnFun-  , SpawnErr(AlreadyStarted)-  , startSupervisor-  , stopSupervisor-  , isSupervisorAlive-  , monitorSupervisor-  , getDiagnosticInfo-  , spawnChild-  , lookupChild-  , stopChild-  ) where--import Control.DeepSeq (NFData(rnf))-import Control.Eff as Eff-import Control.Eff.Concurrent.Api-import Control.Eff.Concurrent.Api.Client-import Control.Eff.Concurrent.Api.Server-import Control.Eff.Concurrent.Api.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 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--- ** Types----- | A function that will initialize the child process.------ The process-id returned from this function is kept private, but--- it is possible to return '(ProcessId, ProcessId)' for example.------ The function is expected to return a value - usually a tuple of 'Server' values for--- the different aspects of a process, such as sending API requests, managing--- observers and receiving other auxiliary API messages.------ @since 0.23.0-type SpawnFun i e o = i -> Eff e (o, ProcessId)---- | Options that control the 'Sup i o' process.------ This contains:------ * a 'SpawnFun'--- * the 'Timeout' after requesting a normal child exit before brutally killing the child.------ @since 0.23.0-data SupConfig i e o =-  MkSupConfig-    { _supConfigSpawnFun         :: SpawnFun i e o-    , _supConfigChildStopTimeout :: Timeout-    }--makeLenses ''SupConfig---- | 'Api' type for supervisor processes.------ The /supervisor/ process contains a 'SpawnFun' from which it can spawn new child processes.------ The supervisor maps an identifier value of type @childId@ to a 'ProcessId' and a--- 'spawnResult' type.------ This 'spawnResult' is likely a tuple of 'Server' process ids that allow type-safe interaction with--- the process.------ Also, this serves as handle or reference for interacting with a supervisor process.------ A value of this type is returned by 'startSupervisor'------ @since 0.23.0-newtype Sup childId spawnResult =-  MkSup (Server (Sup childId spawnResult))-  deriving (Ord, Eq, Typeable, NFData)--instance (PrettyTypeShow (ToPretty childId), PrettyTypeShow (ToPretty spawnResult))-  => Show (Sup childId spawnResult) where-  showsPrec d (MkSup svr) = showsPrec d svr---- | The 'Api' instance contains methods to start, stop and lookup a child--- process, as well as a diagnostic callback.------ @since 0.23.0-data instance  Api (Sup i o) r where-        StartC :: i -> Api (Sup i o) ('Synchronous (Either (SpawnErr i o) o))-        StopC :: i -> Timeout -> Api (Sup i o) ('Synchronous Bool)-        LookupC :: i -> Api (Sup i o) ('Synchronous (Maybe o))-        GetDiagnosticInfo :: Api (Sup i o) ('Synchronous Text)-    deriving Typeable--type instance ToPretty (Sup i o) =-    PutStr "supervisor{" <++> ToPretty i <+> PutStr "=>" <+> ToPretty o <++> PutStr "}"--instance (Show i) => Show (Api (Sup i o) ('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 i) => NFData (Api (Sup i o) ('Synchronous r)) where-  rnf (StartC ci) = rnf ci-  rnf (StopC ci t) = rnf ci `seq` rnf t-  rnf (LookupC ci) = rnf ci-  rnf GetDiagnosticInfo = ()---- | Runtime-Errors occurring when spawning child-processes.------ @since 0.23.0-data SpawnErr i o-  = AlreadyStarted i-                   o-  deriving (Eq, Ord, Show, Typeable, Generic)--instance (NFData i, NFData o) => NFData (SpawnErr i o)----- ** 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 i e o-  . ( HasCallStack-    , Member Logs e-    , Lifted IO e-    , Ord i-    , Tangible i-    , Tangible o-    )-  => SupConfig i (InterruptableProcess e) o-  -> Eff (InterruptableProcess e) (Sup i o)-startSupervisor supConfig = do-  (_sup, supPid) <- spawnApiServerStateful initChildren (onRequest ^: onMessage) onInterrupt-  return (mkSup supPid)-  where-    mkSup supPid = MkSup (asServer supPid)-    initChildren :: Eff (InterruptableProcess e) (Children i o)-    initChildren = return def-    onRequest :: MessageCallback (Sup i o) (State (Children i o) ': InterruptableProcess e)-    onRequest = handleCalls onCall-      where-        onCall ::-             Api (Sup i o) ('Synchronous reply)-          -> (Eff (State (Children i o) ': InterruptableProcess e) (Maybe reply, CallbackResult 'Recoverable) -> st)-          -> st-        onCall GetDiagnosticInfo k = k ((, AwaitNext) . Just . pack . show <$> getChildren @i @o)-        onCall (LookupC i) k = k ((, AwaitNext) . Just . fmap _childOutput <$> lookupChildById @i @o i)-        onCall (StopC i t) k =-          k $ do-            mExisting <- lookupAndRemoveChildById @i @o i-            case mExisting of-              Nothing -> do-                return (Just False, AwaitNext)-              Just existingChild -> do-                stopOrKillChild i existingChild t-                return (Just True, AwaitNext)--        onCall (StartC i) k =-          k $ do-            (o, cPid) <- raise ((supConfig ^. supConfigSpawnFun) i)-            cMon <- monitor cPid-            mExisting <- lookupChildById i-            case mExisting of-              Nothing -> do-                putChild i (MkChild o cPid cMon)-                return (Just (Right o), AwaitNext)-              Just existingChild ->-                return (Just (Left (AlreadyStarted i (existingChild ^. childOutput))), AwaitNext)-    onMessage :: MessageCallback '[] (State (Children i o) ': InterruptableProcess e)-    onMessage = handleProcessDowns onDown <> handleAnyMessages onInfo-      where-        onDown mrChild = do-          oldEntry <- lookupAndRemoveChildByMonitor @i @o mrChild-          case oldEntry of-            Nothing ->-              logWarning ("unexpected process down: " <> pack (show mrChild))-            Just (i, c) -> do-              logInfo (  "process down: "-                      <> pack (show mrChild)-                      <> " for child "-                      <> pack (show i)-                      <> " => "-                      <> pack (show (_childOutput c))-                      )-          return AwaitNext--        onInfo d = do-          logInfo ("unexpected message received: " <> pack (show d))-          return AwaitNext-    onInterrupt :: InterruptCallback (State (Children i o) ': ConsProcess e)-    onInterrupt =-      InterruptCallback $ \e -> do-        let (logSev, exitReason) =-              case e of-                NormalExitRequested ->-                  (debugSeverity, ExitNormally)-                _ ->-                  (warningSeverity, interruptToExit (ErrorInterrupt ("supervisor interrupted: " <> show e)))-        stopAllChildren @i @o (supConfig ^. supConfigChildStopTimeout)-        logWithSeverity logSev ("supervisor stopping: " <> pack (show e))--        pure (StopServer exitReason)---- | Stop the supervisor and shutdown all processes.------ Block until the supervisor has finished.------ @since 0.23.0-stopSupervisor-  :: ( HasCallStack-     , Member Interrupts e-     , SetMember Process (Process q0) e-     , Member Logs e-     , Lifted IO e-     , Ord i-     , Tangible i-     , Tangible o-     )-  => Sup i o-  -> Eff e ()-stopSupervisor (MkSup svr) = do-  logInfo ("stopping supervisor: " <> pack (show svr))-  mr <- monitor (_fromServer svr)-  sendInterrupt (_fromServer svr) NormalExitRequested-  r <- receiveSelectedMessage (selectProcessDown mr)-  logInfo ("supervisor stopped: " <> pack (show svr) <> " " <> pack (show r))---- | Check if a supervisor process is still alive.------ @since 0.23.0-isSupervisorAlive-  :: ( HasCallStack-     , Member Interrupts e-     , Member Logs e-     , Typeable i-     , Typeable o-     , NFData i-     , NFData o-     , Show i-     , Show o-     , SetMember Process (Process q0) e-     )-  => Sup i o-  -> Eff e Bool-isSupervisorAlive (MkSup x) = isProcessAlive (_fromServer x)---- | Monitor a supervisor process.------ @since 0.23.0-monitorSupervisor-  :: ( HasCallStack-     , Member Interrupts e-     , Member Logs e-     , Typeable i-     , Typeable o-     , NFData i-     , NFData o-     , Show i-     , Show o-     , SetMember Process (Process q0) e-     )-  => Sup i o-  -> Eff e MonitorReference-monitorSupervisor (MkSup x) = monitor (_fromServer x)---- | Start, link and monitor a new child process using the 'SpawnFun' passed--- to 'startSupervisor'.------ @since 0.23.0-spawnChild ::-     ( HasCallStack-     , Member Interrupts e-     , Member Logs e-     , Ord i-     , Tangible i-     , Tangible o-     , SetMember Process (Process q0) e-     )-  => Sup i o-  -> i-  -> Eff e (Either (SpawnErr i o) o)-spawnChild (MkSup svr) cId = call svr (StartC cId)---- | Lookup the given child-id and return the output value of the 'SpawnFun'---   if the client process exists.------ @since 0.23.0-lookupChild ::-     ( HasCallStack-     , Member Interrupts e-     , Member Logs e-     , Ord i-     , Tangible i-     , Tangible o-     , SetMember Process (Process q0) e-     )-  => Sup i o-  -> i-  -> Eff e (Maybe o)-lookupChild (MkSup svr) cId = call svr (LookupC 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 ::-     ( HasCallStack-     , Member Interrupts e-     , Member Logs e-     , Ord i-     , Tangible i-     , Tangible o-     , SetMember Process (Process q0) e-     )-  => Sup i o-  -> i-  -> Eff e Bool-stopChild (MkSup svr) cId = call svr (StopC cId (TimeoutMicros 4000000))---- | Return a 'Text' describing the current state of the supervisor.------ @since 0.23.0-getDiagnosticInfo-  :: ( Ord i-     , Tangible i-     , Tangible o-     , Typeable e-     , HasCallStack-     , Member Interrupts e-     , SetMember Process (Process q0) e-     )-  => Sup i o-  -> Eff e Text-getDiagnosticInfo (MkSup s) = call s GetDiagnosticInfo---- Internal Functions--stopOrKillChild-  :: forall i o e q0 .-     ( Ord i-     , Tangible i-     , Tangible o-     , HasCallStack-     , SetMember Process (Process q0) e-     , Member Interrupts e-     , Lifted IO e-     , Lifted IO q0-     , Member Logs e-     , Member (State (Children i o)) e-     )-  => i-  -> Child o-  -> Timeout-  -> Eff e ()-stopOrKillChild cId c stopTimeout =-      do-        sup <- MkSup . asServer @(Sup i o) <$> self-        t <- startTimer stopTimeout-        sendInterrupt (c^.childProcessId) NormalExitRequested-        r1 <- receiveSelectedMessage (   Right <$> selectProcessDown (c^.childMonitoring)-                                     <|> Left  <$> selectTimerElapsed t )-        case r1 of-          Left timerElapsed -> do-            logWarning (pack (show timerElapsed) <> ": child "<> pack (show cId) <>" => " <> pack(show (c^.childOutput)) <>" did not shutdown in time")-            sendShutdown-              (c^.childProcessId)-              (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^.childOutput)) <>" terminated: " <> pack (show (downReason downMsg)))--stopAllChildren-  :: forall i o e q0 .-     ( Ord i-     , Tangible i-     , Tangible o-     , HasCallStack-     , SetMember Process (Process q0) e-     , Lifted IO e-     , Lifted IO q0-     , Member Logs e-     , Member (State (Children i o)) e-     )-  => Timeout -> Eff e ()-stopAllChildren stopTimeout = removeAllChildren @i @o >>= 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,101 +0,0 @@-module Control.Eff.Concurrent.Api.Supervisor.InternalState where--import Control.DeepSeq-import Control.Eff as Eff-import Control.Eff.Concurrent.Process-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 o = MkChild-  { _childOutput :: o-  , _childProcessId :: ProcessId-  , _childMonitoring :: MonitorReference-  }-  deriving (Show, Generic, Typeable, Eq, Ord)--instance (NFData o) => NFData (Child o)--makeLenses ''Child----- | Internal state.-data Children i o = MkChildren-  { _childrenById :: Map i (Child o)-  , _childrenByMonitor :: Map MonitorReference (i, Child o)-  } deriving (Show, Generic, Typeable)--instance Default (Children i o) where-  def = MkChildren def def--instance (NFData i, NFData o) => NFData (Children i o)--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

@@ -376,9 +376,6 @@ selectAnyMessage :: MessageSelector StrictDynamic selectAnyMessage = MessageSelector Just --- | /Cons/ 'Process' onto a list of effects.-type ConsProcess r = Process r ': r- -- | The state that a 'Process' is currently in. data ProcessState =     ProcessBooting              -- ^ The process has just been started but not@@ -576,12 +573,15 @@ -- | 'Interrupt's which are 'Recoverable'. type RecoverableInterrupt = Interrupt 'Recoverable +-- | /Cons/ 'Process' onto a list of effects.+type ConsProcess r = Process r ': r+ -- | 'Exc'eptions containing 'Interrupt's. -- See 'handleInterrupts', 'exitOnInterrupt' or 'provideInterrupts' type Interrupts = Exc (Interrupt 'Recoverable) --- | This adds a layer of the 'Interrupts' effect on top of 'ConsProcess'-type InterruptableProcess e = Interrupts ': ConsProcess e+-- | This adds a layer of the 'Interrupts' effect on top of 'Process'+type InterruptableProcess e = Interrupts ': Process e ': e  -- | Handle all 'Interrupt's of an 'InterruptableProcess' by -- wrapping them up in 'interruptToExit' and then do a process 'Shutdown'.

@@ -38,14 +38,11 @@ import           Control.Concurrent import           Control.Concurrent.STM import           Control.Eff-import           Control.Eff.Concurrent.Api-import           Control.Eff.Concurrent.Api.Client+import           Control.Eff.Concurrent.Protocol+import           Control.Eff.Concurrent.Protocol.Client import           Control.Eff.Concurrent.Process import           Control.Monad import           Data.Foldable-import           Data.Typeable                  ( Typeable )-import           Data.Type.Pretty-import           Control.DeepSeq import           System.Timeout  -- | Contains the communication channels to interact with a scheduler running in@@ -134,13 +131,11 @@ submitCast   :: forall o r    . ( SetMember Lift (Lift IO) r-     , Typeable o-     , PrettyTypeShow (ToPretty o)-     , NFData (Api o 'Asynchronous)+     , TangiblePdu o 'Asynchronous      , Member Interrupts r)   => SchedulerSession r-  -> Server o-  -> Api o 'Asynchronous+  -> Endpoint o+  -> Pdu o 'Asynchronous   -> IO () submitCast sc svr request = submit sc (cast svr request) @@ -148,16 +143,12 @@ submitCall   :: forall o q r    . ( SetMember Lift (Lift IO) r-     , Typeable o-     , PrettyTypeShow (ToPretty o)-     , Typeable q-     , NFData q-     , Show q      , Member Interrupts r-     , NFData (Api o ('Synchronous q))+     , TangiblePdu o ('Synchronous q)+     , Tangible q      )   => SchedulerSession r-  -> Server o-  -> Api o ( 'Synchronous q)+  -> Endpoint o+  -> Pdu o ( 'Synchronous q)   -> IO q submitCall sc svr request = submit sc (call svr request)

@@ -0,0 +1,238 @@+-- | This module contains a mechanism to specify what kind of messages (aka+-- /requests/) a 'Endpoint' ('Process') can handle, and if the caller blocks and+-- waits for an answer, which the server process provides.+--+-- The type magic in the 'Pdu' type family allows to define a related set of /requests/ along+-- with the corresponding responses.+--+-- Request handling can be either blocking, if a response is required, or+-- non-blocking.+--+-- A process can /serve/ a specific 'Pdu' instance by using the functions provided by+-- the "Control.Eff.Concurrent.Pdu.Server" module.+--+-- To enable a process to use such a /service/, the functions provided by+-- the "Control.Eff.Concurrent.Pdu.Client" should be used.+--+module Control.Eff.Concurrent.Protocol+  ( Pdu(..)+  , Synchronicity(..)+  , ProtocolReply+  , Tangible+  , TangiblePdu+  , Endpoint(..)+  , fromEndpoint+  , proxyAsEndpoint+  , asEndpoint+  , EmbedProtocol(..)+  , prettyTypeableShows+  , prettyTypeableShowsPrec+  )+where++import           Control.DeepSeq+import           Control.Eff.Concurrent.Process+import           Control.Lens+import           Data.Kind+import           Type.Reflection+import           Data.Type.Pretty++-- | This data family defines the **protocol data units**(PDU) of a /protocol/.+--+-- A Protocol in the sense of a communication interface description+-- between processes.+--+-- The first parameter is usually a user defined type that identifies the+-- protocol that uses the 'Pdu's are. It maybe a /phantom/ type.+--+-- The second parameter specifies if a specific constructor of an (GADT-like)+-- @Pdu@ instance is 'Synchronous', i.e. returns a result and blocks the caller+-- or if it is 'Asynchronous'+--+-- Example:+--+-- >+-- > data BookShop deriving Typeable+-- >+-- > data instance Pdu BookShop r where+-- >   RentBook  :: BookId   -> Pdu BookShop ('Synchronous (Either RentalError RentalId))+-- >   BringBack :: RentalId -> Pdu BookShop 'Asynchronous+-- >+-- > type BookId = Int+-- > type RentalId = Int+-- > type RentalError = String+-- >+data family Pdu (protocol :: Type) (reply :: Synchronicity)++type instance ToPretty (Pdu x y) =+  PrettySurrounded (PutStr "<") (PutStr ">") ("protocol" <:> ToPretty x <+> ToPretty y)++-- | A set of constraints for types that can evaluated via 'NFData', compared via 'Ord' and presented+-- dynamically via 'Typeable', and represented both as values+-- via 'Show'.+--+-- @since 0.23.0+type Tangible i =+  ( NFData i+  , Typeable i+  , Show i+  )++-- | A 'Constraint' that bundles the requirements for the+-- 'Pdu' values of a protocol.+--+-- This ensures that 'Pdu's can be strictly and deeply evaluated and shown+-- such that for example logging is possible.+--+-- @since 0.24.0+type TangiblePdu p r =+  ( Typeable p+  , Typeable r+  , Tangible (Pdu p r)+  )++-- | The (promoted) constructors of this type specify (at the type level) the+-- reply behavior of a specific constructor of an @Pdu@ instance.+data Synchronicity =+  Synchronous Type -- ^ Specify that handling a request is a blocking operation+                   -- with a specific return type, e.g. @('Synchronous (Either+                   -- RentalError RentalId))@+  | Asynchronous -- ^ Non-blocking, asynchronous, request handling+    deriving Typeable++-- | This type function takes an 'Pdu' and analysis the reply type, i.e. the 'Synchronicity'+-- and evaluates to either @t@ for an+-- @Pdu x ('Synchronous' t)@ or to '()' for an @Pdu x 'Asynchronous'@.+--+-- @since 0.24.0+type family ProtocolReply (s :: Synchronicity) where+  ProtocolReply ('Synchronous t) = t+  ProtocolReply 'Asynchronous = ()++-- | This is a tag-type that wraps around a 'ProcessId' and holds an 'Pdu' index+-- type.+newtype Endpoint protocol = Endpoint { _fromEndpoint :: ProcessId }+  deriving (Eq,Ord,Typeable, NFData)++instance Typeable protocol => Show (Endpoint protocol) where+  showsPrec d (Endpoint c) =+    showParen (d>=10)+    (prettyTypeableShows (SomeTypeRep (typeRep @protocol)) . showsPrec 10 c)++-- | This is equivalent to @'prettyTypeableShowsPrec' 0@+--+-- @since 0.24.0+prettyTypeableShows :: SomeTypeRep -> ShowS+prettyTypeableShows = prettyTypeableShowsPrec 0++-- | An internal utility to print 'Typeable' without the kinds.+-- This is like 'showsPrec' in that it accepts a /precedence/ parameter,+-- and the result is in parentheses when the precedence is higher than 9.+--+-- @since 0.24.0+prettyTypeableShowsPrec :: Int -> SomeTypeRep -> ShowS+prettyTypeableShowsPrec 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)+                     (prettyTypeableShowsPrec 10 <$> conArgs))+            sIn++type instance ToPretty (Endpoint a) = ToPretty a <+> PutStr "endpoint"++makeLenses ''Endpoint++-- | Tag a 'ProcessId' with an 'Pdu' type index to mark it a 'Endpoint' process+-- handling that API+proxyAsEndpoint :: proxy protocol -> ProcessId -> Endpoint protocol+proxyAsEndpoint = const Endpoint++-- | Tag a 'ProcessId' with an 'Pdu' type index to mark it a 'Endpoint' process+-- handling that API+asEndpoint :: forall protocol . ProcessId -> Endpoint protocol+asEndpoint = Endpoint++-- | A class for 'Pdu' instances that embed other 'Pdu'.+-- A 'Prism' for the embedded 'Pdu' is the center of this class+--+-- Laws: @embeddedPdu = prism' embedPdu fromPdu@+--+-- @since 0.24.0+class EmbedProtocol protocol embeddedProtocol where+  -- | A 'Prism' for the embedded 'Pdu's.+  embeddedPdu :: Prism' (Pdu protocol result) (Pdu embeddedProtocol result)+  embeddedPdu = prism' embedPdu fromPdu+  -- | Embed the 'Pdu' value of an embedded protocol into the corresponding+  --  'Pdu' value.+  embedPdu :: Pdu embeddedProtocol r -> Pdu protocol r+  embedPdu = review embeddedPdu+  -- | Examine a 'Pdu' value from the outer protocol, and return it, if it embeds a 'Pdu' of+  -- embedded protocol, otherwise return 'Nothing'/+  fromPdu :: Pdu protocol r -> Maybe (Pdu embeddedProtocol r)+  fromPdu = preview embeddedPdu++instance EmbedProtocol a a where+  embeddedPdu = prism' id Just+  embedPdu = id+  fromPdu = Just++data instance Pdu (a1, a2) x where+        ToPduLeft :: Pdu a1 r -> Pdu (a1, a2) r+        ToPduRight :: Pdu a2 r -> Pdu (a1, a2) r+  deriving Typeable++instance (NFData (Pdu a1 r), NFData (Pdu a2 r)) => NFData (Pdu (a1, a2) r) where+  rnf (ToPduLeft x) = rnf x+  rnf (ToPduRight y) = rnf y++instance (Show (Pdu a1 r), Show (Pdu a2 r)) => Show (Pdu (a1, a2) r) where+  showsPrec d (ToPduLeft x) = showsPrec d x+  showsPrec d (ToPduRight y) = showsPrec d y++instance EmbedProtocol (a1, a2) a1 where+  embedPdu = ToPduLeft+  fromPdu (ToPduLeft l) = Just l+  fromPdu _ = Nothing++instance EmbedProtocol (a1, a2) a2 where+  embeddedPdu =+    prism' ToPduRight $ \case+      ToPduRight r -> Just r+      ToPduLeft _ -> Nothing++data instance Pdu (a1, a2, a3) x where+  ToPdu1 :: Pdu a1 r -> Pdu (a1, a2, a3) r+  ToPdu2 :: Pdu a2 r -> Pdu (a1, a2, a3) r+  ToPdu3 :: Pdu a3 r -> Pdu (a1, a2, a3) r+  deriving Typeable++instance (NFData (Pdu a1 r), NFData (Pdu a2 r), NFData (Pdu a3 r)) => NFData (Pdu (a1, a2, a3) r) where+  rnf (ToPdu1 x) = rnf x+  rnf (ToPdu2 y) = rnf y+  rnf (ToPdu3 z) = rnf z++instance (Show (Pdu a1 r), Show (Pdu a2 r), Show (Pdu a3 r)) => Show (Pdu (a1, a2, a3) r) where+  showsPrec d (ToPdu1 x) = showsPrec d x+  showsPrec d (ToPdu2 y) = showsPrec d y+  showsPrec d (ToPdu3 z) = showsPrec d z++instance EmbedProtocol (a1, a2, a3) a1 where+  embedPdu = ToPdu1+  fromPdu (ToPdu1 l) = Just l+  fromPdu _ = Nothing++instance EmbedProtocol (a1, a2, a3) a2 where+  embedPdu = ToPdu2+  fromPdu (ToPdu2 l) = Just l+  fromPdu _ = Nothing++instance EmbedProtocol (a1, a2, a3) a3 where+  embedPdu = ToPdu3+  fromPdu (ToPdu3 l) = Just l+  fromPdu _ = Nothing+

@@ -0,0 +1,195 @@+-- | Functions for protocol clients.+--+-- This modules is required to write clients that send'Pdu's.+module Control.Eff.Concurrent.Protocol.Client+  ( -- * Calling APIs directly+    cast+  , call+  , callWithTimeout+  -- * Server Process Registration+  , castEndpointReader+  , callEndpointReader+  , ServesProtocol+  , EndpointReader+  , askEndpoint+  , runEndpointReader+  )+where++import           Control.Eff+import           Control.Eff.Reader.Strict+import           Control.Eff.Concurrent.Protocol+import           Control.Eff.Concurrent.Protocol.Request+import           Control.Eff.Concurrent.Process+import           Control.Eff.Concurrent.Process.Timer+import           Control.Eff.Log+import           Data.Typeable                  ( Typeable )+import           GHC.Stack++-- | Send a request 'Pdu' that has no reply and return immediately.+--+-- The type signature enforces that the corresponding 'Pdu' clause is+-- 'Asynchronous'. The operation never fails, if it is important to know if the+-- message was delivered, use 'call' instead.+--+-- The message will be reduced to normal form ('rnf') in the caller process.+cast+  :: forall o' o r q+   . ( HasCallStack+     , SetMember Process (Process q) r+     , Member Interrupts r+     , TangiblePdu o' 'Asynchronous+     , TangiblePdu o 'Asynchronous+     , EmbedProtocol o' o+     )+  => Endpoint o'+  -> Pdu o 'Asynchronous+  -> Eff r ()+cast (Endpoint pid) castMsg = sendMessage pid (Cast (embedPdu @o' castMsg))++-- | Send a request 'Pdu' and wait for the server to return a result value.+--+-- The type signature enforces that the corresponding 'Pdu' clause is+-- 'Synchronous'.+--+-- __Always prefer 'callWithTimeout' over 'call'__+call+  :: forall result protocol' protocol r q+   . ( SetMember Process (Process q) r+     , Member Interrupts r+     , TangiblePdu protocol' ( 'Synchronous result)+     , TangiblePdu protocol ( 'Synchronous result)+     , EmbedProtocol protocol' protocol+     , Tangible result+     , HasCallStack+     )+  => Endpoint protocol'+  -> Pdu protocol ( 'Synchronous result)+  -> Eff r result+call (Endpoint pidInternal) req = do+  callRef <- makeReference+  fromPid <- self+  let requestMessage = Call origin $! (embedPdu @protocol' req)+      origin = RequestOrigin @protocol' @result fromPid callRef+  sendMessage pidInternal requestMessage+  let selectResult :: MessageSelector result+      selectResult =+        let extractResult+              :: Reply protocol' result -> Maybe result+            extractResult (Reply origin' result) =+              if origin == origin' then Just result else Nothing+        in  selectMessageWith extractResult+  resultOrError <- receiveWithMonitor pidInternal selectResult+  either (interrupt . becauseProcessIsDown) return resultOrError+++-- | Send an request 'Pdu' and wait for the server to return a result value.+--+-- The type signature enforces that the corresponding 'Pdu' clause is+-- 'Synchronous'.+--+-- If the server that was called dies, this function interrupts the+-- process with 'ProcessDown'.+-- If the server takes longer to reply than the given timeout, this+-- function interrupts the process with 'TimeoutInterrupt'.+--+-- __Always prefer this function over 'call'__+--+-- @since 0.22.0+callWithTimeout+  :: forall result protocol' protocol r q+   . ( SetMember Process (Process q) r+     , Member Interrupts r+     , TangiblePdu protocol' ( 'Synchronous result)+     , TangiblePdu protocol ( 'Synchronous result)+     , EmbedProtocol protocol' protocol+     , Tangible result+     , Member Logs r+     , Lifted IO q+     , Lifted IO r+     , HasCallStack+     )+  => Endpoint protocol'+  -> Pdu protocol ( 'Synchronous result)+  -> Timeout+  -> Eff r result+callWithTimeout serverP@(Endpoint pidInternal) req timeOut = do+  fromPid <- self+  callRef <- makeReference+  let requestMessage = Call origin $! embedPdu @protocol' req+      origin = RequestOrigin @protocol' @result fromPid callRef+  sendMessage pidInternal requestMessage+  let selectResult =+        let extractResult+              :: Reply protocol' result -> Maybe result+            extractResult (Reply origin' result) =+              if origin == origin' then Just result else Nothing+        in selectMessageWith extractResult+  resultOrError <- receiveSelectedWithMonitorAfter pidInternal selectResult timeOut+  let onTimeout timerRef = do+        let msg = "call timed out after "+                  ++ show timeOut ++ " to server: "+                  ++ show serverP ++ " from "+                  ++ show fromPid ++ " "+                  ++ show timerRef+        logWarning' msg+        interrupt (TimeoutInterrupt msg)+      onProcDown p = do+        logWarning' ("call to dead server: "++ show serverP ++ " from " ++ show fromPid)+        interrupt (becauseProcessIsDown p)+  either (either onProcDown onTimeout) return resultOrError++-- | Instead of passing around a 'Endpoint' value and passing to functions like+-- 'cast' or 'call', a 'Endpoint' can provided by a 'Reader' effect, if there is+-- only a __single server__ for a given 'Pdu' instance. This type alias is+-- convenience to express that an effect has 'Process' and a reader for a+-- 'Endpoint'.+type ServesProtocol o r q =+  ( Typeable o+  , SetMember Process (Process q) r+  , Member (EndpointReader o) r+  )++-- | The reader effect for 'ProcessId's for 'Pdu's, see 'runEndpointReader'+type EndpointReader o = Reader (Endpoint o)++-- | Run a reader effect that contains __the one__ server handling a specific+-- 'Pdu' instance.+runEndpointReader+  :: HasCallStack => Endpoint o -> Eff (EndpointReader o ': r) a -> Eff r a+runEndpointReader = runReader++-- | Get the 'Endpoint' registered with 'runEndpointReader'.+askEndpoint :: Member (EndpointReader o) e => Eff e (Endpoint o)+askEndpoint = ask++-- | Like 'call' but take the 'Endpoint' from the reader provided by+-- 'runEndpointReader'.+callEndpointReader+  :: forall reply o r q .+     ( ServesProtocol o r q+     , HasCallStack+     , Tangible reply+     , TangiblePdu o ( 'Synchronous reply)+     , Member Interrupts r+     )+  => Pdu o ( 'Synchronous reply)+  -> Eff r reply+callEndpointReader method = do+  serverPid <- askEndpoint @o+  call @reply @o @o serverPid method++-- | Like 'cast' but take the 'Endpoint' from the reader provided by+-- 'runEndpointReader'.+castEndpointReader+  :: forall o r q .+     ( ServesProtocol o r q+     , HasCallStack+     , Member Interrupts r+     , TangiblePdu o 'Asynchronous+     )+  => Pdu o 'Asynchronous+  -> Eff r ()+castEndpointReader method = do+  serverPid <- askEndpoint @o+  cast @o @o serverPid method

@@ -0,0 +1,286 @@+-- | Observer Effects+--+-- This module supports the implementation of observers and observables. Expected use+-- case is event propagation.+--+-- @since 0.16.0+module Control.Eff.Concurrent.Protocol.Observer+  ( Observer(..)+  , TangibleObserver+  , Pdu(RegisterObserver, ForgetObserver, Observed)+  , registerObserver+  , forgetObserver+  , handleObservations+  , toObserver+  , toObserverFor+  , ObserverRegistry+  , ObserverState+  , Observers()+  , emptyObservers+  , handleObserverRegistration+  , manageObservers+  , observed+  )+where++import           Control.DeepSeq               (NFData(rnf))+import           Control.Eff+import           Control.Eff.Concurrent.Process+import           Control.Eff.Concurrent.Protocol+import           Control.Eff.Concurrent.Protocol.Client+import           Control.Eff.State.Strict+import           Control.Eff.Log+import           Control.Lens+import           Data.Data                     (typeOf)+import           Data.Dynamic+import           Data.Foldable+import           Data.Proxy+import           Data.Set                       ( Set )+import qualified Data.Set                      as Set+import           Data.Text                      ( pack )+import           Data.Typeable                  ( typeRep )+import           Data.Type.Pretty+import           GHC.Stack++-- * Observers++-- | Describes a process that observes another via 'Asynchronous' 'Pdu' messages.+--+-- An observer consists of a filter and a process id. The filter converts an observation to+-- a message understood by the observer process, and the 'ProcessId' is used to send the message.+--+-- @since 0.16.0+data Observer o where+  Observer+    :: ( Tangible o+       , TangiblePdu p 'Asynchronous+       , Tangible (Endpoint p)+       , Typeable p+       )+    => (o -> Maybe (Pdu p 'Asynchronous)) -> Endpoint p -> Observer o+++-- | The constraints on the type parameters to an 'Observer'+--+-- @since 0.24.0+type TangibleObserver o =+  ( Tangible o, TangiblePdu (Observer o) 'Asynchronous)++type instance ToPretty (Observer o) =+  PrettyParens ("observing" <:> ToPretty o)++instance (NFData o) => NFData (Observer o) where+  rnf (Observer k s) = rnf k `seq` rnf s++instance Show (Observer o) where+  showsPrec d (Observer _ p) = showParen+    (d >= 10)+    (shows (typeRep (Proxy :: Proxy o)) . showString " observer: " . shows p)++instance Ord (Observer o) where+  compare (Observer _ s1) (Observer _ s2) =+    compare (s1 ^. fromEndpoint) (s2 ^. fromEndpoint)++instance Eq (Observer o) where+  (==) (Observer _ s1) (Observer _ s2) =+    (==) (s1 ^. fromEndpoint) (s2 ^. fromEndpoint)++-- | And an 'Observer' to the set of recipients for all observations reported by 'observed'.+--   Note that the observers are keyed by the observing process, i.e. a previous entry for the process+--   contained in the 'Observer' is overwritten. If you want multiple entries for a single process, just+--   combine several filter functions.+--+-- @since 0.16.0+registerObserver+  :: ( SetMember Process (Process q) r+     , HasCallStack+     , Member Interrupts r+     , TangibleObserver o+     , EmbedProtocol x (ObserverRegistry o)+     , TangiblePdu x 'Asynchronous+     )+  => Observer o+  -> Endpoint x+  -> Eff r ()+registerObserver observer observerRegistry =+  cast observerRegistry (RegisterObserver observer)++-- | Send the 'ForgetObserver' message+--+-- @since 0.16.0+forgetObserver+  :: ( SetMember Process (Process q) r+     , HasCallStack+     , Member Interrupts r+     , Typeable o+     , NFData o+     , EmbedProtocol x (ObserverRegistry o)+     , TangiblePdu x 'Asynchronous+     )+  => Observer o+  -> Endpoint x+  -> Eff r ()+forgetObserver observer observerRegistry =+  cast observerRegistry (ForgetObserver observer)++-- ** Observer Support Functions++-- | A minimal Protocol for handling observations.+-- This is one simple way of receiving observations - of course users can use+-- any other 'Asynchronous' 'Pdu' message type for receiving observations.+--+-- @since 0.16.0+data instance Pdu (Observer o) r where+  -- | This message denotes that the given value was 'observed'.+  --+  -- @since 0.16.1+  Observed :: o -> Pdu (Observer o) 'Asynchronous+  deriving Typeable++instance NFData o => NFData (Pdu (Observer o) 'Asynchronous) where+  rnf (Observed o) = rnf o++instance Show o => Show (Pdu (Observer o) r) where+  showsPrec d (Observed o) = showParen (d>=10) (showString "observered: " . shows o)++-- | Based on the 'Pdu' instance for 'Observer' this simplified writing+-- a callback handler for observations. In order to register to+-- and 'ObserverRegistry' use 'toObserver'.+--+-- @since 0.16.0+handleObservations+  :: (HasCallStack, Typeable o, SetMember Process (Process q) r, NFData (Observer o))+  => (o -> Eff r ())+  -> Pdu (Observer o) 'Asynchronous -> Eff r ()+handleObservations k (Observed r) = k r++-- | Use a 'Endpoint' as an 'Observer' for 'handleObservations'.+--+-- @since 0.16.0+toObserver :: TangibleObserver o => Endpoint (Observer o) -> Observer o+toObserver = toObserverFor Observed++-- | Create an 'Observer' that conditionally accepts all observations of the+-- given type and applies the given function to them; the function takes an observation and returns an 'Pdu'+-- cast that the observer server is compatible to.+--+-- @since 0.16.0+toObserverFor+  :: (TangibleObserver o, Typeable a, TangiblePdu a 'Asynchronous)+  => (o -> Pdu a 'Asynchronous)+  -> Endpoint a+  -> Observer o+toObserverFor wrapper = Observer  (Just . wrapper)++-- * Managing Observers++-- | A protocol for managing 'Observer's, encompassing  registration and de-registration of+-- 'Observer's.+--+-- @since 0.16.0+data ObserverRegistry o+  deriving Typeable++type instance ToPretty (ObserverRegistry o) =+  PrettyParens ("observer registry" <:> ToPretty o)++-- | Protocol for managing observers. This can be added to any server for any number of different observation types.+-- The functions 'manageObservers' and 'handleObserverRegistration' are used to include observer handling;+--+-- @since 0.16.0+data instance Pdu (ObserverRegistry o) r where+  -- | This message denotes that the given 'Observer' should receive observations until 'ForgetObserver' is+  --   received.+  --+  -- @since 0.16.1+  RegisterObserver :: NFData o => Observer o -> Pdu (ObserverRegistry o) 'Asynchronous+  -- | This message denotes that the given 'Observer' should not receive observations anymore.+  --+  -- @since 0.16.1+  ForgetObserver :: NFData o => Observer o -> Pdu (ObserverRegistry o) 'Asynchronous+  deriving Typeable++instance NFData (Pdu (ObserverRegistry o) r) where+  rnf (RegisterObserver o) = rnf o+  rnf (ForgetObserver o) = rnf o++instance Show (Pdu (ObserverRegistry o) r) where+  showsPrec d (RegisterObserver o) = showParen (d >= 10) (showString "register observer: " . showsPrec 11 o)+  showsPrec d (ForgetObserver o) = showParen (d >= 10) (showString "forget observer: " . showsPrec 11 o)++-- ** Protocol for integrating 'ObserverRegistry' into processes.++-- | Provide the implementation for the 'ObserverRegistry' Protocol, this handled 'RegisterObserver' and 'ForgetObserver'+-- messages. It also adds the 'ObserverState' constraint to the effect list.+--+-- @since 0.16.0+handleObserverRegistration+  :: forall o q r+   . ( HasCallStack+     , Typeable o+     , SetMember Process (Process q) r+     , Member (ObserverState o) r+     , Member Logs r+     )+  => Pdu (ObserverRegistry o) 'Asynchronous -> Eff r ()+handleObserverRegistration = \case+    RegisterObserver ob -> do+      os <- get @(Observers o)+      logDebug ("registering "+               <> pack (show (typeOf ob))+               <> " current number of observers: "+               <> pack (show (Set.size (view observers os))))+      put (over observers (Set.insert ob) os)++    ForgetObserver ob -> do+      os <- get @(Observers o)+      logDebug ("forgetting "+               <> pack (show (typeOf ob))+               <> " current number of observers: "+               <> pack (show (Set.size (view observers os))))+      put (over observers (Set.delete ob) os)++-- | Keep track of registered 'Observer's.+--+-- Handle the 'ObserverState' introduced by 'handleObserverRegistration'.+--+-- @since 0.16.0+manageObservers :: Eff (ObserverState o ': r) a -> Eff r a+manageObservers = evalState (Observers Set.empty)++-- | The empty 'ObserverState'+--+-- @since 0.24.0+emptyObservers :: Observers o+emptyObservers = Observers Set.empty++-- | Internal state for 'manageObservers'+newtype Observers o =+  Observers { _observers :: Set (Observer o) }+   deriving (Eq, Ord, Typeable, Show, NFData)++-- | Alias for the effect that contains the observers managed by 'manageObservers'+type ObserverState o = State (Observers o)++observers :: Iso' (Observers o) (Set (Observer o))+observers = iso _observers Observers++-- | Report an observation to all observers.+-- The process needs to 'manageObservers' and to 'handleObserverRegistration'.+--+-- @since 0.16.0+observed+  :: forall o r q+   . ( SetMember Process (Process q) r+     , Member (ObserverState o) r+     , Member Interrupts r+     , TangibleObserver o+     )+  => o+  -> Eff r ()+observed observation = do+  os <- view observers <$> get+  mapM_ notifySomeObserver os+ where+  notifySomeObserver (Observer messageFilter receiver) =+    traverse_ (cast receiver) (messageFilter observation)

@@ -0,0 +1,161 @@+-- | 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+  , readObservationQueue+  , tryReadObservationQueue+  , flushObservationQueue+  , withObservationQueue+  , spawnLinkObservationQueueWriter+  )+where++import           Control.Concurrent.STM+import           Control.Eff+import           Control.Eff.Concurrent.Protocol+import           Control.Eff.Concurrent.Protocol.Observer+import           Control.Eff.Concurrent.Protocol.Request+import           Control.Eff.Concurrent.Protocol.Server+import           Control.Eff.Concurrent.Process+import           Control.Eff.ExceptionExtra     ( )+import           Control.Eff.Log+import           Control.Eff.Reader.Strict+import           Control.Exception.Safe        as Safe+import           Control.Monad.IO.Class+import           Control.Monad                  ( unless )+import qualified Data.Text                     as T+import           Data.Typeable+import           GHC.Stack++-- | Contains a 'TBQueue' capturing observations.+-- See 'spawnLinkObservationQueueWriter', 'readObservationQueue'.+newtype ObservationQueue a = ObservationQueue (TBQueue a)++-- | A 'Reader' for an 'ObservationQueue'.+type ObservationQueueReader a = Reader (ObservationQueue a)++logPrefix :: forall o proxy . (HasCallStack, Typeable o) => proxy o -> T.Text+logPrefix px = "observation queue: " <> T.pack (show (typeRep px))++-- | Read queued observations captured and enqueued in the shared 'TBQueue' by 'spawnLinkObservationQueueWriter'.+-- This blocks until something was captured or an interrupt or exceptions was thrown. For a non-blocking+-- variant use 'tryReadObservationQueue' or 'flushObservationQueue'.+readObservationQueue+  :: forall o r+   . ( Member (ObservationQueueReader o) r+     , HasCallStack+     , MonadIO (Eff r)+     , Typeable o+     , Member Logs r+     )+  => Eff r o+readObservationQueue = do+  ObservationQueue q <- ask @(ObservationQueue o)+  liftIO (atomically (readTBQueue q))++-- | Read queued observations captured and enqueued in the shared 'TBQueue' by 'spawnLinkObservationQueueWriter'.+-- Return the oldest enqueued observation immediately or 'Nothing' if the queue is empty.+-- Use 'readObservationQueue' to block until an observation is observed.+tryReadObservationQueue+  :: forall o r+   . ( Member (ObservationQueueReader o) r+     , HasCallStack+     , MonadIO (Eff r)+     , Typeable o+     , Member Logs r+     )+  => Eff r (Maybe o)+tryReadObservationQueue = do+  ObservationQueue q <- ask @(ObservationQueue o)+  liftIO (atomically (tryReadTBQueue q))++-- | Read at once all currently queued observations captured and enqueued+-- in the shared 'TBQueue' by 'spawnLinkObservationQueueWriter'.+-- This returns immediately all currently enqueued observations.+-- For a blocking variant use 'readObservationQueue'.+flushObservationQueue+  :: forall o r+   . ( Member (ObservationQueueReader o) r+     , HasCallStack+     , MonadIO (Eff r)+     , Typeable o+     , Member Logs r+     )+  => Eff r [o]+flushObservationQueue = do+  ObservationQueue q <- ask @(ObservationQueue o)+  liftIO (atomically (flushTBQueue q))++-- | Create a mutable queue for observations. Use 'spawnLinkObservationQueueWriter' for a simple way to get+-- a process that enqueues all observations.+--+-- ==== __Example__+--+-- @+-- withObservationQueue 100 $ do+--   q  <- ask \@(ObservationQueueReader TestEvent)+--   wq <- spawnLinkObservationQueueWriter q+--   registerObserver wq testServer+--   ...+--   cast testServer DoSomething+--   evt <- readObservationQueue \@TestEvent+--   ...+-- @+--+-- @since 0.18.0+withObservationQueue+  :: forall o b e len+   . ( HasCallStack+     , Typeable o+     , Show o+     , Member Logs e+     , Lifted IO e+     , Integral len+     , Member Interrupts e+     )+  => len+  -> Eff (ObservationQueueReader o ': e) b+  -> Eff e b+withObservationQueue queueLimit e = do+  q   <- lift (newTBQueueIO (fromIntegral queueLimit))+  res <- handleInterrupts (return . Left)+                          (Right <$> runReader (ObservationQueue q) e)+  rest <- lift (atomically (flushTBQueue q))+  unless+    (null rest)+    (logNotice (logPrefix (Proxy @o) <> " unread observations: " <> T.pack (show rest)))+  either (\em -> logError (T.pack (show em)) >> lift (throwIO em)) return res++-- | Spawn a process that can be used as an 'Observer' that enqueues the observations into an+--   'ObservationQueue'. See 'withObservationQueue' for an example.+--+-- The observations can be obtained by 'readObservationQueue'. All observations are captured up to+-- the queue size limit, such that the first message received will be first message+-- returned by 'readObservationQueue'.+--+-- @since 0.18.0+spawnLinkObservationQueueWriter+  :: forall o q h+   . ( TangibleObserver o+     , TangiblePdu (Observer o) 'Asynchronous+     , Member Logs q+     , Lifted IO q+     , LogsTo h (InterruptableProcess q)+     , HasCallStack)+  => ObservationQueue o+  -> Eff (InterruptableProcess q) (Observer o)+spawnLinkObservationQueueWriter q = do+  cbo <- startLink (MkObservationQueue q)+  pure (toObserver cbo)++instance (TangibleObserver o, TangiblePdu (Observer o) 'Asynchronous, Lifted IO q, Member Logs q) => Server (ObservationQueue o) (InterruptableProcess q) where+  type Protocol (ObservationQueue o) = Observer o++  data instance StartArgument (ObservationQueue o) (InterruptableProcess q) =+     MkObservationQueue (ObservationQueue o)++  update (MkObservationQueue (ObservationQueue q)) =+    \case+      OnRequest (Cast r) ->+        handleObservations (lift . atomically . writeTBQueue q) r+      otherMsg -> logError ("unexpected: " <> T.pack (show otherMsg))

@@ -0,0 +1,93 @@+-- | Proxies and containers for casts and calls.+--+-- @since 0.15.0+module Control.Eff.Concurrent.Protocol.Request+  ( Request(..)+  , sendReply+  , RequestOrigin(..)+  , Reply(..)+  , makeRequestOrigin+  )+  where++import Control.DeepSeq+import Control.Eff+import Control.Eff.Concurrent.Process+import Control.Eff.Concurrent.Protocol+import Data.Kind (Type)+import Data.Typeable (Typeable)+import GHC.Generics++-- | A wrapper sum type for calls and casts for the 'Pdu's of a protocol+--+-- @since 0.15.0+data Request protocol where+  Call+    :: forall protocol reply.+       ( Tangible reply+       , TangiblePdu protocol ('Synchronous reply)+       )+    => RequestOrigin protocol reply+    -> Pdu protocol ('Synchronous reply)+    -> Request protocol+  Cast+    :: forall protocol. (TangiblePdu protocol 'Asynchronous)+    => Pdu protocol 'Asynchronous+    -> Request protocol+  deriving (Typeable)++instance Show (Request protocol) where+  showsPrec d (Call o r) =+    showParen (d >= 10) (showString "call-request: " . showsPrec 11 o . showString ": " . showsPrec 11 r)+  showsPrec d (Cast r) =+    showParen (d >= 10) (showString "cast-request: " . showsPrec 11 r)++instance NFData (Request protocol) where+  rnf (Call o req) = rnf o `seq` rnf req+  rnf (Cast req) = rnf req++-- | The wrapper around replies to 'Call's.+--+-- @since 0.15.0+data Reply protocol reply where+  Reply :: (Tangible reply) => RequestOrigin protocol reply -> reply -> Reply protocol reply+  deriving (Typeable)++instance NFData (Reply p r) where+  rnf (Reply i r) = rnf i `seq` rnf r++instance Show r => Show (Reply p r) where+  showsPrec d (Reply o r) =+    showParen (d >= 10) (showString "request-reply: " . showsPrec 11 o . showString ": " . showsPrec 11 r)++-- | Wraps the source 'ProcessId' and a unique identifier for a 'Call'.+--+-- @since 0.15.0+data RequestOrigin (proto :: Type) reply = RequestOrigin+  { _requestOriginPid :: !ProcessId+  , _requestOriginCallRef :: !Int+  } deriving (Eq, Ord, Typeable, Generic)++instance Show (RequestOrigin p r) where+  showsPrec d (RequestOrigin o r) =+    showParen (d >= 10) (showString "origin: " . showsPrec 11 o . showChar ' ' . showsPrec 11 r)++-- | Create a new, unique 'RequestOrigin' value for the current process.+--+-- @since 0.24.0+makeRequestOrigin :: (SetMember Process (Process q0) e, '[Interrupts] <:: e) => Eff e (RequestOrigin p r)+makeRequestOrigin = RequestOrigin <$> self <*> makeReference++instance NFData (RequestOrigin p r)++-- | Send a 'Reply' to a 'Call'.+--+-- The reply will be deeply evaluated to 'rnf'.+--+-- @since 0.15.0+sendReply ::+     (SetMember Process (Process q) eff, Member Interrupts eff, Tangible reply, Typeable protocol)+  => RequestOrigin protocol reply+  -> reply+  -> Eff eff ()+sendReply origin reply = sendMessage (_requestOriginPid origin) (Reply origin $! force reply)

@@ -0,0 +1,416 @@+-- | A better, more safe implementation of the Erlang/OTP gen_server behaviour.+--+-- @since 0.24.0+module Control.Eff.Concurrent.Protocol.Server+  ( Server(..)+  , StartArgument(..)+  , ToServerEffects+  , ModelState+  , modifyModel+  , getAndModifyModel+  , modifyAndGetModel+  , getModel+  , putModel+  , getAndPutModel+  , useModel+  , zoomModel+  , SettingsReader+  , askSettings+  , viewSettings+  , Event(..)+  , start+  , startLink+  , protocolServerLoop+  , GenServer(..)+  , ToGenServerEffects+  , GenServerId(..)+  , GenServerProtocol+  , GenServerModel+  , GenServerSettings+  , genServer+  , Stateless+  , ToStatelessEffects+  , statelessGenServer+  -- * Re-exports+  , Request(..)+  , sendReply+  , RequestOrigin(..)+  )+  where++import Control.Applicative+import Control.DeepSeq+import Control.Eff+import Control.Eff.Extend ()+import Control.Eff.Concurrent.Process+import Control.Eff.Concurrent.Process.Timer+import Control.Eff.Concurrent.Protocol+import Control.Eff.Concurrent.Protocol.Request+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+import Data.Typeable+import Data.Type.Pretty+import qualified Data.Text as T+import GHC.Stack (HasCallStack)+import GHC.Generics++-- | A type class for 'Pdu' values that have an implementation+-- which handles the corresponding protocol.+--+-- @since 0.24.0+class+  (Typeable (Protocol a)) =>+      Server (a :: Type) e+  where+  -- | The value that defines what is required to initiate a 'Server'+  -- loop.+  data StartArgument a e+  -- | The index type of the 'Event's that this server processes.+  -- This is the first parameter to the 'Request' and therefore of+  -- the 'Pdu' family.+  type Protocol a :: Type+  type Protocol a = a+  -- | Type of the /model/ data, given to every invocation of 'update'+  -- 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 = ()+  -- | Type of read-only state.+  type Settings a :: Type+  type Settings a = ()++  setup ::+       StartArgument a e+    -> Eff e (Model a, Settings a)++  default setup ::+       (Default (Model a), Default (Settings a))+    => StartArgument a e+    -> Eff e (Model a, Settings a)+  setup _ = pure (def, def)++  update ::+       StartArgument a e+    -> Event (Protocol a)+    -> Eff (ToServerEffects a e) ()++-- | /Cons/ (i.e. prepend) 'ModelState' and 'SettingsReader' to an+-- effect list.+--+-- @since 0.24.0+type ToServerEffects a e =+  ModelState a ': SettingsReader a ': e++-- | The 'Eff'ect type of mutable 'Model' in a 'Server' instance.+--+-- @since 0.24.0+type ModelState a = State (Model a)++-- | Modify the 'Model' of a 'Server'.+--+-- @since 0.24.0+modifyModel :: forall a e . Member (ModelState a) e => (Model a -> Model a) -> Eff e ()+modifyModel f = getModel @a >>= putModel @a . f++-- | Modify the 'Model' of a 'Server' and return the old value.+--+-- @since 0.24.0+getAndModifyModel :: forall a e . Member (ModelState a) e => (Model a -> Model a) -> Eff e (Model a)+getAndModifyModel f = getModel @a <* modify f++-- | Modify the 'Model' of a 'Server' and return the new value.+--+-- @since 0.24.0+modifyAndGetModel :: forall a e . Member (ModelState a) e => (Model a -> Model a) -> Eff e (Model a)+modifyAndGetModel f = modifyModel @a f *> getModel @a++-- | Return the 'Model' of a 'Server'.+--+-- @since 0.24.0+getModel :: forall a e . Member (ModelState a) e => Eff e (Model a)+getModel = get++-- | Return a element selected by a 'Lens' of the 'Model' of a 'Server'.+--+-- @since 0.24.0+useModel :: forall a b e . Member (ModelState a) e => Getting b (Model a) b -> Eff e b+useModel l = view l <$> getModel @a++-- | Overwrite the 'Model' of a 'Server'.+--+-- @since 0.24.0+putModel :: forall a e . Member (ModelState a) e => Model a -> Eff e ()+putModel = put++-- | Overwrite the 'Model' of a 'Server', return the old value.+--+-- @since 0.24.0+getAndPutModel :: forall a e . Member (ModelState a) e => Model a -> Eff e (Model a)+getAndPutModel m = getModel @a <* putModel @a m++-- | Run an action that modifies portions of the 'Model' of a 'Server' defined by the given 'Lens'.+--+-- @since 0.24.0+zoomModel :: forall a b c e. Member (ModelState a) e => Lens' (Model a) b -> Eff (State b ': e) c -> Eff e c+zoomModel l a = do+  m0 <- getModel @a+  (c, m1) <- runState (view l m0) a+  modifyModel @a (l .~ m1)+  return c++-- | The 'Eff'ect type of readonly 'Settings' in a 'Server' instance.+--+-- @since 0.24.0+type SettingsReader a = Reader (Settings a)++-- | Return the read-only 'Settings' of a 'Server'+--+-- @since 0.24.0+askSettings :: forall a e . Member (SettingsReader a) e => Eff e (Settings a)+askSettings = ask++-- | Return the read-only 'Settings' of a 'Server' as viewed through a 'Lens'+--+-- @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++-- | Execute the server loop.+--+-- @since 0.24.0+start+  :: forall a q h+  . ( Server a (InterruptableProcess q)+    , Typeable a+    , LogsTo h (InterruptableProcess q)+    , HasCallStack)+  => StartArgument a (InterruptableProcess q) -> Eff (InterruptableProcess q) (Endpoint (Protocol a))+start a = asEndpoint <$> spawn (protocolServerLoop a)++-- | Execute the server loop.+--+-- @since 0.24.0+startLink+  :: forall a q h . (Typeable a, Server a (InterruptableProcess q), LogsTo h (InterruptableProcess q), HasCallStack)+  => StartArgument a (InterruptableProcess q) -> Eff (InterruptableProcess q) (Endpoint (Protocol a))+startLink a = asEndpoint <$> spawnLink (protocolServerLoop a)++-- | Execute the server loop.+--+-- @since 0.24.0+protocolServerLoop+     :: forall q e h a+     . ( Server a e+       , SetMember Process (Process q) e+       , Member Interrupts e+       , LogsTo h e+       , Typeable a+       )+  => StartArgument a e -> Eff e ()+protocolServerLoop a = do+  (st, env) <- setup a+  _ <- runReader env (runState st (receiveSelectedLoop sel mainLoop))+  return ()+  where+    sel :: MessageSelector (Event (Protocol a))+    sel =+          OnRequest <$> selectMessage @(Request (Protocol a))+      <|> OnDown    <$> selectMessage @ProcessDown+      <|> OnTimeOut <$> selectMessage @TimerElapsed+      <|> OnMessage <$> selectAnyMessage+    handleInt i = update a (OnInterrupt i) *> pure Nothing+    mainLoop ::+         (Typeable a)+      => Either (Interrupt 'Recoverable) (Event (Protocol a))+      -> Eff (ToServerEffects a e) (Maybe ())+    mainLoop (Left i) = handleInt i+    mainLoop (Right i) = update a i *> pure Nothing++-- | Internal protocol to communicate incoming messages and other events to the+-- instances of 'Server'.+--+-- Note that this is required to receive any kind of messages in 'protocolServerLoop'.+--+-- @since 0.24.0+data Event a =+    OnRequest (Request a)+  | OnInterrupt (Interrupt 'Recoverable)+  | OnDown ProcessDown+  | OnTimeOut TimerElapsed+  | OnMessage StrictDynamic+  deriving (Show,Generic,Typeable)++instance NFData a => NFData (Event a) where+   rnf = \case+       OnRequest r      -> rnf r+       OnInterrupt r    -> rnf r+       OnDown r  -> rnf r+       OnTimeOut r -> rnf r+       OnMessage r -> r `seq` ()++type instance ToPretty (Event a) = ToPretty a <+> PutStr "event"++-- * GenServer++-- | A helper for 'Server's that are directly based on logging and 'IO': 'GenIO'+--+-- A record that contains callbacks to provide a 'Server' instance for the+-- @tag@ parameter, .+--+-- The name prefix @Gen@ indicates the inspiration from Erlang's @gen_server@ module.+--+-- @since 0.24.0+data GenServer tag e where+  MkGenServer :: LogIo e =>+      { _setupCallback :: Eff (InterruptableProcess e) (Model (GenServer tag e), Settings (GenServer tag e))+      , _updateCallback+          :: Event (GenServerProtocol tag)+          -> Eff (ToGenServerEffects tag e) ()+      } -> GenServer tag e++-- | Prepend the 'ModelState' for 'GenServerModel' and 'SettingsReader' for 'GenServerSettings' of a 'GenServer'+-- 'Server' to an effect list.+--+-- @since 0.24.0+type ToGenServerEffects tag e = ToServerEffects (GenServer tag e) (InterruptableProcess e)++-- | The name/id of a 'GenServer' for logging purposes.+--+-- @since 0.24.0+newtype GenServerId tag =+  MkGenServerId { _fromGenServerId :: T.Text }+  deriving (Typeable, NFData, Ord, Eq, IsString)++instance Show (GenServerId tag) where+  showsPrec _d (MkGenServerId x) = showString (T.unpack x)++-- | The 'Protocol' un-wrapper type function.+--+-- @since 0.24.0+type family GenServerProtocol tag++-- | Type of state for 'GenServer' based 'Server's+--+-- @since 0.24.0+type family GenServerModel tag++-- | Type of the environment for 'GenServer' based 'Server's+--+-- @since 0.24.0+type family GenServerSettings tag++instance ( Typeable (GenServerProtocol tag)+         , LogIo e )+         => Server (GenServer (tag :: Type) e) (InterruptableProcess e) where+  type Protocol (GenServer tag e) = GenServerProtocol tag+  type Model (GenServer tag e) = GenServerModel tag+  type Settings (GenServer tag e) = GenServerSettings tag+  data instance StartArgument (GenServer tag e) (InterruptableProcess e) =+        MkGenStartArgument+         { _genServerId :: GenServerId tag+         , _genServerCallbacks :: GenServer tag e+         } deriving Typeable+  setup (MkGenStartArgument _ cb) = _setupCallback cb+  update (MkGenStartArgument _ cb) req = _updateCallback cb req++instance NFData (StartArgument (GenServer tag e) (InterruptableProcess e)) where+  rnf (MkGenStartArgument x _) = rnf x++instance Typeable tag => Show (StartArgument (GenServer tag e) (InterruptableProcess e)) where+  showsPrec d (MkGenStartArgument x _) =+    showParen (d>=10)+      ( showsPrec 11 x+      . showChar ' ' . showsTypeRep (typeRep (Proxy @tag))+      . showString " gen-server"+      )++-- ** 'GenServer' based Server constructors+++-- | Create a 'GenServer'.+--+-- This requires the callback for 'Event's, a initial 'Model' and a 'GenServerId'.+--+-- There must be a 'GenServerModel' instance.+-- There must be a 'GenServerSettings' instance.+-- There must be a 'GenServerProtocol' instance.+--+-- This is Haskell, so if this functions is partially applied+-- to some 'Event' callback, you get a function back,+-- that generates 'StartArgument's from 'GenServerId's, like a /factory/+--+-- @since 0.24.0+genServer+  :: forall tag e  .+     ( Typeable tag+     , HasCallStack+     , LogIo e+     , Server (GenServer tag e) (InterruptableProcess e)+     )+  => (GenServerId tag -> Eff (InterruptableProcess e) (GenServerModel tag, GenServerSettings tag))+  -> (GenServerId tag -> Event (GenServerProtocol tag) -> Eff (ToGenServerEffects tag e) ())+  -> GenServerId tag+  -> StartArgument (GenServer tag e) (InterruptableProcess e)+genServer initCb stepCb i =+  MkGenStartArgument+    { _genServerId = i+    , _genServerCallbacks =+        MkGenServer { _setupCallback = initCb i+                    , _updateCallback = stepCb i . coerce+                    }+    }++-- | The type-level tag indicating a stateless 'Server' instance.+--+-- There are 'GenServerModel', 'GenServerSettings' and 'GenServerProtocol' as well as+-- 'ToPretty' instances for this type.+--+-- See also 'ToStatelessEffects'.+--+-- @since 0.24.0+data Stateless tag deriving Typeable++-- | Prepend the 'ModelState' and 'SettingsReader' of a 'Stateless'+-- 'Server' to an effect list. The 'Model' and 'Settings' of a 'Stateless'+-- 'Server' are just @()@ /unit/.+--+-- @since 0.24.0+type ToStatelessEffects e = State () ': Reader () ': e++type instance GenServerModel (Stateless tag) = ()+type instance GenServerSettings (Stateless tag) = ()+type instance GenServerProtocol (Stateless tag) = GenServerProtocol tag+type instance ToPretty (Stateless t) = ToPretty t++-- | Create a 'Stateless' 'GenServer'.+--+-- This requires only the callback for 'Event's+-- and a 'GenServerId'.+--+-- This is Haskell, so if this functions is partially applied+-- to some 'Event' callback, you get a function back,+-- that generates 'StartArgument's from 'GenServerId's, like a /factory/+--+-- @since 0.24.0+statelessGenServer+  :: forall tag e . ( Typeable tag+     , HasCallStack+     , LogIo e+     , Typeable tag+     , Server (GenServer (Stateless tag) e) (InterruptableProcess e)+     )+  => (GenServerId tag -> Event (GenServerProtocol tag) -> Eff (ToStatelessEffects (InterruptableProcess e)) ())+  -> GenServerId tag+  -> StartArgument (GenServer (Stateless tag) e) (InterruptableProcess e)+statelessGenServer stepCb (MkGenServerId i) =+  genServer (const (pure ((), ()))) runStep (MkGenServerId i)+   where+    runStep :: GenServerId (Stateless tag) -> Event (GenServerProtocol (Stateless tag)) -> Eff (ToStatelessEffects (InterruptableProcess e)) ()+    runStep (MkGenServerId i') loopEvent = stepCb (MkGenServerId i') (coerce loopEvent)

@@ -0,0 +1,439 @@+{-# 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+  ( Sup()+  , ChildId+  , StartArgument(MkSupConfig)+  , supConfigChildStopTimeout+  , SpawnErr(AlreadyStarted)+  , startSupervisor+  , stopSupervisor+  , isSupervisorAlive+  , monitorSupervisor+  , getDiagnosticInfo+  , spawnChild+  , lookupChild+  , stopChild+  ) 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.Request+import Control.Eff.Concurrent.Protocol.Server as Server+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 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+-- ** 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 deriving Typeable++-- | 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 (Protocol p))))+        StopC :: ChildId p -> Timeout -> Pdu (Sup p) ('Synchronous Bool)+        LookupC :: ChildId p -> Pdu (Sup p) ('Synchronous (Maybe (Endpoint (Protocol 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+  ( Lifted IO q, LogsTo IO q+  , TangibleSup p+  , Tangible (ChildId p)+  , Server p (InterruptableProcess q)+  ) => Server (Sup p) (InterruptableProcess 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) (InterruptableProcess q) = MkSupConfig+    {+      -- , supConfigChildRestartPolicy :: ChildRestartPolicy+      -- , supConfigResilience :: Resilience+      supConfigChildStopTimeout :: Timeout+    , supConfigStartFun :: ChildId p -> Server.StartArgument p (InterruptableProcess q)+    }++  type Model (Sup p) = Children (ChildId p) p++  setup _cfg = pure (def, ())+  update supConfig (OnRequest (Call orig req)) =+    case req of+      GetDiagnosticInfo ->  do+        p <- (pack . show <$> getChildren @(ChildId p) @p)+        sendReply orig p++      LookupC i -> do+        p <- fmap _childEndpoint <$> lookupChildById @(ChildId p) @p i+        sendReply orig p++      StopC i t -> do+        mExisting <- lookupAndRemoveChildById @(ChildId p) @p i+        case mExisting of+          Nothing -> sendReply orig False+          Just existingChild -> do+            stopOrKillChild i existingChild t+            sendReply orig True++      StartC i -> do+        childEp <- raise (raise (Server.start (supConfigStartFun supConfig i)))+        let childPid = _fromEndpoint childEp+        cMon <- monitor childPid+        mExisting <- lookupChildById @(ChildId p) @p i+        case mExisting of+          Nothing -> do+            putChild i (MkChild @p childEp cMon)+            sendReply orig (Right childEp)+          Just existingChild ->+            sendReply orig (Left (AlreadyStarted i (existingChild ^. childEndpoint)))++  update _supConfig (OnDown (ProcessDown mrChild reason)) = do+      oldEntry <- lookupAndRemoveChildByMonitor @(ChildId p) @p mrChild+      case oldEntry of+        Nothing ->+          logWarning ("unexpected process down: "+                     <> pack (show mrChild)+                     <> " reason: "+                     <> pack (show reason)+                     )+        Just (i, c) -> do+          logInfo (  "process down: "+                  <> pack (show mrChild)+                  <> " reason: "+                  <> pack (show reason)+                  <> " for child "+                  <> pack (show i)+                  <> " => "+                  <> pack (show (_childEndpoint c))+                  )+  update supConfig (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 (Protocol p))+  deriving (Typeable, Generic)++deriving instance Eq (ChildId p) => Eq (SpawnErr p)+deriving instance Ord (ChildId p) => Ord (SpawnErr p)+deriving instance (Typeable (Protocol p), Show (ChildId p)) => Show (SpawnErr p)++instance NFData (ChildId p) => NFData (SpawnErr p)+++-- ** 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+    , LogsTo IO (InterruptableProcess e)+    , Lifted IO e+    , TangibleSup p+    , Server (Sup p) (InterruptableProcess e)+    )+  => StartArgument (Sup p) (InterruptableProcess e)+  -> Eff (InterruptableProcess e) (Endpoint (Sup p))+startSupervisor = Server.start++-- | Stop the supervisor and shutdown all processes.+--+-- Block until the supervisor has finished.+--+-- @since 0.23.0+stopSupervisor+  :: ( HasCallStack+     , Member Interrupts e+     , SetMember Process (Process q0) e+     , 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 Interrupts e+     , Member Logs e+     , Typeable p+     , SetMember Process (Process q0) e+     )+  => 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 Interrupts e+     , Member Logs e+     , SetMember Process (Process q0) e+     , TangibleSup p+     )+  => Endpoint (Sup p)+  -> Eff e MonitorReference+monitorSupervisor x = monitor (_fromEndpoint x)++-- | Start, link and monitor a new child process using the 'SpawnFun' passed+-- to 'startSupervisor'.+--+-- @since 0.23.0+spawnChild+  :: forall p q0 e .+     ( HasCallStack+     , Member Interrupts e+     , Member Logs e+     , SetMember Process (Process q0) e+     , TangibleSup p+     , Typeable (Protocol p)+     )+  => Endpoint (Sup p)+  -> ChildId p+  -> Eff e (Either (SpawnErr p) (Endpoint (Protocol p)))+spawnChild ep cId = call ep (StartC cId)++-- | 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 Interrupts e+     , Member Logs e+     , SetMember Process (Process q0) e+     , TangibleSup p+     , Typeable (Protocol p)+     )+  => Endpoint (Sup p)+  -> ChildId p+  -> Eff e (Maybe (Endpoint (Protocol 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 Interrupts e+     , Member Logs e+     , SetMember Process (Process q0) e+     , 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+     , Member Interrupts e+     , SetMember Process (Process q0) e+     , TangibleSup p+     )+  => Endpoint (Sup p)+  -> Eff e Text+getDiagnosticInfo s = call s (GetDiagnosticInfo @p)++-- Internal Functions++stopOrKillChild+  :: forall p e q0 .+     ( HasCallStack+     , SetMember Process (Process q0) e+     , Member Interrupts e+     , Lifted IO e+     , Lifted IO q0+     , Member Logs e+     , Member (State (Children (ChildId p) p)) e+     , TangibleSup p+     , Typeable (Protocol 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 )+        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+     , SetMember Process (Process q0) e+     , Lifted IO e+     , Lifted IO q0+     , Member Logs e+     , Member (State (Children (ChildId p) p)) e+     , TangibleSup p+     , Typeable (Protocol 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))

@@ -0,0 +1,105 @@+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.Server+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 (Protocol p)+  , _childMonitoring :: MonitorReference+  }+  deriving (Generic, Typeable, Eq, Ord)++instance NFData (Child o)++instance Typeable (Protocol 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

@@ -111,6 +111,7 @@     -- ** 'Logs' Effect Handling   , Logs()   , LogsTo+  , LogIo   , withLogging   , withSomeLogging 

@@ -48,6 +48,7 @@   -- ** 'Logs' Effect Handling   , Logs()   , LogsTo+  , LogIo   , withLogging   , withSomeLogging @@ -176,7 +177,12 @@ -- * 'withLogging' -- * 'withSomeLogging' ---type LogsTo h e = (Member Logs e, HandleLogWriter h, LogWriterEffects h <:: e, SetMember LogWriterReader (LogWriterReader h) e)+type LogsTo h e = (Member Logs e, HandleLogWriter h e, SetMember LogWriterReader (LogWriterReader h) e)++-- | A constraint that required @'LogsTo' 'IO' e@ and @'Lifted' 'IO' e@.+--+-- @since 0.24.0+type LogIo e = (LogsTo IO e, Lifted IO e)  -- | Handle the 'Logs' and 'LogWriterReader' effects. --

@@ -141,28 +141,23 @@ -- * 'LogWriter' Zoo  -- | The instances of this class are the monads that define (side-) effect(s) of writting logs.-class HandleLogWriter (writerEff :: Type -> Type) where-  -- | A list of effects that are required for writing the log messages.-  -- For example 'Lift IO' or '[]' for pure log writers.-  type LogWriterEffects writerEff :: [Type -> Type]+class HandleLogWriter (writerEff :: Type -> Type) e where    -- | Run the side effect of a 'LogWriter' in a compatible 'Eff'.-  handleLogWriterEffect :: (LogWriterEffects writerEff <:: e) => writerEff () -> Eff e ()+  handleLogWriterEffect :: writerEff () -> Eff e ()    -- | Write a message using the 'LogWriter' found in the environment.   --   -- The semantics of this function are a combination of 'runLogWriter' and 'handleLogWriterEffect',   -- with the 'LogWriter' read from a 'LogWriterReader'.-  liftWriteLogMessage :: ( SetMember LogWriterReader (LogWriterReader writerEff) e-                         , LogWriterEffects writerEff <:: e)+  liftWriteLogMessage :: ( SetMember LogWriterReader (LogWriterReader writerEff) e)                       => LogMessage                       -> Eff e ()   liftWriteLogMessage m = do     w <- askLogWriter     handleLogWriterEffect (runLogWriter w m) -instance HandleLogWriter IO where-  type LogWriterEffects IO = '[Lift IO]+instance (Lifted IO e) => HandleLogWriter IO e where   handleLogWriterEffect = send . Lift  -- ** Pure Log Writers@@ -172,8 +167,7 @@   deriving (Applicative, Functor, Monad)  -- | A 'LogWriter' monad for 'Debug.Trace' based pure logging.-instance HandleLogWriter PureLogWriter where-  type LogWriterEffects PureLogWriter = '[]+instance HandleLogWriter PureLogWriter e where   handleLogWriterEffect lw = return (force (runIdentity (force (runPureLogWriter lw))))  -- | This 'LogWriter' will discard all messages.

@@ -1,3 +1,4 @@+{-# LANGUAGE UndecidableInstances #-} -- | Capture 'LogMessage's to a 'Writer'. -- -- See 'Control.Eff.Log.Examples.exampleLogCapture'@@ -30,8 +31,7 @@ -- | A 'LogWriter' monad for pure logging. -- -- The 'HandleLogWriter' instance for this type assumes a 'Writer' effect.-instance HandleLogWriter CaptureLogs where-  type LogWriterEffects CaptureLogs = '[CaptureLogWriter]+instance Member CaptureLogWriter e => HandleLogWriter CaptureLogs e where   handleLogWriterEffect =     traverse_ (tell @LogMessage) . snd . run . runListWriter . unCaptureLogs 

@@ -3,12 +3,12 @@ import Control.Concurrent import Control.Concurrent.STM import Control.Eff-import Control.Eff.Concurrent.Process+import Control.Eff.Concurrent hiding (Timeout)+import Control.Eff.Concurrent.Process.ForkIOScheduler as Scheduler import Control.Eff.Extend-import Control.Eff.Log import Control.Monad (void) import GHC.Stack-import Test.Tasty+import Test.Tasty  import Test.Tasty.HUnit import Test.Tasty.Runners @@ -17,6 +17,13 @@  timeoutSeconds :: Integer -> Timeout timeoutSeconds seconds = Timeout (seconds * 1000000) (show seconds ++ "s")++runTestCase :: TestName -> Eff InterruptableProcEff () -> TestTree+runTestCase msg =+  testCase msg .+  runLift . withTraceLogging "unit-tests" local0 allLogMessages . Scheduler.schedule . handleInterrupts onInt+  where+    onInt = lift . assertFailure . show  withTestLogC :: (e -> IO ()) -> (IO (e -> IO ()) -> TestTree) -> TestTree withTestLogC doSchedule k = k (return doSchedule)

@@ -0,0 +1,111 @@++module GenServerTests+  ( test_genServer+  ) 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.Server as Server+import Control.Lens+import Data.Coerce+import Data.Text as T+import Data.Type.Pretty+import Data.Typeable (Typeable)+import Test.Tasty+import Test.Tasty.HUnit++-- ------------------------------++data Small deriving Typeable++type instance ToPretty Small = PutStr "small"++data instance  Pdu Small r where+        SmallCall :: Bool -> Pdu Small ('Synchronous Bool)+        SmallCast :: String -> Pdu Small 'Asynchronous+    deriving Typeable++instance NFData (Pdu Small r) where+  rnf (SmallCall x) = rnf x+  rnf (SmallCast x) = rnf x++instance Show (Pdu Small r) where+  showsPrec d (SmallCall x) = showParen (d > 10) (showString "SmallCall " . shows x)+  showsPrec d (SmallCast x) = showParen (d > 10) (showString "SmallCast " . showString x)+++-- ----------------------------------------------------------------------------++instance LogIo e => Server Small (InterruptableProcess e) where+  data StartArgument Small (InterruptableProcess e) = MkSmall+  type Model Small = String+  update MkSmall x =+    case x of+      OnRequest msg ->+       logInfo' (show msg)+      other ->+        interrupt (ErrorInterrupt (show other))++-- ----------------------------------------------------------------------------++data Big deriving (Typeable)++type instance ToPretty Big = PutStr "big"++data instance  Pdu Big r where+        BigCall :: Bool -> Pdu Big ('Synchronous Bool)+        BigCast :: String -> Pdu Big 'Asynchronous+        BigSmall :: Pdu Small r -> Pdu Big r+    deriving Typeable++instance NFData (Pdu Big r) where+  rnf (BigCall x) = rnf x+  rnf (BigCast x) = rnf x+  rnf (BigSmall x) = rnf x++instance Show (Pdu Big r) where+  showsPrec d (BigCall x) = showParen (d > 10) (showString "SmallCall " . shows x)+  showsPrec d (BigCast x) = showParen (d > 10) (showString "SmallCast " . showString x)+  showsPrec d (BigSmall x) = showParen (d > 10) (showString "BigSmall " . showsPrec 11 x)++instance EmbedProtocol Big Small where+  embeddedPdu =+    prism'+      BigSmall+      (\case+         (BigSmall x) -> Just x+         _ -> Nothing)++-- ----------------------------------------------------------------------------++instance LogIo e => Server Big (InterruptableProcess e) where+  data instance StartArgument Big (InterruptableProcess e) = MkBig+  type Model Big = String+  update MkBig = \case+    OnRequest msg ->+      case msg of+        Call orig req ->+          case req of+            BigCall o -> do+              logNotice ("BigCall " <> pack (show o))+              sendReply orig o+            BigSmall x -> update MkSmall (OnRequest (Call (coerce orig) x))+        Cast req ->+          case req of+            BigCast o -> putModel @Big o+            BigSmall x -> update MkSmall (OnRequest (Cast 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 <- start 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+  ]

@@ -6,9 +6,9 @@ import           Control.Concurrent.STM import           Control.Eff.Concurrent.Process import           Control.Eff.Concurrent.Process.Timer-import           Control.Eff.Concurrent.Api-import           Control.Eff.Concurrent.Api.Client-import           Control.Eff.Concurrent.Api.Server+import           Control.Eff.Concurrent.Protocol+import           Control.Eff.Concurrent.Protocol.Client+import           Control.Eff.Concurrent.Protocol.Server import qualified Control.Eff.Concurrent.Process.ForkIOScheduler                                                as ForkIO import qualified Control.Eff.Concurrent.Process.SingleThreadedScheduler@@ -63,7 +63,7 @@  allTests   :: forall r-   . (Lifted IO r, LogsTo IO r)+   . (Lifted IO r, LogsTo IO r, Typeable r)   => IO (Eff (InterruptableProcess r) () -> IO ())   -> TestTree allTests schedulerFactory = localOption@@ -88,22 +88,22 @@  type instance ToPretty ReturnToSender = PutStr "ReturnToSender" -data instance Api ReturnToSender r where-  ReturnToSender :: ProcessId -> String -> Api ReturnToSender ('Synchronous Bool)-  StopReturnToSender :: Api ReturnToSender ('Synchronous ())+data instance Pdu ReturnToSender r where+  ReturnToSender :: ProcessId -> String -> Pdu ReturnToSender ('Synchronous Bool)+  StopReturnToSender :: Pdu ReturnToSender ('Synchronous ()) -instance NFData (Api ReturnToSender r) where+instance NFData (Pdu ReturnToSender r) where   rnf (ReturnToSender p s) = rnf p `seq` rnf s   rnf StopReturnToSender = () -deriving instance Show (Api ReturnToSender x)+deriving instance Show (Pdu ReturnToSender x) -deriving instance Typeable (Api ReturnToSender x)+deriving instance Typeable (Pdu ReturnToSender x)  returnToSender   :: forall q r    . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)-  => Server ReturnToSender+  => Endpoint ReturnToSender   -> String   -> Eff r Bool returnToSender toP msg = do@@ -115,30 +115,38 @@ stopReturnToSender   :: forall q r    . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)-  => Server ReturnToSender+  => Endpoint ReturnToSender   -> Eff r () stopReturnToSender toP = call toP StopReturnToSender +type instance GenServerProtocol ReturnToSender = ReturnToSender+ returnToSenderServer   :: forall q-   . (HasCallStack, Member Logs q)-  => Eff (InterruptableProcess q) (Server ReturnToSender)-returnToSenderServer = spawnApiServer-  (handleCalls-    (\m k -> k $ case m of-      StopReturnToSender -> do-        return (Nothing, StopServer testInterruptReason)-      ReturnToSender fromP echoMsg -> do-        sendMessage fromP echoMsg-        yieldProcess-        return (Just True, AwaitNext)+   . (HasCallStack, Lifted IO q, LogsTo IO q, Member Logs q, Typeable q)+  => Eff (InterruptableProcess q) (Endpoint ReturnToSender)+returnToSenderServer = start+  (statelessGenServer @ReturnToSender+    (\_me evt ->+      case evt of+        OnRequest (Call orig msg) ->+          case msg of+            StopReturnToSender -> interrupt testInterruptReason+            ReturnToSender fromP echoMsg -> do+              sendMessage fromP echoMsg+              yieldProcess+              sendReply orig True+        OnInterrupt i ->+          interrupt i+        other -> interrupt (ErrorInterrupt (show other))     )+    "return-to-sender"   )-  stopServerOnInterrupt + selectiveReceiveTests   :: forall r-   . (Lifted IO r, LogsTo IO r)+   . (Lifted IO r, LogsTo IO r, Typeable r)   => IO (Eff (InterruptableProcess r) () -> IO ())   -> TestTree selectiveReceiveTests schedulerFactory = setTravisTestOptions

@@ -1,112 +0,0 @@-module ServerForkIO where--import           Control.Concurrent-import           Control.Eff.Extend-import           Control.Eff.State.Strict-import           Control.Eff.Concurrent-import           Control.Eff.Concurrent.Process.ForkIOScheduler-                                               as Scheduler-import           Control.Monad                  ( )-import           Test.Tasty                    hiding (Timeout)-import           Test.Tasty.HUnit-import           Common-import           Data.Typeable                 hiding ( cast )-import           Data.Type.Pretty-import           Control.DeepSeq--data TestServer deriving Typeable-type instance ToPretty TestServer = PutStr "test"--data instance Api TestServer x where-  TestGetStringLength :: String -> Api TestServer ('Synchronous Int)-  TestSetNextDelay :: Timeout -> Api TestServer 'Asynchronous-  deriving Typeable--instance NFData (Api TestServer x) where-  rnf (TestGetStringLength x) = rnf x-  rnf (TestSetNextDelay x) = rnf x--test_ServerCallTimeout :: TestTree-test_ServerCallTimeout =-  setTravisTestOptions-    $ testCase "Server call timeout"-    $ do-         runLift . withTraceLogging "test" local0 allLogMessages . Scheduler.schedule-               $ do  let backendDelay  = 100000-                         clientTimeout = 1000-                     s <- spawnBackend-                     linkProcess (_fromServer s)-                     logInfo "====================== All Servers Started ========================"-                     cast s (TestSetNextDelay backendDelay)-                     let testData = "this is a nice string" :: String-                         expected = 2 * length testData-                         testAction = callWithTimeout-                                       s-                                       (TestGetStringLength testData)--                     actual <- handleInterrupts-                                  (\i -> do-                                      logNotice' ("caught interrupt: " ++ show i)-                                      logNotice "adapting server delay"-                                      cast s (TestSetNextDelay clientTimeout)-                                      res <- testAction backendDelay-                                      logNotice "Yeah!! Got a result..."-                                      return (res * 2)-                                  ) -                                  (testAction clientTimeout)-                     lift (expected @=? actual)-         threadDelay 1000---spawnRelay :: Timeout -> Server TestServer -> Eff InterruptableProcEff (Server TestServer)-spawnRelay callTimeout target = spawnApiServer hm hi-  where-    hi = InterruptCallback $ \i -> do-            logAlert' (show i)-            pure AwaitNext-    hm :: MessageCallback TestServer InterruptableProcEff-    hm = handleCastsAndCalls onCast onCall-      where-        onCast :: Api TestServer 'Asynchronous -> Eff InterruptableProcEff (CallbackResult 'Recoverable)-        onCast (TestSetNextDelay x) = do-          logDebug "relaying delay"-          cast target (TestSetNextDelay x)-          pure AwaitNext--        onCall :: (NFData l, Typeable l)-               => Api TestServer ('Synchronous l)-               -> (Eff InterruptableProcEff (Maybe l, CallbackResult 'Recoverable) -> k)-               -> k-        onCall (TestGetStringLength s) runHandler = runHandler $ do-          logDebug "relaying get string length"-          l <- callWithTimeout target (TestGetStringLength s) callTimeout-          logDebug' ("got result: " ++ show l)-          return (Just l, AwaitNext)---spawnBackend :: Eff InterruptableProcEff (Server TestServer)-spawnBackend = spawnApiServerStateful (return (1000000 :: Timeout)) hm hi-  where-    hi = InterruptCallback $ \i -> do-            logAlert' (show i)-            pure AwaitNext-    hm :: MessageCallback TestServer (State Timeout ': InterruptableProcEff)-    hm = handleCastsAndCalls onCast onCall-      where-        onCast :: Api TestServer 'Asynchronous -> Eff (State Timeout ': InterruptableProcEff) (CallbackResult 'Recoverable)-        onCast (TestSetNextDelay x) = do-          logDebug' ("setting delay: " ++ show x)-          put x-          pure AwaitNext--        onCall :: (NFData l, Typeable l)-               => Api TestServer ('Synchronous l)-               -> (Eff (State Timeout ': InterruptableProcEff) (Maybe l, CallbackResult 'Recoverable) -> k)-               -> k-        onCall (TestGetStringLength s) runHandler = runHandler $ do-          logDebug' ("calculating string length: " ++ show s)-          t <- get-          logDebug' ("sleeping for: " ++ show t)-          lift (threadDelay (fromTimeoutMicros t))-          logDebug "sending result"-          return (Just (length s), AwaitNext)

@@ -7,8 +7,8 @@ import Control.DeepSeq import Control.Eff import Control.Eff.Concurrent-import Control.Eff.Concurrent.Api.Supervisor as Sup-import Control.Eff.Concurrent.Process.ForkIOScheduler as Scheduler+import Control.Eff.Concurrent.Protocol.Server as Server+import Control.Eff.Concurrent.Protocol.Supervisor as Sup import Control.Eff.Concurrent.Process.Timer import Data.Either (fromRight, isLeft, isRight) import Data.Maybe (fromMaybe)@@ -24,7 +24,7 @@   testGroup     "Supervisor"     (let startTestSup = startTestSupWith ExitWhenRequested (TimeoutMicros 500000)-         startTestSupWith e t = Sup.startSupervisor (MkSupConfig (spawnTestApiProcess e) t)+         startTestSupWith e t = Sup.startSupervisor (MkSupConfig t (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@@ -37,7 +37,7 @@                  () <- receiveMessage                  Sup.stopSupervisor sup              unlinkProcess testWorker-             sup <- receiveMessage :: Eff InterruptableProcEff (Sup.Sup Int (Server TestApi))+             sup <- receiveMessage :: Eff InterruptableProcEff  (Endpoint (Sup.Sup TestProtocol))              supAliveAfter1 <- isSupervisorAlive sup              logInfo ("still alive 1: " <> pack (show supAliveAfter1))              lift (supAliveAfter1 @=? True)@@ -80,7 +80,7 @@                  [ runTestCase "When a supervisor is shut down, all children are shutdown" $ do                      sup <- startTestSup                      child <- spawnTestChild sup childId-                     let childPid = _fromServer child+                     let childPid = _fromEndpoint child                      supMon <- monitorSupervisor sup                      childMon <- monitor childPid                      isProcessAlive childPid >>= lift . assertBool "child process not running"@@ -112,7 +112,7 @@                      "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 = _fromServer child+                     let childPid = _fromEndpoint child                      supMon <- monitorSupervisor sup                      childMon <- monitor childPid                      isProcessAlive childPid >>= lift . assertBool "child process not running"@@ -166,14 +166,14 @@                      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 (_fromServer c)+                     childStillRunning <- isProcessAlive (_fromEndpoint c)                      lift (assertBool "child not running" childStillRunning)-                     someOtherChildStillRunning <- isProcessAlive (_fromServer someOtherChild)+                     someOtherChildStillRunning <- isProcessAlive (_fromEndpoint someOtherChild)                      lift (assertBool "someOtherChild not running" someOtherChildStillRunning)                  , let startTestSupAndChild = do                          sup <- startTestSup                          c <- spawnTestChild sup i-                         cm <- monitor (_fromServer c)+                         cm <- monitor (_fromEndpoint c)                          return (sup, cm)                     in testGroup                          "Stopping children"@@ -186,7 +186,7 @@                              "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 (_fromServer c)+                             cm <- monitor (_fromEndpoint c)                              Sup.stopChild sup i >>= lift . assertBool "child not found"                              (ProcessDown _ r) <- receiveSelectedMessage (selectProcessDown cm)                              case r of@@ -208,7 +208,7 @@                  , let startTestSupAndChild = do                          sup <- startTestSup                          c <- spawnTestChild sup i-                         cm <- monitor (_fromServer c)+                         cm <- monitor (_fromEndpoint c)                          return (sup, c, cm)                     in testGroup                          "Child exit handling"@@ -227,13 +227,13 @@                              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 (_fromServer c) NormalExitRequested+                             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 (_fromServer c) ExitProcessCancelled+                             sendShutdown (_fromEndpoint c) ExitProcessCancelled                              (ProcessDown _ _) <- receiveSelectedMessage (selectProcessDown cm)                              x <- Sup.lookupChild sup i                              lift (assertEqual "lookup should not find a child" Nothing x)@@ -241,57 +241,48 @@                  ]          ]) -runTestCase :: TestName -> Eff InterruptableProcEff () -> TestTree-runTestCase msg =-  testCase msg .-  runLift . withTraceLogging "supervisor-test" local0 allLogMessages . Scheduler.schedule . handleInterrupts onInt-  where-    onInt = lift . assertFailure . show--data TestApiServerMode-  = IgnoreNormalExitRequest-  | ExitWhenRequested-  deriving (Eq)--spawnTestApiProcess :: TestApiServerMode -> Sup.SpawnFun Int InterruptableProcEff (Server TestApi)-spawnTestApiProcess testMode tId =-  spawnApiServer (handleCasts onCast <> handleCalls onCall ^: handleAnyMessages onInfo) (InterruptCallback onInterrupt)-  where-    onCast ::-         Api TestApi 'Asynchronous -> Eff InterruptableProcEff (CallbackResult 'Recoverable)-    onCast (TestInterruptWith i) = do-      logInfo (pack (show tId) <> ": stopping with: " <> pack (show i))-      pure (StopServer i)-    onCall ::-         Api TestApi ('Synchronous r) -> (Eff InterruptableProcEff (Maybe r, CallbackResult 'Recoverable) -> x) -> x-    onCall (TestGetStringLength str) runMe =-      runMe $ do-        logInfo (pack (show tId) <> ": calculating length of: " <> pack str)-        pure (Just (length str), AwaitNext)-    onInfo :: StrictDynamic -> Eff InterruptableProcEff (CallbackResult 'Recoverable)-    onInfo sd = do-      logDebug (pack (show tId) <> ": got some info: " <> pack (show sd))-      pure AwaitNext-    onInterrupt x = do-      logNotice (pack (show tId) <> ": " <> pack (show x))-      if testMode == IgnoreNormalExitRequest-        then do-          logNotice $ pack (show tId) <> ": ignoring normal exit request"-          pure AwaitNext-        else do-          logNotice $ pack (show tId) <> ": exitting normally"-          pure (StopServer (interruptToExit x))--data TestApi+data TestProtocol   deriving (Typeable) -type instance ToPretty TestApi = PutStr "test"+type instance ToPretty TestProtocol = PutStr "test" -data instance  Api TestApi x where-        TestGetStringLength :: String -> Api TestApi ('Synchronous Int)-        TestInterruptWith :: Interrupt 'Recoverable -> Api TestApi 'Asynchronous+data instance  Pdu TestProtocol x where+        TestGetStringLength :: String -> Pdu TestProtocol ('Synchronous Int)+        TestInterruptWith :: Interrupt 'Recoverable -> Pdu TestProtocol 'Asynchronous     deriving Typeable -instance NFData (Api TestApi x) where+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 Server TestProtocol InterruptableProcEff where+  update (TestServerArgs testMode tId) evt =+    case evt of+      OnRequest (Cast (TestInterruptWith i)) -> do+        logInfo (pack (show tId) <> ": stopping with: " <> pack (show i))+        interrupt i+      OnRequest (Call orig (TestGetStringLength str)) -> do+        logInfo (pack (show tId) <> ": calculating length of: " <> pack str)+        sendReply orig (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 InterruptableProcEff = TestServerArgs TestProtocolServerMode Int++type instance ChildId TestProtocol = Int