diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,5 +1,16 @@ # Changelog for extensible-effects-concurrent +## 0.13.2 + +- Add `ProcessFinished` +- Add `tryUninterrupted` +- Add simpler `Server2` + +## 0.13.1 + +- Remove misguided `MonadCatch` constraints in the `ObservationQueueReader` + functions, and use `Interrupts` instead + ## 0.13.0 - Fix bad constraints in `Queue` observer diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -34,21 +34,25 @@ -- The SchedulerProxy paremeter contains the effects of a specific scheduler -- implementation. -newtype WhoAreYou = WhoAreYou ProcessId deriving (Typeable, NFData) -firstExample :: (HasLoggingIO q) => SchedulerProxy q -> Eff (Process q ': q) () +newtype WhoAreYou = WhoAreYou ProcessId deriving (Typeable, NFData, Show) + +firstExample + :: (HasLogging IO q) => SchedulerProxy q -> Eff (InterruptableProcess q) () firstExample px = do person <- spawn (do logInfo "I am waiting for someone to ask me..." WhoAreYou replyPid <- receiveMessage px - sendMessageAs px replyPid "Alice" + sendMessage px replyPid "Alice" logInfo (show replyPid ++ " just needed to know it.") ) me <- self px - sendMessageAs px person (WhoAreYou me) + sendMessage px person (WhoAreYou me) personName <- receiveMessage px logInfo ("I just met " ++ personName) + + ``` **Running** this example causes this output: diff --git a/examples/example-4/Main.hs b/examples/example-4/Main.hs --- a/examples/example-4/Main.hs +++ b/examples/example-4/Main.hs @@ -5,7 +5,6 @@ import Control.Eff.Concurrent import Data.Dynamic import Control.Concurrent -import Control.Applicative import Control.DeepSeq main :: IO () @@ -34,13 +33,3 @@ sendMessage px person (WhoAreYou me) personName <- receiveMessage px logInfo ("I just met " ++ personName) - - -selectInt :: MessageSelector Int -selectInt = selectMessage - -selectString :: MessageSelector String -selectString = selectMessage - -selectIntOrString :: MessageSelector (Either Int String) -selectIntOrString = Left <$> selectInt <|> Right <$> selectString diff --git a/extensible-effects-concurrent.cabal b/extensible-effects-concurrent.cabal --- a/extensible-effects-concurrent.cabal +++ b/extensible-effects-concurrent.cabal @@ -1,5 +1,5 @@ name: extensible-effects-concurrent -version: 0.13.0 +version: 0.13.2 description: Please see the README on GitHub at synopsis: Message passing concurrency as extensible-effect homepage: https://github.com/sheyll/extensible-effects-concurrent#readme @@ -63,6 +63,7 @@ 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, @@ -224,7 +225,7 @@ ghc-options: -Wall -threaded -fno-full-laziness -rtsopts -with-rtsopts=-N build-depends: async - , base >=4.7 && <4.12 + , base >=4.7 && <5 , data-default , extensible-effects-concurrent , extensible-effects >= 3.1.0.2 && < 3.2 diff --git a/src/Control/Eff/Concurrent.hs b/src/Control/Eff/Concurrent.hs --- a/src/Control/Eff/Concurrent.hs +++ b/src/Control/Eff/Concurrent.hs @@ -17,6 +17,9 @@ -- ** /Server/ Functions for Providing APIs module Control.Eff.Concurrent.Api.Server , + -- ** /Server/ Functions for Providing APIs (new experimental) + module Control.Eff.Concurrent.Api.Server2 + , -- ** /Observer/ Functions for Events and Event Listener module Control.Eff.Concurrent.Api.Observer , @@ -99,6 +102,7 @@ , receiveWithMonitor , provideInterruptsShutdown , handleInterrupts + , tryUninterrupted , exitOnInterrupt , logInterrupts , provideInterrupts @@ -151,6 +155,29 @@ , whereIsServer , ServerReader ) +import Control.Eff.Concurrent.Api.Server2 + ( spawnApiServer + , spawnApiServerStateful + , spawnApiServerEffectful + , CallbackResult(..) + , MessageCallback(..) + , handleCasts + , handleCalls + , handleCastsAndCalls + , handleMessages + , handleSelectedMessages + , handleAnyMessages + , handleProcessDowns + , dropUnhandledMessages + , exitOnUnhandled + , logUnhandledMessages + , (^:) + , fallbackHandler + , ToServerPids(..) + , InterruptCallback(..) + , stopServerOnInterrupt + ) + import Control.Eff.Concurrent.Api.Server ( serve , spawnServer @@ -171,7 +198,7 @@ , unhandledCallError , unhandledCastError , defaultTermination - , Servable(..) + -- , Servable(..) , ServerCallback(..) , requestHandlerSelector , terminationHandler diff --git a/src/Control/Eff/Concurrent/Api/Observer/Queue.hs b/src/Control/Eff/Concurrent/Api/Observer/Queue.hs --- a/src/Control/Eff/Concurrent/Api/Observer/Queue.hs +++ b/src/Control/Eff/Concurrent/Api/Observer/Queue.hs @@ -102,7 +102,6 @@ , Member Interrupts r , Lifted IO r , HasCallStack - , MonadCatch (Eff r) ) => SchedulerProxy q -> Int @@ -129,7 +128,6 @@ , Member Interrupts r , Lifted IO q , HasCallStack - , MonadCatch (Eff r) ) => SchedulerProxy q -> Server o @@ -171,15 +169,16 @@ , Typeable a , Show (Observation a) , HasLogging IO e - , MonadCatch (Eff e) , Integral len + , Member Interrupts e ) => len -> Eff (ObservationQueueReader a ': e) b -> Eff e b withQueue queueLimit e = do - q <- liftIO (newTBQueueIO (fromIntegral queueLimit)) - res <- Safe.tryAny (runReader (ObservationQueue q) e) + q <- liftIO (newTBQueueIO (fromIntegral queueLimit)) + res <- handleInterrupts (return . Left) + (Right <$> runReader (ObservationQueue q) e) rest <- liftIO (atomically (flushTBQueue q)) unless (null rest) diff --git a/src/Control/Eff/Concurrent/Api/Server.hs b/src/Control/Eff/Concurrent/Api/Server.hs --- a/src/Control/Eff/Concurrent/Api/Server.hs +++ b/src/Control/Eff/Concurrent/Api/Server.hs @@ -50,6 +50,7 @@ import Control.DeepSeq 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 @@ -70,7 +71,7 @@ -- * the process exits -- * '_callCallback' or '_castCallback' return 'StopApiServer' -- - -- If the process exist peacefully the parameter is 'Nothing', + -- If the process exist peacefully the parameter is 'NotServerCallbacking', -- otherwise @Just "error message..."@ if the process exits with an -- error. -- diff --git a/src/Control/Eff/Concurrent/Api/Server2.hs b/src/Control/Eff/Concurrent/Api/Server2.hs new file mode 100644 --- /dev/null +++ b/src/Control/Eff/Concurrent/Api/Server2.hs @@ -0,0 +1,363 @@ +-- | Experimental new 'Api' server handler. +-- +-- @since 0.13.2 +module Control.Eff.Concurrent.Api.Server2 + ( -- * Starting Api Servers + spawnApiServer + , spawnApiServerStateful + , spawnApiServerEffectful + -- ** Api Server Callbacks + , CallbackResult(..) + , MessageCallback(..) + -- ** Callback Smart Contructors + -- *** Calls and Casts (for 'Api's) + , handleCasts + , handleCalls + , handleCastsAndCalls + -- *** 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.Process +import Control.Monad ( (>=>) ) +import Data.Proxy +import Data.Dynamic +import Control.Applicative +import GHC.Stack +import GHC.Generics +import Control.DeepSeq +import Data.Kind +import Data.Default + +-- | /Server/ 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 (MessageCallback sel cb) (InterruptCallback intCb) = + toServerPids (Proxy @api) + <$> (spawnRaw $ receiveSelectedLoop + (SP @eff) + sel + ( either (fmap Left . intCb) (fmap Right . provideInterrupts . cb) + >=> handleCallbackResult + ) + ) + where + handleCallbackResult + :: Either CallbackResult (Either InterruptReason CallbackResult) + -> Eff (ConsProcess eff) (Maybe ()) + handleCallbackResult (Left HandleNext) = return Nothing + handleCallbackResult (Left (StopServer r)) = exitBecause SP (NotRecovered r) + handleCallbackResult (Right (Right HandleNext)) = return Nothing + handleCallbackResult (Right (Right (StopServer r))) = + intCb r >>= handleCallbackResult . Left + handleCallbackResult (Right (Left r)) = + intCb r >>= handleCallbackResult . Left + +-- | /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) + => Eff (InterruptableProcess eff) state + -> MessageCallback api (State state ': InterruptableProcess eff) + -> InterruptCallback (State state ': ConsProcess eff) + -> Eff (InterruptableProcess eff) (Server api) +spawnApiServerStateful initEffect (MessageCallback sel cb) (InterruptCallback intCb) + = fmap asServer $ 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 HandleNext -> return Nothing + where + invokeIntCb j = do + l <- intCb j + case l of + HandleNext -> 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 + , 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) (Server api) +spawnApiServerEffectful handleServerInteralEffects scb (InterruptCallback intCb) + = asServer <$> (spawn (handleServerInteralEffects (go scb))) + where + go (MessageCallback sel cb) = 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 HandleNext) = return Nothing + handleCallbackResult (Left (StopServer r)) = + exitBecause SP (NotRecovered r) + handleCallbackResult (Right (Right HandleNext)) = 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 + HandleNext :: 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 (Show, Typeable, Generic) + +instance NFData CallbackResult + +-- | 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 HandleNext)) + +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 + , NFData (Api api 'Asynchronous) + , Typeable (Api api 'Asynchronous) + ) + => (Api api 'Asynchronous -> Eff eff CallbackResult) + -> MessageCallback api eff +handleCasts = MessageCallback selectMessage + +-- | A smart constructor for 'MessageCallback's +-- +-- @since 0.13.2 +handleCalls + :: forall api eff reply + . ( HasCallStack + , NFData (Api api ( 'Synchronous reply)) + , Typeable (Api api ( 'Synchronous reply)) + ) + => (Api api ( 'Synchronous reply) -> Eff eff CallbackResult) + -> MessageCallback api eff +handleCalls = MessageCallback selectMessage + +-- | A smart constructor for 'MessageCallback's +-- +-- @since 0.13.2 +handleCastsAndCalls + :: forall api eff reply + . ( HasCallStack + , NFData (Api api ( 'Synchronous reply)) + , Typeable (Api api ( 'Synchronous reply)) + , NFData (Api api 'Asynchronous) + , Typeable (Api api 'Asynchronous) + ) + => (Api api 'Asynchronous -> Eff eff CallbackResult) + -> (Api api ( 'Synchronous reply) -> Eff eff CallbackResult) + -> MessageCallback api eff +handleCastsAndCalls onCast onCall = handleCalls onCall <> handleCasts onCast + +-- | 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 HandleNext)) + +-- | 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 HandleNext + +-- | 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) diff --git a/src/Control/Eff/Concurrent/Process.hs b/src/Control/Eff/Concurrent/Process.hs --- a/src/Control/Eff/Concurrent/Process.hs +++ b/src/Control/Eff/Concurrent/Process.hs @@ -86,6 +86,7 @@ -- ** Process Interrupt Handling , provideInterruptsShutdown , handleInterrupts + , tryUninterrupted , exitOnInterrupt , logInterrupts , provideInterrupts @@ -470,6 +471,7 @@ -- | Get the 'ExitRecover'y toExitRecovery :: ExitReason r -> ExitRecovery toExitRecovery = \case + ProcessFinished -> Recoverable (ProcessNotRunning _) -> Recoverable (LinkedProcessCrashed _) -> Recoverable (ProcessError _) -> Recoverable @@ -495,12 +497,21 @@ -- | Get the 'ExitSeverity' of a 'ExitReason'. toExitSeverity :: ExitReason e -> ExitSeverity toExitSeverity = \case - ExitNormally -> NormalExit - _ -> Crash + ExitNormally -> NormalExit + ProcessFinished -> NormalExit + _ -> Crash -- | A sum-type with reasons for why a process exists the scheduling loop, -- this includes errors, that can occur when scheduleing messages. data ExitReason (t :: ExitRecovery) where + -- | A process has finished a unit of work and might exit or work on + -- something else. This is primarily used for interupting infinite + -- server loops, allowing for additional cleanup work before + -- exitting (e.g. with 'ExitNormally') + -- + -- @since 0.13.2 + ProcessFinished + :: ExitReason 'Recoverable -- | A process that should be running was not running. ProcessNotRunning :: ProcessId -> ExitReason 'Recoverable @@ -529,6 +540,7 @@ showsPrec d = showParen (d>=10) . (\case + ProcessFinished -> showString "process finished" ProcessNotRunning p -> showString "process not running: " . shows p LinkedProcessCrashed m -> showString "linked process " . shows m . showString " crashed" @@ -546,6 +558,7 @@ instance Exc.Exception (ExitReason 'NoRecovery ) instance NFData (ExitReason x) where + rnf ProcessFinished = rnf () rnf (ProcessNotRunning !l) = rnf l rnf (LinkedProcessCrashed !l) = rnf l rnf (ProcessError !l) = rnf l @@ -555,6 +568,9 @@ rnf Killed = rnf () instance Ord (ExitReason x) where + compare ProcessFinished ProcessFinished = EQ + compare ProcessFinished _ = LT + compare _ ProcessFinished = GT compare (ProcessNotRunning l) (ProcessNotRunning r) = compare l r compare (ProcessNotRunning _) _ = LT compare _ (ProcessNotRunning _) = GT @@ -575,6 +591,7 @@ compare Killed Killed = EQ instance Eq (ExitReason x) where + (==) ProcessFinished ProcessFinished = True (==) (ProcessNotRunning l) (ProcessNotRunning r) = (==) l r (==) ExitNormally ExitNormally = True (==) (LinkedProcessCrashed l) (LinkedProcessCrashed r) = l == r @@ -588,6 +605,7 @@ -- | A predicate for linked process __crashes__. isBecauseDown :: Maybe ProcessId -> ExitReason r -> Bool isBecauseDown mp = \case + ProcessFinished -> False ProcessNotRunning _ -> False LinkedProcessCrashed p -> maybe True (== p) mp ProcessError _ -> False @@ -626,6 +644,16 @@ -> Eff r a handleInterrupts = flip catchError +-- | Like 'handleInterrupts', but instead of passing the 'InterruptReason' +-- to a handler function, 'Either' is returned. +-- +-- @since 0.13.2 +tryUninterrupted + :: (HasCallStack, Member Interrupts r) + => Eff r a + -> Eff r (Either InterruptReason a) +tryUninterrupted = handleInterrupts (pure . Left) . fmap Right + -- | Handle interrupts by logging them with `logProcessExit` and otherwise -- ignoring them. logInterrupts @@ -697,6 +725,7 @@ fromSomeExitReason :: SomeExitReason -> Either (ExitReason 'NoRecovery) InterruptReason fromSomeExitReason (SomeExitReason e) = case e of + recoverable@ProcessFinished -> Right recoverable recoverable@(ProcessNotRunning _) -> Right recoverable recoverable@(LinkedProcessCrashed _) -> Right recoverable recoverable@(ProcessError _) -> Right recoverable diff --git a/test/Common.hs b/test/Common.hs --- a/test/Common.hs +++ b/test/Common.hs @@ -30,7 +30,7 @@ withTestLogC doSchedule k = k (return (\e -> withAsyncLogChannel - 1000 + (1000 :: Integer) (multiMessageLogWriter (\writeWith -> writeWith diff --git a/test/ForkIOScheduler.hs b/test/ForkIOScheduler.hs --- a/test/ForkIOScheduler.hs +++ b/test/ForkIOScheduler.hs @@ -12,11 +12,14 @@ import Control.Eff.Concurrent.Process.Timer import Control.Eff.Concurrent.Process.ForkIOScheduler as Scheduler -import Control.Monad ( void, replicateM_ ) +import Control.Monad ( void + , replicateM_ + ) import Test.Tasty import Test.Tasty.HUnit import Data.Dynamic import Common +import Numeric.Natural import Data.Default test_IOExceptionsIsolated :: TestTree @@ -31,7 +34,7 @@ $ do aVar <- newEmptyTMVarIO withAsyncLogChannel - 1000 + (1000 :: Natural) def -- (singleMessageLogWriter (putStrLn . renderLogMessage)) (Scheduler.defaultMainWithLogChannel (do @@ -99,7 +102,7 @@ (testCase "spawn a child and return" (withAsyncLogChannel - 1000 + (1000 :: Natural) def (Scheduler.defaultMainWithLogChannel (void (spawn (void (receiveAnyMessage forkIoScheduler)))) @@ -112,7 +115,7 @@ (testCase "spawn a child and exit normally" (withAsyncLogChannel - 1000 + (1000 :: Natural) def (Scheduler.defaultMainWithLogChannel (do @@ -131,7 +134,7 @@ (testCase "spawn a child with a busy send loop and exit normally" (withAsyncLogChannel - 1000 + (1000 :: Natural) def (Scheduler.defaultMainWithLogChannel (do @@ -152,7 +155,7 @@ (testCase "spawn a child and let it return and return" (withAsyncLogChannel - 1000 + (1000 :: Natural) def (Scheduler.defaultMainWithLogChannel (do @@ -169,7 +172,7 @@ (testCase "spawn a child and let it exit and exit" (withAsyncLogChannel - 1000 + (1000 :: Natural) def (Scheduler.defaultMainWithLogChannel (do @@ -192,7 +195,7 @@ test_timer = setTravisTestOptions $ testCase "flush via timer" - $ withAsyncLogChannel 1000 def + $ withAsyncLogChannel (1000 :: Natural) def $ Scheduler.defaultMainWithLogChannel $ do let n = 100 diff --git a/test/ProcessBehaviourTestCases.hs b/test/ProcessBehaviourTestCases.hs --- a/test/ProcessBehaviourTestCases.hs +++ b/test/ProcessBehaviourTestCases.hs @@ -172,12 +172,14 @@ 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 () + $ replicateM_ 10 (sendMessage px me True) + >> sendMessage px me () spawn_ - $ replicateM_ 10 (sendMessage px me "123") >> sendMessage px me () + $ 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 @@ -863,16 +865,18 @@ foo1 = void (receiveAnyMessage SP) foo2 foo1Pid = do linkProcess SP foo1Pid - ("unlink foo1", barPid) <- receiveMessage SP + (r1, barPid) <- receiveMessage SP + lift ("unlink foo1" @=? r1) unlinkProcess SP foo1Pid sendMessage SP barPid ("unlinked foo1", foo1Pid) - "the end" <- receiveMessage SP + 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) - ("unlinked foo1", foo1Pid) <- receiveMessage SP + (r1, foo1Pid) <- receiveMessage SP + lift ("unlinked foo1" @=? r1) handleInterrupts (const (return ())) (do @@ -993,7 +997,8 @@ me <- self SP other <- spawn (do - () <- receiveMessage @() SP + r <- receiveMessage @() SP + lift (r @?= ()) sendMessage SP me (123 :: Int) ) pd1 <- receiveAfter @() SP 10000 @@ -1012,7 +1017,8 @@ other <- spawn (do replicateM_ n $ sendMessage SP me "bad message" - () <- receiveMessage @() SP + r <- receiveMessage @() SP + lift (r @?= ()) replicateM_ n $ sendMessage SP me testMsg ) receiveAfter @Float SP 100 >>= lift . (@?= Nothing)