-- | Observer Effects
--
-- This module supports the implementation of observerRegistry and observables. Expected use
-- case is event propagation.
--
-- The observable event sources and the observers are usually server processes for a
-- protocol that embeds the 'ObserverRegistry' and 'Observer' 'Pdu's respectively.
--
-- A generic FIFO queue based observer can be found in "Control.Eff.Concurrent.Protocol.Observer.Queue".
--
-- @since 0.16.0
module Control.Eff.Concurrent.Protocol.Observer
( Observer (..),
ObservationSink (),
IsObservable,
CanObserve,
Pdu (RegisterObserver, ForgetObserver, Observed),
registerObserver,
forgetObserver,
forgetObserverUnsafe,
ObserverRegistry (..),
ObserverRegistryState,
observerRegistryNotify,
evalObserverRegistryState,
emptyObserverRegistry,
observerRegistryHandlePdu,
observerRegistryRemoveProcess,
)
where
import Control.DeepSeq (NFData (rnf))
import Control.Eff
import Control.Eff.Concurrent.Process
import Control.Eff.Concurrent.Protocol
import Control.Eff.Concurrent.Protocol.Client
import Control.Eff.Concurrent.Protocol.Wrapper (Request (Cast))
import Control.Eff.Log
import Control.Eff.State.Strict
import Control.Lens
import Control.Monad
import Data.Dynamic
import Data.Kind
import Data.Map (Map)
import qualified Data.Map as Map
import Data.Proxy
import Data.Semigroup
import GHC.Generics
import GHC.Stack
-- * Observers
-- ** Observables
-- | A /protocol/ to communicate 'Observed' events from a sources to many sinks.
--
-- A sink is any process that serves a protocol with a 'Pdu' instance that embeds
-- the 'Observer' Pdu via an 'HasPduPrism' instance.
--
-- This type has /dual use/, for one it serves as type-index for 'Pdu', i.e.
-- 'HasPdu' respectively, and secondly it contains an 'ObservationSink' and
-- a 'MonitorReference'.
--
-- The 'ObservationSink' is used to serialize and send the 'Observed' events,
-- while the 'ProcessId' serves as key for internal maps.
--
-- @since 0.28.0
newtype Observer event
= MkObserver (Arg ProcessId (ObservationSink event))
deriving (Eq, Ord, Typeable)
instance ToTypeLogMsg event => ToLogMsg (Observer event) where
toLogMsg (MkObserver (Arg t _)) = toTypeLogMsg (Proxy @(Observer event)) <> toLogMsg t
instance ToTypeLogMsg event => ToTypeLogMsg (Observer event) where
toTypeLogMsg _ = toTypeLogMsg (Proxy @event) <> packLogMsg "_observer"
instance NFData (Observer event) where
rnf (MkObserver (Arg x y)) = rnf x `seq` rnf y
instance (Tangible event) => HasPdu (Observer event) where
data Pdu (Observer event) r where
Observed :: event -> Pdu (Observer event) 'Asynchronous
deriving (Typeable)
instance NFData event => NFData (Pdu (Observer event) r) where
rnf (Observed event) = rnf event
instance ToLogMsg event => ToLogMsg (Pdu (Observer event) r) where
toLogMsg (Observed event) =
packLogMsg "observed: " <> toLogMsg event
-- | The Information necessary to wrap an 'Observed' event to a process specific
-- message, e.g. the embedded 'Observer' 'Pdu' instance, and the 'MonitorReference' of
-- the destination process.
--
-- @since 0.28.0
data ObservationSink event = MkObservationSink
{ _observerSerializer :: Serializer (Pdu (Observer event) 'Asynchronous),
_observerMonitorReference :: MonitorReference
}
deriving (Generic, Typeable)
instance NFData (ObservationSink event) where
rnf (MkObservationSink s p) = s `seq` rnf p
-- | Convenience type alias.
--
-- @since 0.28.0
type IsObservable eventSource event =
( Tangible event,
Embeds eventSource (ObserverRegistry event),
HasPdu eventSource,
ToTypeLogMsg event,
ToLogMsg event
)
-- | Convenience type alias.
--
-- @since 0.28.0
type CanObserve eventSink event =
( Tangible event,
Embeds eventSink (Observer event),
HasPdu eventSink
)
-- | And an 'Observer' to the set of recipients for all observations reported by 'observerRegistryNotify'.
-- Note that the observerRegistry are keyed by the observing process, i.e. a previous entry for the process
-- contained in the 'Observer' is overwritten. If you want multiple entries for a single process, just
-- combine several filter functions.
--
-- @since 0.16.0
registerObserver ::
forall event eventSink eventSource r q.
( HasProcesses r q,
TangiblePdu eventSource 'Asynchronous,
IsObservable eventSource event,
TangiblePdu eventSink 'Asynchronous,
CanObserve eventSink event
) =>
Endpoint eventSource ->
Endpoint eventSink ->
Eff r ()
registerObserver eventSource eventSink =
cast eventSource (RegisterObserver serializer (eventSink ^. fromEndpoint))
where
serializer =
MkSerializer
( toMessage
. Cast
. embedPdu @eventSink @(Observer event) @('Asynchronous)
)
-- | Send the 'ForgetObserver' message
--
-- @since 0.16.0
forgetObserver ::
forall event eventSink eventSource r q.
( HasProcesses r q,
TangiblePdu eventSource 'Asynchronous,
IsObservable eventSource event
) =>
Endpoint eventSource ->
Endpoint eventSink ->
Eff r ()
forgetObserver eventSource eventSink =
forgetObserverUnsafe @event @eventSource eventSource (eventSink ^. fromEndpoint)
-- | Send the 'ForgetObserver' message, use a raw 'ProcessId' as parameter.
--
-- @since 0.28.0
forgetObserverUnsafe ::
forall event eventSource r q.
( HasProcesses r q,
TangiblePdu eventSource 'Asynchronous,
IsObservable eventSource event
) =>
Endpoint eventSource ->
ProcessId ->
Eff r ()
forgetObserverUnsafe eventSource eventSink =
cast eventSource (ForgetObserver @event eventSink)
-- ** Observer Support Functions
-- * Managing Observers
-- | A protocol for managing 'Observer's, encompassing registration and de-registration of
-- 'Observer's.
--
-- @since 0.28.0
data ObserverRegistry (event :: Type) = MkObserverRegistry
{_observerRegistry :: Map ProcessId (ObservationSink event)}
deriving (Typeable)
instance ToTypeLogMsg event => ToTypeLogMsg (ObserverRegistry event) where
toTypeLogMsg _ = toTypeLogMsg (Proxy @event) <> packLogMsg "_observer_registry_event"
instance (Tangible event) => HasPdu (ObserverRegistry event) where
data Pdu (ObserverRegistry event) r where
-- | This message denotes that the given 'Observer' should receive observations until 'ForgetObserver' is
-- received.
--
-- @since 0.28.0
RegisterObserver :: Serializer (Pdu (Observer event) 'Asynchronous) -> ProcessId -> Pdu (ObserverRegistry event) 'Asynchronous
-- | This message denotes that the given 'Observer' should not receive observations anymore.
--
-- @since 0.16.1
ForgetObserver :: ProcessId -> Pdu (ObserverRegistry event) 'Asynchronous
-- -- | This message denotes that a monitored process died
-- --
-- -- @since 0.28.0
-- ObserverMightBeDown :: MonitorReference -> Pdu (ObserverRegistry event) ( 'Synchronous Bool)
deriving (Typeable)
instance NFData (Pdu (ObserverRegistry event) r) where
rnf (RegisterObserver ser pid) = rnf ser `seq` rnf pid
rnf (ForgetObserver pid) = rnf pid
instance ToTypeLogMsg event => ToLogMsg (Pdu (ObserverRegistry event) r) where
toLogMsg (RegisterObserver _ser pid) = packLogMsg "register " <> toTypeLogMsg (Proxy @event) <> packLogMsg " observer " <> toLogMsg pid
toLogMsg (ForgetObserver pid) = packLogMsg "forget " <> toTypeLogMsg (Proxy @event) <> packLogMsg " observer " <> toLogMsg pid
-- ** Protocol for integrating 'ObserverRegistry' into processes.
-- | Provide the implementation for the 'ObserverRegistry' Protocol, this handled 'RegisterObserver' and 'ForgetObserver'
-- messages. It also adds the 'ObserverRegistryState' constraint to the effect list.
--
-- @since 0.28.0
observerRegistryHandlePdu ::
forall event q r.
( HasCallStack,
ToTypeLogMsg event,
HasProcesses r q,
Member (ObserverRegistryState event) r,
Member Logs r
) =>
Pdu (ObserverRegistry event) 'Asynchronous ->
Eff r ()
observerRegistryHandlePdu = \case
RegisterObserver ser pid -> do
monRef <- monitor pid
let sink = MkObservationSink ser monRef
observer = MkObserver (Arg pid sink)
modify @(ObserverRegistry event) (over observerRegistry (Map.insert pid sink))
os <- get @(ObserverRegistry event)
logDebug
(LABEL "registered" observer)
( LABEL
"current number of observers" -- TODO put this info into the process details
(Map.size (view observerRegistry os))
)
ForgetObserver ob -> do
wasRemoved <- observerRegistryRemoveProcess @event ob
unless wasRemoved $
logDebug (LABEL "unknown observer " ob)
-- | Remove the entry in the 'ObserverRegistry' for the 'ProcessId'
-- and return 'True' if there was an entry, 'False' otherwise.
--
-- @since 0.28.0
observerRegistryRemoveProcess ::
forall event q r.
( HasCallStack,
ToTypeLogMsg event,
HasProcesses r q,
Member (ObserverRegistryState event) r,
Member Logs r
) =>
ProcessId ->
Eff r Bool
observerRegistryRemoveProcess ob = do
mSink <- view (observerRegistry . at ob) <$> get @(ObserverRegistry event)
modify @(ObserverRegistry event) (observerRegistry . at ob .~ Nothing)
os <- get @(ObserverRegistry event)
maybe
(pure False)
(foundIt os)
mSink
where
foundIt os sink@(MkObservationSink _ monRef) = do
demonitor monRef
logDebug
(LABEL "removed" (MkObserver $ Arg ob sink))
(LABEL "current number of observers" (Map.size (view observerRegistry os)))
pure True
-- | Keep track of registered 'Observer's.
--
-- Handle the 'ObserverRegistryState' effect, i.e. run 'evalState' on an 'emptyObserverRegistry'.
--
-- @since 0.28.0
evalObserverRegistryState :: Eff (ObserverRegistryState event ': r) a -> Eff r a
evalObserverRegistryState = evalState emptyObserverRegistry
-- | The empty 'ObserverRegistryState'
--
-- @since 0.28.0
emptyObserverRegistry :: ObserverRegistry event
emptyObserverRegistry = MkObserverRegistry Map.empty
-- | Alias for the effect that contains the observers managed by 'evalObserverRegistryState'
type ObserverRegistryState event = State (ObserverRegistry event)
-- | An 'Iso' for the 'Map' used internally.
observerRegistry :: Iso' (ObserverRegistry event) (Map ProcessId (ObservationSink event))
observerRegistry = iso _observerRegistry MkObserverRegistry
-- | Report an observation to all observers.
-- The process needs to 'evalObserverRegistryState' and to 'observerRegistryHandlePdu'.
--
-- @since 0.28.0
observerRegistryNotify ::
forall event r q.
( HasProcesses r q,
Member (ObserverRegistryState event) r
) =>
event ->
Eff r ()
observerRegistryNotify observation = do
os <- view observerRegistry <$> get
mapM_ notifySomeObserver (Map.assocs os)
where
notifySomeObserver (destination, (MkObservationSink serializer _)) =
sendAnyMessage destination (runSerializer serializer (Observed observation))