extensible-effects-concurrent 0.15.0 → 0.16.0

raw patch · 22 files changed

+2077/−2926 lines, 22 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• ChangeLog.md +18/−0
• README.md +18/−25
• examples/example-1/Main.hs +45/−68
• examples/example-2/Main.hs +42/−188
• examples/example-4/Main.hs +9/−9
• extensible-effects-concurrent.cabal +12/−7
• src/Control/Eff/Concurrent.hs +10/−47
• src/Control/Eff/Concurrent/Api.hs +1/−5
• src/Control/Eff/Concurrent/Api/Client.hs +14/−18
• src/Control/Eff/Concurrent/Api/Observer.hs +159/−188
• src/Control/Eff/Concurrent/Api/Observer/Queue.hs +45/−75
• src/Control/Eff/Concurrent/Api/Request.hs +0/−1
• src/Control/Eff/Concurrent/Api/Server.hs +406/−351
• src/Control/Eff/Concurrent/Api/Server2.hs +0/−504
• src/Control/Eff/Concurrent/Process.hs +49/−100
• src/Control/Eff/Concurrent/Process/Interactive.hs +3/−3
• src/Control/Eff/Concurrent/Process/SingleThreadedScheduler.hs +130/−137
• src/Control/Eff/Concurrent/Process/Timer.hs +22/−31
• test/ForkIOScheduler.hs +28/−39
• test/LoopTests.hs +85/−90
• test/ProcessBehaviourTestCases.hs +965/−1023
• test/SingleThreadedScheduler.hs +16/−17

@@ -1,5 +1,23 @@ # Changelog for extensible-effects-concurrent +## 0.16.0++API Stabilization and cleanup release with major API changes.++- Replace `Control.Eff.Concurrent.Api.Server` with+   `Control.Eff.Concurrent.Api.Server2` and rename+   `Control.Eff.Concurrent.Api.Server2` to+   `Control.Eff.Concurrent.Api.Server`++- Rewrite `Observer` and related modules like `Observer.Queue`+  - Remove all type classes+  - Rely on `Server2`+  - Remove `CallBackObserver`+  - Remove the observer support code in `Server2`++- Remove the `SchedulerProxy` parameter and tell library users to enable `AllowAmbiguousTypes` and `TypeApplications`+  - Remove dependent support code like `HasScheduler`+ ## 0.15.0  - Add `Api` `Request` and `Reply` types

@@ -12,6 +12,16 @@  - Memory Leak Free `forever` +## GHC Extensions++In order to use the library you might need to activate some extension+in order to fight some ambiguous types, stemming from the flexibility to+choose different Scheduler implementations.++- AllowAmbiguousTypes+- TypeApplications++ ## Example  ```haskell@@ -28,7 +38,7 @@ main = defaultMain   (do     lift (threadDelay 100000) -- because of async logging-    firstExample forkIoScheduler+    firstExample     lift (threadDelay 100000) -- ... async logging   )     -- The SchedulerProxy paremeter contains the effects of a specific scheduler@@ -38,18 +48,18 @@ newtype WhoAreYou = WhoAreYou ProcessId deriving (Typeable, NFData, Show)  firstExample-  :: (HasLogging IO q) => SchedulerProxy q -> Eff (InterruptableProcess q) ()-firstExample px = do+  :: (HasLogging IO q) => Eff (InterruptableProcess q) ()+firstExample = do   person <- spawn     (do       logInfo "I am waiting for someone to ask me..."-      WhoAreYou replyPid <- receiveMessage px-      sendMessage px replyPid "Alice"+      WhoAreYou replyPid <- receiveMessage+      sendMessage replyPid "Alice"       logInfo (show replyPid ++ " just needed to know it.")     )-  me <- self px-  sendMessage px person (WhoAreYou me)-  personName <- receiveMessage px+  me <- self+  sendMessage person (WhoAreYou me)+  personName <- receiveMessage   logInfo ("I just met " ++ personName)  @@ -83,23 +93,6 @@  [![extensible-effects-concurrent LTS](http://stackage.org/package/extensible-effects-concurrent/badge/lts)](http://stackage.org/lts/package/extensible-effects-concurrent) -### Scheduler Variation--The ambiguity-flexibility trade-off introduced by using extensible effects-kicks in because the `Process` type has a `Spawn` clause, which needs to-know the `Eff`ects.--That is resolved by an omnipresent scheduler proxy parameter.--I will resolve this issue in one of these ways, but haven't decided:--- By using backpack - best option, apart from missing stack support--- By duplicating the code for each scheduler implementation--- By using implicit parameters (experimental use of that technique is in-  the logging part) - problem is that implicit parameter sometimes act weired-  and also might break compiler inlineing.  ### Other 

@@ -27,109 +27,86 @@ deriving instance Show (Api TestApi x)  main :: IO ()-main = defaultMain (example forkIoScheduler)+main = defaultMain example -mainProcessSpawnsAChildAndReturns-  :: (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)-  => SchedulerProxy q-  -> Eff r ()-mainProcessSpawnsAChildAndReturns px =-  void (spawn (void (receiveAnyMessage px)))+mainProcessSpawnsAChildAndReturns :: HasCallStack => Eff (InterruptableProcess q) ()+mainProcessSpawnsAChildAndReturns = void (spawn (void receiveAnyMessage)) -example-  :: ( HasCallStack-     , SetMember Process (Process q) r-     , Member Interrupts r-     , HasLogging IO r-     , HasLogging IO q-     )-  => SchedulerProxy q-  -> Eff r ()-example px = do-  me <- self px+example:: ( HasCallStack, HasLogging IO q) => Eff (InterruptableProcess q) ()+example = do+  me <- self   logInfo ("I am " ++ show me)-  server <- testServerLoop px+  server <- testServerLoop   logInfo ("Started server " ++ show server)   let go = do         lift (putStr "Enter something: ")         x <- lift getLine         case x of           ('K' : rest) -> do-            callRegistered px (TerminateError rest)+            callRegistered (TerminateError rest)             go           ('S' : _) -> do-            callRegistered px Terminate+            callRegistered Terminate             go           ('C' : _) -> do-            castRegistered px (Shout x)+            castRegistered (Shout x)             go           ('R' : rest) -> do-            replicateM_ (read rest) (castRegistered px (Shout x))+            replicateM_ (read rest) (castRegistered (Shout x))             go           ('q' : _) -> logInfo "Done."           _         -> do-            res <- callRegistered px (SayHello x)+            res <- callRegistered (SayHello x)             logInfo ("Result: " ++ show res)             go   registerServer server go  testServerLoop-  :: forall r q+  :: forall q    . ( HasCallStack-     , SetMember Process (Process q) r      , HasLogging IO q-     , Member Interrupts r      )-  => SchedulerProxy q-  -> Eff r (Server TestApi)-testServerLoop px = spawnServer px-  $ apiHandler handleCastTest handleCallTest handleTerminateTest+  => Eff (InterruptableProcess q) (Server TestApi)+testServerLoop = spawnApiServer+  (handleCastTest <> handleCalls handleCallTest) handleTerminateTest  where-  handleCastTest-    :: Api TestApi 'Asynchronous -> Eff (InterruptableProcess q) ApiServerCmd-  handleCastTest (Shout x) = do-    me <- self px+  handleCastTest = handleCasts $ \(Shout x) -> do+    me <- self     logInfo (show me ++ " Shouting: " ++ x)-    return HandleNextRequest-  handleCallTest-    :: Api TestApi ( 'Synchronous x)-    -> (x -> Eff (InterruptableProcess q) ())-    -> Eff (InterruptableProcess q) ApiServerCmd-  handleCallTest (SayHello "e1") _reply = do-    me <- self px+    return AwaitNext+  handleCallTest :: Api TestApi ('Synchronous r) -> (Eff (InterruptableProcess q) (Maybe r, CallbackResult) -> xxx) -> xxx+  handleCallTest (SayHello "e1") k = k $ do+    me <- self     logInfo (show me ++ " raising an error")     interrupt (ProcessError "No body loves me... :,(")-  handleCallTest (SayHello "e2") _reply = do-    me <- self px+  handleCallTest (SayHello "e2") k = k $ do+    me <- self     logInfo (show me ++ " throwing a MyException ")-    lift (Exc.throw MyException)-  handleCallTest (SayHello "self") reply = do-    me <- self px+    void (lift (Exc.throw MyException))+    pure (Nothing, AwaitNext)+  handleCallTest (SayHello "self") k = k $ do+    me <- self     logInfo (show me ++ " casting to self")-    cast px (asServer @TestApi me) (Shout "from me")-    void (reply False)-    return HandleNextRequest-  handleCallTest (SayHello "stop") reply = do-    me <- self px+    cast (asServer @TestApi me) (Shout "from me")+    return (Just False, AwaitNext)+  handleCallTest (SayHello "stop") k = k $ do+    me <- self     logInfo (show me ++ " stopping me")-    void (reply False)-    return (StopApiServer (ProcessError "test error"))-  handleCallTest (SayHello x) reply = do-    me <- self px+    return (Just False, StopServer (ProcessError "test error"))+  handleCallTest (SayHello x) k = k $ do+    me <- self     logInfo (show me ++ " Got Hello: " ++ x)-    void (reply (length x > 3))-    return HandleNextRequest-  handleCallTest Terminate reply = do-    me <- self px+    return (Just (length x > 3), AwaitNext)+  handleCallTest Terminate k = k $ do+    me <- self     logInfo (show me ++ " exiting")-    void (reply ())-    exitNormally px-  handleCallTest (TerminateError msg) reply = do-    me <- self px+    pure (Just (), StopServer ProcessFinished)+  handleCallTest (TerminateError msg) k = k $ do+    me <- self     logInfo (show me ++ " exiting with error: " ++ msg)-    void (reply ())-    exitWithError px msg-  handleTerminateTest msg = do-    me <- self px+    pure (Just (), StopServer (ProcessError msg))+  handleTerminateTest = InterruptCallback $ \msg -> do+    me <- self     logInfo (show me ++ " is exiting: " ++ show msg)     logProcessExit msg+    pure (StopServer msg)

@@ -11,11 +11,7 @@ import           Control.Concurrent  main :: IO ()-main = do-  defaultMain (void (altCounterExample SchedulerProxy))--  defaultMain (void (counterExample SchedulerProxy))-  print (schedulePure (counterExample SchedulerProxy))+main = defaultMain (void counterExample)  -- * First API @@ -24,36 +20,33 @@ data instance Api Counter x where   Inc :: Api Counter 'Asynchronous   Cnt :: Api Counter ('Synchronous Integer)-  ObserveCounter :: SomeObserver Counter -> Api Counter 'Asynchronous-  UnobserveCounter :: SomeObserver Counter -> Api Counter 'Asynchronous   deriving Typeable -altCounterExample+counterExample   :: (Member (Logs LogMessage) q, Lifted IO q)-  => SchedulerProxy q-  -> Eff (InterruptableProcess q) ()-altCounterExample px = do-  (c, (_sdp, cp)) <- altCounter+  => Eff (InterruptableProcess q) ()+counterExample = do+  (c, (co, (_sdp, cp))) <- spawnCounter   lift (threadDelay 500000)-  o <- logCounterObservations px+  o <- logCounterObservations   lift (threadDelay 500000)-  registerObserver px o c+  registerObserver o co   lift (threadDelay 500000)-  cast px c Inc+  cast c Inc   lift (threadDelay 500000)-  sendMessage px cp "test 123"-  cast px c Inc+  sendMessage cp "test 123"+  cast c Inc   lift (threadDelay 500000)-  cast px c Inc-  sendMessage px cp (12312312 :: Int)+  cast c Inc+  sendMessage cp (12312312 :: Int)   lift (threadDelay 500000)-  cast px c Inc+  cast c Inc   lift (threadDelay 500000)-  cast px c Inc+  cast c Inc   lift (threadDelay 500000)-  cast px c Inc+  cast c Inc   lift (threadDelay 500000)-  r <- call px c Cnt+  r <- call c Cnt   lift (threadDelay 500000)   lift (putStrLn ("r: " ++ show r))   lift (threadDelay 500000)@@ -65,17 +58,23 @@   Whoopediedoo :: Bool -> Api SupiDupi ('Synchronous (Maybe ()))   deriving Typeable -altCounter+data CounterChanged = CounterChanged Integer+  deriving (Show, Typeable)++spawnCounter   :: (Member (Logs LogMessage) q)   => Eff        (InterruptableProcess q)-       (Server Counter, (Server SupiDupi, ProcessId))-altCounter = spawnApiServerEffectful-  (manageObservers @Counter . evalState (0 :: Integer) . evalState+       ( Server Counter+       , ( Server (ObserverRegistry CounterChanged)+         , (Server SupiDupi, ProcessId)+         )+       )+spawnCounter = spawnApiServerEffectful+  (manageObservers @CounterChanged . evalState (0 :: Integer) . evalState     (Nothing :: Maybe (RequestOrigin (Api SupiDupi ( 'Synchronous (Maybe ())))))   )   (  handleCalls-      SP       (\case         Cnt ->           ($ do@@ -88,7 +87,7 @@          Inc -> do            val <- get @Integer            let val' = val + 1-           notifyObservers SP (CountChanged val')+           observed (CounterChanged val')            put val'            when (val' > 5) $ do              get@@ -102,15 +101,10 @@                )             return AwaitNext-         ObserveCounter s -> do-           addObserver s-           return AwaitNext-         UnobserveCounter s -> do-           removeObserver s-           return AwaitNext        )+  ^: handleObserverRegistration   ^: handleCallsDeferred-       SP+        (\origin ->          (\case            Whoopediedoo c -> do@@ -132,156 +126,16 @@  deriving instance Show (Api Counter x) -instance Observable Counter where-  data Observation Counter where-    CountChanged :: Integer -> Observation Counter-    deriving (Show, Typeable)-  registerObserverMessage = ObserveCounter-  forgetObserverMessage = UnobserveCounter- logCounterObservations-  :: ( SetMember Process (Process q) r-     , Member (Logs LogMessage) q-     , Member (Logs LogMessage) r-     , Member Interrupts r-     )-  => SchedulerProxy q-  -> Eff r (Server (CallbackObserver Counter))-logCounterObservations px = spawnCallbackObserver-  px-  (\fromSvr msg -> do-    me <- self px-    logInfo (show me ++ " observed on: " ++ show fromSvr ++ ": " ++ show msg)-    return HandleNextRequest-  )---counterHandler-  :: forall r q-   . ( Member (State (Observers Counter)) r-     , Member (State Integer) r-     , SetMember Process (Process q) r-     , Member (Logs LogMessage) q-     , Member (Logs LogMessage) r-     , Member Interrupts r-     )-  => SchedulerProxy q-  -> ApiHandler Counter r-counterHandler px = castAndCallHandler handleCastCounter handleCallCounter- where-  handleCastCounter :: Api Counter 'Asynchronous -> Eff r ApiServerCmd-  handleCastCounter (ObserveCounter o) =-    addObserver o >> return HandleNextRequest-  handleCastCounter (UnobserveCounter o) =-    removeObserver o >> return HandleNextRequest-  handleCastCounter Inc = do-    logInfo "Inc"-    modify (+ (1 :: Integer))-    currentCount <- get-    notifyObservers px (CountChanged currentCount)-    return HandleNextRequest-  handleCallCounter-    :: Api Counter ( 'Synchronous x) -> (x -> Eff r ()) -> Eff r ApiServerCmd-  handleCallCounter Cnt reply = do-    c <- get-    logInfo ("Cnt is " ++ show c)-    reply c-    return HandleNextRequest---- * Second API--data PocketCalc deriving Typeable--data instance Api PocketCalc x where-  Add :: Integer -> Api PocketCalc ('Synchronous Integer)-  AAdd :: Integer -> Api PocketCalc 'Asynchronous--deriving instance Show (Api PocketCalc x)--pocketCalcHandler-  :: forall r q-   . ( Member (State Integer) r-     , SetMember Process (Process q) r-     , Member (Logs LogMessage) r-     )-  => SchedulerProxy q-  -> ApiHandler PocketCalc r-pocketCalcHandler _ = castAndCallHandler handleCastCalc handleCallCalc- where-  handleCastCalc :: Api PocketCalc 'Asynchronous -> Eff r ApiServerCmd-  handleCastCalc (AAdd x) = do-    logInfo ("AsyncAdd " ++ show x)-    modify (+ x)-    c <- get @Integer-    logInfo ("Accumulator is " ++ show c)-    return HandleNextRequest-  handleCallCalc-    :: Api PocketCalc ( 'Synchronous x) -> (x -> Eff r ()) -> Eff r ApiServerCmd-  handleCallCalc (Add x) reply = do-    logInfo ("Add " ++ show x)-    modify (+ x)-    c <- get-    logInfo ("Accumulator is " ++ show c)-    reply c-    return HandleNextRequest--serverLoop-  :: forall r q-   . ( Member (Logs LogMessage) q-     , Member (Logs LogMessage) r-     , Member Interrupts r-     , SetMember Process (Process q) r-     )-  => SchedulerProxy q-  -> Eff r ()-serverLoop px = evalState @Integer-  0-  (manageObservers @Counter (serve px (counterHandler px, pocketCalcHandler px))-  )---- ** Counter client-counterExample-  :: ( SetMember Process (Process q) r-     , Member (Logs LogMessage) q-     , Member (Logs LogMessage) r-     , Member Interrupts r-     , q <:: r-     )-  => SchedulerProxy q-  -> Eff r Integer-counterExample px = execState (0 :: Integer) $ do-  let cnt sv = do-        r <- call px sv Cnt-        logInfo (show sv ++ " " ++ show r)-        modify (+ r)-  pid1 <- spawn (serverLoop px)-  pid2 <- spawn (serverLoop px)-  let cntServer1  = asServer @Counter pid1-      cntServer2  = asServer @Counter pid2-      calcServer1 = asServer @PocketCalc pid1-      calcServer2 = asServer @PocketCalc pid2-  cast px cntServer1 Inc-  cnt cntServer1-  cnt cntServer2-  co1 <- logCounterObservations px-  co2 <- logCounterObservations px-  registerObserver px co1 cntServer1-  registerObserver px co2 cntServer2-  cast px calcServer1 (AAdd 12313)-  cast px cntServer1  Inc-  cnt cntServer1-  cast px cntServer2 Inc-  cnt cntServer2-  registerObserver px co2 cntServer1-  registerObserver px co1 cntServer2-  cast px cntServer1 Inc-  void (call px calcServer2 (Add 878))-  cnt cntServer1-  cast px cntServer2 Inc-  cnt cntServer2-  forgetObserver px co2 cntServer1-  cast px cntServer1 Inc-  cnt cntServer1-  cast px cntServer2 Inc-  cnt cntServer2-  void $ sendShutdown px pid2 (NotRecovered (ProcessError "test test test"))+  :: (Member (Logs LogMessage) q)+  => Eff (InterruptableProcess q) (Observer CounterChanged)+logCounterObservations = do+  svr <- spawnApiServer+    (handleObservations+      (\msg -> do+        logInfo ("observed: " ++ show msg)+        return AwaitNext+      )+    )+    stopServerOnInterrupt+  pure (toObserver svr)

@@ -11,25 +11,25 @@ main = defaultMain   (do     lift (threadDelay 100000) -- because of async logging-    firstExample forkIoScheduler+    firstExample     lift (threadDelay 100000) -- ... async logging   )     -- The SchedulerProxy paremeter contains the effects of a specific scheduler     -- implementation. + newtype WhoAreYou = WhoAreYou ProcessId deriving (Typeable, NFData, Show) -firstExample-  :: (HasLogging IO q) => SchedulerProxy q -> Eff (InterruptableProcess q) ()-firstExample px = do+firstExample :: (HasLogging IO q) => Eff (InterruptableProcess q) ()+firstExample = do   person <- spawn     (do       logInfo "I am waiting for someone to ask me..."-      WhoAreYou replyPid <- receiveMessage px-      sendMessage px replyPid "Alice"+      WhoAreYou replyPid <- receiveMessage+      sendMessage replyPid "Alice"       logInfo (show replyPid ++ " just needed to know it.")     )-  me <- self px-  sendMessage px person (WhoAreYou me)-  personName <- receiveMessage px+  me <- self+  sendMessage person (WhoAreYou me)+  personName <- receiveMessage   logInfo ("I just met " ++ personName)

@@ -1,5 +1,5 @@ name:           extensible-effects-concurrent-version:        0.15.0+version:        0.16.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@@ -63,7 +63,6 @@                   Control.Eff.Concurrent,                   Control.Eff.Concurrent.Api,                   Control.Eff.Concurrent.Api.Client,-                  Control.Eff.Concurrent.Api.Server2,                   Control.Eff.Concurrent.Api.Server,                   Control.Eff.Concurrent.Process,                   Control.Eff.Concurrent.Process.Timer,@@ -76,6 +75,7 @@                 Control.Eff.Concurrent.Api.Request,                 Paths_extensible_effects_concurrent   default-extensions:+                     AllowAmbiguousTypes,                      BangPatterns,                      ConstraintKinds,                      DeriveFoldable,@@ -115,7 +115,8 @@               , lens   ghc-options: -Wall -threaded -fno-full-laziness -rtsopts -with-rtsopts=-N   default-language: Haskell2010-  default-extensions:   BangPatterns+  default-extensions:   AllowAmbiguousTypes+                      , BangPatterns                       , DataKinds                       , FlexibleContexts                       , FlexibleInstances@@ -142,7 +143,8 @@               , lens   ghc-options: -Wall -threaded -fno-full-laziness -rtsopts -with-rtsopts=-N   default-language: Haskell2010-  default-extensions:   BangPatterns+  default-extensions:   AllowAmbiguousTypes+                      , BangPatterns                       , DataKinds                       , FlexibleContexts                       , FlexibleInstances@@ -171,7 +173,8 @@               , lens   ghc-options: -Wall -threaded -fno-full-laziness -rtsopts -with-rtsopts=-N   default-language: Haskell2010-  default-extensions:   BangPatterns+  default-extensions:   AllowAmbiguousTypes+                      , BangPatterns                       , DataKinds                       , FlexibleContexts                       , FlexibleInstances@@ -197,7 +200,8 @@               , deepseq   ghc-options: -Wall -threaded -fno-full-laziness -rtsopts -with-rtsopts=-N   default-language: Haskell2010-  default-extensions:   BangPatterns+  default-extensions:   AllowAmbiguousTypes+                      , BangPatterns                       , DataKinds                       , FlexibleContexts                       , FlexibleInstances@@ -243,7 +247,8 @@               , HUnit               , stm   default-language: Haskell2010-  default-extensions:   BangPatterns+  default-extensions:   AllowAmbiguousTypes+                      , BangPatterns                       , DataKinds                       , FlexibleContexts                       , FlexibleInstances

@@ -18,7 +18,7 @@     module Control.Eff.Concurrent.Api.Server   ,     -- ** /Server/ Functions for Providing APIs (new experimental)-    module Control.Eff.Concurrent.Api.Server2+    module Control.Eff.Concurrent.Api.Server   ,     -- ** Encapsulate 'Api's 'Cast's as well as 'Call's and their 'Reply's     module Control.Eff.Concurrent.Api.Request@@ -51,9 +51,6 @@                                                 , ConsProcess                                                 , ResumeProcess(..)                                                 , SchedulerProxy(..)-                                                , HasScheduler-                                                , getSchedulerProxy-                                                , withSchedulerProxy                                                 , thisSchedulerProxy                                                 , ProcessState(..)                                                 , yieldProcess@@ -165,7 +162,7 @@                                                 , RequestOrigin(..)                                                 , sendReply                                                 )-import           Control.Eff.Concurrent.Api.Server2+import           Control.Eff.Concurrent.Api.Server                                                 ( spawnApiServer                                                 , spawnLinkApiServer                                                 , spawnApiServerStateful@@ -189,52 +186,19 @@                                                 , ToServerPids(..)                                                 , InterruptCallback(..)                                                 , stopServerOnInterrupt-                                                , handleObservations-                                                , Observing()                                                 )--import           Control.Eff.Concurrent.Api.Server-                                                ( serve-                                                , spawnServer-                                                , spawnServerWithEffects-                                                , ApiHandler(..)-                                                , castCallback-                                                , callCallback-                                                , terminateCallback-                                                , apiHandler-                                                , apiHandlerForever-                                                , castHandler-                                                , castHandlerForever-                                                , callHandler-                                                , callHandlerForever-                                                , castAndCallHandler-                                                , castAndCallHandlerForever-                                                , ApiServerCmd(..)-                                                , unhandledCallError-                                                , unhandledCastError-                                                , defaultTermination-                                                -- , Servable(..)-                                                , ServerCallback(..)-                                                , requestHandlerSelector-                                                , terminationHandler-                                                ) import           Control.Eff.Concurrent.Api.Observer                                                 ( Observer(..)-                                                , Observable(..)-                                                , ObserverState-                                                , notifyObserver                                                 , registerObserver                                                 , forgetObserver-                                                , SomeObserver(..)-                                                , notifySomeObserver-                                                , Observers()+                                                , handleObservations+                                                , toObserver+                                                , toObserverFor+                                                , ObserverRegistry+                                                , ObserverState+                                                , handleObserverRegistration                                                 , manageObservers-                                                , addObserver-                                                , removeObserver-                                                , notifyObservers-                                                , CallbackObserver-                                                , spawnCallbackObserver-                                                , spawnLoggingObserver+                                                , observed                                                 ) import           Control.Eff.Concurrent.Api.Observer.Queue                                                 ( ObservationQueue()@@ -242,8 +206,7 @@                                                 , readObservationQueue                                                 , tryReadObservationQueue                                                 , flushObservationQueue-                                                , enqueueObservationsRegistered-                                                , enqueueObservations+                                                , spawnLinkObserverationQueue                                                 ) import           Control.Eff.Concurrent.Process.ForkIOScheduler                                                 ( schedule

@@ -24,15 +24,12 @@   ) where -import           Control.DeepSeq-import           Control.Eff import           Control.Eff.Concurrent.Process import           Control.Lens import           Data.Kind import           Data.Typeable                  ( Typeable                                                 , typeRep                                                 )-import           GHC.Generics  -- | This data family defines an API, a communication interface description -- between at least two processes. The processes act as __servers__ or@@ -78,8 +75,7 @@  instance Typeable api => Show (Server api) where   showsPrec d s@(Server c) =-    showParen (d >= 10)-      (showsPrec 11 (typeRep s) . showsPrec 11 c)+    showParen (d >= 10) (showsPrec 11 (typeRep s) . showsPrec 11 c)  makeLenses ''Server 

@@ -36,11 +36,10 @@      , Typeable o      , Typeable (Api o 'Asynchronous)      )-  => SchedulerProxy q-  -> Server o+  => Server o   -> Api o 'Asynchronous   -> Eff r ()-cast px (Server pid) castMsg = sendMessage px pid (Cast $! castMsg)+cast (Server pid) castMsg = sendMessage pid (Cast $! castMsg)  -- | Send an 'Api' request and wait for the server to return a result value. --@@ -57,15 +56,14 @@      , NFData result      , Show result      )-  => SchedulerProxy q-  -> Server api+  => Server api   -> Api api ( 'Synchronous result)   -> Eff r result-call px (Server pidInternal) req = do-  fromPid <- self px-  callRef <- makeReference px+call (Server pidInternal) req = do+  fromPid <- self+  callRef <- makeReference   let requestMessage = Call callRef fromPid $! req-  sendMessage px pidInternal requestMessage+  sendMessage pidInternal requestMessage   let selectResult :: MessageSelector result       selectResult =         let extractResult@@ -73,7 +71,7 @@             extractResult (Reply _pxResult callRefMsg result) =               if callRefMsg == callRef then Just result else Nothing         in  selectMessageWith extractResult-  rres <- receiveWithMonitor px pidInternal selectResult+  rres <- receiveWithMonitor pidInternal selectResult   either (interrupt . becauseProcessIsDown) return rres  -- | Instead of passing around a 'Server' value and passing to functions like@@ -110,20 +108,18 @@      , Show reply      , Member Interrupts r      )-  => SchedulerProxy q-  -> Api o ( 'Synchronous reply)+  => Api o ( 'Synchronous reply)   -> Eff r reply-callRegistered px method = do+callRegistered method = do   serverPid <- whereIsServer-  call px serverPid method+  call serverPid method  -- | Like 'cast' but take the 'Server' from the reader provided by -- 'registerServer'. castRegistered   :: (Typeable o, ServesApi o r q, HasCallStack, Member Interrupts r)-  => SchedulerProxy q-  -> Api o 'Asynchronous+  => Api o 'Asynchronous   -> Eff r ()-castRegistered px method = do+castRegistered method = do   serverPid <- whereIsServer-  cast px serverPid method+  cast serverPid method

@@ -1,247 +1,218 @@ -- | Observer Effects ----- This module supports the implementation of observers and observables. One use--- case is event propagation. The tools in this module are tailored towards--- 'Api' servers/clients.+-- This module supports the implementation of observers and observables. Expected use+-- case is event propagation.+--+-- @since 0.16.0 module Control.Eff.Concurrent.Api.Observer-  ( -- * Observation API-    Observer(..)-  , Observable(..)-  , notifyObserver+  ( Observer(..)   , registerObserver   , forgetObserver-  -- ** Generalized observation-  , SomeObserver(..)-  , notifySomeObserver-  , Observers()+  , handleObservations+  , toObserver+  , toObserverFor+  , ObserverRegistry   , ObserverState+  , handleObserverRegistration   , manageObservers-  , addObserver-  , removeObserver-  , notifyObservers-  -- * Callback 'Observer'-  , CallbackObserver-  , spawnCallbackObserver-  , spawnLoggingObserver+  , observed   ) where -import           GHC.Stack-import           Data.Dynamic-import           Data.Set                       ( Set )-import qualified Data.Set                      as Set import           Control.Eff import           Control.Eff.Concurrent.Process import           Control.Eff.Concurrent.Api import           Control.Eff.Concurrent.Api.Client import           Control.Eff.Concurrent.Api.Server-import           Control.Eff.Log import           Control.Eff.State.Strict import           Control.Lens+import           Data.Dynamic+import           Data.Foldable+import           Data.Proxy+import           Data.Set                       ( Set )+import qualified Data.Set                      as Set+import           Data.Typeable                  ( typeRep )+import           GHC.Stack --- | An 'Api' index that support observation of the--- another 'Api' that is 'Observable'.-class (Typeable p, Observable o) => Observer p o where-  -- | Wrap the 'Observation' and the 'ProcessId' (i.e. the 'Server')-  -- that caused the observation into an 'Api' value that the-  -- 'Observable' understands.-  observationMessage :: Server o -> Observation o -> Api p 'Asynchronous+-- * Observers --- | An 'Api' index that supports registration and de-registration of--- 'Observer's.-class (Typeable o, Typeable (Observation o)) => Observable o where-  -- | Type of observations visible on this observable-  data Observation o-  -- | Return the 'Api' value for the 'cast_' that registeres an observer-  registerObserverMessage :: SomeObserver o -> Api o 'Asynchronous-  -- | Return the 'Api' value for the 'cast_' that de-registeres an observer-  forgetObserverMessage :: SomeObserver o -> Api o 'Asynchronous+-- | Describes a process that observes another via 'Asynchronous' 'Api' messages.+--+-- An observer consists of a filter and a process id. The filter converts an observation to+-- a message understood by the observer process, and the 'ProcessId' is used to send the message.+--+-- @since 0.16.0+data Observer o where+  Observer+    :: (Show (Server p), Typeable p, Typeable o)+    => (o -> Maybe (Api p 'Asynchronous)) -> Server p -> Observer o --- | Send an 'Observation' to an 'Observer'-notifyObserver-  :: ( SetMember Process (Process q) r-     , Observable o-     , Observer p o-     , HasCallStack-     , Member Interrupts r-     )-  => SchedulerProxy q-  -> Server p-  -> Server o-  -> Observation o-  -> Eff r ()-notifyObserver px observer observed observation =-  cast px observer (observationMessage observed observation)+instance Show (Observer o) where+  showsPrec d (Observer _ p) = showParen+    (d >= 10)+    (shows (typeRep (Proxy :: Proxy o)) . showString " observer: " . shows p) --- | Send the 'registerObserverMessage'+instance Ord (Observer o) where+  compare (Observer _ s1) (Observer _ s2) =+    compare (s1 ^. fromServer) (s2 ^. fromServer)++instance Eq (Observer o) where+  (==) (Observer _ s1) (Observer _ s2) =+    (==) (s1 ^. fromServer) (s2 ^. fromServer)++-- | And an 'Observer' to the set of reciepients for all observations reported by 'observed'.+--   Note that the observers are keyed by the observing process, i.e. a previous entry for the process+--   contained in the 'Observer' is overwritten. If you want multiple entries for a single process, just+--   combine several filter functions.+--+-- @since 0.16.0 registerObserver   :: ( SetMember Process (Process q) r-     , Observable o-     , Observer p o      , HasCallStack      , Member Interrupts r+     , Typeable o      )-  => SchedulerProxy q-  -> Server p-  -> Server o+  => Observer o+  -> Server (ObserverRegistry o)   -> Eff r ()-registerObserver px observer observed =-  cast px observed (registerObserverMessage (SomeObserver observer))+registerObserver observer observerRegistry =+  cast observerRegistry (RegisterObserver observer)  -- | Send the 'forgetObserverMessage'+--+-- @since 0.16.0 forgetObserver   :: ( SetMember Process (Process q) r-     , Observable o-     , Observer p o+     , HasCallStack      , Member Interrupts r+     , Typeable o      )-  => SchedulerProxy q-  -> Server p-  -> Server o+  => Observer o+  -> Server (ObserverRegistry o)   -> Eff r ()-forgetObserver px observer observed =-  cast px observed (forgetObserverMessage (SomeObserver observer))+forgetObserver observer observerRegistry =+  cast observerRegistry (ForgetObserver observer) --- | An existential wrapper around a 'Server' of an 'Observer'.--- Needed to support different types of observers to observe the--- same 'Observable' in a general fashion.-data SomeObserver o where-  SomeObserver :: (Show (Server p), Typeable p, Observer p o) => Server p -> SomeObserver o+-- ** Observer Support Functions -deriving instance Show (SomeObserver o)+-- | A minimal Api for handling observations.+-- This is one simple way of receiving observations - of course users can use+-- any other 'Asynchronous' 'Api' message type for receiving observations.+--+-- @since 0.16.0+data instance Api (Observer o) r where+  Observed :: o -> Api (Observer o) 'Asynchronous -instance Ord (SomeObserver o) where-  compare (SomeObserver (Server o1)) (SomeObserver (Server o2)) =-    compare o1 o2+-- | Based on the 'Api' instance for 'Observer' this simplified writing+-- a callback handler for observations. In order to register to+-- and 'ObservationRegistry' use 'toObserver'.+--+-- @since 0.16.0+handleObservations+  :: (HasCallStack, Typeable o, SetMember Process (Process q) r)+  => (o -> Eff r CallbackResult)+  -> MessageCallback (Observer o) r+handleObservations k = handleCasts+  (\case+    Observed o -> k o+  ) -instance Eq (SomeObserver o) where-  (==) (SomeObserver (Server o1)) (SomeObserver (Server o2)) =-    o1 == o2+-- | Use a 'Server' as an 'Observer' for 'handleObserved'.+--+-- @since 0.16.0+toObserver :: Typeable o => Server (Observer o) -> Observer o+toObserver = toObserverFor Observed --- | Send an 'Observation' to 'SomeObserver'.-notifySomeObserver-  :: ( SetMember Process (Process q) r-     , Observable o-     , HasCallStack-     , Member Interrupts r-     )-  => SchedulerProxy q-  -> Server o-  -> Observation o-  -> SomeObserver o-  -> Eff r ()-notifySomeObserver px observed observation (SomeObserver observer) =-  notifyObserver px observer observed observation+-- | Create an 'Observer' that conditionally accepts all observeration of the+-- given type and applies the given function to them; the function takes an observation and returns an 'Api'+-- cast that the observer server is compatible to.+--+-- @since 0.16.0+toObserverFor+  :: (Typeable a, Typeable o)+  => (o -> Api a 'Asynchronous)+  -> Server a+  -> Observer o+toObserverFor wrapper = Observer (Just . wrapper) --- ** Manage 'Observers's+-- * Managing Observers --- | Internal state for 'manageObservers'-data Observers o =-  Observers { _observers :: Set (SomeObserver o) }+-- | An 'Api' for managing 'Observer's, encompassing  registration and de-registration of+-- 'Observer's.+--+-- @since 0.16.0+data ObserverRegistry o --- | Alias for the effect that contains the observers managed by 'manageObservers'-type ObserverState o = State (Observers o)+-- | Api for managing observers. This can be added to any server for any number of different observation types.+-- The functions 'manageObservers' and 'handleObserverApi' are used to include observer handling;+--+-- @since 0.16.0+data instance Api (ObserverRegistry o) r where+  RegisterObserver :: Observer o -> Api (ObserverRegistry o) 'Asynchronous+  ForgetObserver :: Observer o -> Api (ObserverRegistry o) 'Asynchronous -observers :: Iso' (Observers o) (Set (SomeObserver o))-observers = iso _observers Observers --- | Keep track of registered 'Observer's Observers can be added and removed,--- and an 'Observation' can be sent to all registerd observers at once.-manageObservers :: Eff (ObserverState o ': r) a -> Eff r a-manageObservers = evalState (Observers Set.empty)---- | Add an 'Observer' to the 'Observers' managed by 'manageObservers'.-addObserver-  :: (SetMember Process (Process q) r, Member (ObserverState o) r, Observable o)-  => SomeObserver o-  -> Eff r ()-addObserver = modify . over observers . Set.insert---- | Delete an 'Observer' from the 'Observers' managed by 'manageObservers'.-removeObserver-  :: ( SetMember Process (Process q) r-     , Member (ObserverState o) r-     , Observable o-     , Member Interrupts r-     )-  => SomeObserver o-  -> Eff r ()-removeObserver = modify . over observers . Set.delete-+-- ** Api for integrating 'ObserverRegistry' into processes. --- | Send an 'Observation' to all 'SomeObserver's in the 'Observers' state.-notifyObservers-  :: forall o r q-   . ( Observable o+-- | Provide the implementation for the 'Observerd' Api, this handled 'RegisterObserver' and 'ForgetObserver'+-- messages. It also adds the 'ObserverState' constraint to the effect list.+--+-- @since 0.16.0+handleObserverRegistration+  :: forall o q r+   . ( HasCallStack+     , Typeable o      , SetMember Process (Process q) r      , Member (ObserverState o) r-     , Member Interrupts r      )-  => SchedulerProxy q-  -> Observation o-  -> Eff r ()-notifyObservers px observation = do-  me <- asServer @o <$> self px-  os <- view observers <$> get-  mapM_ (notifySomeObserver px me observation) os+  => MessageCallback (ObserverRegistry o) r+handleObserverRegistration = handleCasts+  (\case+    RegisterObserver ob ->+      get @(Observers o)+        >>= put+        .   over observers (Set.insert ob)+        >>  pure AwaitNext+    ForgetObserver ob ->+      get @(Observers o)+        >>= put+        .   over observers (Set.delete ob)+        >>  pure AwaitNext+  ) --- | An 'Observer' that schedules the observations to an effectful callback.-data CallbackObserver o-  deriving Typeable -data instance Api (CallbackObserver o) r where-  CbObserved :: (Typeable o, Typeable (Observation o)) =>-             Server o -> Observation o -> Api (CallbackObserver o) 'Asynchronous-  deriving Typeable+-- | Keep track of registered 'Observer's Observers can be added and removed,+-- and an 'Observation' can be sent to all registerd observers at once.+--+-- @since 0.16.0+manageObservers :: Eff (ObserverState o ': r) a -> Eff r a+manageObservers = evalState (Observers Set.empty) -deriving instance Show (Observation o) => Show (Api (CallbackObserver o) r)+-- | Internal state for 'manageObservers'+data Observers o =+  Observers { _observers :: Set (Observer o) } -instance (Observable o) => Observer (CallbackObserver o) o where-  observationMessage = CbObserved+-- | Alias for the effect that contains the observers managed by 'manageObservers'+type ObserverState o = State (Observers o) --- | Start a new process for an 'Observer' that schedules--- all observations to an effectful callback.-spawnCallbackObserver-  :: forall o r q-   . ( SetMember Process (Process q) r-     , Typeable o-     , Show (Observation o)-     , Observable o-     , Member (Logs LogMessage) q-     , Member Interrupts r-     , HasCallStack-     )-  => SchedulerProxy q-  -> (  Server o-     -> Observation o-     -> Eff (InterruptableProcess q) ApiServerCmd-     )-  -> Eff r (Server (CallbackObserver o))-spawnCallbackObserver px onObserve = spawnServerWithEffects-  px-  (castHandler handleCastCbo)-  id-  where handleCastCbo (CbObserved fromSvr v) = onObserve fromSvr v+observers :: Iso' (Observers o) (Set (Observer o))+observers = iso _observers Observers --- | Start a new process for an 'Observer' that schedules--- all observations to an effectful callback.+-- | Report an observation to all observers.+-- The process needs to 'manageObservers' and to 'handleObserverRegistration'. ----- @since 0.3.0.0-spawnLoggingObserver+-- @since 0.16.0+observed   :: forall o r q    . ( SetMember Process (Process q) r-     , Typeable o-     , Show (Observation o)-     , Observable o-     , Member (Logs LogMessage) q-     , Member (Logs LogMessage) r+     , Member (ObserverState o) r      , Member Interrupts r-     , HasCallStack      )-  => SchedulerProxy q-  -> Eff r (Server (CallbackObserver o))-spawnLoggingObserver px = spawnCallbackObserver-  px-  (\s o ->-    logDebug (show s ++ " OBSERVED: " ++ show o) >> return HandleNextRequest-  )+  => o+  -> Eff r ()+observed observation = do+  os <- view observers <$> get+  mapM_ notifySomeObserver os+ where+  notifySomeObserver (Observer messageFilter receiver) =+    traverse_ (cast receiver) (messageFilter observation)

@@ -1,17 +1,17 @@--- | Capture 'Observation's and enqueue then into an STM 'TBQeueu'.+-- | A small process to capture and _share_ observation's by enqueueing them into an STM 'TBQeueu'. module Control.Eff.Concurrent.Api.Observer.Queue   ( ObservationQueue()   , ObservationQueueReader   , readObservationQueue   , tryReadObservationQueue   , flushObservationQueue-  , enqueueObservationsRegistered-  , enqueueObservations+  , spawnLinkObserverationQueue   ) where  import           Control.Concurrent.STM import           Control.Eff+import           Control.Eff.Extend import           Control.Eff.ExceptionExtra     ( ) import           Control.Eff.Lift import           Control.Eff.Concurrent.Process@@ -29,8 +29,8 @@ import           GHC.Stack  -- | Contains a 'TBQueue' capturing observations received by 'enqueueObservationsRegistered'--- or 'enqueueObservations'.-newtype ObservationQueue a = ObservationQueue (TBQueue (Observation a))+-- or 'spawnLinkObserverationQueue'.+newtype ObservationQueue a = ObservationQueue (TBQueue a)  -- | A 'Reader' for an 'ObservationQueue'. type ObservationQueueReader a = Reader (ObservationQueue a)@@ -38,9 +38,8 @@ logPrefix :: forall o proxy . (HasCallStack, Typeable o) => proxy o -> String logPrefix px = "observation queue: " ++ show (typeRep px) --- | Read queued observations captured by observing a 'Server' that implements--- an 'Observable' 'Api' using 'enqueueObservationsRegistered' or 'enqueueObservations'.--- This blocks until the next 'Observation' received. For a non-blocking+-- | Read queued observations captured and enqueued in the shared 'TBQueue' by 'spawnLinkObserverationQueue'.+-- 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@@ -50,14 +49,13 @@      , Typeable o      , HasLogging IO r      )-  => Eff r (Observation o)+  => Eff r o readObservationQueue = do   ObservationQueue q <- ask @(ObservationQueue o)   liftIO (atomically (readTBQueue q)) --- | Read queued observations captured by observing a 'Server' that implements--- an 'Observable' 'Api' using 'enqueueObservationsRegistered' or 'enqueueObservations'.--- Return the next 'Observation' immediately or 'Nothing' if the queue is empty.+-- | Read queued observations captured and enqueued in the shared 'TBQueue' by 'spawnLinkObserverationQueue'.+-- 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@@ -67,12 +65,12 @@      , Typeable o      , HasLogging IO r      )-  => Eff r (Maybe (Observation o))+  => Eff r (Maybe o) tryReadObservationQueue = do   ObservationQueue q <- ask @(ObservationQueue o)   liftIO (atomically (tryReadTBQueue q)) --- | Read all currently queued 'Observation's captured by 'enqueueObservations'.+-- | Read at once all currently queued observations captured and enqueued in the shared 'TBQueue' by 'spawnLinkObserverationQueue'. -- This returns immediately all currently enqueued 'Observation's. For a blocking -- variant use 'readObservationQueue'. flushObservationQueue@@ -83,58 +81,29 @@      , Typeable o      , HasLogging IO r      )-  => Eff r [Observation o]+  => Eff r [o] flushObservationQueue = do   ObservationQueue q <- ask @(ObservationQueue o)   liftIO (atomically (flushTBQueue q)) --- | Observe a(the) registered 'Server' that implements an 'Observable' 'Api'.--- Based on 'enqueueObservations'.-enqueueObservationsRegistered-  :: forall o r q a-   . ( ServesApi o r q-     , SetMember Process (Process q) r-     , Typeable o-     , Show (Observation o)-     , Observable o-     , HasLogging IO q-     , HasLogging IO r-     , Member Interrupts r-     , Lifted IO r-     , HasCallStack-     )-  => SchedulerProxy q-  -> Int-  -> Eff (ObservationQueueReader o ': r) a-  -> Eff r a-enqueueObservationsRegistered px queueLimit k = do-  oSvr <- whereIsServer @o-  enqueueObservations px oSvr queueLimit k --- | Observe a 'Server' that implements an 'Observable' 'Api', the 'Observation's+-- | Capture an observation.+data instance Api (ObservationQueue a) r where+  EnqueueObservation :: a -> Api (ObservationQueue a) 'Asynchronous+  StopObservationQueue :: Api (ObservationQueue a) ('Synchronous ())++-- | Observe a 'Server' that implements an 'Observable' 'Api'. The observations -- can be obtained by 'readObservationQueue'. All observations are captured up to -- the queue size limit, such that the first message received will be first message -- returned by 'readObservationQueue'.------ This function captures runtime exceptions and cleans up accordingly.-enqueueObservations-  :: forall o r q a-   . ( SetMember Process (Process q) r-     , Typeable o-     , Show (Observation o)-     , Observable o-     , HasLogging IO r-     , HasLogging IO q-     , Member Interrupts r-     , Lifted IO q-     , HasCallStack-     )-  => SchedulerProxy q-  -> Server o+spawnLinkObserverationQueue+  :: forall o q a+   . (Typeable o, Show o, HasLogging IO q, Lifted IO q, HasCallStack)+  => Server (ObserverRegistry o)   -> Int-  -> Eff (ObservationQueueReader o ': r) a-  -> Eff r a-enqueueObservations px oSvr queueLimit k = withQueue+  -> Eff (ObservationQueueReader o ': InterruptableProcess q) a+  -> Eff (InterruptableProcess q) a+spawnLinkObserverationQueue oSvr queueLimit k = withQueue   queueLimit   (do     ObservationQueue q <- ask @(ObservationQueue o)@@ -144,36 +113,37 @@                 (logPrefix (Proxy @o))                 queueLimit         )-      cbo <- spawnCallbackObserver-        px-        (\_from observation -> do-          liftIO (atomically (writeTBQueue q observation))-          return HandleNextRequest-        )-      logDebug-        (printf "%s started observer process %s"-                (logPrefix (Proxy @o))-                (show cbo)+      cbo <- raise+        (spawnLinkApiServer+          (handleCasts+            (\case+              EnqueueObservation o -> do+                lift (atomically (writeTBQueue q o))+                pure AwaitNext+            )+          )+          stopServerOnInterrupt         )-      registerObserver SchedulerProxy cbo oSvr+      let thisObserver = toObserverFor EnqueueObservation cbo+      registerObserver thisObserver oSvr       res <- k-      forgetObserver SchedulerProxy cbo oSvr-      sendShutdown px (_fromServer cbo) ExitNormally+      forgetObserver thisObserver oSvr+      call cbo StopObservationQueue       logDebug (printf "%s stopped observer process" (logPrefix (Proxy @o)))       return res   )  withQueue-  :: forall a b e len+  :: forall o b e len    . ( HasCallStack-     , Typeable a-     , Show (Observation a)+     , Typeable o+     , Show o      , HasLogging IO e      , Integral len      , Member Interrupts e      )   => len-  -> Eff (ObservationQueueReader a ': e) b+  -> Eff (ObservationQueueReader o ': e) b   -> Eff e b withQueue queueLimit e = do   q   <- liftIO (newTBQueueIO (fromIntegral queueLimit))@@ -182,5 +152,5 @@   rest <- liftIO (atomically (flushTBQueue q))   unless     (null rest)-    (logNotice (logPrefix (Proxy @a) ++ " unread observations: " ++ show rest))+    (logNotice (logPrefix (Proxy @o) ++ " unread observations: " ++ show rest))   either (\em -> logError (show em) >> liftIO (throwIO em)) return res

@@ -82,6 +82,5 @@   -> reply   -> Eff eff () sendReply origin reply = sendMessage-  SP   (_requestOriginPid origin)   (Reply (Proxy @request) (_requestOriginCallRef origin) $! reply)

@@ -1,416 +1,471 @@--- | Functions to implement 'Api' __servers__.+-- | Support code to implement 'Api' _server_ processes.+--+-- @since 0.16.0 module Control.Eff.Concurrent.Api.Server-  (-  -- * Api Server-    serve-  , spawnServer-  , spawnServerWithEffects-  -- * Api Callbacks-  , ApiHandler(..)-  , castCallback-  , callCallback-  , terminateCallback-  , apiHandler-  , apiHandlerForever-  , castHandler-  , castHandlerForever-  , callHandler-  , callHandlerForever-  , castAndCallHandler-  , castAndCallHandlerForever-  , ApiServerCmd(..)-  , unhandledCallError-  , unhandledCastError-  , defaultTermination-  -- * Callback Composition-  , Servable(..)-  , ServerCallback(..)-  , requestHandlerSelector-  , terminationHandler+  ( -- * Starting Api Servers+    spawnApiServer+  , spawnLinkApiServer+  , spawnApiServerStateful+  , spawnApiServerEffectful+  , spawnLinkApiServerEffectful+  , apiServerLoop+  -- ** Api Server Callbacks+  , CallbackResult(..)+  , MessageCallback(..)+  -- ** Callback Smart Contructors+  -- *** Calls and Casts (for 'Api's)+  , handleCasts+  , handleCalls+  , handleCastsAndCalls+  , handleCallsDeferred+  -- *** Generic Message Handler+  , handleMessages+  , handleSelectedMessages+  , handleAnyMessages+  , handleProcessDowns+  -- *** Fallback Handler+  , dropUnhandledMessages+  , exitOnUnhandled+  , logUnhandledMessages+  -- ** Api Composition+  , (^:)+  , fallbackHandler+  , ToServerPids(..)+  -- ** Interrupt handler+  , InterruptCallback(..)+  , stopServerOnInterrupt   ) where  import           Control.Eff+import           Control.Eff.Extend+import           Control.Eff.Log+import           Control.Eff.State.Lazy import           Control.Eff.Concurrent.Api import           Control.Eff.Concurrent.Api.Request import           Control.Eff.Concurrent.Process-import           Control.Eff.Exception-import           Control.Eff.Log-import           Control.Lens+import           Control.Monad                  ( (>=>) ) import           Data.Proxy-import           Data.Typeable                  ( Typeable-                                                , typeRep-                                                ) import           Data.Dynamic import           Control.Applicative-import           Data.Kind import           GHC.Stack-import           Data.Maybe-import           GHC.Generics import           Control.DeepSeq+import           Data.Kind+import           Data.Foldable import           Data.Default ---- | A record of callbacks, handling requests sent to a /server/ 'Process', all--- belonging to a specific 'Api' family instance.--- The values of this type can be 'serve'ed or combined via 'Servable' or--- 'ServerCallback's.-data ApiHandler api eff where-  ApiHandler ::-     { -- | A cast will not return a result directly. This is used for async-       -- methods. This returns an 'ApiServerCmd' to the server loop.-       _castCallback-         :: Maybe (Api api 'Asynchronous -> Eff eff ApiServerCmd)-      -- | A call is a blocking operation, the caller is blocked until this-      -- handler calls the reply continuation.-      -- This returns an 'ApiServerCmd' to the server loop.-     , _callCallback-         :: forall reply . Maybe (Api api ('Synchronous reply) -> (reply -> Eff eff ()) -> Eff eff ApiServerCmd)-     -- | This callback is called with @Nothing@ if one of these things happen:-     ---     --  * the process exits-     --  * '_callCallback' or '_castCallback' return 'StopApiServer'-     ---     -- If the process exist peacefully the parameter is 'NotServerCallbacking',-     -- otherwise @Just "error message..."@ if the process exits with an-     -- error.-     ---     -- The default behavior is defined in 'defaultTermination'.-     , _terminateCallback-         :: Maybe (ExitReason 'Recoverable -> Eff eff ())-     } -> ApiHandler api eff+-- | /Serve/ an 'Api' in a newly spawned process.+--+-- @since 0.13.2+spawnApiServer+  :: forall api eff+   . (ToServerPids api, HasCallStack)+  => MessageCallback api (InterruptableProcess eff)+  -> InterruptCallback (ConsProcess eff)+  -> Eff (InterruptableProcess eff) (ServerPids api)+spawnApiServer scb (InterruptCallback icb) = toServerPids (Proxy @api)+  <$> spawn (apiServerLoop scb (InterruptCallback (raise . icb))) +-- | /Serve/ an 'Api' in a newly spawned -and linked - process.+--+-- @since 0.14.2+spawnLinkApiServer+  :: forall api eff+   . (ToServerPids api, HasCallStack)+  => MessageCallback api (InterruptableProcess eff)+  -> InterruptCallback (ConsProcess eff)+  -> Eff (InterruptableProcess eff) (ServerPids api)+spawnLinkApiServer scb (InterruptCallback icb) = toServerPids (Proxy @api)+  <$> spawnLink (apiServerLoop scb (InterruptCallback (raise . icb))) -instance Default (ApiHandler api eff) where-  def = ApiHandler { _castCallback = def-                   , _callCallback = def-                   , _terminateCallback = def-                   }+-- | /Server/ an 'Api' in a newly spawned process; the callbacks have access+-- to some state initialed by the function in the first parameter.+--+-- @since 0.13.2+spawnApiServerStateful+  :: forall api eff state+   . (HasCallStack, ToServerPids api)+  => Eff (InterruptableProcess eff) state+  -> MessageCallback api (State state ': InterruptableProcess eff)+  -> InterruptCallback (State state ': ConsProcess eff)+  -> Eff (InterruptableProcess eff) (ServerPids api)+spawnApiServerStateful initEffect (MessageCallback sel cb) (InterruptCallback intCb)+  = fmap (toServerPids (Proxy @api)) $ spawnRaw $ do+    state <- provideInterruptsShutdown initEffect+    evalState state $ receiveSelectedLoop sel $ \msg -> case msg of+      Left  m -> invokeIntCb m+      Right m -> do+        s <- get+        r <- raise (provideInterrupts (evalState s (cb m)))+        case r of+          Left  i              -> invokeIntCb i+          Right (StopServer i) -> invokeIntCb i+          Right AwaitNext      -> return Nothing+ where+  invokeIntCb j = do+    l <- intCb j+    case l of+      AwaitNext                  -> return Nothing+      StopServer ProcessFinished -> return (Just ())+      StopServer k               -> exitBecause (NotRecovered k) --- | Create an 'ApiHandler' with a '_castCallback', a '_callCallback'  and---  a '_terminateCallback' implementation.-apiHandler-  :: (Api api 'Asynchronous -> Eff e ApiServerCmd)-  -> (  forall r-      . Api api ( 'Synchronous r)-     -> (r -> Eff e ())-     -> Eff e ApiServerCmd+-- | /Server/ an 'Api' in a newly spawned process; The caller provides an+-- effect handler for arbitrary effects used by the server callbacks.+--+-- @since 0.13.2+spawnApiServerEffectful+  :: forall api eff serverEff+   . ( HasCallStack+     , ToServerPids api+     , Member Interrupts serverEff+     , SetMember Process (Process eff) serverEff      )-  -> (ExitReason 'Recoverable -> Eff e ())-  -> ApiHandler api e-apiHandler c d e = ApiHandler-  { _castCallback      = Just c-  , _callCallback      = Just d-  , _terminateCallback = Just e-  }+  => (forall b . Eff serverEff b -> Eff (InterruptableProcess eff) b)+  -> MessageCallback api serverEff+  -> InterruptCallback serverEff+  -> Eff (InterruptableProcess eff) (ServerPids api)+spawnApiServerEffectful handleServerInteralEffects scb icb =+  toServerPids (Proxy @api)+    <$> spawn (handleServerInteralEffects (apiServerLoop scb icb)) --- | Like 'apiHandler' but the server will loop until an error is raised or--- the process exits.--- The callback actions won't decide wether to stop the--- server or not, instead the 'ApiServerCmd' 'HandleNextRequest' is used.-apiHandlerForever-  :: (Api api 'Asynchronous -> Eff e ())-  -> (forall r . Api api ( 'Synchronous r) -> (r -> Eff e ()) -> Eff e ())-  -> (ExitReason 'Recoverable -> Eff e ())-  -> ApiHandler api e-apiHandlerForever c d = apiHandler-  (\someCast -> c someCast >> return HandleNextRequest)-  (\someCall k -> d someCall k >> return HandleNextRequest) --- | Create an 'ApiHandler' with only a '_castCallback' implementation.-castHandler-  :: (Api api 'Asynchronous -> Eff eff ApiServerCmd) -> ApiHandler api eff-castHandler c = def { _castCallback = Just c }---- | Like 'castHandler' but the server will loop until an error is raised or--- the process exits. See 'apiHandlerForver'.-castHandlerForever-  :: (Api api 'Asynchronous -> Eff eff ()) -> ApiHandler api eff-castHandlerForever c =-  castHandler (\someCast -> c someCast >> return HandleNextRequest)---- | Create an 'ApiHandler' with only a '_callCallback' implementation.-callHandler-  :: (  forall r-      . Api api ( 'Synchronous r)-     -> (r -> Eff e ())-     -> Eff e ApiServerCmd+-- | /Server/ an 'Api' in a newly spawned process; The caller provides an+-- effect handler for arbitrary effects used by the server callbacks.+-- Links to the calling process like 'linkProcess' would.+--+-- @since 0.14.2+spawnLinkApiServerEffectful+  :: forall api eff serverEff+   . ( HasCallStack+     , ToServerPids api+     , Member Interrupts serverEff+     , SetMember Process (Process eff) serverEff      )-  -> ApiHandler api e-callHandler c = def { _callCallback = Just c }---- | Like 'callHandler' but the server will loop until an error is raised or--- the process exits. See 'apiHandlerForver'.-callHandlerForever-  :: (forall r . Api api ( 'Synchronous r) -> (r -> Eff e ()) -> Eff e ())-  -> ApiHandler api e-callHandlerForever d =-  callHandler (\someCall k -> d someCall k >> return HandleNextRequest)+  => (forall b . Eff serverEff b -> Eff (InterruptableProcess eff) b)+  -> MessageCallback api serverEff+  -> InterruptCallback serverEff+  -> Eff (InterruptableProcess eff) (ServerPids api)+spawnLinkApiServerEffectful handleServerInteralEffects scb icb =+  toServerPids (Proxy @api)+    <$> spawnLink (handleServerInteralEffects (apiServerLoop scb icb)) --- | Create an 'ApiHandler' with only a '_castCallback' and '_callCallback' implementation.-castAndCallHandler-  :: (Api api 'Asynchronous -> Eff e ApiServerCmd)-  -> (  forall r-      . Api api ( 'Synchronous r)-     -> (r -> Eff e ())-     -> Eff e ApiServerCmd+-- | Receive loop for 'Api' 'call's. This starts a receive loop for+-- a 'MessageCallback'. It is used behind the scenes by 'spawnLinkApiServerEffectful'+-- and 'spawnApiServerEffectful'.+--+-- @since 0.14.2+apiServerLoop+  :: forall api eff serverEff+   . ( HasCallStack+     , ToServerPids api+     , Member Interrupts serverEff+     , SetMember Process (Process eff) serverEff      )-  -> ApiHandler api e-castAndCallHandler c d = def { _castCallback = Just c, _callCallback = Just d }---- | Like 'castAndCallHandler' but the server will loop until an error is raised or--- the process exits. See 'apiHandlerForver'.-castAndCallHandlerForever-  :: (Api api 'Asynchronous -> Eff e ())-  -> (forall r . Api api ( 'Synchronous r) -> (r -> Eff e ()) -> Eff e ())-  -> ApiHandler api e-castAndCallHandlerForever c d = castAndCallHandler-  (\someCast -> c someCast >> return HandleNextRequest)-  (\someCall k -> d someCall k >> return HandleNextRequest)+  => MessageCallback api serverEff+  -> InterruptCallback serverEff+  -> Eff serverEff ()+apiServerLoop (MessageCallback sel cb) (InterruptCallback intCb) =+  receiveSelectedLoop+    sel+    (   either (fmap Left . intCb) (fmap Right . tryUninterrupted . cb)+    >=> handleCallbackResult+    )+ where+  handleCallbackResult+    :: Either CallbackResult (Either InterruptReason CallbackResult)+    -> Eff serverEff (Maybe ())+  handleCallbackResult (Left AwaitNext) = return Nothing+  handleCallbackResult (Left (StopServer r)) = exitBecause (NotRecovered r)+  handleCallbackResult (Right (Right AwaitNext)) = return Nothing+  handleCallbackResult (Right (Right (StopServer r))) =+    intCb r >>= handleCallbackResult . Left+  handleCallbackResult (Right (Left r)) =+    intCb r >>= handleCallbackResult . Left  -- | A command to the server loop started e.g. by 'server' or 'spawnServerWithEffects'. -- Typically returned by an 'ApiHandler' member to indicate if the server -- should continue or stop.-data ApiServerCmd where+--+-- @since 0.13.2+data CallbackResult where   -- | Tell the server to keep the server loop running-  HandleNextRequest  :: ApiServerCmd+  AwaitNext :: CallbackResult   -- | Tell the server to exit, this will make 'serve' stop handling requests without   -- exitting the process. '_terminateCallback' will be invoked with the given   -- optional reason.-  StopApiServer :: ExitReason 'Recoverable -> ApiServerCmd-  --  SendReply :: reply -> ApiServerCmd () -> ApiServerCmd (reply -> Eff eff ())-  deriving (Show, Typeable, Generic)+  StopServer :: InterruptReason -> CallbackResult+  --  SendReply :: reply -> CallbackResult () -> CallbackResult (reply -> Eff eff ())+  deriving ( Typeable ) -instance NFData ApiServerCmd -makeLenses ''ApiHandler---- | Building block for composition of 'ApiHandler'.--- A wrapper for 'ApiHandler'. Use this to combine 'ApiHandler', allowing a--- process to implement several 'Api' instances. The termination will be evenly--- propagated.--- Create this via e.g. 'Servable' instances--- To serve multiple apis use '<>' to combine server callbacks, e.g.+-- | An existential wrapper around  a 'MessageSelector' and a function that+-- handles the selected message. The @api@ type parameter is a phantom type. ----- @@@--- let f = apiHandlerServerCallback px $ ApiHandler ...---     g = apiHandlerServerCallback px $ ApiHandler ...---     h = f <> g--- in serve px h--- @@@+-- The return value if the handler function is a 'CallbackResult'. ---data ServerCallback eff =-  ServerCallback { _requestHandlerSelector :: MessageSelector (Eff eff ApiServerCmd)-                 , _terminationHandler :: ExitReason 'Recoverable -> Eff eff ()-                 }--makeLenses ''ServerCallback+-- @since 0.13.2+data MessageCallback api eff where+   MessageCallback :: MessageSelector a -> (a -> Eff eff CallbackResult) -> MessageCallback api eff -instance Semigroup (ServerCallback eff) where-  l <> r = l & requestHandlerSelector .~-                  selectDynamicMessageLazy (\x ->-                    runMessageSelector (view requestHandlerSelector l) x <|>-                    runMessageSelector (view requestHandlerSelector r) x)-             & terminationHandler .~-                  (\reason ->-                      do (l^.terminationHandler) reason-                         (r^.terminationHandler) reason)+instance Semigroup (MessageCallback api eff) where+  (MessageCallback selL runL) <> (MessageCallback selR runR) =+    MessageCallback (Left <$> selL <|> Right <$> selR) (either runL runR) -instance Monoid (ServerCallback eff) where+instance Monoid (MessageCallback api eff) where   mappend = (<>)-  mempty = ServerCallback-              { _requestHandlerSelector = selectDynamicMessageLazy (const Nothing)-              , _terminationHandler = const (return ())-              }+  mempty  = MessageCallback selectAnyMessageLazy (const (pure AwaitNext)) --- | Helper type class to allow composition of 'ApiHandler'.-class Servable a where-  -- | The effect of the callbacks-  type ServerEff a :: [Type -> Type]-  -- | The is used to let the spawn function return multiple 'Server' 'ProcessId's-  -- in a type safe way, e.g. for a tuple instance of this class-  -- @(Server a, Server b)@-  type ServerPids a-  -- | The is used to let the spawn function return multiple 'Server' 'ProcessId's-  -- in a type safe way.-  toServerPids :: proxy a -> ProcessId -> ServerPids a-  -- | Convert the value to a 'ServerCallback'-  toServerCallback-    :: (Member Interrupts (ServerEff a), SetMember Process (Process effScheduler) (ServerEff a))-    => SchedulerProxy effScheduler -> a -> ServerCallback (ServerEff a)+instance Default (MessageCallback api eff) where+  def = mempty -instance Servable (ServerCallback eff)  where-  type ServerEff (ServerCallback eff) = eff-  type ServerPids (ServerCallback eff) = ProcessId-  toServerCallback  = const id-  toServerPids = const id+-- | A smart constructor for 'MessageCallback's+--+-- @since 0.13.2+handleMessages+  :: forall eff a+   . (HasCallStack, NFData a, Typeable a)+  => (a -> Eff eff CallbackResult)+  -> MessageCallback '[] eff+handleMessages = MessageCallback selectMessage -instance Typeable a => Servable (ApiHandler a eff)  where-  type ServerEff (ApiHandler a eff) = eff-  type ServerPids (ApiHandler a eff) = Server a-  toServerCallback  = apiHandlerServerCallback-  toServerPids _ = asServer+-- | A smart constructor for 'MessageCallback's+--+-- @since 0.13.2+handleSelectedMessages+  :: forall eff a+   . HasCallStack+  => MessageSelector a+  -> (a -> Eff eff CallbackResult)+  -> MessageCallback '[] eff+handleSelectedMessages = MessageCallback -instance (ServerEff a ~ ServerEff b, Servable a, Servable b) => Servable (a, b) where-  type ServerPids (a, b) = (ServerPids a, ServerPids b)-  type ServerEff (a, b) = ServerEff a-  toServerCallback px (a, b) = toServerCallback px a <> toServerCallback px b-  toServerPids _ pid =-    ( toServerPids (Proxy :: Proxy a) pid-    , toServerPids (Proxy :: Proxy b) pid-    )+-- | A smart constructor for 'MessageCallback's+--+-- @since 0.13.2+handleAnyMessages+  :: forall eff+   . HasCallStack+  => (Dynamic -> Eff eff CallbackResult)+  -> MessageCallback '[] eff+handleAnyMessages = MessageCallback selectAnyMessageLazy --- | Receive and process incoming requests until the process exits.-serve-  :: forall a effScheduler-   . ( Servable a-     , SetMember Process (Process effScheduler) (ServerEff a)-     , Member Interrupts (ServerEff a)-     , HasCallStack-     )-  => SchedulerProxy effScheduler-  -> a-  -> Eff (ServerEff a) ()-serve px a =-  let serverCb = toServerCallback px a-      stopServer reason = do-        (serverCb ^. terminationHandler) reason-        return (Just ())-  in  receiveSelectedLoop px (serverCb ^. requestHandlerSelector) $ \case-        Left  reason   -> stopServer reason-        Right handleIt -> handleIt >>= \case-          HandleNextRequest    -> return Nothing-          StopApiServer reason -> stopServer reason+-- | A smart constructor for 'MessageCallback's+--+-- @since 0.13.2+handleCasts+  :: forall api eff+   . (HasCallStack, Typeable api, Typeable (Api api 'Asynchronous))+  => (Api api 'Asynchronous -> Eff eff CallbackResult)+  -> MessageCallback api eff+handleCasts h = MessageCallback+  (selectMessageWithLazy+    (\case+      cr@(Cast _ :: Request api) -> Just cr+      _callReq                   -> Nothing+    )+  )+  (\(Cast req :: Request api) -> h req) --- | Spawn a new process, that will receive and process incoming requests--- until the process exits.-spawnServer-  :: forall a effScheduler eff-   . ( Servable a-     , ServerEff a ~ (InterruptableProcess effScheduler)+-- | A smart constructor for 'MessageCallback's+--+-- > handleCalls SP+-- >   (\ (RentBook bookId customerId) runCall ->+-- >      runCall $ do+-- >          rentalIdE <- rentBook bookId customerId+-- >          case rentalIdE of+-- >            -- on fail we just don't send a reply, let the caller run into+-- >            -- timeout+-- >            Left err -> return (Nothing, AwaitNext)+-- >            Right rentalId -> return (Just rentalId, AwaitNext))+--+-- @since 0.13.2+handleCalls+  :: forall api eff effScheduler+   . ( HasCallStack+     , Typeable api      , SetMember Process (Process effScheduler) eff      , Member Interrupts eff-     , HasCallStack      )-  => SchedulerProxy effScheduler-  -> a-  -> Eff eff (ServerPids a)-spawnServer px a = spawnServerWithEffects px a id---- | Spawn a new process, that will receive and process incoming requests--- until the process exits. Also handle all internal effects.-spawnServerWithEffects-  :: forall a effScheduler eff-   . ( Servable a-     , SetMember Process (Process effScheduler) (ServerEff a)-     , SetMember Process (Process effScheduler) eff-     , Member Interrupts eff-     , Member Interrupts (ServerEff a)-     , HasCallStack-     )-  => SchedulerProxy effScheduler-  -> a-  -> (  Eff (ServerEff a) ()-     -> Eff (InterruptableProcess effScheduler) ()+  => (  forall secret reply+      . (Typeable reply, Typeable (Api api ( 'Synchronous reply)))+     => Api api ( 'Synchronous reply)+     -> (Eff eff (Maybe reply, CallbackResult) -> secret)+     -> secret      )-  -> Eff eff (ServerPids a)-spawnServerWithEffects px a handleEff = do-  pid <- spawn (handleEff (serve px a))-  return (toServerPids (Proxy @a) pid)+  -> MessageCallback api eff+handleCalls h = MessageCallback+  (selectMessageWithLazy+    (\case+      (Cast _ :: Request api) -> Nothing+      callReq                 -> Just callReq+    )+  )+  (\(Call callRef fromPid req :: Request api) -> h+    req+    (\resAction -> do+      (mReply, cbResult) <- resAction+      traverse_ (sendReply (mkRequestOrigin req fromPid callRef)) mReply+      return cbResult+    )+  ) --- | Wrap an 'ApiHandler' into a composable 'ServerCallback' value.-apiHandlerServerCallback-  :: forall eff effScheduler api++-- | A smart constructor for 'MessageCallback's+--+-- @since 0.13.2+handleCastsAndCalls+  :: forall api eff effScheduler    . ( HasCallStack      , Typeable api+     , Typeable (Api api 'Asynchronous)      , SetMember Process (Process effScheduler) eff      , Member Interrupts eff      )-  => SchedulerProxy effScheduler-  -> ApiHandler api eff-  -> ServerCallback eff-apiHandlerServerCallback px handlers = mempty-  { _requestHandlerSelector = selectHandlerMethod px handlers-  , _terminationHandler     = fromMaybe (const (return ()))-                                        (_terminateCallback handlers)-  }+  => (Api api 'Asynchronous -> Eff eff CallbackResult)+  -> (  forall secret reply+      . (Typeable reply, Typeable (Api api ( 'Synchronous reply)))+     => Api api ( 'Synchronous reply)+     -> (Eff eff (Maybe reply, CallbackResult) -> secret)+     -> secret+     )+  -> MessageCallback api eff+handleCastsAndCalls onCast onCall = handleCalls onCall <> handleCasts onCast --- | Try to parse an incoming message to an API request, and apply either--- the 'handleCall' method or the 'handleCast' method to it.-selectHandlerMethod-  :: forall eff effScheduler api++-- | A variation of 'handleCalls' that allows to defer a reply to a call.+--+-- @since 0.14.2+handleCallsDeferred+  :: forall api eff effScheduler    . ( HasCallStack      , Typeable api      , SetMember Process (Process effScheduler) eff      , Member Interrupts eff      )-  => SchedulerProxy effScheduler-  -> ApiHandler api eff-  -> MessageSelector (Eff eff ApiServerCmd)-selectHandlerMethod px handlers =-  selectDynamicMessageLazy (fmap (applyHandlerMethod px handlers) . fromDynamic)---- | Apply either the '_callCallback', '_castCallback' or the '_terminateCallback'--- callback to an incoming request.-applyHandlerMethod-  :: forall eff effScheduler api-   . ( Typeable api-     , SetMember Process (Process effScheduler) eff-     , Member Interrupts eff-     , HasCallStack+  => (  forall reply+      . (Typeable reply, Typeable (Api api ( 'Synchronous reply)))+     => RequestOrigin (Api api ( 'Synchronous reply))+     -> Api api ( 'Synchronous reply)+     -> Eff eff CallbackResult      )-  => SchedulerProxy effScheduler-  -> ApiHandler api eff-  -> Request api-  -> Eff eff ApiServerCmd-applyHandlerMethod px handlers (Cast request) =-  fromMaybe (unhandledCastError px) (_castCallback handlers) request-applyHandlerMethod px handlers (Call callRef fromPid request) = fromMaybe-  (unhandledCallError px)-  (_callCallback handlers)-  request-  (sendReply (mkRequestOrigin request fromPid callRef))+  -> MessageCallback api eff+handleCallsDeferred h = MessageCallback+  (selectMessageWithLazy+    (\case+      (Cast _ :: Request api) -> Nothing+      callReq                 -> Just callReq+    )+  )+  (\(Call callRef fromPid req :: Request api) ->+    h (RequestOrigin fromPid callRef) req+  ) --- | A default handler to use in '_callCallback' in 'ApiHandler'. It will call--- 'raiseError' with a nice error message.-unhandledCallError-  :: forall p x r q-   . ( Typeable p-     , HasCallStack-     , SetMember Process (Process q) r-     , Member Interrupts r-     )-  => SchedulerProxy q-  -> Api p ( 'Synchronous x)-  -> (x -> Eff r ())-  -> Eff r ApiServerCmd-unhandledCallError _px _api _ = throwError-  (ProcessError ("unhandled call on api: " ++ show (typeRep (Proxy @p))))+type family ResponseType request where+  ResponseType (Api a ('Synchronous r)) = r --- | A default handler to use in '_castCallback' in 'ApiHandler'. It will call--- 'raiseError' with a nice error message.-unhandledCastError-  :: forall p r q-   . ( Typeable p-     , HasCallStack-     , SetMember Process (Process q) r-     , Member Interrupts r-     )-  => SchedulerProxy q-  -> Api p 'Asynchronous-  -> Eff r ApiServerCmd-unhandledCastError _px _api = throwError-  (ProcessError ("unhandled cast on api: " ++ show (typeRep (Proxy @p)))) --- | Either do nothing, if the error message is @Nothing@,--- or call 'exitWithError' with the error message.-defaultTermination-  :: forall q r-   . ( HasCallStack-     , SetMember Process (Process q) r-     , Member (Logs LogMessage) r-     )-  => SchedulerProxy q-  -> ExitReason 'Recoverable-  -> Eff r ()-defaultTermination _px r = logNotice ("server process terminating " ++ show r)+-- | A smart constructor for 'MessageCallback's+--+-- @since 0.13.2+handleProcessDowns+  :: forall eff+   . HasCallStack+  => (MonitorReference -> Eff eff CallbackResult)+  -> MessageCallback '[] eff+handleProcessDowns k = MessageCallback selectMessage (k . downReference)++-- | Compose two 'Api's to a type-leve pair of them.+--+-- > handleCalls api1calls ^: handleCalls api2calls ^:+--+-- @since 0.13.2+(^:)+  :: forall (api1 :: Type) (apis2 :: [Type]) eff+   . HasCallStack+  => MessageCallback api1 eff+  -> MessageCallback apis2 eff+  -> MessageCallback (api1 ': apis2) eff+(MessageCallback selL runL) ^: (MessageCallback selR runR) =+  MessageCallback (Left <$> selL <|> Right <$> selR) (either runL runR)++infixr 5 ^:++-- | Make a fallback handler, i.e. a handler to which no other can be composed+-- to from the right.+--+-- @since 0.13.2+fallbackHandler+  :: forall api eff+   . HasCallStack+  => MessageCallback api eff+  -> MessageCallback '[] eff+fallbackHandler (MessageCallback s r) = MessageCallback s r++-- | A 'fallbackHandler' that drops the left-over messages.+--+-- @since 0.13.2+dropUnhandledMessages :: forall eff . HasCallStack => MessageCallback '[] eff+dropUnhandledMessages =+  MessageCallback selectAnyMessageLazy (const (return AwaitNext))++-- | A 'fallbackHandler' that terminates if there are unhandled messages.+--+-- @since 0.13.2+exitOnUnhandled :: forall eff . HasCallStack => MessageCallback '[] eff+exitOnUnhandled = MessageCallback selectAnyMessageLazy $ \msg ->+  return (StopServer (ProcessError ("unhandled message " ++ show msg)))++-- | A 'fallbackHandler' that drops the left-over messages.+--+-- @since 0.13.2+logUnhandledMessages+  :: forall eff+   . (Member (Logs LogMessage) eff, HasCallStack)+  => MessageCallback '[] eff+logUnhandledMessages = MessageCallback selectAnyMessageLazy $ \msg -> do+  logWarning ("ignoring unhandled message " ++ show msg)+  return AwaitNext++-- | Helper type class for the return values of 'spawnApiServer' et al.+--+-- @since 0.13.2+class ToServerPids (t :: k) where+  type ServerPids t+  toServerPids :: proxy t -> ProcessId -> ServerPids t++instance ToServerPids '[] where+  type ServerPids '[] = ProcessId+  toServerPids _ = id++instance+  forall (api1 :: Type) (api2 :: [Type])+  . (ToServerPids api1, ToServerPids api2)+  => ToServerPids (api1 ': api2) where+  type ServerPids (api1 ': api2) = (ServerPids api1, ServerPids api2)+  toServerPids _ p =+    (toServerPids (Proxy @api1) p, toServerPids (Proxy @api2) p)++instance+  forall (api1 :: Type)+  . (ToServerPids api1)+  => ToServerPids api1 where+  type ServerPids api1 = Server api1+  toServerPids _ = asServer++-- | Just a wrapper around a function that will be applied to the result of+-- a 'MessageCallback's 'StopServer' clause, or an 'InterruptReason' caught during+-- the execution of @receive@ or a 'MessageCallback'+--+-- @since 0.13.2+data InterruptCallback eff where+   InterruptCallback ::+     (InterruptReason -> Eff eff CallbackResult) -> InterruptCallback eff++instance Default (InterruptCallback eff) where+  def = stopServerOnInterrupt++-- | A smart constructor for 'InterruptCallback's+--+-- @since 0.13.2+stopServerOnInterrupt :: forall eff . HasCallStack => InterruptCallback eff+stopServerOnInterrupt = InterruptCallback (pure . StopServer)

@@ -1,504 +0,0 @@--- | Experimental new 'Api' server handler.------ @since 0.13.2-module Control.Eff.Concurrent.Api.Server2-  ( -- * Starting Api Servers-    spawnApiServer-  , spawnLinkApiServer-  , spawnApiServerStateful-  , spawnApiServerEffectful-  , spawnLinkApiServerEffectful-  , apiServerLoop-  -- ** Api Server Callbacks-  , CallbackResult(..)-  , MessageCallback(..)-  -- ** Callback Smart Contructors-  -- *** Calls and Casts (for 'Api's)-  , handleCasts-  , handleCalls-  , handleCastsAndCalls-  , handleCallsDeferred-  -- *** Generic Message Handler-  , handleMessages-  , handleSelectedMessages-  , handleAnyMessages-  , handleProcessDowns-  -- *** Fallback Handler-  , dropUnhandledMessages-  , exitOnUnhandled-  , logUnhandledMessages-  -- ** Api Composition-  , (^:)-  , fallbackHandler-  , ToServerPids(..)-  -- ** Interrupt handler-  , InterruptCallback(..)-  , stopServerOnInterrupt-  -- * Observers-  , handleObservations-  , Observing()-  )-where--import           Control.Eff-import           Control.Eff.Extend-import           Control.Eff.Log-import           Control.Eff.State.Lazy-import           Control.Eff.Concurrent.Api-import           Control.Eff.Concurrent.Api.Observer-import           Control.Eff.Concurrent.Api.Request-import           Control.Eff.Concurrent.Process-import           Control.Monad                  ( (>=>) )-import           Data.Proxy-import           Data.Dynamic-import           Control.Applicative-import           GHC.Stack-import           Control.DeepSeq-import           Data.Kind-import           Data.Foldable-import           Data.Default---- | /Serve/ an 'Api' in a newly spawned process.------ @since 0.13.2-spawnApiServer-  :: forall api eff-   . (ToServerPids api, HasCallStack)-  => MessageCallback api (InterruptableProcess eff)-  -> InterruptCallback (ConsProcess eff)-  -> Eff (InterruptableProcess eff) (ServerPids api)-spawnApiServer scb (InterruptCallback icb) = toServerPids (Proxy @api)-  <$> spawn (apiServerLoop scb (InterruptCallback (raise . icb)))---- | /Serve/ an 'Api' in a newly spawned -and linked - process.------ @since 0.14.2-spawnLinkApiServer-  :: forall api eff-   . (ToServerPids api, HasCallStack)-  => MessageCallback api (InterruptableProcess eff)-  -> InterruptCallback (ConsProcess eff)-  -> Eff (InterruptableProcess eff) (ServerPids api)-spawnLinkApiServer scb (InterruptCallback icb) = toServerPids (Proxy @api)-  <$> spawnLink (apiServerLoop scb (InterruptCallback (raise . icb)))---- | /Server/ an 'Api' in a newly spawned process; the callbacks have access--- to some state initialed by the function in the first parameter.------ @since 0.13.2-spawnApiServerStateful-  :: forall api eff state-   . (HasCallStack, ToServerPids api)-  => Eff (InterruptableProcess eff) state-  -> MessageCallback api (State state ': InterruptableProcess eff)-  -> InterruptCallback (State state ': ConsProcess eff)-  -> Eff (InterruptableProcess eff) (ServerPids api)-spawnApiServerStateful initEffect (MessageCallback sel cb) (InterruptCallback intCb)-  = fmap (toServerPids (Proxy @api)) $ spawnRaw $ do-    state <- provideInterruptsShutdown initEffect-    evalState state $ receiveSelectedLoop (SP @eff) sel $ \msg -> case msg of-      Left  m -> invokeIntCb m-      Right m -> do-        s <- get-        r <- raise (provideInterrupts (evalState s (cb m)))-        case r of-          Left  i              -> invokeIntCb i-          Right (StopServer i) -> invokeIntCb i-          Right AwaitNext      -> return Nothing- where-  invokeIntCb j = do-    l <- intCb j-    case l of-      AwaitNext                  -> return Nothing-      StopServer ProcessFinished -> return (Just ())-      StopServer k               -> exitBecause SP (NotRecovered k)---- | /Server/ an 'Api' in a newly spawned process; The caller provides an--- effect handler for arbitrary effects used by the server callbacks.------ @since 0.13.2-spawnApiServerEffectful-  :: forall api eff serverEff-   . ( HasCallStack-     , ToServerPids api-     , Member Interrupts serverEff-     , SetMember Process (Process eff) serverEff-     )-  => (forall b . Eff serverEff b -> Eff (InterruptableProcess eff) b)-  -> MessageCallback api serverEff-  -> InterruptCallback serverEff-  -> Eff (InterruptableProcess eff) (ServerPids api)-spawnApiServerEffectful handleServerInteralEffects scb icb =-  toServerPids (Proxy @api)-    <$> spawn (handleServerInteralEffects (apiServerLoop scb icb))----- | /Server/ an 'Api' in a newly spawned process; The caller provides an--- effect handler for arbitrary effects used by the server callbacks.--- Links to the calling process like 'linkProcess' would.------ @since 0.14.2-spawnLinkApiServerEffectful-  :: forall api eff serverEff-   . ( HasCallStack-     , ToServerPids api-     , Member Interrupts serverEff-     , SetMember Process (Process eff) serverEff-     )-  => (forall b . Eff serverEff b -> Eff (InterruptableProcess eff) b)-  -> MessageCallback api serverEff-  -> InterruptCallback serverEff-  -> Eff (InterruptableProcess eff) (ServerPids api)-spawnLinkApiServerEffectful handleServerInteralEffects scb icb =-  toServerPids (Proxy @api)-    <$> spawnLink (handleServerInteralEffects (apiServerLoop scb icb))---- | Receive loop for 'Api' 'call's. This starts a receive loop for--- a 'MessageCallback'. It is used behind the scenes by 'spawnLinkApiServerEffectful'--- and 'spawnApiServerEffectful'.------ @since 0.14.2-apiServerLoop-  :: forall api eff serverEff-   . ( HasCallStack-     , ToServerPids api-     , Member Interrupts serverEff-     , SetMember Process (Process eff) serverEff-     )-  => MessageCallback api serverEff-  -> InterruptCallback serverEff-  -> Eff serverEff ()-apiServerLoop (MessageCallback sel cb) (InterruptCallback intCb) =-  receiveSelectedLoop-    (SP @eff)-    sel-    (   either (fmap Left . intCb) (fmap Right . tryUninterrupted . cb)-    >=> handleCallbackResult-    )- where-  handleCallbackResult-    :: Either CallbackResult (Either InterruptReason CallbackResult)-    -> Eff serverEff (Maybe ())-  handleCallbackResult (Left AwaitNext) = return Nothing-  handleCallbackResult (Left (StopServer r)) = exitBecause SP (NotRecovered r)-  handleCallbackResult (Right (Right AwaitNext)) = return Nothing-  handleCallbackResult (Right (Right (StopServer r))) =-    intCb r >>= handleCallbackResult . Left-  handleCallbackResult (Right (Left r)) =-    intCb r >>= handleCallbackResult . Left---- | A command to the server loop started e.g. by 'server' or 'spawnServerWithEffects'.--- Typically returned by an 'ApiHandler' member to indicate if the server--- should continue or stop.------ @since 0.13.2-data CallbackResult where-  -- | Tell the server to keep the server loop running-  AwaitNext :: CallbackResult-  -- | Tell the server to exit, this will make 'serve' stop handling requests without-  -- exitting the process. '_terminateCallback' will be invoked with the given-  -- optional reason.-  StopServer :: InterruptReason -> CallbackResult-  --  SendReply :: reply -> CallbackResult () -> CallbackResult (reply -> Eff eff ())-  deriving ( Typeable )----- | An existential wrapper around  a 'MessageSelector' and a function that--- handles the selected message. The @api@ type parameter is a phantom type.------ The return value if the handler function is a 'CallbackResult'.------ @since 0.13.2-data MessageCallback api eff where-   MessageCallback :: MessageSelector a -> (a -> Eff eff CallbackResult) -> MessageCallback api eff--instance Semigroup (MessageCallback api eff) where- (MessageCallback selL runL) <> (MessageCallback selR runR) =-    MessageCallback (Left <$> selL <|> Right <$> selR) (either runL runR)--instance Monoid (MessageCallback api eff) where-  mappend = (<>)-  mempty = MessageCallback selectAnyMessageLazy (const (pure AwaitNext))--instance Default (MessageCallback api eff) where-  def = mempty---- | A smart constructor for 'MessageCallback's------ @since 0.13.2-handleMessages-  :: forall eff a-   . (HasCallStack, NFData a, Typeable a)-  => (a -> Eff eff CallbackResult)-  -> MessageCallback '[] eff-handleMessages = MessageCallback selectMessage---- | A smart constructor for 'MessageCallback's------ @since 0.13.2-handleSelectedMessages-  :: forall eff a-   . HasCallStack-  => MessageSelector a-  -> (a -> Eff eff CallbackResult)-  -> MessageCallback '[] eff-handleSelectedMessages = MessageCallback---- | A smart constructor for 'MessageCallback's------ @since 0.13.2-handleAnyMessages-  :: forall eff-   . HasCallStack-  => (Dynamic -> Eff eff CallbackResult)-  -> MessageCallback '[] eff-handleAnyMessages = MessageCallback selectAnyMessageLazy---- | A smart constructor for 'MessageCallback's------ @since 0.13.2-handleCasts-  :: forall api eff-   . (HasCallStack, Typeable api, Typeable (Api api 'Asynchronous))-  => (Api api 'Asynchronous -> Eff eff CallbackResult)-  -> MessageCallback api eff-handleCasts h = MessageCallback-  (selectMessageWithLazy-    (\case-      cr@(Cast _ :: Request api) -> Just cr-      _callReq                   -> Nothing-    )-  )-  (\(Cast req :: Request api) -> h req)---- | A smart constructor for 'MessageCallback's------ > handleCalls SP--- >   (\ (RentBook bookId customerId) runCall ->--- >      runCall $ do--- >          rentalIdE <- rentBook bookId customerId--- >          case rentalIdE of--- >            -- on fail we just don't send a reply, let the caller run into--- >            -- timeout--- >            Left err -> return (Nothing, AwaitNext)--- >            Right rentalId -> return (Just rentalId, AwaitNext))------ @since 0.13.2-handleCalls-  :: forall api eff effScheduler-   . ( HasCallStack-     , Typeable api-     , SetMember Process (Process effScheduler) eff-     , Member Interrupts eff-     )-  => SchedulerProxy effScheduler-  -> (  forall secret reply-      . (Typeable reply, Typeable (Api api ( 'Synchronous reply)))-     => Api api ( 'Synchronous reply)-     -> (Eff eff (Maybe reply, CallbackResult) -> secret)-     -> secret-     )-  -> MessageCallback api eff-handleCalls px h = MessageCallback-  (selectMessageWithLazy-    (\case-      (Cast _ :: Request api) -> Nothing-      callReq                 -> Just callReq-    )-  )-  (\(Call callRef fromPid req :: Request api) -> h-    req-    (\resAction -> do-      (mReply, cbResult) <- resAction-      traverse_ (sendReply (mkRequestOrigin req fromPid callRef)) mReply-      return cbResult-    )-  )----- | A smart constructor for 'MessageCallback's------ @since 0.13.2-handleCastsAndCalls-  :: forall api eff effScheduler-   . ( HasCallStack-     , Typeable api-     , Typeable (Api api 'Asynchronous)-     , SetMember Process (Process effScheduler) eff-     , Member Interrupts eff-     )-  => SchedulerProxy effScheduler-  -> (Api api 'Asynchronous -> Eff eff CallbackResult)-  -> (  forall secret reply-      . (Typeable reply, Typeable (Api api ( 'Synchronous reply)))-     => Api api ( 'Synchronous reply)-     -> (Eff eff (Maybe reply, CallbackResult) -> secret)-     -> secret-     )-  -> MessageCallback api eff-handleCastsAndCalls px onCast onCall =-  handleCalls px onCall <> handleCasts onCast----- | A variation of 'handleCalls' that allows to defer a reply to a call.------ @since 0.14.2-handleCallsDeferred-  :: forall api eff effScheduler-   . ( HasCallStack-     , Typeable api-     , SetMember Process (Process effScheduler) eff-     , Member Interrupts eff-     )-  => SchedulerProxy effScheduler-  -> (  forall reply-      . (Typeable reply, Typeable (Api api ( 'Synchronous reply)))-     => RequestOrigin (Api api ( 'Synchronous reply))-     -> Api api ( 'Synchronous reply)-     -> Eff eff CallbackResult-     )-  -> MessageCallback api eff-handleCallsDeferred px h = MessageCallback-  (selectMessageWithLazy-    (\case-      (Cast _ :: Request api) -> Nothing-      callReq                 -> Just callReq-    )-  )-  (\(Call callRef fromPid req :: Request api) ->-    h (RequestOrigin fromPid callRef) req-  )--type family ResponseType request where-  ResponseType (Api a ('Synchronous r)) = r----- | A smart constructor for 'MessageCallback's------ @since 0.13.2-handleProcessDowns-  :: forall eff-   . HasCallStack-  => (MonitorReference -> Eff eff CallbackResult)-  -> MessageCallback '[] eff-handleProcessDowns k = MessageCallback selectMessage (k . downReference)---- | Compose two 'Api's to a type-leve pair of them.------ > handleCalls api1calls ^: handleCalls api2calls ^:------ @since 0.13.2-(^:)-  :: forall (api1 :: Type) (apis2 :: [Type]) eff-   . HasCallStack-  => MessageCallback api1 eff-  -> MessageCallback apis2 eff-  -> MessageCallback (api1 ': apis2) eff-(MessageCallback selL runL) ^: (MessageCallback selR runR) =-  MessageCallback (Left <$> selL <|> Right <$> selR) (either runL runR)--infixr 5 ^:---- | Make a fallback handler, i.e. a handler to which no other can be composed--- to from the right.------ @since 0.13.2-fallbackHandler-  :: forall api eff-   . HasCallStack-  => MessageCallback api eff-  -> MessageCallback '[] eff-fallbackHandler (MessageCallback s r) = MessageCallback s r---- | A 'fallbackHandler' that drops the left-over messages.------ @since 0.13.2-dropUnhandledMessages :: forall eff . HasCallStack => MessageCallback '[] eff-dropUnhandledMessages =-  MessageCallback selectAnyMessageLazy (const (return AwaitNext))---- | A 'fallbackHandler' that terminates if there are unhandled messages.------ @since 0.13.2-exitOnUnhandled :: forall eff . HasCallStack => MessageCallback '[] eff-exitOnUnhandled = MessageCallback selectAnyMessageLazy $ \msg ->-  return (StopServer (ProcessError ("unhandled message " ++ show msg)))---- | A 'fallbackHandler' that drops the left-over messages.------ @since 0.13.2-logUnhandledMessages-  :: forall eff-   . (Member (Logs LogMessage) eff, HasCallStack)-  => MessageCallback '[] eff-logUnhandledMessages = MessageCallback selectAnyMessageLazy $ \msg -> do-  logWarning ("ignoring unhandled message " ++ show msg)-  return AwaitNext---- | Helper type class for the return values of 'spawnApiServer' et al.------ @since 0.13.2-class ToServerPids (t :: k) where-  type ServerPids t-  toServerPids :: proxy t -> ProcessId -> ServerPids t--instance ToServerPids '[] where-  type ServerPids '[] = ProcessId-  toServerPids _ = id--instance-  forall (api1 :: Type) (api2 :: [Type])-  . (ToServerPids api1, ToServerPids api2)-  => ToServerPids (api1 ': api2) where-  type ServerPids (api1 ': api2) = (ServerPids api1, ServerPids api2)-  toServerPids _ p =-    (toServerPids (Proxy @api1) p, toServerPids (Proxy @api2) p)--instance-  forall (api1 :: Type)-  . (ToServerPids api1)-  => ToServerPids api1 where-  type ServerPids api1 = Server api1-  toServerPids _ = asServer---- | Just a wrapper around a function that will be applied to the result of--- a 'MessageCallback's 'StopServer' clause, or an 'InterruptReason' caught during--- the execution of @receive@ or a 'MessageCallback'------ @since 0.13.2-data InterruptCallback eff where-   InterruptCallback ::-     (InterruptReason -> Eff eff CallbackResult) -> InterruptCallback eff--instance Default (InterruptCallback eff) where-  def = stopServerOnInterrupt---- | A smart constructor for 'InterruptCallback's------ @since 0.13.2-stopServerOnInterrupt :: forall eff . HasCallStack => InterruptCallback eff-stopServerOnInterrupt = InterruptCallback (pure . StopServer)---- | Apply a given callback function to incoming 'Observeration's.------ @since 0.14.1-handleObservations-  :: Typeable o-  => (Server o -> Observation o -> Eff e CallbackResult)-  -> MessageCallback (Observing o) e-handleObservations cb = handleCasts $ \(Observered s o) -> cb s o---- | An 'Api' type for generic 'Observer's, see 'handleObservations'.------ @since 0.14.1-data Observing o-  deriving Typeable---- | An 'Api' instance for generic 'Observer's, see 'handleObservations'.------ @since 0.14.1-data instance Api (Observing o) 'Asynchronous where-  Observered :: Server o -> Observation o -> Api (Observing o) 'Asynchronous--instance (Typeable o, Observable o) => Observer (Observing o) o where-  observationMessage = Observered

@@ -22,9 +22,6 @@   , ResumeProcess(..)   -- ** Scheduler Effect Identification   , SchedulerProxy(..)-  , HasScheduler-  , getSchedulerProxy-  , withSchedulerProxy   , thisSchedulerProxy   -- ** Process State   , ProcessState(..)@@ -404,27 +401,6 @@   -- | Like 'SP' but different   Scheduler :: SchedulerProxy q --- | A constraint for the implicit 'SchedulerProxy' parameter.--- Use 'getSchedulerProxy' to query it. _EXPERIMENTAL_------ @since 0.12.0-type HasScheduler q = (?_schedulerProxy :: SchedulerProxy q)---- | Get access to the 'SchedulerProxy' for the current scheduler effects.---  _EXPERIMENTAL_------ @since 0.12.0-getSchedulerProxy :: HasScheduler q => SchedulerProxy q-getSchedulerProxy = ?_schedulerProxy---- | Set the 'SchedulerProxy' to use, this satisfies 'HasScheduler' .---  _EXPERIMENTAL_------ @since 0.12.0-withSchedulerProxy :: SchedulerProxy q -> (HasScheduler q => a) -> a-withSchedulerProxy px x = let ?_schedulerProxy = px in x-- -- | /Cons/ 'Process' onto a list of effects. type ConsProcess r = Process r ': r @@ -667,10 +643,9 @@ -- via a call to 'interrupt'. exitOnInterrupt   :: (HasCallStack, Member Interrupts r, SetMember Process (Process q) r)-  => SchedulerProxy q-  -> Eff r a+  => Eff r a   -> Eff r a-exitOnInterrupt px = handleInterrupts (exitBecause px . NotRecovered)+exitOnInterrupt = handleInterrupts (exitBecause . NotRecovered)  -- | Handle 'InterruptReason's arising during process operations, e.g. -- when a linked process crashes while we wait in a 'receiveSelectedMessage'@@ -807,9 +782,8 @@ yieldProcess   :: forall r q    . (SetMember Process (Process q) r, HasCallStack, Member Interrupts r)-  => SchedulerProxy q-  -> Eff r ()-yieldProcess _ = executeAndResumeOrThrow YieldProcess+  => Eff r ()+yieldProcess = executeAndResumeOrThrow YieldProcess  -- | Send a message to a process addressed by the 'ProcessId'. -- See 'SendMessage'.@@ -821,11 +795,10 @@      , Member Interrupts r      , Typeable o      )-  => SchedulerProxy q-  -> ProcessId+  => ProcessId   -> o   -> Eff r ()-sendMessage _ pid message =+sendMessage pid message =   executeAndResumeOrThrow (SendMessage pid $! toDyn $! message)  -- | Send a 'Dynamic' value to a process addressed by the 'ProcessId'.@@ -833,11 +806,10 @@ sendAnyMessage   :: forall r q    . (SetMember Process (Process q) r, HasCallStack, Member Interrupts r)-  => SchedulerProxy q-  -> ProcessId+  => ProcessId   -> Dynamic   -> Eff r ()-sendAnyMessage _ pid message =+sendAnyMessage pid message =   rnf pid `seq` executeAndResumeOrThrow (SendMessage pid $! message)  -- | Exit a process addressed by the 'ProcessId'. The process will exit,@@ -846,11 +818,10 @@ sendShutdown   :: forall r q    . (SetMember Process (Process q) r, HasCallStack, Member Interrupts r)-  => SchedulerProxy q-  -> ProcessId+  => ProcessId   -> ExitReason 'NoRecovery   -> Eff r ()-sendShutdown _ pid s =+sendShutdown pid s =   pid `deepseq` s `deepseq` executeAndResumeOrThrow (SendShutdown pid s)  -- | Interrupts a process addressed by the 'ProcessId'. The process might exit,@@ -859,11 +830,10 @@ sendInterrupt   :: forall r q    . (SetMember Process (Process q) r, HasCallStack, Member Interrupts r)-  => SchedulerProxy q-  -> ProcessId+  => ProcessId   -> InterruptReason   -> Eff r ()-sendInterrupt _ pid s =+sendInterrupt pid s =   pid `deepseq` s `deepseq` executeAndResumeOrThrow (SendInterrupt pid s)  -- | Start a new process, the new process will execute an effect, the function@@ -923,10 +893,9 @@ isProcessAlive   :: forall r q    . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)-  => SchedulerProxy q-  -> ProcessId+  => ProcessId   -> Eff r Bool-isProcessAlive _px pid =+isProcessAlive  pid =   isJust <$> executeAndResumeOrThrow (GetProcessState pid)  -- | Block until a message was received.@@ -934,9 +903,8 @@ receiveAnyMessage   :: forall r q    . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)-  => SchedulerProxy q-  -> Eff r Dynamic-receiveAnyMessage _ =+  => Eff r Dynamic+receiveAnyMessage =   executeAndResumeOrThrow (ReceiveSelectedMessage selectAnyMessageLazy)  -- | Block until a message was received, that is not 'Nothing' after applying@@ -949,10 +917,9 @@      , SetMember Process (Process q) r      , Member Interrupts r      )-  => SchedulerProxy q-  -> MessageSelector a+  => MessageSelector a   -> Eff r a-receiveSelectedMessage _ f = executeAndResumeOrThrow (ReceiveSelectedMessage f)+receiveSelectedMessage f = executeAndResumeOrThrow (ReceiveSelectedMessage f)  -- | Receive and cast the message to some 'Typeable' instance. -- See 'ReceiveSelectedMessage' for more documentation.@@ -965,9 +932,8 @@      , SetMember Process (Process q) r      , Member Interrupts r      )-  => SchedulerProxy q-  -> Eff r a-receiveMessage px = receiveSelectedMessage px (MessageSelector fromDynamic)+  => Eff r a+receiveMessage = receiveSelectedMessage (MessageSelector fromDynamic)  -- | Remove and return all messages currently enqueued in the process message -- queue.@@ -978,7 +944,6 @@    . ( HasCallStack      , SetMember Process (Process q) r      , Member Interrupts r-     , HasScheduler q      )   => Eff r [Dynamic] flushMessages =@@ -995,50 +960,45 @@ receiveSelectedLoop   :: forall r q a endOfLoopResult    . (SetMember Process (Process q) r, HasCallStack)-  => SchedulerProxy q-  -> MessageSelector a+  => MessageSelector a   -> (Either InterruptReason a -> Eff r (Maybe endOfLoopResult))   -> Eff r endOfLoopResult-receiveSelectedLoop px selectMesage handlers = do+receiveSelectedLoop selectMesage handlers = do   mReq <- send (ReceiveSelectedMessage @q @a selectMesage)   mRes <- case mReq of     Interrupted reason  -> handlers (Left reason)     ResumeWith  message -> handlers (Right message)-  maybe (receiveSelectedLoop px selectMesage handlers) return mRes+  maybe (receiveSelectedLoop selectMesage handlers) return mRes  -- | Like 'receiveSelectedLoop' but /not selective/. -- See also 'selectAnyMessageLazy', 'receiveSelectedLoop'. receiveAnyLoop   :: forall r q endOfLoopResult    . (SetMember Process (Process q) r, HasCallStack)-  => SchedulerProxy q-  -> (Either InterruptReason Dynamic -> Eff r (Maybe endOfLoopResult))+  => (Either InterruptReason Dynamic -> Eff r (Maybe endOfLoopResult))   -> Eff r endOfLoopResult-receiveAnyLoop px = receiveSelectedLoop px selectAnyMessageLazy+receiveAnyLoop = receiveSelectedLoop selectAnyMessageLazy  -- | Like 'receiveSelectedLoop' but refined to casting to a specific 'Typeable' -- using 'selectMessageLazy'. receiveLoop   :: forall r q a endOfLoopResult    . (SetMember Process (Process q) r, HasCallStack, Typeable a)-  => SchedulerProxy q-  -> (Either InterruptReason a -> Eff r (Maybe endOfLoopResult))+  => (Either InterruptReason a -> Eff r (Maybe endOfLoopResult))   -> Eff r endOfLoopResult-receiveLoop px = receiveSelectedLoop px selectMessageLazy+receiveLoop = receiveSelectedLoop selectMessageLazy  -- | Returns the 'ProcessId' of the current process. self   :: (HasCallStack, SetMember Process (Process q) r)-  => SchedulerProxy q-  -> Eff r ProcessId-self _px = executeAndResumeOrExit SelfPid+  => Eff r ProcessId+self = executeAndResumeOrExit SelfPid  -- | Generate a unique 'Int' for the current process. makeReference   :: (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)-  => SchedulerProxy q-  -> Eff r Int-makeReference _px = executeAndResumeOrThrow MakeReference+  => Eff r Int+makeReference = executeAndResumeOrThrow MakeReference  -- | A value that contains a unique reference of a process -- monitoring.@@ -1072,10 +1032,9 @@ monitor   :: forall r q    . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)-  => SchedulerProxy q-  -> ProcessId+  => ProcessId   -> Eff r MonitorReference-monitor _px = executeAndResumeOrThrow . Monitor . force+monitor = executeAndResumeOrThrow . Monitor . force  -- | Remove a monitor created with 'monitor'. --@@ -1083,10 +1042,9 @@ demonitor   :: forall r q    . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)-  => SchedulerProxy q-  -> MonitorReference+  => MonitorReference   -> Eff r ()-demonitor _px = executeAndResumeOrThrow . Demonitor . force+demonitor = executeAndResumeOrThrow . Demonitor . force  -- | 'monitor' another process before while performing an action -- and 'demonitor' afterwards.@@ -1098,11 +1056,10 @@      , SetMember Process (Process q) r      , Member Interrupts r      )-  => SchedulerProxy q-  -> ProcessId+  => ProcessId   -> (MonitorReference -> Eff r a)   -> Eff r a-withMonitor px pid eff = monitor px pid >>= \ref -> eff ref <* demonitor px ref+withMonitor pid eff = monitor pid >>= \ref -> eff ref <* demonitor ref  -- | A 'MessageSelector' for receiving either a monitor of the -- given process or another message.@@ -1116,15 +1073,12 @@      , Typeable a      , Show a      )-  => SchedulerProxy q-  -> ProcessId+  => ProcessId   -> MessageSelector a   -> Eff r (Either ProcessDown a)-receiveWithMonitor px pid sel = withMonitor-  px+receiveWithMonitor pid sel = withMonitor   pid   (\ref -> receiveSelectedMessage-    px     (Left <$> selectProcessDown ref <|> Right <$> sel)   ) @@ -1176,10 +1130,9 @@ linkProcess   :: forall r q    . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)-  => SchedulerProxy q-  -> ProcessId+  => ProcessId   -> Eff r ()-linkProcess _px = executeAndResumeOrThrow . Link . force+linkProcess = executeAndResumeOrThrow . Link . force  -- | Unlink the calling proccess from the other process. --@@ -1187,36 +1140,32 @@ unlinkProcess   :: forall r q    . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)-  => SchedulerProxy q-  -> ProcessId+  => ProcessId   -> Eff r ()-unlinkProcess _px = executeAndResumeOrThrow . Unlink . force+unlinkProcess = executeAndResumeOrThrow . Unlink . force  -- | Exit the process with a 'ProcessExitReaon'. exitBecause   :: forall r q a    . (HasCallStack, SetMember Process (Process q) r)-  => SchedulerProxy q-  -> ExitReason 'NoRecovery+  => ExitReason 'NoRecovery   -> Eff r a-exitBecause _ = send . Shutdown @q . force+exitBecause = send . Shutdown @q . force  -- | Exit the process. exitNormally   :: forall r q a    . (HasCallStack, SetMember Process (Process q) r)-  => SchedulerProxy q-  -> Eff r a-exitNormally px = exitBecause px ExitNormally+  => Eff r a+exitNormally = exitBecause  ExitNormally  -- | Exit the process with an error. exitWithError   :: forall r q a    . (HasCallStack, SetMember Process (Process q) r)-  => SchedulerProxy q-  -> String+  => String   -> Eff r a-exitWithError px = exitBecause px . NotRecovered . ProcessError+exitWithError = exitBecause . NotRecovered . ProcessError  -- | Each process is identified by a single process id, that stays constant -- throughout the life cycle of a process. Also, message sending relies on these

@@ -86,7 +86,7 @@       Right nextAction -> do         res <- nextAction         traverse_ (lift . putStrLn . (">>> " ++)) res-        yieldProcess SP+        yieldProcess         readEvalPrintLoop queueVar    where     readAction = lift $ atomically $ do@@ -138,7 +138,7 @@   -> Server o   -> Api o 'Asynchronous   -> IO ()-submitCast sc svr request = submit sc (cast SchedulerProxy svr request)+submitCast sc svr request = submit sc (cast svr request)  -- | Combination of 'submit' and 'cast'. submitCall@@ -154,4 +154,4 @@   -> Server o   -> Api o ( 'Synchronous q)   -> IO q-submitCall sc svr request = submit sc (call SchedulerProxy svr request)+submitCall sc svr request = submit sc (call svr request)

@@ -63,10 +63,11 @@     . showString " msgQs: "     . appEndo         (foldMap-          (\(pid, msgs) -> Endo (showString "  " . shows pid . showString ": ")-            <> foldMap (\m -> Endo (shows (dynTypeRep m))) (toList msgs)+          (\(pid, msgs) ->+            Endo (showString "  " . shows pid . showString ": ")+              <> foldMap (\m -> Endo (shows (dynTypeRep m))) (toList msgs)           )-          (sts ^.. msgQs . itraversed . withIndex  )+          (sts ^.. msgQs . itraversed . withIndex)         )     ) @@ -301,26 +302,25 @@     -> OnYield r a  instance Show (OnYield r a) where-  show =-    \case-      OnFlushMessages _ -> "OnFlushMessages"-      OnYield _ -> "OnYield"-      OnSelf _ -> "OnSelf"-      OnSpawn False _ _ -> "OnSpawn"-      OnSpawn True _ _ -> "OnSpawn (link)"-      OnDone _ -> "OnDone"-      OnShutdown e -> "OnShutdown " ++ show e-      OnInterrupt e _ -> "OnInterrupt "++ show e-      OnSend toP _ _ -> "OnSend " ++ show toP-      OnRecv _ _ -> "OnRecv"-      OnGetProcessState p _ -> "OnGetProcessState " ++ show p-      OnSendShutdown p e _ -> "OnSendShutdow " ++ show p ++ " " ++ show e-      OnSendInterrupt p e _ -> "OnSendInterrupt " ++ show p ++ " " ++ show e-      OnMakeReference _ -> "OnMakeReference"-      OnMonitor p _ -> "OnMonitor " ++ show p-      OnDemonitor p _ -> "OnDemonitor " ++ show p-      OnLink p _ -> "OnLink " ++ show p-      OnUnlink p _ -> "OnUnlink " ++ show p+  show = \case+    OnFlushMessages _     -> "OnFlushMessages"+    OnYield         _     -> "OnYield"+    OnSelf          _     -> "OnSelf"+    OnSpawn False _ _     -> "OnSpawn"+    OnSpawn True  _ _     -> "OnSpawn (link)"+    OnDone     _          -> "OnDone"+    OnShutdown e          -> "OnShutdown " ++ show e+    OnInterrupt e _       -> "OnInterrupt " ++ show e+    OnSend toP _ _        -> "OnSend " ++ show toP+    OnRecv            _ _ -> "OnRecv"+    OnGetProcessState p _ -> "OnGetProcessState " ++ show p+    OnSendShutdown  p e _ -> "OnSendShutdow " ++ show p ++ " " ++ show e+    OnSendInterrupt p e _ -> "OnSendInterrupt " ++ show p ++ " " ++ show e+    OnMakeReference _     -> "OnMakeReference"+    OnMonitor   p _       -> "OnMonitor " ++ show p+    OnDemonitor p _       -> "OnDemonitor " ++ show p+    OnLink      p _       -> "OnLink " ++ show p+    OnUnlink    p _       -> "OnUnlink " ++ show p  runAsCoroutinePure   :: forall v r m@@ -357,9 +357,8 @@ handleProcess _sts Empty =   return $ Left (NotRecovered (ProcessError "no main process")) -handleProcess sts allProcs@((!processState, !pid) :<| rest)-  = let-      handleExit res = if pid == 0+handleProcess sts allProcs@((!processState, !pid) :<| rest) =+  let handleExit res = if pid == 0         then return res         else do           let (downPids, stsNew) = removeLinksTo pid sts@@ -382,128 +381,126 @@               )             )             nextTargets-    in-      case processState of-        OnDone     r    -> handleExit (Right r)+  in+    case processState of+      OnDone     r    -> handleExit (Right r) -        OnShutdown e    -> handleExit (Left e)+      OnShutdown e    -> handleExit (Left e) -        OnInterrupt e k -> do-          nextK <- diskontinue sts k e-          handleProcess sts (rest :|> (nextK, pid))+      OnInterrupt e k -> do+        nextK <- diskontinue sts k e+        handleProcess sts (rest :|> (nextK, pid)) -        OnSendInterrupt targetPid sr k -> doSendInterrupt targetPid sr k+      OnSendInterrupt targetPid sr k -> doSendInterrupt targetPid sr k -        OnSendShutdown  targetPid sr k -> do-          let allButTarget =-                Seq.filter (\(_, e) -> e /= pid && e /= targetPid) allProcs-              targets = Seq.filter (\(_, e) -> e == targetPid) allProcs-              suicide = targetPid == pid-          if suicide-            then handleExit (Left sr)-            else do-              let deliverTheGoodNews (targetState, tPid) = do-                    nextTargetState <- case targetState of-                      OnSendInterrupt _ _ _tk -> return (OnShutdown sr)-                      OnSendShutdown  _ _ _tk -> return (OnShutdown sr)-                      OnFlushMessages _tk     -> return (OnShutdown sr)-                      OnYield _tk             -> return (OnShutdown sr)-                      OnSelf  _tk             -> return (OnShutdown sr)-                      OnSend _ _ _tk          -> return (OnShutdown sr)-                      OnRecv _ _tk            -> return (OnShutdown sr)-                      OnSpawn _ _ _tk         -> return (OnShutdown sr)-                      OnDone x                -> return (OnDone x)-                      OnGetProcessState _ _tk -> return (OnShutdown sr)-                      OnShutdown sr'          -> return (OnShutdown sr')-                      OnInterrupt _er _tk     -> return (OnShutdown sr)-                      OnMakeReference _tk     -> return (OnShutdown sr)-                      OnMonitor   _ _tk       -> return (OnShutdown sr)-                      OnDemonitor _ _tk       -> return (OnShutdown sr)-                      OnLink      _ _tk       -> return (OnShutdown sr)-                      OnUnlink    _ _tk       -> return (OnShutdown sr)-                    return (nextTargetState, tPid)-              nextTargets <- _runEff sts $ traverse deliverTheGoodNews targets-              nextK       <- kontinue sts k ()-              handleProcess-                sts-                (allButTarget Seq.>< (nextTargets :|> (nextK, pid)))+      OnSendShutdown  targetPid sr k -> do+        let allButTarget =+              Seq.filter (\(_, e) -> e /= pid && e /= targetPid) allProcs+            targets = Seq.filter (\(_, e) -> e == targetPid) allProcs+            suicide = targetPid == pid+        if suicide+          then handleExit (Left sr)+          else do+            let deliverTheGoodNews (targetState, tPid) = do+                  nextTargetState <- case targetState of+                    OnSendInterrupt _ _ _tk -> return (OnShutdown sr)+                    OnSendShutdown  _ _ _tk -> return (OnShutdown sr)+                    OnFlushMessages _tk     -> return (OnShutdown sr)+                    OnYield         _tk     -> return (OnShutdown sr)+                    OnSelf          _tk     -> return (OnShutdown sr)+                    OnSend _ _ _tk          -> return (OnShutdown sr)+                    OnRecv _ _tk            -> return (OnShutdown sr)+                    OnSpawn _ _ _tk         -> return (OnShutdown sr)+                    OnDone x                -> return (OnDone x)+                    OnGetProcessState _ _tk -> return (OnShutdown sr)+                    OnShutdown sr'          -> return (OnShutdown sr')+                    OnInterrupt _er _tk     -> return (OnShutdown sr)+                    OnMakeReference _tk     -> return (OnShutdown sr)+                    OnMonitor   _ _tk       -> return (OnShutdown sr)+                    OnDemonitor _ _tk       -> return (OnShutdown sr)+                    OnLink      _ _tk       -> return (OnShutdown sr)+                    OnUnlink    _ _tk       -> return (OnShutdown sr)+                  return (nextTargetState, tPid)+            nextTargets <- _runEff sts $ traverse deliverTheGoodNews targets+            nextK       <- kontinue sts k ()+            handleProcess sts+                          (allButTarget Seq.>< (nextTargets :|> (nextK, pid))) -        OnSelf k -> do-          nextK <- kontinue sts k pid-          handleProcess sts (rest :|> (nextK, pid))+      OnSelf k -> do+        nextK <- kontinue sts k pid+        handleProcess sts (rest :|> (nextK, pid)) -        OnMakeReference k -> do-          let (ref, stsNext) = incRef sts-          nextK <- kontinue sts k ref-          handleProcess stsNext (rest :|> (nextK, pid))+      OnMakeReference k -> do+        let (ref, stsNext) = incRef sts+        nextK <- kontinue sts k ref+        handleProcess stsNext (rest :|> (nextK, pid)) -        OnYield k -> do-          sts ^. yieldEff-          nextK <- kontinue sts k ()-          handleProcess sts (rest :|> (nextK, pid))+      OnYield k -> do+        sts ^. yieldEff+        nextK <- kontinue sts k ()+        handleProcess sts (rest :|> (nextK, pid)) -        OnSend toPid msg k -> do-          nextK <- kontinue sts k ()-          handleProcess (enqueueMsg toPid msg sts) (rest :|> (nextK, pid))+      OnSend toPid msg k -> do+        nextK <- kontinue sts k ()+        handleProcess (enqueueMsg toPid msg sts) (rest :|> (nextK, pid)) -        OnGetProcessState toPid k -> do-          nextK <- kontinue sts k (getProcessState toPid sts)-          handleProcess sts (rest :|> (nextK, pid))+      OnGetProcessState toPid k -> do+        nextK <- kontinue sts k (getProcessState toPid sts)+        handleProcess sts (rest :|> (nextK, pid)) -        OnSpawn link f k -> do-          let (newPid, newSts) =-                newProcessQ (if link then Just pid else Nothing) sts-          fk    <- runAsCoroutinePure (newSts ^. runEff) (f >> exitNormally SP)-          nextK <- kontinue newSts k newPid-          handleProcess newSts (rest :|> (fk, newPid) :|> (nextK, pid))+      OnSpawn link f k -> do+        let (newPid, newSts) =+              newProcessQ (if link then Just pid else Nothing) sts+        fk    <- runAsCoroutinePure (newSts ^. runEff) (f >> exitNormally)+        nextK <- kontinue newSts k newPid+        handleProcess newSts (rest :|> (fk, newPid) :|> (nextK, pid)) -        OnFlushMessages k -> do-          let (msgs, newSts) = flushMsgs pid sts-          nextK <- kontinue newSts k msgs-          handleProcess newSts (rest :|> (nextK, pid))+      OnFlushMessages k -> do+        let (msgs, newSts) = flushMsgs pid sts+        nextK <- kontinue newSts k msgs+        handleProcess newSts (rest :|> (nextK, pid)) -        recv@(OnRecv messageSelector k) ->-          case receiveMsg pid messageSelector sts of-            Nothing -> do-              nextK <- diskontinue-                sts-                k-                (ProcessError (show pid ++ " has no message queue"))+      recv@(OnRecv messageSelector k) ->+        case receiveMsg pid messageSelector sts of+          Nothing -> do+            nextK <- diskontinue+              sts+              k+              (ProcessError (show pid ++ " has no message queue")) +            handleProcess sts (rest :|> (nextK, pid))+          Just Nothing -> if Seq.length rest == 0+            then do+              nextK <- diskontinue sts+                                   k+                                   (ProcessError (show pid ++ " deadlocked"))               handleProcess sts (rest :|> (nextK, pid))-            Just Nothing -> if Seq.length rest == 0-              then do-                nextK <- diskontinue-                  sts-                  k-                  (ProcessError (show pid ++ " deadlocked"))-                handleProcess sts (rest :|> (nextK, pid))-              else handleProcess sts (rest :|> (recv, pid))-            Just (Just (result, newSts)) -> do-              nextK <- kontinue newSts k result-              handleProcess newSts (rest :|> (nextK, pid))+            else handleProcess sts (rest :|> (recv, pid))+          Just (Just (result, newSts)) -> do+            nextK <- kontinue newSts k result+            handleProcess newSts (rest :|> (nextK, pid)) -        OnMonitor toPid k -> do-          let (ref, stsNew) = addMonitoring pid toPid sts-          nextK <- kontinue stsNew k ref-          handleProcess stsNew (rest :|> (nextK, pid))+      OnMonitor toPid k -> do+        let (ref, stsNew) = addMonitoring pid toPid sts+        nextK <- kontinue stsNew k ref+        handleProcess stsNew (rest :|> (nextK, pid)) -        OnDemonitor monRef k -> do-          let stsNew = removeMonitoring monRef sts-          nextK <- kontinue stsNew k ()-          handleProcess stsNew (rest :|> (nextK, pid))+      OnDemonitor monRef k -> do+        let stsNew = removeMonitoring monRef sts+        nextK <- kontinue stsNew k ()+        handleProcess stsNew (rest :|> (nextK, pid)) -        OnLink toPid k -> do-          let (downInterrupts, stsNew) = addLink pid toPid sts-          nextK <- case downInterrupts of-            Nothing -> kontinue stsNew k ()-            Just i  -> diskontinue stsNew k i-          handleProcess stsNew (rest :|> (nextK, pid))+      OnLink toPid k -> do+        let (downInterrupts, stsNew) = addLink pid toPid sts+        nextK <- case downInterrupts of+          Nothing -> kontinue stsNew k ()+          Just i  -> diskontinue stsNew k i+        handleProcess stsNew (rest :|> (nextK, pid)) -        OnUnlink toPid k -> do-          let (_, stsNew) = removeLinksTo toPid sts-          nextK <- kontinue stsNew k ()-          handleProcess stsNew (rest :|> (nextK, pid))+      OnUnlink toPid k -> do+        let (_, stsNew) = removeLinksTo toPid sts+        nextK <- kontinue stsNew k ()+        handleProcess stsNew (rest :|> (nextK, pid))  where   doSendInterrupt targetPid sr k = do     let suicide = targetPid == pid@@ -524,8 +521,8 @@             OnSendInterrupt _ _ tk -> tk (Interrupted sr)             OnSendShutdown  _ _ tk -> tk (Interrupted sr)             OnFlushMessages tk     -> tk (Interrupted sr)-            OnYield tk             -> tk (Interrupted sr)-            OnSelf  tk             -> tk (Interrupted sr)+            OnYield         tk     -> tk (Interrupted sr)+            OnSelf          tk     -> tk (Interrupted sr)             OnSend _ _ tk          -> tk (Interrupted sr)             OnRecv _ tk            -> tk (Interrupted sr)             OnSpawn _ _ tk         -> tk (Interrupted sr)@@ -544,11 +541,7 @@  -- | The concrete list of 'Eff'ects for running this pure scheduler on @IO@ and -- with string logging.-type LoggingAndIo =-              '[ Logs LogMessage-               , LogWriterReader LogMessage IO-               , Lift IO-               ]+type LoggingAndIo = '[Logs LogMessage, LogWriterReader LogMessage IO, Lift IO]  -- | A 'SchedulerProxy' for 'LoggingAndIo'. singleThreadedIoScheduler :: SchedulerProxy LoggingAndIo

@@ -44,11 +44,10 @@      , NFData a      , Show a      )-  => SchedulerProxy q-  -> Timeout+  => Timeout   -> Eff r (Maybe a)-receiveAfter px t =-  either (const Nothing) Just <$> receiveSelectedAfter px (selectMessage @a) t+receiveAfter t =+  either (const Nothing) Just <$> receiveSelectedAfter (selectMessage @a) t  -- | Wait for a message of the given type for the given time. When no message -- arrives in time, return 'Left' 'TimerElapsed'. This is based on@@ -63,16 +62,14 @@      , Member Interrupts r      , Show a      )-  => SchedulerProxy q-  -> MessageSelector a+  => MessageSelector a   -> Timeout   -> Eff r (Either TimerElapsed a)-receiveSelectedAfter px sel t = do-  timerRef <- startTimer px t+receiveSelectedAfter sel t = do+  timerRef <- startTimer t   res      <- receiveSelectedMessage-    px     (Left <$> selectTimerElapsed timerRef <|> Right <$> sel)-  cancelTimer px timerRef+  cancelTimer timerRef   return res  -- | A 'MessageSelector' matching 'TimerElapsed' messages created by@@ -92,7 +89,7 @@  instance Show Timeout where   showsPrec d (TimeoutMicros t) =-    showParen (d>=10) (showString "timeout: " . shows t . showString " µs")+    showParen (d >= 10) (showString "timeout: " . shows t . showString " µs")  -- | The reference to a timer started by 'startTimer', required to stop -- a timer via 'cancelTimer'.@@ -103,7 +100,7 @@  instance Show TimerReference where   showsPrec d (TimerReference t) =-    showParen (d>=10) (showString "timer: " . shows t)+    showParen (d >= 10) (showString "timer: " . shows t)  -- | A value to be sent when timer started with 'startTimer' has elapsed. --@@ -113,7 +110,7 @@  instance Show TimerElapsed where   showsPrec d (TimerElapsed t) =-    showParen (d>=10) (shows t . showString " elapsed")--+    showParen (d >= 10) (shows t . showString " elapsed")-- -- @since 0.12.0  @@ -131,20 +128,16 @@      , Typeable message      , NFData message      )-  => SchedulerProxy q-  -> ProcessId+  => ProcessId   -> Timeout   -> (TimerReference -> message)   -> Eff r TimerReference-sendAfter px pid (TimeoutMicros 0) mkMsg = TimerReference <$> spawn-  (   yieldProcess px-  >>  self px-  >>= (sendMessage SP pid . force . mkMsg . TimerReference)-  )-sendAfter px pid (TimeoutMicros t) mkMsg = TimerReference <$> spawn+sendAfter pid (TimeoutMicros 0) mkMsg = TimerReference <$> spawn+  (yieldProcess >> self >>= (sendMessage pid . force . mkMsg . TimerReference))+sendAfter pid (TimeoutMicros t) mkMsg = TimerReference <$> spawn   (   liftIO (threadDelay t)-  >>  self px-  >>= (sendMessage SP pid . force . mkMsg . TimerReference)+  >>  self+  >>= (sendMessage pid . force . mkMsg . TimerReference)   )  -- | Start a new timer, after the time has elapsed, 'TimerElapsed' is sent to@@ -161,12 +154,11 @@      , SetMember Process (Process q) r      , Member Interrupts r      )-  => SchedulerProxy q-  -> Timeout+  => Timeout   -> Eff r TimerReference-startTimer px t = do-  p <- self px-  sendAfter px p t TimerElapsed+startTimer t = do+  p <- self+  sendAfter p t TimerElapsed  -- | Cancel a timer started with 'startTimer'. --@@ -178,7 +170,6 @@      , SetMember Process (Process q) r      , Member Interrupts r      )-  => SchedulerProxy q-  -> TimerReference+  => TimerReference   -> Eff r ()-cancelTimer px (TimerReference tr) = sendShutdown px tr ExitNormally+cancelTimer (TimerReference tr) = sendShutdown tr ExitNormally

@@ -45,21 +45,14 @@                   doExit                 lift (threadDelay 100000) -                me <- self forkIoScheduler-                spawn_-                  (lift (threadDelay 10000) >> sendMessage forkIoScheduler me ()-                  )-                eres <- receiveWithMonitor forkIoScheduler-                                           p1-                                           (selectMessage @())+                me <- self+                spawn_ (lift (threadDelay 10000) >> sendMessage me ())+                eres <- receiveWithMonitor p1 (selectMessage @())                 case eres of                   Left  _down -> lift (atomically (putTMVar aVar False))-                  Right ()    -> withMonitor forkIoScheduler p1 $ \ref -> do-                    sendShutdown forkIoScheduler-                                 p1-                                 (NotRecovered (ProcessError "test 123"))-                    _down <- receiveSelectedMessage forkIoScheduler-                                                    (selectProcessDown ref)+                  Right ()    -> withMonitor p1 $ \ref -> do+                    sendShutdown p1 (NotRecovered (ProcessError "test 123"))+                    _down <- receiveSelectedMessage (selectProcessDown ref)                     lift (atomically (putTMVar aVar True))               )             )@@ -105,7 +98,7 @@       (1000 :: Natural)       def       (Scheduler.defaultMainWithLogChannel-        (void (spawn (void (receiveAnyMessage forkIoScheduler))))+        (void (spawn (void receiveAnyMessage)))       )     )   )@@ -119,8 +112,8 @@       def       (Scheduler.defaultMainWithLogChannel         (do-          void (spawn (void (receiveAnyMessage forkIoScheduler)))-          void (exitNormally forkIoScheduler)+          void (spawn (void receiveAnyMessage))+          void exitNormally           fail "This should not happen!!"         )       )@@ -138,11 +131,8 @@         def         (Scheduler.defaultMainWithLogChannel           (do-            void-              (spawn-                (foreverCheap (void (sendMessage forkIoScheduler 1000 "test")))-              )-            void (exitNormally forkIoScheduler)+            void (spawn (foreverCheap (void (sendMessage 1000 "test"))))+            void exitNormally             fail "This should not happen!!"           )         )@@ -159,8 +149,8 @@       def       (Scheduler.defaultMainWithLogChannel         (do-          child <- spawn (void (receiveMessage @String forkIoScheduler))-          sendMessage forkIoScheduler child "test"+          child <- spawn (void (receiveMessage @String))+          sendMessage child "test"           return ()         )       )@@ -177,14 +167,13 @@       (Scheduler.defaultMainWithLogChannel         (do           child <- spawn $ void $ provideInterrupts $ exitOnInterrupt-            forkIoScheduler             (do-              void (receiveMessage @String forkIoScheduler)-              void (exitNormally forkIoScheduler)+              void (receiveMessage @String)+              void exitNormally               error "This should not happen (child)!!"             )-          sendMessage forkIoScheduler child "test"-          void (exitNormally forkIoScheduler)+          sendMessage child "test"+          void exitNormally           error "This should not happen!!"         )       )@@ -200,24 +189,24 @@     $ do         let n = 100             testMsg :: Float-            testMsg = 123-            flushMsgs px = do-              res <- receiveSelectedAfter px (selectDynamicMessageLazy Just) 0+            testMsg   = 123+            flushMsgs = do+              res <- receiveSelectedAfter (selectDynamicMessageLazy Just) 0               case res of                 Left  _to -> return ()-                Right _   -> flushMsgs px-        me <- self SP+                Right _   -> flushMsgs+        me <- self         spawn_           (do-            replicateM_ n $ sendMessage SP me "bad message"-            replicateM_ n $ sendMessage SP me (3123 :: Integer)-            sendMessage SP me testMsg+            replicateM_ n $ sendMessage me "bad message"+            replicateM_ n $ sendMessage me (3123 :: Integer)+            sendMessage me testMsg           )         do-          res <- receiveAfter @Float SP 1000000+          res <- receiveAfter @Float 1000000           lift (res @?= Just testMsg)-        flushMsgs SP-        res <- receiveSelectedAfter SP (selectDynamicMessageLazy Just) 10000+        flushMsgs+        res <- receiveSelectedAfter (selectDynamicMessageLazy Just) 10000         case res of           Left  _ -> return ()           Right x -> lift (False @? "unexpected message in queue " ++ show x)

@@ -16,99 +16,94 @@                                                as Scheduler  test_loopTests :: TestTree-test_loopTests-    = let soMany = 1000000-      in-          setTravisTestOptions $ testGroup-              "Loops without space leaks"-              [ testCase-                      "scheduleMonadIOEff with many yields from replicateCheapM_"-                  $ do-                        res <--                            Scheduler.scheduleIOWithLogging-                                (multiMessageLogWriter-                                    ($! (putStrLn . (">>> " ++)))-                                )-                            $ replicateCheapM_ soMany-                            $ yieldProcess SP-                        res @=? Right ()-              , testCase-                      "replicateCheapM_ of strict Int increments via the state effect"-                  $ do-                        let-                            res = run-                                (execState-                                    (0 :: Int)-                                    ( replicateCheapM_ soMany-                                    $ modify (force . (+ 1))-                                    )-                                )-                        res @=? soMany-              , testCase-                      "'foreverCheap' inside a child process and 'replicateCheapM_' in the main process"-                  $ do-                        res <--                            Scheduler.scheduleIOWithLogging-                                    (multiMessageLogWriter-                                        ($! (putStrLn . (">>> " ++)))-                                    )+test_loopTests =+    let soMany = 1000000+    in+        setTravisTestOptions $ testGroup+            "Loops without space leaks"+            [ testCase+                    "scheduleMonadIOEff with many yields from replicateCheapM_"+                $ do+                      res <-+                          Scheduler.scheduleIOWithLogging+                                  (multiMessageLogWriter+                                      ($! (putStrLn . (">>> " ++)))+                                  )+                              $ replicateCheapM_ soMany yieldProcess+                      res @=? Right ()+            , testCase+                    "replicateCheapM_ of strict Int increments via the state effect"+                $ do+                      let+                          res = run+                              (execState+                                  (0 :: Int)+                                  ( replicateCheapM_ soMany+                                  $ modify (force . (+ 1))+                                  )+                              )+                      res @=? soMany+            , testCase+                    "'foreverCheap' inside a child process and 'replicateCheapM_' in the main process"+                $ do+                      res <-+                          Scheduler.scheduleIOWithLogging+                                  (multiMessageLogWriter+                                      ($! (putStrLn . (">>> " ++)))+                                  ) -                                $ do-                                      me <- self SP-                                      spawn_-                                          (foreverCheap $ sendMessage SP me ()-                                          )-                                      replicateCheapM_-                                          soMany-                                          (void (receiveMessage @() SP))+                              $ do+                                    me <- self+                                    spawn_ (foreverCheap $ sendMessage me ())+                                    replicateCheapM_+                                        soMany+                                        (void (receiveMessage @())) -                        res @=? Right ()-              ]+                      res @=? Right ()+            ]   test_loopWithLeaksTests :: TestTree-test_loopWithLeaksTests-    = let soMany = 1000000-      in-          setTravisTestOptions $ testGroup-              "Loops WITH space leaks"-              [ testCase "scheduleMonadIOEff with many yields from replicateM_"-                  $ do-                        res <--                            Scheduler.scheduleIOWithLogging-                                (multiMessageLogWriter-                                    ($! (putStrLn . (">>> " ++)))-                                )-                            $ replicateM_ soMany-                            $ yieldProcess SP-                        res @=? Right ()-              , testCase-                      "replicateM_ of strict Int increments via the state effect"-                  $ do-                        let-                            res =-                                run-                                    (execState-                                        (0 :: Int)-                                        ( replicateM_ soMany-                                        $ modify (force . (+ 1))-                                        )-                                    )-                        res @=? soMany-              , testCase-                      "'forever' inside a child process and 'replicateM_' in the main process"-                  $ do-                        res <--                            Scheduler.scheduleIOWithLogging-                                    (multiMessageLogWriter-                                        ($! (putStrLn . (">>> " ++)))-                                    )-                                $ do-                                      me <- self SP-                                      spawn_ (forever $ sendMessage SP me ())-                                      replicateM_-                                          soMany-                                          (void (receiveMessage @() SP))+test_loopWithLeaksTests =+    let soMany = 1000000+    in+        setTravisTestOptions $ testGroup+            "Loops WITH space leaks"+            [ testCase "scheduleMonadIOEff with many yields from replicateM_"+                $ do+                      res <-+                          Scheduler.scheduleIOWithLogging+                                  (multiMessageLogWriter+                                      ($! (putStrLn . (">>> " ++)))+                                  )+                              $ replicateM_ soMany yieldProcess+                      res @=? Right ()+            , testCase+                    "replicateM_ of strict Int increments via the state effect"+                $ do+                      let+                          res =+                              run+                                  (execState+                                      (0 :: Int)+                                      ( replicateM_ soMany+                                      $ modify (force . (+ 1))+                                      )+                                  )+                      res @=? soMany+            , testCase+                    "'forever' inside a child process and 'replicateM_' in the main process"+                $ do+                      res <-+                          Scheduler.scheduleIOWithLogging+                                  (multiMessageLogWriter+                                      ($! (putStrLn . (">>> " ++)))+                                  )+                              $ do+                                    me <- self+                                    spawn_ (forever $ sendMessage me ())+                                    replicateM_ soMany+                                                (void (receiveMessage @())) -                        res @=? Right ()-              ]+                      res @=? Right ()+            ]

@@ -4,1029 +4,971 @@ import           Data.Foldable                  ( traverse_ ) import qualified Data.Dynamic                  as Dynamic import           Data.Typeable-import           Data.Default-import           Control.Exception-import           Control.Concurrent-import           Control.Concurrent.STM-import           Control.Eff.Concurrent.Process-import           Control.Eff.Concurrent.Process.Timer-import           Control.Eff.Concurrent.Api-import           Control.Eff.Concurrent.Api.Client-import           Control.Eff.Concurrent.Api.Server-import qualified Control.Eff.Concurrent.Process.ForkIOScheduler-                                               as ForkIO-import qualified Control.Eff.Concurrent.Process.SingleThreadedScheduler-                                               as SingleThreaded-import           Control.Eff-import           Control.Eff.Extend-import           Control.Eff.Log-import           Control.Eff.Loop-import           Control.Eff.Lift-import           Control.Monad-import           Test.Tasty-import           Test.Tasty.HUnit-import           Common-import           Control.Applicative-import           Data.Void--testInterruptReason :: InterruptReason-testInterruptReason = ProcessError "test interrupt"--test_forkIo :: TestTree-test_forkIo = setTravisTestOptions $ withTestLogC-  (\c lc -> handleLoggingAndIO_ (ForkIO.schedule c) lc)-  (\factory -> testGroup "ForkIOScheduler" [allTests factory])---test_singleThreaded :: TestTree-test_singleThreaded = setTravisTestOptions $ withTestLogC-  (\e logC ->-        -- void (runLift (logToChannel logC (SingleThreaded.schedule (return ()) e)))-    let runEff-          :: Eff '[Logs LogMessage, LogWriterReader LogMessage IO, Lift IO] a-          -> IO a-        runEff = flip handleLoggingAndIO logC-    in  void $ SingleThreaded.scheduleM runEff yield e-  )-  (\factory -> testGroup "SingleThreadedScheduler" [allTests factory])--allTests-  :: forall r-   . (Member (Logs LogMessage) r, SetMember Lift (Lift IO) r)-  => IO (Eff (InterruptableProcess r) () -> IO ())-  -> TestTree-allTests schedulerFactory = localOption-  (timeoutSeconds 300)-  (testGroup-    "Process"-    [ errorTests schedulerFactory-    , sendShutdownTests schedulerFactory-    , concurrencyTests schedulerFactory-    , exitTests schedulerFactory-    , pingPongTests schedulerFactory-    , yieldLoopTests schedulerFactory-    , selectiveReceiveTests schedulerFactory-    , linkingTests schedulerFactory-    , monitoringTests schedulerFactory-    , timerTests schedulerFactory-    ]-  )--data ReturnToSender-  deriving Typeable--data instance Api ReturnToSender r where-  ReturnToSender :: ProcessId -> String -> Api ReturnToSender ('Synchronous Bool)-  StopReturnToSender :: Api ReturnToSender ('Synchronous ())--deriving instance Show (Api ReturnToSender x)--deriving instance Typeable (Api ReturnToSender x)--returnToSender-  :: forall q r-   . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)-  => SchedulerProxy q-  -> Server ReturnToSender-  -> String-  -> Eff r Bool-returnToSender px toP msg = do-  me      <- self px-  _       <- call px toP (ReturnToSender me msg)-  msgEcho <- receiveMessage @String px-  return (msgEcho == msg)--stopReturnToSender-  :: forall q r-   . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)-  => SchedulerProxy q-  -> Server ReturnToSender-  -> Eff r ()-stopReturnToSender px toP = call px toP StopReturnToSender--returnToSenderServer-  :: forall q r-   . ( HasCallStack-     , SetMember Process (Process q) r-     , Member (Logs LogMessage) q-     , Member Interrupts r-     )-  => SchedulerProxy q-  -> Eff r (Server ReturnToSender)-returnToSenderServer px = asServer <$> spawn-  (serve px $ def-    { _callCallback = Just-                        (\m k -> case m of-                          StopReturnToSender -> do-                            k ()-                            return (StopApiServer testInterruptReason)-                          ReturnToSender fromP echoMsg -> do-                            sendMessage px fromP echoMsg-                            yieldProcess px-                            k True-                            return HandleNextRequest-                        )-    }-  )--selectiveReceiveTests-  :: forall r-   . (Member (Logs LogMessage) r, SetMember Lift (Lift IO) r)-  => IO (Eff (InterruptableProcess r) () -> IO ())-  -> TestTree-selectiveReceiveTests schedulerFactory = setTravisTestOptions-  (testGroup-    "selective receive tests"-    [ testCase "send 10 messages (from 1..10) and receive messages from 10 to 1"-    $ applySchedulerFactory schedulerFactory-    $ do-        let-          nMax = 10-          receiverLoop donePid = go nMax-           where-            go :: Int -> Eff (InterruptableProcess r) ()-            go 0 = sendMessage SP donePid True-            go n = do-              void $ receiveSelectedMessage SP (filterMessage (== n))-              go (n - 1)--          senderLoop receviverPid =-            traverse_ (sendMessage SP receviverPid) [1 .. nMax]--        me          <- self SP-        receiverPid <- spawn (receiverLoop me)-        spawn_ (senderLoop receiverPid)-        ok <- receiveMessage @Bool SP-        lift (ok @? "selective receive failed")-    , testCase "receive a message while waiting for a call reply"-    $ applySchedulerFactory schedulerFactory-    $ do-        srv <- returnToSenderServer SP-        ok  <- returnToSender SP srv "test"-        ()  <- stopReturnToSender SP srv-        lift (ok @? "selective receive failed")-    , testCase "flush messages"-    $ applySchedulerFactory schedulerFactory-    $ withSchedulerProxy SP-    $ do-        let px = SP-        me <- self px-        spawn_-          $  replicateM_ 10 (sendMessage px me True)-          >> sendMessage px me ()-        spawn_-          $  replicateM_ 10 (sendMessage px me (123.23411 :: Float))-          >> sendMessage px me ()-        spawn_-          $  replicateM_ 10 (sendMessage px me "123")-          >> sendMessage px me ()-        ()   <- receiveMessage px-        ()   <- receiveMessage px-        ()   <- receiveMessage px-        -- replicateCheapM_ 40 (yieldProcess px)-        msgs <- flushMessages-        lift (length msgs @?= 30)-    ]-  )---yieldLoopTests-  :: forall r-   . (Member (Logs LogMessage) r, SetMember Lift (Lift IO) r)-  => IO (Eff (InterruptableProcess r) () -> IO ())-  -> TestTree-yieldLoopTests schedulerFactory-  = let maxN = 100000-    in-      setTravisTestOptions-        (testGroup-          "yield tests"-          [ testCase-            "yield many times (replicateM_)"-            (applySchedulerFactory schedulerFactory-                                   (replicateM_ maxN (yieldProcess SP))-            )-          , testCase-            "yield many times (forM_)"-            (applySchedulerFactory-              schedulerFactory-              (forM_ [1 :: Int .. maxN] (\_ -> yieldProcess SP))-            )-          , testCase-            "construct an effect with an exit first, followed by many yields"-            (applySchedulerFactory-              schedulerFactory-              (do-                void (exitNormally SP)-                replicateM_ 1000000000000 (yieldProcess SP)-              )-            )-          ]-        )---data Ping = Ping ProcessId-  deriving (Eq, Show)--data Pong = Pong-  deriving (Eq, Show)--pingPongTests-  :: forall r-   . (Member (Logs LogMessage) r, SetMember Lift (Lift IO) r)-  => IO (Eff (InterruptableProcess r) () -> IO ())-  -> TestTree-pingPongTests schedulerFactory = testGroup-  "Yield Tests"-  [ testCase "ping pong a message between two processes, both don't yield"-  $ applySchedulerFactory schedulerFactory-  $ do-      let pongProc = foreverCheap $ do-            Ping pinger <- receiveMessage SP-            sendMessage SP pinger Pong-          pingProc ponger parent = do-            me <- self SP-            sendMessage SP ponger (Ping me)-            Pong <- receiveMessage SP-            sendMessage SP parent True-      pongPid <- spawn pongProc-      me      <- self SP-      spawn_ (pingProc pongPid me)-      ok <- receiveMessage @Bool SP-      lift (ok @? "ping pong failed")-  , testCase "ping pong a message between two processes, with massive yielding"-  $ applySchedulerFactory schedulerFactory-  $ do-      yieldProcess SP-      let pongProc = foreverCheap $ do-            yieldProcess SP-            Ping pinger <- receiveMessage SP-            yieldProcess SP-            sendMessage SP pinger Pong-            yieldProcess SP-          pingProc ponger parent = do-            yieldProcess SP-            me <- self SP-            yieldProcess SP-            sendMessage SP ponger (Ping me)-            yieldProcess SP-            Pong <- receiveMessage SP-            yieldProcess SP-            sendMessage SP parent True-            yieldProcess SP-      yieldProcess SP-      pongPid <- spawn pongProc-      yieldProcess SP-      me <- self SP-      yieldProcess SP-      spawn_ (pingProc pongPid me)-      yieldProcess SP-      ok <- receiveMessage @Bool SP-      yieldProcess SP-      lift (ok @? "ping pong failed")-      yieldProcess SP-  , testCase-    "the first message is not delayed, not even in cooperative scheduling (because of yield)"-  $ applySchedulerFactory schedulerFactory-  $ do-      pongVar <- lift newEmptyMVar-      let pongProc = foreverCheap $ do-            Pong <- receiveMessage SP-            lift (putMVar pongVar Pong)-      ponger <- spawn pongProc-      sendMessage SP ponger Pong-      let waitLoop = do-            p <- lift (tryTakeMVar pongVar)-            case p of-              Nothing -> do-                yieldProcess SP-                waitLoop-              Just r -> return r-      p <- waitLoop-      lift (p == Pong @? "ping pong failed")-  ]--errorTests-  :: forall r-   . (Member (Logs LogMessage) r, SetMember Lift (Lift IO) r)-  => IO (Eff (InterruptableProcess r) () -> IO ())-  -> TestTree-errorTests schedulerFactory-  = let-      px :: SchedulerProxy r-      px = SchedulerProxy-    in-      testGroup-        "causing and handling errors"-        [ testGroup-            "exitWithError"-            [ testCase "unhandled exitWithError"-            $ applySchedulerFactory schedulerFactory-            $ do-                void $ exitWithError px "test error"-                error "This should not happen"-            , testCase "cannot catch exitWithError"-            $ applySchedulerFactory schedulerFactory-            $ do-                void $ exitWithError px "test error 4"-                error "This should not happen"-            , testCase "multi process exitWithError"-            $ scheduleAndAssert schedulerFactory-            $ \assertEff -> do-                me <- self px-                let n = 15-                traverse_-                  (\(i :: Int) -> spawn $ foreverCheap-                    (void-                      (  sendMessage px (888888 + fromIntegral i) "test message"-                      >> yieldProcess px-                      )-                    )-                  )-                  [0 .. n]-                traverse_-                  (\(i :: Int) -> spawn $ do-                    void $ sendMessage px me i-                    void (exitWithError px (show i ++ " died"))-                    error "this should not be reached"-                  )-                  [0 .. n]-                oks <- replicateM (length [0 .. n]) (receiveMessage px)-                assertEff "" (sort oks == [0 .. n])-            ]-        ]--concurrencyTests-  :: forall r-   . (Member (Logs LogMessage) r, SetMember Lift (Lift IO) r)-  => IO (Eff (InterruptableProcess r) () -> IO ())-  -> TestTree-concurrencyTests schedulerFactory-  = let-      px :: SchedulerProxy r-      px = SchedulerProxy-      n  = 100-    in-      testGroup-        "concurrency tests"-        [ testCase-          "when main process exits the scheduler kills/cleans and returns"-        $ applySchedulerFactory schedulerFactory-        $ do-            me <- self px-            traverse_-              (const-                (spawn-                  (do-                    m <- receiveAnyMessage px-                    void (sendMessage px me m)-                  )-                )-              )-              [1 .. n]-            lift (threadDelay 1000)-        , testCase "new processes are executed before the parent process"-        $ scheduleAndAssert schedulerFactory-        $ \assertEff -> do -- start massive amount of children that exit as soon as they are-               -- executed, this will only work smoothly when scheduler schedules-               -- the new child before the parent-            traverse_ (const (spawn (exitNormally px))) [1 .. n]-            assertEff "" True-        , testCase "two concurrent processes"-        $ scheduleAndAssert schedulerFactory-        $ \assertEff -> do-            me     <- self px-            child1 <- spawn-              (do-                m <- receiveAnyMessage px-                void (sendAnyMessage px me m)-              )-            child2 <- spawn (foreverCheap (void (sendMessage px 888888 "")))-            sendMessage px child1 "test"-            i <- receiveMessage px-            sendInterrupt px child2 testInterruptReason-            assertEff "" (i == "test")-        , testCase "most processes send foreverCheap"-        $ scheduleAndAssert schedulerFactory-        $ \assertEff -> do-            me <- self px-            traverse_-              (\(i :: Int) -> spawn $ do-                when (i `rem` 5 == 0) $ void $ sendMessage px me i-                foreverCheap $ void (sendMessage px 888 "test message to 888")-              )-              [0 .. n]-            oks <- replicateM (length [0, 5 .. n]) (receiveMessage px)-            assertEff "" (sort oks == [0, 5 .. n])-        , testCase "most processes self foreverCheap"-        $ scheduleAndAssert schedulerFactory-        $ \assertEff -> do-            me <- self px-            traverse_-              (\(i :: Int) -> spawn $ do-                when (i `rem` 5 == 0) $ void $ sendMessage px me i-                foreverCheap $ void (self px)-              )-              [0 .. n]-            oks <- replicateM (length [0, 5 .. n]) (receiveMessage px)-            assertEff "" (sort oks == [0, 5 .. n])-        , testCase "most processes sendShutdown foreverCheap"-        $ scheduleAndAssert schedulerFactory-        $ \assertEff -> do-            me <- self px-            traverse_-              (\(i :: Int) -> spawn $ do-                when (i `rem` 5 == 0) $ void $ sendMessage px me i-                foreverCheap $ void (sendShutdown px 999 ExitNormally)-              )-              [0 .. n]-            oks <- replicateM (length [0, 5 .. n]) (receiveMessage px)-            assertEff "" (sort oks == [0, 5 .. n])-        , testCase "most processes spawn foreverCheap"-        $ scheduleAndAssert schedulerFactory-        $ \assertEff -> do-            me <- self px-            traverse_-              (\(i :: Int) -> spawn $ do-                when (i `rem` 5 == 0) $ void $ sendMessage px me i-                parent <- self px-                foreverCheap-                  $ void-                      (spawn-                        (void (sendMessage px parent "test msg from child"))-                      )-              )-              [0 .. n]-            oks <- replicateM (length [0, 5 .. n]) (receiveMessage px)-            assertEff "" (sort oks == [0, 5 .. n])-        , testCase "most processes receive foreverCheap"-        $ scheduleAndAssert schedulerFactory-        $ \assertEff -> do-            me <- self px-            traverse_-              (\(i :: Int) -> spawn $ do-                when (i `rem` 5 == 0) $ void $ sendMessage px me i-                foreverCheap $ void (receiveAnyMessage px)-              )-              [0 .. n]-            oks <- replicateM (length [0, 5 .. n]) (receiveMessage px)-            assertEff "" (sort oks == [0, 5 .. n])-        ]--exitTests-  :: forall r-   . (Member (Logs LogMessage) r, SetMember Lift (Lift IO) r)-  => IO (Eff (InterruptableProcess r) () -> IO ())-  -> TestTree-exitTests schedulerFactory =-  let-    px :: SchedulerProxy r-    px = SchedulerProxy-  in-    testGroup "process exit tests"-      $ [ testGroup-          "async exceptions"-          [ testCase-                (  "a process dies immediately if a "-                ++ show e-                ++ " is thrown, while "-                ++ busyWith-                )-              $ do-                  tidVar           <- newEmptyTMVarIO-                  schedulerDoneVar <- newEmptyTMVarIO-                  void $ forkIO $ do-                    void-                      $ try @SomeException-                      $ void-                      $ applySchedulerFactory schedulerFactory-                      $ do-                          tid <- lift $ myThreadId-                          lift $ atomically $ putTMVar tidVar tid-                          foreverCheap busyEffect-                    atomically $ putTMVar schedulerDoneVar ()-                  tid <- atomically $ takeTMVar tidVar-                  threadDelay 1000-                  throwTo tid e-                  void $ atomically $ takeTMVar schedulerDoneVar-          | e <- [ThreadKilled, UserInterrupt, HeapOverflow, StackOverflow]-          , (busyWith, busyEffect) <--            [ ( "receiving"-              , void-                (send-                  (ReceiveSelectedMessage @r (filterMessage (== "test message"))-                  )-                )-              )-            , ( "sending"-              , void-                (send (SendMessage @r 44444 (Dynamic.toDyn "test message")))-              )-            , ( "sending shutdown"-              , void (send (SendShutdown @r 44444 ExitNormally))-              )-            , ("selfpid-ing", void (send (SelfPid @r)))-            , ( "spawn-ing"-              , void-                (send-                  (Spawn @r-                    (void-                      (send (ReceiveSelectedMessage @r selectAnyMessageLazy))-                    )-                  )-                )-              )-            , ("sleeping", lift (threadDelay 100000))-            ]-          ]-        , testGroup-          "main thread exit not blocked by"-          [ testCase ("a child process, busy with " ++ busyWith)-            $ applySchedulerFactory schedulerFactory-            $ do-                void $ spawn $ foreverCheap busyEffect-                lift (threadDelay 10000)-          | (busyWith, busyEffect) <--            [ ( "receiving"-              , void-                (send-                  (ReceiveSelectedMessage @r (filterMessage (== "test message"))-                  )-                )-              )-            , ( "sending"-              , void-                (send (SendMessage @r 44444 (Dynamic.toDyn "test message")))-              )-            , ( "sending shutdown"-              , void (send (SendShutdown @r 44444 ExitNormally))-              )-            , ("selfpid-ing", void (send (SelfPid @r)))-            , ( "spawn-ing"-              , void-                (send-                  (Spawn @r-                    (void-                      (send (ReceiveSelectedMessage @r selectAnyMessageLazy))-                    )-                  )-                )-              )-            ]-          ]-        , testGroup-          "one process exits, the other continues unimpaired"-          [ testCase-              (  "process 2 exits with: "-              ++ howToExit-              ++ " - while process 1 is busy with: "-              ++ busyWith-              )-            $ scheduleAndAssert schedulerFactory-            $ \assertEff -> do-                p1 <- spawn $ foreverCheap busyEffect-                lift (threadDelay 1000)-                void $ spawn $ do-                  lift (threadDelay 1000)-                  doExit-                lift (threadDelay 100000)-                wasRunningP1 <- isProcessAlive SP p1-                sendShutdown px p1 ExitNormally-                lift (threadDelay 100000)-                stillRunningP1 <- isProcessAlive SP p1-                assertEff "the other process did not die still running"-                          (not stillRunningP1 && wasRunningP1)-          | (busyWith , busyEffect) <--            [ ( "receiving"-              , void-                (send-                  (ReceiveSelectedMessage @r (filterMessage (== "test message"))-                  )-                )-              )-            , ( "sending"-              , void-                (send (SendMessage @r 44444 (Dynamic.toDyn "test message")))-              )-            , ( "sending shutdown"-              , void (send (SendShutdown @r 44444 ExitNormally))-              )-            , ("selfpid-ing", void (send (SelfPid @r)))-            , ( "spawn-ing"-              , void-                (send-                  (Spawn @r-                    (void-                      (send (ReceiveSelectedMessage @r selectAnyMessageLazy))-                    )-                  )-                )-              )-            ]-          , (howToExit, doExit    ) <--            [ ("normally"        , void (exitNormally px))-            , ("simply returning", return ())-            , ( "raiseError"-              , void (interrupt (ProcessError "test error raised"))-              )-            , ("exitWithError", void (exitWithError px "test error exit"))-            , ( "sendShutdown to self"-              , do-                me <- self px-                void (sendShutdown px me ExitNormally)-              )-            ]-          ]-        ]---sendShutdownTests-  :: forall r-   . (Member (Logs LogMessage) r, SetMember Lift (Lift IO) r)-  => IO (Eff (InterruptableProcess r) () -> IO ())-  -> TestTree-sendShutdownTests schedulerFactory =-  let px :: SchedulerProxy r-      px = SchedulerProxy-  in  testGroup-        "sendShutdown"-        [ testCase "... self" $ applySchedulerFactory schedulerFactory $ do-          me <- self px-          void $ send (SendShutdown @r me ExitNormally)-          interrupt (ProcessError "sendShutdown must not return")-        , testCase "sendInterrupt to self"-        $ scheduleAndAssert schedulerFactory-        $ \assertEff -> do-            me <- self px-            r  <- send (SendInterrupt @r me (ProcessError "123"))-            assertEff-              "Interrupted must be returned"-              (case r of-                Interrupted (ProcessError "123") -> True-                _ -> False-              )-        , testGroup-          "... other process"-          [ testCase "while it is sending"-          $ scheduleAndAssert schedulerFactory-          $ \assertEff -> do-              me    <- self px-              other <- spawn-                (do-                  untilInterrupted-                    (SendMessage @r 666666 (Dynamic.toDyn "test"))-                  void (sendMessage px me "OK")-                )-              void (sendInterrupt px other testInterruptReason)-              a <- receiveMessage px-              assertEff "" (a == "OK")-          , testCase "while it is receiving"-          $ scheduleAndAssert schedulerFactory-          $ \assertEff -> do-              me    <- self px-              other <- spawn-                (do-                  untilInterrupted-                    (ReceiveSelectedMessage @r selectAnyMessageLazy)-                  void (sendMessage px me "OK")-                )-              void (sendInterrupt px other testInterruptReason)-              a <- receiveMessage px-              assertEff "" (a == "OK")-          , testCase "while it is self'ing"-          $ scheduleAndAssert schedulerFactory-          $ \assertEff -> do-              me    <- self px-              other <- spawn-                (do-                  untilInterrupted (SelfPid @r)-                  void (sendMessage px me "OK")-                )-              void (sendInterrupt px other (ProcessError "testError"))-              a <- receiveMessage px-              assertEff "" (a == "OK")-          , testCase "while it is spawning"-          $ scheduleAndAssert schedulerFactory-          $ \assertEff -> do-              me    <- self px-              other <- spawn-                (do-                  untilInterrupted (Spawn @r (return ()))-                  void (sendMessage px me "OK")-                )-              void (sendInterrupt px other testInterruptReason)-              a <- receiveMessage px-              assertEff "" (a == "OK")-          , testCase "while it is sending shutdown messages"-          $ scheduleAndAssert schedulerFactory-          $ \assertEff -> do-              me    <- self px-              other <- spawn-                (do-                  untilInterrupted (SendShutdown @r 777 ExitNormally)-                  void (sendMessage px me "OK")-                )-              void (sendInterrupt px other testInterruptReason)-              a <- receiveMessage px-              assertEff "" (a == "OK")-          , testCase "handleInterrupt handles my own interrupts"-          $ scheduleAndAssert schedulerFactory-          $ \assertEff ->-              handleInterrupts-                  (\e -> return (ProcessError "test" == e))-                  (interrupt (ProcessError "test") >> return False)-                >>= assertEff "exception handler not invoked"-          ]-        ]--linkingTests-  :: forall r-   . (Member (Logs LogMessage) r, SetMember Lift (Lift IO) r)-  => IO (Eff (InterruptableProcess r) () -> IO ())-  -> TestTree-linkingTests schedulerFactory = setTravisTestOptions-  (testGroup-    "process linking tests"-    [ testCase "link process with it self"-    $ applySchedulerFactory schedulerFactory-    $ do-        me <- self SP-        handleInterrupts-          (\er -> lift (False @? ("unexpected interrupt: " ++ show er)))-          (do-            linkProcess SP me-            lift (threadDelay 10000)-          )-    , testCase "link with not running process"-    $ applySchedulerFactory schedulerFactory-    $ do-        let testPid = 234234234-        handleInterrupts-          (lift . (@?= LinkedProcessCrashed testPid))-          (do-            linkProcess SP testPid-            void (receiveMessage @Void SP)-          )-    , testCase "linked process exit message is NormalExit"-    $ applySchedulerFactory schedulerFactory-    $ do-        foo <- spawn (void (receiveMessage @Void SP))-        handleInterrupts-          (lift . (\e -> e /= LinkedProcessCrashed foo @? show e))-          (do-            linkProcess SP foo-            sendShutdown SP foo ExitNormally-            lift (threadDelay 1000)-          )-    , testCase "linked process exit message is Crash"-    $ applySchedulerFactory schedulerFactory-    $ do-        foo <- spawn (void (receiveMessage @Void SP))-        handleInterrupts-          (lift . (@?= LinkedProcessCrashed foo))-          (do-            linkProcess SP foo-            sendShutdown SP foo Killed-            void (receiveMessage @Void SP)-          )-    , testCase "link multiple times"-    $ applySchedulerFactory schedulerFactory-    $ do-        foo <- spawn (void (receiveMessage @Void SP))-        handleInterrupts-          (lift . (@?= LinkedProcessCrashed foo))-          (do-            linkProcess SP foo-            linkProcess SP foo-            linkProcess SP foo-            linkProcess SP foo-            linkProcess SP foo-            linkProcess SP foo-            linkProcess SP foo-            sendShutdown SP foo Killed-            void (receiveMessage @Void SP)-          )-    , testCase "unlink multiple times"-    $ applySchedulerFactory schedulerFactory-    $ do-        foo <- spawn (void (receiveMessage @Void SP))-        handleInterrupts-          (lift . (False @?) . show)-          (do-            spawn_ (void (receiveAnyMessage SP))-            linkProcess SP foo-            linkProcess SP foo-            linkProcess SP foo-            unlinkProcess SP foo-            unlinkProcess SP foo-            unlinkProcess SP foo-            unlinkProcess SP foo-            withMonitor SP foo $ \ref -> do-              sendShutdown SP foo Killed-              void (receiveSelectedMessage SP (selectProcessDown ref))-          )-    , testCase "spawnLink" $ applySchedulerFactory schedulerFactory $ do-      let foo = void (receiveMessage @Void SP)-      handleInterrupts (\er -> lift (isBecauseDown Nothing er @? show er)) $ do-        x <- spawnLink foo-        sendShutdown SP x Killed-        void (receiveMessage @Void SP)-    , testCase "ignore normal exit"-    $ applySchedulerFactory schedulerFactory-    $ do-        mainProc <- self SP-        let linkingServer = void $ exitOnInterrupt SP $ do-              logNotice "linker"-              foreverCheap $ do-                x <- receiveMessage SP-                linkProcess SP x-                sendMessage SP mainProc True-        linker <- spawnLink linkingServer-        logNotice "mainProc"-        do-          x <- spawnLink (logNotice "x 1" >> void (receiveMessage @Void SP))-          withMonitor SP x $ \xRef -> do-            sendMessage SP linker x-            void $ receiveSelectedMessage SP (filterMessage id)-            sendShutdown SP x ExitNormally-            void (receiveSelectedMessage SP (selectProcessDown xRef))-        do-          x <- spawnLink (logNotice "x 2" >> void (receiveMessage @Void SP))-          withMonitor SP x $ \xRef -> do-            sendMessage SP linker x-            void $ receiveSelectedMessage SP (filterMessage id)-            sendShutdown SP x ExitNormally-            void (receiveSelectedMessage SP (selectProcessDown xRef))-        handleInterrupts (lift . (LinkedProcessCrashed linker @=?)) $ do-          sendShutdown SP linker Killed-          void (receiveMessage @Void SP)-    , testCase "unlink" $ applySchedulerFactory schedulerFactory $ do-      let-        foo1 = void (receiveAnyMessage SP)-        foo2 foo1Pid = do-          linkProcess SP foo1Pid-          (r1, barPid) <- receiveMessage SP-          lift ("unlink foo1" @=? r1)-          unlinkProcess SP foo1Pid-          sendMessage SP barPid ("unlinked foo1", foo1Pid)-          receiveMessage SP >>= lift . (@?= "the end")-          exitWithError SP "foo two"-        bar foo2Pid parentPid = do-          linkProcess SP foo2Pid-          me <- self SP-          sendMessage SP foo2Pid ("unlink foo1", me)-          (r1, foo1Pid) <- receiveMessage SP-          lift ("unlinked foo1" @=? r1)-          handleInterrupts-            (const (return ()))-            (do-              linkProcess SP foo1Pid-              sendShutdown SP foo1Pid Killed-              void (receiveMessage @Void SP)-            )-          handleInterrupts-            (\er ->-              void-                (sendMessage SP parentPid (LinkedProcessCrashed foo2Pid == er))-            )-            (do-              sendMessage SP foo2Pid "the end"-              void (receiveAnyMessage SP)-            )-      foo1Pid <- spawn foo1-      foo2Pid <- spawn (foo2 foo1Pid)-      me      <- self SP-      barPid  <- spawn (bar foo2Pid me)-      handleInterrupts-        (\er -> lift (LinkedProcessCrashed barPid @?= er))-        (do-          res <- receiveMessage @Bool SP-          lift (threadDelay 100000)-          lift (res @?= True)-        )-    ]-  )--monitoringTests-  :: forall r-   . (Member (Logs LogMessage) r, SetMember Lift (Lift IO) r)-  => IO (Eff (InterruptableProcess r) () -> IO ())-  -> TestTree-monitoringTests schedulerFactory = setTravisTestOptions-  (testGroup-    "process monitoring tests"-    [ testCase "monitored process not running"-    $ applySchedulerFactory schedulerFactory-    $ do-        let badPid = 132123-        ref <- monitor SP badPid-        pd  <- receiveSelectedMessage SP (selectProcessDown ref)-        lift (downReason pd @?= SomeExitReason (ProcessNotRunning badPid))-        lift (threadDelay 10000)-    , testCase "monitored process exit normally"-    $ applySchedulerFactory schedulerFactory-    $ do-        target <- spawn (receiveMessage SP >>= exitBecause SP)-        ref    <- monitor SP target-        sendMessage SP target ExitNormally-        pd <- receiveSelectedMessage SP (selectProcessDown ref)-        lift (downReason pd @?= SomeExitReason ExitNormally)-        lift (threadDelay 10000)-    , testCase "multiple monitors some demonitored"-    $ applySchedulerFactory schedulerFactory-    $ do-        target <- spawn (receiveMessage SP >>= exitBecause SP)-        ref1   <- monitor SP target-        ref2   <- monitor SP target-        ref3   <- monitor SP target-        ref4   <- monitor SP target-        ref5   <- monitor SP target-        demonitor SP ref3-        demonitor SP ref5-        sendMessage SP target ExitNormally-        pd1 <- receiveSelectedMessage SP (selectProcessDown ref1)-        lift (downReason pd1 @?= SomeExitReason ExitNormally)-        pd2 <- receiveSelectedMessage SP (selectProcessDown ref2)-        lift (downReason pd2 @?= SomeExitReason ExitNormally)-        pd4 <- receiveSelectedMessage SP (selectProcessDown ref4)-        lift (downReason pd4 @?= SomeExitReason ExitNormally)-        lift (threadDelay 10000)-    , testCase "monitored process killed"-    $ applySchedulerFactory schedulerFactory-    $ do-        target <- spawn (receiveMessage SP >>= exitBecause SP)-        ref    <- monitor SP target-        sendMessage SP target Killed-        pd <- receiveSelectedMessage SP (selectProcessDown ref)-        lift (downReason pd @?= SomeExitReason Killed)-        lift (threadDelay 10000)-    , testCase "demonitored process killed"-    $ applySchedulerFactory schedulerFactory-    $ do-        target <- spawn (receiveMessage SP >>= exitBecause SP)-        ref    <- monitor SP target-        demonitor SP ref-        sendMessage SP target Killed-        me <- self SP-        spawn_ (lift (threadDelay 10000) >> sendMessage SP me ())-        pd <- receiveSelectedMessage-          SP-          (Right <$> selectProcessDown ref <|> Left <$> selectMessage @())-        lift (pd @?= Left ())-        lift (threadDelay 10000)-    ]-  )--timerTests-  :: forall r-   . (Member (Logs LogMessage) r, SetMember Lift (Lift IO) r)-  => IO (Eff (InterruptableProcess r) () -> IO ())-  -> TestTree-timerTests schedulerFactory = setTravisTestOptions-  (testGroup-    "process timer tests"-    [ testCase "receiveAfter into timeout"-    $ applySchedulerFactory schedulerFactory-    $ do-        pd <- receiveAfter @Void SP 1000-        lift (pd @?= Nothing)-        lift (threadDelay 10000)-    , testCase "receiveAfter no timeout"-    $ applySchedulerFactory schedulerFactory-    $ do-        me    <- self SP-        other <- spawn-          (do-            r <- receiveMessage @() SP-            lift (r @?= ())-            sendMessage SP me (123 :: Int)-          )-        pd1 <- receiveAfter @() SP 10000-        lift (pd1 @?= Nothing)-        sendMessage SP other ()-        pd2 <- receiveAfter @Int SP 10000-        lift (pd2 @?= Just 123)-        lift (threadDelay 10000)-    , testCase "many receiveAfters"-    $ applySchedulerFactory schedulerFactory-    $ do-        let n = 5-            testMsg :: Float-            testMsg = 123-        me    <- self SP-        other <- spawn-          (do-            replicateM_ n $ sendMessage SP me "bad message"-            r <- receiveMessage @() SP-            lift (r @?= ())-            replicateM_ n $ sendMessage SP me testMsg-          )-        receiveAfter @Float SP 100 >>= lift . (@?= Nothing)-        sendMessage SP other ()-        replicateM_-          n-          (do-            res <- receiveAfter @Float SP 10000+import           Control.Exception+import           Control.Concurrent+import           Control.Concurrent.STM+import           Control.Eff.Concurrent.Process+import           Control.Eff.Concurrent.Process.Timer+import           Control.Eff.Concurrent.Api+import           Control.Eff.Concurrent.Api.Client+import           Control.Eff.Concurrent.Api.Server+import qualified Control.Eff.Concurrent.Process.ForkIOScheduler+                                               as ForkIO+import qualified Control.Eff.Concurrent.Process.SingleThreadedScheduler+                                               as SingleThreaded+import           Control.Eff+import           Control.Eff.Extend+import           Control.Eff.Log+import           Control.Eff.Loop+import           Control.Eff.Lift+import           Control.Monad+import           Test.Tasty+import           Test.Tasty.HUnit+import           Common+import           Control.Applicative+import           Data.Void++testInterruptReason :: InterruptReason+testInterruptReason = ProcessError "test interrupt"++test_forkIo :: TestTree+test_forkIo = setTravisTestOptions $ withTestLogC+  (\c lc -> handleLoggingAndIO_ (ForkIO.schedule c) lc)+  (\factory -> testGroup "ForkIOScheduler" [allTests factory])+++test_singleThreaded :: TestTree+test_singleThreaded = setTravisTestOptions $ withTestLogC+  (\e logC ->+        -- void (runLift (logToChannel logC (SingleThreaded.schedule (return ()) e)))+    let runEff+          :: Eff '[Logs LogMessage, LogWriterReader LogMessage IO, Lift IO] a+          -> IO a+        runEff = flip handleLoggingAndIO logC+    in  void $ SingleThreaded.scheduleM runEff yield e+  )+  (\factory -> testGroup "SingleThreadedScheduler" [allTests factory])++allTests+  :: forall r+   . (Member (Logs LogMessage) r, SetMember Lift (Lift IO) r)+  => IO (Eff (InterruptableProcess r) () -> IO ())+  -> TestTree+allTests schedulerFactory = localOption+  (timeoutSeconds 300)+  (testGroup+    "Process"+    [ errorTests schedulerFactory+    , sendShutdownTests schedulerFactory+    , concurrencyTests schedulerFactory+    , exitTests schedulerFactory+    , pingPongTests schedulerFactory+    , yieldLoopTests schedulerFactory+    , selectiveReceiveTests schedulerFactory+    , linkingTests schedulerFactory+    , monitoringTests schedulerFactory+    , timerTests schedulerFactory+    ]+  )++data ReturnToSender+  deriving Typeable++data instance Api ReturnToSender r where+  ReturnToSender :: ProcessId -> String -> Api ReturnToSender ('Synchronous Bool)+  StopReturnToSender :: Api ReturnToSender ('Synchronous ())++deriving instance Show (Api ReturnToSender x)++deriving instance Typeable (Api ReturnToSender x)++returnToSender+  :: forall q r+   . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)+  => Server ReturnToSender+  -> String+  -> Eff r Bool+returnToSender toP msg = do+  me      <- self+  _       <- call toP (ReturnToSender me msg)+  msgEcho <- receiveMessage @String+  return (msgEcho == msg)++stopReturnToSender+  :: forall q r+   . (HasCallStack, SetMember Process (Process q) r, Member Interrupts r)+  => Server ReturnToSender+  -> Eff r ()+stopReturnToSender toP = call toP StopReturnToSender++returnToSenderServer+  :: forall q . ( HasCallStack, Member (Logs LogMessage) q )+  => Eff (InterruptableProcess q) (Server ReturnToSender)+returnToSenderServer = spawnApiServer+  (handleCalls (\m k -> k $+                        case m of+                          StopReturnToSender -> do+                            return (Nothing, StopServer testInterruptReason)+                          ReturnToSender fromP echoMsg -> do+                            sendMessage fromP echoMsg+                            yieldProcess+                            return (Just True, AwaitNext)+                        )+  ) stopServerOnInterrupt++selectiveReceiveTests+  :: forall r+   . (Member (Logs LogMessage) r, SetMember Lift (Lift IO) r)+  => IO (Eff (InterruptableProcess r) () -> IO ())+  -> TestTree+selectiveReceiveTests schedulerFactory = setTravisTestOptions+  (testGroup+    "selective receive tests"+    [ testCase "send 10 messages (from 1..10) and receive messages from 10 to 1"+    $ applySchedulerFactory schedulerFactory+    $ do+        let+          nMax = 10+          receiverLoop donePid = go nMax+           where+            go :: Int -> Eff (InterruptableProcess r) ()+            go 0 = sendMessage donePid True+            go n = do+              void $ receiveSelectedMessage (filterMessage (== n))+              go (n - 1)++          senderLoop receviverPid =+            traverse_ (sendMessage receviverPid) [1 .. nMax]++        me          <- self+        receiverPid <- spawn (receiverLoop me)+        spawn_ (senderLoop receiverPid)+        ok <- receiveMessage @Bool+        lift (ok @? "selective receive failed")+    , testCase "receive a message while waiting for a call reply"+    $ applySchedulerFactory schedulerFactory+    $ do+        srv <- returnToSenderServer+        ok  <- returnToSender srv "test"+        ()  <- stopReturnToSender srv+        lift (ok @? "selective receive failed")+    , testCase "flush messages" $ applySchedulerFactory schedulerFactory $ do+      me <- self+      spawn_ $ replicateM_ 10 (sendMessage me True) >> sendMessage me ()+      spawn_+        $  replicateM_ 10 (sendMessage me (123.23411 :: Float))+        >> sendMessage me ()+      spawn_ $ replicateM_ 10 (sendMessage me "123") >> sendMessage me ()+      ()   <- receiveMessage+      ()   <- receiveMessage+      ()   <- receiveMessage+      -- replicateCheapM_ 40 yieldProcess+      msgs <- flushMessages+      lift (length msgs @?= 30)+    ]+  )+++yieldLoopTests+  :: forall r+   . (Member (Logs LogMessage) r, SetMember Lift (Lift IO) r)+  => IO (Eff (InterruptableProcess r) () -> IO ())+  -> TestTree+yieldLoopTests schedulerFactory =+  let maxN = 100000+  in  setTravisTestOptions+        (testGroup+          "yield tests"+          [ testCase+            "yield many times (replicateM_)"+            (applySchedulerFactory schedulerFactory+                                   (replicateM_ maxN yieldProcess)+            )+          , testCase+            "yield many times (forM_)"+            (applySchedulerFactory+              schedulerFactory+              (forM_ [1 :: Int .. maxN] (\_ -> yieldProcess))+            )+          , testCase+            "construct an effect with an exit first, followed by many yields"+            (applySchedulerFactory+              schedulerFactory+              (do+                void exitNormally+                replicateM_ 1000000000000 yieldProcess+              )+            )+          ]+        )+++data Ping = Ping ProcessId+  deriving (Eq, Show)++data Pong = Pong+  deriving (Eq, Show)++pingPongTests+  :: forall r+   . (Member (Logs LogMessage) r, SetMember Lift (Lift IO) r)+  => IO (Eff (InterruptableProcess r) () -> IO ())+  -> TestTree+pingPongTests schedulerFactory = testGroup+  "Yield Tests"+  [ testCase "ping pong a message between two processes, both don't yield"+  $ applySchedulerFactory schedulerFactory+  $ do+      let pongProc = foreverCheap $ do+            Ping pinger <- receiveMessage+            sendMessage pinger Pong+          pingProc ponger parent = do+            me <- self+            sendMessage ponger (Ping me)+            Pong <- receiveMessage+            sendMessage parent True+      pongPid <- spawn pongProc+      me      <- self+      spawn_ (pingProc pongPid me)+      ok <- receiveMessage @Bool+      lift (ok @? "ping pong failed")+  , testCase "ping pong a message between two processes, with massive yielding"+  $ applySchedulerFactory schedulerFactory+  $ do+      yieldProcess+      let pongProc = foreverCheap $ do+            yieldProcess+            Ping pinger <- receiveMessage+            yieldProcess+            sendMessage pinger Pong+            yieldProcess+          pingProc ponger parent = do+            yieldProcess+            me <- self+            yieldProcess+            sendMessage ponger (Ping me)+            yieldProcess+            Pong <- receiveMessage+            yieldProcess+            sendMessage parent True+            yieldProcess+      yieldProcess+      pongPid <- spawn pongProc+      yieldProcess+      me <- self+      yieldProcess+      spawn_ (pingProc pongPid me)+      yieldProcess+      ok <- receiveMessage @Bool+      yieldProcess+      lift (ok @? "ping pong failed")+      yieldProcess+  , testCase+    "the first message is not delayed, not even in cooperative scheduling (because of yield)"+  $ applySchedulerFactory schedulerFactory+  $ do+      pongVar <- lift newEmptyMVar+      let pongProc = foreverCheap $ do+            Pong <- receiveMessage+            lift (putMVar pongVar Pong)+      ponger <- spawn pongProc+      sendMessage ponger Pong+      let waitLoop = do+            p <- lift (tryTakeMVar pongVar)+            case p of+              Nothing -> do+                yieldProcess+                waitLoop+              Just r -> return r+      p <- waitLoop+      lift (p == Pong @? "ping pong failed")+  ]++errorTests+  :: forall r+   . (Member (Logs LogMessage) r, SetMember Lift (Lift IO) r)+  => IO (Eff (InterruptableProcess r) () -> IO ())+  -> TestTree+errorTests schedulerFactory = testGroup+  "causing and handling errors"+  [ testGroup+      "exitWithError"+      [ testCase "unhandled exitWithError"+      $ applySchedulerFactory schedulerFactory+      $ do+          void $ exitWithError "test error"+          error "This should not happen"+      , testCase "cannot catch exitWithError"+      $ applySchedulerFactory schedulerFactory+      $ do+          void $ exitWithError "test error 4"+          error "This should not happen"+      , testCase "multi process exitWithError"+      $ scheduleAndAssert schedulerFactory+      $ \assertEff -> do+          me <- self+          let n = 15+          traverse_+            (\(i :: Int) -> spawn $ foreverCheap+              (void+                (  sendMessage (888888 + fromIntegral i) "test message"+                >> yieldProcess+                )+              )+            )+            [0 .. n]+          traverse_+            (\(i :: Int) -> spawn $ do+              void $ sendMessage me i+              void (exitWithError (show i ++ " died"))+              error "this should not be reached"+            )+            [0 .. n]+          oks <- replicateM (length [0 .. n]) receiveMessage+          assertEff "" (sort oks == [0 .. n])+      ]+  ]++concurrencyTests+  :: forall r+   . (Member (Logs LogMessage) r, SetMember Lift (Lift IO) r)+  => IO (Eff (InterruptableProcess r) () -> IO ())+  -> TestTree+concurrencyTests schedulerFactory =+  let n = 100+  in+    testGroup+      "concurrency tests"+      [ testCase+        "when main process exits the scheduler kills/cleans and returns"+      $ applySchedulerFactory schedulerFactory+      $ do+          me <- self+          traverse_+            (const+              (spawn+                (do+                  m <- receiveAnyMessage+                  void (sendMessage me m)+                )+              )+            )+            [1 .. n]+          lift (threadDelay 1000)+      , testCase "new processes are executed before the parent process"+      $ scheduleAndAssert schedulerFactory+      $ \assertEff -> do -- start massive amount of children that exit as soon as they are+             -- executed, this will only work smoothly when scheduler schedules+             -- the new child before the parent+          traverse_ (const (spawn exitNormally)) [1 .. n]+          assertEff "" True+      , testCase "two concurrent processes"+      $ scheduleAndAssert schedulerFactory+      $ \assertEff -> do+          me     <- self+          child1 <- spawn+            (do+              m <- receiveAnyMessage+              void (sendAnyMessage me m)+            )+          child2 <- spawn (foreverCheap (void (sendMessage 888888 "")))+          sendMessage child1 "test"+          i <- receiveMessage+          sendInterrupt child2 testInterruptReason+          assertEff "" (i == "test")+      , testCase "most processes send foreverCheap"+      $ scheduleAndAssert schedulerFactory+      $ \assertEff -> do+          me <- self+          traverse_+            (\(i :: Int) -> spawn $ do+              when (i `rem` 5 == 0) $ void $ sendMessage me i+              foreverCheap $ void (sendMessage 888 "test message to 888")+            )+            [0 .. n]+          oks <- replicateM (length [0, 5 .. n]) receiveMessage+          assertEff "" (sort oks == [0, 5 .. n])+      , testCase "most processes self foreverCheap"+      $ scheduleAndAssert schedulerFactory+      $ \assertEff -> do+          me <- self+          traverse_+            (\(i :: Int) -> spawn $ do+              when (i `rem` 5 == 0) $ void $ sendMessage me i+              foreverCheap $ void self+            )+            [0 .. n]+          oks <- replicateM (length [0, 5 .. n]) receiveMessage+          assertEff "" (sort oks == [0, 5 .. n])+      , testCase "most processes sendShutdown foreverCheap"+      $ scheduleAndAssert schedulerFactory+      $ \assertEff -> do+          me <- self+          traverse_+            (\(i :: Int) -> spawn $ do+              when (i `rem` 5 == 0) $ void $ sendMessage me i+              foreverCheap $ void (sendShutdown 999 ExitNormally)+            )+            [0 .. n]+          oks <- replicateM (length [0, 5 .. n]) receiveMessage+          assertEff "" (sort oks == [0, 5 .. n])+      , testCase "most processes spawn foreverCheap"+      $ scheduleAndAssert schedulerFactory+      $ \assertEff -> do+          me <- self+          traverse_+            (\(i :: Int) -> spawn $ do+              when (i `rem` 5 == 0) $ void $ sendMessage me i+              parent <- self+              foreverCheap $ void+                (spawn (void (sendMessage parent "test msg from child")))+            )+            [0 .. n]+          oks <- replicateM (length [0, 5 .. n]) receiveMessage+          assertEff "" (sort oks == [0, 5 .. n])+      , testCase "most processes receive foreverCheap"+      $ scheduleAndAssert schedulerFactory+      $ \assertEff -> do+          me <- self+          traverse_+            (\(i :: Int) -> spawn $ do+              when (i `rem` 5 == 0) $ void $ sendMessage me i+              foreverCheap $ void receiveAnyMessage+            )+            [0 .. n]+          oks <- replicateM (length [0, 5 .. n]) receiveMessage+          assertEff "" (sort oks == [0, 5 .. n])+      ]++exitTests+  :: forall r+   . (Member (Logs LogMessage) r, SetMember Lift (Lift IO) r)+  => IO (Eff (InterruptableProcess r) () -> IO ())+  -> TestTree+exitTests schedulerFactory =+  testGroup "process exit tests"+    $ [ testGroup+        "async exceptions"+        [ testCase+              (  "a process dies immediately if a "+              ++ show e+              ++ " is thrown, while "+              ++ busyWith+              )+            $ do+                tidVar           <- newEmptyTMVarIO+                schedulerDoneVar <- newEmptyTMVarIO+                void $ forkIO $ do+                  void+                    $ try @SomeException+                    $ void+                    $ applySchedulerFactory schedulerFactory+                    $ do+                        tid <- lift $ myThreadId+                        lift $ atomically $ putTMVar tidVar tid+                        foreverCheap busyEffect+                  atomically $ putTMVar schedulerDoneVar ()+                tid <- atomically $ takeTMVar tidVar+                threadDelay 1000+                throwTo tid e+                void $ atomically $ takeTMVar schedulerDoneVar+        | e <- [ThreadKilled, UserInterrupt, HeapOverflow, StackOverflow]+        , (busyWith, busyEffect) <-+          [ ( "receiving"+            , void+              (send+                (ReceiveSelectedMessage @r (filterMessage (== "test message")))+              )+            )+          , ( "sending"+            , void (send (SendMessage @r 44444 (Dynamic.toDyn "test message")))+            )+          , ( "sending shutdown"+            , void (send (SendShutdown @r 44444 ExitNormally))+            )+          , ("selfpid-ing", void (send (SelfPid @r)))+          , ( "spawn-ing"+            , void+              (send+                (Spawn @r+                  (void (send (ReceiveSelectedMessage @r selectAnyMessageLazy)))+                )+              )+            )+          , ("sleeping", lift (threadDelay 100000))+          ]+        ]+      , testGroup+        "main thread exit not blocked by"+        [ testCase ("a child process, busy with " ++ busyWith)+          $ applySchedulerFactory schedulerFactory+          $ do+              void $ spawn $ foreverCheap busyEffect+              lift (threadDelay 10000)+        | (busyWith, busyEffect) <-+          [ ( "receiving"+            , void+              (send+                (ReceiveSelectedMessage @r (filterMessage (== "test message")))+              )+            )+          , ( "sending"+            , void (send (SendMessage @r 44444 (Dynamic.toDyn "test message")))+            )+          , ( "sending shutdown"+            , void (send (SendShutdown @r 44444 ExitNormally))+            )+          , ("selfpid-ing", void (send (SelfPid @r)))+          , ( "spawn-ing"+            , void+              (send+                (Spawn @r+                  (void (send (ReceiveSelectedMessage @r selectAnyMessageLazy)))+                )+              )+            )+          ]+        ]+      , testGroup+        "one process exits, the other continues unimpaired"+        [ testCase+            (  "process 2 exits with: "+            ++ howToExit+            ++ " - while process 1 is busy with: "+            ++ busyWith+            )+          $ scheduleAndAssert schedulerFactory+          $ \assertEff -> do+              p1 <- spawn $ foreverCheap busyEffect+              lift (threadDelay 1000)+              void $ spawn $ do+                lift (threadDelay 1000)+                doExit+              lift (threadDelay 100000)+              wasRunningP1 <- isProcessAlive p1+              sendShutdown p1 ExitNormally+              lift (threadDelay 100000)+              stillRunningP1 <- isProcessAlive p1+              assertEff "the other process did not die still running"+                        (not stillRunningP1 && wasRunningP1)+        | (busyWith , busyEffect) <-+          [ ( "receiving"+            , void+              (send+                (ReceiveSelectedMessage @r (filterMessage (== "test message")))+              )+            )+          , ( "sending"+            , void (send (SendMessage @r 44444 (Dynamic.toDyn "test message")))+            )+          , ( "sending shutdown"+            , void (send (SendShutdown @r 44444 ExitNormally))+            )+          , ("selfpid-ing", void (send (SelfPid @r)))+          , ( "spawn-ing"+            , void+              (send+                (Spawn @r+                  (void (send (ReceiveSelectedMessage @r selectAnyMessageLazy)))+                )+              )+            )+          ]+        , (howToExit, doExit    ) <-+          [ ("normally"        , void exitNormally)+          , ("simply returning", return ())+          , ("raiseError", void (interrupt (ProcessError "test error raised")))+          , ("exitWithError"   , void (exitWithError "test error exit"))+          , ( "sendShutdown to self"+            , do+              me <- self+              void (sendShutdown me ExitNormally)+            )+          ]+        ]+      ]+++sendShutdownTests+  :: forall r+   . (Member (Logs LogMessage) r, SetMember Lift (Lift IO) r)+  => IO (Eff (InterruptableProcess r) () -> IO ())+  -> TestTree+sendShutdownTests schedulerFactory = testGroup+  "sendShutdown"+  [ testCase "... self" $ applySchedulerFactory schedulerFactory $ do+    me <- self+    void $ send (SendShutdown @r me ExitNormally)+    interrupt (ProcessError "sendShutdown must not return")+  , testCase "sendInterrupt to self"+  $ scheduleAndAssert schedulerFactory+  $ \assertEff -> do+      me <- self+      r  <- send (SendInterrupt @r me (ProcessError "123"))+      assertEff+        "Interrupted must be returned"+        (case r of+          Interrupted (ProcessError "123") -> True+          _ -> False+        )+  , testGroup+    "... other process"+    [ testCase "while it is sending"+    $ scheduleAndAssert schedulerFactory+    $ \assertEff -> do+        me    <- self+        other <- spawn+          (do+            untilInterrupted (SendMessage @r 666666 (Dynamic.toDyn "test"))+            void (sendMessage me "OK")+          )+        void (sendInterrupt other testInterruptReason)+        a <- receiveMessage+        assertEff "" (a == "OK")+    , testCase "while it is receiving"+    $ scheduleAndAssert schedulerFactory+    $ \assertEff -> do+        me    <- self+        other <- spawn+          (do+            untilInterrupted (ReceiveSelectedMessage @r selectAnyMessageLazy)+            void (sendMessage me "OK")+          )+        void (sendInterrupt other testInterruptReason)+        a <- receiveMessage+        assertEff "" (a == "OK")+    , testCase "while it is self'ing"+    $ scheduleAndAssert schedulerFactory+    $ \assertEff -> do+        me    <- self+        other <- spawn+          (do+            untilInterrupted (SelfPid @r)+            void (sendMessage me "OK")+          )+        void (sendInterrupt other (ProcessError "testError"))+        a <- receiveMessage+        assertEff "" (a == "OK")+    , testCase "while it is spawning"+    $ scheduleAndAssert schedulerFactory+    $ \assertEff -> do+        me    <- self+        other <- spawn+          (do+            untilInterrupted (Spawn @r (return ()))+            void (sendMessage me "OK")+          )+        void (sendInterrupt other testInterruptReason)+        a <- receiveMessage+        assertEff "" (a == "OK")+    , testCase "while it is sending shutdown messages"+    $ scheduleAndAssert schedulerFactory+    $ \assertEff -> do+        me    <- self+        other <- spawn+          (do+            untilInterrupted (SendShutdown @r 777 ExitNormally)+            void (sendMessage me "OK")+          )+        void (sendInterrupt other testInterruptReason)+        a <- receiveMessage+        assertEff "" (a == "OK")+    , testCase "handleInterrupt handles my own interrupts"+    $ scheduleAndAssert schedulerFactory+    $ \assertEff ->+        handleInterrupts (\e -> return (ProcessError "test" == e))+                         (interrupt (ProcessError "test") >> return False)+          >>= assertEff "exception handler not invoked"+    ]+  ]++linkingTests+  :: forall r+   . (Member (Logs LogMessage) r, SetMember Lift (Lift IO) r)+  => IO (Eff (InterruptableProcess r) () -> IO ())+  -> TestTree+linkingTests schedulerFactory = setTravisTestOptions+  (testGroup+    "process linking tests"+    [ testCase "link process with it self"+    $ applySchedulerFactory schedulerFactory+    $ do+        me <- self+        handleInterrupts+          (\er -> lift (False @? ("unexpected interrupt: " ++ show er)))+          (do+            linkProcess me+            lift (threadDelay 10000)+          )+    , testCase "link with not running process"+    $ applySchedulerFactory schedulerFactory+    $ do+        let testPid = 234234234+        handleInterrupts+          (lift . (@?= LinkedProcessCrashed testPid))+          (do+            linkProcess testPid+            void (receiveMessage @Void)+          )+    , testCase "linked process exit message is NormalExit"+    $ applySchedulerFactory schedulerFactory+    $ do+        foo <- spawn (void (receiveMessage @Void))+        handleInterrupts+          (lift . (\e -> e /= LinkedProcessCrashed foo @? show e))+          (do+            linkProcess foo+            sendShutdown foo ExitNormally+            lift (threadDelay 1000)+          )+    , testCase "linked process exit message is Crash"+    $ applySchedulerFactory schedulerFactory+    $ do+        foo <- spawn (void (receiveMessage @Void))+        handleInterrupts+          (lift . (@?= LinkedProcessCrashed foo))+          (do+            linkProcess foo+            sendShutdown foo Killed+            void (receiveMessage @Void)+          )+    , testCase "link multiple times"+    $ applySchedulerFactory schedulerFactory+    $ do+        foo <- spawn (void (receiveMessage @Void))+        handleInterrupts+          (lift . (@?= LinkedProcessCrashed foo))+          (do+            linkProcess foo+            linkProcess foo+            linkProcess foo+            linkProcess foo+            linkProcess foo+            linkProcess foo+            linkProcess foo+            sendShutdown foo Killed+            void (receiveMessage @Void)+          )+    , testCase "unlink multiple times"+    $ applySchedulerFactory schedulerFactory+    $ do+        foo <- spawn (void (receiveMessage @Void))+        handleInterrupts+          (lift . (False @?) . show)+          (do+            spawn_ (void receiveAnyMessage)+            linkProcess foo+            linkProcess foo+            linkProcess foo+            unlinkProcess foo+            unlinkProcess foo+            unlinkProcess foo+            unlinkProcess foo+            withMonitor foo $ \ref -> do+              sendShutdown foo Killed+              void (receiveSelectedMessage (selectProcessDown ref))+          )+    , testCase "spawnLink" $ applySchedulerFactory schedulerFactory $ do+      let foo = void (receiveMessage @Void)+      handleInterrupts (\er -> lift (isBecauseDown Nothing er @? show er)) $ do+        x <- spawnLink foo+        sendShutdown x Killed+        void (receiveMessage @Void)+    , testCase "ignore normal exit"+    $ applySchedulerFactory schedulerFactory+    $ do+        mainProc <- self+        let linkingServer = void $ exitOnInterrupt $ do+              logNotice "linker"+              foreverCheap $ do+                x <- receiveMessage+                linkProcess x+                sendMessage mainProc True+        linker <- spawnLink linkingServer+        logNotice "mainProc"+        do+          x <- spawnLink (logNotice "x 1" >> void (receiveMessage @Void))+          withMonitor x $ \xRef -> do+            sendMessage linker x+            void $ receiveSelectedMessage (filterMessage id)+            sendShutdown x ExitNormally+            void (receiveSelectedMessage (selectProcessDown xRef))+        do+          x <- spawnLink (logNotice "x 2" >> void (receiveMessage @Void))+          withMonitor x $ \xRef -> do+            sendMessage linker x+            void $ receiveSelectedMessage (filterMessage id)+            sendShutdown x ExitNormally+            void (receiveSelectedMessage (selectProcessDown xRef))+        handleInterrupts (lift . (LinkedProcessCrashed linker @=?)) $ do+          sendShutdown linker Killed+          void (receiveMessage @Void)+    , testCase "unlink" $ applySchedulerFactory schedulerFactory $ do+      let+        foo1 = void receiveAnyMessage+        foo2 foo1Pid = do+          linkProcess foo1Pid+          (r1, barPid) <- receiveMessage+          lift ("unlink foo1" @=? r1)+          unlinkProcess foo1Pid+          sendMessage barPid ("unlinked foo1", foo1Pid)+          receiveMessage >>= lift . (@?= "the end")+          exitWithError "foo two"+        bar foo2Pid parentPid = do+          linkProcess foo2Pid+          me <- self+          sendMessage foo2Pid ("unlink foo1", me)+          (r1, foo1Pid) <- receiveMessage+          lift ("unlinked foo1" @=? r1)+          handleInterrupts+            (const (return ()))+            (do+              linkProcess foo1Pid+              sendShutdown foo1Pid Killed+              void (receiveMessage @Void)+            )+          handleInterrupts+            (\er -> void+              (sendMessage parentPid (LinkedProcessCrashed foo2Pid == er))+            )+            (do+              sendMessage foo2Pid "the end"+              void receiveAnyMessage+            )+      foo1Pid <- spawn foo1+      foo2Pid <- spawn (foo2 foo1Pid)+      me      <- self+      barPid  <- spawn (bar foo2Pid me)+      handleInterrupts+        (\er -> lift (LinkedProcessCrashed barPid @?= er))+        (do+          res <- receiveMessage @Bool+          lift (threadDelay 100000)+          lift (res @?= True)+        )+    ]+  )++monitoringTests+  :: forall r+   . (Member (Logs LogMessage) r, SetMember Lift (Lift IO) r)+  => IO (Eff (InterruptableProcess r) () -> IO ())+  -> TestTree+monitoringTests schedulerFactory = setTravisTestOptions+  (testGroup+    "process monitoring tests"+    [ testCase "monitored process not running"+    $ applySchedulerFactory schedulerFactory+    $ do+        let badPid = 132123+        ref <- monitor badPid+        pd  <- receiveSelectedMessage (selectProcessDown ref)+        lift (downReason pd @?= SomeExitReason (ProcessNotRunning badPid))+        lift (threadDelay 10000)+    , testCase "monitored process exit normally"+    $ applySchedulerFactory schedulerFactory+    $ do+        target <- spawn (receiveMessage >>= exitBecause)+        ref    <- monitor target+        sendMessage target ExitNormally+        pd <- receiveSelectedMessage (selectProcessDown ref)+        lift (downReason pd @?= SomeExitReason ExitNormally)+        lift (threadDelay 10000)+    , testCase "multiple monitors some demonitored"+    $ applySchedulerFactory schedulerFactory+    $ do+        target <- spawn (receiveMessage >>= exitBecause)+        ref1   <- monitor target+        ref2   <- monitor target+        ref3   <- monitor target+        ref4   <- monitor target+        ref5   <- monitor target+        demonitor ref3+        demonitor ref5+        sendMessage target ExitNormally+        pd1 <- receiveSelectedMessage (selectProcessDown ref1)+        lift (downReason pd1 @?= SomeExitReason ExitNormally)+        pd2 <- receiveSelectedMessage (selectProcessDown ref2)+        lift (downReason pd2 @?= SomeExitReason ExitNormally)+        pd4 <- receiveSelectedMessage (selectProcessDown ref4)+        lift (downReason pd4 @?= SomeExitReason ExitNormally)+        lift (threadDelay 10000)+    , testCase "monitored process killed"+    $ applySchedulerFactory schedulerFactory+    $ do+        target <- spawn (receiveMessage >>= exitBecause)+        ref    <- monitor target+        sendMessage target Killed+        pd <- receiveSelectedMessage (selectProcessDown ref)+        lift (downReason pd @?= SomeExitReason Killed)+        lift (threadDelay 10000)+    , testCase "demonitored process killed"+    $ applySchedulerFactory schedulerFactory+    $ do+        target <- spawn (receiveMessage >>= exitBecause)+        ref    <- monitor target+        demonitor ref+        sendMessage target Killed+        me <- self+        spawn_ (lift (threadDelay 10000) >> sendMessage me ())+        pd <- receiveSelectedMessage++          (Right <$> selectProcessDown ref <|> Left <$> selectMessage @())+        lift (pd @?= Left ())+        lift (threadDelay 10000)+    ]+  )++timerTests+  :: forall r+   . (Member (Logs LogMessage) r, SetMember Lift (Lift IO) r)+  => IO (Eff (InterruptableProcess r) () -> IO ())+  -> TestTree+timerTests schedulerFactory = setTravisTestOptions+  (testGroup+    "process timer tests"+    [ testCase "receiveAfter into timeout"+    $ applySchedulerFactory schedulerFactory+    $ do+        pd <- receiveAfter @Void 1000+        lift (pd @?= Nothing)+        lift (threadDelay 10000)+    , testCase "receiveAfter no timeout"+    $ applySchedulerFactory schedulerFactory+    $ do+        me    <- self+        other <- spawn+          (do+            r <- receiveMessage @()+            lift (r @?= ())+            sendMessage me (123 :: Int)+          )+        pd1 <- receiveAfter @() 10000+        lift (pd1 @?= Nothing)+        sendMessage other ()+        pd2 <- receiveAfter @Int 10000+        lift (pd2 @?= Just 123)+        lift (threadDelay 10000)+    , testCase "many receiveAfters"+    $ applySchedulerFactory schedulerFactory+    $ do+        let n = 5+            testMsg :: Float+            testMsg = 123+        me    <- self+        other <- spawn+          (do+            replicateM_ n $ sendMessage me "bad message"+            r <- receiveMessage @()+            lift (r @?= ())+            replicateM_ n $ sendMessage me testMsg+          )+        receiveAfter @Float 100 >>= lift . (@?= Nothing)+        sendMessage other ()+        replicateM_+          n+          (do+            res <- receiveAfter @Float 10000             lift (res @?= Just testMsg)           ) 

@@ -18,19 +18,19 @@       @=? Scheduler.schedulePure               (do                   adderChild <- spawn $ do-                      (from, arg1, arg2) <- receiveMessage SP-                      sendMessage SP from ((arg1 + arg2) :: Int)-                      foreverCheap $ void $ receiveAnyMessage SP+                      (from, arg1, arg2) <- receiveMessage+                      sendMessage from ((arg1 + arg2) :: Int)+                      foreverCheap $ void $ receiveAnyMessage                    multChild <- spawn $ do-                      (from, arg1, arg2) <- receiveMessage SP-                      sendMessage SP from ((arg1 * arg2) :: Int)+                      (from, arg1, arg2) <- receiveMessage+                      sendMessage from ((arg1 * arg2) :: Int) -                  me <- self SP-                  sendMessage SP adderChild (me, 3 :: Int, 4 :: Int)-                  x <- receiveMessage @Int SP-                  sendMessage SP multChild (me, x, 6 :: Int)-                  receiveMessage @Int SP+                  me <- self+                  sendMessage adderChild (me, 3 :: Int, 4 :: Int)+                  x <- receiveMessage @Int+                  sendMessage multChild (me, x, 6 :: Int)+                  receiveMessage @Int               )     ] @@ -41,9 +41,8 @@         "spawn a child and exit normally"         (Scheduler.defaultMainSingleThreaded             (do-                void-                    (spawn (void (receiveAnyMessage singleThreadedIoScheduler)))-                void (exitNormally singleThreadedIoScheduler)+                void (spawn (void receiveAnyMessage))+                void exitNormally                 fail "This should not happen!!"             )         )@@ -58,12 +57,12 @@             (do                 child <- spawn                     (do-                        void (receiveMessage @String singleThreadedIoScheduler)-                        void (exitNormally singleThreadedIoScheduler)+                        void (receiveMessage @String)+                        void exitNormally                         error "This should not happen (child)!!"                     )-                sendMessage singleThreadedIoScheduler child (toDyn "test")-                void (exitNormally singleThreadedIoScheduler)+                sendMessage child (toDyn "test")+                void exitNormally                 error "This should not happen!!"             )         )