diff --git a/.gitignore b/.gitignore --- a/.gitignore +++ b/.gitignore @@ -11,3 +11,4 @@ cabal.config *~ *.swp +.stack-work/ diff --git a/CHANGELOG.markdown b/CHANGELOG.markdown --- a/CHANGELOG.markdown +++ b/CHANGELOG.markdown @@ -1,3 +1,10 @@ +0.14.0.0 +-------- +* Fix deadlock issues in subscription code. +* Improve cluster connection fiability. +* Internal refactoring. +* Typeful `EventType` + 0.13.1.7 -------- * Report stream name when facing a 'Stream not found' error from the server on catchup subscription. diff --git a/Database/EventStore.hs b/Database/EventStore.hs --- a/Database/EventStore.hs +++ b/Database/EventStore.hs @@ -45,6 +45,7 @@ -- * Event , Event , EventData + , EventType(..) , createEvent , withJson , withJsonAndMetadata @@ -108,11 +109,13 @@ , SubscriptionClosed(..) , SubscriptionId , Subscription - , S.Running(..) - , S.SubDropReason(..) + , SubDropReason(..) , waitConfirmation + , unsubscribeConfirmed + , unsubscribeConfirmedSTM + , waitUnsubscribeConfirmed -- * Volatile Subscription - , S.Regular + , Regular , subscribe , subscribeToAll , getSubId @@ -125,17 +128,18 @@ , getSubLastCommitPos , getSubLastEventNumber -- * Catch-up Subscription - , S.Catchup + , Catchup , subscribeFrom , subscribeToAllFrom , waitTillCatchup , hasCaughtUp + , hasCaughtUpSTM -- * Persistent Subscription - , S.Persistent + , Persistent , PersistentSubscriptionSettings(..) , SystemConsumerStrategy(..) , NakAction(..) - , S.PersistActionException(..) + , PersistActionException(..) , acknowledge , acknowledgeEvents , failed @@ -194,14 +198,14 @@ -------------------------------------------------------------------------------- import ClassyPrelude hiding (Builder, group) import Data.List.NonEmpty(NonEmpty(..), nonEmpty) -import Data.UUID import Network.Connection (TLSSettings) -------------------------------------------------------------------------------- import Database.EventStore.Internal.Command import Database.EventStore.Internal.Connection import Database.EventStore.Internal.Discovery -import qualified Database.EventStore.Internal.Manager.Subscription as S +import Database.EventStore.Internal.Subscription +import Database.EventStore.Internal.Manager.Subscription.Driver hiding (unsubscribe) import Database.EventStore.Internal.Manager.Subscription.Message import Database.EventStore.Internal.Operation (OperationError(..)) import qualified Database.EventStore.Internal.Operations as Op @@ -279,213 +283,6 @@ sendEvents mgr evt_stream exp_ver [evt] -------------------------------------------------------------------------------- --- | Represents a subscription id. -newtype SubscriptionId = SubId UUID deriving (Eq, Ord, Show) - --------------------------------------------------------------------------------- --- | Determines whether or not any link events encontered in the stream will be --- resolved. -getSubResolveLinkTos :: Subscription S.Regular -> Bool -getSubResolveLinkTos = S._subTos . _subInner - --------------------------------------------------------------------------------- --- | Non blocking version of `waitTillCatchup`. -hasCaughtUp :: Subscription S.Catchup -> IO Bool -hasCaughtUp sub = atomically $ _hasCaughtUp sub - --------------------------------------------------------------------------------- --- | Waits until 'CatchupSubscription' subscription catch-up its stream. -waitTillCatchup :: Subscription S.Catchup -> IO () -waitTillCatchup sub = atomically $ do - caughtUp <- _hasCaughtUp sub - when (not caughtUp) retrySTM - --------------------------------------------------------------------------------- -_hasCaughtUp :: Subscription S.Catchup -> STM Bool -_hasCaughtUp Subscription{..} = do - res <- _subState - case res of - SubState sm _ -> return $ S.hasCaughtUp sm - SubException e -> throwSTM e - --------------------------------------------------------------------------------- --- | Tracks a 'Subcription' lifecycle. It holds a 'Subscription' state machine --- and `SubDropReason` if any. -data SubState a - = SubState (S.Subscription a) (Maybe S.SubDropReason) - | forall e. Exception e => SubException e - -- ^ Hack used to cover a special exception that can arise from - -- a catchup subscription for instance. One example is when asking a - -- catchup subscription on stream that doesn't exist yet. - --------------------------------------------------------------------------------- --- | It's possible to subscribe to a stream and be notified when new events are --- written to that stream. There are three types of subscription which are --- available, all of which can be useful in different situations. --- --- * 'S.Regular' --- --- * 'S.Catchup' --- --- * 'S.Persistent' -data Subscription a = - Subscription - { _subState :: STM (SubState a) - , _subSetState :: SubState a -> STM () - , _subRun :: TMVar S.Running - , _subStream :: Text - , _subProd :: Production - , _subInner :: a - } - --------------------------------------------------------------------------------- --- | Gets the ID of the subscription. -getSubId :: Subscription a -> IO SubscriptionId -getSubId Subscription{..} = atomically $ do - run <- readTMVar _subRun - return $ SubId $ S.runningUUID run - --------------------------------------------------------------------------------- --- | Gets the subscription stream name. -getSubStream :: Subscription a -> Text -getSubStream = _subStream - --------------------------------------------------------------------------------- --- | Asynchronously unsubscribe from the the stream. -unsubscribe :: Subscription a -> IO () -unsubscribe Subscription{..} = do - run <- atomically $ readTMVar _subRun - pushUnsubscribe _subProd run - --------------------------------------------------------------------------------- --- | If the subscription is on the $all stream. -isSubscribedToAll :: Subscription a -> Bool -isSubscribedToAll = (== "") . getSubStream - --------------------------------------------------------------------------------- --- | The last commit position seen on the subscription (if this a subscription --- to $all stream). -getSubLastCommitPos :: Subscription a -> IO Int64 -getSubLastCommitPos Subscription{..} = atomically $ do - run <- readTMVar _subRun - return $ S.runningLastCommitPosition run - --------------------------------------------------------------------------------- --- | The last event number seen on the subscription (if this is a subscription --- to a single stream). -getSubLastEventNumber :: Subscription a -> IO (Maybe Int32) -getSubLastEventNumber Subscription{..} = atomically $ do - run <- readTMVar _subRun - return $ S.runningLastEventNumber run - --------------------------------------------------------------------------------- --- | Awaits for the next event. -nextEvent :: Subscription a -> IO ResolvedEvent -nextEvent sub = atomically $ do - m <- _nextEventMaybe sub - case m of - Nothing -> retrySTM - Just e -> return e - --------------------------------------------------------------------------------- --- | Non blocking version of 'nextEvent'. -nextEventMaybe :: Subscription a -> IO (Maybe ResolvedEvent) -nextEventMaybe = atomically . _nextEventMaybe - --------------------------------------------------------------------------------- --- | Waits until the `Subscription` has been confirmed. -waitConfirmation :: Subscription a -> IO () -waitConfirmation s = atomically $ do - _ <- readTMVar $ _subRun s - return () - --------------------------------------------------------------------------------- -_nextEventMaybe :: Subscription a -> STM (Maybe ResolvedEvent) -_nextEventMaybe Subscription{..} = do - st <- _subState - case st of - SubException e -> throwSTM e - SubState sub close -> do - run <- readTMVar _subRun - let (res, nxt) = S.readNext sub - case res of - Nothing -> do - case close of - Nothing -> return Nothing - Just err -> throwSTM $ SubscriptionClosed run err - Just e -> do - _subSetState $ SubState nxt close - return $ Just e - --------------------------------------------------------------------------------- --- | Acknowledges those event ids have been successfully processed. -notifyEventsProcessed :: Subscription S.Persistent -> [UUID] -> IO () -notifyEventsProcessed Subscription{..} evts = do - run <- atomically $ readTMVar _subRun - pushAckPersist _subProd run evts - --------------------------------------------------------------------------------- --- | Acknowledges that 'ResolvedEvent' has been successfully processed. -acknowledge :: Subscription S.Persistent -> ResolvedEvent -> IO () -acknowledge sub e = notifyEventsProcessed sub [resolvedEventOriginalId e] - --------------------------------------------------------------------------------- --- | Acknowledges those 'ResolvedEvent's have been successfully processed. -acknowledgeEvents :: Subscription S.Persistent -> [ResolvedEvent] -> IO () -acknowledgeEvents sub = notifyEventsProcessed sub . fmap resolvedEventOriginalId - --------------------------------------------------------------------------------- --- | Mark a message that has failed processing. The server will take action --- based upon the action parameter. -failed :: Subscription S.Persistent - -> ResolvedEvent - -> NakAction - -> Maybe Text - -> IO () -failed sub e a r = notifyEventsFailed sub a r [resolvedEventOriginalId e] - --------------------------------------------------------------------------------- --- | Mark messages that have failed processing. The server will take action --- based upon the action parameter. -eventsFailed :: Subscription S.Persistent - -> [ResolvedEvent] - -> NakAction - -> Maybe Text - -> IO () -eventsFailed sub evts a r = - notifyEventsFailed sub a r $ fmap resolvedEventOriginalId evts - --------------------------------------------------------------------------------- --- | Acknowledges those event ids have failed to be processed successfully. -notifyEventsFailed :: Subscription S.Persistent - -> NakAction - -> Maybe Text - -> [UUID] - -> IO () -notifyEventsFailed Subscription{..} act res evts = do - run <- atomically $ readTMVar _subRun - pushNakPersist _subProd run act res evts - --------------------------------------------------------------------------------- --- | Modifies 'SubState' internal state machine, letting any 'SubDropReason' --- untouched. -modifySubSM :: (S.Subscription a -> S.Subscription a) - -> SubState a - -> SubState a -modifySubSM k (SubState sm r) = SubState (k sm) r -modifySubSM _ s = s - --------------------------------------------------------------------------------- --- | This exception is raised when the user tries to get the next event from a --- 'Subscription' that is already closed. -data SubscriptionClosed = - SubscriptionClosed S.Running S.SubDropReason - deriving (Show, Typeable) - --------------------------------------------------------------------------------- - -instance Exception SubscriptionClosed --------------------------------------------------------------------------------- -- | Sends a list of 'Event' to given stream. sendEvents :: Connection -> Text -- ^ Stream name @@ -664,33 +461,41 @@ Position c_pos p_pos = pos -------------------------------------------------------------------------------- +mkSubEnv :: Connection -> SubEnv +mkSubEnv Connection{..} = + SubEnv + { subSettings = _settings + , subPushOp = pushOperation _prod + , subPushConnect = \k cmd -> + case cmd of + PushRegular stream tos -> + pushConnectStream _prod k stream tos + PushPersistent group stream size -> + pushConnectPersist _prod k group stream size + , subPushUnsub = pushUnsubscribe _prod + , subAckCmd = \cmd run uuids -> + case cmd of + AckCmd -> pushAckPersist _prod run uuids + NakCmd act res -> pushNakPersist _prod run act res uuids + , subForceReconnect = \node -> + pushForceReconnect _prod node + } + +-------------------------------------------------------------------------------- -- | Subcribes to given stream. subscribe :: Connection -> Text -- ^ Stream name -> Bool -- ^ Resolve Link Tos - -> IO (Subscription S.Regular) -subscribe Connection{..} stream_id res_lnk_tos = do - mvar <- newEmptyTMVarIO - var <- newTVarIO $ SubState S.regularSubscription Nothing - let mk r = putTMVar mvar r - recv = readTVar var - send = writeTVar var - dropped r = do - SubState sm _ <- readTVar var - writeTVar var $ SubState sm (Just r) - cb = createSubAsync mk recv send dropped - pushConnectStream _prod cb stream_id res_lnk_tos - let getState = readTVar var - setState = writeTVar var - return $ Subscription getState setState mvar stream_id _prod - (S.Regular res_lnk_tos) + -> IO (Subscription Regular) +subscribe conn streamId resLnkTos = + regularSub (mkSubEnv conn) streamId resLnkTos -------------------------------------------------------------------------------- -- | Subcribes to $all stream. subscribeToAll :: Connection -> Bool -- ^ Resolve Link Tos - -> IO (Subscription S.Regular) -subscribeToAll conn res_lnk_tos = subscribe conn "" res_lnk_tos + -> IO (Subscription Regular) +subscribeToAll conn = subscribe conn "" -------------------------------------------------------------------------------- -- | Subscribes to given stream. If last checkpoint is defined, this will @@ -702,11 +507,11 @@ -> Bool -- ^ Resolve Link Tos -> Maybe Int32 -- ^ Last checkpoint -> Maybe Int32 -- ^ Batch size - -> IO (Subscription S.Catchup) -subscribeFrom conn stream_id res_lnk_tos last_chk_pt batch_m = - subscribeFromCommon conn stream_id res_lnk_tos batch_m tpe + -> IO (Subscription Catchup) +subscribeFrom conn streamId resLnkTos lastChkPt batch = + subscribeFromCommon conn resLnkTos batch tpe where - tpe = Op.RegularCatchup stream_id (fromMaybe 0 last_chk_pt) + tpe = Op.RegularCatchup streamId (fromMaybe 0 lastChkPt) -------------------------------------------------------------------------------- -- | Same as 'subscribeFrom' but applied to $all stream. @@ -714,62 +519,26 @@ -> Bool -- ^ Resolve Link Tos -> Maybe Position -- ^ Last checkpoint -> Maybe Int32 -- ^ Batch size - -> IO (Subscription S.Catchup) -subscribeToAllFrom conn res_lnk_tos last_chk_pt batch_m = - subscribeFromCommon conn "" res_lnk_tos batch_m tpe + -> IO (Subscription Catchup) +subscribeToAllFrom conn resLnkTos lastChkPt batch = + subscribeFromCommon conn resLnkTos batch tpe where - Position c_pos p_pos = fromMaybe positionStart last_chk_pt - tpe = Op.AllCatchup c_pos p_pos + Position cPos pPos = fromMaybe positionStart lastChkPt + tpe = Op.AllCatchup cPos pPos -------------------------------------------------------------------------------- subscribeFromCommon :: Connection - -> Text -> Bool -> Maybe Int32 -> Op.CatchupState - -> IO (Subscription S.Catchup) -subscribeFromCommon Connection{..} stream_id res_lnk_tos batch_m tpe = do - mvarRun <- newEmptyTMVarIO - mvarSub <- newEmptyTMVarIO - let readFrom res = - case res of - -- We want to notify the user that something went wrong in the - -- first phase of a catchup subscription (e.g. reading the - -- stream forward until we catchup to stream's end). This - -- prevents a deadlock on user side in case where the user calls - -- `waitTillCatchup` on a stream that doesn't exist. - Left e -> atomically $ do - isEmpty <- isEmptyTMVar mvarSub - if isEmpty - then putTMVar mvarSub (SubException e) - else () <$ swapTMVar mvarSub (SubException e) - - Right (xs, eos, chk) -> atomically $ do - -- When a catchup subscription receives events for the - -- first time. - whenM (isEmptyTMVar mvarSub) $ do - let initState = SubState S.catchupSubscription Nothing - putTMVar mvarSub initState - - subState <- takeTMVar mvarSub - let nxtSubState = - modifySubSM (S.batchRead xs eos chk) subState - - putTMVar mvarSub nxtSubState - mk = putTMVar mvarRun - rcv = readTMVar mvarSub - send = \x -> () <$ swapTMVar mvarSub x - dropped r = do - SubState sm _ <- takeTMVar mvarSub - putTMVar mvarSub $ SubState sm (Just r) - op = Op.catchup _settings tpe res_lnk_tos batch_m - cb = createSubAsync mk rcv send dropped - - pushOperation _prod readFrom op - pushConnectStream _prod cb stream_id res_lnk_tos - let getState = readTMVar mvarSub - setState = \x -> () <$ swapTMVar mvarSub x - return $ Subscription getState setState mvarRun stream_id _prod S.Catchup + -> IO (Subscription Catchup) +subscribeFromCommon conn resLnkTos batch tpe = + catchupSub (mkSubEnv conn) params + where + params = CatchupParams { catchupResLnkTos = resLnkTos + , catchupState = tpe + , catchupBatchSize = batch + } -------------------------------------------------------------------------------- -- | Asynchronously sets the metadata for a stream. @@ -799,7 +568,7 @@ -> Text -> Text -> PersistentSubscriptionSettings - -> IO (Async (Maybe S.PersistActionException)) + -> IO (Async (Maybe PersistActionException)) createPersistentSubscription Connection{..} group stream sett = do mvar <- newEmptyTMVarIO let _F res = atomically $ @@ -815,7 +584,7 @@ -> Text -> Text -> PersistentSubscriptionSettings - -> IO (Async (Maybe S.PersistActionException)) + -> IO (Async (Maybe PersistActionException)) updatePersistentSubscription Connection{..} group stream sett = do mvar <- newEmptyTMVarIO let _F res = atomically $ @@ -830,7 +599,7 @@ deletePersistentSubscription :: Connection -> Text -> Text - -> IO (Async (Maybe S.PersistActionException)) + -> IO (Async (Maybe PersistActionException)) deletePersistentSubscription Connection{..} group stream = do mvar <- newEmptyTMVarIO let _F res = atomically $ @@ -847,22 +616,9 @@ -> Text -> Text -> Int32 - -> IO (Subscription S.Persistent) -connectToPersistentSubscription Connection{..} group stream bufSize = do - mvar <- newEmptyTMVarIO - var <- newTVarIO $ SubState S.persistentSubscription Nothing - let mk r = putTMVar mvar r - recv = readTVar var - send = writeTVar var - dropped r = do - SubState sm _ <- readTVar var - writeTVar var $ SubState sm (Just r) - cb = createSubAsync mk recv send dropped - pushConnectPersist _prod cb group stream bufSize - let getState = readTVar var - setState = writeTVar var - return $ Subscription getState setState mvar stream _prod - (S.Persistent group) + -> IO (Subscription Persistent) +connectToPersistentSubscription conn group stream bufSize = + persistentSub (mkSubEnv conn) group stream bufSize -------------------------------------------------------------------------------- createOpAsync :: IO (Either OperationError a -> IO (), Async a) @@ -872,25 +628,3 @@ res <- readMVar mvar either throwIO return res return (putMVar mvar, as) - --------------------------------------------------------------------------------- -createSubAsync :: (S.Running -> STM ()) - -> STM (SubState a) - -> (SubState a -> STM ()) - -> (S.SubDropReason -> STM ()) - -> (S.SubConnectEvent -> IO ()) -createSubAsync mk rcv send quit = go - where - go (S.SubConfirmed run) = atomically $ mk run - go (S.EventAppeared e) = atomically $ do - st <- rcv - case st of - SubState sm close -> - let nxt = S.eventArrived e sm in - send $ SubState nxt close - SubException _ -> - -- At this moment [07 October 2016], this can only happen during - -- the first phase of a catchup subscription where the user - -- asked for a subscription on a stream that doesn't exist. - return () - go (S.Dropped r) = atomically $ quit r diff --git a/Database/EventStore/Internal/Command.hs b/Database/EventStore/Internal/Command.hs --- a/Database/EventStore/Internal/Command.hs +++ b/Database/EventStore/Internal/Command.hs @@ -13,21 +13,15 @@ module Database.EventStore.Internal.Command (Command(..)) where -------------------------------------------------------------------------------- -import Data.Word -import Numeric +import ClassyPrelude -------------------------------------------------------------------------------- -import ClassyPrelude +import Database.EventStore.Internal.Utils -------------------------------------------------------------------------------- -- | Internal command representation. newtype Command = Command { cmdWord8 :: Word8 } deriving (Eq, Ord, Num) -------------------------------------------------------------------------------- -padding :: String -> String -padding [x] = ['0',x] -padding xs = xs - --------------------------------------------------------------------------------- instance Show Command where - show (Command w) = "0x" ++ padding (showHex w "") + show (Command w) = prettyWord8 w diff --git a/Database/EventStore/Internal/Connection.hs b/Database/EventStore/Internal/Connection.hs --- a/Database/EventStore/Internal/Connection.hs +++ b/Database/EventStore/Internal/Connection.hs @@ -22,6 +22,7 @@ , connRecv , connIsClosed , newConnection + , connForceReconnect ) where -------------------------------------------------------------------------------- @@ -36,6 +37,7 @@ -------------------------------------------------------------------------------- import Database.EventStore.Internal.Command +import Database.EventStore.Internal.EndPoint import Database.EventStore.Internal.Discovery import Database.EventStore.Internal.Types import Database.EventStore.Logging @@ -63,6 +65,7 @@ Close :: In () Send :: Package -> In () Recv :: In Package + ForceReconnect :: NodeEndPoints -> In () -------------------------------------------------------------------------------- -- | Represents connection logic action to carry out. @@ -131,6 +134,11 @@ _ -> return False -------------------------------------------------------------------------------- +-- | Forces reconnection on given node. +connForceReconnect :: InternalConnection -> NodeEndPoints -> IO () +connForceReconnect conn = execute conn . ForceReconnect + +-------------------------------------------------------------------------------- -- Connection Logic -------------------------------------------------------------------------------- onlineLogic :: forall a. TMVar State @@ -175,6 +183,7 @@ handleInput _ conn Recv = recv conn handleInput uuid _ Id = return uuid handleInput _ conn Close = liftIO $ connectionClose conn +handleInput _ _ ForceReconnect{} = return () -------------------------------------------------------------------------------- -- | Main connection logic. It will automatically reconnect to the server when @@ -188,7 +197,10 @@ Errored e -> throwIO e WithConnection uuid conn op -> handleInput uuid conn op CreateConnection op -> do - (uuid, conn) <- openConnection iconn + (uuid, conn) <- + case op of + ForceReconnect node -> openConnection iconn (Just node) + _ -> openConnection iconn Nothing atomically $ putTMVar (_var iconn) (Online uuid conn) handleInput uuid conn op @@ -198,8 +210,14 @@ reachedMaxAttempt (AtMost n) cur = n <= cur -------------------------------------------------------------------------------- -openConnection :: InternalConnection -> IO (UUID, Connection) -openConnection InternalConnection{..} = attempt 1 +isSsl :: Settings -> Bool +isSsl = isJust . s_ssl + +-------------------------------------------------------------------------------- +openConnection :: InternalConnection + -> Maybe NodeEndPoints + -> IO (UUID, Connection) +openConnection InternalConnection{..} nodeM = attempt 1 where delay = s_reconnect_delay_secs _setts * secs @@ -212,17 +230,31 @@ attempt trialCount = do _settingsLog _setts (Info $ Connecting trialCount) - old <- readIORef _last - ept_opt <- runDiscovery _disc old - case ept_opt of - Nothing -> handleFailure trialCount - Just ept -> do - let host = endPointIp ept + case nodeM of + Just node -> do + let ept = if isSsl _setts + then let Just pt = secureEndPoint node + in pt + else tcpEndPoint node + + host = endPointIp ept port = endPointPort ept res <- tryAny $ connect _setts _ctx host port case res of Left _ -> handleFailure trialCount Right st -> st <$ writeIORef _last (Just ept) + Nothing -> do + old <- readIORef _last + ept_opt <- runDiscovery _disc old + case ept_opt of + Nothing -> handleFailure trialCount + Just ept -> do + let host = endPointIp ept + port = endPointPort ept + res <- tryAny $ connect _setts _ctx host port + case res of + Left _ -> handleFailure trialCount + Right st -> st <$ writeIORef _last (Just ept) -------------------------------------------------------------------------------- secs :: Int diff --git a/Database/EventStore/Internal/Discovery.hs b/Database/EventStore/Internal/Discovery.hs --- a/Database/EventStore/Internal/Discovery.hs +++ b/Database/EventStore/Internal/Discovery.hs @@ -49,6 +49,9 @@ import System.Random -------------------------------------------------------------------------------- +import Database.EventStore.Internal.EndPoint + +-------------------------------------------------------------------------------- data DnsDiscoveryException = MaxDiscoveryAttemptReached ByteString | DNSDiscoveryError DNSError @@ -56,18 +59,6 @@ -------------------------------------------------------------------------------- instance Exception DnsDiscoveryException - --------------------------------------------------------------------------------- --- | Gathers both an IPv4 and a port. -data EndPoint = - EndPoint - { endPointIp :: !String - , endPointPort :: !Int - } deriving Show - --------------------------------------------------------------------------------- -emptyEndPoint :: EndPoint -emptyEndPoint = EndPoint "" 0 -------------------------------------------------------------------------------- httpRequest :: EndPoint -> String -> IO Request diff --git a/Database/EventStore/Internal/EndPoint.hs b/Database/EventStore/Internal/EndPoint.hs new file mode 100644 --- /dev/null +++ b/Database/EventStore/Internal/EndPoint.hs @@ -0,0 +1,34 @@ +-------------------------------------------------------------------------------- +-- | +-- Module : Database.EventStore.Internal.EndPoint +-- Copyright : (C) 2016 Yorick Laupa +-- License : (see the file LICENSE) +-- +-- Maintainer : Yorick Laupa +-- Stability : provisional +-- Portability : non-portable +-- +-------------------------------------------------------------------------------- +module Database.EventStore.Internal.EndPoint where + +-------------------------------------------------------------------------------- +import ClassyPrelude + +-------------------------------------------------------------------------------- +-- | Gathers both an IPv4 and a port. +data EndPoint = + EndPoint + { endPointIp :: !String + , endPointPort :: !Int + } deriving Show + +-------------------------------------------------------------------------------- +emptyEndPoint :: EndPoint +emptyEndPoint = EndPoint "" 0 + +-------------------------------------------------------------------------------- +data NodeEndPoints = + NodeEndPoints + { tcpEndPoint :: !EndPoint + , secureEndPoint :: !(Maybe EndPoint) + } deriving Show diff --git a/Database/EventStore/Internal/Execution/Production.hs b/Database/EventStore/Internal/Execution/Production.hs --- a/Database/EventStore/Internal/Execution/Production.hs +++ b/Database/EventStore/Internal/Execution/Production.hs @@ -36,6 +36,7 @@ , pushNakPersist , pushUnsubscribe , prodWaitTillClosed + , pushForceReconnect ) where -------------------------------------------------------------------------------- @@ -50,14 +51,16 @@ -------------------------------------------------------------------------------- import Database.EventStore.Internal.Connection import Database.EventStore.Internal.Discovery +import Database.EventStore.Internal.EndPoint import Database.EventStore.Internal.Generator -import Database.EventStore.Internal.Manager.Subscription hiding +import Database.EventStore.Internal.Manager.Subscription.Driver hiding ( submitPackage , unsubscribe , ackPersist , nakPersist , abort ) +import Database.EventStore.Internal.Manager.Subscription.Model import Database.EventStore.Internal.Operation import Database.EventStore.Internal.Processor import Database.EventStore.Internal.Types @@ -191,6 +194,7 @@ Text Text | AckPersist Running [UUID] | NakPersist Running NakAction (Maybe Text) [UUID] + | ForceReconnect NodeEndPoints -------------------------------------------------------------------------------- pushCmd :: Production -> Msg -> IO () @@ -286,6 +290,10 @@ prodWaitTillClosed (Prod _ disposed) = atomically disposed -------------------------------------------------------------------------------- +pushForceReconnect :: Production -> NodeEndPoints -> IO () +pushForceReconnect prod n = pushCmd prod (ForceReconnect n) + +-------------------------------------------------------------------------------- newtype Job = Job (IO ()) -------------------------------------------------------------------------------- @@ -374,6 +382,9 @@ writeCycleQueue _pkgQueue pkg go nxt go (Await new_proc) = return new_proc + go (ForceReconnectCmd node nxt) = do + writeCycleQueue _queue (ForceReconnect node) + go nxt -------------------------------------------------------------------------------- -- | First execution mode. It spawns initial reader, runner and writer threads. @@ -390,6 +401,13 @@ cruising env -------------------------------------------------------------------------------- +data ForceReconnectException = ForceReconnectionException NodeEndPoints + deriving (Show, Typeable) + +-------------------------------------------------------------------------------- +instance Exception ForceReconnectException + +-------------------------------------------------------------------------------- -- | Crusing execution mode. Reads and handle message coming from the channel as -- those are arrived. That mode is used when the connection to the server is -- still live. We might have deconnection once in a while but at the end, if @@ -400,6 +418,7 @@ s <- readTVarIO _state case msg of Stopped _ e -> throwIO e + ForceReconnect node -> throwIO $ ForceReconnectionException node Arrived pkg -> do let sm = submitPackage pkg $ _proc s atomically $ do @@ -517,6 +536,14 @@ raiseException e _ = throwIO e -------------------------------------------------------------------------------- +handles :: SomeException -> [Handler IO a] -> IO a +handles e [] = throwIO e +handles e (Handler k:hs) = + case fromException e of + Just t -> k t + _ -> handles e hs + +-------------------------------------------------------------------------------- -- | Main Production execution model entry point. newExecutionModel :: Settings -> Discovery -> IO Production newExecutionModel setts disc = do @@ -535,15 +562,23 @@ case res of Left e -> do _settingsLog setts (Error $ UnexpectedException e) - case fromException e of - Just (_ :: ConnectionException) -> atomically $ do - writeTVar nxt_sub (raiseException e) - putTMVar disposed () - _ -> do new_conn <- newConnection setts disc - writeIORef conn_ref new_conn - _ <- forkFinally (bootstrap env) handler - return () - + handles e + [ Handler $ \(_ :: ConnectionException) -> + atomically $ do + writeTVar nxt_sub (raiseException e) + putTMVar disposed () + , Handler $ \(ForceReconnectionException node) -> do + new_conn <- newConnection setts disc + connForceReconnect new_conn node + writeIORef conn_ref new_conn + _ <- forkFinally (bootstrap env) handler + return () + , Handler $ \(_ :: SomeException) -> do + new_conn <- newConnection setts disc + writeIORef conn_ref new_conn + _ <- forkFinally (bootstrap env) handler + return () + ] _ -> atomically $ putTMVar disposed () _ <- forkFinally (bootstrap env) handler return $ Prod nxt_sub $ do diff --git a/Database/EventStore/Internal/Manager/Operation/Model.hs b/Database/EventStore/Internal/Manager/Operation/Model.hs --- a/Database/EventStore/Internal/Manager/Operation/Model.hs +++ b/Database/EventStore/Internal/Manager/Operation/Model.hs @@ -77,6 +77,7 @@ -- ^ Asks for sending the given 'Package'. | Await (Model r) -- ^ waits for more input. + | NotHandled MasterInfo (Transition r) -------------------------------------------------------------------------------- -- | Main 'Operation' bookkeeping state machine. @@ -139,20 +140,40 @@ -------------------------------------------------------------------------------- runPackage :: Settings -> State r -> Package -> Maybe (Transition r) -runPackage setts st Package{..} = do +runPackage setts st pkg@Package{..} = do Elem op resp_cmd cont cb <- lookup packageCorrelation $ _pending st let nxt_ps = deleteMap packageCorrelation $ _pending st nxt_st = st { _pending = nxt_ps } - if resp_cmd /= packageCmd - then - let r = cb $ Left $ InvalidServerResponse resp_cmd packageCmd in - return $ Produce r (Await $ Model $ execute setts nxt_st) - else - case runGet decodeMessage packageData of - Left e -> + case packageCmd of + -- Bad request + 0xF0 -> do + let reason = packageDataAsText pkg + resp = ServerError reason + value = cb $ Left $ resp + return $ Produce value (Await $ Model $ execute setts nxt_st) + 0xF4 -> do + let value = cb $ Left NotAuthenticatedOp + return $ Produce value (Await $ Model $ execute setts nxt_st) + -- Not handled + 0xF1 -> do + msg <- maybeDecodeMessage packageData + let reason = getField $ notHandledReason msg + case reason of + N_NotMaster -> do + info <- getField $ notHandledAdditionalInfo msg + let next = runOperation setts cb op op nxt_st + return $ NotHandled (masterInfo info) next + -- In this case with just retry the operation. + _ -> return $ runOperation setts cb op op nxt_st + _ | packageCmd /= resp_cmd -> do + let r = cb $ Left $ InvalidServerResponse resp_cmd packageCmd + return $ Produce r (Await $ Model $ execute setts nxt_st) + | otherwise -> + case runGet decodeMessage packageData of + Left e -> let r = cb $ Left $ ProtobufDecodingError e in return $ Produce r (Await $ Model $ execute setts nxt_st) - Right m -> return $ runOperation setts cb op (cont m) nxt_st + Right m -> return $ runOperation setts cb op (cont m) nxt_st -------------------------------------------------------------------------------- abortOperations :: Settings -> State r -> Transition r @@ -174,3 +195,10 @@ execute setts st (New op cb) = Just $ runOperation setts cb op op st execute setts st (Pkg pkg) = runPackage setts st pkg execute setts st Abort = Just $ abortOperations setts st + +-------------------------------------------------------------------------------- +maybeDecodeMessage :: Decode a => ByteString -> Maybe a +maybeDecodeMessage bytes = + case runGet decodeMessage bytes of + Right a -> Just a + _ -> Nothing diff --git a/Database/EventStore/Internal/Manager/Subscription.hs b/Database/EventStore/Internal/Manager/Subscription.hs deleted file mode 100644 --- a/Database/EventStore/Internal/Manager/Subscription.hs +++ /dev/null @@ -1,220 +0,0 @@ -{-# LANGUAGE GADTs #-} -{-# LANGUAGE Rank2Types #-} -{-# LANGUAGE ScopedTypeVariables #-} --------------------------------------------------------------------------------- --- | --- Module : Database.EventStore.Internal.Manager.Subscription --- Copyright : (C) 2015 Yorick Laupa --- License : (see the file LICENSE) --- --- Maintainer : Yorick Laupa --- Stability : provisional --- Portability : non-portable --- --- Main subscription state machine declaration module. It also declares every --- functions required to drive a 'Subscription'. --------------------------------------------------------------------------------- -module Database.EventStore.Internal.Manager.Subscription - ( module Database.EventStore.Internal.Manager.Subscription.Driver - , Regular(..) - , Persistent(..) - , Catchup(..) - , Subscription - , Running(..) - , Checkpoint(..) - , regularSubscription - , catchupSubscription - , persistentSubscription - , eventArrived - , readNext - , batchRead - , hasCaughtUp - , runningUUID - , runningLastEventNumber - , runningLastCommitPosition - ) where - --------------------------------------------------------------------------------- -import Data.Int - --------------------------------------------------------------------------------- -import ClassyPrelude -import Data.Sequence (ViewL(..), viewl, dropWhileL) - --------------------------------------------------------------------------------- -import Database.EventStore.Internal.Manager.Subscription.Driver -import Database.EventStore.Internal.Manager.Subscription.Model -import Database.EventStore.Internal.Types - --------------------------------------------------------------------------------- --- | Also referred as volatile subscription. For example, if a stream has 100 --- events in it when a subscriber connects, the subscriber can expect to see --- event number 101 onwards until the time the subscription is closed or --- dropped. -data Regular = Regular { _subTos :: Bool } - --------------------------------------------------------------------------------- --- | This kind of subscription specifies a starting point, in the form of an --- event number or transaction file position. The given function will be --- called for events from the starting point until the end of the stream, and --- then for subsequently written events. --- --- For example, if a starting point of 50 is specified when a stream has 100 --- events in it, the subscriber can expect to see events 51 through 100, and --- then any events subsequently written until such time as the subscription is --- dropped or closed. -data Catchup = Catchup - --------------------------------------------------------------------------------- --- | The server remembers the state of the subscription. This allows for many --- different modes of operations compared to a regular or catchup subscription --- where the client holds the subscription state. --- (Need EventStore >= v3.1.0). -data Persistent = Persistent { _perGroup :: Text } - --------------------------------------------------------------------------------- --- | Represents the different type of inputs a subscription state-machine can --- handle. -data Input t a where - -- A event has written to the stream. Subscription state machine should - -- store that event withing its state. - Arrived :: ResolvedEvent -> Input t (Subscription t) - -- The user asks for the next event coming from the server. - ReadNext :: Input t (Maybe ResolvedEvent, Subscription t) - -- A batch read has been made. It's only use for 'Catchup' subscription - -- type. It gives the list of read events and indicates if it reaches the - -- end of the stream along with the next checkpoint to point at. - BatchRead :: [ResolvedEvent] - -> Bool - -> Checkpoint - -> Input Catchup (Subscription Catchup) - -- Used only for 'Catchup' subscription type. Asks if the subscription - -- read every events up to the checkpoint given by the user. - CaughtUp :: Input Catchup Bool - --------------------------------------------------------------------------------- --- | Main subscription state machine. -newtype Subscription t = Subscription (forall a. Input t a -> a) - --------------------------------------------------------------------------------- --- | Submit a new event to the subscription state machine. Internally, --- that event should be stored into the subscription buffer. -eventArrived :: ResolvedEvent -> Subscription t -> Subscription t -eventArrived e (Subscription k) = k (Arrived e) - --------------------------------------------------------------------------------- --- | Reads the next available event. Returns 'Nothing' it there is any. When --- returning an event, it will be removed from the subscription buffer. -readNext :: Subscription t -> (Maybe ResolvedEvent, Subscription t) -readNext (Subscription k) = k ReadNext - --------------------------------------------------------------------------------- --- | Submits a list of events read from a stream. It's only used by a 'Catchup' --- subscription. -batchRead :: [ResolvedEvent] - -> Bool -- ^ If it reaches the end of the stream. - -> Checkpoint - -> Subscription Catchup - -> Subscription Catchup -batchRead es eos nxt (Subscription k) = k (BatchRead es eos nxt) - --------------------------------------------------------------------------------- --- | Indicates if the subscription caught up the end of the stream, meaning the --- subscription is actually live. Only used by 'Catchup' subscription. -hasCaughtUp :: Subscription Catchup -> Bool -hasCaughtUp (Subscription k) = k CaughtUp - --------------------------------------------------------------------------------- --- | Main 'Regular' subscription state machine. -regularSubscription :: Subscription Regular -regularSubscription = baseSubscription - --------------------------------------------------------------------------------- --- | Main 'Persistent' subscription state machine. -persistentSubscription :: Subscription Persistent -persistentSubscription = baseSubscription - --------------------------------------------------------------------------------- --- | Represents the next checkpoint to reach on a catchup subscription. Wheither --- it's a regular stream or the $all stream, it either point to an 'Int32' or --- a 'Position'. -data Checkpoint = CheckpointNumber Int32 | CheckpointPosition Position - --------------------------------------------------------------------------------- --- | Depending either if the subscription concerns a regular stream or $all, --- indicates if an event number (or 'Position') is lesser that the current the --- given 'CheckPoint'. -beforeChk :: Checkpoint -> ResolvedEvent -> Bool -beforeChk (CheckpointNumber num) re = - recordedEventNumber (resolvedEventOriginal re) < num -beforeChk (CheckpointPosition pos) re = - maybe False (< pos) $ resolvedEventPosition re - --------------------------------------------------------------------------------- --- | That subscription state machine accumulates events coming from batch read --- and any real time change made on a stream. That state machine will not --- served any recent change made on the stream until it reaches the end of the --- stream. On every batch read, it makes sure events contained in that batch --- are deleted from the subscription buffer in order to avoid duplicates. That --- implemention has been chosen to avoid potential message lost between the --- moment with reach the end of the stream and the delay required by asking --- for a subscription. -catchupSubscription :: Subscription Catchup -catchupSubscription = Subscription $ catchingUp empty empty - where - catchingUp :: forall a. Seq ResolvedEvent - -> Seq ResolvedEvent - -> Input Catchup a - -> a - catchingUp b s (Arrived e) = - Subscription $ catchingUp b (s `snoc` e) - catchingUp b s ReadNext = - case viewl b of - EmptyL -> (Nothing, Subscription $ catchingUp b s) - e :< rest -> (Just e, Subscription $ catchingUp rest s) - catchingUp b s (BatchRead es eos nxt_pt) = - let nxt_b = foldl' snoc b es - nxt_s = dropWhileL (beforeChk nxt_pt) s - nxt = if eos - then Subscription $ caughtUp nxt_b nxt_s - else Subscription $ catchingUp nxt_b nxt_s in - nxt - catchingUp _ _ CaughtUp = False - - caughtUp :: forall a. Seq ResolvedEvent - -> Seq ResolvedEvent - -> Input Catchup a - -> a - caughtUp b s (Arrived e) = Subscription $ caughtUp b (s `snoc` e) - caughtUp b s ReadNext = - case viewl b of - EmptyL -> live s ReadNext - e :< rest -> - case viewl rest of - EmptyL -> (Just e, Subscription $ live s) - _ -> (Just e, Subscription $ caughtUp rest s) - caughtUp b s (BatchRead _ _ _) = Subscription $ caughtUp b s - caughtUp _ _ CaughtUp = False - - live :: forall a. Seq ResolvedEvent -> Input Catchup a -> a - live s (Arrived e) = Subscription $ live (s `snoc` e) - live s ReadNext = - case viewl s of - EmptyL -> (Nothing, Subscription $ live s) - e :< rest -> (Just e, Subscription $ live rest) - live s (BatchRead _ _ _) = Subscription $ live s - live _ CaughtUp = True - --------------------------------------------------------------------------------- --- | Base subscription used for 'Regular' or 'Persistent' subscription. -baseSubscription :: forall t. Subscription t -baseSubscription = Subscription $ go empty - where - go :: forall a. Seq ResolvedEvent -> Input t a -> a - go s (Arrived e) = Subscription $ go (s `snoc` e) - go s ReadNext = - case viewl s of - EmptyL -> (Nothing, Subscription $ go s) - e :< rest -> (Just e, Subscription $ go rest) - go _ BatchRead{} = error "impossible: base subscription" - go _ CaughtUp = error "impossible: base subscription" diff --git a/Database/EventStore/Internal/Manager/Subscription/Command.hs b/Database/EventStore/Internal/Manager/Subscription/Command.hs new file mode 100644 --- /dev/null +++ b/Database/EventStore/Internal/Manager/Subscription/Command.hs @@ -0,0 +1,103 @@ +-------------------------------------------------------------------------------- +-- | +-- Module : Database.EventStore.Internal.Manager.Subscription.Command +-- Copyright : (C) 2016 Yorick Laupa +-- License : (see the file LICENSE) +-- +-- Maintainer : Yorick Laupa +-- Stability : provisional +-- Portability : non-portable +-- +-------------------------------------------------------------------------------- +module Database.EventStore.Internal.Manager.Subscription.Command where + +-------------------------------------------------------------------------------- +import ClassyPrelude +import Data.ProtocolBuffers +import Data.Serialize + +-------------------------------------------------------------------------------- +import Database.EventStore.Internal.Command +import Database.EventStore.Internal.Manager.Subscription.Message +import Database.EventStore.Internal.Manager.Subscription.Types +import Database.EventStore.Internal.Types + +-------------------------------------------------------------------------------- +data ServerMessage + = EventAppearedMsg !ResolvedEvent + | PersistentEventAppearedMsg !ResolvedEvent + | ConfirmationMsg !Int64 !(Maybe Int32) + | PersistentConfirmationMsg !Text !Int64 !(Maybe Int32) + | PersistentCreatedMsg CreatePersistentSubscriptionResult + | PersistentUpdatedMsg UpdatePersistentSubscriptionResult + | PersistentDeletedMsg DeletePersistentSubscriptionResult + | DroppedMsg SubDropReason + | BadRequestMsg !(Maybe Text) + | NotHandledMsg !NotHandledReason !(Maybe MasterInfo) + | NotAuthenticatedMsg !(Maybe Text) + | UnknownMsg !(Maybe Command) + +-------------------------------------------------------------------------------- +toSubDropReason :: DropReason -> SubDropReason +toSubDropReason D_Unsubscribed = SubUnsubscribed +toSubDropReason D_NotFound = SubNotFound +toSubDropReason D_AccessDenied = SubAccessDenied +toSubDropReason D_PersistentSubscriptionDeleted = SubPersistDeleted +toSubDropReason D_SubscriberMaxCountReached = SubSubscriberMaxCountReached + +-------------------------------------------------------------------------------- +decodeServerMessage :: Package -> ServerMessage +decodeServerMessage pkg = fromMaybe (UnknownMsg $ Just cmd) $ go cmd + where + cmd = packageCmd pkg + go 0xC2 = do + msg <- maybeDecodeMessage $ packageData pkg + let evt = newResolvedEventFromBuf $ getField $ streamResolvedEvent msg + return $ EventAppearedMsg evt + go 0xC7 = do + msg <- maybeDecodeMessage $ packageData pkg + let evt = newResolvedEvent $ getField $ psseaEvt msg + return $ PersistentEventAppearedMsg evt + go 0xC1 = do + msg <- maybeDecodeMessage $ packageData pkg + let lcp = getField $ subscribeLastCommitPos msg + len = getField $ subscribeLastEventNumber msg + return $ ConfirmationMsg lcp len + go 0xC6 = do + msg <- maybeDecodeMessage $ packageData pkg + let lcp = getField $ pscLastCommitPos msg + sid = getField $ pscId msg + len = getField $ pscLastEvtNumber msg + return $ PersistentConfirmationMsg sid lcp len + go 0xC9 = + fmap (PersistentCreatedMsg . getField . cpscResult) + $ maybeDecodeMessage + $ packageData pkg + go 0xCF = + fmap (PersistentUpdatedMsg . getField . upscResult) + $ maybeDecodeMessage + $ packageData pkg + go 0xCB = + fmap (PersistentDeletedMsg . getField . dpscResult) + $ maybeDecodeMessage + $ packageData pkg + go 0xC4 = do + msg <- maybeDecodeMessage $ packageData pkg + let reason = fromMaybe D_Unsubscribed $ getField $ dropReason msg + return $ DroppedMsg $ toSubDropReason reason + go 0xF0 = return $ BadRequestMsg $ packageDataAsText pkg + go 0xF4 = return $ NotAuthenticatedMsg $ packageDataAsText pkg + go 0xF1 = do + msg <- maybeDecodeMessage $ packageData pkg + let info = fmap masterInfo $ getField $ notHandledAdditionalInfo msg + reason = getField $ notHandledReason msg + return $ NotHandledMsg reason info + + go _ = Nothing + +-------------------------------------------------------------------------------- +maybeDecodeMessage :: Decode a => ByteString -> Maybe a +maybeDecodeMessage bytes = + case runGet decodeMessage bytes of + Right a -> Just a + _ -> Nothing diff --git a/Database/EventStore/Internal/Manager/Subscription/Driver.hs b/Database/EventStore/Internal/Manager/Subscription/Driver.hs --- a/Database/EventStore/Internal/Manager/Subscription/Driver.hs +++ b/Database/EventStore/Internal/Manager/Subscription/Driver.hs @@ -1,8 +1,10 @@ -{-# LANGUAGE DeriveDataTypeable #-} -{-# LANGUAGE GADTs #-} -{-# LANGUAGE Rank2Types #-} -{-# LANGUAGE RecordWildCards #-} -{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE GADTs #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE Rank2Types #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE ScopedTypeVariables #-} -------------------------------------------------------------------------------- -- | -- Module : Database.EventStore.Internal.Manager.Subscription.Driver @@ -41,16 +43,17 @@ import Data.Maybe -------------------------------------------------------------------------------- -import ClassyPrelude -import Data.Serialize -import Data.ProtocolBuffers +import ClassyPrelude hiding (group) +import Control.Monad.State import Data.UUID -------------------------------------------------------------------------------- import Database.EventStore.Internal.Generator +import Database.EventStore.Internal.Manager.Subscription.Command import Database.EventStore.Internal.Manager.Subscription.Message import Database.EventStore.Internal.Manager.Subscription.Model import Database.EventStore.Internal.Manager.Subscription.Packages +import Database.EventStore.Internal.Manager.Subscription.Types import Database.EventStore.Internal.Types -------------------------------------------------------------------------------- @@ -63,22 +66,7 @@ | SubConfirmed Running -- ^ Subscription connection is confirmed. It means that subscription can -- receive events from the server. - --------------------------------------------------------------------------------- --- | Indicates why a subscription has been dropped. -data SubDropReason - = SubUnsubscribed - -- ^ Subscription connection has been closed by the user. - | SubAccessDenied - -- ^ The current user is not allowed to operate on the supplied stream. - | SubNotFound - -- ^ Given stream name doesn't exist. - | SubPersistDeleted - -- ^ Given stream is deleted. - | SubAborted - -- ^ Occurs when the user shutdown the connection from the server or if - -- the connection to the server is no longer possible. - deriving (Show, Eq) + | Unsubscribed -------------------------------------------------------------------------------- -- | Enumerates all persistent action exceptions. @@ -245,14 +233,7 @@ updateRException UPS_AccessDenied = Just PersistActionAccessDenied -------------------------------------------------------------------------------- -toSubDropReason :: DropReason -> SubDropReason -toSubDropReason D_Unsubscribed = SubUnsubscribed -toSubDropReason D_NotFound = SubNotFound -toSubDropReason D_AccessDenied = SubAccessDenied -toSubDropReason D_PersistentSubscriptionDeleted = SubPersistDeleted - -------------------------------------------------------------------------------- --------------------------------------------------------------------------------- -- | Type of inputs handled by the 'Subscription' driver. data In r a where -- A command consists of receiving some parameters, updating the @@ -310,9 +291,15 @@ -- confirmed, the driver will return the provided final value. -------------------------------------------------------------------------------- +cmdSubCallback :: Cmd r -> Maybe (SubConnectEvent -> r) +cmdSubCallback (ConnectReg k _ _) = Just k +cmdSubCallback (ConnectPersist k _ _ _) = Just k +cmdSubCallback _ = Nothing + +-------------------------------------------------------------------------------- -- | Driver internal state. -data State r = - State +data Internal r = + Internal { _model :: !Model -- ^ Subscription model. , _gen :: !Generator @@ -323,155 +310,241 @@ } -------------------------------------------------------------------------------- -initState :: Generator -> State r -initState gen = State newModel gen mempty +initInternal :: Generator -> Internal r +initInternal gen = Internal newModel gen mempty -------------------------------------------------------------------------------- -- | Subscription driver state machine. newtype Driver r = Driver (forall a. In r a -> a) -------------------------------------------------------------------------------- --- | Creates a new subscription 'Driver' state machine. -newDriver :: forall r. Settings -> Generator -> Driver r -newDriver setts gen = Driver $ go (initState gen) - where - go :: forall a. State r -> In r a -> a - go st@State{..} (Pkg Package{..}) = do - elm <- lookup packageCorrelation _reg - case packageCmd of - 0xC2 -> do - _ <- querySubscription packageCorrelation _model - msg <- maybeDecodeMessage packageData - let e = getField $ streamResolvedEvent msg - evt = newResolvedEventFromBuf e - app = EventAppeared evt - ConnectReg k _ _ = elm - return (k app, Driver $ go st) +newtype DriverM r m a = DriverM (ReaderT Settings (StateT (Internal r) m) a) + deriving ( Functor + , Applicative + , Monad + , MonadReader Settings + , MonadState (Internal r) + ) - 0xC7 -> do - _ <- querySubscription packageCorrelation _model - msg <- maybeDecodeMessage packageData - let e = getField $ psseaEvt msg - evt = newResolvedEvent e - app = EventAppeared evt - ConnectPersist k _ _ _ = elm - return (k app, Driver $ go st) +-------------------------------------------------------------------------------- +instance MonadTrans (DriverM r) where + lift m = DriverM $ lift $ lift m - 0xC1 -> do - msg <- maybeDecodeMessage packageData - let lcp = getField $ subscribeLastCommitPos msg - len = getField $ subscribeLastEventNumber msg - meta = RegularMeta lcp len - ConnectReg k _ _ = elm - nxt_m = confirmedSubscription packageCorrelation meta _model - run <- querySubscription packageCorrelation nxt_m - let nxt_st = st { _model = nxt_m } - evt = SubConfirmed run - return (k evt, Driver $ go nxt_st) +-------------------------------------------------------------------------------- +noop :: DriverM r Maybe a +noop = lift Nothing - 0xC6 -> do - msg <- maybeDecodeMessage packageData - let lcp = getField $ pscLastCommitPos msg - sid = getField $ pscId msg - len = getField $ pscLastEvtNumber msg - meta = PersistMeta sid lcp len - ConnectPersist k _ _ _ = elm - nxt_m = confirmedSubscription packageCorrelation meta _model - run <- querySubscription packageCorrelation nxt_m - let nxt_st = st { _model = nxt_m } - evt = SubConfirmed run - return (k evt, Driver $ go nxt_st) +-------------------------------------------------------------------------------- +modelSubRunning :: UUID -> DriverM r Maybe Running +modelSubRunning uuid = do + model <- gets _model + lift $ querySubscription uuid $ model - 0xC9 -> confirmPAction elm (getField . cpscResult) createRException - 0xCF -> confirmPAction elm (getField . upscResult) updateRException - 0xCB -> confirmPAction elm (getField . dpscResult) deleteRException +-------------------------------------------------------------------------------- +modelSubConfirmed :: Monad m => UUID -> Meta -> DriverM r m () +modelSubConfirmed uuid meta = do + model <- gets _model + let nxt = confirmedSubscription uuid meta model + modify $ \s -> s { _model = nxt } - 0xC4 -> do - run <- querySubscription packageCorrelation _model - msg <- maybeDecodeMessage packageData - let reason = fromMaybe D_Unsubscribed $ getField - $ dropReason msg - nxt_m = unsubscribed run _model - dreason = toSubDropReason reason - evt = Dropped dreason - nxt_reg = deleteMap packageCorrelation _reg - nxt_st = st { _model = nxt_m - , _reg = nxt_reg } - case elm of - ConnectReg k _ _ -> return (k evt, Driver $ go nxt_st) - ConnectPersist k _ _ _ -> return (k evt, Driver $ go nxt_st) - _ -> Nothing +-------------------------------------------------------------------------------- +modelActionConfirmed :: Monad m => UUID -> DriverM r m () +modelActionConfirmed uuid = + modify $ \s -> s { _model = confirmedAction uuid $ _model s } - _ -> Nothing - where - confirmPAction :: Decode m - => Cmd r - -> (m -> e) - -> (e -> Maybe PersistActionException) - -> Maybe (r, Driver r) - confirmPAction (ApplyPersistAction k g n c) fd em = do - msg <- maybeDecodeMessage packageData - _ <- queryPersistentAction packageCorrelation _model - let nxt_m = confirmedAction packageCorrelation _model - nxt_rg = deleteMap packageCorrelation _reg - nxt_st = st { _model = nxt_m - , _reg = nxt_rg - } - evt = ConfirmedAction packageCorrelation g n c - case em $ fd msg of - Just e -> return (k $ Left e, Driver $ go nxt_st) - Nothing -> return (k $ Right evt, Driver $ go nxt_st) - confirmPAction _ _ _ = Nothing +-------------------------------------------------------------------------------- +modelUnsubscribed :: UUID -> DriverM r Maybe () +modelUnsubscribed uuid = do + run <- modelSubRunning uuid + model <- gets _model + modify $ \s -> s { _model = unsubscribed run model } - go st@State{..} (Cmd cmd) = - case cmd of - ConnectReg _ s tos -> - let (u, nxt_g) = nextUUID _gen - pkg = createConnectRegularPackage setts u s tos - nxt_m = connectReg s tos u _model - nxt_st = st { _model = nxt_m - , _gen = nxt_g - , _reg = insertMap u cmd _reg } in - (pkg, Driver $ go nxt_st) - ConnectPersist _ gn n b -> - let (u, nxt_g) = nextUUID _gen - pkg = createConnectPersistPackage setts u gn n b - nxt_m = connectPersist gn n b u _model - nxt_st = st { _model = nxt_m - , _gen = nxt_g - , _reg = insertMap u cmd _reg } in - (pkg, Driver $ go nxt_st) - Unsubscribe r -> - let pkg = createUnsubscribePackage setts $ runningUUID r in - (pkg, Driver $ go st) - ApplyPersistAction _ gn n a -> - let (u, nxt_g) = nextUUID _gen - pkg = createPersistActionPackage setts u gn n a - nxt_m = persistAction gn n u a _model - nxt_st = st { _model = nxt_m - , _gen = nxt_g - , _reg = insertMap u cmd _reg } in - (pkg, Driver $ go nxt_st) - PersistAck _ run evts -> - let RunningPersist _ _ _ _ sid _ _ = run - u = runningUUID run - pkg = createAckPackage setts u sid evts in - (pkg, Driver $ go st) - PersistNak _ run na r evts -> - let RunningPersist _ _ _ _ sid _ _ = run - u = runningUUID run - pkg = createNakPackage setts u sid na r evts in - (pkg, Driver $ go st) - go st Abort = (fmap snd $ mapToList $ _reg st) >>= _F - where - _F (ConnectReg k _ _) = [k $ Dropped SubAborted] - _F (ConnectPersist k _ _ _) = [k $ Dropped SubAborted] - _F (ApplyPersistAction k _ _ _) = [k $ Left PersistActionAborted] - _F _ = [] +-------------------------------------------------------------------------------- +registerDelete :: Monad m => UUID -> DriverM r m () +registerDelete uuid = do + reg <- gets _reg + let nxtR = deleteMap uuid reg + modify $ \s -> s { _reg = nxtR } -------------------------------------------------------------------------------- -maybeDecodeMessage :: Decode a => ByteString -> Maybe a -maybeDecodeMessage bytes = - case runGet decodeMessage bytes of - Right a -> Just a - _ -> Nothing +registerAdd :: Monad m => UUID -> Cmd r -> DriverM r m () +registerAdd uuid cmd = do + reg <- gets _reg + modify $ \s -> s { _reg = insertMap uuid cmd reg } + +-------------------------------------------------------------------------------- +modelPersistentAction :: UUID -> DriverM r Maybe PendingAction +modelPersistentAction uuid = do + model <- gets _model + lift $ queryPersistentAction uuid model + +-------------------------------------------------------------------------------- +freshUUID :: Monad m => DriverM r m UUID +freshUUID = do + (uuid, nxtG) <- gets (nextUUID . _gen) + modify $ \s -> s { _gen = nxtG } + return uuid + +-------------------------------------------------------------------------------- +modelConnectReg :: Monad m => Text -> Bool -> DriverM r m UUID +modelConnectReg stream tos = do + uuid <- freshUUID + model <- gets _model + modify $ \s -> s { _model = connectReg stream tos uuid model } + return uuid + +-------------------------------------------------------------------------------- +modelConnectPersist :: Monad m => Text -> Text -> Int32 -> DriverM r m UUID +modelConnectPersist group name batch = do + uuid <- freshUUID + model <- gets _model + modify $ \s -> s { _model = connectPersist group name batch uuid model } + return uuid + +-------------------------------------------------------------------------------- +modelPersistAction :: Monad m + => Text + -> Text + -> PersistAction + -> DriverM r m UUID +modelPersistAction group name action = do + uuid <- freshUUID + model <- gets _model + modify $ \s -> s { _model = persistAction group name uuid action model } + return uuid + +-------------------------------------------------------------------------------- +runDriverM :: Monad m + => Settings + -> Internal r + -> DriverM r m a + -> m (a, Driver r) +runDriverM setts st (DriverM m) = do + (a, nxtSt) <- runStateT (runReaderT m setts) st + return (a, Driver $ handleDriver setts nxtSt) + +-------------------------------------------------------------------------------- +runDriver :: Settings + -> Internal r + -> DriverM r Identity a + -> (a, Driver r) +runDriver setts st action = runIdentity $ runDriverM setts st action + +-------------------------------------------------------------------------------- +-- Driver main state machine. +handleDriver :: Settings -> Internal r -> In r a -> a +handleDriver setts st (Pkg pkg) = do + let corrId = packageCorrelation pkg + cmd <- lookup corrId $ _reg st + let action = handleMsg corrId cmd $ decodeServerMessage pkg + runDriverM setts st action +handleDriver setts st (Cmd cmd) = + let action = handleCmd cmd in + runDriver setts st action +handleDriver _ st Abort = (fmap snd $ mapToList $ _reg st) >>= _F + where + _F (ConnectReg k _ _) = [k $ Dropped SubAborted] + _F (ConnectPersist k _ _ _) = [k $ Dropped SubAborted] + _F (ApplyPersistAction k _ _ _) = [k $ Left PersistActionAborted] + _F _ = [] + +-------------------------------------------------------------------------------- +-- | Handles 'Package's coming from the server. +handleMsg :: UUID -> Cmd r -> ServerMessage -> DriverM r Maybe r +handleMsg corrId = go + where + go (ConnectReg k _ _) (EventAppearedMsg evt) = do + _ <- modelSubRunning corrId + return $ k $ EventAppeared evt + go (ConnectPersist k _ _ _) (PersistentEventAppearedMsg evt) = do + _ <- modelSubRunning corrId + return $ k $ EventAppeared evt + go (ConnectReg k _ _) (ConfirmationMsg lcp len) = do + let meta = RegularMeta lcp len + modelSubConfirmed corrId meta + run <- modelSubRunning corrId + return $ k $ SubConfirmed run + go (ConnectPersist k _ _ _) (PersistentConfirmationMsg sid lcp len) = do + let meta = PersistMeta sid lcp len + modelSubConfirmed corrId meta + run <- modelSubRunning corrId + return $ k $ SubConfirmed run + go cmd (PersistentCreatedMsg res) = + confirmPAction cmd $ createRException res + go cmd (PersistentUpdatedMsg res) = + confirmPAction cmd $ updateRException res + go cmd (PersistentDeletedMsg res) = + confirmPAction cmd $ deleteRException res + go cmd (DroppedMsg reason) = do + modelUnsubscribed corrId + registerDelete corrId + let evt = + case reason of + SubUnsubscribed -> Unsubscribed + _ -> Dropped reason + k <- lift $ cmdSubCallback cmd + return $ k evt + go cmd (BadRequestMsg msg) = + go cmd (DroppedMsg $ SubServerError msg) + go cmd (NotAuthenticatedMsg msg) = + go cmd (DroppedMsg $ SubNotAuthenticated msg) + go cmd (NotHandledMsg reason info) = + go cmd (DroppedMsg $ SubNotHandled reason info) + go cmd (UnknownMsg pkgCmdM) = do + k <- lift $ cmdSubCallback cmd + let msgM = fmap (\c -> "unknown command: " <> tshow c) pkgCmdM + return $ k $ Dropped $ SubServerError msgM + go cmd _ = do + k <- lift $ cmdSubCallback cmd + let msg = "Logic error in Subscription Driver (the impossible happened)" + return $ k $ Dropped $ SubClientError msg + + confirmPAction :: Cmd r + -> Maybe PersistActionException + -> DriverM r Maybe r + confirmPAction (ApplyPersistAction k g n c) em = do + _ <- modelPersistentAction corrId + modelActionConfirmed corrId + registerDelete corrId + let evt = ConfirmedAction corrId g n c + case em of + Just e -> return $ k $ Left e + Nothing -> return $ k $ Right evt + confirmPAction _ _ = noop + +-------------------------------------------------------------------------------- +-- | Handles commands coming from the user. +handleCmd :: Monad m => Cmd r -> DriverM r m Package +handleCmd cmd@(ConnectReg _ s tos) = do + setts <- ask + uuid <- modelConnectReg s tos + registerAdd uuid cmd + return $ createConnectRegularPackage setts uuid s tos +handleCmd cmd@(ConnectPersist _ gn n b) = do + setts <- ask + uuid <- modelConnectPersist gn n b + registerAdd uuid cmd + return $ createConnectPersistPackage setts uuid gn n b +handleCmd (Unsubscribe r) = do + setts <- ask + return $ createUnsubscribePackage setts $ runningUUID r +handleCmd cmd@(ApplyPersistAction _ gn n a) = do + setts <- ask + uuid <- modelPersistAction gn n a + registerAdd uuid cmd + return $ createPersistActionPackage setts uuid gn n a +handleCmd (PersistAck _ run evts) = do + setts <- ask + let RunningPersist _ _ _ _ sid _ _ = run + uuid = runningUUID run + return $ createAckPackage setts uuid sid evts +handleCmd (PersistNak _ run na r evts) = do + setts <- ask + let RunningPersist _ _ _ _ sid _ _ = run + uuid = runningUUID run + return $ createNakPackage setts uuid sid na r evts + +-------------------------------------------------------------------------------- +-- | Creates a new subscription 'Driver' state machine. +newDriver :: forall r. Settings -> Generator -> Driver r +newDriver setts gen = Driver $ handleDriver setts (initInternal gen) diff --git a/Database/EventStore/Internal/Manager/Subscription/Message.hs b/Database/EventStore/Internal/Manager/Subscription/Message.hs --- a/Database/EventStore/Internal/Manager/Subscription/Message.hs +++ b/Database/EventStore/Internal/Manager/Subscription/Message.hs @@ -76,6 +76,7 @@ | D_AccessDenied | D_NotFound | D_PersistentSubscriptionDeleted + | D_SubscriberMaxCountReached deriving (Enum, Eq, Show) -------------------------------------------------------------------------------- diff --git a/Database/EventStore/Internal/Manager/Subscription/Types.hs b/Database/EventStore/Internal/Manager/Subscription/Types.hs new file mode 100644 --- /dev/null +++ b/Database/EventStore/Internal/Manager/Subscription/Types.hs @@ -0,0 +1,40 @@ +-------------------------------------------------------------------------------- +-- | +-- Module : Database.EventStore.Internal.Manager.Subscription.Types +-- Copyright : (C) 2016 Yorick Laupa +-- License : (see the file LICENSE) +-- +-- Maintainer : Yorick Laupa +-- Stability : provisional +-- Portability : non-portable +-- +-------------------------------------------------------------------------------- +module Database.EventStore.Internal.Manager.Subscription.Types where + +-------------------------------------------------------------------------------- +import ClassyPrelude + +-------------------------------------------------------------------------------- +import Database.EventStore.Internal.Types + +-------------------------------------------------------------------------------- +-- | Indicates why a subscription has been dropped. +data SubDropReason + = SubUnsubscribed + -- ^ Subscription connection has been closed by the user. + | SubAccessDenied + -- ^ The current user is not allowed to operate on the supplied stream. + | SubNotFound + -- ^ Given stream name doesn't exist. + | SubPersistDeleted + -- ^ Given stream is deleted. + | SubAborted + -- ^ Occurs when the user shutdown the connection from the server or if + -- the connection to the server is no longer possible. + | SubNotAuthenticated (Maybe Text) + | SubServerError (Maybe Text) + -- ^ Unexpected error from the server. + | SubNotHandled !NotHandledReason !(Maybe MasterInfo) + | SubClientError !Text + | SubSubscriberMaxCountReached + deriving (Show, Eq) diff --git a/Database/EventStore/Internal/Operation.hs b/Database/EventStore/Internal/Operation.hs --- a/Database/EventStore/Internal/Operation.hs +++ b/Database/EventStore/Internal/Operation.hs @@ -50,6 +50,7 @@ | ProtobufDecodingError String | ServerError (Maybe Text) -- ^ Reason | InvalidOperation Text + | NotAuthenticatedOp -- ^ Invalid operation state. If happens, it's a driver bug. | Aborted -- ^ Occurs when the user asked to close the connection or if the diff --git a/Database/EventStore/Internal/Operation/Catchup.hs b/Database/EventStore/Internal/Operation/Catchup.hs --- a/Database/EventStore/Internal/Operation/Catchup.hs +++ b/Database/EventStore/Internal/Operation/Catchup.hs @@ -14,7 +14,10 @@ -------------------------------------------------------------------------------- module Database.EventStore.Internal.Operation.Catchup ( CatchupState(..) + , CatchupOpResult(..) + , Checkpoint(..) , catchup + , catchupStreamName ) where -------------------------------------------------------------------------------- @@ -25,7 +28,6 @@ import ClassyPrelude -------------------------------------------------------------------------------- -import Database.EventStore.Internal.Manager.Subscription (Checkpoint(..)) import Database.EventStore.Internal.Operation import Database.EventStore.Internal.Operation.Read.Common import Database.EventStore.Internal.Operation.ReadAllEvents @@ -34,6 +36,12 @@ import Database.EventStore.Internal.Types -------------------------------------------------------------------------------- +-- | Represents the next checkpoint to reach on a catchup subscription. Wheither +-- it's a regular stream or the $all stream, it either point to an 'Int32' or +-- a 'Position'. +data Checkpoint = CheckpointNumber Int32 | CheckpointPosition Position + +-------------------------------------------------------------------------------- defaultBatchSize :: Int32 defaultBatchSize = 500 @@ -52,17 +60,24 @@ -- the $all stream. -------------------------------------------------------------------------------- -streamName :: CatchupState -> Text -streamName (RegularCatchup stream _) = stream -streamName _ = "$all" +catchupStreamName :: CatchupState -> Text +catchupStreamName (RegularCatchup stream _) = stream +catchupStreamName _ = "$all" -------------------------------------------------------------------------------- +data CatchupOpResult = + CatchupOpResult { catchupReadEvents :: ![ResolvedEvent] + , catchupEndOfStream :: !Bool + , catchupCheckpoint :: !Checkpoint + } + +-------------------------------------------------------------------------------- -- | Stream catching up operation. catchup :: Settings -> CatchupState -> Bool -> Maybe Int32 - -> Operation ([ResolvedEvent], Bool, Checkpoint) + -> Operation CatchupOpResult -- ([ResolvedEvent], Bool, Checkpoint) catchup setts init_tpe tos bat_siz = go init_tpe where batch = fromMaybe defaultBatchSize bat_siz @@ -85,15 +100,15 @@ tmp_tpe = AllCatchup nxt_c nxt_p chk = CheckpointPosition $ sliceNext as return (sliceEOS as, sliceEvents as, chk, tmp_tpe) - Left rr -> fromReadResult (streamName tpe) rr $ \as -> + Left rr -> fromReadResult (catchupStreamName tpe) rr $ \as -> let RegularCatchup s _ = tpe nxt = sliceNext as tmp_tpe = RegularCatchup s nxt chk = CheckpointNumber nxt in return (sliceEOS as, sliceEvents as, chk, tmp_tpe) - yield (evts, eos, nchk) - when (not eos) $ go nxt_tpe + yield (CatchupOpResult evts eos nchk) + unless eos $ go nxt_tpe -------------------------------------------------------------------------------- fromReadResult :: Text diff --git a/Database/EventStore/Internal/Operation/StreamMetadata.hs b/Database/EventStore/Internal/Operation/StreamMetadata.hs --- a/Database/EventStore/Internal/Operation/StreamMetadata.hs +++ b/Database/EventStore/Internal/Operation/StreamMetadata.hs @@ -60,7 +60,7 @@ setMetaStream setts s v meta = let stream = metaStream s json = streamMetadataJSON meta - evt = createEvent "$metadata" Nothing (withJson json) + evt = createEvent StreamMetadataType Nothing (withJson json) inner = writeEvents setts stream v [evt] in foreach inner yield diff --git a/Database/EventStore/Internal/Operation/Write/Common.hs b/Database/EventStore/Internal/Operation/Write/Common.hs --- a/Database/EventStore/Internal/Operation/Write/Common.hs +++ b/Database/EventStore/Internal/Operation/Write/Common.hs @@ -44,7 +44,7 @@ evt_data_bytes evt_metadata_bytes where - evt_type = eventType evt + evt_type = eventTypeText $ eventType evt evt_id = eventId evt evt_data_bytes = eventDataBytes $ eventData evt evt_data_type = eventDataType $ eventData evt diff --git a/Database/EventStore/Internal/Processor.hs b/Database/EventStore/Internal/Processor.hs --- a/Database/EventStore/Internal/Processor.hs +++ b/Database/EventStore/Internal/Processor.hs @@ -38,13 +38,15 @@ import Data.UUID -------------------------------------------------------------------------------- +import Database.EventStore.Internal.EndPoint import Database.EventStore.Internal.Generator import Database.EventStore.Internal.Operation hiding (SM(..)) import Database.EventStore.Internal.Types -------------------------------------------------------------------------------- import qualified Database.EventStore.Internal.Manager.Operation.Model as Op -import qualified Database.EventStore.Internal.Manager.Subscription as Sub +import qualified Database.EventStore.Internal.Manager.Subscription.Driver as Sub +import qualified Database.EventStore.Internal.Manager.Subscription.Model as Sub -------------------------------------------------------------------------------- -- | Type of inputs handled by the 'Processor' driver. @@ -217,6 +219,7 @@ -- ^ Indicates to send the given 'Package'. | Await (Processor r) -- ^ Waits for more input. + | ForceReconnectCmd NodeEndPoints (Transition r) -------------------------------------------------------------------------------- -- | Processor state-machine. @@ -230,6 +233,9 @@ Transmit pkg (loopOpTransition st nxt) loopOpTransition st (Op.Await m) = let nxt_st = st { _opModel = m } in Await $ Processor $ execute nxt_st +loopOpTransition st (Op.NotHandled info nxt) = + let node = masterInfoNodeEndPoints info in + ForceReconnectCmd node (loopOpTransition st nxt) -------------------------------------------------------------------------------- abortTransition :: State r -> Op.Transition r -> [r] -> Transition r diff --git a/Database/EventStore/Internal/Subscription.hs b/Database/EventStore/Internal/Subscription.hs new file mode 100644 --- /dev/null +++ b/Database/EventStore/Internal/Subscription.hs @@ -0,0 +1,794 @@ +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE GADTs #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE Rank2Types #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE ScopedTypeVariables #-} +-------------------------------------------------------------------------------- +-- | +-- Module : Database.EventStore.Internal.Subscription +-- Copyright : (C) 2015 Yorick Laupa +-- License : (see the file LICENSE) +-- +-- Maintainer : Yorick Laupa +-- Stability : provisional +-- Portability : non-portable +-- +-- Main subscription state machine declaration module. It also declares every +-- functions required to drive a 'Subscription'. +-------------------------------------------------------------------------------- +module Database.EventStore.Internal.Subscription + ( Regular(..) + , Persistent(..) + , Catchup(..) + , CatchupParams(..) + , Running(..) + , Checkpoint(..) + , Subscription + , SubscriptionId + , SubscriptionClosed(..) + , SubEnv(..) + , PushCmd(..) + , AckCmd(..) + , catchupSub + , regularSub + , persistentSub + , hasCaughtUp + , getSubId + , getSubStream + , unsubscribe + , isSubscribedToAll + , getSubLastCommitPos + , getSubLastEventNumber + , nextEventMaybeSTM + , nextEvent + , nextEventMaybe + , waitConfirmation + , getSubResolveLinkTos + , waitTillCatchup + , hasCaughtUpSTM + , notifyEventsProcessed + , acknowledge + , acknowledgeEvents + , failed + , eventsFailed + , notifyEventsFailed + , unsubscribeConfirmed + , waitUnsubscribeConfirmed + , unsubscribeConfirmedSTM + ) where + +-------------------------------------------------------------------------------- +import Data.Int + +-------------------------------------------------------------------------------- +import ClassyPrelude +import Data.Sequence (ViewL(..), viewl, dropWhileL) +import Data.UUID + +-------------------------------------------------------------------------------- +import Database.EventStore.Internal.EndPoint +import Database.EventStore.Internal.Manager.Subscription.Driver hiding (unsubscribe) +import Database.EventStore.Internal.Manager.Subscription.Model +import Database.EventStore.Internal.Operation +import Database.EventStore.Internal.Operation.Catchup +import Database.EventStore.Internal.Types + +-------------------------------------------------------------------------------- +-- | Also referred as volatile subscription. For example, if a stream has 100 +-- events in it when a subscriber connects, the subscriber can expect to see +-- event number 101 onwards until the time the subscription is closed or +-- dropped. +data Regular = Regular { _subTos :: Bool } + +-------------------------------------------------------------------------------- +-- | This kind of subscription specifies a starting point, in the form of an +-- event number or transaction file position. The given function will be +-- called for events from the starting point until the end of the stream, and +-- then for subsequently written events. +-- +-- For example, if a starting point of 50 is specified when a stream has 100 +-- events in it, the subscriber can expect to see events 51 through 100, and +-- then any events subsequently written until such time as the subscription is +-- dropped or closed. +data Catchup = Catchup + +-------------------------------------------------------------------------------- +-- | The server remembers the state of the subscription. This allows for many +-- different modes of operations compared to a regular or catchup subscription +-- where the client holds the subscription state. +-- (Need EventStore >= v3.1.0). +data Persistent = Persistent { _perGroup :: Text } + +-------------------------------------------------------------------------------- +-- | Represents the different type of inputs a subscription state-machine can +-- handle. +data Input t a where + -- A event has written to the stream. Subscription state machine should + -- store that event withing its state. + Arrived :: ResolvedEvent -> Input t (SubStateMachine t) + -- The user asks for the next event coming from the server. + ReadNext :: Input t (Maybe (ResolvedEvent, SubStateMachine t)) + -- A batch read has been made. It's only use for 'Catchup' subscription + -- type. It gives the list of read events and indicates if it reaches the + -- end of the stream along with the next checkpoint to point at. + BatchRead :: [ResolvedEvent] + -> Bool + -> Checkpoint + -> Input Catchup (SubStateMachine Catchup) + -- Used only for 'Catchup' subscription type. Asks if the subscription + -- read every events up to the checkpoint given by the user. + CaughtUp :: Input Catchup Bool + + -- Returns the last event number read by the user. + LastEventNum :: Input Catchup (Int32, Maybe Position) + +-------------------------------------------------------------------------------- +-- | Main subscription state machine. +newtype SubStateMachine t = SubStateMachine (forall a. Input t a -> a) + +-------------------------------------------------------------------------------- +type PushOp a = (Either OperationError a -> IO ()) -> Operation a -> IO () + +-------------------------------------------------------------------------------- +type PushConnect = (SubConnectEvent -> IO ()) -> PushCmd -> IO () + +-------------------------------------------------------------------------------- +data PushCmd + = PushRegular Text Bool + | PushPersistent Text Text Int32 + +-------------------------------------------------------------------------------- +data AckCmd = AckCmd | NakCmd NakAction (Maybe Text) + +-------------------------------------------------------------------------------- +data SubEnv = + SubEnv { subSettings :: Settings + , subPushOp :: forall a. PushOp a + , subPushConnect :: PushConnect + , subPushUnsub :: Running -> IO () + , subAckCmd :: AckCmd -> Running -> [UUID] -> IO () + , subForceReconnect :: NodeEndPoints -> IO () + } + +-------------------------------------------------------------------------------- +data SubState t + = SubOnline (SubStateMachine t) + | SubDropped SubDropReason + | forall e. Exception e => SubException e + | SubUserUnsubscribed + +-------------------------------------------------------------------------------- +-- | Modifies 'SubState' internal state machine, letting any 'SubDropReason' +-- untouched. +modifySubSM :: (SubStateMachine a -> SubStateMachine a) + -> SubState a + -> SubState a +modifySubSM k (SubOnline sm) = SubOnline (k sm) +modifySubSM _ s = s + +-------------------------------------------------------------------------------- +data CatchupParams = + CatchupParams { catchupResLnkTos :: !Bool + , catchupState :: !CatchupState + , catchupBatchSize :: !(Maybe Int32) + } + +-------------------------------------------------------------------------------- +-- | Represents a subscription life cycle. +data SubLifeCycle a = + SubLifeCycle + { + onConfirm :: Running -> IO () + -- ^ When the server's confirmed this subscription's been created. + , readState :: STM (SubState a) + -- ^ Reads this subscription internal state. + , writeState :: SubState a -> STM () + -- ^ Modifies this subscription internal state. + , onError :: SubDropReason -> IO () + -- ^ When an error's occured. + , onUserUnsubscribed :: IO () + -- ^ When the server confirmed the subscription is no longer live. + -- This action is triggered because the user asks to unsubscribe. + , retrySub :: IO () + -- ^ Retry the all subscription, this behavior is transparent to the user. + } + +-------------------------------------------------------------------------------- +-- | It's possible to subscribe to a stream and be notified when new events are +-- written to that stream. There are three types of subscription which are +-- available, all of which can be useful in different situations. +-- +-- * 'Regular' +-- +-- * 'Catchup' +-- +-- * 'Persistent' +data Subscription t = + Subscription { subStream :: Text + , subLifeCycle :: SubLifeCycle t + , subEnv :: SubEnv + , subRun :: TMVar Running + , subType :: t + } + +-------------------------------------------------------------------------------- +-- | This exception is raised when the user tries to get the next event from a +-- 'Subscription' that is already closed. +data SubscriptionClosed + = SubscriptionClosed SubDropReason + | SubscriptionUnsubscribedByUser + deriving (Show, Typeable) + +-------------------------------------------------------------------------------- +instance Exception SubscriptionClosed + +-------------------------------------------------------------------------------- +catchupSub :: SubEnv -> CatchupParams -> IO (Subscription Catchup) +catchupSub env params = do + mvarRun <- newEmptyTMVarIO + mvarState <- newEmptyTMVarIO + + let streamId = catchupStreamName $ catchupState params + pushCmd = PushRegular streamId (catchupResLnkTos params) + lcycle = + SubLifeCycle + { onConfirm = atomically . putTMVar mvarRun + , readState = readTMVar mvarState + , writeState = \s -> () <$ swapTMVar mvarState s + , onError = \e -> + case e of + SubAborted -> + tryRetryCatcupSubscription pushCmd env mvarState + lcycle params + SubNotHandled reason infoM -> + subNotHandledMsg env lcycle reason infoM + _ -> atomically $ do + s <- takeTMVar mvarState + case s of + SubOnline{} -> putTMVar mvarState $ SubDropped e + _ -> putTMVar mvarState s + , onUserUnsubscribed = atomically $ do + _ <- takeTMVar mvarState + putTMVar mvarState SubUserUnsubscribed + , retrySub = tryRetryCatcupSubscription pushCmd env mvarState + lcycle params + } + + op = createCatchupOperation env params + + subPushOp env (catchupOpEventHandler mvarState) op + subPushConnect env (subEventHandler lcycle) pushCmd + return $ Subscription streamId lcycle env mvarRun Catchup + +-------------------------------------------------------------------------------- +tryRetryCatcupSubscription :: PushCmd + -> SubEnv + -> TMVar (SubState Catchup) + -> SubLifeCycle Catchup + -> CatchupParams + -> IO () +tryRetryCatcupSubscription pushCmd env mvarState lcycle params = do + state <- atomically $ takeTMVar mvarState + + case state of + -- In this case, we do our best to re-engage the + -- catchup subscription where it was at before + -- losing the connection with the server. + SubOnline sm -> do + let (num, posM) = lastEventNumSM sm + newStart = + case catchupState params of + RegularCatchup stream _ -> + RegularCatchup stream num + AllCatchup{} -> + case posM of + Just (Position npc npp) -> + AllCatchup npc npp + _ -> catchupState params + + newParams = params { catchupState = newStart } + + newOp = createCatchupOperation env newParams + + subPushOp env (catchupOpEventHandler mvarState) + newOp + subPushConnect env (subEventHandler lcycle) + pushCmd + _ -> atomically $ putTMVar mvarState state + +-------------------------------------------------------------------------------- +regularSub :: SubEnv -> Text -> Bool -> IO (Subscription Regular) +regularSub env streamId resLnkTos = do + mvarRun <- newEmptyTMVarIO + varState <- newTVarIO $ SubOnline regularSubscription + let lcycle = + SubLifeCycle + { onConfirm = atomically . putTMVar mvarRun + , readState = readTVar varState + , writeState = writeTVar varState + , onError = \r -> atomically $ do + s <- readTVar varState + case s of + SubOnline{} -> writeTVar varState $ SubDropped r + _ -> return () + , onUserUnsubscribed = + atomically $ writeTVar varState SubUserUnsubscribed + , retrySub = do + atomically $ do + state <- readState lcycle + case state of + SubOnline{} -> return () + _ -> writeState lcycle $ SubOnline regularSubscription + subPushConnect env (subEventHandler lcycle) + (PushRegular streamId resLnkTos) + } + + subPushConnect env (subEventHandler lcycle) (PushRegular streamId resLnkTos) + return $ Subscription streamId lcycle env mvarRun (Regular resLnkTos) + +-------------------------------------------------------------------------------- +persistentSub :: SubEnv -> Text -> Text -> Int32 -> IO (Subscription Persistent) +persistentSub env grp stream bufSize = do + mvarRun <- newEmptyTMVarIO + varState <- newTVarIO $ SubOnline persistentSubscription + let lcycle = + SubLifeCycle + { onConfirm = atomically . putTMVar mvarRun + , readState = readTVar varState + , writeState = writeTVar varState + , onError = \r -> atomically $ do + s <- readTVar varState + case s of + SubOnline{} -> writeTVar varState $ SubDropped r + _ -> return () + , onUserUnsubscribed = + atomically $ writeTVar varState SubUserUnsubscribed + , retrySub = do + atomically $ writeState lcycle + $ SubOnline persistentSubscription + subPushConnect env (subEventHandler lcycle) pushCmd + } + + pushCmd = PushPersistent grp stream bufSize + + subPushConnect env (subEventHandler lcycle) pushCmd + return $ Subscription stream lcycle env mvarRun (Persistent grp) + +-------------------------------------------------------------------------------- +subNotHandledMsg :: SubEnv + -> SubLifeCycle s + -> NotHandledReason + -> Maybe MasterInfo + -> IO () +subNotHandledMsg env _ N_NotMaster (Just info) = + subForceReconnect env $ masterInfoNodeEndPoints info +subNotHandledMsg _ lcycle N_NotMaster _ = + atomically $ writeState lcycle + $ SubDropped $ SubServerError (Just msg) + where + msg = "Been asked to connect to new master node \ + \ but no master info been sent." +subNotHandledMsg _ lcycle N_NotReady _ = retrySub lcycle +subNotHandledMsg _ lcycle N_TooBusy _ = retrySub lcycle + +-------------------------------------------------------------------------------- +createCatchupOperation :: SubEnv -> CatchupParams -> Operation CatchupOpResult +createCatchupOperation env params = + catchup (subSettings env) + (catchupState params) + (catchupResLnkTos params) + (catchupBatchSize params) + +-------------------------------------------------------------------------------- +-- | We want to notify the user that something went wrong in the first phase of +-- a catchup subscription (e.g. reading the stream forward until we catchup to +-- stream's end). This prevents a deadlock on user side in case where the user +-- calls `waitTillCatchup` on a stream that doesn't exist. +catchupOpEventHandler :: TMVar (SubState Catchup) + -> Either OperationError CatchupOpResult + -> IO () +catchupOpEventHandler mvarState (Left e) = atomically $ do + isEmpty <- isEmptyTMVar mvarState + if isEmpty + then putTMVar mvarState (SubException e) + else () <$ swapTMVar mvarState (SubException e) +catchupOpEventHandler mvarState (Right res) = atomically $ do + -- When a catchup subscription receives events for the + -- first time. + whenM (isEmptyTMVar mvarState) $ do + let initState = SubOnline catchupSubscription + putTMVar mvarState initState + + subState <- takeTMVar mvarState + let cmd = batchReadSM (catchupReadEvents res) + (catchupEndOfStream res) + (catchupCheckpoint res) + + nxtSubState = modifySubSM cmd subState + + putTMVar mvarState nxtSubState + +-------------------------------------------------------------------------------- +-- | Subscription event handler. Used during a subscription lifetime. +subEventHandler :: SubLifeCycle a -> SubConnectEvent -> IO () +subEventHandler lcycle (SubConfirmed run) = onConfirm lcycle run +subEventHandler lcycle (EventAppeared e) = atomically $ do + st <- readState lcycle + case st of + SubOnline sm -> + writeState lcycle $ SubOnline $ eventArrivedSM e sm + SubException _ -> + -- At this moment [07 October 2016], this can only happen during + -- the first phase of a catchup subscription where the user + -- asked for a subscription on a stream that doesn't exist. + return () + _ -> error "Impossible: subEventHandler" +subEventHandler lcycle Unsubscribed = onUserUnsubscribed lcycle +subEventHandler lcycle (Dropped r) = onError lcycle r + +-------------------------------------------------------------------------------- +-- | Submit a new event to the subscription state machine. Internally, +-- that event should be stored into the subscription buffer. +eventArrivedSM :: ResolvedEvent -> SubStateMachine t -> SubStateMachine t +eventArrivedSM e (SubStateMachine k) = k (Arrived e) + +-------------------------------------------------------------------------------- +-- | Reads the next available event. Returns 'Nothing' it there is any. When +-- returning an event, it will be removed from the subscription buffer. +readNextSM :: SubStateMachine t -> Maybe (ResolvedEvent, SubStateMachine t) +readNextSM (SubStateMachine k) = k ReadNext + +-------------------------------------------------------------------------------- +-- | Submits a list of events read from a stream. It's only used by a 'Catchup' +-- subscription. +batchReadSM :: [ResolvedEvent] + -> Bool -- ^ If it reaches the end of the stream. + -> Checkpoint + -> SubStateMachine Catchup + -> SubStateMachine Catchup +batchReadSM es eos nxt (SubStateMachine k) = k (BatchRead es eos nxt) + +-------------------------------------------------------------------------------- +-- | Indicates if the subscription caught up the end of the stream, meaning the +-- subscription is actually live. Only used by 'Catchup' subscription. +hasCaughtUpSM :: SubStateMachine Catchup -> Bool +hasCaughtUpSM (SubStateMachine k) = k CaughtUp + +-------------------------------------------------------------------------------- +-- | Last event number read by the user. +lastEventNumSM :: SubStateMachine Catchup -> (Int32, Maybe Position) +lastEventNumSM (SubStateMachine k) = k LastEventNum + +-------------------------------------------------------------------------------- +-- | Main 'Regular' subscription state machine. +regularSubscription :: SubStateMachine Regular +regularSubscription = baseSubStateMachine + +-------------------------------------------------------------------------------- +-- | Main 'Persistent' subscription state machine. +persistentSubscription :: SubStateMachine Persistent +persistentSubscription = baseSubStateMachine + +-------------------------------------------------------------------------------- +-- | Depending either if the subscription concerns a regular stream or $all, +-- indicates if an event number (or 'Position') is lesser that the current the +-- given 'CheckPoint'. +beforeChk :: Checkpoint -> ResolvedEvent -> Bool +beforeChk (CheckpointNumber num) re = + recordedEventNumber (resolvedEventOriginal re) < num +beforeChk (CheckpointPosition pos) re = + maybe False (< pos) $ resolvedEventPosition re + +-------------------------------------------------------------------------------- +-- | This data structure is only used by catchup subscription state machine. +data CatchupSMState = + CatchupSMState { csmReadSeq :: !(Seq ResolvedEvent) + -- ^ This sequence is used to pack events coming from reading + -- a stream forward. + , csmLiveSeq :: !(Seq ResolvedEvent) + -- ^ This sequence is used to pack events coming from live + -- subscription. + , csmLastNum :: !(Maybe Int32) + -- ^ Tracks the last event read. + , csmLastPos :: !(Maybe Position) + } + +-------------------------------------------------------------------------------- +initialCatchupSMState :: CatchupSMState +initialCatchupSMState = CatchupSMState empty empty Nothing Nothing + +-------------------------------------------------------------------------------- +insertReadEvents :: [ResolvedEvent] + -> Checkpoint + -> CatchupSMState + -> CatchupSMState +insertReadEvents es chp s = + s { csmReadSeq = foldl' snoc (csmReadSeq s) es + , csmLiveSeq = dropWhileL (beforeChk chp) (csmLiveSeq s) + } + +-------------------------------------------------------------------------------- +insertLiveEvent :: ResolvedEvent -> CatchupSMState -> CatchupSMState +insertLiveEvent e s = s { csmLiveSeq = csmLiveSeq s `snoc` e } + +-------------------------------------------------------------------------------- +readNextFromBatchSeq :: CatchupSMState -> Maybe (ResolvedEvent, CatchupSMState) +readNextFromBatchSeq s = + case viewl $ csmReadSeq s of + EmptyL -> Nothing + e :< rest -> + let newLast = recordedEventNumber $ resolvedEventOriginal e + nxtS = s { csmReadSeq = rest + , csmLastNum = Just newLast + , csmLastPos = resolvedEventPosition e + } in + Just (e, nxtS) + +-------------------------------------------------------------------------------- +readNextFromLiveSeq :: CatchupSMState -> Maybe (ResolvedEvent, CatchupSMState) +readNextFromLiveSeq s = + case viewl $ csmLiveSeq s of + EmptyL -> Nothing + e :< rest -> + let newLast = recordedEventNumber $ resolvedEventOriginal e + nxtS = s { csmLiveSeq = rest + , csmLastNum = Just newLast + , csmLastPos = resolvedEventPosition e + } in + Just (e, nxtS) + +-------------------------------------------------------------------------------- +lastEventNumber :: CatchupSMState -> (Int32, Maybe Position) +lastEventNumber s = (fromMaybe 0 $ csmLastNum s, csmLastPos s) + +-------------------------------------------------------------------------------- +isBatchReqEmpty :: CatchupSMState -> Bool +isBatchReqEmpty s = + case viewl $ csmReadSeq s of + EmptyL -> True + _ -> False + +-------------------------------------------------------------------------------- +-- | That subscription state machine accumulates events coming from batch read +-- and any real time change made on a stream. That state machine will not +-- served any recent change made on the stream until it reaches the end of the +-- stream. On every batch read, it makes sure events contained in that batch +-- are deleted from the subscription buffer in order to avoid duplicates. That +-- implemention has been chosen to avoid potential message lost between the +-- moment with reach the end of the stream and the delay required by asking +-- for a subscription. +catchupSubscription :: SubStateMachine Catchup +catchupSubscription = SubStateMachine $ catchingUp initialCatchupSMState + where + catchingUp :: forall a. CatchupSMState -> Input Catchup a -> a + catchingUp s (Arrived e) = + SubStateMachine $ catchingUp $ insertLiveEvent e s + catchingUp s ReadNext = + let _F (e, sm) = (e, SubStateMachine $ catchingUp sm) in + fmap _F $ readNextFromBatchSeq s + catchingUp s (BatchRead es eos chk) = + let nxtS = insertReadEvents es chk s in + SubStateMachine $ + if eos + then caughtUp nxtS + else catchingUp nxtS + catchingUp _ CaughtUp = False + catchingUp s LastEventNum = lastEventNumber s + + caughtUp :: forall a. CatchupSMState -> Input Catchup a -> a + caughtUp s (Arrived e) = SubStateMachine $ caughtUp $ insertLiveEvent e s + caughtUp s ReadNext = + case readNextFromBatchSeq s of + Nothing -> live s ReadNext + Just (e, nxtS) -> + if isBatchReqEmpty nxtS + then Just (e, SubStateMachine $ live s) + else Just (e, SubStateMachine $ caughtUp nxtS) + caughtUp s BatchRead{} = SubStateMachine $ caughtUp s + caughtUp _ CaughtUp = False + caughtUp s LastEventNum = lastEventNumber s + + live :: forall a. CatchupSMState -> Input Catchup a -> a + live s (Arrived e) = SubStateMachine $ live $ insertLiveEvent e s + live s ReadNext = + let _F (e, sm) = (e, SubStateMachine $ live sm) in + fmap _F $ readNextFromLiveSeq s + live s BatchRead{} = SubStateMachine $ live s + live _ CaughtUp = True + live s LastEventNum = lastEventNumber s + +-------------------------------------------------------------------------------- +-- | Base subscription used for 'Regular' or 'Persistent' subscription. +baseSubStateMachine :: forall t. SubStateMachine t +baseSubStateMachine = SubStateMachine $ go empty + where + go :: forall a. Seq ResolvedEvent -> Input t a -> a + go s (Arrived e) = SubStateMachine $ go (s `snoc` e) + go s ReadNext = + case viewl s of + EmptyL -> Nothing + e :< rest -> Just (e, SubStateMachine $ go rest) + go _ _ = error "impossible: base subscription" + +-------------------------------------------------------------------------------- +-- Subscription API +-------------------------------------------------------------------------------- +-- | Represents a subscription id. +newtype SubscriptionId = SubId UUID deriving (Eq, Ord, Show) + +-------------------------------------------------------------------------------- +-- | Gets the ID of the subscription. +getSubId :: Subscription a -> IO SubscriptionId +getSubId Subscription{..} = atomically $ do + run <- readTMVar subRun + return $ SubId $ runningUUID run + +-------------------------------------------------------------------------------- +-- | Gets the subscription stream name. +getSubStream :: Subscription a -> Text +getSubStream = subStream + +-------------------------------------------------------------------------------- +-- | Asynchronously unsubscribe from the the stream. +unsubscribe :: Subscription a -> IO () +unsubscribe Subscription{..} = do + run <- atomically $ readTMVar subRun + subPushUnsub subEnv run + +-------------------------------------------------------------------------------- +-- | If the subscription is on the $all stream. +isSubscribedToAll :: Subscription a -> Bool +isSubscribedToAll = (== "") . getSubStream + +-------------------------------------------------------------------------------- +-- | The last commit position seen on the subscription (if this a subscription +-- to $all stream). +getSubLastCommitPos :: Subscription a -> IO Int64 +getSubLastCommitPos Subscription{..} = atomically $ do + run <- readTMVar subRun + return $ runningLastCommitPosition run + +-------------------------------------------------------------------------------- +-- | The last event number seen on the subscription (if this is a subscription +-- to a single stream). +getSubLastEventNumber :: Subscription a -> IO (Maybe Int32) +getSubLastEventNumber Subscription{..} = atomically $ do + run <- readTMVar subRun + return $ runningLastEventNumber run + +-------------------------------------------------------------------------------- +-- | Asks for the next incoming event like 'nextEventMaybe' while still being +-- in the the 'STM'. +nextEventMaybeSTM :: Subscription a -> STM (Maybe ResolvedEvent) +nextEventMaybeSTM Subscription{..} = do + st <- readState subLifeCycle + case st of + SubException e -> throwSTM e + SubDropped r -> throwSTM $ SubscriptionClosed r + SubOnline sub -> do + case readNextSM sub of + Just (e, nxt) -> + Just e <$ writeState subLifeCycle (SubOnline nxt) + _ -> return Nothing + SubUserUnsubscribed -> throwSTM SubscriptionUnsubscribedByUser + +-------------------------------------------------------------------------------- +-- | Awaits for the next event. +nextEvent :: Subscription a -> IO ResolvedEvent +nextEvent sub = atomically $ do + m <- nextEventMaybeSTM sub + case m of + Nothing -> retrySTM + Just e -> return e + +-------------------------------------------------------------------------------- +-- | Non blocking version of 'nextEvent'. +nextEventMaybe :: Subscription a -> IO (Maybe ResolvedEvent) +nextEventMaybe = atomically . nextEventMaybeSTM + +-------------------------------------------------------------------------------- +-- | Waits until the `Subscription` has been confirmed. +waitConfirmation :: Subscription a -> IO () +waitConfirmation s = atomically $ do + _ <- readTMVar $ subRun s + return () + +-------------------------------------------------------------------------------- +-- | Determines whether or not any link events encontered in the stream will be +-- resolved. +getSubResolveLinkTos :: Subscription Regular -> Bool +getSubResolveLinkTos = _subTos . subType + +-------------------------------------------------------------------------------- +-- | Non blocking version of `waitTillCatchup`. +hasCaughtUp :: Subscription Catchup -> IO Bool +hasCaughtUp sub = atomically $ hasCaughtUpSTM sub + +-------------------------------------------------------------------------------- +-- | Waits until 'CatchupSubscription' subscription catch-up its stream. +waitTillCatchup :: Subscription Catchup -> IO () +waitTillCatchup sub = atomically $ + unlessM (hasCaughtUpSTM sub) retrySTM + +-------------------------------------------------------------------------------- +-- | Like 'hasCaughtUp' but lives in 'STM' monad. +hasCaughtUpSTM :: Subscription Catchup -> STM Bool +hasCaughtUpSTM Subscription{..} = do + res <- readState subLifeCycle + case res of + SubOnline sm -> return $ hasCaughtUpSM sm + SubException e -> throwSTM e + SubDropped r -> throwSTM $ SubscriptionClosed r + SubUserUnsubscribed -> throwSTM SubscriptionUnsubscribedByUser + +-------------------------------------------------------------------------------- +-- | Like 'unsubscribeConfirmed' but lives in 'STM' monad. +unsubscribeConfirmedSTM :: Subscription a -> STM Bool +unsubscribeConfirmedSTM Subscription{..} = do + res <- readState subLifeCycle + case res of + SubOnline _ -> return False + SubException e -> throwSTM e + SubDropped r -> throwSTM $ SubscriptionClosed r + SubUserUnsubscribed -> return True + +-------------------------------------------------------------------------------- +-- | Non blocking version of `waitUnsubscribeConfirmed`. +unsubscribeConfirmed :: Subscription a -> IO Bool +unsubscribeConfirmed = atomically . unsubscribeConfirmedSTM + +-------------------------------------------------------------------------------- +-- | Wait until unsubscription has been confirmed by the server. +waitUnsubscribeConfirmed :: Subscription a -> IO () +waitUnsubscribeConfirmed sub = atomically $ + unlessM (unsubscribeConfirmedSTM sub) retrySTM + +-------------------------------------------------------------------------------- +-- | Acknowledges those event ids have been successfully processed. +notifyEventsProcessed :: Subscription Persistent -> [UUID] -> IO () +notifyEventsProcessed Subscription{..} evts = do + run <- atomically $ readTMVar subRun + subAckCmd subEnv AckCmd run evts + +-------------------------------------------------------------------------------- +-- | Acknowledges that 'ResolvedEvent' has been successfully processed. +acknowledge :: Subscription Persistent -> ResolvedEvent -> IO () +acknowledge sub e = notifyEventsProcessed sub [resolvedEventOriginalId e] + +-------------------------------------------------------------------------------- +-- | Acknowledges those 'ResolvedEvent's have been successfully processed. +acknowledgeEvents :: Subscription Persistent -> [ResolvedEvent] -> IO () +acknowledgeEvents sub = notifyEventsProcessed sub . fmap resolvedEventOriginalId + +-------------------------------------------------------------------------------- +-- | Mark a message that has failed processing. The server will take action +-- based upon the action parameter. +failed :: Subscription Persistent + -> ResolvedEvent + -> NakAction + -> Maybe Text + -> IO () +failed sub e a r = notifyEventsFailed sub a r [resolvedEventOriginalId e] + +-------------------------------------------------------------------------------- +-- | Mark messages that have failed processing. The server will take action +-- based upon the action parameter. +eventsFailed :: Subscription Persistent + -> [ResolvedEvent] + -> NakAction + -> Maybe Text + -> IO () +eventsFailed sub evts a r = + notifyEventsFailed sub a r $ fmap resolvedEventOriginalId evts + +-------------------------------------------------------------------------------- +-- | Acknowledges those event ids have failed to be processed successfully. +notifyEventsFailed :: Subscription Persistent + -> NakAction + -> Maybe Text + -> [UUID] + -> IO () +notifyEventsFailed Subscription{..} act res evts = do + run <- atomically $ readTMVar subRun + subAckCmd subEnv (NakCmd act res) run evts diff --git a/Database/EventStore/Internal/Types.hs b/Database/EventStore/Internal/Types.hs --- a/Database/EventStore/Internal/Types.hs +++ b/Database/EventStore/Internal/Types.hs @@ -36,6 +36,7 @@ -------------------------------------------------------------------------------- import Database.EventStore.Internal.Command +import Database.EventStore.Internal.EndPoint import Database.EventStore.Logging -------------------------------------------------------------------------------- @@ -57,18 +58,61 @@ -------------------------------------------------------------------------------- -- Event -------------------------------------------------------------------------------- +-- | Constants for System event types. +data EventType + = StreamDeletedType + -- ^ Event type for stream deleted. + | StatsCollectionType + -- ^ Event type for statistics. + | LinkToType + -- ^ Event type for linkTo. + | StreamMetadataType + -- ^ Event type for stream metadata. + | SettingsType + -- ^ Event type for the system settings. + | UserDefined Text + -- ^ Event defined by the user. + deriving Eq + +-------------------------------------------------------------------------------- +instance IsString EventType where + fromString = eventTypeFromStr + +-------------------------------------------------------------------------------- +instance Show EventType where + show = unpack . eventTypeText + +-------------------------------------------------------------------------------- +eventTypeText :: EventType -> Text +eventTypeText StreamDeletedType = "$streamDeleted" +eventTypeText StatsCollectionType = "$statsCollected" +eventTypeText LinkToType = "$>" +eventTypeText StreamMetadataType = "$metadata" +eventTypeText SettingsType = "$settings" +eventTypeText (UserDefined t) = t + +-------------------------------------------------------------------------------- +eventTypeFromStr :: String -> EventType +eventTypeFromStr "$streamDeleted" = StreamDeletedType +eventTypeFromStr "$statsCollected" = StatsCollectionType +eventTypeFromStr "$>" = LinkToType +eventTypeFromStr "$metadata" = StreamMetadataType +eventTypeFromStr "$settings" = SettingsType +eventTypeFromStr t = UserDefined $ pack t + +-------------------------------------------------------------------------------- -- | Contains event information like its type and data. Only used for write -- queries. data Event = Event - { eventType :: !Text + { eventType :: !EventType , eventId :: !(Maybe UUID) , eventData :: !EventData } deriving (Eq, Show) -------------------------------------------------------------------------------- -- | Create an 'Event' meant to be persisted. -createEvent :: Text -- ^ Event type +createEvent :: EventType -- ^ Event type -> Maybe UUID -- ^ Event ID, generated if 'Nothing' -> EventData -- ^ Event data -> Event @@ -259,6 +303,78 @@ instance Decode ResolvedIndexedEvent -------------------------------------------------------------------------------- +data NotHandledReason + = N_NotReady + | N_TooBusy + | N_NotMaster + deriving (Enum, Eq, Show) + +-------------------------------------------------------------------------------- +data MasterInfoBuf + = MasterInfoBuf + { bufMasterExternalTcpAddr :: Required 1 (Value Text) + , bufMasterExternalTcpPort :: Required 2 (Value Int32) + , bufMasterExternalHttpAddr :: Required 3 (Value Text) + , bufMasterExternalHttpPort :: Required 4 (Value Int32) + , bufMasterExternalSecureTcpAddr :: Optional 5 (Value Text) + , bufMasterExternalSecureTcpPort :: Optional 6 (Value Int32) + } deriving (Generic, Show) + +-------------------------------------------------------------------------------- +instance Decode MasterInfoBuf + +-------------------------------------------------------------------------------- +data MasterInfo + = MasterInfo + { masterExternalTcpAddr :: !String + , masterExternalTcpPort :: !Int + , masterExternalHttpAddr :: !String + , masterExternalHttpPort :: !Int + , masterExternalSecureTcpAddr :: !(Maybe String) + , masterExternalSecureTcpPort :: !(Maybe Int) + } deriving (Show, Eq) + +-------------------------------------------------------------------------------- +masterInfo :: MasterInfoBuf -> MasterInfo +masterInfo buf = + MasterInfo + { masterExternalTcpAddr = + unpack $ getField $ bufMasterExternalTcpAddr buf + , masterExternalTcpPort = + fromIntegral $ getField $ bufMasterExternalTcpPort buf + , masterExternalHttpAddr = + unpack $ getField $ bufMasterExternalHttpAddr buf + , masterExternalHttpPort = + fromIntegral $ getField $ bufMasterExternalHttpPort buf + , masterExternalSecureTcpAddr = + fmap unpack $ getField $ bufMasterExternalSecureTcpAddr buf + , masterExternalSecureTcpPort = + fmap fromIntegral $ getField $ bufMasterExternalSecureTcpPort buf + } + +-------------------------------------------------------------------------------- +masterInfoNodeEndPoints :: MasterInfo -> NodeEndPoints +masterInfoNodeEndPoints info = + NodeEndPoints + (EndPoint (masterExternalTcpAddr info) + (fromIntegral $ masterExternalTcpPort info)) + securePoint + where + securePoint = + EndPoint <$> masterExternalSecureTcpAddr info + <*> (fmap fromIntegral $ masterExternalSecureTcpPort info) + +-------------------------------------------------------------------------------- +data NotHandledBuf + = NotHandledBuf + { notHandledReason :: Required 1 (Enumeration NotHandledReason) + , notHandledAdditionalInfo :: Optional 2 (Message MasterInfoBuf) + } deriving (Generic, Show) + +-------------------------------------------------------------------------------- +instance Decode NotHandledBuf + +-------------------------------------------------------------------------------- -- | Represents a serialized event sent by the server in a subscription context. data ResolvedEventBuf = ResolvedEventBuf @@ -489,6 +605,13 @@ , packageCred :: !(Maybe Credentials) } deriving Show + +-------------------------------------------------------------------------------- +packageDataAsText :: Package -> Maybe Text +packageDataAsText = go . decodeUtf8 . packageData + where + go "" = Nothing + go t = Just t -------------------------------------------------------------------------------- -- | Constructs a heartbeat response given the 'UUID' of heartbeat request. diff --git a/Database/EventStore/Internal/Utils.hs b/Database/EventStore/Internal/Utils.hs new file mode 100644 --- /dev/null +++ b/Database/EventStore/Internal/Utils.hs @@ -0,0 +1,27 @@ +-------------------------------------------------------------------------------- +-- | +-- Module : Database.EventStore.Internal.Utils +-- Copyright : (C) 2016 Yorick Laupa +-- License : (see the file LICENSE) +-- +-- Maintainer : Yorick Laupa +-- Stability : provisional +-- Portability : non-portable +-- +-------------------------------------------------------------------------------- +module Database.EventStore.Internal.Utils (prettyWord8) where + +-------------------------------------------------------------------------------- +import Numeric + +-------------------------------------------------------------------------------- +import ClassyPrelude + +-------------------------------------------------------------------------------- +prettyWord8 :: Word8 -> String +prettyWord8 w = "0x" ++ padding (showHex w "") + +-------------------------------------------------------------------------------- +padding :: String -> String +padding [x] = ['0',x] +padding xs = xs diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -5,6 +5,7 @@ [![Build Status](https://travis-ci.org/YoEight/eventstore.svg?branch=master)](https://travis-ci.org/YoEight/eventstore) That driver supports 100% of EventStore features ! +More information about the GetEventStore database can be found there: https://geteventstore.com/ Requirements ============ diff --git a/eventstore.cabal b/eventstore.cabal --- a/eventstore.cabal +++ b/eventstore.cabal @@ -8,9 +8,9 @@ -- for standards guiding when and how versions should be incremented. -- http://www.haskell.org/haskellwiki/Package_versioning_policy -- PVP summary: +-+------- breaking API changes --- | | +----- non-breaking API additions --- | | | +--- code changes with no API change -version: 0.13.1.7 +-- | | +----- non-breaking API additions +-- | | | +--- code changes with no API change +version: 0.14.0.0 tested-with: GHC >= 7.8.3 && < 7.11 @@ -65,18 +65,22 @@ other-modules: Database.EventStore.Internal.Command Database.EventStore.Internal.Connection Database.EventStore.Internal.Discovery + Database.EventStore.Internal.EndPoint Database.EventStore.Internal.Execution.Production Database.EventStore.Internal.Generator Database.EventStore.Internal.Operation Database.EventStore.Internal.Processor Database.EventStore.Internal.Stream + Database.EventStore.Internal.Subscription Database.EventStore.Internal.Types + Database.EventStore.Internal.Utils Database.EventStore.Internal.Manager.Operation.Model - Database.EventStore.Internal.Manager.Subscription + Database.EventStore.Internal.Manager.Subscription.Command Database.EventStore.Internal.Manager.Subscription.Driver Database.EventStore.Internal.Manager.Subscription.Message Database.EventStore.Internal.Manager.Subscription.Model Database.EventStore.Internal.Manager.Subscription.Packages + Database.EventStore.Internal.Manager.Subscription.Types Database.EventStore.Internal.Operations Database.EventStore.Internal.Operation.Catchup Database.EventStore.Internal.Operation.DeleteStream @@ -118,6 +122,7 @@ , dotnet-timespan , connection ==0.2.* , classy-prelude ==1.* + , mtl -- Directories containing source files. -- hs-source-dirs: diff --git a/tests/Tests.hs b/tests/Tests.hs --- a/tests/Tests.hs +++ b/tests/Tests.hs @@ -237,6 +237,7 @@ loop alljss unsubscribe sub + waitUnsubscribeConfirmed sub let action = do _ <- nextEvent sub return False