extensible-effects-concurrent 0.9.0 → 0.9.2

raw patch · 25 files changed

+1044/−578 lines, 25 filesdep −logging-effectdep −randomdep −taggeddep ~basedep ~extensible-effectsnew-component:exe:extensible-effects-concurrent-example-3new-component:exe:extensible-effects-concurrent-example-4

Dependencies removed: logging-effect, random, tagged, transformers

Dependency ranges changed: base, extensible-effects

Files

• ChangeLog.md +14/−1
• README.md +98/−3
• examples/example-1/Main.hs +8/−14
• examples/example-2/Main.hs +15/−28
• examples/example-3/Main.hs +44/−0
• examples/example-4/Main.hs +34/−0
• extensible-effects-concurrent.cabal +60/−12
• src/Control/Eff/Concurrent.hs +15/−2
• src/Control/Eff/Concurrent/Api.hs +3/−8
• src/Control/Eff/Concurrent/Api/Client.hs +6/−4
• src/Control/Eff/Concurrent/Api/Observer.hs +4/−6
• src/Control/Eff/Concurrent/Api/Observer/Queue.hs +11/−21
• src/Control/Eff/Concurrent/Api/Server.hs +8/−6
• src/Control/Eff/Concurrent/Process.hs +125/−27
• src/Control/Eff/Concurrent/Process/ForkIOScheduler.hs +24/−19
• src/Control/Eff/Concurrent/Process/Interactive.hs +2/−1
• src/Control/Eff/Concurrent/Process/SingleThreadedScheduler.hs +21/−12
• src/Control/Eff/Log/Channel.hs +69/−174
• src/Control/Eff/Log/Handler.hs +179/−57
• src/Control/Eff/Log/Message.hs +108/−39
• test/Common.hs +15/−18
• test/ForkIOScheduler.hs +56/−35
• test/LoggingTests.hs +8/−12
• test/LoopTests.hs +62/−43
• test/ProcessBehaviourTestCases.hs +55/−36

@@ -1,5 +1,18 @@ # Changelog for extensible-effects-concurrent +## 0.9.2++- Try to adapt the dependency versions to make hackage happy again++## 0.9.1++- Add smart constructors for `MessageSelector`+- Remove `ReceiveMessage` `Process` action+- Rename `ReceiveMessageSuchThat` to `ReceiveSelectedMessage`+- Improve some Show instances, e.g. ProcessId+- Rewrite Logging API:+  - Vastly simplified API+ ## 0.9.0  - Make `ForkIOScheduler` faster and more robust@@ -18,7 +31,7 @@   unique `Int`s - Rename `spawnServer` to `spawnServerWithEffects` and add a simpler version of   `spawnServerWithEffects` called `spawnServer`-- Make all  `ApiHandler` handler callbacks optional (by changing the type to `Maybe ...`)+- Make all `ApiHandler` handler callbacks optional (by changing the type to `Maybe ...`) - `ApiHandler` must now return an `ApiServerCmd`. - Add `ApiServerCmd` which allows handler functions to leave to server loop without   exitting the process

@@ -4,10 +4,105 @@  [![Build Status](https://travis-ci.org/sheyll/extensible-effects-concurrent.svg?branch=master)](https://travis-ci.org/sheyll/extensible-effects-concurrent) -[![Hackage](https://img.shields.io/badge/hackage-extensible-effects-concurrent-green.svg?style=flat)](http://hackage.haskell.org/package/extensible-effects-concurrent) +[![Hackage](https://img.shields.io/hackage/v/extensible-effects-concurrent.svg?style=flat)](http://hackage.haskell.org/package/extensible-effects-concurrent) -[![extensible-effects-concurrent LTS](http://stackage.org/package/extensible-effects-concurrent/badge/lts)](http://stackage.org/lts/package/extensible-effects-concurrent)+Also included: +- Logging++- Memory Leak Free `forever`+ ## Example -TODO+```haskell+module Main where++import           Control.Eff+import           Control.Eff.Lift+import           Control.Eff.Concurrent+import           Data.Dynamic+import           Control.Concurrent+import           Control.DeepSeq++main :: IO ()+main = defaultMain+  (do+    lift (threadDelay 100000) -- because of async logging+    firstExample forkIoScheduler+    lift (threadDelay 100000) -- ... async logging+  )+    -- 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) ()+firstExample px = do+  person <- spawn+    (do+      logInfo "I am waiting for someone to ask me..."+      WhoAreYou replyPid <- receiveMessageAs px+      sendMessageAs px replyPid "Alice"+      logInfo (show replyPid ++ " just needed to know it.")+    )+  me <- self px+  sendMessageAs px person (WhoAreYou me)+  personName <- receiveMessageAs px+  logInfo ("I just met " ++ personName)+```++**Running** this example causes this output:+(_not entirely true because of async logging, but true enough_)++```text+2018-11-05T10:50:42 DEBUG     scheduler loop entered                                                   ForkIOScheduler.hs line 131+2018-11-05T10:50:42 DEBUG            !1 [ThreadId 11] enter process                                                            ForkIOScheduler.hs line 437+2018-11-05T10:50:42 NOTICE           !1 [ThreadId 11] ++++++++ main process started ++++++++                                   ForkIOScheduler.hs line 394+2018-11-05T10:50:42 DEBUG            !2 [ThreadId 12] enter process                                                            ForkIOScheduler.hs line 437+2018-11-05T10:50:42 INFO             !2 [ThreadId 12] I am waiting for someone to ask me...                                               Main.hs line 27+2018-11-05T10:50:42 INFO             !2 [ThreadId 12] !1 just needed to know it.                                                          Main.hs line 30+2018-11-05T10:50:42 DEBUG            !2 [ThreadId 12] returned                                                                 ForkIOScheduler.hs line 440+2018-11-05T10:50:42 INFO             !1 [ThreadId 11] I just met Alice                                                                    Main.hs line 35+2018-11-05T10:50:42 NOTICE           !1 [ThreadId 11] ++++++++ main process returned ++++++++                                  ForkIOScheduler.hs line 396+2018-11-05T10:50:42 DEBUG            !1 [ThreadId 11] returned                                                                 ForkIOScheduler.hs line 440+2018-11-05T10:50:42 DEBUG     scheduler loop returned                                                  ForkIOScheduler.hs line 133+2018-11-05T10:50:42 DEBUG     scheduler cleanup begin                                                  ForkIOScheduler.hs line 137+2018-11-05T10:50:42 NOTICE    cancelling processes: []                                                 ForkIOScheduler.hs line 149+2018-11-05T10:50:42 DEBUG     scheduler cleanup done                                                   ForkIOScheduler.hs line 141+```++## TODO++### Stackage++Still todo...++[![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++- Process Linking/Monitoring++- Scheduler `ekg` Monitoring++- Timers and Timeouts (e.g. in `receive`)++- Rename stuff

@@ -38,10 +38,8 @@ example   :: ( HasCallStack      , SetMember Process (Process q) r-     , Member (Logs LogMessage) r-     , Member (Logs LogMessage) q-     , SetMember Lift (Lift IO) q-     , SetMember Lift (Lift IO) r+     , HasLoggingIO r+     , HasLoggingIO q      )   => SchedulerProxy q   -> Eff r ()@@ -75,15 +73,11 @@  testServerLoop   :: forall r q-   . ( HasCallStack-     , Member (Logs LogMessage) q-     , SetMember Lift (Lift IO) q-     , SetMember Process (Process q) r-     )+   . (HasCallStack, SetMember Process (Process q) r, HasLoggingIO q)   => SchedulerProxy q   -> Eff r (Server TestApi) testServerLoop px = spawnServer px-  $ ApiHandler (Just handleCastTest) (Just handleCallTest) Nothing -- (Just handleTerminateTest)+  $ apiHandler handleCastTest handleCallTest handleTerminateTest  where   handleCastTest     :: Api TestApi 'Asynchronous -> Eff (Process q ': q) ApiServerCmd@@ -143,7 +137,7 @@     logInfo (show me ++ " exiting with error: " ++ msg)     void (reply ())     exitWithError px msg-  -- handleTerminateTest msg = do-  --   me <- self px-  --   logInfo (show me ++ " is exiting: " ++ show msg)-  --   maybe (exitNormally px) (exitWithError px) msg+  handleTerminateTest msg = do+    me <- self px+    logInfo (show me ++ " is exiting: " ++ show msg)+    maybe (exitNormally px) (exitWithError px) msg

@@ -3,10 +3,6 @@  import           Data.Dynamic import           Control.Eff-import qualified Control.Eff.Concurrent.Process.ForkIOScheduler-                                               as ForkIO-import qualified Control.Eff.Concurrent.Process.SingleThreadedScheduler-                                               as Pure import           Control.Eff.Concurrent import           Control.Eff.State.Strict import           Control.Eff.Lift@@ -14,8 +10,8 @@  main :: IO () main = do-  ForkIO.defaultMain (counterExample SchedulerProxy)-  Pure.defaultMain (counterExample SchedulerProxy)+  defaultMain (void (counterExample SchedulerProxy))+  print (schedulePure (counterExample SchedulerProxy))  -- * First API @@ -37,7 +33,7 @@   forgetObserverMessage = UnobserveCounter  logCounterObservations-  :: (SetMember Process (Process q) r, Member (Logs LogMessage) q, Lifted IO q)+  :: (SetMember Process (Process q) r, HasLogging m q)   => SchedulerProxy q   -> Eff r (Server (CallbackObserver Counter)) logCounterObservations px = spawnCallbackObserver@@ -49,12 +45,13 @@   )  counterHandler-  :: forall r q+  :: forall r q m    . ( Member (State (Observers Counter)) r      , Member (State Integer) r-     , Member (Logs LogMessage) r+     , HasLogging m q+     , HasLogging m r      , SetMember Process (Process q) r-     , Lifted IO r+     , Lifted m r      )   => SchedulerProxy q   -> ApiHandler Counter r@@ -90,11 +87,10 @@ deriving instance Show (Api PocketCalc x)  pocketCalcHandler-  :: forall r q+  :: forall r q m    . ( Member (State Integer) r-     , Member (Logs LogMessage) r+     , HasLogging m r      , SetMember Process (Process q) r-     , Lifted IO r      )   => SchedulerProxy q   -> ApiHandler PocketCalc r@@ -118,11 +114,8 @@     return HandleNextRequest  serverLoop-  :: forall r q-   . ( Member (Logs LogMessage) r-     , SetMember Process (Process q) r-     , Lifted IO r-     )+  :: forall r q m+   . (HasLogging m r, HasLogging m q, SetMember Process (Process q) r)   => SchedulerProxy q   -> Eff r () serverLoop px = evalState @Integer@@ -130,22 +123,16 @@   (manageObservers @Counter (serve px (counterHandler px, pocketCalcHandler px))   ) - -- ** Counter client counterExample-  :: ( SetMember Process (Process q) r-     , Member (Logs LogMessage) q-     , Member (Logs LogMessage) r-     , Lifted IO q-     , Lifted IO r-     , q <:: r-     )+  :: (SetMember Process (Process q) r, HasLogging m r, HasLogging m q, q <:: r)   => SchedulerProxy q-  -> Eff r ()-counterExample px = do+  -> 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

@@ -0,0 +1,44 @@+module Main where++import           Control.Concurrent+import           Control.Eff.Lift+import           Control.Eff.Log+import           Control.Exception             as IOException+import           System.Directory+import           System.FilePath+import           System.IO++main :: IO ()+main = withAsyncLogChannel+  1000+  (ioLogMessageWriter+    (fileAppender "extensible-effects-concurrent-example-3.log")+  )+  (handleLoggingAndIO+    (do+      logInfo "test 1"+      lift (threadDelay 1000000)+      logDebug "test 2"+      lift (threadDelay 1000000)+      logCritical "test 3"+      lift (threadDelay 1000000)+    )+  )++fileAppender :: FilePath -> LogWriter String IO+fileAppender fnIn = multiMessageLogWriter+  (\writeLogMessageWith -> bracket+    (do+      fnCanon <- canonicalizePath fnIn+      createDirectoryIfMissing True (takeDirectory fnCanon)+      h <- openFile fnCanon AppendMode+      hSetBuffering h (BlockBuffering (Just 1024))+      return h+    )+    (\h -> IOException.try @SomeException (hFlush h) >> hClose h)+    (writeLogMessageWith . hPutStrLn)+  )++fileAppenderSimple :: FilePath -> LogWriter String IO+fileAppenderSimple fnIn =+  singleMessageLogWriter (\msg -> withFile fnIn AppendMode (`hPutStrLn` msg))

@@ -0,0 +1,34 @@+module Main where++import           Control.Eff+import           Control.Eff.Lift+import           Control.Eff.Concurrent+import           Data.Dynamic+import           Control.Concurrent+import           Control.DeepSeq++main :: IO ()+main = defaultMain+  (do+    lift (threadDelay 100000) -- because of async logging+    firstExample forkIoScheduler+    lift (threadDelay 100000) -- ... async logging+  )+    -- 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) ()+firstExample px = do+  person <- spawn+    (do+      logInfo "I am waiting for someone to ask me..."+      WhoAreYou replyPid <- receiveMessageAs px+      sendMessageAs px replyPid "Alice"+      logInfo (show replyPid ++ " just needed to know it.")+    )+  me <- self px+  sendMessageAs px person (WhoAreYou me)+  personName <- receiveMessageAs px+  logInfo ("I just met " ++ personName)

@@ -1,5 +1,5 @@ name:           extensible-effects-concurrent-version:        0.9.0+version:        0.9.2 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@@ -34,10 +34,9 @@       src   build-depends:       async >= 2.2 && <3,-      base >=4.9 && <5,+      base >=4.7 && <4.12,       data-default,       deepseq,-      directory,       enclosed-exceptions >= 1.0 && < 1.1,       filepath,       time,@@ -45,15 +44,11 @@       containers >=0.5.8 && <0.7,       QuickCheck <2.12,       lens,-      logging-effect,-      transformers,       parallel,       process,       monad-control,-      random,-      extensible-effects >= 3,-      stm >= 2.4.5 && <2.6,-      tagged+      extensible-effects >= 3.1 && <4,+      stm >= 2.4.5 && <2.6   autogen-modules: Paths_extensible_effects_concurrent   exposed-modules:                   Control.Eff.Loop,@@ -107,7 +102,7 @@   main-is: Main.hs   hs-source-dirs: examples/example-1   build-depends:-                base >=4.9 && <5+                base >=4.7 && <4.12               , data-default               , extensible-effects-concurrent               , extensible-effects >= 3@@ -133,7 +128,7 @@   main-is: Main.hs   hs-source-dirs: examples/example-2   build-depends:-                base >=4.9 && <5+                base >=4.7 && <4.12               , data-default               , extensible-effects-concurrent               , extensible-effects >= 3@@ -155,6 +150,59 @@                       , TypeFamilies                       , TypeOperators +executable extensible-effects-concurrent-example-3+  main-is: Main.hs+  hs-source-dirs: examples/example-3+  build-depends:+                base >=4.7 && <4.12+              , data-default+              , directory+              , extensible-effects-concurrent+              , extensible-effects >= 3+              , filepath+              , lens+  ghc-options: -Wall -threaded -fno-full-laziness -rtsopts -with-rtsopts=-N+  default-language: Haskell2010+  default-extensions:   BangPatterns+                      , DataKinds+                      , FlexibleContexts+                      , FlexibleInstances+                      , FunctionalDependencies+                      , GADTs+                      , GeneralizedNewtypeDeriving+                      , RankNTypes+                      , ScopedTypeVariables+                      , StandaloneDeriving+                      , TemplateHaskell+                      , TypeApplications+                      , TypeFamilies+                      , TypeOperators++executable extensible-effects-concurrent-example-4+  main-is: Main.hs+  hs-source-dirs: examples/example-4+  build-depends:+                base >=4.7 && <4.12+              , extensible-effects-concurrent+              , extensible-effects >= 3+              , deepseq+  ghc-options: -Wall -threaded -fno-full-laziness -rtsopts -with-rtsopts=-N+  default-language: Haskell2010+  default-extensions:   BangPatterns+                      , DataKinds+                      , FlexibleContexts+                      , FlexibleInstances+                      , FunctionalDependencies+                      , GADTs+                      , GeneralizedNewtypeDeriving+                      , RankNTypes+                      , ScopedTypeVariables+                      , StandaloneDeriving+                      , TemplateHaskell+                      , TypeApplications+                      , TypeFamilies+                      , TypeOperators+ test-suite extensible-effects-concurrent-test   type: exitcode-stdio-1.0   main-is: Driver.hs@@ -170,7 +218,7 @@               , SingleThreadedScheduler   ghc-options: -Wall -threaded -fno-full-laziness -rtsopts -with-rtsopts=-N   build-depends:-                base >=4.9 && <5+                base >=4.7 && <4.12               , data-default               , extensible-effects-concurrent               , extensible-effects >= 3

@@ -42,7 +42,20 @@                                                 , ConsProcess                                                 , ResumeProcess(..)                                                 , SchedulerProxy(..)-                                                , MessageSelector(..)+                                                , MessageSelector+                                                  ( runMessageSelector+                                                  )+                                                , selectMessage+                                                , selectMessageLazy+                                                , selectMessageProxy+                                                , selectMessageProxyLazy+                                                , filterMessage+                                                , filterMessageLazy+                                                , selectMessageWith+                                                , selectMessageWithLazy+                                                , selectDynamicMessage+                                                , selectDynamicMessageLazy+                                                , selectAnyMessageLazy                                                 , ProcessState(..)                                                 , ProcessExitReason(..)                                                 , ShutdownRequest(..)@@ -57,7 +70,7 @@                                                 , spawn_                                                 , receiveMessage                                                 , receiveMessageAs-                                                , receiveMessageSuchThat+                                                , receiveSelectedMessage                                                 , receiveLoop                                                 , receiveLoopAs                                                 , receiveLoopSuchThat

@@ -72,16 +72,11 @@ newtype Server api = Server { _fromServer :: ProcessId }   deriving (Eq,Ord,Typeable) -instance Read (Server api) where-  readsPrec _ ('[':'#':rest1) =-    case reads (dropWhile (/= '#') rest1) of-      [(c, ']':rest2)] -> [(Server c, rest2)]-      _ -> []-  readsPrec _ _ = []  instance Typeable api => Show (Server api) where-  show s@(Server c) =-    "[#" ++ show (typeRep s) ++ "#" ++ show c ++ "]"+  showsPrec d s@(Server c) =+    showParen (d >= 10)+      (showsPrec 11 (typeRep s) . showsPrec 11 c)  makeLenses ''Server 

@@ -23,11 +23,11 @@ import           Control.Eff.Concurrent.Api import           Control.Eff.Concurrent.Api.Internal import           Control.Eff.Concurrent.Process-import           Control.Monad                  ( (>=>) ) import           Data.Dynamic import           Data.Typeable                  ( Typeable                                                 , typeRep                                                 )+import           Control.DeepSeq import           GHC.Stack  -- | Send an 'Api' request that has no return value and return as fast as@@ -78,6 +78,7 @@      , Typeable (Api api ( 'Synchronous result))      , Typeable result      , HasCallStack+     , NFData result      )   => SchedulerProxy q   -> Server api@@ -95,8 +96,8 @@             let extractResult :: Response api result -> Maybe result                 extractResult (Response _pxResult callRefMsg result) =                   if callRefMsg == callRef then Just result else Nothing-            in  MessageSelector (fromDynamic >=> extractResult)-      in  receiveMessageSuchThat px selectResult+            in  selectMessageWith extractResult+      in  receiveSelectedMessage px selectResult     else raiseError       px       (  "failed to send a call for: '"@@ -135,7 +136,7 @@ -- | Like 'call' but take the 'Server' from the reader provided by -- 'registerServer'. callRegistered-  :: (Typeable reply, ServesApi o r q, HasCallStack)+  :: (Typeable reply, ServesApi o r q, HasCallStack, NFData reply)   => SchedulerProxy q   -> Api o ( 'Synchronous reply)   -> Eff r reply@@ -154,6 +155,7 @@      , Typeable reply      , ServesApi o r q      , HasCallStack+     , NFData (f reply)      )   => SchedulerProxy q   -> Api o ( 'Synchronous (f reply))

@@ -36,7 +36,6 @@ import           Control.Eff.Concurrent.Api.Client import           Control.Eff.Concurrent.Api.Server import           Control.Eff.Log-import           Control.Eff.Lift import           Control.Eff.State.Strict import           Control.Lens @@ -180,12 +179,12 @@ -- | Start a new process for an 'Observer' that schedules -- all observations to an effectful callback. spawnCallbackObserver-  :: forall o r q+  :: forall o r q logWriter    . ( SetMember Process (Process q) r      , Typeable o      , Show (Observation o)      , Observable o-     , Member (Logs LogMessage) q+     , HasLogging logWriter q      , HasCallStack      )   => SchedulerProxy q@@ -206,13 +205,12 @@ -- -- @since 0.3.0.0 spawnLoggingObserver-  :: forall o r q+  :: forall o r q logWriter    . ( SetMember Process (Process q) r      , Typeable o      , Show (Observation o)      , Observable o-     , Member (Logs LogMessage) q-     , Lifted IO q+     , HasLogging logWriter q      , HasCallStack      )   => SchedulerProxy q

@@ -35,7 +35,7 @@ type ObservationQueueReader a = Reader (ObservationQueue a)  logPrefix :: forall o proxy . (HasCallStack, Typeable o) => proxy o -> String-logPrefix px = "<observation queue " ++ show (typeRep px) ++ ">"+logPrefix px = "observation queue: " ++ show (typeRep px)  -- | Read queued observations captured by observing a 'Server' that implements -- an 'Observable' 'Api' using 'enqueueObservationsRegistered' or 'enqueueObservations'.@@ -47,7 +47,7 @@      , HasCallStack      , MonadIO (Eff r)      , Typeable o-     , Member (Logs LogMessage) r+     , HasLoggingIO r      )   => Eff r (Observation o) readObservationQueue = do@@ -65,7 +65,7 @@      , HasCallStack      , MonadIO (Eff r)      , Typeable o-     , Member (Logs LogMessage) r+     , HasLoggingIO r      )   => Eff r (Maybe (Observation o)) tryReadObservationQueue = do@@ -82,7 +82,7 @@      , HasCallStack      , MonadIO (Eff r)      , Typeable o-     , Member (Logs LogMessage) r+     , HasLoggingIO r      )   => Eff r [Observation o] flushObservationQueue = do@@ -99,11 +99,9 @@      , Typeable o      , Show (Observation o)      , Observable o-     , Member (Logs LogMessage) q-     , MonadIO (Eff (Process q ': q))-     , MonadIO (Eff r)-     , SetMember Lift (Lift IO) r-     , Member (Logs LogMessage) r+     , HasLoggingIO q+     , HasLoggingIO r+     , Lifted IO r      , HasCallStack      )   => SchedulerProxy q@@ -126,11 +124,9 @@      , Typeable o      , Show (Observation o)      , Observable o-     , Member (Logs LogMessage) q-     , MonadIO (Eff (Process q ': q))-     , MonadIO (Eff r)-     , SetMember Lift (Lift IO) r-     , Member (Logs LogMessage) r+     , HasLoggingIO r+     , HasLoggingIO q+     , Lifted IO q      , HasCallStack      )   => SchedulerProxy q@@ -180,13 +176,7 @@  withQueue   :: forall a b e-   . ( HasCallStack-     , MonadIO (Eff e)-     , Member (Logs LogMessage) e-     , Typeable a-     , Show (Observation a)-     , SetMember Lift (Lift IO) e-     )+   . (HasCallStack, Typeable a, Show (Observation a), HasLoggingIO e)   => Int   -> Eff (ObservationQueueReader a ': e) b   -> Eff e b

@@ -44,6 +44,8 @@ import           Data.Kind import           GHC.Stack import           Data.Maybe+import           GHC.Generics+import           Control.DeepSeq import           Data.Default  -- | A record of callbacks, handling requests sent to a /server/ 'Process', all@@ -174,8 +176,9 @@   -- optional reason.   StopApiServer :: Maybe String -> ApiServerCmd   --  SendReply :: reply -> ApiServerCmd () -> ApiServerCmd (reply -> Eff eff ())-  deriving (Show, Typeable)+  deriving (Show, Typeable, Generic) +instance NFData ApiServerCmd  makeLenses ''ApiHandler @@ -202,7 +205,7 @@  instance Semigroup (ServerCallback eff) where   l <> r = l & requestHandlerSelector .~-                  MessageSelector (\x ->+                  selectDynamicMessageLazy (\x ->                     runMessageSelector (view requestHandlerSelector l) x <|>                     runMessageSelector (view requestHandlerSelector r) x)              & terminationHandler .~@@ -213,7 +216,7 @@ instance Monoid (ServerCallback eff) where   mappend = (<>)   mempty = ServerCallback-              { _requestHandlerSelector = MessageSelector (const Nothing)+              { _requestHandlerSelector = selectDynamicMessageLazy (const Nothing)               , _terminationHandler = const (return ())               } @@ -335,7 +338,7 @@   -> ApiHandler api eff   -> MessageSelector (Eff eff ApiServerCmd) selectHandlerMethod px handlers =-  MessageSelector (fmap (applyHandlerMethod px handlers) . fromDynamic)+  selectDynamicMessageLazy (fmap (applyHandlerMethod px handlers) . fromDynamic)  -- | Apply either the '_callCallback', '_castCallback' or the '_terminateCallback' -- callback to an incoming request.@@ -392,5 +395,4 @@   => SchedulerProxy q   -> Maybe String   -> Eff r ()-defaultTermination px =-  maybe (return ()) (exitWithError px)+defaultTermination px = maybe (return ()) (exitWithError px)

@@ -19,7 +19,18 @@   , ConsProcess   , ResumeProcess(..)   , SchedulerProxy(..)-  , MessageSelector(..)+  , MessageSelector(runMessageSelector)+  , selectMessage+  , selectMessageLazy+  , selectMessageProxy+  , selectMessageProxyLazy+  , filterMessage+  , filterMessageLazy+  , selectMessageWith+  , selectMessageWithLazy+  , selectDynamicMessage+  , selectDynamicMessageLazy+  , selectAnyMessageLazy   , ProcessState(..)   , ProcessExitReason(..)   , ShutdownRequest(..)@@ -34,7 +45,7 @@   , spawn_   , receiveMessage   , receiveMessageAs-  , receiveMessageSuchThat+  , receiveSelectedMessage   , receiveLoop   , receiveLoopAs   , receiveLoopSuchThat@@ -60,8 +71,9 @@ import           Control.Eff.Log.Handler import           Control.Eff.Log.Message import           Control.Lens-import           Control.Monad                  ( void )-import           Control.Monad.IO.Class+import           Control.Monad                  ( void+                                                , (>=>)+                                                ) import           Data.Default import           Data.Dynamic import           Data.Kind@@ -115,12 +127,11 @@   -- destination process does not exist, or does not accept messages of the   -- given type.   SendMessage :: ProcessId -> Dynamic -> Process r (ResumeProcess Bool)-  -- | Receive a message. This should block until an a message was received. The-  -- message is returned as a 'ProcessMessage' value. The function should also-  -- return if an exception was caught or a shutdown was requested.-  ReceiveMessage :: Process r (ResumeProcess Dynamic)-  -- | Wait for the next message that matches a criterium. Similar to 'ReceiveMessage'.-  ReceiveMessageSuchThat :: MessageSelector a -> Process r (ResumeProcess a)+  -- | Receive a message that matches a criterium.+  -- This should block until an a message was received. The message is returned+  -- as a 'ProcessMessage' value. The function should also return if an exception+  -- was caught or a shutdown was requested.+  ReceiveSelectedMessage :: forall r a . MessageSelector a -> Process r (ResumeProcess a)   -- | Generate a unique 'Int' for the current process.   MakeReference :: Process r (ResumeProcess Int) @@ -147,8 +158,7 @@       . showChar ' '       . showsPrec 10 sr       )-    ReceiveMessage           -> showString "ReceiveMessage"-    ReceiveMessageSuchThat _ -> showString "ReceiveMessageSuchThat"+    ReceiveSelectedMessage _ -> showString "ReceiveSelectedMessage"     MakeReference            -> showString "MakeReference"  -- | Every 'Process' action returns it's actual result wrapped in this type. It@@ -172,12 +182,99 @@ instance NFData1 ResumeProcess  -- | A function that deciced if the next message will be received by--- 'ReceiveMessageSuchThat'. It conveniently is an instance of 'Monoid'+-- 'ReceiveSelectedMessage'. It conveniently is an instance of 'Monoid' -- with first come, first serve bais. newtype MessageSelector a =   MessageSelector {runMessageSelector :: Dynamic -> Maybe a }   deriving (Semigroup, Monoid, Functor) ++-- | Create a message selector for a value that can be obtained by 'fromDynamic'.+-- It will also 'force' the result.+--+-- @since 0.9.1+selectMessage :: (NFData t, Typeable t) => MessageSelector t+selectMessage = selectDynamicMessage fromDynamic++-- | Create a message selector for a value that can be obtained by 'fromDynamic'.+-- It will also 'force' the result.+--+-- @since 0.9.1+selectMessageLazy :: Typeable t => MessageSelector t+selectMessageLazy = selectDynamicMessageLazy fromDynamic++-- | Create a message selector from a predicate. It will 'force' the result.+--+-- @since 0.9.1+filterMessage :: (Typeable a, NFData a) => (a -> Bool) -> MessageSelector a+filterMessage predicate = selectDynamicMessage+  (\d -> case fromDynamic d of+    Just a | predicate a -> Just a+    _                    -> Nothing+  )++-- | Create a message selector from a predicate. It will 'force' the result.+--+-- @since 0.9.1+filterMessageLazy :: Typeable a => (a -> Bool) -> MessageSelector a+filterMessageLazy predicate = selectDynamicMessageLazy+  (\d -> case fromDynamic d of+    Just a | predicate a -> Just a+    _                    -> Nothing+  )++-- | Select a message of type @a@ and apply the given function to it.+-- If the function returns 'Just' The 'ReceiveSelectedMessage' function will+-- return the result (sans @Maybe@). It will 'force' the result.+--+-- @since 0.9.1+selectMessageWith+  :: (Typeable a, NFData b) => (a -> Maybe b) -> MessageSelector b+selectMessageWith f = selectDynamicMessage (fromDynamic >=> f)++-- | Select a message of type @a@ and apply the given function to it.+-- If the function returns 'Just' The 'ReceiveSelectedMessage' function will+-- return the result (sans @Maybe@). It will 'force' the result.+--+-- @since 0.9.1+selectMessageWithLazy :: Typeable a => (a -> Maybe b) -> MessageSelector b+selectMessageWithLazy f = selectDynamicMessageLazy (fromDynamic >=> f)++-- | Create a message selector. It will 'force' the result.+--+-- @since 0.9.1+selectDynamicMessage :: NFData a => (Dynamic -> Maybe a) -> MessageSelector a+selectDynamicMessage = MessageSelector . (force .)++-- | Create a message selector.+--+-- @since 0.9.1+selectDynamicMessageLazy :: (Dynamic -> Maybe a) -> MessageSelector a+selectDynamicMessageLazy = MessageSelector++-- | Create a message selector that will match every message. This is /lazy/+-- because the result is not 'force'ed.+--+-- @since 0.9.1+selectAnyMessageLazy :: MessageSelector Dynamic+selectAnyMessageLazy = MessageSelector Just++-- | Create a message selector for a value that can be obtained by 'fromDynamic'+-- with a proxy argument. It will also 'force' the result.+--+-- @since 0.9.1+selectMessageProxy+  :: forall proxy t . (NFData t, Typeable t) => proxy t -> MessageSelector t+selectMessageProxy _ = selectDynamicMessage fromDynamic++-- | Create a message selector for a value that can be obtained by 'fromDynamic'+-- with a proxy argument. It will also 'force' the result.+--+-- @since 0.9.1+selectMessageProxyLazy+  :: forall proxy t . (Typeable t) => proxy t -> MessageSelector t+selectMessageProxyLazy _ = selectDynamicMessageLazy fromDynamic+ -- | Every function for 'Process' things needs such a proxy value -- for the low-level effect list, i.e. the effects identified by -- @__r__@ in @'Process' r : r@, this might be dependent on the@@ -376,32 +473,33 @@    . (HasCallStack, SetMember Process (Process q) r)   => SchedulerProxy q   -> Eff r Dynamic-receiveMessage _ = executeAndResume ReceiveMessage+receiveMessage _ =+  executeAndResume (ReceiveSelectedMessage selectAnyMessageLazy)  -- | Block until a message was received, that is not 'Nothing' after applying -- a callback to it.--- See 'ReceiveMessageSuchThat' for more documentation.-receiveMessageSuchThat+-- See 'ReceiveSelectedMessage' for more documentation.+receiveSelectedMessage   :: forall r q a    . (HasCallStack, Typeable a, SetMember Process (Process q) r)   => SchedulerProxy q   -> MessageSelector a   -> Eff r a-receiveMessageSuchThat _ f = executeAndResume (ReceiveMessageSuchThat f)+receiveSelectedMessage _ f = executeAndResume (ReceiveSelectedMessage f)  -- | Receive and cast the message to some 'Typeable' instance.--- See 'ReceiveMessageSuchThat' for more documentation.--- This will wait for a message of the return type using 'receiveMessageSuchThat'+-- See 'ReceiveSelectedMessage' for more documentation.+-- This will wait for a message of the return type using 'receiveSelectedMessage' receiveMessageAs   :: forall a r q    . (HasCallStack, Typeable a, SetMember Process (Process q) r)   => SchedulerProxy q   -> Eff r a-receiveMessageAs px = receiveMessageSuchThat px (MessageSelector fromDynamic)+receiveMessageAs px = receiveSelectedMessage px (MessageSelector fromDynamic)  -- | Enter a loop to receive messages and pass them to a callback, until the -- function returns 'Just' a result.--- See 'receiveMesage' or 'ReceiveMessage' for more documentation.+-- See 'selectAnyMessageLazy' or 'ReceiveSelectedMessage' for more documentation. receiveLoop   :: forall r q endOfLoopResult    . (SetMember Process (Process q) r, HasCallStack)@@ -409,7 +507,7 @@   -> (Either (Maybe String) Dynamic -> Eff r (Maybe endOfLoopResult))   -> Eff r endOfLoopResult receiveLoop px handlers = do-  mReq <- send (ReceiveMessage @q)+  mReq <- send (ReceiveSelectedMessage @q selectAnyMessageLazy)   mRes <- case mReq of     RetryLastAction                          -> return Nothing     ShutdownRequested ExitNormally           -> handlers (Left Nothing)@@ -428,7 +526,7 @@   -> (Either (Maybe String) a -> Eff r (Maybe endOfLoopResult))   -> Eff r endOfLoopResult receiveLoopAs px handlers = do-  mReq <- send (ReceiveMessageSuchThat @a @q (MessageSelector fromDynamic))+  mReq <- send (ReceiveSelectedMessage @q @a (MessageSelector fromDynamic))   mRes <- case mReq of     RetryLastAction                          -> return Nothing     ShutdownRequested ExitNormally           -> handlers (Left Nothing)@@ -448,7 +546,7 @@   -> (Either (Maybe String) a -> Eff r (Maybe endOfLoopResult))   -> Eff r endOfLoopResult receiveLoopSuchThat px selectMesage handlers = do-  mReq <- send (ReceiveMessageSuchThat @a @q selectMesage)+  mReq <- send (ReceiveSelectedMessage @q @a selectMesage)   mRes <- case mReq of     RetryLastAction                          -> return Nothing     ShutdownRequested ExitNormally           -> handlers (Left Nothing)@@ -541,7 +639,7 @@  -- | Log the 'ProcessExitReaons' logProcessExit-  :: ('[Logs LogMessage] <:: e, MonadIO (Eff e), HasCallStack)+  :: (HasCallStack, HasLogWriter LogMessage h e)   => ProcessExitReason   -> Eff e () logProcessExit ex = withFrozenCallStack $ case ex of@@ -550,5 +648,5 @@   ProcessShutDown (ExitWithError m) -> logError ("exit with error: " ++ show m)   ProcessCaughtIOException w m ->     logError ("runtime exception: " ++ m ++ " caught here: " ++ w)-  ProcessRaisedError m -> logError ("unhandled process exception: " ++ show m)-  _                    -> logError ("scheduler error: " ++ show ex)+  ProcessRaisedError m  -> logError ("unhandled process exception: " ++ show m)+  SchedulerShuttingDown -> logInfo "scheduler schutting down"

@@ -38,7 +38,6 @@ import           Control.Eff.Lift import           Control.Eff.Log import           Control.Eff.Reader.Strict     as Reader-import           Control.Monad.Log              ( runLoggingT ) import           Control.Lens import           Control.Monad                  ( when                                                 , void@@ -123,7 +122,7 @@ -- | Create a new 'SchedulerState' run an IO action, catching all exceptions, -- and when the actions returns, clean up and kill all processes. withNewSchedulerState-  :: HasCallStack => Eff SchedulerIO () -> Eff LoggingAndIO ()+  :: (HasLoggingIO SchedulerIO, HasCallStack) => Eff SchedulerIO () -> Eff LoggingAndIO () withNewSchedulerState mainProcessAction =   Exceptions.tryAny (lift (atomically newSchedulerState))     >>= either@@ -182,20 +181,22 @@  -- | Start the message passing concurrency system then execute a 'Process' on -- top of 'SchedulerIO' effect. All logging is sent to standard output.-defaultMain :: HasCallStack => Eff ProcEff () -> IO ()-defaultMain c = runLoggingT-  (logChannelBracket 128-                     Nothing-                     (handleLoggingAndIO_ (schedule c))-  )-  (printLogMessage :: LogMessage -> IO ())+defaultMain+  :: HasCallStack => (HasLogWriterProxy IO => Eff ProcEff ()) -> IO ()+defaultMain c =+  withAsyncLogChannel 1024+    (multiMessageLogWriter ($ printLogMessage))+    (handleLoggingAndIO_ (schedule c))  -- | Start the message passing concurrency system then execute a 'Process' on -- top of 'SchedulerIO' effect. All logging is sent to standard output. defaultMainWithLogChannel-  :: HasCallStack => LogChannel LogMessage -> Eff ProcEff () -> IO ()-defaultMainWithLogChannel logC c =-  closeLogChannelAfter logC (handleLoggingAndIO_ (schedule c) logC)+  :: HasCallStack+  => (HasLogWriterProxy IO => Eff ProcEff ())+  -> LogChannel LogMessage+  -> IO ()+defaultMainWithLogChannel c =+  handleLoggingAndIO_ (schedule c)  -- | A 'SchedulerProxy' for 'SchedulerIO' forkIoScheduler :: SchedulerProxy SchedulerIO@@ -221,7 +222,7 @@ -- ** MessagePassing execution  handleProcess-  :: HasCallStack+  :: (HasLoggingIO SchedulerIO, HasCallStack)   => ProcessInfo   -> Eff ProcEff ProcessExitReason   -> Eff SchedulerIO ProcessExitReason@@ -317,8 +318,7 @@           then diskontinue (ProcessShutDown r)           else kontinue (ShutdownRequested shutdownRequest)       Spawn _                  -> kontinue (ShutdownRequested shutdownRequest)-      ReceiveMessage           -> kontinue (ShutdownRequested shutdownRequest)-      ReceiveMessageSuchThat _ -> kontinue (ShutdownRequested shutdownRequest)+      ReceiveSelectedMessage _ -> kontinue (ShutdownRequested shutdownRequest)       SelfPid                  -> kontinue (ShutdownRequested shutdownRequest)       MakeReference            -> kontinue (ShutdownRequested shutdownRequest)       YieldProcess             -> kontinue (ShutdownRequested shutdownRequest)@@ -341,8 +341,7 @@           then kontinue nextRef (ShutdownRequested msg)           else interpretSendShutdown toPid msg >>= kontinue nextRef . ResumeWith       Spawn child              -> spawnNewProcess child >>= kontinue nextRef . ResumeWith . fst-      ReceiveMessage           -> interpretReceive (MessageSelector Just) >>= kontinue nextRef-      ReceiveMessageSuchThat f -> interpretReceive f >>= kontinue nextRef+      ReceiveSelectedMessage f -> interpretReceive f >>= kontinue nextRef       SelfPid                  -> kontinue nextRef (ResumeWith myPid)       MakeReference            -> kontinue (nextRef + 1) (ResumeWith nextRef)       YieldProcess             -> kontinue nextRef (ResumeWith  ())@@ -384,7 +383,10 @@ -- | This is the main entry point to running a message passing concurrency -- application. This function takes a 'Process' on top of the 'SchedulerIO' -- effect and a 'LogChannel' for concurrent logging.-schedule :: HasCallStack => Eff ProcEff () -> Eff LoggingAndIO ()+schedule+  :: (HasLoggingIO SchedulerIO, HasCallStack)+  => Eff ProcEff ()+  -> Eff LoggingAndIO () schedule procEff =   liftBaseWith (\runS ->     Async.withAsync (runS $ withNewSchedulerState $ do@@ -400,7 +402,10 @@       )   ) >>= restoreM -spawnNewProcess :: Eff ProcEff () -> Eff SchedulerIO (ProcessId, Async ProcessExitReason)+spawnNewProcess+  :: (HasLoggingIO SchedulerIO, HasCallStack)+  => Eff ProcEff ()+  -> Eff SchedulerIO (ProcessId, Async ProcessExitReason) spawnNewProcess mfa = do   schedulerState <- getSchedulerState   liftBaseWith

@@ -45,6 +45,7 @@ import           Control.Monad import           Data.Foldable import           Data.Typeable                  ( Typeable )+import Control.DeepSeq import           System.Timeout  -- | Contains the communication channels to interact with a scheduler running in@@ -140,7 +141,7 @@ -- | Combination of 'submit' and 'cast'. submitCall   :: forall o q r-   . (SetMember Lift (Lift IO) r, Typeable o, Typeable q)+   . (SetMember Lift (Lift IO) r, Typeable o, Typeable q, NFData q)   => SchedulerSession r   -> Server o   -> Api o ( 'Synchronous q)

@@ -36,8 +36,16 @@ -- @schedulePure == runIdentity . 'scheduleM' (Identity . run)  (return ())@ -- -- @since 0.3.0.2-schedulePure :: Eff (ConsProcess '[]) a -> Either String a-schedulePure = runIdentity . scheduleM (Identity . run) (return ())+schedulePure+  :: (  HasLogWriterProxy Identity+     => Eff (ConsProcess '[LogsM LogMessage Identity, Lift Identity]) a+     )+  -> Either String a+schedulePure e = runIdentity+  (scheduleM (runLift . handleLogs traceLogMessageWriter)+             (return ())+             (usingLogWriterProxy (LogWriterProxy @Identity) e)+  )  -- | Invoke 'schedule' with @lift 'Control.Concurrent.yield'@ as yield effect. -- @scheduleIO runEff == 'scheduleM' (runLift . runEff) (liftIO 'yield')@@@ -69,10 +77,10 @@ -- @since 0.4.0.0 scheduleIOWithLogging   :: (NFData l)-  => (l -> IO ())+  => LogWriter l IO   -> Eff (ConsProcess '[Logs l, Lift IO]) a   -> IO (Either String a)-scheduleIOWithLogging handleLog = scheduleIO (handleLogsWith handleLog)+scheduleIOWithLogging h = scheduleIO (handleLogs h)  -- | Handle the 'Process' effect, as well as all lower effects using an effect handler function. --@@ -91,14 +99,16 @@ -- -- @since 0.4.0.0 scheduleM-  :: Monad m+  :: forall m r a+   . Monad m   => (forall b . Eff r b -> m b)   -> m () -- ^ An that performs a __yield__ w.r.t. the underlying effect   --  @r@. E.g. if @Lift IO@ is present, this might be:   --  @lift 'Control.Concurrent.yield'.-  -> Eff (ConsProcess r) a+  -> (HasLogWriterProxy m => Eff (ConsProcess r) a)   -> m (Either String a)-scheduleM runEff yieldEff e = do+scheduleM runEff yieldEff e' = do+  let e = (usingLogWriterProxy (LogWriterProxy @m) e')   y <- runAsCoroutinePure runEff e   handleProcess runEff                 yieldEff@@ -247,7 +257,7 @@                         (msgQs & at newPid ?~ Seq.empty)                         (rest :|> (nextK, pid) :|> (fk, newPid)) -        recv@(OnRecv selectMessage k) -> case msgQs ^. at pid of+        recv@(OnRecv messageSelector k) -> case msgQs ^. at pid of           Nothing -> do             nextK <- runEff $ k (OnError (show pid ++ " has no message queue!"))             handleProcess runEff@@ -277,7 +287,7 @@                 partitionMessages (m :<| msgRest) acc  = maybe                   (partitionMessages msgRest (acc :|> m))                   (\res -> Just (res, acc Seq.>< msgRest))-                  (runMessageSelector selectMessage m)+                  (runMessageSelector messageSelector m)             in  case partitionMessages messages Empty of                   Nothing -> handleProcess runEff                                            yieldEff@@ -309,8 +319,7 @@   cont (Shutdown   !sr      )     _k = return (OnShutdown sr)   cont (RaiseError !e       )     _k = return (OnRaiseError e)   cont (SendMessage !tp !msg)     k  = return (OnSend tp msg k)-  cont ReceiveMessage             k  = return (OnRecv (MessageSelector Just) k)-  cont (ReceiveMessageSuchThat f) k  = return (OnRecv f k)+  cont (ReceiveSelectedMessage f) k  = return (OnRecv f k)   cont (SendShutdown !pid !sr   ) k  = return (OnSendShutdown pid sr k)   cont MakeReference              k  = return (OnMakeReference k) @@ -334,5 +343,5 @@ defaultMain =   void     . runLift-    . handleLogsWithLoggingTHandler ($ printLogMessage)+    . handleLogs (multiMessageLogWriter ($ printLogMessage))     . scheduleMonadIOEff

@@ -1,207 +1,102 @@ -- | Concurrent Logging module Control.Eff.Log.Channel   ( LogChannel()-  , logToChannel-  , nullLogChannel-  , forkLogger-  , filterLogChannel-  , joinLogChannel-  , killLogChannel-  , closeLogChannelAfter-  , logChannelBracket-  , logChannelPutIO-  , JoinLogChannelException()-  , KillLogChannelException()+  , withAsyncLogChannel   , handleLoggingAndIO   , handleLoggingAndIO_   ) where -import           Control.Concurrent+import           Control.Concurrent.Async import           Control.Concurrent.STM import           Control.Eff                   as Eff import           Control.Eff.Lift-import           Control.Exception              ( bracket )-import qualified Control.Exception             as Exc+import           Control.Exception              ( evaluate ) import           Control.Monad                  ( void-                                                , when                                                 , unless                                                 )-import           Control.Monad.Log             as ExtLog-                                         hiding ( )-import           Control.Monad.Trans.Control import           Control.Eff.Log.Handler-import qualified Control.Eff.Lift              as Eff import           Data.Foldable                  ( traverse_ ) import           Data.Kind                      ( )-import           Data.String-import           Data.Typeable+import           Control.DeepSeq import           GHC.Stack --- | A log channel processes logs from the 'Logs' effect by en-queuing them in a--- shared queue read from a seperate processes. A channel can contain log--- message filters.-data LogChannel message =-   FilteredLogChannel (message -> Bool) (LogChannel message)-   -- ^ filter log messages- | DiscardLogs-   -- ^ discard all log messages- | ConcurrentLogChannel-   { fromLogChannel :: TBQueue message-   , _logChannelThread :: ThreadId-   }-   -- ^ send all log messages to a log process --- | Send the log messages to a 'LogChannel'.-logToChannel-  :: forall r message a-   . (SetMember Eff.Lift (Eff.Lift IO) r)-  => LogChannel message-  -> Eff (Logs message ': r) a-  -> Eff r a-logToChannel logChan =-  handleLogsWithLoggingTHandler ($ logChannelPutIO logChan)---- | Enqueue a log message into a log channel-logChannelPutIO :: LogChannel message -> message -> IO ()-logChannelPutIO DiscardLogs               _ = return ()-logChannelPutIO (FilteredLogChannel f lc) m = when (f m) (logChannelPutIO lc m)-logChannelPutIO c                         m = atomically $ do-  dropMessage <- isFullTBQueue (fromLogChannel c)-  unless dropMessage (writeTBQueue (fromLogChannel c) m)---- | Create a 'LogChannel' that will discard all messages sent--- via 'forwardLogstochannel' or 'logChannelPutIO'.-nullLogChannel :: LogChannel message-nullLogChannel = DiscardLogs---- | Fork a new process, that applies a monadic action to all log messages sent--- via 'logToChannel' or 'logChannelPutIO'.-forkLogger-  :: forall message-   . (Typeable message)+-- | Fork a new process in which the given log message writer, will listen+-- on a message queue in a 'LogChannel', which is passed to the second function.+-- If the function returns or throws, the logging process will be killed.+--+-- Log messages are deeply evaluated before being sent to the logger process,+-- to prevent that lazy evaluation leads to heavy work being done in the+-- logger process instead of the caller process.+--+-- Example usage, a super stupid log to file:+--+-- >+-- > main =+-- >   withAsyncLogChannel+-- >      1000+-- >      (singleMessageLogWriter putStrLn)+-- >      (handleLoggingAndIO+-- >        (do logMsg "test 1"+-- >            logMsg "test 2"+-- >            logMsg "test 3"))+-- >+--+withAsyncLogChannel+  :: forall message a+   . (NFData message)   => Int -- ^ Size of the log message input queue. If the queue is full, message         -- are dropped silently.-  -> (message -> IO ()) -- ^ An IO action to log the messages-  -> Maybe message -- ^ Optional __first__ message to log-  -> IO (LogChannel message)-forkLogger queueLen handle mFirstMsg = do-  msgQ <- atomically-    (do-      tq <- newTBQueue (fromIntegral @Int queueLen)-      mapM_ (writeTBQueue tq) mFirstMsg-      return tq-    )-  thread <- forkFinally (logLoop msgQ) (writeLastLogs msgQ)-  return (ConcurrentLogChannel msgQ thread)+  -> LogWriter message IO -- ^ An IO action to write the log messages+  -> (LogChannel message -> IO a)+  -> IO a+withAsyncLogChannel queueLen ioWriter action = do+  msgQ <- newTBQueueIO (fromIntegral @Int queueLen)+  withAsync (logLoop msgQ) (action . ConcurrentLogChannel msgQ)  where-  writeLastLogs :: TBQueue message -> Either Exc.SomeException () -> IO ()-  writeLastLogs tq ee = do-    logMessages <- atomically $ flushTBQueue tq-    case ee of-      Right _  -> return ()-      Left  se -> case Exc.fromException se of-        Just JoinLogChannelException -> traverse_ handle logMessages-        Nothing                      -> case Exc.fromException se of-          Just KillLogChannelException -> return ()-          Nothing                      -> mapM_ handle logMessages--  logLoop :: TBQueue message -> IO ()   logLoop tq = do-    m <- atomically $ do+    ms <- atomically $ do       h <- readTBQueue tq       t <- flushTBQueue tq       return (h : t)-    traverse_ handle m+    writeAllLogMessages ioWriter ms     logLoop tq --- | Filter logs sent to a 'LogChannel' using a predicate.-filterLogChannel-  :: (message -> Bool) -> LogChannel message -> LogChannel message-filterLogChannel = FilteredLogChannel---- | Run an action and close a 'LogChannel' created by 'nullLogChannel', 'forkLogger'--- or 'filterLogChannel' afterwards using 'joinLogChannel'. If a--- 'Exc.SomeException' was thrown, the log channel is killed with--- 'killLogChannel', and the exception is re-thrown.-closeLogChannelAfter-  :: (Typeable message, IsString message) => LogChannel message -> IO a -> IO a-closeLogChannelAfter logC ioAction = do-  res <- closeLogAndRethrow `Exc.handle` ioAction-  closeLogSuccess-  return res- where-  closeLogAndRethrow :: Exc.SomeException -> IO a-  closeLogAndRethrow se = do-    void $ Exc.try @Exc.SomeException $ killLogChannel logC-    Exc.throw se--  closeLogSuccess :: IO ()-  closeLogSuccess = joinLogChannel logC---- | Close a log channel created by e.g. 'forkLogger'. Message already enqueue--- are handled. Subsequent log message--- will not be handled anymore. If the log channel must be closed immediately,--- use 'killLogChannel' instead.-joinLogChannel :: (Typeable message) => LogChannel message -> IO ()-joinLogChannel DiscardLogs                = return ()-joinLogChannel (FilteredLogChannel _f lc) = joinLogChannel lc-joinLogChannel (ConcurrentLogChannel _tq thread) =-  throwTo thread JoinLogChannelException---- | Close a log channel quickly, without logging messages already in the queue.--- Subsequent logging requests will not be handled anymore. If the log channel--- must be closed without loosing any messages, use 'joinLogChannel' instead.-killLogChannel :: (Typeable message) => LogChannel message -> IO ()-killLogChannel DiscardLogs                = return ()-killLogChannel (FilteredLogChannel _f lc) = killLogChannel lc-killLogChannel (ConcurrentLogChannel _tq thread) =-  throwTo thread KillLogChannelException---- | Internal exception to shutdown a 'LogChannel' process created by--- 'forkLogger'. This exception is handled such that all message already--- en-queued are handled and then an optional final message is written.-data JoinLogChannelException = JoinLogChannelException-  deriving (Show, Typeable)--instance Exc.Exception JoinLogChannelException---- | Internal exception to **immediately** shutdown a 'LogChannel' process--- created by 'forkLogger', other than 'JoinLogChannelException' the message queue--- will not be flushed, not further messages will be logged, except for the--- optional final message.-data KillLogChannelException = KillLogChannelException-  deriving (Show, Typeable)--instance Exc.Exception KillLogChannelException---- | Wrap 'LogChannel' creation and destruction around a monad action in--- 'bracket'y manner. This function uses 'joinLogChannel', so en-queued messages--- are flushed on exit. The resulting action is a 'LoggingT' action, which--- is essentially a reader for a log handler function in 'IO'.-logChannelBracket-  :: (Typeable message)-  => Int -- ^ Size of the log message input queue. If the queue is full, message-        -- are dropped silently.-  -> Maybe message -- ^ Optional __first__ message to log-  -> (LogChannel message -> IO a) -- ^ An IO action that will use the-                                -- 'LogChannel', after the action returns (even-                                -- because of an exception) the log channel is-                                -- destroyed.-  -> LoggingT message IO a-logChannelBracket queueLen mWelcome f = control-  (\runInIO -> do-    let logHandler = void . runInIO . logMessage-    bracket (forkLogger queueLen logHandler mWelcome) joinLogChannel f-  )---- | Handle the 'LoggingAndIO' effects, using a 'LogChannel' for the 'Logs'--- effect.+-- | Fork an IO based log writer thread and set the 'LogWriter' to an action+-- that will send all logs to that thread via a bounded queue.+-- When the queue is full, flush it handleLoggingAndIO-  :: HasCallStack => Eff '[Logs m, Lift IO] a -> LogChannel m -> IO a-handleLoggingAndIO e lc = runLift (logToChannel lc e)+  :: (NFData m, HasCallStack)+  => (HasLogWriterProxy IO => Eff '[Logs m, Lift IO] a)+  -> LogChannel m+  -> IO a+handleLoggingAndIO e lc = runLift+  (handleLogs (foldingLogWriter (traverse_ logChannelPutIO)) e)+ where+  logQ = fromLogChannel lc+  logChannelPutIO (force -> me) = do+    !m <- evaluate me+    atomically+      (do+        dropMessage <- isFullTBQueue logQ+        unless dropMessage (writeTBQueue logQ m)+      )  -- | Like 'handleLoggingAndIO' but return @()@. handleLoggingAndIO_-  :: HasCallStack => Eff '[Logs m, Lift IO] a -> LogChannel m -> IO ()-handleLoggingAndIO_ = ((.) . (.)) void handleLoggingAndIO+  :: (NFData m, HasCallStack)+  => (HasLogWriterProxy IO => Eff '[Logs m, Lift IO] a)+  -> LogChannel m+  -> IO ()+handleLoggingAndIO_ e lc = void (handleLoggingAndIO e lc)++-- | A log channel processes logs from the 'Logs' effect by en-queuing them in a+-- shared queue read from a seperate processes. A channel can contain log+-- message filters.+data LogChannel message =+   ConcurrentLogChannel+   { fromLogChannel :: TBQueue message+   , _logChannelThread :: Async ()+   }+   -- ^ send all log messages to a log process

@@ -1,47 +1,155 @@+{-# LANGUAGE ImplicitParams #-}+ -- | A logging effect based on 'Control.Monad.Log.MonadLog'. module Control.Eff.Log.Handler-  (-    -- * Logging Effect-    Logs-  , LogWriter(..)-  , askLogWriter-  , logMsg-  , interceptLogging-  , foldLogMessages+  ( logMsg+  , logMsgs+  , LogsM+  , Logs+  , handleLogs   , ignoreLogs-  , handleLogsWith-  , handleLogsWithLoggingTHandler+  , mapLogMessages+  , filterLogMessages+  , interceptLogging+  , HasLogWriter+  , HasLogWriterIO+  , LogWriter()+  , writeAllLogMessages+  , foldingLogWriter+  , singleMessageLogWriter+  , multiMessageLogWriter+  , HasLogWriterProxy+  , LogWriterProxy(..)+  , usingLogWriterProxy+  , logWriterProxy+  , askLogWriter   ) where  import           Control.DeepSeq import           Control.Eff                   as Eff import           Control.Eff.Reader.Strict-import qualified Control.Eff.Lift              as Eff-import qualified Control.Monad.Log             as Log-import           Control.Monad.IO.Class-import           Data.Foldable                  ( traverse_ )-import           Data.Kind                      ( )+import           Control.Eff.Lift              as Eff+import           Data.Foldable                  ( traverse_, toList )+import           Data.Kind+import           Data.Functor.Identity+import           GHC.Stack+import           Data.Default  -- | A function that takes a log message and returns an effect that -- /logs/ the message.-newtype LogWriter message = LogWriter { runLogWriter :: message -> IO () }+newtype LogWriter message writerM =+  LogWriter+  { runLogWriter+      :: forall f. (HasCallStack, Traversable f, Foldable f, Functor f)+      => f message+      -> writerM ()+  } +instance Applicative w => Default (LogWriter m w) where+  def = LogWriter (const (pure ()))++-- | Efficiently apply the 'LogWriter' to a 'Foldable' container of log+-- messages.+writeAllLogMessages+  :: (Foldable f, Functor f, Traversable f, Applicative writerM, HasCallStack)+  => LogWriter message writerM -> f message -> writerM ()+writeAllLogMessages = runLogWriter++-- | Create a 'LogWriter' from a function that can write+-- a 'Foldable' container.+foldingLogWriter+  :: (forall f. (Foldable f, Traversable f, HasCallStack) => f message -> writerM ())+  -> LogWriter message writerM+foldingLogWriter = LogWriter++-- | Create a 'LogWriter' from a function that is applied to each+-- individual log message. NOTE: This is probably the simplest,+-- but also the most inefficient and annoying way to make+-- a 'LogWriter'. Better use 'foldingLogWriter' or even+-- 'multiMessageLogWriter'.+singleMessageLogWriter+  :: (Applicative writerM, HasCallStack)+  => (message -> writerM ())+  -> LogWriter message writerM+singleMessageLogWriter writeMessage =+  foldingLogWriter (traverse_ writeMessage)++-- | Create a 'LogWriter' from a function that is applied to each+-- individual log message. Don't be scared by the type signature,+-- here is an example file appender that re-opens the log file+-- everytime a bunch of log messages are written:+--+-- > fileAppender fn = multiMessageLogWriter+-- >   (\writeLogMessageWith ->+-- >      withFile fn AppendMode (writeLogMessageWith . hPutStrLn))+--+multiMessageLogWriter+  :: (Applicative writerM, HasCallStack)+  => (((message -> writerM ()) -> writerM ()) -> writerM ())+  -> LogWriter message writerM+multiMessageLogWriter withMessageWriter =+  foldingLogWriter+    (\xs -> withMessageWriter (\writer -> traverse_ writer xs))+ -- | Logging effect type, parameterized by a log message type. This is a reader -- for a 'LogWriter'.-type Logs message = Reader (LogWriter message)+type LogsM message writerM = Reader (LogWriter message writerM) --- | Log a message.-logMsg :: (NFData m, MonadIO (Eff r), Member (Logs m) r) => m -> Eff r ()-logMsg !(force -> m) = do-  lw <- ask-  liftIO (runLogWriter lw m)+-- | The logging effect on IO.+type Logs message = LogsM message IO +-- | A constraint that combines constraints for logging into any+-- log writer monad.+type HasLogWriter message logWriterMonad effects =+  ( Member (LogsM message logWriterMonad) effects+  , NFData message+  , Monad logWriterMonad+  , HasLogWriterProxy logWriterMonad+  , Lifted logWriterMonad effects ) +-- | A constraint that combines constraints for IO based logging.+type HasLogWriterIO message effects =+  ( Member (Logs message) effects+  , NFData message+  , HasLogWriterProxy IO+  , Lifted IO effects )++-- | Type constraint alias for the implicit 'LogWriterProxy'+type HasLogWriterProxy m = ( ?logWriterProxy :: LogWriterProxy m )++-- | A proxy type to carry the 'LogWriter' effects.+data LogWriterProxy (e :: Type -> Type) = LogWriterProxy++-- | Set the implicit parameter defined in 'HasLogWriterProxy' for function+-- applications like 'logMsg'+usingLogWriterProxy+  :: forall e r a+  . LogWriterProxy e+  -> (HasLogWriterProxy e => Eff r a) -> Eff r a+usingLogWriterProxy px e = let ?logWriterProxy = px in e++-- | Return the current 'LogWriterProxy'+logWriterProxy :: HasLogWriterProxy e => LogWriterProxy e+logWriterProxy = ?logWriterProxy+ -- | Get the current 'LogWriter'-askLogWriter :: (Member (Logs m) r) => Eff r (LogWriter m)+askLogWriter :: forall m h r . (HasLogWriter m h r) => Eff r (LogWriter m h) askLogWriter = ask +-- | Log a message.+logMsg :: (NFData m, HasLogWriter m h r) => m -> Eff r ()+logMsg !(force -> m) = do+  lw <- askLogWriter+  lift (runLogWriter lw (Identity m))++-- | Log a bunch of messages. This might be more efficient than calling 'logMsg'+-- multiple times.+logMsgs :: (NFData (f m), HasLogWriter m h r, Functor f, Traversable f, Foldable f) => f m -> Eff r ()+logMsgs !(force -> m) = do+  lw <- askLogWriter+  lift (runLogWriter lw m)+ -- | Change, add or remove log messages and perform arbitrary actions upon -- intercepting a log message. --@@ -60,45 +168,59 @@ -- sending/receiving, to intercept the messages and execute some code and then -- return a new message. interceptLogging-  :: forall r m a-   . (Member (Logs m) r, Eff.Lifted IO r)-  => (m -> Eff '[Logs m, Eff.Lift IO] ())-  -> Eff r a+  :: forall r m h a+   . (HasLogWriter m h r)+  => (m -> Eff '[LogsM m h, Lift h] ())+  -> (HasLogWriterProxy h => Eff r a)   -> Eff r a-interceptLogging interceptor action = do-  old <- ask-  let new = LogWriter (Eff.runLift . runReader old . interceptor)-  local (const new) action+interceptLogging interceptor  =+  let replaceWriter old =+        multiMessageLogWriter+          ($ (runLift . runReader old . interceptor))+  in local replaceWriter --- | Intercept logging to change, add or remove log messages.------ This is without side effects, hence faster than 'interceptLogging'.-foldLogMessages-  :: forall r m a f-   . (NFData m, Foldable f, Member (Logs m) r, Eff.Lifted IO r)-  => (m -> f m)-  -> Eff r a+-- | Map a pure function over log messages.+mapLogMessages+  :: forall m h r a+   . (HasLogWriter m h r)+  => (m -> m)+  -> (HasLogWriterProxy h => Eff r a)   -> Eff r a-foldLogMessages f = interceptLogging (traverse_ logMsg . f)+mapLogMessages f =+  local @(LogWriter m h)+    (\(LogWriter old) ->+      LogWriter (\xs ->+        old (fmap f xs))) --- | Throw away all log messages.-ignoreLogs :: forall message r a . Eff (Logs message ': r) a -> Eff r a-ignoreLogs = runReader (LogWriter (const (pure ())))+-- | Filter which messages are handled.+filterLogMessages+  :: forall r m h a+   . (HasLogWriter m h r, HasCallStack)+  => (m -> Bool)+  -> (HasLogWriterProxy h => Eff r a)+  -> Eff r a+filterLogMessages predicate =+  local+    @(LogWriter m h)+    (\(LogWriter old) ->+      LogWriter (old . filter predicate . toList)) --- | Apply a function that returns an effect to each log message.-handleLogsWith-  :: forall message r a-   . (message -> IO ())-  -> Eff (Logs message ': r) a+-- | Define the the 'LogWriter' callback and set the 'HasLogWriterProxy'+-- constraint in the given log writer action. This assumes that the monad used+-- is 'Lift'ed to @r@.+handleLogs+  :: forall message writerM r a+   . (Lifted writerM r)+  => LogWriter message writerM+  -> (HasLogWriterProxy writerM => Eff (LogsM message writerM ': r) a)   -> Eff r a-handleLogsWith = runReader . LogWriter+handleLogs lw e =+  usingLogWriterProxy (LogWriterProxy @writerM) (runReader lw e) --- | Handle the 'Logs' effect using 'Log.LoggingT' 'Log.Handler's.-handleLogsWithLoggingTHandler-  :: forall r message a-   . (Eff.Lifted IO r)-  => (forall b . (Log.Handler IO message -> IO b) -> IO b)-  -> Eff (Logs message ': r) a+-- | Throw away all log messages.+ignoreLogs+  :: forall message m r a+  . (Lifted m r, Applicative m)+  => (HasLogWriterProxy m => Eff (LogsM message m ': r) a)   -> Eff r a-handleLogsWithLoggingTHandler foldHandler =-  handleLogsWith (foldHandler . flip ($!))+ignoreLogs = handleLogs def

@@ -2,9 +2,14 @@ -- logging them. module Control.Eff.Log.Message   ( LogMessage(..)+  , HasLogging+  , HasLoggingIO   , renderRFC5424   , printLogMessage-  , relogAsDebugMessages+  , ioLogMessageHandler+  , ioLogMessageWriter+  , traceLogMessageWriter+  , renderLogMessage   , increaseLogMessageDistance   , dropDistantLogMessages   , logWithSeverity@@ -85,10 +90,12 @@ import           Control.Monad                  ( (>=>) ) import           Control.Monad.IO.Class import           Data.Default+import           Data.Foldable import           Data.Maybe import           Data.String import           Data.Time.Clock import           Data.Time.Format+import           Debug.Trace import           GHC.Generics import           GHC.Stack import           System.FilePath.Posix@@ -110,6 +117,13 @@              , _lmDistance :: Int }   deriving (Eq, Generic) +-- | A convenient alias for the constraints that enable logging of 'LogMessage's+-- in the monad, which is 'Lift'ed into a given @Eff@ effect list.+type HasLogging writerM effect = (HasLogWriter LogMessage writerM effect)++-- | Like 'HasLogging' but with 'IO' as effect.+type HasLoggingIO effect = (HasLogWriterIO LogMessage effect)+ showLmMessage :: LogMessage -> [String] showLmMessage (LogMessage _f _s _ts _hn _an _pid _mi _sd ti loc msg _dist) =   if null msg@@ -127,7 +141,13 @@           )           loc +-- | A 'LogWriter' that applys 'renderLogMessage' to the log message and then+-- traces it using 'traceM'.+traceLogMessageWriter :: Monad m => LogWriter LogMessage m+traceLogMessageWriter =+  foldingLogWriter (traverse_ (traceM . renderLogMessage)) + -- | Render a 'LogMessage' human readable. renderLogMessage :: LogMessage -> String renderLogMessage l@(LogMessage _f s ts hn an pid mi sd _ _ _ _) =@@ -237,53 +257,62 @@ printLogMessage :: LogMessage -> IO () printLogMessage = setLogMessageTimestamp >=> putStrLn . renderLogMessage +-- | Set a timestamp (if not set), the thread id (if not set) using IO actions+-- then /write/ the log message using the 'IO' and 'String' based 'LogWriter'.+ioLogMessageWriter+  :: HasCallStack => LogWriter String IO -> LogWriter LogMessage IO+ioLogMessageWriter delegatee = foldingLogWriter+  (   traverse setLogMessageTimestamp+  >=> traverse setLogMessageThreadId+  >=> (writeAllLogMessages delegatee . fmap renderLogMessage)+  )++-- | Use 'ioLogMessageWriter' to /handle/ logging using 'handleLogs'.+ioLogMessageHandler+  :: (HasCallStack, Lifted IO e)+  => LogWriter String IO+  -> (HasLogWriterProxy IO => Eff (Logs LogMessage ': e) a)+  -> Eff e a+ioLogMessageHandler delegatee = handleLogs (ioLogMessageWriter delegatee)+ -- | An IO action that sets the current UTC time (see 'enableLogMessageTimestamps') -- in 'lmTimestamp'. setLogMessageTimestamp :: MonadIO m => LogMessage -> m LogMessage-setLogMessageTimestamp m = do-  now <- liftIO getCurrentTime-  return (m & lmTimestamp ?~ now)+setLogMessageTimestamp m = if isNothing (m ^. lmTimestamp)+  then do+    now <- liftIO getCurrentTime+    return (m & lmTimestamp ?~ now)+  else return m  -- | An IO action appends the the 'ThreadId' of the calling process (see 'myThreadId') -- to 'lmMessage'. setLogMessageThreadId :: MonadIO m => LogMessage -> m LogMessage-setLogMessageThreadId m = do-  t <- liftIO myThreadId-  return (m & lmThreadId ?~ t)+setLogMessageThreadId m = if isNothing (m ^. lmThreadId)+  then do+    t <- liftIO myThreadId+    return (m & lmThreadId ?~ t)+  else return m  -- | Increase the /distance/ of log messages by one. -- Logs can be filtered by their distance with 'dropDistantLogMessages' increaseLogMessageDistance-  :: (HasCallStack, Member (Logs LogMessage) e, MonadIO (Eff e), Lifted IO e)-  => Eff e a-  -> Eff e a-increaseLogMessageDistance = interceptLogging (logMsg . over lmDistance (+ 1))+  :: (HasCallStack, HasLogWriter LogMessage h e) => Eff e a -> Eff e a+increaseLogMessageDistance = mapLogMessages (over lmDistance (+ 1))  -- | Drop all log messages with an 'lmDistance' greater than the given -- value.-dropDistantLogMessages-  :: (SetMember Lift (Lift IO) r, Member (Logs LogMessage) r)-  => Int-  -> Eff r a-  -> Eff r a-dropDistantLogMessages maxDistance = foldLogMessages-  (\lm -> if lm ^. lmDistance > maxDistance then Nothing else Just lm)---- | Handle a 'Logs' effect for 'String' messages by re-logging the messages--- as 'LogMessage's with 'debugSeverity'.-relogAsDebugMessages-  :: (HasCallStack, Member (Logs LogMessage) e, MonadIO (Eff e))-  => Eff (Logs String ': e) a-  -> Eff e a-relogAsDebugMessages e = withFrozenCallStack-  (do-    LogWriter logMsgLogger <- askLogWriter-    handleLogsWith (logMsgLogger . debugMessage) e-  )+dropDistantLogMessages :: (HasLogging m r) => Int -> Eff r a -> Eff r a+dropDistantLogMessages maxDistance =+  filterLogMessages (\lm -> lm ^. lmDistance <= maxDistance)  -- | Log a 'String' as 'LogMessage' with a given 'Severity'. logWithSeverity-  :: (HasCallStack, Member (Logs LogMessage) e, MonadIO (Eff e))+  :: ( HasLogWriterProxy h+     , HasCallStack+     , Monad h+     , Member (LogsM LogMessage h) e+     , Lifted h e+     )   => Severity   -> String   -> Eff e ()@@ -296,56 +325,96 @@  -- | Log a 'String' as 'emergencySeverity'. logEmergency-  :: (HasCallStack, Member (Logs LogMessage) e, MonadIO (Eff e))+  :: ( HasCallStack+     , HasLogWriterProxy h+     , Monad h+     , Member (LogsM LogMessage h) e+     , Lifted h e+     )   => String   -> Eff e () logEmergency = withFrozenCallStack (logWithSeverity emergencySeverity)  -- | Log a message with 'alertSeverity'. logAlert-  :: (HasCallStack, Member (Logs LogMessage) e, MonadIO (Eff e))+  :: ( HasCallStack+     , HasLogWriterProxy h+     , Monad h+     , Member (LogsM LogMessage h) e+     , Lifted h e+     )   => String   -> Eff e () logAlert = withFrozenCallStack (logWithSeverity alertSeverity)  -- | Log a 'criticalSeverity' message. logCritical-  :: (HasCallStack, Member (Logs LogMessage) e, MonadIO (Eff e))+  :: ( HasCallStack+     , HasLogWriterProxy h+     , Monad h+     , Member (LogsM LogMessage h) e+     , Lifted h e+     )   => String   -> Eff e () logCritical = withFrozenCallStack (logWithSeverity criticalSeverity)  -- | Log a 'errorSeverity' message. logError-  :: (HasCallStack, Member (Logs LogMessage) e, MonadIO (Eff e))+  :: ( HasCallStack+     , HasLogWriterProxy h+     , Monad h+     , Member (LogsM LogMessage h) e+     , Lifted h e+     )   => String   -> Eff e () logError = withFrozenCallStack (logWithSeverity errorSeverity)  -- | Log a 'warningSeverity' message. logWarning-  :: (HasCallStack, Member (Logs LogMessage) e, MonadIO (Eff e))+  :: ( HasCallStack+     , HasLogWriterProxy h+     , Monad h+     , Member (LogsM LogMessage h) e+     , Lifted h e+     )   => String   -> Eff e () logWarning = withFrozenCallStack (logWithSeverity warningSeverity)  -- | Log a 'noticeSeverity' message. logNotice-  :: (HasCallStack, Member (Logs LogMessage) e, MonadIO (Eff e))+  :: ( HasCallStack+     , HasLogWriterProxy h+     , Monad h+     , Member (LogsM LogMessage h) e+     , Lifted h e+     )   => String   -> Eff e () logNotice = withFrozenCallStack (logWithSeverity noticeSeverity)  -- | Log a 'informationalSeverity' message. logInfo-  :: (HasCallStack, Member (Logs LogMessage) e, MonadIO (Eff e))+  :: ( HasCallStack+     , HasLogWriterProxy h+     , Monad h+     , Member (LogsM LogMessage h) e+     , Lifted h e+     )   => String   -> Eff e () logInfo = withFrozenCallStack (logWithSeverity informationalSeverity)  -- | Log a 'debugSeverity' message. logDebug-  :: (HasCallStack, Member (Logs LogMessage) e, MonadIO (Eff e))+  :: ( HasLogWriterProxy h+     , HasCallStack+     , Monad h+     , Member (LogsM LogMessage h) e+     , Lifted h e+     )   => String   -> Eff e () logDebug = withFrozenCallStack (logWithSeverity debugSeverity)

@@ -6,11 +6,14 @@ import           Control.Eff.Extend import           Control.Eff.Log import           Control.Eff.Lift-import           Control.Monad                  ( void )+import           Control.Monad                  ( void+                                                , when+                                                ) import           Test.Tasty import           Test.Tasty.HUnit import           Test.Tasty.Runners import           GHC.Stack+import           Control.Lens  setTravisTestOptions :: TestTree -> TestTree setTravisTestOptions =@@ -23,26 +26,20 @@   :: (e -> LogChannel LogMessage -> IO ())   -> (IO (e -> IO ()) -> TestTree)   -> TestTree-withTestLogC doSchedule k = withResource-  testLogC-  testLogJoin-  (\logCFactory -> k-    (return-      (\e -> do-        logC <- logCFactory-        doSchedule e logC-        -- doSchedule e nullLogChannel+withTestLogC doSchedule k = k+  (return+    (\e -> withAsyncLogChannel+      1000+      (multiMessageLogWriter+        (\writeWith ->+          writeWith+            (\m -> when (view lmSeverity m < noticeSeverity) (printLogMessage m)+            )+        )       )+      (doSchedule e)     )   )--testLogC :: IO (LogChannel LogMessage)-testLogC =-  -- filterLogChannel ((< informationalSeverity) . _lmSeverity) <$>-  forkLogger 1000 printLogMessage Nothing--testLogJoin :: LogChannel LogMessage -> IO ()-testLogJoin = joinLogChannel  untilShutdown :: Member t r => t (ResumeProcess v) -> Eff r () untilShutdown pa = do

@@ -3,9 +3,10 @@ import           Control.Exception import           Control.Concurrent import           Control.Concurrent.STM-import           Control.Eff.Loop import           Control.Eff.Extend import           Control.Eff.Lift+import           Control.Eff.Log+import           Control.Eff.Loop import           Control.Eff.Concurrent.Process import           Control.Eff.Concurrent.Process.ForkIOScheduler                                                as Scheduler@@ -14,6 +15,7 @@ import           Test.Tasty.HUnit import           Data.Dynamic import           Common+import           Data.Default  test_IOExceptionsIsolated :: TestTree test_IOExceptionsIsolated = setTravisTestOptions $ testGroup@@ -26,23 +28,30 @@         )       $ do           aVar <- newEmptyTMVarIO-          lc   <- testLogC-          Scheduler.defaultMainWithLogChannel lc $ do-            p1 <- spawn $ foreverCheap busyEffect-            lift (threadDelay 1000)-            void $ spawn $ do-              lift (threadDelay 1000)-              doExit-            lift (threadDelay 100000)-            wasStillRunningP1 <- sendShutdownChecked forkIoScheduler-                                                     p1-                                                     ExitNormally-            lift (atomically (putTMVar aVar wasStillRunningP1))+          withAsyncLogChannel+            1000+            def+            (Scheduler.defaultMainWithLogChannel+              (do+                p1 <- spawn $ foreverCheap busyEffect+                lift (threadDelay 1000)+                void $ spawn $ do+                  lift (threadDelay 1000)+                  doExit+                lift (threadDelay 100000)+                wasStillRunningP1 <- sendShutdownChecked forkIoScheduler+                                                         p1+                                                         ExitNormally+                lift (atomically (putTMVar aVar wasStillRunningP1))+              )+            )            wasStillRunningP1 <- atomically (takeTMVar aVar)           assertBool "the other process was still running" wasStillRunningP1   | (busyWith , busyEffect) <--    [ ("receiving", void (send (ReceiveMessage @SchedulerIO)))+    [ ( "receiving"+      , void (send (ReceiveSelectedMessage @SchedulerIO selectAnyMessageLazy))+      )     , ( "sending"       , void (send (SendMessage @SchedulerIO 44444 (toDyn "test message")))       )@@ -52,7 +61,14 @@     , ("selfpid-ing", void (send (SelfPid @SchedulerIO)))     , ( "spawn-ing"       , void-        (send (Spawn @SchedulerIO (void (send (ReceiveMessage @SchedulerIO)))))+        (send+          (Spawn @SchedulerIO+            (void+              (send (ReceiveSelectedMessage @SchedulerIO selectAnyMessageLazy)+              )+            )+          )+        )       )     ]   , (howToExit, doExit    ) <-@@ -67,11 +83,12 @@ test_mainProcessSpawnsAChildAndReturns = setTravisTestOptions   (testCase     "spawn a child and return"-    (do-      lc <- testLogC-      Scheduler.defaultMainWithLogChannel-        lc+    (withAsyncLogChannel+      1000+      def+      (Scheduler.defaultMainWithLogChannel         (void (spawn (void (receiveMessage forkIoScheduler))))+      )     )   ) @@ -79,15 +96,16 @@ test_mainProcessSpawnsAChildAndExitsNormally = setTravisTestOptions   (testCase     "spawn a child and exit normally"-    (do-      lc <- testLogC-      Scheduler.defaultMainWithLogChannel-        lc+    (withAsyncLogChannel+      1000+      def+      (Scheduler.defaultMainWithLogChannel         (do           void (spawn (void (receiveMessage forkIoScheduler)))           void (exitNormally forkIoScheduler)           fail "This should not happen!!"         )+      )     )   ) @@ -97,10 +115,10 @@   setTravisTestOptions     (testCase       "spawn a child with a busy send loop and exit normally"-      (do-        lc <- testLogC-        Scheduler.defaultMainWithLogChannel-          lc+      (withAsyncLogChannel+        1000+        def+        (Scheduler.defaultMainWithLogChannel           (do             void               (spawn@@ -111,6 +129,7 @@             void (exitNormally forkIoScheduler)             fail "This should not happen!!"           )+        )       )     ) @@ -119,15 +138,16 @@ test_mainProcessSpawnsAChildBothReturn = setTravisTestOptions   (testCase     "spawn a child and let it return and return"-    (do-      lc <- testLogC-      Scheduler.defaultMainWithLogChannel-        lc+    (withAsyncLogChannel+      1000+      def+      (Scheduler.defaultMainWithLogChannel         (do           child <- spawn (void (receiveMessageAs @String forkIoScheduler))           True  <- sendMessageChecked forkIoScheduler child (toDyn "test")           return ()         )+      )     )   ) @@ -135,10 +155,10 @@ test_mainProcessSpawnsAChildBothExitNormally = setTravisTestOptions   (testCase     "spawn a child and let it exit and exit"-    (do-      lc <- testLogC-      Scheduler.defaultMainWithLogChannel-        lc+    (withAsyncLogChannel+      1000+      def+      (Scheduler.defaultMainWithLogChannel         (do           child <- spawn             (do@@ -150,5 +170,6 @@           void (exitNormally forkIoScheduler)           error "This should not happen!!"         )+      )     )   )

@@ -8,11 +8,8 @@ import           Common import           Control.Concurrent.STM import           Control.DeepSeq-import           Control.Monad.IO.Class -demo-  :: (Member (Logs String) e, Member (Logs OtherLogMsg) e, MonadIO (Eff e))-  => Eff e ()+demo :: (HasLogWriterIO String e, HasLogWriterIO OtherLogMsg e) => Eff e () demo = do   logMsg "jo"   logMsg (OtherLogMsg "test 123")@@ -25,12 +22,15 @@       otherLogsQueue  <- newTQueueIO @OtherLogMsg       stringLogsQueue <- newTQueueIO @String       runLift-        (handleLogsWith-          (atomically . writeTQueue stringLogsQueue)-          (handleLogsWith (atomically . writeTQueue otherLogsQueue)-                          (interceptLogging reverseStringLogs demo)+        (handleLogs+          (multiMessageLogWriter ($ (atomically . writeTQueue stringLogsQueue)))+          (handleLogs+            (multiMessageLogWriter ($ (atomically . writeTQueue otherLogsQueue))+            )+            (mapLogMessages @String reverse  demo)           )         )+       otherLogs <- atomically (flushTQueue otherLogsQueue)       strLogs   <- atomically (flushTQueue stringLogsQueue)       assertEqual "string logs" ["oj"]                   strLogs@@ -38,7 +38,3 @@   ]  newtype OtherLogMsg = OtherLogMsg String deriving (Eq, NFData, Show)--reverseStringLogs-  :: ('[Logs String, Lift IO] <:: e, MonadIO (Eff e)) => String -> Eff e ()-reverseStringLogs = logMsg . reverse

@@ -21,27 +21,38 @@       in           setTravisTestOptions $ testGroup               "Loops without space leaks"-              [ testCase "scheduleMonadIOEff with many yields from replicateCheapM_" $ do-                  res <--                      Scheduler.scheduleIOWithLogging-                          ($! (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"+              [ 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 . (">>> " ++)))+                                    )+                                 $ do                                       me <- self SP                                       spawn_@@ -61,35 +72,43 @@       in           setTravisTestOptions $ testGroup               "Loops WITH space leaks"-              [ testCase "scheduleMonadIOEff with many yields from replicateM_" $ do-                    res <--                        Scheduler.scheduleIOWithLogging-                            ($! (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+              [ 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 $ sendMessageAs SP me ()-                                            )-                                        replicateM_-                                            soMany-                                            (void (receiveMessageAs @() SP))+                                      me <- self SP+                                      spawn_ (forever $ sendMessageAs SP me ())+                                      replicateM_+                                          soMany+                                          (void (receiveMessageAs @() SP))                          res @=? Right ()-            ]+              ]

@@ -2,9 +2,7 @@  import           Data.List                      ( sort ) import           Data.Foldable                  ( traverse_ )-import           Data.Dynamic                   ( fromDynamic-                                                , toDyn-                                                )+import           Data.Dynamic                   ( toDyn ) import           Data.Typeable import           Data.Default import           Control.Exception@@ -30,23 +28,20 @@ import           Debug.Trace  test_forkIo :: TestTree-test_forkIo = setTravisTestOptions-  (withTestLogC (\c lc -> handleLoggingAndIO_ (ForkIO.schedule c) lc)-                (\factory -> testGroup "ForkIOScheduler" [allTests factory])-  )+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 ->+test_singleThreaded = setTravisTestOptions $ withTestLogC+  (\e logC ->         -- void (runLift (logToChannel logC (SingleThreaded.schedule (return ()) e)))-      let runEff :: Eff '[Logs LogMessage, Lift IO] a -> IO a-          runEff = runLift . logToChannel logC-      in  void $ SingleThreaded.scheduleM runEff yield e-    )-    (\factory -> testGroup "SingleThreadedScheduler" [allTests factory])+    let runEff :: Eff '[Logs LogMessage, Lift IO] a -> IO a+        runEff = flip handleLoggingAndIO logC+    in  void $ SingleThreaded.scheduleM runEff yield e   )-+  (\factory -> testGroup "SingleThreadedScheduler" [allTests factory])  allTests   :: forall r@@ -67,7 +62,6 @@     ]   ) - data ReturnToSender   deriving Typeable @@ -142,14 +136,8 @@             go :: Int -> Eff (Process r ': r) ()             go 0 = sendMessageAs SP donePid True             go n = do-              void $ receiveMessageSuchThat-                SP-                (MessageSelector-                  (\m -> do-                    i <- fromDynamic m-                    if i == n then Just i else Nothing-                  )-                )+              void $ receiveSelectedMessage+                        SP (filterMessage (==n))               go (n - 1)            senderLoop receviverPid =@@ -517,7 +505,9 @@                   void $ atomically $ takeTMVar schedulerDoneVar           | e <- [ThreadKilled, UserInterrupt, HeapOverflow, StackOverflow]           , (busyWith, busyEffect) <--            [ ("receiving", void (send (ReceiveMessage @r)))+            [ ( "receiving"+              , void (send (ReceiveSelectedMessage @r (filterMessage (=="test message"))))+              )             , ( "sending"               , void (send (SendMessage @r 44444 (toDyn "test message")))               )@@ -526,7 +516,14 @@               )             , ("selfpid-ing", void (send (SelfPid @r)))             , ( "spawn-ing"-              , void (send (Spawn @r (void (send (ReceiveMessage @r)))))+              , void+                (send+                  (Spawn @r+                    (void+                      (send (ReceiveSelectedMessage @r selectAnyMessageLazy))+                    )+                  )+                )               )             , ("sleeping", lift (threadDelay 100000))             ]@@ -539,7 +536,9 @@                 void $ spawn $ foreverCheap busyEffect                 lift (threadDelay 10000)           | (busyWith, busyEffect) <--            [ ("receiving", void (send (ReceiveMessage @r)))+            [ ( "receiving"+              , void (send (ReceiveSelectedMessage @r (filterMessage (=="test message"))))+              )             , ( "sending"               , void (send (SendMessage @r 44444 (toDyn "test message")))               )@@ -548,7 +547,14 @@               )             , ("selfpid-ing", void (send (SelfPid @r)))             , ( "spawn-ing"-              , void (send (Spawn @r (void (send (ReceiveMessage @r)))))+              , void+                (send+                  (Spawn @r+                    (void+                      (send (ReceiveSelectedMessage @r selectAnyMessageLazy))+                    )+                  )+                )               )             ]           ]@@ -572,7 +578,9 @@                 assertEff "the other process was still running"                           wasStillRunningP1           | (busyWith , busyEffect) <--            [ ("receiving", void (send (ReceiveMessage @r)))+            [ ( "receiving"+              , void (send (ReceiveSelectedMessage @r (filterMessage (=="test message"))))+              )             , ( "sending"               , void (send (SendMessage @r 44444 (toDyn "test message")))               )@@ -581,7 +589,14 @@               )             , ("selfpid-ing", void (send (SelfPid @r)))             , ( "spawn-ing"-              , void (send (Spawn @r (void (send (ReceiveMessage @r)))))+              , void+                (send+                  (Spawn @r+                    (void+                      (send (ReceiveSelectedMessage @r selectAnyMessageLazy))+                    )+                  )+                )               )             ]           , (howToExit, doExit    ) <-@@ -604,10 +619,12 @@    . (Member (Logs LogMessage) r, SetMember Lift (Lift IO) r)   => IO (Eff (Process r ': r) () -> IO ())   -> TestTree-sendShutdownTests schedulerFactory =-  let px :: SchedulerProxy r+sendShutdownTests schedulerFactory+  = let+      px :: SchedulerProxy r       px = SchedulerProxy-  in  testGroup+    in+      testGroup         "sendShutdown"         [ testCase "... self" $ applySchedulerFactory schedulerFactory $ do           me <- self px@@ -618,7 +635,8 @@         $ \assertEff -> do             me <- self px             r  <- send (SendShutdown @r me ExitNormally)-            traceShowM (show me ++": returned from SendShutdow to self " ++ show r)+            traceShowM+              (show me ++ ": returned from SendShutdow to self " ++ show r)             assertEff               "ShutdownRequested must be returned"               (case r of@@ -645,7 +663,8 @@               me    <- self px               other <- spawn                 (do-                  untilShutdown (ReceiveMessage @r)+                  untilShutdown+                    (ReceiveSelectedMessage @r selectAnyMessageLazy)                   void (sendMessage px me (toDyn "OK"))                 )               void (sendShutdown px other ExitNormally)