extensible-effects-concurrent 0.13.2 → 0.14.0

raw patch · 4 files changed

+164/−36 lines, 4 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• ChangeLog.md +4/−0
• examples/example-2/Main.hs +71/−0
• extensible-effects-concurrent.cabal +6/−1
• src/Control/Eff/Concurrent/Api/Server2.hs +83/−35

@@ -1,5 +1,9 @@ # Changelog for extensible-effects-concurrent +## 0.14.0++- Fix/Improve Server2+ ## 0.13.2  - Add `ProcessFinished`

@@ -3,12 +3,16 @@  import           Data.Dynamic import           Control.Eff+import           Control.Eff.Lift import           Control.Eff.Concurrent import           Control.Eff.State.Strict import           Control.Monad+import           Control.Concurrent  main :: IO () main = do+  defaultMain (void (altCounterExample SchedulerProxy))+   defaultMain (void (counterExample SchedulerProxy))   print (schedulePure (counterExample SchedulerProxy)) @@ -21,7 +25,73 @@   Cnt :: Api Counter ('Synchronous Integer)   ObserveCounter :: SomeObserver Counter -> Api Counter 'Asynchronous   UnobserveCounter :: SomeObserver Counter -> Api Counter 'Asynchronous+  deriving Typeable +altCounterExample+  :: (Member (Logs LogMessage) q, Lifted IO q)+  => SchedulerProxy q+  -> Eff (InterruptableProcess q) ()+altCounterExample px = do+  (c, cp) <- altCounter+  lift (threadDelay 500000)+  o <- logCounterObservations px+  lift (threadDelay 500000)+  registerObserver px o c+  lift (threadDelay 500000)+  cast px c Inc+  lift (threadDelay 500000)+  sendMessage px cp "test 123"+  cast px c Inc+  lift (threadDelay 500000)+  cast px c Inc+  sendMessage px cp (12312312 :: Int)+  lift (threadDelay 500000)+  cast px c Inc+  lift (threadDelay 500000)+  cast px c Inc+  lift (threadDelay 500000)+  cast px c Inc+  lift (threadDelay 500000)+  r <- call px c Cnt+  lift (threadDelay 500000)+  lift (putStrLn ("r: " ++ show r))+  lift (threadDelay 500000)+  lift (threadDelay 500000)+++altCounter+  :: (Member (Logs LogMessage) q)+  => Eff (InterruptableProcess q) (Server Counter, ProcessId)+altCounter = spawnApiServerEffectful+  (manageObservers @Counter . evalState (0 :: Integer))+  (  handleCalls+      SP+      (\case+        Cnt ->+          ($ do+            val <- get+            return (Just val, AwaitNext)+          )+      )+  <> handleCasts+       (\case+         Inc -> do+           val <- get @Integer+           let val' = val + 1+           notifyObservers SP (CountChanged val')+           put val'+           return AwaitNext+         ObserveCounter s -> do+           addObserver s+           return AwaitNext+         UnobserveCounter s -> do+           removeObserver s+           return AwaitNext+       )+  ^: logUnhandledMessages+  )+  stopServerOnInterrupt+ deriving instance Show (Api Counter x)  instance Observable Counter where@@ -46,6 +116,7 @@     logInfo (show me ++ " observed on: " ++ show fromSvr ++ ": " ++ show msg)     return HandleNextRequest   )+  counterHandler   :: forall r q

@@ -1,5 +1,5 @@ name:           extensible-effects-concurrent-version:        0.13.2+version:        0.14.0 description:    Please see the README on GitHub at <https://github.com/sheyll/extensible-effects-concurrent#readme> synopsis:       Message passing concurrency as extensible-effect homepage:       https://github.com/sheyll/extensible-effects-concurrent#readme@@ -122,6 +122,7 @@                       , FunctionalDependencies                       , GADTs                       , GeneralizedNewtypeDeriving+                      , LambdaCase                       , RankNTypes                       , ScopedTypeVariables                       , StandaloneDeriving@@ -148,6 +149,7 @@                       , FunctionalDependencies                       , GADTs                       , GeneralizedNewtypeDeriving+                      , LambdaCase                       , RankNTypes                       , ScopedTypeVariables                       , StandaloneDeriving@@ -176,6 +178,7 @@                       , FunctionalDependencies                       , GADTs                       , GeneralizedNewtypeDeriving+                      , LambdaCase                       , RankNTypes                       , ScopedTypeVariables                       , StandaloneDeriving@@ -201,6 +204,7 @@                       , FunctionalDependencies                       , GADTs                       , GeneralizedNewtypeDeriving+                      , LambdaCase                       , RankNTypes                       , ScopedTypeVariables                       , StandaloneDeriving@@ -246,6 +250,7 @@                       , FunctionalDependencies                       , GADTs                       , GeneralizedNewtypeDeriving+                      , LambdaCase                       , RankNTypes                       , ScopedTypeVariables                       , StandaloneDeriving

@@ -38,15 +38,16 @@ import           Control.Eff.Log import           Control.Eff.State.Lazy import           Control.Eff.Concurrent.Api+import           Control.Eff.Concurrent.Api.Internal import           Control.Eff.Concurrent.Process import           Control.Monad                  ( (>=>) ) import           Data.Proxy import           Data.Dynamic import           Control.Applicative import           GHC.Stack-import           GHC.Generics import           Control.DeepSeq import           Data.Kind+import           Data.Foldable import           Data.Default  -- | /Server/ an 'Api' in a newly spawned process.@@ -71,9 +72,9 @@   handleCallbackResult     :: Either CallbackResult (Either InterruptReason CallbackResult)     -> Eff (ConsProcess eff) (Maybe ())-  handleCallbackResult (Left HandleNext) = return Nothing+  handleCallbackResult (Left AwaitNext) = return Nothing   handleCallbackResult (Left (StopServer r)) = exitBecause SP (NotRecovered r)-  handleCallbackResult (Right (Right HandleNext)) = return Nothing+  handleCallbackResult (Right (Right AwaitNext)) = return Nothing   handleCallbackResult (Right (Right (StopServer r))) =     intCb r >>= handleCallbackResult . Left   handleCallbackResult (Right (Left r)) =@@ -85,13 +86,13 @@ -- @since 0.13.2 spawnApiServerStateful   :: forall api eff state-   . (HasCallStack)+   . (HasCallStack, ToServerPids api)   => Eff (InterruptableProcess eff) state   -> MessageCallback api (State state ': InterruptableProcess eff)   -> InterruptCallback (State state ': ConsProcess eff)-  -> Eff (InterruptableProcess eff) (Server api)+  -> Eff (InterruptableProcess eff) (ServerPids api) spawnApiServerStateful initEffect (MessageCallback sel cb) (InterruptCallback intCb)-  = fmap asServer $ spawnRaw $ do+  = fmap (toServerPids (Proxy @api)) $ spawnRaw $ do     state <- provideInterruptsShutdown initEffect     evalState state $ receiveSelectedLoop (SP @eff) sel $ \msg -> case msg of       Left  m -> invokeIntCb m@@ -101,12 +102,12 @@         case r of           Left  i              -> invokeIntCb i           Right (StopServer i) -> invokeIntCb i-          Right HandleNext     -> return Nothing+          Right AwaitNext      -> return Nothing  where   invokeIntCb j = do     l <- intCb j     case l of-      HandleNext                 -> return Nothing+      AwaitNext                  -> return Nothing       StopServer ProcessFinished -> return (Just ())       StopServer k               -> exitBecause SP (NotRecovered k) @@ -117,15 +118,16 @@ spawnApiServerEffectful   :: forall api eff serverEff    . ( HasCallStack+     , ToServerPids api      , Member Interrupts serverEff      , SetMember Process (Process eff) serverEff      )   => (forall b . Eff serverEff b -> Eff (InterruptableProcess eff) b)   -> MessageCallback api serverEff   -> InterruptCallback serverEff-  -> Eff (InterruptableProcess eff) (Server api)+  -> Eff (InterruptableProcess eff) (ServerPids api) spawnApiServerEffectful handleServerInteralEffects scb (InterruptCallback intCb)-  = asServer <$> (spawn (handleServerInteralEffects (go scb)))+  = toServerPids (Proxy @api) <$> spawn (handleServerInteralEffects (go scb))  where   go (MessageCallback sel cb) = receiveSelectedLoop     (SP @eff)@@ -137,10 +139,10 @@     handleCallbackResult       :: Either CallbackResult (Either InterruptReason CallbackResult)       -> Eff serverEff (Maybe ())-    handleCallbackResult (Left HandleNext) = return Nothing+    handleCallbackResult (Left AwaitNext) = return Nothing     handleCallbackResult (Left (StopServer r)) =       exitBecause SP (NotRecovered r)-    handleCallbackResult (Right (Right HandleNext)) = return Nothing+    handleCallbackResult (Right (Right AwaitNext)) = return Nothing     handleCallbackResult (Right (Right (StopServer r))) =       intCb r >>= handleCallbackResult . Left     handleCallbackResult (Right (Left r)) =@@ -153,15 +155,14 @@ -- @since 0.13.2 data CallbackResult where   -- | Tell the server to keep the server loop running-  HandleNext :: CallbackResult+  AwaitNext :: CallbackResult   -- | Tell the server to exit, this will make 'serve' stop handling requests without   -- exitting the process. '_terminateCallback' will be invoked with the given   -- optional reason.   StopServer :: InterruptReason -> CallbackResult   --  SendReply :: reply -> CallbackResult () -> CallbackResult (reply -> Eff eff ())-  deriving (Show, Typeable, Generic)+  deriving ( Typeable ) -instance NFData CallbackResult  -- | An existential wrapper around  a 'MessageSelector' and a function that -- handles the selected message. The @api@ type parameter is a phantom type.@@ -178,7 +179,7 @@  instance Monoid (MessageCallback api eff) where   mappend = (<>)-  mempty = MessageCallback selectAnyMessageLazy (const (pure HandleNext))+  mempty = MessageCallback selectAnyMessageLazy (const (pure AwaitNext))  instance Default (MessageCallback api eff) where   def = mempty@@ -219,42 +220,89 @@ -- @since 0.13.2 handleCasts   :: forall api eff-   . ( HasCallStack-     , NFData (Api api 'Asynchronous)-     , Typeable (Api api 'Asynchronous)-     )+   . (HasCallStack, Typeable api, Typeable (Api api 'Asynchronous))   => (Api api 'Asynchronous -> Eff eff CallbackResult)   -> MessageCallback api eff-handleCasts = MessageCallback selectMessage+handleCasts h = MessageCallback+  (selectMessageWithLazy+    (\case+      cr@(Cast _ :: Request api) -> Just cr+      _callReq                   -> Nothing+    )+  )+  (\(Cast req :: Request api) -> h req)  -- | A smart constructor for 'MessageCallback's --+-- > handleCalls SP+-- >   (\ (RentBook bookId customerId) runCall ->+-- >      runCall $ do+-- >          rentalIdE <- rentBook bookId customerId+-- >          case rentalIdE of+-- >            -- on fail we just don't send a reply, let the caller run into+-- >            -- timeout+-- >            Left err -> return (Nothing, AwaitNext)+-- >            Right rentalId -> return (Just rentalId, AwaitNext))+-- -- @since 0.13.2 handleCalls-  :: forall api eff reply+  :: forall api eff effScheduler    . ( HasCallStack-     , NFData (Api api ( 'Synchronous reply))-     , Typeable (Api api ( 'Synchronous reply))+     , Typeable api+     , SetMember Process (Process effScheduler) eff+     , Member Interrupts eff      )-  => (Api api ( 'Synchronous reply) -> Eff eff CallbackResult)+  => SchedulerProxy effScheduler+  -> (  forall secret reply+      . (Typeable reply, Typeable (Api api ( 'Synchronous reply)))+     => Api api ( 'Synchronous reply)+     -> (Eff eff (Maybe reply, CallbackResult) -> secret)+     -> secret+     )   -> MessageCallback api eff-handleCalls = MessageCallback selectMessage+handleCalls px h = MessageCallback+  (selectMessageWithLazy+    (\case+      (Cast _ :: Request api) -> Nothing+      callReq                 -> Just callReq+    )+  )+  (\(Call callRef fromPid req :: Request api) -> h+    req+    (\resAction -> do+      (mReply, cbResult) <- resAction+      traverse_ (sendReply fromPid callRef) mReply+      return cbResult+    )+  )+ where+  sendReply :: (Typeable reply) => ProcessId -> Int -> reply -> Eff eff ()+  sendReply fromPid callRef reply =+    sendMessage px fromPid (Response (Proxy @api) callRef $! reply) + -- | A smart constructor for 'MessageCallback's -- -- @since 0.13.2 handleCastsAndCalls-  :: forall api eff reply+  :: forall api eff effScheduler    . ( HasCallStack-     , NFData (Api api ( 'Synchronous reply))-     , Typeable (Api api ( 'Synchronous reply))-     , NFData (Api api 'Asynchronous)+     , Typeable api      , Typeable (Api api 'Asynchronous)+     , SetMember Process (Process effScheduler) eff+     , Member Interrupts eff      )-  => (Api api 'Asynchronous -> Eff eff CallbackResult)-  -> (Api api ( 'Synchronous reply) -> Eff eff CallbackResult)+  => SchedulerProxy effScheduler+  -> (Api api 'Asynchronous -> Eff eff CallbackResult)+  -> (  forall secret reply+      . (Typeable reply, Typeable (Api api ( 'Synchronous reply)))+     => Api api ( 'Synchronous reply)+     -> (Eff eff (Maybe reply, CallbackResult) -> secret)+     -> secret+     )   -> MessageCallback api eff-handleCastsAndCalls onCast onCall = handleCalls onCall <> handleCasts onCast+handleCastsAndCalls px onCast onCall =+  handleCalls px onCall <> handleCasts onCast  -- | A smart constructor for 'MessageCallback's --@@ -298,7 +346,7 @@ -- @since 0.13.2 dropUnhandledMessages :: forall eff . HasCallStack => MessageCallback '[] eff dropUnhandledMessages =-  MessageCallback selectAnyMessageLazy (const (return HandleNext))+  MessageCallback selectAnyMessageLazy (const (return AwaitNext))  -- | A 'fallbackHandler' that terminates if there are unhandled messages. --@@ -316,7 +364,7 @@   => MessageCallback '[] eff logUnhandledMessages = MessageCallback selectAnyMessageLazy $ \msg -> do   logWarning ("ignoring unhandled message " ++ show msg)-  return HandleNext+  return AwaitNext  -- | Helper type class for the return values of 'spawnApiServer' et al. --