extensible-effects-concurrent 0.25.1 → 0.26.0

raw patch · 13 files changed

+145/−90 lines, 13 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• ChangeLog.md +4/−0
• examples/example-1/Main.hs +6/−6
• examples/example-2/Main.hs +5/−7
• extensible-effects-concurrent.cabal +1/−1
• src/Control/Eff/Concurrent.hs +19/−0
• src/Control/Eff/Concurrent/Protocol/Client.hs +1/−1
• src/Control/Eff/Concurrent/Protocol/EffectfulServer.hs +6/−15
• src/Control/Eff/Concurrent/Protocol/Request.hs +86/−34
• src/Control/Eff/Concurrent/Protocol/StatefulServer.hs +0/−7
• src/Control/Eff/Concurrent/Protocol/Supervisor.hs +7/−7
• test/GenServerTests.hs +5/−8
• test/ProcessBehaviourTestCases.hs +3/−2
• test/SupervisorTests.hs +2/−2

@@ -1,5 +1,9 @@ # Changelog for extensible-effects-concurrent +## 0.26.0+- Introduce `ReplyTarget` +- Change the `sendReply` signature to accept a `ReplyTarget`+ ## 0.25.1 - Add `castSingleton` and `callSingleton`, which use the `EndpointReader` and `EmbedProtocol` type class. - Change `toObserver` to accept an Endpoint of a protocol that embeds `Observer x`

@@ -78,18 +78,18 @@ testServerLoop = start (genServer (const id) handleReq "test-server-1")  where   handleReq :: GenServerId TestProtocol -> Event TestProtocol -> Eff Effects ()-  handleReq _myId (OnCall ser orig cm) =+  handleReq _myId (OnCall rt cm) =     case cm of       Terminate -> do         me <- self         logInfo (T.pack (show me ++ " exiting"))-        sendReply ser orig ()+        sendReply rt ()         interrupt NormalExitRequested        TerminateError e -> do         me <- self         logInfo (T.pack (show me ++ " exiting with error: " ++ e))-        sendReply ser orig ()+        sendReply rt ()         interrupt (ErrorInterrupt e)        SayHello mx ->@@ -108,18 +108,18 @@             me <- self             logInfo (T.pack (show me ++ " casting to self"))             cast (asEndpoint @TestProtocol me) (Shout "from me")-            sendReply ser orig False+            sendReply rt False            "stop" -> do             me <- self             logInfo (T.pack (show me ++ " stopping me"))-            sendReply ser orig False+            sendReply rt False             interrupt (ErrorInterrupt "test error")            x -> do             me <- self             logInfo (T.pack (show me ++ " Got Hello: " ++ x))-            sendReply ser orig (length x > 3)+            sendReply rt (length x > 3)    handleReq _myId (OnCast (Shout x)) = do     me <- self

@@ -93,9 +93,7 @@   type instance Model SupiCounter =     ( Integer     , Observers CounterChanged-    , Maybe ( Serializer (Reply SupiCounter (Maybe ()))-            , RequestOrigin SupiCounter (Maybe ())-            )+    , Maybe (ReplyTarget SupiCounter (Maybe ()))     )    data instance StartArgument SupiCounter (Processes q) = MkEmptySupiCounter@@ -103,13 +101,13 @@   setup _ = return ((0, emptyObservers, Nothing), ())    update _ = \case-    OnCall ser orig callReq ->+    OnCall rt callReq ->       case callReq of         ToPdu1 Cnt ->-          sendReply ser orig =<< useModel @SupiCounter _1+          sendReply rt =<< useModel @SupiCounter _1         ToPdu2 _ -> error "unreachable"         ToPdu3 (Whoopediedoo c) ->-          modifyModel @SupiCounter (_3 .~ if c then Just (ser, orig) else Nothing)+          modifyModel @SupiCounter (_3 .~ if c then Just rt else Nothing)      OnCast castReq ->       case castReq of@@ -118,7 +116,7 @@           zoomModel @SupiCounter _2 (observed (CounterChanged val'))           when (val' > 5) $             getAndModifyModel @SupiCounter (_3 .~ Nothing)-            >>= traverse_ (\(ser, orig) -> sendReply ser orig (Just ())) . view _3+            >>= traverse_ (\rt' -> sendReply rt' (Just ())) . view _3         ToPdu2 x ->           zoomModel @SupiCounter _2 (handleObserverRegistration x)         ToPdu3 _ -> error "unreachable"

@@ -1,6 +1,6 @@ cabal-version:  2.0 name:           extensible-effects-concurrent-version:        0.25.1+version:        0.26.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

@@ -31,6 +31,9 @@     -- ** /Client/ Functions for Consuming APIs     module Control.Eff.Concurrent.Protocol.Client   ,+    -- ** /Protocol-Server/ Support Functions for building protocol servers+    module Control.Eff.Concurrent.Protocol.Request+  ,     -- ** /Observer/ Functions for Events and Event Listener     module Control.Eff.Concurrent.Protocol.Observer   ,@@ -230,6 +233,22 @@                                                 , flushObservationQueue                                                 , withObservationQueue                                                 , spawnLinkObservationQueueWriter+                                                )+import           Control.Eff.Concurrent.Protocol.Request+                                                ( Request(..)+                                                , sendReply+                                                , ReplyTarget(..)+                                                , replyTarget+                                                , embeddedReplyTarget+                                                , replyTargetOrigin+                                                , replyTargetSerializer+                                                , toEmbeddedReplyTarget+                                                , RequestOrigin(..)+                                                , embedRequestOrigin+                                                , toEmbeddedOrigin+                                                , Reply(..)+                                                , embedReplySerializer+                                                , makeRequestOrigin                                                 ) import           Control.Eff.Concurrent.Process.ForkIOScheduler                                                 ( schedule

@@ -205,7 +205,7 @@ -- This function makes use of AmbigousTypes and TypeApplications. -- -- When not working with an embedded 'Pdu' use 'callEndpointReader'.---+--  -- @since 0.25.1 callSingleton   :: forall outer inner reply q e

@@ -12,13 +12,6 @@   , GenServer   , GenServerId(..)   , genServer-  -- * Re-exports-  , RequestOrigin(..)-  , Reply(..)-  , sendReply-  , sendEmbeddedReply-  , toEmbeddedOrigin-  , embedReplySerializer   )   where @@ -150,7 +143,7 @@       <|> OnMessage <$> selectAnyMessage       where         onRequest :: Request (ServerPdu a) -> Event (ServerPdu a)-        onRequest (Call o m) = OnCall (MkSerializer toStrictDynamic) o m+        onRequest (Call o m) = OnCall (replyTarget (MkSerializer toStrictDynamic) o) m         onRequest (Cast m) = OnCast m     handleInt i = onEvent a (OnInterrupt i) *> pure Nothing     mainLoop :: (Typeable a)@@ -159,18 +152,16 @@     mainLoop (Left i) = handleInt i     mainLoop (Right i) = onEvent a i *> pure Nothing --- | Internal protocol to communicate incoming messages and other events to the+-- | This event sum-type is used to communicate incoming messages and other events to the -- instances of 'Server'. ----- Note that this is required to receive any kind of messages in 'protocolServerLoop'.--- -- @since 0.24.0 data Event a where   -- | A 'Synchronous' message was received. If an implementation wants to delegate nested 'Pdu's, it can-  -- 'contramap' the reply 'Serializer' such that the 'Reply' received by the caller has the correct type.+  -- use 'toEmbeddedReplyTarget' to convert a 'ReplyTarget' safely to the embedded protocol.   --   -- @since 0.24.1-  OnCall :: forall a r. (Tangible r, TangiblePdu a ('Synchronous r)) => Serializer (Reply a r) -> RequestOrigin a r -> Pdu a ('Synchronous r) -> Event a+  OnCall :: forall a r. (Tangible r, TangiblePdu a ('Synchronous r)) => ReplyTarget a r -> Pdu a ('Synchronous r) -> Event a   OnCast :: forall a. TangiblePdu a 'Asynchronous => Pdu a 'Asynchronous -> Event a   OnInterrupt  :: (Interrupt 'Recoverable) -> Event a   OnDown  :: ProcessDown -> Event a@@ -183,7 +174,7 @@     showParen (d>=10) $       showString "event: "       . case e of-          OnCall _ o p -> shows (Call o p)+          OnCall o p -> shows (Call (view replyTargetOrigin o) p)           OnCast p -> shows (Cast p)           OnInterrupt r -> shows r           OnDown r -> shows r@@ -192,7 +183,7 @@  instance NFData a => NFData (Event a) where    rnf = \case-       OnCall _ o p -> rnf o `seq` rnf p+       OnCall o p -> rnf o `seq` rnf p        OnCast p -> rnf p        OnInterrupt r -> rnf r        OnDown r  -> rnf r

@@ -4,7 +4,12 @@ module Control.Eff.Concurrent.Protocol.Request   ( Request(..)   , sendReply-  , sendEmbeddedReply+  , ReplyTarget(..)+  , replyTarget+  , replyTargetOrigin+  , replyTargetSerializer+  , embeddedReplyTarget+  , toEmbeddedReplyTarget   , RequestOrigin(..)   , embedRequestOrigin   , toEmbeddedOrigin@@ -18,10 +23,10 @@ import Control.Eff import Control.Eff.Concurrent.Process import Control.Eff.Concurrent.Protocol-import Data.Functor.Contravariant+import Control.Lens import Data.Kind (Type) import Data.Typeable (Typeable)-import Data.Tagged+import Data.Semigroup import GHC.Generics  -- | A wrapper sum type for calls and casts for the 'Pdu's of a protocol@@ -95,37 +100,6 @@  instance NFData (RequestOrigin p r) --- | Send a 'Reply' to a 'Call'.------ The reply will be deeply evaluated to 'rnf'.------ To send replies for 'EmbedProtocol' instances use 'sendEmbeddedReply',--- or 'embedReplySerializer' and 'toEmbeddedOrigin'.------ @since 0.15.0-sendReply ::-  forall protocol reply q eff .-     (SetMember Process (Process q) eff, Member Interrupts eff, Tangible reply, Typeable protocol)-  => Serializer (Reply protocol reply) -> RequestOrigin protocol reply -> reply -> Eff eff ()-sendReply ser o r = sendAnyMessage (_requestOriginPid o) $! runSerializer ser $! Reply o r---- | Send a 'Reply' to a 'Call' for an embedded 'Pdu'.------ The reply will be deeply evaluated to 'rnf'.------ @since 0.25.1-sendEmbeddedReply ::-  forall outer inner reply q eff .-     ( SetMember Process (Process q) eff-     , Member Interrupts eff-     , TangiblePdu outer ('Synchronous reply)-     , TangiblePdu inner ('Synchronous reply)-     , Tangible reply-     , EmbedProtocol outer inner-     )-  => Serializer (Reply outer reply) -> RequestOrigin outer reply -> Tagged inner reply -> Eff eff ()-sendEmbeddedReply ser o = sendReply @outer @reply (embedReplySerializer ser) (toEmbeddedOrigin o) . unTagged- -- | Turn an 'RequestOrigin' to an origin for an embedded request (See 'EmbedProtocol'). -- -- This is useful of a server delegates the @calls@ and @casts@ for an embedded protocol@@ -171,3 +145,81 @@ -- @since 0.24.2 embedReply :: forall outer inner reply . EmbedProtocol outer inner => Reply inner reply -> Reply outer reply embedReply (Reply (RequestOrigin !pid !ref) !v) = Reply (RequestOrigin pid ref) v+++-- | Answer a 'Call' by sending the reply value to the client process.+--+-- The 'ProcessId', the 'RequestOrigin' and the 'Reply' 'Serializer' are+-- stored in the 'ReplyTarget'.+--+-- @since 0.25.1+sendReply+  :: ( SetMember Process (Process q) eff+     , Member Interrupts eff+     , Tangible reply+     , Typeable protocol+     )+  => ReplyTarget protocol reply+  -> reply+  -> Eff eff ()+sendReply (MkReplyTarget (Arg o ser)) r =+  sendAnyMessage (_requestOriginPid o) $! runSerializer ser $! Reply o r++-- | Target of a 'Call' reply.+--+-- This combines a 'RequestOrigin' with a 'Serializer' for a 'Reply' using 'Arg'.+-- There are to smart constructors for this type: 'replyTarget' and 'embeddedReplyTarget'.+--+-- Because of 'Arg' the 'Eq' and 'Ord' instances are implemented via+-- the 'RequestOrigin' instances.+--+-- @since 0.26.0+newtype ReplyTarget p r =+  MkReplyTarget (Arg (RequestOrigin p r) (Serializer (Reply p r)))+    deriving (Eq, Ord, Typeable)++instance Show (ReplyTarget p r) where+  showsPrec d (MkReplyTarget (Arg o _s)) = showParen (d>=10) (showString "reply-target: " . shows o)++instance NFData (ReplyTarget p r) where+  rnf (MkReplyTarget (Arg x y)) = rnf x `seq` y `seq` ()++-- | Smart constructor for a 'ReplyTarget'.+--+-- To build a @ReplyTarget@ for an 'EmbedProtocol' instance use 'embeddedReplyTarget'.+--+-- @since 0.26.0+replyTarget :: Serializer (Reply p reply) -> RequestOrigin p reply -> ReplyTarget p reply+replyTarget ser orig =  MkReplyTarget (Arg orig ser)++-- | A simple 'Lens' for the 'RequestOrigin' of a 'ReplyTarget'.+--+-- @since 0.26.0+replyTargetOrigin :: Lens' (ReplyTarget p reply) (RequestOrigin p reply)+replyTargetOrigin f (MkReplyTarget (Arg o x)) =+  (\o' -> MkReplyTarget (Arg o' x)) <$> f o++-- | A simple 'Lens' for the 'Reply' 'Serializer' of a 'ReplyTarget'.+--+-- @since 0.26.0+replyTargetSerializer :: Lens' (ReplyTarget p reply) (Serializer (Reply p reply))+replyTargetSerializer f (MkReplyTarget (Arg x o)) =+  (\o' -> MkReplyTarget (Arg x o')) <$> f o++-- | Smart constructor for an /embedded/ 'ReplyTarget'.+--+-- This combines 'replyTarget' and 'toEmbeddedReplyTarget'.+--+-- @since 0.26.0+embeddedReplyTarget :: EmbedProtocol outer inner => Serializer (Reply outer reply) -> RequestOrigin outer reply -> ReplyTarget inner reply+embeddedReplyTarget ser orig = toEmbeddedReplyTarget $ replyTarget ser orig++-- | Convert a 'ReplyTarget' to be usable for /embedded/ replies.+--+-- This combines a 'toEmbeddedOrigin' with 'embedReplySerializer' to produce a+-- 'ReplyTarget' that can be passed to functions defined soley on an embedded protocol.+--+-- @since 0.26.0+toEmbeddedReplyTarget :: EmbedProtocol outer inner => ReplyTarget outer r -> ReplyTarget inner r+toEmbeddedReplyTarget (MkReplyTarget (Arg orig ser)) =+  MkReplyTarget (Arg (toEmbeddedOrigin orig) (embedReplySerializer ser))

@@ -19,12 +19,6 @@   , viewSettings   -- * Re-exports   , Effectful.Event(..)-  , RequestOrigin(..)-  , Reply(..)-  , sendReply-  , sendEmbeddedReply-  , toEmbeddedOrigin-  , embedReplySerializer   )   where @@ -33,7 +27,6 @@ import Control.Eff.Concurrent.Process import Control.Eff.Concurrent.Protocol import qualified Control.Eff.Concurrent.Protocol.EffectfulServer as Effectful-import Control.Eff.Concurrent.Protocol.Request import Control.Eff.Log import Control.Eff.Reader.Strict import Control.Eff.State.Strict

@@ -169,23 +169,23 @@   type Model (Sup p) = Children (ChildId p) p    setup _cfg = pure (def, ())-  update supConfig (OnCall ser orig req) =+  update supConfig (OnCall rt req) =     case req of       GetDiagnosticInfo ->  do         p <- (pack . show <$> getChildren @(ChildId p) @p)-        sendReply ser orig p+        sendReply rt p        LookupC i -> do         p <- fmap _childEndpoint <$> lookupChildById @(ChildId p) @p i-        sendReply ser orig p+        sendReply rt p        StopC i t -> do         mExisting <- lookupAndRemoveChildById @(ChildId p) @p i         case mExisting of-          Nothing -> sendReply ser orig False+          Nothing -> sendReply rt False           Just existingChild -> do             stopOrKillChild i existingChild t-            sendReply ser orig True+            sendReply rt True        StartC i -> do         childEp <- raise (raise (Server.start (supConfigStartFun supConfig i)))@@ -195,9 +195,9 @@         case mExisting of           Nothing -> do             putChild i (MkChild @p childEp cMon)-            sendReply ser orig (Right childEp)+            sendReply rt (Right childEp)           Just existingChild ->-            sendReply ser orig (Left (AlreadyStarted i (existingChild ^. childEndpoint)))+            sendReply rt (Left (AlreadyStarted i (existingChild ^. childEndpoint)))    update _supConfig (OnDown (ProcessDown mrChild reason)) = do       oldEntry <- lookupAndRemoveChildByMonitor @(ChildId p) @p mrChild

@@ -10,10 +10,7 @@ import Control.Eff.Concurrent.Protocol.Supervisor as Sup import qualified Control.Eff.Concurrent.Protocol.EffectfulServer as E import qualified Control.Eff.Concurrent.Protocol.StatefulServer as S-import Control.Eff.Concurrent.Protocol.Request import Control.Lens-import Data.Coerce-import Data.Functor.Contravariant ((>$<)) import Data.Text as T import Data.Type.Pretty import Data.Typeable (Typeable)@@ -48,8 +45,8 @@   type Model Small = String   update MkSmall x =     case x of-      E.OnCall ser o (SmallCall f) ->-       sendReply ser o f+      E.OnCall rt (SmallCall f) ->+       sendReply rt f       E.OnCast msg ->        logInfo' (show msg)       other ->@@ -92,12 +89,12 @@   data instance StartArgument Big (Processes e) = MkBig   type Model Big = String   update MkBig = \case-    E.OnCall ser orig req ->+    E.OnCall rt req ->           case req of             BigCall o -> do               logNotice ("BigCall " <> pack (show o))-              sendReply ser orig o-            BigSmall x -> S.update MkSmall (S.OnCall (coerce >$< ser) (coerce orig) x)+              sendReply rt o+            BigSmall x -> S.update MkSmall (S.OnCall (toEmbeddedReplyTarget rt) x)     E.OnCast req ->         case req of           BigCast o -> S.putModel @Big o

@@ -7,6 +7,7 @@ import           Control.Eff.Concurrent.Process import           Control.Eff.Concurrent.Process.Timer import           Control.Eff.Concurrent.Protocol+import           Control.Eff.Concurrent.Protocol.Request import           Control.Eff.Concurrent.Protocol.Client import           Control.Eff.Concurrent.Protocol.EffectfulServer import qualified Control.Eff.Concurrent.Process.ForkIOScheduler@@ -131,13 +132,13 @@     (const id)     (\_me evt ->       case evt of-        OnCall ser orig msg ->+        OnCall rt msg ->           case msg of             StopReturnToSender -> interrupt testInterruptReason             ReturnToSender fromP echoMsg -> do               sendMessage fromP echoMsg               yieldProcess-              sendReply ser orig True+              sendReply rt True         OnInterrupt i ->           interrupt i         other -> interrupt (ErrorInterrupt (show other))

@@ -271,9 +271,9 @@       OnCast (TestInterruptWith i) -> do         logInfo (pack (show tId) <> ": stopping with: " <> pack (show i))         interrupt i-      OnCall ser orig (TestGetStringLength str) -> do+      OnCall rt (TestGetStringLength str) -> do         logInfo (pack (show tId) <> ": calculating length of: " <> pack str)-        sendReply ser orig (length str)+        sendReply rt (length str)       OnInterrupt x -> do         logNotice (pack (show tId) <> ": " <> pack (show x))         if testMode == IgnoreNormalExitRequest