extensible-effects-concurrent 0.27.0 → 0.28.0

raw patch · 23 files changed

+1124/−623 lines, 23 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• ChangeLog.md +8/−0
• examples/example-2/Main.hs +17/−15
• extensible-effects-concurrent.cabal +3/−1
• src/Control/Eff/Concurrent.hs +39/−57
• src/Control/Eff/Concurrent/Misc.hs +49/−0
• src/Control/Eff/Concurrent/Process.hs +116/−75
• src/Control/Eff/Concurrent/Process/ForkIOScheduler.hs +5/−6
• src/Control/Eff/Concurrent/Process/SingleThreadedScheduler.hs +3/−3
• src/Control/Eff/Concurrent/Process/Timer.hs +6/−12
• src/Control/Eff/Concurrent/Protocol.hs +52/−60
• src/Control/Eff/Concurrent/Protocol/CallbackServer.hs +5/−5
• src/Control/Eff/Concurrent/Protocol/Client.hs +30/−34
• src/Control/Eff/Concurrent/Protocol/EffectfulServer.hs +15/−15
• src/Control/Eff/Concurrent/Protocol/Observer.hs +229/−195
• src/Control/Eff/Concurrent/Protocol/Observer/Queue.hs +145/−72
• src/Control/Eff/Concurrent/Protocol/StatefulServer.hs +6/−4
• src/Control/Eff/Concurrent/Protocol/Supervisor.hs +13/−29
• src/Control/Eff/Concurrent/Protocol/Wrapper.hs +2/−8
• test/Common.hs +7/−1
• test/GenServerTests.hs +3/−3
• test/ObserverTests.hs +343/−0
• test/ProcessBehaviourTestCases.hs +6/−6
• test/SupervisorTests.hs +22/−22

@@ -1,5 +1,13 @@ # Changelog for extensible-effects-concurrent +## 0.28.0+- Simplify `Protocol.Observer` registration API+- Rewrite `Protocol.Observer.Queue` API+- Add the `ProcessId` to the `ProcessDown` message++## 0.27.1+- Introduce `HasProcesses` and `HasSafeProcesses` everywhere+ ## 0.27.0 - Improve/fix `EmbedProtocol` type class - Add a _this_ like parameter to the methods of `EffectfulServer`

@@ -41,7 +41,7 @@   lift (threadDelay 500000)   o <- logCounterObservations   lift (threadDelay 500000)-  registerObserver o c+  registerObserver @CounterChanged c o   lift (threadDelay 500000)   cast c Inc   lift (threadDelay 500000)@@ -92,13 +92,13 @@    type instance Model SupiCounter =     ( Integer-    , Observers CounterChanged+    , ObserverRegistry CounterChanged     , Maybe (ReplyTarget SupiCounter (Maybe ()))     )    data instance StartArgument SupiCounter (Processes q) = MkEmptySupiCounter -  setup _ _ = return ((0, emptyObservers, Nothing), ())+  setup _ _ = return ((0, emptyObserverRegistry, Nothing), ())    update _ _ = \case     OnCall rt callReq ->@@ -113,14 +113,20 @@       case castReq of         ToPdu1 Inc -> do           val' <- view _1 <$> modifyAndGetModel @SupiCounter (_1 %~ (+ 1))-          zoomModel @SupiCounter _2 (observed (CounterChanged val'))+          zoomModel @SupiCounter _2 (observerRegistryNotify (CounterChanged val'))           when (val' > 5) $             getAndModifyModel @SupiCounter (_3 .~ Nothing)             >>= traverse_ (\rt' -> sendReply rt' (Just ())) . view _3         ToPdu2 x ->-          zoomModel @SupiCounter _2 (handleObserverRegistration x)+          zoomModel @SupiCounter _2 (observerRegistryHandlePdu x)         ToPdu3 _ -> error "unreachable" +    OnDown pd -> do+      wasRemoved <- zoomModel @SupiCounter _2 (observerRegistryRemoveProcess @CounterChanged (downProcess pd))+      if wasRemoved+        then logDebug ("removed: "    <> T.pack (show pd))+        else logError ("unexpected: " <> T.pack (show pd))+     other -> logWarning (T.pack (show other))  spawnCounter :: (LogIo q) => Eff (Processes q) ( Endpoint SupiCounter )@@ -131,16 +137,12 @@  logCounterObservations   :: (LogIo q, Typeable q)-  => Eff (Processes q) (Observer CounterChanged)-logCounterObservations = do-  svr <- start OCCStart-  pure (toObserver svr)+  => Eff (Processes q) (Endpoint (Observer CounterChanged))+logCounterObservations = start OCCStart  instance Member Logs q => Server (Observer CounterChanged) (Processes q) where   data StartArgument (Observer CounterChanged) (Processes q) = OCCStart-  type Model (Observer CounterChanged) = Observers CounterChanged-  setup _ _ = pure (emptyObservers, ())-  update _ _ =-    \case-      OnCast r -> handleObservations (\msg -> logInfo' ("observed: " ++ show msg)) r-      wtf -> logNotice (T.pack (show wtf))+  update _ _ e =+    case e of+      OnCast (Observed msg) -> logInfo' ("observerRegistryNotify: " ++ show msg)+      _ -> logNotice (T.pack (show e))

@@ -1,6 +1,6 @@ cabal-version:  2.0 name:           extensible-effects-concurrent-version:        0.27.0+version:        0.28.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@@ -99,6 +99,7 @@                   Control.Eff.Concurrent.Protocol.Observer                   Control.Eff.Concurrent.Protocol.Observer.Queue   other-modules:+                Control.Eff.Concurrent.Misc,                 Control.Eff.Concurrent.Protocol.Supervisor.InternalState,                 Paths_extensible_effects_concurrent   default-extensions:@@ -276,6 +277,7 @@               , LoggingTests               , LogMessageIdeaTest               , LoopTests+              , ObserverTests               , ProcessBehaviourTestCases               , SupervisorTests               , SingleThreadedScheduler

@@ -37,9 +37,6 @@     -- ** /Observer/ Functions for Events and Event Listener     module Control.Eff.Concurrent.Protocol.Observer   ,-    -- *** Capture /Observation/ in a FIFO Queue-    module Control.Eff.Concurrent.Protocol.Observer.Queue-  ,     -- * Utilities     -- ** Logging Effect     module Control.Eff.Log@@ -74,7 +71,12 @@   ) where -import           Control.Eff.Concurrent.Process ( Process(..)+import           Control.Eff.Concurrent.Process+                                                ( Process(..)+                                                , SafeProcesses+                                                , Processes+                                                , HasProcesses+                                                , HasSafeProcesses                                                 , ProcessTitle(..)                                                 , fromProcessTitle                                                 , ProcessDetails(..)@@ -86,73 +88,58 @@                                                 , Serializer(..)                                                 , ProcessId(..)                                                 , fromProcessId-                                                , SafeProcesses-                                                , ResumeProcess(..)-                                                , ProcessState(..)                                                 , yieldProcess                                                 , sendMessage                                                 , sendAnyMessage-                                                , sendShutdown-                                                , sendInterrupt                                                 , makeReference-                                                , getProcessState-                                                , updateProcessDetails-                                                , receiveAnyMessage                                                 , receiveMessage                                                 , receiveSelectedMessage                                                 , flushMessages-                                                , receiveAnyLoop+                                                , receiveAnyMessage                                                 , receiveLoop                                                 , receiveSelectedLoop-                                                , MessageSelector-                                                  ( runMessageSelector-                                                  )-                                                , selectMessage+                                                , receiveAnyLoop+                                                , MessageSelector(runMessageSelector)                                                 , selectMessage                                                 , filterMessage-                                                , filterMessage                                                 , selectMessageWith-                                                , selectMessageWith                                                 , selectDynamicMessage-                                                , selectDynamicMessage                                                 , selectAnyMessage                                                 , self                                                 , isProcessAlive+                                                , getProcessState+                                                , updateProcessDetails                                                 , spawn                                                 , spawn_                                                 , spawnLink                                                 , spawnRaw                                                 , spawnRaw_+                                                , ResumeProcess(..)+                                                , executeAndResume+                                                , executeAndResumeOrExit+                                                , executeAndResumeOrThrow+                                                , interrupt+                                                , sendInterrupt                                                 , exitBecause                                                 , exitNormally                                                 , exitWithError+                                                , sendShutdown -- TODO rename to 'sendExit'                                                 , linkProcess                                                 , unlinkProcess                                                 , monitor                                                 , demonitor                                                 , ProcessDown(..)                                                 , selectProcessDown+                                                , selectProcessDownByProcessId                                                 , becauseProcessIsDown                                                 , MonitorReference(..)                                                 , withMonitor                                                 , receiveWithMonitor-                                                , provideInterruptsShutdown-                                                , handleInterrupts-                                                , tryUninterrupted-                                                , exitOnInterrupt-                                                , logInterrupts-                                                , provideInterrupts-                                                , mergeEitherInterruptAndExitReason-                                                , interrupt-                                                , executeAndResume-                                                , executeAndResumeOrExit-                                                , executeAndResumeOrThrow                                                 , Interrupt(..)+                                                , Interrupts                                                 , interruptToExit                                                 , ExitRecovery(..)                                                 , RecoverableInterrupt-                                                , Interrupts-                                                , Processes                                                 , ExitSeverity(..)                                                 , SomeExitReason(SomeExitReason)                                                 , toExitRecovery@@ -163,6 +150,13 @@                                                 , toCrashReason                                                 , fromSomeExitReason                                                 , logProcessExit+                                                , provideInterruptsShutdown+                                                , handleInterrupts+                                                , tryUninterrupted+                                                , exitOnInterrupt+                                                , logInterrupts+                                                , provideInterrupts+                                                , mergeEitherInterruptAndExitReason                                                 ) import           Control.Eff.Concurrent.Process.Timer                                                 ( Timeout(fromTimeoutMicros)@@ -207,32 +201,20 @@                                                 ) import           Control.Eff.Concurrent.Protocol.Observer                                                 ( Observer(..)-                                                , Pdu-                                                  ( RegisterObserver-                                                  , ForgetObserver-                                                  , Observed-                                                  )+                                                , ObservationSink+                                                , IsObservable+                                                , CanObserve+                                                , Pdu(RegisterObserver, ForgetObserver, Observed)                                                 , registerObserver                                                 , forgetObserver-                                                , handleObservations-                                                , toObserver-                                                , toObserverFor-                                                , ObserverRegistry-                                                , ObserverState-                                                , Observers()-                                                , emptyObservers-                                                , handleObserverRegistration-                                                , manageObservers-                                                , observed-                                                )-import           Control.Eff.Concurrent.Protocol.Observer.Queue-                                                ( ObservationQueue()-                                                , ObservationQueueReader-                                                , readObservationQueue-                                                , tryReadObservationQueue-                                                , flushObservationQueue-                                                , withObservationQueue-                                                , spawnLinkObservationQueueWriter+                                                , forgetObserverUnsafe+                                                , ObserverRegistry(..)+                                                , ObserverRegistryState+                                                , evalObserverRegistryState+                                                , emptyObserverRegistry+                                                , observerRegistryHandlePdu+                                                , observerRegistryRemoveProcess+                                                , observerRegistryNotify                                                 ) import           Control.Eff.Concurrent.Protocol.Wrapper                                                 ( Request(..)

@@ -0,0 +1,49 @@+-- | Internal module containing internal helpers that didn't+-- make it into their own library.+module Control.Eff.Concurrent.Misc+  ( showSTypeRepPrec+  , showSTypeRep+  , showSTypeable+  , showSPrecTypeable+  )+  where++import           Data.Dynamic+import           Data.Typeable ()+import           Type.Reflection++-- | Render a 'Typeable' to a 'ShowS'.+--+-- @since 0.28.0+showSTypeable :: forall message . Typeable message => ShowS+showSTypeable = showSTypeRep (SomeTypeRep (typeRep @message))++-- | Render a 'Typeable' to a 'ShowS' with a precedence parameter.+--+-- @since 0.28.0+showSPrecTypeable :: forall message . Typeable message => Int -> ShowS+showSPrecTypeable d = showSTypeRepPrec d (SomeTypeRep (typeRep @message))++-- | This is equivalent to @'showSTypeRepPrec' 0@+--+-- @since 0.24.0+showSTypeRep :: SomeTypeRep -> ShowS+showSTypeRep = showSTypeRepPrec 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+showSTypeRepPrec :: Int -> SomeTypeRep -> ShowS+showSTypeRepPrec d (SomeTypeRep tr) sIn =+  let (con, conArgs) = splitApps tr+   in case conArgs of+        [] -> showString (tyConName con) sIn+        _ ->+          showParen+            (d >= 10)+            (showString (tyConName con) . showChar ':' .+              foldr1 (\f acc -> showChar '-' . f . acc)+                     (showSTypeRepPrec 10 <$> conArgs))+            sIn

@@ -13,9 +13,14 @@ --    "Control.Eff.Concurrent.Process.SingleThreadedScheduler" module Control.Eff.Concurrent.Process   ( -- * Process Effect-    -- ** Effect Type Handling     Process(..)-    -- ** Process Info+    -- ** Process Effect Aliases+  , SafeProcesses+  , Processes+    -- ** Process Effect Contraints+  , HasProcesses+  , HasSafeProcesses+      -- ** Process Info   , ProcessTitle(..)   , fromProcessTitle   , ProcessDetails(..)@@ -27,12 +32,9 @@   , unwrapStrictDynamic   , Serializer(..) -     -- ** ProcessId Type   , ProcessId(..)   , fromProcessId-  , SafeProcesses-  , ResumeProcess(..)   -- ** Process State   , ProcessState(..)   -- ** Yielding@@ -40,8 +42,6 @@   -- ** Sending Messages   , sendMessage   , sendAnyMessage-  , sendShutdown-  , sendInterrupt   -- ** Utilities   , makeReference   -- ** Receiving Messages@@ -59,7 +59,7 @@   , selectMessageWith   , selectDynamicMessage   , selectAnyMessage-  -- ** Process Life Cycle Management+  -- ** Process State Reflection   , self   , isProcessAlive   , getProcessState@@ -70,42 +70,39 @@   , spawnLink   , spawnRaw   , spawnRaw_-  -- ** Process Exit or Interrupt Recoverable+  -- ** Process Operation Execution+  , ResumeProcess(..)+  , executeAndResume+  , executeAndResumeOrExit+  , executeAndResumeOrThrow+  -- ** Exits and Interrupts+  -- *** Interrupting Processes+  , interrupt+  , sendInterrupt+  -- *** Exitting Processes   , exitBecause   , exitNormally   , exitWithError-  -- ** Links+  , sendShutdown -- TODO rename to 'sendExit'+  -- *** Linking Processes   , linkProcess   , unlinkProcess-  -- ** Monitors+  -- *** Monitor Processes   , monitor   , demonitor   , ProcessDown(..)   , selectProcessDown+  , selectProcessDownByProcessId   , becauseProcessIsDown   , MonitorReference(..)   , withMonitor   , receiveWithMonitor-  -- ** Process Interrupt Recoverable Handling-  , provideInterruptsShutdown-  , handleInterrupts-  , tryUninterrupted-  , exitOnInterrupt-  , logInterrupts-  , provideInterrupts-  , mergeEitherInterruptAndExitReason-  , interrupt-  -- ** Process Operation Execution-  , executeAndResume-  , executeAndResumeOrExit-  , executeAndResumeOrThrow-  -- ** Exit Or Interrupt Recoverable Reasons+  -- *** Exit and Interrupt Reasons   , Interrupt(..)+  , Interrupts   , interruptToExit   , ExitRecovery(..)   , RecoverableInterrupt-  , Interrupts-  , Processes   , ExitSeverity(..)   , SomeExitReason(SomeExitReason)   , toExitRecovery@@ -116,10 +113,20 @@   , toCrashReason   , fromSomeExitReason   , logProcessExit+  -- ** Control Flow+  -- *** Process Interrupt Recoverable Handling+  , provideInterruptsShutdown+  , handleInterrupts+  , tryUninterrupted+  , exitOnInterrupt+  , logInterrupts+  , provideInterrupts+  , mergeEitherInterruptAndExitReason   ) where  import           Control.Applicative+import           Control.Eff.Concurrent.Misc import           Control.DeepSeq import           Control.Eff import           Control.Eff.Exception@@ -341,6 +348,12 @@     { runSerializer :: message -> StrictDynamic     } deriving (Typeable) +instance NFData (Serializer message) where+  rnf (MkSerializer !s) = s `seq` ()++instance Typeable message => Show (Serializer message) where+  showsPrec d _x = showParen (d >= 10) (showSTypeable @message . showString "-serializer")+ instance Contravariant Serializer where   contramap f (MkSerializer b) = MkSerializer (b . f) @@ -646,10 +659,32 @@ -- | This adds a layer of the 'Interrupts' effect on top of 'Processes' type Processes e = Interrupts ': SafeProcesses e +-- | A constraint for an effect set that requires the presence of 'Processes'.+--+-- This constrains the effect list to look like this:+-- @[e1 ... eN, 'Interrupts', 'Process' [e(N+1) .. e(N+k)], e(N+1) .. e(N+k)]@+--+-- It contrains @e@ beyond 'HasSafeProcesses' to encompass 'Interrupts'.+--+-- @since 0.27.1+type HasProcesses e inner = (HasSafeProcesses e inner, Member Interrupts e)+ -- | /Cons/ 'Process' onto a list of effects. This is called @SafeProcesses@ because -- the the actions cannot be interrupted in. type SafeProcesses r = Process r ': r +-- | A constraint for an effect set that requires the presence of 'SafeProcesses'.+--+-- This constrains the effect list to look like this:+-- @[e1 ... eN, 'Process' [e(N+1) .. e(N+k)], e(N+1) .. e(N+k)]@+--+-- It contrains @e@ to support the (only) 'Process' effect.+--+-- This is more relaxed that 'HasProcesses' since it does not require 'Interrupts'.+--+-- @since 0.27.1+type HasSafeProcesses e inner = (SetMember Process (Process inner) e)+ -- | 'Exc'eptions containing 'Interrupt's. -- See 'handleInterrupts', 'exitOnInterrupt' or 'provideInterrupts' type Interrupts = Exc (Interrupt 'Recoverable)@@ -697,7 +732,7 @@ -- when a linked process crashes while we wait in a 'receiveSelectedMessage' -- via a call to 'interrupt'. exitOnInterrupt-  :: (HasCallStack, Member Interrupts r, SetMember Process (Process q) r)+  :: (HasCallStack, HasProcesses r q)   => Eff r a   -> Eff r a exitOnInterrupt = handleInterrupts (exitBecause . interruptToExit)@@ -793,7 +828,7 @@ -- whenever the exit reason satisfies 'isRecoverable', return the exit reason. executeAndResume   :: forall q r v-   . (SetMember Process (Process q) r, HasCallStack)+   . (HasSafeProcesses r q, HasCallStack)   => Process q (ResumeProcess v)   -> Eff r (Either (Interrupt 'Recoverable) v) executeAndResume processAction = do@@ -807,7 +842,7 @@ -- interrupts. executeAndResumeOrExit   :: forall r q v-   . (SetMember Process (Process q) r, HasCallStack)+   . (HasSafeProcesses r q, HasCallStack)   => Process q (ResumeProcess v)   -> Eff r v executeAndResumeOrExit processAction = do@@ -821,7 +856,7 @@ -- interrupts. executeAndResumeOrThrow   :: forall q r v-   . (SetMember Process (Process q) r, HasCallStack, Member Interrupts r)+   . (HasProcesses r q, HasCallStack)   => Process q (ResumeProcess v)   -> Eff r v executeAndResumeOrThrow processAction = do@@ -836,7 +871,7 @@ -- for more information. yieldProcess   :: forall r q-   . (SetMember Process (Process q) r, HasCallStack, Member Interrupts r)+   . (HasProcesses r q, HasCallStack)   => Eff r () yieldProcess = executeAndResumeOrThrow YieldProcess @@ -846,9 +881,8 @@ -- The message will be reduced to normal form ('rnf') by/in the caller process. sendMessage   :: forall r q o-   . ( SetMember Process (Process q) r+   . ( HasProcesses r q      , HasCallStack-     , Member Interrupts r      , Typeable o      , NFData o      )@@ -863,7 +897,7 @@ -- See 'SendMessage'. sendAnyMessage   :: forall r q-   . (SetMember Process (Process q) r, HasCallStack, Member Interrupts r)+   . (HasCallStack, HasProcesses r q)   => ProcessId   -> StrictDynamic   -> Eff r ()@@ -875,7 +909,7 @@ -- See 'SendShutdown'. sendShutdown   :: forall r q-   . (SetMember Process (Process q) r, HasCallStack, Member Interrupts r)+   . (HasCallStack, HasProcesses r q)   => ProcessId   -> Interrupt 'NoRecovery   -> Eff r ()@@ -887,7 +921,7 @@ -- | Like 'sendInterrupt', but also return @True@ iff the process to exit exists. sendInterrupt   :: forall r q-   . (SetMember Process (Process q) r, HasCallStack, Member Interrupts r)+   . (HasCallStack, HasProcesses r q)   => ProcessId   -> Interrupt 'Recoverable   -> Eff r ()@@ -901,7 +935,7 @@ -- 'spawnRaw'. spawn   :: forall r q-   . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)+   . (HasCallStack, HasProcesses r q)   => ProcessTitle   -> Eff (Processes q) ()   -> Eff r ProcessId@@ -911,7 +945,7 @@ -- | Like 'spawn' but return @()@. spawn_   :: forall r q-   . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)+   . (HasCallStack, HasProcesses r q)   => ProcessTitle   -> Eff (Processes q) ()   -> Eff r ()@@ -922,7 +956,7 @@ -- @since 0.12.0 spawnLink   :: forall r q-   . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)+   . (HasCallStack, HasProcesses r q)   => ProcessTitle   -> Eff (Processes q) ()   -> Eff r ProcessId@@ -935,7 +969,7 @@ -- suited. spawnRaw   :: forall r q-   . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)+   . (HasCallStack, HasProcesses r q)   => ProcessTitle   -> Eff (SafeProcesses q) ()   -> Eff r ProcessId@@ -944,7 +978,7 @@ -- | Like 'spawnRaw' but return @()@. spawnRaw_   :: forall r q-   . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)+   . (HasCallStack, HasProcesses r q)   => ProcessTitle   -> Eff (SafeProcesses q) ()   -> Eff r ()@@ -955,7 +989,7 @@ -- @since 0.12.0 isProcessAlive   :: forall r q-   . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)+   . (HasCallStack, HasProcesses r q)   => ProcessId   -> Eff r Bool isProcessAlive pid = isJust <$> executeAndResumeOrThrow (GetProcessState pid)@@ -967,7 +1001,7 @@ -- @since 0.24.1 getProcessState   :: forall r q-   . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)+   . (HasCallStack, HasProcesses r q)   => ProcessId   -> Eff r (Maybe (ProcessTitle, ProcessDetails, ProcessState)) getProcessState pid = executeAndResumeOrThrow (GetProcessState pid)@@ -977,7 +1011,7 @@ -- @since 0.24.1 updateProcessDetails   :: forall r q-   . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)+   . (HasCallStack, HasProcesses r q)   => ProcessDetails   -> Eff r () updateProcessDetails pd = executeAndResumeOrThrow (UpdateProcessDetails pd)@@ -986,7 +1020,7 @@ -- See 'ReceiveSelectedMessage' for more documentation. receiveAnyMessage   :: forall r q-   . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)+   . (HasCallStack, HasProcesses r q)   => Eff r StrictDynamic receiveAnyMessage =   executeAndResumeOrThrow (ReceiveSelectedMessage selectAnyMessage)@@ -998,8 +1032,7 @@   :: forall r q a    . ( HasCallStack      , Show a-     , SetMember Process (Process q) r-     , Member Interrupts r+     , HasProcesses r q      )   => MessageSelector a   -> Eff r a@@ -1014,8 +1047,7 @@      , Typeable a      , NFData a      , Show a-     , SetMember Process (Process q) r-     , Member Interrupts r+     , HasProcesses r q      )   => Eff r a receiveMessage = receiveSelectedMessage (MessageSelector fromStrictDynamic)@@ -1026,7 +1058,7 @@ -- @since 0.12.0 flushMessages   :: forall r q-   . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)+   . (HasCallStack, HasProcesses r q)   => Eff r [StrictDynamic] flushMessages = executeAndResumeOrThrow @q FlushMessages @@ -1040,7 +1072,7 @@ -- See also 'ReceiveSelectedMessage' for more documentation. receiveSelectedLoop   :: forall r q a endOfLoopResult-   . (SetMember Process (Process q) r, HasCallStack)+   . (HasSafeProcesses r q, HasCallStack)   => MessageSelector a   -> (Either (Interrupt 'Recoverable) a -> Eff r (Maybe endOfLoopResult))   -> Eff r endOfLoopResult@@ -1055,7 +1087,7 @@ -- See also 'selectAnyMessage', 'receiveSelectedLoop'. receiveAnyLoop   :: forall r q endOfLoopResult-   . (SetMember Process (Process q) r, HasCallStack)+   . (HasSafeProcesses r q, HasCallStack)   => (Either (Interrupt 'Recoverable) StrictDynamic -> Eff r (Maybe endOfLoopResult))   -> Eff r endOfLoopResult receiveAnyLoop = receiveSelectedLoop selectAnyMessage@@ -1064,18 +1096,18 @@ -- using 'selectMessage'. receiveLoop   :: forall r q a endOfLoopResult-   . (SetMember Process (Process q) r, HasCallStack, NFData a, Typeable a)+   . (HasSafeProcesses r q, HasCallStack, NFData a, Typeable a)   => (Either (Interrupt 'Recoverable) a -> Eff r (Maybe endOfLoopResult))   -> Eff r endOfLoopResult receiveLoop = receiveSelectedLoop selectMessage  -- | Returns the 'ProcessId' of the current process.-self :: (HasCallStack, SetMember Process (Process q) r) => Eff r ProcessId+self :: (HasCallStack, HasSafeProcesses r q) => Eff r ProcessId self = executeAndResumeOrExit SelfPid  -- | Generate a unique 'Int' for the current process. makeReference-  :: (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)+  :: (HasCallStack, HasProcesses r q)   => Eff r Int makeReference = executeAndResumeOrThrow MakeReference @@ -1109,7 +1141,7 @@ -- @since 0.12.0 monitor   :: forall r q-   . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)+   . (HasCallStack, HasProcesses r q)   => ProcessId   -> Eff r MonitorReference monitor = executeAndResumeOrThrow . Monitor . force@@ -1119,7 +1151,7 @@ -- @since 0.12.0 demonitor   :: forall r q-   . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)+   . (HasCallStack, HasProcesses r q)   => MonitorReference   -> Eff r () demonitor = executeAndResumeOrThrow . Demonitor . force@@ -1129,11 +1161,7 @@ -- -- @since 0.12.0 withMonitor-  :: ( HasCallStack-     , Member Interrupts r-     , SetMember Process (Process q) r-     , Member Interrupts r-     )+  :: (HasCallStack, HasProcesses r q)   => ProcessId   -> (MonitorReference -> Eff r a)   -> Eff r a@@ -1145,9 +1173,7 @@ -- @since 0.12.0 receiveWithMonitor   :: ( HasCallStack-     , Member Interrupts r-     , SetMember Process (Process q) r-     , Member Interrupts r+     , HasProcesses r q      , Typeable a      , Show a      )@@ -1169,6 +1195,7 @@   ProcessDown     { downReference :: !MonitorReference     , downReason    :: !SomeExitReason+    , downProcess   :: !ProcessId     }   deriving (Typeable, Generic, Eq, Ord) @@ -1186,21 +1213,35 @@   showsPrec d =     showParen (d >= 10)       . (\case-          ProcessDown ref reason ->-            showString "monitored process down "-              . showsPrec 11 ref+          ProcessDown ref reason pid ->+            showString "down: "+              . shows pid               . showChar ' '+              . shows ref+              . showChar ' '               . showsPrec 11 reason         )  -- | A 'MessageSelector' for the 'ProcessDown' message of a specific -- process. --+-- The parameter is the value obtained by 'monitor'.+-- -- @since 0.12.0 selectProcessDown :: MonitorReference -> MessageSelector ProcessDown selectProcessDown ref0 =-  filterMessage (\(ProcessDown ref _reason) -> ref0 == ref)+  filterMessage (\(ProcessDown ref _reason _pid) -> ref0 == ref) +-- | A 'MessageSelector' for the 'ProcessDown' message. of a specific+-- process.+--+-- In contrast to 'selectProcessDown' this function matches the 'ProcessId'.+--+-- @since 0.28.0+selectProcessDownByProcessId :: ProcessId -> MessageSelector ProcessDown+selectProcessDownByProcessId pid0 =+  filterMessage (\(ProcessDown _ref _reason pid) -> pid0 == pid)+ -- | Connect the calling process to another process, such that -- if one of the processes crashes (i.e. 'isCrash' returns 'True'), the other -- is shutdown with the 'Interrupt' 'LinkedProcessCrashed'.@@ -1208,7 +1249,7 @@ -- @since 0.12.0 linkProcess   :: forall r q-   . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)+   . (HasCallStack, HasProcesses r q)   => ProcessId   -> Eff r () linkProcess = executeAndResumeOrThrow . Link . force@@ -1218,7 +1259,7 @@ -- @since 0.12.0 unlinkProcess   :: forall r q-   . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)+   . (HasCallStack, HasProcesses r q)   => ProcessId   -> Eff r () unlinkProcess = executeAndResumeOrThrow . Unlink . force@@ -1226,20 +1267,20 @@ -- | Exit the process with a 'Interrupt'. exitBecause   :: forall r q a-   . (HasCallStack, SetMember Process (Process q) r)+   . (HasCallStack, HasSafeProcesses r q)   => Interrupt 'NoRecovery   -> Eff r a exitBecause = send . Shutdown @q . force  -- | Exit the process. exitNormally-  :: forall r q a . (HasCallStack, SetMember Process (Process q) r) => Eff r a+  :: forall r q a . (HasCallStack, HasSafeProcesses r q) => Eff r a exitNormally = exitBecause ExitNormally  -- | Exit the process with an error. exitWithError   :: forall r q a-   . (HasCallStack, SetMember Process (Process q) r)+   . (HasCallStack,HasSafeProcesses r q)   => String   -> Eff r a exitWithError = exitBecause . interruptToExit . ErrorInterrupt

@@ -131,7 +131,7 @@ -- | Add monitor: If the process is dead, enqueue a 'ProcessDown' message into the -- owners message queue addMonitoring-  :: ProcessId -> ProcessId -> SchedulerState -> STM MonitorReference+  :: HasCallStack => ProcessId -> ProcessId -> SchedulerState -> STM MonitorReference addMonitoring target owner schedulerState = do   aNewMonitorIndex <- readTVar (schedulerState ^. nextMonitorIndex)   modifyTVar' (schedulerState ^. nextMonitorIndex) (+ 1)@@ -143,13 +143,13 @@                        (Set.insert (monitorRef, owner))       else         let processDownMessage =-              ProcessDown monitorRef (SomeExitReason (OtherProcessNotRunning target))+              ProcessDown monitorRef (SomeExitReason (OtherProcessNotRunning target)) target         in  enqueueMessageOtherProcess owner                                        (toStrictDynamic processDownMessage)                                        schedulerState   return monitorRef -removeMonitoring :: MonitorReference -> SchedulerState -> STM ()+removeMonitoring :: HasCallStack => MonitorReference -> SchedulerState -> STM () removeMonitoring monitorRef schedulerState = modifyTVar'   (schedulerState ^. processMonitors)   (Set.filter (\(ref, _) -> ref /= monitorRef))@@ -163,7 +163,7 @@   go (mr, owner) = when     (monitoredProcess mr == downPid)     (do-      let processDownMessage = ProcessDown mr reason+      let processDownMessage = ProcessDown mr reason downPid       enqueueMessageOtherProcess owner (toStrictDynamic processDownMessage) schedulerState       removeMonitoring mr schedulerState     )@@ -191,8 +191,7 @@  -- | Create a new 'SchedulerState' run an IO action, catching all exceptions, -- and when the actions returns, clean up and kill all processes.-withNewSchedulerState-  :: (HasCallStack) => Eff BaseEffects () -> Eff LoggingAndIo ()+withNewSchedulerState :: HasCallStack => Eff BaseEffects () -> Eff LoggingAndIo () withNewSchedulerState mainProcessAction = Safe.bracketWithError   (lift (atomically newSchedulerState))   (\exceptions schedulerState -> do

@@ -154,7 +154,7 @@         (\res -> Just (res, acc Seq.>< msgRest))         (runMessageSelector messageSelector m) --- | Add monitor: If the process is dead, enqueue a ProcessDown message into the+-- | Add monitor: If the process is dead, enqueue a 'ProcessDown' message into the -- owners message queue addMonitoring   :: ProcessId -> ProcessId -> STS m r -> (MonitorReference, STS m r)@@ -166,7 +166,7 @@       pt <- use msgQs       if Map.member target pt         then monitors %= Set.insert (mref, owner)-        else let pdown = ProcessDown mref (SomeExitReason (OtherProcessNotRunning target))+        else let pdown = ProcessDown mref (SomeExitReason (OtherProcessNotRunning target)) target               in State.modify' (enqueueMsg owner (toStrictDynamic pdown))     return mref @@ -180,7 +180,7 @@  where   go (mr, owner) = when     (monitoredProcess mr == downPid)-    (let pdown = ProcessDown mr reason+    (let pdown = ProcessDown mr reason downPid      in  State.modify' (enqueueMsg owner (toStrictDynamic pdown) . removeMonitoring mr)     ) 

@@ -39,8 +39,7 @@   :: forall a r q    . ( Lifted IO q      , HasCallStack-     , SetMember Process (Process q) r-     , Member Interrupts r+     , HasProcesses r q      , Typeable a      , NFData a      , Show a@@ -59,8 +58,7 @@   :: forall a r q    . ( Lifted IO q      , HasCallStack-     , SetMember Process (Process q) r-     , Member Interrupts r+     , HasProcesses r q      , Show a      )   => MessageSelector a@@ -80,8 +78,7 @@   :: forall a r q    . ( Lifted IO q      , HasCallStack-     , SetMember Process (Process q) r-     , Member Interrupts r+     , HasProcesses r q      , Show a      )   => ProcessId@@ -150,8 +147,7 @@   :: forall r q message    . ( Lifted IO q      , HasCallStack-     , SetMember Process (Process q) r-     , Member Interrupts r+     , HasProcesses r q      , Typeable message      , NFData message      )@@ -179,8 +175,7 @@   :: forall r q    . ( Lifted IO q      , HasCallStack-     , SetMember Process (Process q) r-     , Member Interrupts r+     , HasProcesses r q      )   => Timeout   -> Eff r TimerReference@@ -195,8 +190,7 @@   :: forall r q    . ( Lifted IO q      , HasCallStack-     , SetMember Process (Process q) r-     , Member Interrupts r+     , HasProcesses r q      )   => TimerReference   -> Eff r ()

@@ -47,11 +47,10 @@   , EmbedProtocol(..)   , toEmbeddedEndpoint   , fromEmbeddedEndpoint-  , prettyTypeableShows-  , prettyTypeableShowsPrec   ) where +import           Control.Eff.Concurrent.Misc import           Control.DeepSeq import           Control.Eff.Concurrent.Process import           Control.Lens@@ -63,8 +62,28 @@ import           Type.Reflection  --- | This data family defines the __protocol data units__ (PDU) of a /protocol/.+-- | A server process for /protocol/. --+-- Protocols are represented by phantom types, which are used in different places to+-- index type families and type class instances.+--+-- A 'Process' can send and receive any messages. An 'Endpoint'+-- wraps around a 'ProcessId' and carries a phantom type to indicate+-- the kinds of messages accepted by the process.+--+-- As a metaphor, communication between processes can be thought of waiting for+-- and sending __protocol data units__ belonging to some protocol.+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)+    (showSTypeRep (SomeTypeRep (typeRep @protocol)) . showsPrec 10 c)++-- | This type class and the associated data family defines the+-- __protocol data units__ (PDU) of a /protocol/.+-- -- A Protocol in the sense of a communication interface description -- between processes. --@@ -80,7 +99,7 @@ -- > -- > data BookShop deriving Typeable -- >--- > instance HasPdu BookShop r where+-- > instance Typeable r => HasPdu BookShop r where -- >   data instance Pdu BookShop r where -- >     RentBook  :: BookId   -> Pdu BookShop ('Synchronous (Either RentalError RentalId)) -- >     BringBack :: RentalId -> Pdu BookShop 'Asynchronous@@ -92,7 +111,7 @@ -- > -- -- @since 0.25.1-class (NFData (Pdu protocol reply), Show (Pdu protocol reply), Typeable protocol, Typeable reply) => HasPdu (protocol :: Type) (reply :: Synchronicity) where+class (Tangible (Pdu protocol reply), Typeable protocol, Typeable reply) => HasPdu (protocol :: Type) (reply :: Synchronicity) where    -- | The __protocol data unit__ type for the given protocol.   data family Pdu protocol reply@@ -134,6 +153,7 @@   ( Typeable p   , Typeable r   , Tangible (Pdu p r)+  , HasPdu p r   )  -- | The (promoted) constructors of this type specify (at the type level) the@@ -154,45 +174,11 @@   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 -instance (HasPdu a1 r, HasPdu a2 r) => HasPdu (a1, a2) r where+instance (Typeable r, HasPdu a1 r, HasPdu a2 r) => HasPdu (a1, a2) r where   data instance Pdu (a1, a2) r where           ToPduLeft :: Pdu a1 r -> Pdu (a1, a2) r           ToPduRight :: Pdu a2 r -> Pdu (a1, a2) r@@ -208,7 +194,7 @@           Nothing ->             Nothing -instance (HasPdu a1 r, HasPdu a2 r, HasPdu a3 r) => HasPdu (a1, a2, a3) r where+instance (Typeable r, HasPdu a1 r, HasPdu a2 r, HasPdu a3 r) => HasPdu (a1, a2, a3) r where   data instance Pdu (a1, a2, a3) r where     ToPdu1 :: Pdu a1 r -> Pdu (a1, a2, a3) r     ToPdu2 :: Pdu a2 r -> Pdu (a1, a2, a3) r@@ -229,7 +215,7 @@               Nothing ->                 Nothing -instance (HasPdu a1 r, HasPdu a2 r, HasPdu a3 r, HasPdu a4 r) => HasPdu (a1, a2, a3, a4) r where+instance (Typeable r, HasPdu a1 r, HasPdu a2 r, HasPdu a3 r, HasPdu a4 r) => HasPdu (a1, a2, a3, a4) r where   data instance Pdu (a1, a2, a3, a4) r where     ToPdu1Of4 :: Pdu a1 r -> Pdu (a1, a2, a3, a4) r     ToPdu2Of4 :: Pdu a2 r -> Pdu (a1, a2, a3, a4) r@@ -255,7 +241,7 @@                   Nothing ->                     Nothing -instance (HasPdu a1 r, HasPdu a2 r, HasPdu a3 r, HasPdu a4 r, HasPdu a5 r) => HasPdu (a1, a2, a3, a4, a5) r where+instance (Typeable r, HasPdu a1 r, HasPdu a2 r, HasPdu a3 r, HasPdu a4 r, HasPdu a5 r) => HasPdu (a1, a2, a3, a4, a5) r where   data instance Pdu (a1, a2, a3, a4, a5) r where     ToPdu1Of5 :: Pdu a1 r -> Pdu (a1, a2, a3, a4, a5) r     ToPdu2Of5 :: Pdu a2 r -> Pdu (a1, a2, a3, a4, a5) r@@ -302,10 +288,16 @@ -- Laws: @embeddedPdu = prism' embedPdu fromPdu@ -- -- @since 0.24.0-class EmbedProtocol protocol embeddedProtocol (result :: Synchronicity) where+class+  ( HasPdu protocol         result+  , HasPdu embeddedProtocol result+  )+  => EmbedProtocol protocol embeddedProtocol (result :: Synchronicity) 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 result -> Pdu protocol result@@ -332,7 +324,7 @@ fromEmbeddedEndpoint ::  forall outer inner r . EmbedProtocol outer inner r => Endpoint inner -> Endpoint outer fromEmbeddedEndpoint = coerce -instance EmbedProtocol a a r where+instance HasPdu a r => EmbedProtocol a a r where   embeddedPdu = prism' id Just   embedPdu = id   fromPdu = Just@@ -345,12 +337,12 @@   showsPrec d (ToPduLeft x) = showsPrec d x   showsPrec d (ToPduRight y) = showsPrec d y -instance EmbedProtocol (a1, a2) a1 r where+instance (Typeable r, HasPdu a1 r, HasPdu a2 r) => EmbedProtocol (a1, a2) a1 r where   embedPdu = ToPduLeft   fromPdu (ToPduLeft l) = Just l   fromPdu _ = Nothing -instance EmbedProtocol (a1, a2) a2 r where+instance (Typeable r, HasPdu a1 r, HasPdu a2 r) => EmbedProtocol (a1, a2) a2 r where   embeddedPdu =     prism' ToPduRight $ \case       ToPduRight r -> Just r@@ -366,17 +358,17 @@   showsPrec d (ToPdu2 y) = showsPrec d y   showsPrec d (ToPdu3 z) = showsPrec d z -instance EmbedProtocol (a1, a2, a3) a1 r where+instance (Typeable r, HasPdu a1 r, HasPdu a2 r, HasPdu a3 r) => EmbedProtocol (a1, a2, a3) a1 r where   embedPdu = ToPdu1   fromPdu (ToPdu1 l) = Just l   fromPdu _ = Nothing -instance EmbedProtocol (a1, a2, a3) a2 r where+instance (Typeable r, HasPdu a1 r, HasPdu a2 r, HasPdu a3 r) => EmbedProtocol (a1, a2, a3) a2 r where   embedPdu = ToPdu2   fromPdu (ToPdu2 l) = Just l   fromPdu _ = Nothing -instance EmbedProtocol (a1, a2, a3) a3 r where+instance (Typeable r, HasPdu a1 r, HasPdu a2 r, HasPdu a3 r) => EmbedProtocol (a1, a2, a3) a3 r where   embedPdu = ToPdu3   fromPdu (ToPdu3 l) = Just l   fromPdu _ = Nothing@@ -393,27 +385,27 @@   showsPrec d (ToPdu3Of4 z) = showsPrec d z   showsPrec d (ToPdu4Of4 w) = showsPrec d w -instance EmbedProtocol (a1, a2, a3, a4) a1 r where+instance (Typeable r, HasPdu a1 r, HasPdu a2 r, HasPdu a3 r, HasPdu a4 r) => EmbedProtocol (a1, a2, a3, a4) a1 r where   embedPdu = ToPdu1Of4   fromPdu (ToPdu1Of4 l) = Just l   fromPdu _ = Nothing -instance EmbedProtocol (a1, a2, a3, a4) a2 r where+instance (Typeable r, HasPdu a1 r, HasPdu a2 r, HasPdu a3 r, HasPdu a4 r) => EmbedProtocol (a1, a2, a3, a4) a2 r where   embedPdu = ToPdu2Of4   fromPdu (ToPdu2Of4 l) = Just l   fromPdu _ = Nothing -instance EmbedProtocol (a1, a2, a3, a4) a3 r where+instance (Typeable r, HasPdu a1 r, HasPdu a2 r, HasPdu a3 r, HasPdu a4 r) => EmbedProtocol (a1, a2, a3, a4) a3 r where   embedPdu = ToPdu3Of4   fromPdu (ToPdu3Of4 l) = Just l   fromPdu _ = Nothing -instance EmbedProtocol (a1, a2, a3, a4) a4 r where+instance (Typeable r, HasPdu a1 r, HasPdu a2 r, HasPdu a3 r, HasPdu a4 r) => EmbedProtocol (a1, a2, a3, a4) a4 r where   embedPdu = ToPdu4Of4   fromPdu (ToPdu4Of4 l) = Just l   fromPdu _ = Nothing -instance (NFData (Pdu a1 r), NFData (Pdu a2 r), NFData (Pdu a3 r), NFData (Pdu a4 r), NFData (Pdu a5 r)) => NFData (Pdu (a1, a2, a3, a4, a5) r) where+instance (Typeable r, NFData (Pdu a1 r), NFData (Pdu a2 r), NFData (Pdu a3 r), NFData (Pdu a4 r), NFData (Pdu a5 r)) => NFData (Pdu (a1, a2, a3, a4, a5) r) where   rnf (ToPdu1Of5 x) = rnf x   rnf (ToPdu2Of5 y) = rnf y   rnf (ToPdu3Of5 z) = rnf z@@ -427,27 +419,27 @@   showsPrec d (ToPdu4Of5 w) = showsPrec d w   showsPrec d (ToPdu5Of5 v) = showsPrec d v -instance EmbedProtocol (a1, a2, a3, a4, a5) a1 r where+instance (Typeable r, HasPdu a1 r, HasPdu a2 r, HasPdu a3 r, HasPdu a4 r, HasPdu a5 r) => EmbedProtocol (a1, a2, a3, a4, a5) a1 r where   embedPdu = ToPdu1Of5   fromPdu (ToPdu1Of5 l) = Just l   fromPdu _ = Nothing -instance EmbedProtocol (a1, a2, a3, a4, a5) a2 r where+instance (Typeable r, HasPdu a1 r, HasPdu a2 r, HasPdu a3 r, HasPdu a4 r, HasPdu a5 r) => EmbedProtocol (a1, a2, a3, a4, a5) a2 r where   embedPdu = ToPdu2Of5   fromPdu (ToPdu2Of5 l) = Just l   fromPdu _ = Nothing -instance EmbedProtocol (a1, a2, a3, a4, a5) a3 r where+instance (Typeable r, HasPdu a1 r, HasPdu a2 r, HasPdu a3 r, HasPdu a4 r, HasPdu a5 r) => EmbedProtocol (a1, a2, a3, a4, a5) a3 r where   embedPdu = ToPdu3Of5   fromPdu (ToPdu3Of5 l) = Just l   fromPdu _ = Nothing -instance EmbedProtocol (a1, a2, a3, a4, a5) a4 r where+instance (Typeable r, HasPdu a1 r, HasPdu a2 r, HasPdu a3 r, HasPdu a4 r, HasPdu a5 r) => EmbedProtocol (a1, a2, a3, a4, a5) a4 r where   embedPdu = ToPdu4Of5   fromPdu (ToPdu4Of5 l) = Just l   fromPdu _ = Nothing -instance EmbedProtocol (a1, a2, a3, a4, a5) a5 r where+instance (Typeable r, HasPdu a1 r, HasPdu a2 r, HasPdu a3 r, HasPdu a4 r, HasPdu a5 r) => EmbedProtocol (a1, a2, a3, a4, a5) a5 r where   embedPdu = ToPdu5Of5   fromPdu (ToPdu5Of5 l) = Just l   fromPdu _ = Nothing

@@ -16,11 +16,12 @@  import Control.DeepSeq import Control.Eff-import Control.Eff.Extend ()+import Control.Eff.Concurrent.Misc import Control.Eff.Concurrent.Process import Control.Eff.Concurrent.Protocol import qualified Control.Eff.Concurrent.Protocol.EffectfulServer as E import Control.Eff.Concurrent.Protocol.EffectfulServer (Event(..))+import Control.Eff.Extend () import Control.Eff.Log import Data.Kind import Data.String@@ -71,8 +72,7 @@ -- @since 0.27.0 type TangibleCallbacks tag eLoop e =        ( LogIo e-       , SetMember Process (Process e) eLoop-       , Member Interrupts eLoop+       , HasProcesses eLoop e        , Typeable e        , Typeable eLoop        , Typeable tag@@ -91,7 +91,7 @@       (d >= 10)       (showString (T.unpack x)       . showString " :: "-      . prettyTypeableShows (typeOf px)+      . showSTypeRep (typeOf px)       )  instance (TangibleCallbacks tag eLoop e) => E.Server (Server (tag :: Type) eLoop) (Processes e) where@@ -113,7 +113,7 @@   showsPrec d svr =     showParen (d>=10)       ( showsPrec 11 (genServerId svr)-      . showChar ' ' . prettyTypeableShows (typeRep (Proxy @tag))+      . showChar ' ' . showSTypeRep (typeRep (Proxy @tag))       . showString " callback-server"       ) 

@@ -36,18 +36,17 @@ -- -- The message will be reduced to normal form ('rnf') in the caller process. cast-  :: forall o' o r q+  :: forall destination protocol r q    . ( HasCallStack-     , SetMember Process (Process q) r-     , Member Interrupts r-     , HasPdu o' 'Asynchronous-     , HasPdu o 'Asynchronous-     , EmbedProtocol o' o 'Asynchronous+     , HasProcesses r q+     , HasPdu destination 'Asynchronous+     , HasPdu protocol 'Asynchronous+     , EmbedProtocol destination protocol 'Asynchronous      )-  => Endpoint o'-  -> Pdu o 'Asynchronous+  => Endpoint destination+  -> Pdu protocol 'Asynchronous   -> Eff r ()-cast (Endpoint pid) castMsg = sendMessage pid (Cast (embedPdu @o' castMsg))+cast (Endpoint pid) castMsg = sendMessage pid (Cast (embedPdu @destination castMsg))  -- | Send a request 'Pdu' and wait for the server to return a result value. --@@ -56,28 +55,27 @@ -- -- __Always prefer 'callWithTimeout' over 'call'__ call-  :: forall result protocol' protocol r q-   . ( SetMember Process (Process q) r-     , Member Interrupts r-     , TangiblePdu protocol' ( 'Synchronous result)+  :: forall result destination protocol r q+   . ( HasProcesses r q+     , TangiblePdu destination ( 'Synchronous result)      , TangiblePdu protocol ( 'Synchronous result)-     , EmbedProtocol protocol' protocol ( 'Synchronous result)+     , EmbedProtocol destination protocol ( 'Synchronous result)      , Tangible result      , HasCallStack      )-  => Endpoint protocol'+  => Endpoint destination   -> 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+  let requestMessage = Call origin $! (embedPdu @destination req)+      origin = RequestOrigin @destination @result fromPid callRef   sendMessage pidInternal requestMessage   let selectResult :: MessageSelector result       selectResult =         let extractResult-              :: Reply protocol' result -> Maybe result+              :: Reply destination result -> Maybe result             extractResult (Reply origin' result) =               if origin == origin' then Just result else Nothing         in  selectMessageWith extractResult@@ -99,31 +97,30 @@ -- -- @since 0.22.0 callWithTimeout-  :: forall result protocol' protocol r q-   . ( SetMember Process (Process q) r-     , Member Interrupts r-     , TangiblePdu protocol' ( 'Synchronous result)+  :: forall result destination protocol r q+   . ( HasProcesses r q+     , TangiblePdu destination ( 'Synchronous result)      , TangiblePdu protocol ( 'Synchronous result)-     , EmbedProtocol protocol' protocol ( 'Synchronous result)+     , EmbedProtocol destination protocol ( 'Synchronous result)      , Tangible result      , Member Logs r      , Lifted IO q      , Lifted IO r      , HasCallStack      )-  => Endpoint protocol'+  => Endpoint destination   -> 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+  let requestMessage = Call origin $! embedPdu @destination req+      origin = RequestOrigin @destination @result fromPid callRef   sendMessage pidInternal requestMessage   let selectResult =         let extractResult-              :: Reply protocol' result -> Maybe result+              :: Reply destination result -> Maybe result             extractResult (Reply origin' result) =               if origin == origin' then Just result else Nothing         in selectMessageWith extractResult@@ -146,9 +143,9 @@ -- 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 =+type ServesProtocol o r q =          -- TODO remove!   ( Typeable o-  , SetMember Process (Process q) r+  , HasSafeProcesses r q   , Member (EndpointReader o) r   ) @@ -176,6 +173,7 @@      , Tangible reply      , TangiblePdu o ( 'Synchronous reply)      , Member Interrupts r+           )   => Pdu o ( 'Synchronous reply)   -> Eff r reply@@ -205,15 +203,14 @@ -- This function makes use of AmbigousTypes and TypeApplications. -- -- When not working with an embedded 'Pdu' use 'callEndpointReader'.--- +-- -- @since 0.25.1 callSingleton   :: forall outer inner reply q e   . ( HasCallStack     , EmbedProtocol outer inner  ( 'Synchronous reply)     , Member (EndpointReader outer) e-    , SetMember Process (Process q) e-    , Member Interrupts e+    , HasProcesses e q     , TangiblePdu outer ('Synchronous reply)     , TangiblePdu inner ('Synchronous reply)     , Tangible reply@@ -234,8 +231,7 @@   . ( HasCallStack     , EmbedProtocol outer inner 'Asynchronous     , Member (EndpointReader outer) e-    , SetMember Process (Process q) e-    , Member Interrupts e+    , HasProcesses e q     , HasPdu outer 'Asynchronous     , HasPdu inner 'Asynchronous     )

@@ -13,6 +13,7 @@ import Control.Applicative import Control.DeepSeq import Control.Eff+import Control.Eff.Concurrent.Misc import Control.Eff.Extend () import Control.Eff.Concurrent.Process import Control.Eff.Concurrent.Process.Timer@@ -63,7 +64,7 @@   serverTitle :: Init a e -> ProcessTitle    default serverTitle :: Typeable (ServerPdu a) => Init a e -> ProcessTitle-  serverTitle _ = fromString $ prettyTypeableShows (typeRep (Proxy @(ServerPdu a))) "-server"+  serverTitle _ = fromString $ showSTypeable @(ServerPdu a) "-server"    -- | Process the effects of the implementation   runEffects :: Endpoint (ServerPdu a) -> Init a e -> Eff (ServerEffects a e) x -> Eff e x@@ -82,32 +83,32 @@ -- -- @since 0.24.0 start-  :: forall a q h+  :: forall a r q h   . ( Server a (Processes q)     , Typeable a     , Typeable (ServerPdu a)     , LogsTo h (Processes q)-    , SetMember Process (Process q) (ServerEffects a (Processes q))-    , Member Interrupts             (ServerEffects a (Processes q))+    , HasProcesses (ServerEffects a (Processes q)) q+    , HasProcesses r q     , HasCallStack)   => Init a (Processes q)-  -> Eff (Processes q) (Endpoint (ServerPdu a))+  -> Eff r (Endpoint (ServerPdu a)) start a = asEndpoint <$> spawn (serverTitle a) (protocolServerLoop a)  -- | Execute the server loop. -- -- @since 0.24.0 startLink-  :: forall a q h+  :: forall a r q h   . ( Typeable a     , Typeable (ServerPdu a)     , Server a (Processes q)     , LogsTo h (Processes q)-    , SetMember Process (Process q) (ServerEffects a (Processes q))-    , Member Interrupts (ServerEffects a (Processes q))+    , HasProcesses (ServerEffects a (Processes q)) q+    , HasProcesses r q     , HasCallStack)   => Init a (Processes q)-  -> Eff (Processes q) (Endpoint (ServerPdu a))+  -> Eff r (Endpoint (ServerPdu a)) startLink a = asEndpoint <$> spawnLink (serverTitle a) (protocolServerLoop a)  -- | Execute the server loop.@@ -117,8 +118,7 @@      :: forall q h a      . ( Server a (Processes q)        , LogsTo h (Processes q)-       , SetMember Process (Process q) (ServerEffects a (Processes q))-       , Member Interrupts (ServerEffects a (Processes q))+       , HasProcesses (ServerEffects a (Processes q)) q        , Typeable a        , Typeable (ServerPdu a)        )@@ -133,10 +133,10 @@   where     lmAddEp myEp = lmProcessId ?~ myEp     sel :: MessageSelector (Event (ServerPdu a))-    sel = onRequest <$> selectMessage @(Request (ServerPdu a))-      <|> OnDown    <$> selectMessage @ProcessDown-      <|> OnTimeOut <$> selectMessage @TimerElapsed-      <|> OnMessage <$> selectAnyMessage+    sel =  onRequest <$> selectMessage @(Request (ServerPdu a))+       <|> OnDown    <$> selectMessage @ProcessDown+       <|> OnTimeOut <$> selectMessage @TimerElapsed+       <|> OnMessage <$> selectAnyMessage       where         onRequest :: Request (ServerPdu a) -> Event (ServerPdu a)         onRequest (Call o m) = OnCall (replyTarget (MkSerializer toStrictDynamic) o) m

@@ -1,296 +1,330 @@ -- | Observer Effects ----- This module supports the implementation of observers and observables. Expected use+-- This module supports the implementation of observerRegistry and observables. Expected use -- case is event propagation. --+-- The observable event sources and the observers are usually server processes for a+-- protocol that embeds the 'ObserverRegistry' and 'Observer' 'Pdu's respectively.+--+-- A generic FIFO queue based observer can be found in "Control.Eff.Concurrent.Protocol.Observer.Queue".+-- -- @since 0.16.0 module Control.Eff.Concurrent.Protocol.Observer   ( Observer(..)-  , TangibleObserver+  , ObservationSink()+  , IsObservable+  , CanObserve   , Pdu(RegisterObserver, ForgetObserver, Observed)   , registerObserver   , forgetObserver-  , handleObservations-  , toObserver-  , toObserverFor-  , ObserverRegistry-  , ObserverState-  , Observers()-  , emptyObservers-  , handleObserverRegistration-  , manageObservers-  , observed+  , forgetObserverUnsafe+  , ObserverRegistry(..)+  , ObserverRegistryState+  , observerRegistryNotify+  , evalObserverRegistryState+  , emptyObserverRegistry+  , observerRegistryHandlePdu+  , observerRegistryRemoveProcess   ) where -import           Control.DeepSeq               (NFData(rnf))+import           Control.DeepSeq               ( NFData(rnf) ) import           Control.Eff+import           Control.Eff.Concurrent.Misc import           Control.Eff.Concurrent.Process import           Control.Eff.Concurrent.Protocol import           Control.Eff.Concurrent.Protocol.Client+import           Control.Eff.Concurrent.Protocol.Wrapper (Request(Cast)) import           Control.Eff.State.Strict import           Control.Eff.Log import           Control.Lens-import           Data.Data                     (typeOf)+import           Control.Monad import           Data.Dynamic-import           Data.Foldable import           Data.Kind-import           Data.Proxy-import           Data.Set                       ( Set )-import qualified Data.Set                      as Set+import           Data.Semigroup+import           Data.Map                       ( Map )+import qualified Data.Map                      as Map import           Data.Text                      ( pack )-import           Data.Typeable                  ( typeRep ) import           Data.Type.Pretty+import           GHC.Generics import           GHC.Stack  -- * Observers --- | Describes a process that observes another via 'Asynchronous' 'Pdu' messages.+-- ** Observables++-- | A /protocol/ to communicate 'Observed' events from a sources to many sinks. ----- 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.+-- A sink is any process that serves a protocol with a 'Pdu' instance that embeds+-- the 'Observer' Pdu via an 'EmbedProtocol' instance. ----- @since 0.16.0-data Observer o where-  Observer-    :: ( Tangible o-       , HasPdu p 'Asynchronous-       , Tangible (Endpoint p)-       , Typeable p-       )-    => (o -> Maybe (Pdu p 'Asynchronous)) -> Endpoint p -> Observer o+-- This type has /dual use/, for one it serves as type-index for 'Pdu', i.e.+-- 'HasPdu' respectively, and secondly it contains an 'ObservationSink' and+-- a 'MonitorReference'.+--+-- The 'ObservationSink' is used to serialize and send the 'Observed' events,+-- while the 'ProcessId' serves as key for internal maps.+--+-- @since 0.28.0+newtype Observer event =+  MkObserver (Arg ProcessId (ObservationSink event))+  deriving (Eq, Ord, Typeable) +instance NFData (Observer event) where+  rnf (MkObserver (Arg x y)) = rnf x `seq` rnf y --- | The constraints on the type parameters to an 'Observer'------ @since 0.24.0-type TangibleObserver o =-  ( Tangible o, HasPdu (Observer o) 'Asynchronous)+instance Typeable event => Show (Observer event) where+  showsPrec d (MkObserver (Arg x (MkObservationSink _ m))) =+    showParen (d>=10) (showSTypeable @event . showString "-observer: " . shows x . showChar ' ' . shows m ) -type instance ToPretty (Observer o) =-  PrettyParens ("observing" <:> ToPretty o)+type instance ToPretty (Observer event) =+  PrettyParens ("observing" <:> ToPretty event) -instance (NFData o) => NFData (Observer o) where-  rnf (Observer k s) = rnf k `seq` rnf s+instance (Typeable r, Tangible event) => HasPdu (Observer event) (r :: Synchronicity) where+  data Pdu (Observer event) r where+    Observed :: event -> Pdu (Observer event) 'Asynchronous+    deriving Typeable -instance Show (Observer o) where-  showsPrec d (Observer _ p) = showParen-    (d >= 10)-    (shows (typeRep (Proxy :: Proxy o)) . showString " observer: " . shows p)+instance NFData event => NFData (Pdu (Observer event) r) where+  rnf (Observed event) = rnf event -instance Ord (Observer o) where-  compare (Observer _ s1) (Observer _ s2) =-    compare (s1 ^. fromEndpoint) (s2 ^. fromEndpoint)+instance Show event => Show (Pdu (Observer event) r) where+  showsPrec d (Observed event) =+    showParen (d >= 10) (showString "observered: " . shows event) -instance Eq (Observer o) where-  (==) (Observer _ s1) (Observer _ s2) =-    (==) (s1 ^. fromEndpoint) (s2 ^. fromEndpoint)+-- | The Information necessary to wrap an 'Observed' event to a process specific+-- message, e.g. the embedded 'Observer' 'Pdu' instance, and the 'MonitorReference' of+-- the destination process.+--+-- @since 0.28.0+data ObservationSink event =+  MkObservationSink+    { _observerSerializer :: Serializer (Pdu (Observer event) 'Asynchronous)+    , _observerMonitorReference  :: MonitorReference+    }+  deriving (Generic, Typeable) --- | 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+instance NFData (ObservationSink event) where+  rnf (MkObservationSink s p) = s `seq` rnf p++-- | Convenience type alias.+--+-- @since 0.28.0+type IsObservable eventSource event =+  ( Tangible event+  , EmbedProtocol eventSource (ObserverRegistry event) 'Asynchronous+  )++-- | Convenience type alias.+--+-- @since 0.28.0+type CanObserve eventSink event =+  ( Tangible event+  , EmbedProtocol eventSink (Observer event) 'Asynchronous+  )++-- | And an 'Observer' to the set of recipients for all observations reported by 'observerRegistryNotify'.+--   Note that the observerRegistry 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) 'Asynchronous-     , HasPdu x 'Asynchronous+  :: forall event eventSink eventSource r q .+     ( HasCallStack+     , HasProcesses r q+     , IsObservable eventSource event+     , CanObserve eventSink event      )-  => Observer o-  -> Endpoint x+  => Endpoint eventSource+  -> Endpoint eventSink   -> Eff r ()-registerObserver observer observerRegistry =-  cast observerRegistry (RegisterObserver observer)+registerObserver eventSource eventSink =+   cast eventSource (RegisterObserver serializer (eventSink ^. fromEndpoint))+    where+       serializer =+        MkSerializer+          ( toStrictDynamic+          . Cast+          . embedPdu @eventSink @(Observer event) @( 'Asynchronous )+          ) + -- | Send the 'ForgetObserver' message -- -- @since 0.16.0 forgetObserver-  :: ( SetMember Process (Process q) r+  :: forall event eventSink eventSource r q .+     ( HasProcesses r q      , HasCallStack-     , Member Interrupts r-     , Typeable o-     , NFData o-     , EmbedProtocol x (ObserverRegistry o) 'Asynchronous-     , HasPdu x 'Asynchronous+     , IsObservable eventSource event+     , CanObserve eventSink event      )-  => Observer o-  -> Endpoint x+  => Endpoint eventSource+  -> Endpoint eventSink   -> 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-instance (NFData o, Show o, Typeable o, Typeable r) => HasPdu (Observer o) r where-  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) r) 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+forgetObserver eventSource eventSink =+  forgetObserverUnsafe @event @eventSource eventSource (eventSink ^. fromEndpoint) --- | Use a 'Endpoint' as an 'Observer' for 'handleObservations'.+-- | Send the 'ForgetObserver' message, use a raw 'ProcessId' as parameter. ----- @since 0.16.0-toObserver-  :: forall o p-  . ( HasPdu p 'Asynchronous-    , EmbedProtocol p (Observer o) 'Asynchronous-    , TangibleObserver o-    )-  => Endpoint p-  -> Observer o-toObserver = toObserverFor (embedPdu @p . Observed)+-- @since 0.28.0+forgetObserverUnsafe+  :: forall event eventSource r q .+     ( HasProcesses r q+     , HasCallStack+     , IsObservable eventSource event+     )+  => Endpoint eventSource+  -> ProcessId+  -> Eff r ()+forgetObserverUnsafe eventSource eventSink =+  cast eventSource (ForgetObserver @event eventSink) --- | 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, HasPdu a 'Asynchronous)-  => (o -> Pdu a 'Asynchronous)-  -> Endpoint a-  -> Observer o-toObserverFor wrapper = Observer  (Just . wrapper)+-- ** Observer Support Functions  -- * Managing Observers  -- | A protocol for managing 'Observer's, encompassing  registration and de-registration of -- 'Observer's. ----- @since 0.16.0-data ObserverRegistry (o :: Type)+-- @since 0.28.0+data ObserverRegistry (event :: Type) = MkObserverRegistry+  { _observerRegistry :: Map ProcessId (ObservationSink event) }   deriving Typeable -type instance ToPretty (ObserverRegistry o) =-  PrettyParens ("observer registry" <:> ToPretty o)+type instance ToPretty (ObserverRegistry event) =+  PrettyParens ("observer registry" <:> ToPretty event) -instance (Typeable o, Typeable r) => HasPdu (ObserverRegistry o) r where+instance (Tangible event, Typeable r) => HasPdu (ObserverRegistry event) r where   -- | 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;+  -- The functions 'evalObserverRegistryState' and 'observerRegistryHandlePdu' are used to include observer handling;   --   -- @since 0.16.0-  data instance Pdu (ObserverRegistry o) r where+  data instance Pdu (ObserverRegistry event) 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+    -- @since 0.28.0+    RegisterObserver :: Serializer (Pdu (Observer event) 'Asynchronous) -> ProcessId -> Pdu (ObserverRegistry event) '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+    ForgetObserver ::  ProcessId -> Pdu (ObserverRegistry event) 'Asynchronous+--    -- | This message denotes that a monitored process died+--    --+--    -- @since 0.28.0+--    ObserverMightBeDown :: MonitorReference -> Pdu (ObserverRegistry event) ( 'Synchronous Bool)     deriving Typeable -instance NFData (Pdu (ObserverRegistry o) r) where-  rnf (RegisterObserver o) = rnf o-  rnf (ForgetObserver o) = rnf o+instance NFData (Pdu (ObserverRegistry event) r) where+  rnf (RegisterObserver ser pid) = rnf ser `seq` rnf pid+  rnf (ForgetObserver pid) = rnf pid -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)+instance Typeable event => Show (Pdu (ObserverRegistry event) r) where+  showsPrec d (RegisterObserver ser pid) = showParen (d >= 10) (showString "register observer: " . shows ser . showChar ' ' . shows pid)+  showsPrec d (ForgetObserver p) = showParen (d >= 10) (showString "forget observer: " . shows p)  -- ** 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.+-- messages. It also adds the 'ObserverRegistryState' constraint to the effect list. ----- @since 0.16.0-handleObserverRegistration-  :: forall o q r+-- @since 0.28.0+observerRegistryHandlePdu+  :: forall event q r    . ( HasCallStack-     , Typeable o-     , SetMember Process (Process q) r-     , Member (ObserverState o) r+     , Typeable event+     , HasProcesses r q+     , Member (ObserverRegistryState event) 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)+  => Pdu (ObserverRegistry event) 'Asynchronous -> Eff r ()+observerRegistryHandlePdu = \case+    RegisterObserver ser pid -> do+      monRef <- monitor pid+      let sink = MkObservationSink ser monRef+          observer = MkObserver (Arg pid sink)+      modify @(ObserverRegistry event) (over observerRegistry (Map.insert pid sink))+      os <- get @(ObserverRegistry event)+      logDebug ( "registered "+               <> pack (show observer)+               <> " current number of observers: "  -- TODO put this info into the process details+               <> pack (show (Map.size (view observerRegistry 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)+      wasRemoved <- observerRegistryRemoveProcess @event ob+      unless wasRemoved $+        logDebug (pack (show ("unknown observer " ++ show ob))) ++-- | Remove the entry in the 'ObserverRegistry' for the 'ProcessId'+-- and return 'True' if there was an entry, 'False' otherwise.+--+-- @since 0.28.0+observerRegistryRemoveProcess+  :: forall event q r+   . ( HasCallStack+     , Typeable event+     , HasProcesses r q+     , Member (ObserverRegistryState event) r+     , Member Logs r+     )+  => ProcessId -> Eff r Bool+observerRegistryRemoveProcess ob = do+  mSink <- view (observerRegistry . at ob) <$> get @(ObserverRegistry event)+  modify @(ObserverRegistry event) (observerRegistry . at ob .~ Nothing)+  os <- get @(ObserverRegistry event)+  maybe+    (pure False)+    (foundIt os)+    mSink+ where+  foundIt os sink@(MkObservationSink _ monRef) = do+    demonitor monRef+    logDebug (  "removed: "+           <> (pack $ show $ MkObserver $ Arg ob sink)+           <> " current number of observers: "+           <> pack (show (Map.size (view observerRegistry os))))+    pure True+ -- | Keep track of registered 'Observer's. ----- Handle the 'ObserverState' introduced by 'handleObserverRegistration'.+-- Handle the 'ObserverRegistryState' effect, i.e. run 'evalState' on an 'emptyObserverRegistry'. ----- @since 0.16.0-manageObservers :: Eff (ObserverState o ': r) a -> Eff r a-manageObservers = evalState (Observers Set.empty)+-- @since 0.28.0+evalObserverRegistryState :: HasCallStack => Eff (ObserverRegistryState event ': r) a -> Eff r a+evalObserverRegistryState = evalState emptyObserverRegistry --- | The empty 'ObserverState'+-- | The empty 'ObserverRegistryState' ----- @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)+-- @since 0.28.0+emptyObserverRegistry :: ObserverRegistry event+emptyObserverRegistry = MkObserverRegistry Map.empty --- | Alias for the effect that contains the observers managed by 'manageObservers'-type ObserverState o = State (Observers o)+-- | Alias for the effect that contains the observers managed by 'evalObserverRegistryState'+type ObserverRegistryState event = State (ObserverRegistry event) -observers :: Iso' (Observers o) (Set (Observer o))-observers = iso _observers Observers+-- | An 'Iso' for the 'Map' used internally.+observerRegistry :: Iso' (ObserverRegistry event) (Map ProcessId (ObservationSink event))+observerRegistry = iso _observerRegistry MkObserverRegistry  -- | Report an observation to all observers.--- The process needs to 'manageObservers' and to 'handleObserverRegistration'.+-- The process needs to 'evalObserverRegistryState' and to 'observerRegistryHandlePdu'. ----- @since 0.16.0-observed-  :: forall o r q-   . ( SetMember Process (Process q) r-     , Member (ObserverState o) r-     , Member Interrupts r-     , TangibleObserver o+-- @since 0.28.0+observerRegistryNotify+  :: forall event r q+   . ( HasProcesses r q+     , Member (ObserverRegistryState event) r+     , Tangible event+     , HasCallStack      )-  => o+  => event   -> Eff r ()-observed observation = do-  os <- view observers <$> get-  mapM_ notifySomeObserver os+observerRegistryNotify observation = do+  os <- view observerRegistry <$> get+  mapM_ notifySomeObserver (Map.assocs os)  where-  notifySomeObserver (Observer messageFilter receiver) =-    traverse_ (cast receiver) (messageFilter observation)+  notifySomeObserver (destination,  (MkObservationSink serializer _)) =+    sendAnyMessage destination (runSerializer serializer (Observed observation))

@@ -3,11 +3,10 @@ module Control.Eff.Concurrent.Protocol.Observer.Queue   ( ObservationQueue(..)   , ObservationQueueReader-  , readObservationQueue-  , tryReadObservationQueue-  , flushObservationQueue-  , withObservationQueue-  , spawnLinkObservationQueueWriter+  , observe+  , await+  , tryRead+  , flush   ) where @@ -21,100 +20,138 @@ import           Control.Eff.Log import           Control.Eff.Reader.Strict import           Control.Exception.Safe        as Safe+import           Control.Lens import           Control.Monad.IO.Class-import           Control.Monad                  ( unless )+import           Control.Monad                  ( unless, when ) import qualified Data.Text                     as T import           Data.Typeable import           GHC.Stack  -- | Contains a 'TBQueue' capturing observations.--- See 'spawnLinkObservationQueueWriter', 'readObservationQueue'.+-- See 'observe'. 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 :: forall event proxy . (HasCallStack, Typeable event) => proxy event -> T.Text logPrefix px = "observation queue: " <> T.pack (show (typeRep px)) --- | Read queued observations captured and enqueued in the shared 'TBQueue' by 'spawnLinkObservationQueueWriter'.+-- | Read queued observations captured and enqueued in the shared 'ObservationQueue' by 'observe'.+-- -- This blocks until something was captured or an interrupt or exceptions was thrown. For a non-blocking--- variant use 'tryReadObservationQueue' or 'flushObservationQueue'.-readObservationQueue-  :: forall o r-   . ( Member (ObservationQueueReader o) r+-- variant use 'tryRead' or 'flush'.+--+-- @since 0.28.0+await+  :: forall event r+   . ( Member (ObservationQueueReader event) r      , HasCallStack      , MonadIO (Eff r)-     , Typeable o+     , Typeable event      , Member Logs r      )-  => Eff r o-readObservationQueue = do-  ObservationQueue q <- ask @(ObservationQueue o)+  => Eff r event+await = do+  ObservationQueue q <- ask @(ObservationQueue event)   liftIO (atomically (readTBQueue q)) --- | Read queued observations captured and enqueued in the shared 'TBQueue' by 'spawnLinkObservationQueueWriter'.+-- | Read queued observations captured and enqueued in the shared 'ObservationQueue' by 'observe'.+-- -- Return the oldest enqueued observation immediately or 'Nothing' if the queue is empty.--- Use 'readObservationQueue' to block until an observation is observed.-tryReadObservationQueue-  :: forall o r-   . ( Member (ObservationQueueReader o) r+-- Use 'await' to block until an observation is observerRegistryNotify.+--+-- @since 0.28.0+tryRead+  :: forall event r+   . ( Member (ObservationQueueReader event) r      , HasCallStack      , MonadIO (Eff r)-     , Typeable o+     , Typeable event      , Member Logs r      )-  => Eff r (Maybe o)-tryReadObservationQueue = do-  ObservationQueue q <- ask @(ObservationQueue o)+  => Eff r (Maybe event)+tryRead = do+  ObservationQueue q <- ask @(ObservationQueue event)   liftIO (atomically (tryReadTBQueue q))  -- | Read at once all currently queued observations captured and enqueued--- in the shared 'TBQueue' by 'spawnLinkObservationQueueWriter'.+-- in the shared 'ObservationQueue' by 'observe'.+-- -- This returns immediately all currently enqueued observations.--- For a blocking variant use 'readObservationQueue'.-flushObservationQueue-  :: forall o r-   . ( Member (ObservationQueueReader o) r+-- For a blocking variant use 'await'.+--+-- @since 0.28.0+flush+  :: forall event r+   . ( Member (ObservationQueueReader event) r      , HasCallStack      , MonadIO (Eff r)-     , Typeable o+     , Typeable event      , Member Logs r      )-  => Eff r [o]-flushObservationQueue = do-  ObservationQueue q <- ask @(ObservationQueue o)+  => Eff r [event]+flush = do+  ObservationQueue q <- ask @(ObservationQueue event)   liftIO (atomically (flushTBQueue q)) --- | Create a mutable queue for observations. Use 'spawnLinkObservationQueueWriter' for a simple way to get--- a process that enqueues all observations.+-- | Listen to, and capture observations in an 'ObservationQueue'. --+-- Fork an 'Observer' process that runs only while the body expression is executed.+-- Register the observer to the observable process passed to this function.+--+-- The captured observations can be obtained by 'await',+-- 'tryRead' and 'flush'.+--+-- The queue size is limited to the given number.+-- -- ==== __Example__ -- -- @--- withObservationQueue 100 $ do---   q  <- ask \@(ObservationQueue TestEvent)---   wq <- spawnLinkObservationQueueWriter q---   registerObserver wq testServer---   ...---   cast testServer DoSomething---   evt <- readObservationQueue \@TestEvent---   ...+--+-- import qualified Control.Eff.Concurrent.Observer.Queue as OQ+--+-- foo =+--   do+--     observed <- start SomeObservable+--     OQ.observe 100 observed $ do+--       ...+--       cast observed DoSomething+--       evt <- OQ.await \@TestEvent+--       ... -- @ ----- @since 0.18.0+-- @since 0.28.0+observe+  :: forall event eventSource e q len b+  . ( HasCallStack+    , HasProcesses e q+    , LogIo q+    , LogIo e+    , IsObservable eventSource event+    , Integral len+    , Server (ObservationQueue event) (Processes q)+    )+  => len+  -> Endpoint eventSource+  -> Eff (ObservationQueueReader event ': e) b+  -> Eff e b+observe queueLimit eventSource e =+  withObservationQueue queueLimit (withWriter @event eventSource e)++ withObservationQueue-  :: forall o b e len+  :: forall event b e len    . ( HasCallStack-     , Typeable o-     , Show o+     , Typeable event+     , Show event      , Member Logs e      , Lifted IO e      , Integral len      , Member Interrupts e      )   => len-  -> Eff (ObservationQueueReader o ': e) b+  -> Eff (ObservationQueueReader event ': e) b   -> Eff e b withObservationQueue queueLimit e = do   q   <- lift (newTBQueueIO (fromIntegral queueLimit))@@ -123,39 +160,75 @@   rest <- lift (atomically (flushTBQueue q))   unless     (null rest)-    (logNotice (logPrefix (Proxy @o) <> " unread observations: " <> T.pack (show rest)))+    (logNotice (logPrefix (Proxy @event) <> " 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 observations can be obtained by 'await'. All observations are captured up to -- the queue size limit, such that the first message received will be first message--- returned by 'readObservationQueue'.+-- returned by 'await'. ----- @since 0.18.0-spawnLinkObservationQueueWriter-  :: forall o q h-   . ( TangibleObserver o-     , HasPdu (Observer o) 'Asynchronous-     , Member Logs q+-- @since 0.28.0+spawnWriter+  :: forall event r q h+   . ( Member Logs q      , Lifted IO q      , LogsTo h (Processes q)-     , HasCallStack)-  => ObservationQueue o-  -> Eff (Processes q) (Observer o)-spawnLinkObservationQueueWriter q = do-  cbo <- startLink (MkObservationQueue q)-  pure (toObserver cbo)+     , HasProcesses r q+     , Typeable event+     , HasCallStack+     , Server (ObservationQueue event) (Processes q)+     )+  => ObservationQueue event+  -> Eff r (Endpoint (Observer event))+spawnWriter q =+  start @_ @r @q @h (MkObservationQueue q) -instance (TangibleObserver o, HasPdu (Observer o) 'Asynchronous, Lifted IO q, Member Logs q) => Server (ObservationQueue o) (Processes q) where-  type Protocol (ObservationQueue o) = Observer o+-- | Spawn a process that can be used as an 'Observer' that enqueues the observations into an+--   'ObservationQueue'. See 'withObservationQueue'.+--+-- The observations can be obtained by 'await'. All observations are captured up to+-- the queue size limit, such that the first message received will be first message+-- returned by 'await'.+--+-- @since 0.28.0+withWriter+  :: forall event eventSource e q b+  . ( HasCallStack+    , HasProcesses e q+    , LogIo q+    , IsObservable eventSource event+    , Member (ObservationQueueReader event) e+    )+  => Endpoint eventSource+  -> Eff e b+  -> Eff e b+withWriter eventSource e = do+  q <- ask @(ObservationQueue event)+  w <- spawnWriter @event q+  registerObserver @event eventSource w+  res <- e+  forgetObserver @event eventSource w+  sendShutdown (w^.fromEndpoint) ExitNormally+  pure res -  data instance StartArgument (ObservationQueue o) (Processes q) =-     MkObservationQueue (ObservationQueue o) +instance (Typeable event, Lifted IO q, Member Logs q) => Server (ObservationQueue event) (Processes q) where+  type Protocol (ObservationQueue event) = Observer event++  data instance StartArgument (ObservationQueue event) (Processes q) =+     MkObservationQueue (ObservationQueue event)+   update _ (MkObservationQueue (ObservationQueue q)) =     \case-      OnCast r ->-        handleObservations (lift . atomically . writeTBQueue q) r-      otherMsg -> logError ("unexpected: " <> T.pack (show otherMsg))+      OnCast (Observed !event) -> do+        isFull <- lift . atomically $ do+          isFull <- isFullTBQueue q+          unless isFull (writeTBQueue q event)+          pure isFull+        when isFull $+          logWarning "queue full"+      otherMsg ->+        logError ("unexpected: " <> T.pack (show otherMsg))

@@ -116,28 +116,30 @@ -- -- @since 0.24.0 start-  :: forall a q h+  :: forall a r q h   . ( HasCallStack     , Typeable a     , LogsTo h (Processes q)     , Effectful.Server (Stateful a) (Processes q)     , Server a (Processes q)+    , HasProcesses r q     )-  => StartArgument a (Processes q) -> Eff (Processes q) (Endpoint (Protocol a))+  => StartArgument a (Processes q) -> Eff r (Endpoint (Protocol a)) start = Effectful.start . Init  -- | Execute the server loop. -- -- @since 0.24.0 startLink-  :: forall a q h+  :: forall a r q h   . ( HasCallStack     , Typeable a     , LogsTo h (Processes q)     , Effectful.Server (Stateful a) (Processes q)     , Server a (Processes q)+    , HasProcesses r q     )-  => StartArgument a (Processes q) -> Eff (Processes q) (Endpoint (Protocol a))+  => StartArgument a (Processes q) -> Eff r (Endpoint (Protocol a)) startLink = Effectful.startLink . Init  

@@ -199,20 +199,12 @@           Just existingChild ->             sendReply rt (Left (AlreadyStarted i (existingChild ^. childEndpoint))) -  update _ _supConfig (OnDown (ProcessDown mrChild reason)) = do-      oldEntry <- lookupAndRemoveChildByMonitor @(ChildId p) @p mrChild+  update _ _supConfig (OnDown pd) = do+      oldEntry <- lookupAndRemoveChildByMonitor @(ChildId p) @p (downReference pd)       case oldEntry of-        Nothing ->-          logWarning ("unexpected process down: "-                     <> pack (show mrChild)-                     <> " reason: "-                     <> pack (show reason)-                     )+        Nothing -> logWarning ("unexpected: " <> pack (show pd))         Just (i, c) -> do-          logInfo (  "process down: "-                  <> pack (show mrChild)-                  <> " reason: "-                  <> pack (show reason)+          logInfo (  pack (show pd)                   <> " for child "                   <> pack (show i)                   <> " => "@@ -270,8 +262,7 @@ -- @since 0.23.0 stopSupervisor   :: ( HasCallStack-     , Member Interrupts e-     , SetMember Process (Process q0) e+     , HasProcesses e q      , Member Logs e      , Lifted IO e      , TangibleSup p@@ -291,10 +282,9 @@ isSupervisorAlive   :: forall p q0 e .      ( HasCallStack-     , Member Interrupts e      , Member Logs e      , Typeable p-     , SetMember Process (Process q0) e+     , HasProcesses e q0      )   => Endpoint (Sup p)   -> Eff e Bool@@ -306,9 +296,8 @@ monitorSupervisor   :: forall p q0 e .      ( HasCallStack-     , Member Interrupts e      , Member Logs e-     , SetMember Process (Process q0) e+     , HasProcesses e q0      , TangibleSup p      )   => Endpoint (Sup p)@@ -322,9 +311,8 @@ spawnChild   :: forall p q0 e .      ( HasCallStack-     , Member Interrupts e      , Member Logs e-     , SetMember Process (Process q0) e+     , HasProcesses e q0      , TangibleSup p      , Typeable (Protocol p)      )@@ -340,9 +328,8 @@ lookupChild ::     forall p e q0 .      ( HasCallStack-     , Member Interrupts e      , Member Logs e-     , SetMember Process (Process q0) e+     , HasProcesses e q0      , TangibleSup p      , Typeable (Protocol p)      )@@ -360,9 +347,8 @@ stopChild ::     forall p e q0 .      ( HasCallStack-     , Member Interrupts e      , Member Logs e-     , SetMember Process (Process q0) e+     , HasProcesses e q0      , TangibleSup p      )   => Endpoint (Sup p)@@ -376,8 +362,7 @@ getDiagnosticInfo   :: forall p e q0 .      ( HasCallStack-     , Member Interrupts e-     , SetMember Process (Process q0) e+     , HasProcesses e q0      , TangibleSup p      )   => Endpoint (Sup p)@@ -389,8 +374,7 @@ stopOrKillChild   :: forall p e q0 .      ( HasCallStack-     , SetMember Process (Process q0) e-     , Member Interrupts e+     , HasProcesses e q0      , Lifted IO e      , Lifted IO q0      , Member Logs e@@ -424,7 +408,7 @@ stopAllChildren   :: forall p e q0 .      ( HasCallStack-     , SetMember Process (Process q0) e+     , HasProcesses e q0      , Lifted IO e      , Lifted IO q0      , Member Logs e

@@ -90,12 +90,7 @@ -- | Create a new, unique 'RequestOrigin' value for the current process. -- -- @since 0.24.0-makeRequestOrigin-  :: ( Typeable r-     , NFData r-     , SetMember Process (Process q0) e-     , '[Interrupts] <:: e)-  => Eff e (RequestOrigin p r)+makeRequestOrigin :: (Typeable r, NFData r, HasProcesses e q0) => Eff e (RequestOrigin p r) makeRequestOrigin = RequestOrigin <$> self <*> makeReference  instance NFData (RequestOrigin p r)@@ -154,8 +149,7 @@ -- -- @since 0.25.1 sendReply-  :: ( SetMember Process (Process q) eff-     , Member Interrupts eff+  :: ( HasProcesses eff q      , Tangible reply      , Typeable protocol      )

@@ -8,7 +8,7 @@ import Control.Eff.Extend import Control.Monad (void) import GHC.Stack-import Test.Tasty +import Test.Tasty import Test.Tasty.HUnit import Test.Tasty.Runners @@ -58,3 +58,9 @@ applySchedulerFactory factory procAction = do   scheduler <- factory   scheduler (procAction >> lift (threadDelay 20000))++awaitProcessDown :: (HasCallStack, HasProcesses r q) => ProcessId -> Eff r ProcessDown+awaitProcessDown p = do+  m <- monitor p+  receiveSelectedMessage (selectProcessDown m)+

@@ -75,7 +75,7 @@   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 r where+instance  Typeable r => EmbedProtocol Big Small r where   embeddedPdu =     prism'       BigSmall@@ -94,11 +94,11 @@             BigCall o -> do               logNotice ("BigCall " <> pack (show o))               sendReply rt o-            BigSmall x -> S.update (toEmbeddedEndpoint me) MkSmall (S.OnCall (toEmbeddedReplyTarget rt) x)+            BigSmall x -> S.update (toEmbeddedEndpoint @_ @_ @('Synchronous Bool) me) MkSmall (S.OnCall (toEmbeddedReplyTarget rt) x)     E.OnCast req ->         case req of           BigCast o -> S.putModel @Big o-          BigSmall x -> S.update (toEmbeddedEndpoint me) MkSmall (S.OnCast x)+          BigSmall x -> S.update (toEmbeddedEndpoint @_ @_ @('Asynchronous) me) MkSmall (S.OnCast x)     other ->       interrupt (ErrorInterrupt (show other)) -- ----------------------------------------------------------------------------

@@ -0,0 +1,343 @@+{-# LANGUAGE UndecidableInstances #-}+module ObserverTests+  ( test_observer+  ) where++import Common+import Control.DeepSeq+import Control.Eff+import Control.Eff.Concurrent+import qualified Control.Eff.Concurrent.Protocol.EffectfulServer as S+import qualified Control.Eff.Concurrent.Protocol.Observer.Queue as OQ+import qualified Control.Eff.Concurrent.Protocol.StatefulServer as M+import Control.Lens+import Control.Monad+import Data.Text as T+import Data.Typeable (Typeable)+import Test.Tasty+import Test.Tasty.HUnit+++test_observer :: HasCallStack => TestTree+test_observer = testGroup "observer" (basicTests ++ [observerQueueTests])+++basicTests :: HasCallStack => [TestTree]+basicTests =+  [runTestCase "when no observer is present, nothing crashes"+      $ do+            testObservable <- S.start TestObservableServerInit+            call testObservable (SendTestEvent "1")+            cast testObservable StopTestObservable++            void $ awaitProcessDown (testObservable ^. fromEndpoint)++  , runTestCase "observers receive only messages sent after registration"+      $ do+            testObservable <- S.start TestObservableServerInit+            call testObservable (SendTestEvent "1")+            call testObservable (SendTestEvent "2")+            call testObservable (SendTestEvent "3")+            testObserver <- M.start MkTestObserver+            registerObserver @String testObservable testObserver+            call testObservable (SendTestEvent "4")+            call testObservable (SendTestEvent "5")+            es <- call testObserver GetCapturedEvents+            lift (["4", "5"] @=? es)+            cast testObservable StopTestObservable+            void $ awaitProcessDown (testObservable ^. fromEndpoint)+  , runTestCase "observers receive only messages sent before de-registration"+      $ do+            testObservable <- S.start TestObservableServerInit+            call testObservable (SendTestEvent "1")+            call testObservable (SendTestEvent "2")+            call testObservable (SendTestEvent "3")+            testObserver <- M.start MkTestObserver+            registerObserver @String testObservable testObserver+            call testObservable (SendTestEvent "4")+            call testObservable (SendTestEvent "5")+            forgetObserver @String testObservable testObserver+            call testObservable (SendTestEvent "6")+            call testObservable (SendTestEvent "7")+            es <- call testObserver GetCapturedEvents+            lift (["4", "5"] @=? es)+            cast testObservable StopTestObservable+            void $ awaitProcessDown (testObservable ^. fromEndpoint)+  , runTestCase "observers receive only messages sent between registration and deregistration"+      $ do+            testObservable <- S.start TestObservableServerInit+            call testObservable (SendTestEvent "1")+            call testObservable (SendTestEvent "2")+            call testObservable (SendTestEvent "3")+            testObserver <- M.start MkTestObserver+            registerObserver @String testObservable testObserver+            call testObservable (SendTestEvent "4")+            call testObservable (SendTestEvent "5")+            forgetObserver @String testObservable testObserver+            call testObservable (SendTestEvent "6")+            registerObserver @String testObservable testObserver+            call testObservable (SendTestEvent "7")+            es <- call testObserver GetCapturedEvents+            lift (["4", "5", "7"] @=? es)+            cast testObservable StopTestObservable+            void $ awaitProcessDown (testObservable ^. fromEndpoint)+  , runTestCase "all observers receive all messages sent between registration and deregistration"+      $ do+            testObservable <- S.start TestObservableServerInit+            call testObservable (SendTestEvent "1")+            call testObservable (SendTestEvent "2")+            call testObservable (SendTestEvent "3")+            testObserver1 <- M.start MkTestObserver+            testObserver2 <- M.start MkTestObserver+            testObserver3 <- M.start MkTestObserver+            registerObserver @String testObservable testObserver1+            call testObservable (SendTestEvent "4")+            registerObserver @String testObservable testObserver2+            call testObservable (SendTestEvent "5")+            registerObserver @String testObservable testObserver3+            call testObservable (SendTestEvent "6")+            forgetObserver @String testObservable testObserver1+            call testObservable (SendTestEvent "7")+            forgetObserver @String testObservable testObserver2+            call testObservable (SendTestEvent "8")+            forgetObserver @String testObservable testObserver3+            call testObservable (SendTestEvent "9")+            es1 <- call testObserver1 GetCapturedEvents+            lift (["4", "5", "6"] @=? es1)+            es2 <- call testObserver2 GetCapturedEvents+            lift (["5", "6", "7"] @=? es2)+            es3 <- call testObserver3 GetCapturedEvents+            lift (["6", "7", "8"] @=? es3)+            cast testObservable StopTestObservable+            void $ awaitProcessDown (testObservable ^. fromEndpoint)+  , runTestCase "when an observer exits, the messages are still deliviered to the others"+      $ do+            testObservable <- S.start TestObservableServerInit+            call testObservable (SendTestEvent "1")+            call testObservable (SendTestEvent "2")+            call testObservable (SendTestEvent "3")+            testObserver1 <- M.start MkTestObserver+            testObserver2 <- M.start MkTestObserver+            registerObserver @String testObservable testObserver1+            registerObserver @String testObservable testObserver2+            call testObservable (SendTestEvent "4")+            call testObservable (SendTestEvent "5")+            sendShutdown (testObserver2^.fromEndpoint) ExitNormally+            void $ awaitProcessDown (testObserver2^.fromEndpoint)+            call testObservable (SendTestEvent "6")+            es1 <- call testObserver1 GetCapturedEvents+            lift (["4", "5", "6"] @=? es1)+            cast testObservable StopTestObservable+            void $ awaitProcessDown (testObservable ^. fromEndpoint)+  , runTestCase "when an observer registers multiple times, it still gets the messages only once"+      $ do+            testObservable <- S.start TestObservableServerInit+            call testObservable (SendTestEvent "1")+            call testObservable (SendTestEvent "2")+            call testObservable (SendTestEvent "3")+            testObserver1 <- M.start MkTestObserver+            registerObserver @String testObservable testObserver1+            registerObserver @String testObservable testObserver1+            call testObservable (SendTestEvent "4")+            call testObservable (SendTestEvent "5")+            call testObservable (SendTestEvent "6")+            es1 <- call testObserver1 GetCapturedEvents+            lift (["4", "5", "6"] @=? es1)+            cast testObservable StopTestObservable+            void $ awaitProcessDown (testObservable ^. fromEndpoint)+  , runTestCase "when an observer is forgotton multiple times, nothing bad happens"+      $ do+            testObservable <- S.start TestObservableServerInit+            call testObservable (SendTestEvent "1")+            call testObservable (SendTestEvent "2")+            call testObservable (SendTestEvent "3")+            testObserver1 <- M.start MkTestObserver+            registerObserver @String testObservable testObserver1+            call testObservable (SendTestEvent "4")+            call testObservable (SendTestEvent "5")+            forgetObserver @String testObservable testObserver1+            forgetObserver @String testObservable testObserver1+            forgetObserver @String testObservable testObserver1+            call testObservable (SendTestEvent "6")+            es1 <- call testObserver1 GetCapturedEvents+            lift (["4", "5"] @=? es1)+            cast testObservable StopTestObservable+            void $ awaitProcessDown (testObservable ^. fromEndpoint)+  ]++observerQueueTests :: TestTree+observerQueueTests =+  testGroup "observer-queue"+  [ runTestCase "tryRead"+      $ do+          testObservable <- S.start TestObservableServerInit+          let len :: Int+              len = 1+          OQ.observe  @String len testObservable $ do+            OQ.tryRead @String >>= lift . (Nothing @=?)+            call testObservable (SendTestEvent "1")+            OQ.tryRead @String >>= lift . (Just "1" @=?)++          cast testObservable StopTestObservable+          void $ awaitProcessDown (testObservable ^. fromEndpoint)+  , runTestCase "observe then read"+      $ do+          testObservable <- S.start TestObservableServerInit+          let len :: Int+              len = 1+          OQ.observe  @String len testObservable $ do+            call testObservable (SendTestEvent "1")+            OQ.await @String >>= lift . ("1" @=?)+            call testObservable (SendTestEvent "2")+            OQ.await @String >>= lift . ("2" @=?)+            call testObservable (SendTestEvent "3")+            OQ.await @String >>= lift . ("3" @=?)++          cast testObservable StopTestObservable+          void $ awaitProcessDown (testObservable ^. fromEndpoint)+  , runTestCase "FIFO"+      $ do+          testObservable <- S.start TestObservableServerInit+          let len :: Int+              len = 3+          OQ.observe  @String len testObservable $ do+            call testObservable (SendTestEvent "1")+            call testObservable (SendTestEvent "2")+            call testObservable (SendTestEvent "3")+            OQ.await @String >>= lift . ("1" @=?)+            OQ.await @String >>= lift . ("2" @=?)+            OQ.await @String >>= lift . ("3" @=?)++          cast testObservable StopTestObservable+          void $ awaitProcessDown (testObservable ^. fromEndpoint)+  , runTestCase "flush"+      $ do+          testObservable <- S.start TestObservableServerInit+          let len :: Int+              len = 3+          OQ.observe  @String len testObservable $ do+            call testObservable (SendTestEvent "1")+            call testObservable (SendTestEvent "2")+            call testObservable (SendTestEvent "3")+            OQ.flush @String >>= lift . (["1", "2", "3"] @=?)+            OQ.tryRead @String >>= lift . (Nothing @=?)++          cast testObservable StopTestObservable+          void $ awaitProcessDown (testObservable ^. fromEndpoint)+  , runTestCase "when the queue is full, new observations are dropped"+      $ do+            testObservable <- S.start TestObservableServerInit+            let len :: Int+                len = 2+            OQ.observe  @String len testObservable $ do+              call testObservable (SendTestEvent "1")+              call testObservable (SendTestEvent "2")+              call testObservable (SendTestEvent "3")+              call testObservable (SendTestEvent "4")+              OQ.await @String >>= lift . ("1" @=?)+              OQ.await @String >>= lift . ("2" @=?)+              OQ.tryRead @String >>= lift . (Nothing @=?)++            cast testObservable StopTestObservable+            void $ awaitProcessDown (testObservable ^. fromEndpoint)+  , runTestCase "flush after queue full"+      $ do+          testObservable <- S.start TestObservableServerInit+          let len :: Int+              len = 3+          OQ.observe  @String len testObservable $ do+            call testObservable (SendTestEvent "1")+            call testObservable (SendTestEvent "2")+            call testObservable (SendTestEvent "3")+            call testObservable (SendTestEvent "4")+            call testObservable (SendTestEvent "5")+            OQ.flush @String >>= lift . (["1", "2", "3"] @=?)+            OQ.flush @String >>= lift . ([] @=?)+            OQ.tryRead @String >>= lift . (Nothing @=?)+            call testObservable (SendTestEvent "6")+            OQ.await @String >>= lift . ("6" @=?)++          cast testObservable StopTestObservable+          void $ awaitProcessDown (testObservable ^. fromEndpoint)+  ]+++data TestObservable deriving Typeable++instance Typeable r => HasPdu TestObservable r  where+     data Pdu TestObservable r where+      SendTestEvent :: String -> Pdu TestObservable ('Synchronous ())+      StopTestObservable :: Pdu TestObservable 'Asynchronous+      TestObsReg :: Pdu (ObserverRegistry String) 'Asynchronous -> Pdu TestObservable 'Asynchronous+      deriving (Typeable)++instance EmbedProtocol TestObservable (ObserverRegistry String) 'Asynchronous where+  embeddedPdu = prism' TestObsReg $+    \case+      TestObsReg e -> Just e+      _ -> Nothing++instance Typeable r => NFData (Pdu TestObservable r) where+  rnf (SendTestEvent s) = rnf s+  rnf StopTestObservable = ()+  rnf (TestObsReg x) = rnf x++instance Typeable r => Show (Pdu TestObservable r) where+  show (SendTestEvent x) = "SendTestEvent " ++ show x+  show StopTestObservable = "StopTestObservable"+  show (TestObsReg x) = "TestObsReg " ++ show x++instance (LogIo r, HasProcesses r q) => S.Server TestObservable r where+  data Init TestObservable r = TestObservableServerInit+  type ServerEffects TestObservable r = ObserverRegistryState String ': r+  runEffects _ _ = evalObserverRegistryState+  onEvent _ _ =+    \case+      S.OnCall rt e ->+        case e of+          SendTestEvent x -> observerRegistryNotify x >> sendReply rt ()+      S.OnCast (TestObsReg x) -> observerRegistryHandlePdu x+      S.OnCast StopTestObservable -> exitNormally+      S.OnDown pd -> do+        logDebug ("inspecting: " <> pack (show pd))+        wasHandled <- observerRegistryRemoveProcess @String (downProcess pd)+        unless wasHandled $+          logError ("the process down message was not handled: " <> pack (show pd))+      other ->+        logError ("unexpected: " <> pack (show other))+++data TestObserver deriving Typeable++instance Typeable r => HasPdu TestObserver r where+  data Pdu TestObserver r where+    GetCapturedEvents :: Pdu TestObserver ('Synchronous [String])+    OnTestEvent :: Pdu (Observer String) 'Asynchronous -> Pdu TestObserver 'Asynchronous+    deriving Typeable++instance NFData (Pdu TestObserver r) where+  rnf GetCapturedEvents = ()+  rnf (OnTestEvent e) =  rnf e++instance Show (Pdu TestObserver r) where+  showsPrec _ GetCapturedEvents = showString "GetCapturedEvents"+  showsPrec d (OnTestEvent e) = showParen (d>=10) (showString "OnTestEvent " . shows e)++instance EmbedProtocol TestObserver (Observer String) 'Asynchronous where+  embeddedPdu = prism' OnTestEvent $+    \case+      OnTestEvent e -> Just e+++instance (LogIo r, HasProcesses r q) => M.Server TestObserver r where+  data StartArgument TestObserver r = MkTestObserver+  type Model TestObserver = [String]+  setup _ MkTestObserver = pure ([], ())+  update _ MkTestObserver e =+    case e of+      M.OnCall rt GetCapturedEvents ->+        M.getAndPutModel @TestObserver [] >>= sendReply rt+      M.OnCast (OnTestEvent (Observed x)) ->+        M.modifyModel @TestObserver (++ [x])+      _ ->+        logError ("unexpected: " <> pack (show e))

@@ -107,7 +107,7 @@  returnToSender   :: forall q r-   . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)+   . (HasCallStack, HasProcesses r q)   => Endpoint ReturnToSender   -> String   -> Eff r Bool@@ -119,7 +119,7 @@  stopReturnToSender   :: forall q r-   . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)+   . (HasCallStack, HasProcesses r q)   => Endpoint ReturnToSender   -> Eff r () stopReturnToSender toP = call toP StopReturnToSender@@ -821,12 +821,12 @@         l <- receiveMessage         _ <- monitor l         sendMessage l ()-        pL@(ProcessDown _ _) <- receiveMessage+        pL@(ProcessDown _ _ _) <- receiveMessage         logCritical' ("linked process down: " <> show pL)         _ <- monitor u         mpU <- receiveAfter (TimeoutMicros 1000)         case mpU of-          Just (pU@(ProcessDown _ _)) ->+          Just (pU@(ProcessDown _ _ _)) ->             error ("unlinked process down: " <> show pU)           Nothing ->  logInfo "passed" @@ -845,12 +845,12 @@         l <- receiveMessage         _ <- monitor l         sendMessage l ()-        pL@(ProcessDown _ _) <- receiveMessage+        pL@(ProcessDown _ _ _) <- receiveMessage         logCritical' ("linked process down: " <> show pL)         _ <- monitor u         mpU <- receiveAfter (TimeoutMicros 1000)         case mpU of-          Just (pU@(ProcessDown _ _)) ->+          Just (pU@(ProcessDown _ _ _)) ->             logInfo' ("unlinked process down: " <> show pU)           Nothing ->  error "linked process not exited!" 

@@ -47,11 +47,11 @@              logInfo ("still alive 2: " <> pack (show supAliveAfter2))              lift (supAliveAfter2 @=? True)              sendMessage testWorker ()-             _ <- monitor testWorker-             d1@(ProcessDown _ _) <- receiveMessage+             testWorkerMonitorRef <- monitor testWorker+             d1 <- receiveSelectedMessage (selectProcessDown testWorkerMonitorRef)              logInfo ("got test worker down: " <> pack (show d1))-             _ <- monitorSupervisor sup-             d2@(ProcessDown _ _) <- receiveMessage+             testSupervisorMonitorRef <- monitorSupervisor sup+             d2 <- receiveSelectedMessage (selectProcessDown testSupervisorMonitorRef)              logInfo ("got supervisor down: " <> pack (show d2))              supAliveAfterOwnerExited <- isSupervisorAlive sup              logInfo ("still alive after owner exited: " <> pack (show supAliveAfterOwnerExited))@@ -60,19 +60,19 @@              "Diagnostics"              [ runTestCase "When only time passes the diagnostics do not change" $ do                  sup <- startTestSup-                 diag1 <- Sup.getDiagnosticInfo sup+                 info1 <- Sup.getDiagnosticInfo sup                  lift (threadDelay 10000)-                 diag2 <- Sup.getDiagnosticInfo sup-                 lift (assertEqual "diagnostics should not differ: " diag1 diag2)+                 info2 <- Sup.getDiagnosticInfo sup+                 lift (assertEqual "diagnostics should not differ: " info1 info2)              , runTestCase "When a child is started the diagnostics change" $ do                  sup <- startTestSup-                 diag1 <- Sup.getDiagnosticInfo sup-                 logInfo ("got diagnostics: " <> diag1)+                 info1 <- Sup.getDiagnosticInfo sup+                 logInfo ("got diagnostics: " <> info1)                  let childId = 1                  _child <- fromRight (error "failed to spawn child") <$> Sup.spawnChild sup childId-                 diag2 <- Sup.getDiagnosticInfo sup-                 logInfo ("got diagnostics: " <> diag2)-                 lift $ assertBool ("diagnostics should differ: " ++ show (diag1, diag2)) (diag1 /= diag2)+                 info2 <- Sup.getDiagnosticInfo sup+                 logInfo ("got diagnostics: " <> info2)+                 lift $ assertBool ("diagnostics should differ: " ++ show (info1, info2)) (info1 /= info2)              ]          , let childId = 1             in testGroup@@ -87,10 +87,10 @@                      isSupervisorAlive sup >>= lift . assertBool "supervisor process not running"                      call child (TestGetStringLength "123") >>= lift . assertEqual "child not working" 3                      stopSupervisor sup-                     d1@(ProcessDown mon1 er1) <-+                     d1@(ProcessDown mon1 er1 _) <-                        fromMaybe (error "receive timeout 1") <$> receiveAfter (TimeoutMicros 1000000)                      logInfo ("got process down: " <> pack (show d1))-                     d2@(ProcessDown mon2 er2) <-+                     d2@(ProcessDown mon2 er2 _) <-                        fromMaybe (error "receive timeout 2") <$> receiveAfter (TimeoutMicros 1000000)                      logInfo ("got process down: " <> pack (show d2))                      case if mon1 == supMon && mon2 == childMon@@ -119,10 +119,10 @@                      isSupervisorAlive sup >>= lift . assertBool "supervisor process not running"                      call child (TestGetStringLength "123") >>= lift . assertEqual "child not working" 3                      stopSupervisor sup-                     d1@(ProcessDown mon1 er1) <-+                     d1@(ProcessDown mon1 er1 _) <-                        fromMaybe (error "receive timeout 1") <$> receiveAfter (TimeoutMicros 1000000)                      logInfo ("got process down: " <> pack (show d1))-                     d2@(ProcessDown mon2 er2) <-+                     d2@(ProcessDown mon2 er2 _) <-                        fromMaybe (error "receive timeout 2") <$> receiveAfter (TimeoutMicros 1000000)                      logInfo ("got process down: " <> pack (show d2))                      case if mon1 == supMon && mon2 == childMon@@ -180,7 +180,7 @@                          [ runTestCase "When a child is started it can be stopped" $ do                              (sup, cm) <- startTestSupAndChild                              Sup.stopChild sup i >>= lift . assertBool "child not found"-                             (ProcessDown _ r) <- receiveSelectedMessage (selectProcessDown cm)+                             (ProcessDown _ r _) <- receiveSelectedMessage (selectProcessDown cm)                              lift (assertEqual "bad exit reason" (SomeExitReason ExitNormally) r)                          , runTestCase                              "When a child is stopped but doesn't exit voluntarily, it is kill after some time" $ do@@ -188,7 +188,7 @@                              c <- spawnTestChild sup i                              cm <- monitor (_fromEndpoint c)                              Sup.stopChild sup i >>= lift . assertBool "child not found"-                             (ProcessDown _ r) <- receiveSelectedMessage (selectProcessDown cm)+                             (ProcessDown _ r _) <- receiveSelectedMessage (selectProcessDown cm)                              case r of                                SomeExitReason (ExitUnhandledError _) -> return ()                                _ -> lift (assertFailure ("bad exit reason: " ++ show r))@@ -215,26 +215,26 @@                          [ runTestCase "When a child exits normally, lookupChild will not find it" $ do                              (sup, c, cm) <- startTestSupAndChild                              cast c (TestInterruptWith NormalExitRequested)-                             (ProcessDown _ r) <- receiveSelectedMessage (selectProcessDown cm)+                             (ProcessDown _ r _) <- receiveSelectedMessage (selectProcessDown cm)                              lift (assertEqual "bad exit reason" (SomeExitReason ExitNormally) r)                              x <- Sup.lookupChild sup i                              lift (assertEqual "lookup should not find a child" Nothing x)                          , runTestCase "When a child exits with an error, lookupChild will not find it" $ do                              (sup, c, cm) <- startTestSupAndChild                              cast c (TestInterruptWith (ErrorInterrupt "test error reason"))-                             (ProcessDown _ _) <- receiveSelectedMessage (selectProcessDown cm)+                             (ProcessDown _ _ _) <- receiveSelectedMessage (selectProcessDown cm)                              x <- Sup.lookupChild sup i                              lift (assertEqual "lookup should not find a child" Nothing x)                          , runTestCase "When a child is interrupted from another process and dies, lookupChild will not find it" $ do                              (sup, c, cm) <- startTestSupAndChild                              sendInterrupt (_fromEndpoint c) NormalExitRequested-                             (ProcessDown _ _) <- receiveSelectedMessage (selectProcessDown cm)+                             (ProcessDown _ _ _) <- receiveSelectedMessage (selectProcessDown cm)                              x <- Sup.lookupChild sup i                              lift (assertEqual "lookup should not find a child" Nothing x)                          , runTestCase "When a child is shutdown from another process and dies, lookupChild will not find it" $ do                              (sup, c, cm) <- startTestSupAndChild                              sendShutdown (_fromEndpoint c) ExitProcessCancelled-                             (ProcessDown _ _) <- receiveSelectedMessage (selectProcessDown cm)+                             (ProcessDown _ _ _) <- receiveSelectedMessage (selectProcessDown cm)                              x <- Sup.lookupChild sup i                              lift (assertEqual "lookup should not find a child" Nothing x)                          ]