diff --git a/CHANGELOG.markdown b/CHANGELOG.markdown --- a/CHANGELOG.markdown +++ b/CHANGELOG.markdown @@ -1,3 +1,10 @@ +1.4.0 +----- +* Internal: Refactor operation and registry tcp internals. +* Introduces `MonitoringBackend` to stop relying exclusively on `ekg-core`. +* Supports GHC 8.8.*. +* Exposes a stream processing interface for subscriptions. + 1.3.3 ----- * Add `Pinned` system consumer strategy. diff --git a/Database/EventStore.hs b/Database/EventStore.hs --- a/Database/EventStore.hs +++ b/Database/EventStore.hs @@ -52,8 +52,6 @@ , eventNumber , rawEventNumber , eventNumberToInt64 - -- * Common Operation types - , OperationMaxAttemptReached(..) -- * Read Operations , StreamMetadataResult(..) , BatchResult @@ -110,30 +108,21 @@ , transactionWrite -- * Subscription , SubscriptionClosed(..) - , SubscriptionId , SubscriptionStream(..) , Subscription , SubDropReason(..) , SubDetails - , waitConfirmation - , unsubscribeConfirmed - , unsubscribeConfirmedSTM - , waitUnsubscribeConfirmed - , nextEventMaybeSTM - , getSubscriptionDetailsSTM + , SubAction(..) , unsubscribe + , nextSubEvent + , streamSubEvents + , streamSubResolvedEvents -- * Volatile Subscription , RegularSubscription , subscribe - , getSubscriptionId - , nextEvent - , nextEventMaybe -- * Catch-up Subscription , CatchupSubscription , subscribeFrom - , waitTillCatchup - , hasCaughtUp - , hasCaughtUpSTM -- * Persistent Subscription , PersistentSubscription , PersistentSubscriptionSettings(..) @@ -184,6 +173,9 @@ , LogLevel(..) , LogType(..) , LoggerFilter(..) + -- * Monitoring + , MonitoringBackend(..) + , noopMonitoringBackend -- * Misc , Command , DropReason(..) @@ -226,5 +218,4 @@ import Database.EventStore.Internal.Subscription.Persistent import Database.EventStore.Internal.Subscription.Types import Database.EventStore.Internal.Subscription.Regular -import Database.EventStore.Internal.Manager.Operation.Registry import Database.EventStore.Internal.Types diff --git a/Database/EventStore/Internal.hs b/Database/EventStore/Internal.hs --- a/Database/EventStore/Internal.hs +++ b/Database/EventStore/Internal.hs @@ -23,7 +23,6 @@ import Data.Maybe -------------------------------------------------------------------------------- -import Database.EventStore.Internal.Callback import Database.EventStore.Internal.Communication import Database.EventStore.Internal.Connection (connectionBuilder) import Database.EventStore.Internal.Control hiding (subscribe) @@ -121,11 +120,8 @@ -> [Event] -> Maybe Credentials -> IO (Async WriteResult) -sendEvents Connection{..} evt_stream exp_ver evts cred = do - p <- newPromise - let op = Op.writeEvents _settings (streamIdRaw evt_stream) exp_ver cred evts - publishWith _exec (SubmitOperation p op) - async (retrieve p) +sendEvents Connection{..} evt_stream exp_ver evts cred = + Op.writeEvents _settings _exec (streamIdRaw evt_stream) exp_ver cred evts -------------------------------------------------------------------------------- -- | Deletes given stream. @@ -135,11 +131,8 @@ -> Maybe Bool -- ^ Hard delete -> Maybe Credentials -> IO (Async Op.DeleteResult) -deleteStream Connection{..} evt_stream exp_ver hard_del cred = do - p <- newPromise - let op = Op.deleteStream _settings (streamIdRaw evt_stream) exp_ver hard_del cred - publishWith _exec (SubmitOperation p op) - async (retrieve p) +deleteStream Connection{..} stream expVer hardDel cred = + Op.deleteStream _settings _exec (streamIdRaw stream) expVer hardDel cred -------------------------------------------------------------------------------- -- | Represents a multi-request transaction with the EventStore. @@ -170,11 +163,9 @@ -> Maybe Credentials -> IO (Async Transaction) startTransaction conn@Connection{..} evt_stream exp_ver cred = do - p <- newPromise - let op = Op.transactionStart _settings (streamIdRaw evt_stream) exp_ver cred - publishWith _exec (SubmitOperation p op) + as <- Op.transactionStart _settings _exec (streamIdRaw evt_stream) exp_ver cred async $ do - tid <- retrieve p + tid <- wait as return Transaction { _tStream = streamIdRaw evt_stream , _tTransId = TransactionId tid @@ -189,23 +180,17 @@ -> Maybe Credentials -> IO (Async ()) transactionWrite Transaction{..} evts cred = do - p <- newPromise let Connection{..} = _tConn raw_id = _unTransId _tTransId - op = Op.transactionWrite _settings _tStream _tExpVer raw_id evts cred - publishWith _exec (SubmitOperation p op) - async (retrieve p) + Op.transactionWrite _settings _exec _tStream _tExpVer raw_id evts cred -------------------------------------------------------------------------------- -- | Asynchronously commits this transaction. transactionCommit :: Transaction -> Maybe Credentials -> IO (Async WriteResult) transactionCommit Transaction{..} cred = do - p <- newPromise let Connection{..} = _tConn raw_id = _unTransId _tTransId - op = Op.transactionCommit _settings _tStream _tExpVer raw_id cred - publishWith _exec (SubmitOperation p op) - async (retrieve p) + Op.transactionCommit _settings _exec _tStream _tExpVer raw_id cred -------------------------------------------------------------------------------- -- | There isn't such of thing in EventStore parlance. Basically, if you want to @@ -221,13 +206,10 @@ -> ResolveLink -> Maybe Credentials -> IO (Async (ReadResult EventNumber Op.ReadEvent)) -readEvent Connection{..} stream_id evtNum resLinkTos cred = do - p <- newPromise - let evt_num = eventNumberToInt64 evtNum - res_link_tos = resolveLinkToBool resLinkTos - op = Op.readEvent _settings (streamIdRaw stream_id) evt_num res_link_tos cred - publishWith _exec (SubmitOperation p op) - async (retrieve p) +readEvent Connection{..} stream evtNum resLinkTos cred = do + let evtNumRaw = eventNumberToInt64 evtNum + linkTos = resolveLinkToBool resLinkTos + Op.readEvent _settings _exec (streamIdRaw stream) evtNumRaw linkTos cred -------------------------------------------------------------------------------- -- | When batch-reading a stream, this type-level function maps the result you @@ -271,20 +253,15 @@ -> Maybe Credentials -> IO (Async (BatchResult t)) readEventsCommon Connection{..} dir streamId start cnt resLinkTos cred = do - p <- newPromise let res_link_tos = resolveLinkToBool resLinkTos - op = - case streamId of - StreamName{} -> - let name = streamIdRaw streamId - evtNum = eventNumberToInt64 start in - Op.readStreamEvents _settings dir name evtNum cnt res_link_tos cred - All -> - let Position c_pos p_pos = start in - Op.readAllEvents _settings c_pos p_pos cnt res_link_tos dir cred - - publishWith _exec (SubmitOperation p op) - async (retrieve p) + case streamId of + StreamName{} -> + let name = streamIdRaw streamId + evtNum = eventNumberToInt64 start in + Op.readStreamEvents _settings _exec dir name evtNum cnt res_link_tos cred + All -> + let Position c_pos p_pos = start in + Op.readAllEvents _settings _exec c_pos p_pos cnt res_link_tos dir cred -------------------------------------------------------------------------------- -- | Subscribes to a stream. @@ -340,11 +317,8 @@ -> Maybe Credentials -> IO (Async WriteResult) setStreamMetadata Connection{..} evt_stream exp_ver metadata cred = do - p <- newPromise let name = streamIdRaw evt_stream - op = Op.setMetaStream _settings name exp_ver cred metadata - publishWith _exec (SubmitOperation p op) - async (retrieve p) + Op.setMetaStream _settings _exec name exp_ver cred metadata -------------------------------------------------------------------------------- -- | Asynchronously gets the metadata of a stream. @@ -352,11 +326,8 @@ -> StreamName -> Maybe Credentials -> IO (Async StreamMetadataResult) -getStreamMetadata Connection{..} evt_stream cred = do - p <- newPromise - let op = Op.readMetaStream _settings (streamIdRaw evt_stream) cred - publishWith _exec (SubmitOperation p op) - async (retrieve p) +getStreamMetadata Connection{..} stream cred = + Op.readMetaStream _settings _exec (streamIdRaw stream) cred -------------------------------------------------------------------------------- -- | Asynchronously create a persistent subscription group on a stream. @@ -366,11 +337,8 @@ -> PersistentSubscriptionSettings -> Maybe Credentials -> IO (Async (Maybe PersistActionException)) -createPersistentSubscription Connection{..} grp stream sett cred = do - p <- newPromise - let op = Op.createPersist grp (streamIdRaw stream) sett cred - publishWith _exec (SubmitOperation p op) - async (persistAsync p) +createPersistentSubscription Connection{..} grp stream sett cred = + Op.createPersist _exec grp (streamIdRaw stream) sett cred -------------------------------------------------------------------------------- -- | Asynchronously update a persistent subscription group on a stream. @@ -380,11 +348,8 @@ -> PersistentSubscriptionSettings -> Maybe Credentials -> IO (Async (Maybe PersistActionException)) -updatePersistentSubscription Connection{..} grp stream sett cred = do - p <- newPromise - let op = Op.updatePersist grp (streamIdRaw stream) sett cred - publishWith _exec (SubmitOperation p op) - async (persistAsync p) +updatePersistentSubscription Connection{..} grp stream sett cred = + Op.updatePersist _exec grp (streamIdRaw stream) sett cred -------------------------------------------------------------------------------- -- | Asynchronously delete a persistent subscription group on a stream. @@ -393,11 +358,8 @@ -> StreamName -> Maybe Credentials -> IO (Async (Maybe PersistActionException)) -deletePersistentSubscription Connection{..} grp stream cred = do - p <- newPromise - let op = Op.deletePersist grp (streamIdRaw stream) cred - publishWith _exec (SubmitOperation p op) - async (persistAsync p) +deletePersistentSubscription Connection{..} grp stream cred = + Op.deletePersist _exec grp (streamIdRaw stream) cred -------------------------------------------------------------------------------- -- | Asynchronously connect to a persistent subscription given a group on a @@ -410,8 +372,3 @@ -> IO PersistentSubscription connectToPersistentSubscription Connection{..} group stream bufSize cred = newPersistentSubscription _exec group stream bufSize cred - --------------------------------------------------------------------------------- -persistAsync :: Callback (Maybe PersistActionException) - -> IO (Maybe PersistActionException) -persistAsync = either throw return <=< tryRetrieve diff --git a/Database/EventStore/Internal/Callback.hs b/Database/EventStore/Internal/Callback.hs deleted file mode 100644 --- a/Database/EventStore/Internal/Callback.hs +++ /dev/null @@ -1,111 +0,0 @@ -{-# LANGUAGE GADTs #-} -{-# LANGUAGE RankNTypes #-} -{-# LANGUAGE ScopedTypeVariables #-} --------------------------------------------------------------------------------- --- | --- Module : Database.EventStore.Internal.Callback --- Copyright : (C) 2017 Yorick Laupa --- License : (see the file LICENSE) --- --- Maintainer : Yorick Laupa --- Stability : provisional --- Portability : non-portable --- --------------------------------------------------------------------------------- -module Database.EventStore.Internal.Callback - ( Callback - , newPromise - , newCallback - , fulfill - , reject - , retrieve - , tryRetrieve - , fromEither - ) where - --------------------------------------------------------------------------------- -import Database.EventStore.Internal.Prelude - --------------------------------------------------------------------------------- -newtype Callback a = - Callback { runCallback :: forall b. Stage a b -> IO b } - --------------------------------------------------------------------------------- -data Stage a b where - Fulfill :: a -> Stage a () - Reject :: Exception e => e -> Stage a () - Retrieve :: Stage a (Either SomeException a) - --------------------------------------------------------------------------------- -fulfill :: MonadIO m => Callback a -> a -> m () -fulfill cb a = liftIO $ runCallback cb (Fulfill a) - --------------------------------------------------------------------------------- -reject :: (Exception e, MonadIO m) => Callback a -> e -> m () -reject cb e = liftIO $ runCallback cb (Reject e) - --------------------------------------------------------------------------------- -tryRetrieve :: Callback a -> IO (Either SomeException a) -tryRetrieve cb = runCallback cb Retrieve - --------------------------------------------------------------------------------- -retrieve :: Callback a -> IO a -retrieve p = do - tryRetrieve p >>= \case - Left e -> throw e - Right a -> return a - --------------------------------------------------------------------------------- -fromEither :: Exception e => Callback a -> Either e a -> IO () -fromEither p (Left e) = reject p e -fromEither p (Right a) = fulfill p a - --------------------------------------------------------------------------------- -newPromise :: IO (Callback a) -newPromise = do - mvar <- newEmptyTMVarIO - return $ promise mvar - --------------------------------------------------------------------------------- -newCallback :: (Either SomeException a -> IO ()) -> IO (Callback a) -newCallback k = do - mvar <- newEmptyTMVarIO - return $ callback mvar k - --------------------------------------------------------------------------------- -promise :: forall a. TMVar (Either SomeException a) -> Callback a -promise mvar = Callback go - where - go :: forall b. Stage a b -> IO b - go (Fulfill a) = atomically $ - whenM (isEmptyTMVar mvar) $ - putTMVar mvar (Right a) - - go (Reject e) = atomically $ - whenM (isEmptyTMVar mvar) $ - putTMVar mvar (Left $ toException e) - - go Retrieve = atomically $ readTMVar mvar - --------------------------------------------------------------------------------- -callback :: forall a. TMVar (Either SomeException a) - -> (Either SomeException a -> IO ()) - -> Callback a -callback mvar k = Callback go - where - go :: forall b. Stage a b -> IO b - go (Fulfill a) = do - atomically $ - unlessM (tryPutTMVar mvar (Right a)) $ do - _ <- swapTMVar mvar (Right a) - return () - k (Right a) - go (Reject e) = do - let err = Left $ toException e - atomically $ do - unlessM (tryPutTMVar mvar err) $ do - _ <- swapTMVar mvar err - - return () - k err - go Retrieve = atomically $ readTMVar mvar diff --git a/Database/EventStore/Internal/Communication.hs b/Database/EventStore/Internal/Communication.hs --- a/Database/EventStore/Internal/Communication.hs +++ b/Database/EventStore/Internal/Communication.hs @@ -18,8 +18,7 @@ import Data.Typeable -------------------------------------------------------------------------------- -import Database.EventStore.Internal.Callback -import Database.EventStore.Internal.Operation +import Database.EventStore.Internal.Operation (Mailbox, Lifetime) import Database.EventStore.Internal.Prelude import Database.EventStore.Internal.Types @@ -51,11 +50,6 @@ deriving Typeable -------------------------------------------------------------------------------- -data SubmitOperation = - forall a. SubmitOperation (Callback a) (Operation a) - deriving Typeable - --------------------------------------------------------------------------------- data ServiceTerminated = ServiceTerminated Service deriving Typeable -------------------------------------------------------------------------------- @@ -65,3 +59,6 @@ -------------------------------------------------------------------------------- newtype SendPackage = SendPackage Package deriving Typeable + +-------------------------------------------------------------------------------- +data Transmit = Transmit Mailbox Lifetime Package deriving Typeable diff --git a/Database/EventStore/Internal/ConnectionManager.hs b/Database/EventStore/Internal/ConnectionManager.hs --- a/Database/EventStore/Internal/ConnectionManager.hs +++ b/Database/EventStore/Internal/ConnectionManager.hs @@ -25,7 +25,6 @@ import Data.Time -------------------------------------------------------------------------------- -import Database.EventStore.Internal.Callback import Database.EventStore.Internal.Command import Database.EventStore.Internal.Communication import Database.EventStore.Internal.Connection @@ -36,7 +35,7 @@ import Database.EventStore.Internal.Operation import Database.EventStore.Internal.Operation.Authenticate (newAuthenticatePkg) import Database.EventStore.Internal.Operation.Identify (newIdentifyPkg) -import qualified Database.EventStore.Internal.OperationManager as Operation +import qualified Database.EventStore.Internal.Manager.Operation.Registry as Operation import Database.EventStore.Internal.Prelude import Database.EventStore.Internal.Stopwatch import Database.EventStore.Internal.Types @@ -70,7 +69,7 @@ Init -> False Connected{} -> True Closed -> True - Connecting a s -> + Connecting _ s -> case s of Reconnecting -> False EndpointDiscovery -> False @@ -156,7 +155,7 @@ , _stage :: IORef Stage , _last :: IORef (Maybe EndPoint) , _sending :: TVar Bool - , _opMgr :: Operation.Manager + , _opMgr :: Operation.Registry , _stopwatch :: Stopwatch , _lastCheck :: IORef NominalDiffTime , _lastConnected :: IORef Bool @@ -171,20 +170,20 @@ monitorIncrPkgCount -------------------------------------------------------------------------------- -connectionManager :: ConnectionBuilder +connectionManager :: Settings + -> ConnectionBuilder -> Discovery -> Hub -> IO () -connectionManager builder disc mainBus = do +connectionManager setts builder disc mainBus = do stageRef <- newIORef Init let mkInternal = Internal disc builder stageRef - connRef = ConnectionRef $ lookingUpConnectionWhenConnected stageRef stopwatch <- newStopwatch timeoutCheck <- stopwatchElapsed stopwatch internal <- mkInternal <$> newIORef Nothing <*> newTVarIO False - <*> Operation.new connRef + <*> Operation.registryNew (s_operationTimeout setts) (s_operationRetry setts) <*> return stopwatch <*> newIORef timeoutCheck <*> newIORef False @@ -195,7 +194,7 @@ subscribe mainBus (onEstablish internal) subscribe mainBus (onEstablished internal) subscribe mainBus (onArrived internal) - subscribe mainBus (onSubmitOperation internal) + subscribe mainBus (onTransmit internal) subscribe mainBus (onConnectionError internal) subscribe mainBus (onConnectionClosed internal) subscribe mainBus (onCloseConnection internal) @@ -318,7 +317,8 @@ -- ms. This could lead the first operation to take time before gettings. -- FIXME: We might consider doing that hack only if it's the first time -- we connect with the server. - Operation.check _opMgr + pkgs <- Operation.registryCheckAndRetry _opMgr (connectionId conn) + traverse_ (enqueuePackage conn) pkgs _ -> pure () -------------------------------------------------------------------------------- @@ -330,7 +330,7 @@ closeConnection self@Internal{..} cause = do $logDebug [i|CloseConnection: #{cause}.|] mConn <- lookupConnectionAndSwitchToClosed self - Operation.cleanup _opMgr + Operation.registryAbort _opMgr traverse_ (closeTcpConnection self cause) mConn $logInfo [i|CloseConnection: connection cleanup done for [#{cause}].|] publish (FatalException cause) @@ -423,12 +423,15 @@ (defaultConnecting -> True) -> manageHeartbeats self - Connected _ -> do + Connected conn -> do elapsed <- stopwatchElapsed _stopwatch timeoutCheckStart <- readIORef _lastCheck when (elapsed - timeoutCheckStart >= s_operationTimeout setts) $ do - Operation.check _opMgr + $logDebug "Start check and retry..." + pkgs <- Operation.registryCheckAndRetry _opMgr (connectionId conn) + traverse_ (enqueuePackage conn) pkgs + $logDebug "Completed check and retry" atomicWriteIORef _lastCheck elapsed _ -> return () @@ -561,12 +564,9 @@ | packageCmd == heartbeatRequestCmd = enqueuePackage conn heartbeatResponse | otherwise = - Operation.handle _opMgr pkg >>= \case - Nothing -> $logWarn [i|Package not handled: #{pkg}|] - Just decision -> - case decision of - Operation.Handled -> return () - Operation.Reconnect node -> forceReconnect self node + Operation.registryHandle _opMgr pkg >>= \case + Nothing -> pure () + Just node -> forceReconnect self node closedOrInit = \case Init -> True @@ -605,17 +605,21 @@ onShutdown self@Internal{..} _ = do $logDebug "Shutting down..." mConn <- lookupConnectionAndSwitchToClosed self - Operation.cleanup _opMgr + Operation.registryAbort _opMgr traverse_ dispose mConn $logDebug "Shutdown properly." publish (ServiceTerminated ConnectionManager) -------------------------------------------------------------------------------- -onSubmitOperation :: Internal -> SubmitOperation -> EventStore () -onSubmitOperation Internal{..} (SubmitOperation callback op) = +onTransmit :: Internal -> Transmit -> EventStore () +onTransmit Internal{..} (Transmit m lifetime pkg) = readIORef _stage >>= \case - Closed -> reject callback Aborted - _ -> Operation.submit _opMgr op callback + Closed + -> mailboxFail m Aborted + Connected conn + -> do enqueuePackage conn pkg + Operation.registryRegister _opMgr (connectionId conn) lifetime pkg m + _ -> Operation.registryPostpone _opMgr m lifetime pkg -------------------------------------------------------------------------------- onCloseConnection :: Internal -> CloseConnection -> EventStore () @@ -637,13 +641,6 @@ Identification _ _ conn -> Just conn _ -> Nothing go _ = Nothing - --------------------------------------------------------------------------------- -lookingUpConnectionWhenConnected :: IORef Stage -> EventStore (Maybe Connection) -lookingUpConnectionWhenConnected = fmap go . readIORef - where - go (Connected conn) = Just conn - go _ = Nothing -------------------------------------------------------------------------------- onSendPackage :: Internal -> SendPackage -> EventStore () diff --git a/Database/EventStore/Internal/Control.hs b/Database/EventStore/Internal/Control.hs --- a/Database/EventStore/Internal/Control.hs +++ b/Database/EventStore/Internal/Control.hs @@ -66,9 +66,6 @@ import Control.Monad.Reader import Data.UUID import Data.UUID.V4 -import System.Metrics -import System.Metrics.Counter hiding (add) -import System.Metrics.Distribution -------------------------------------------------------------------------------- import Database.EventStore.Internal.Logger @@ -77,10 +74,10 @@ -------------------------------------------------------------------------------- data Env = - Env { __logRef :: LoggerRef - , __settings :: Settings - , __bus :: Bus - , __monitor :: Maybe Monitoring + Env { __logRef :: !LoggerRef + , __settings :: !Settings + , __bus :: !Bus + , __monitor :: !MonitoringBackend } -------------------------------------------------------------------------------- @@ -107,7 +104,7 @@ getSettings = __settings <$> getEnv -------------------------------------------------------------------------------- -freshUUID :: EventStore UUID +freshUUID :: MonadIO m => m UUID freshUUID = liftIO nextRandom -------------------------------------------------------------------------------- @@ -302,7 +299,7 @@ , _busEventHandlers :: IORef EventHandlers , _busQueue :: TBMQueue Message , _workerAsync :: Async () - , _monitoring :: Maybe Monitoring + , _monitoring :: MonitoringBackend } -------------------------------------------------------------------------------- @@ -324,7 +321,7 @@ Bus ref setts <$> newIORef mempty <*> newTBMQueueIO 500 <*> async (worker b) - <*> traverse configureMonitoring (s_monitoring setts) + <*> pure (s_monitoring setts) return bus @@ -375,55 +372,33 @@ -------------------------------------------------------------------------------- -- Monitoring -------------------------------------------------------------------------------- -data Monitoring = - Monitoring - { _pkgCount :: Counter - , _connDrops :: Counter - , _dataTx :: Distribution - , _forceReco :: Counter - , _heartTimeout :: Counter - } -------------------------------------------------------------------------------- -configureMonitoring :: Store -> IO Monitoring -configureMonitoring store = - Monitoring <$> createCounter "eventstore.packages.received" store - <*> createCounter "eventstore.connection.drops" store - <*> createDistribution "eventstore.data.transmitted" store - <*> createCounter "eventstore.force_reconnect" store - <*> createCounter "eventstore.heartbeat.timeouts" store - --------------------------------------------------------------------------------- monitorIncrPkgCount :: EventStore () monitorIncrPkgCount = do Env{..} <- getEnv - for_ __monitor $ \Monitoring{..}-> - liftIO $ inc _pkgCount + liftIO $ monitoringBackendIncrPkgCount __monitor -------------------------------------------------------------------------------- monitorIncrConnectionDrop :: EventStore () monitorIncrConnectionDrop = do Env{..} <- getEnv - for_ __monitor $ \Monitoring{..}-> - liftIO $ inc _connDrops + liftIO $ monitoringBackendIncrConnectionDrop __monitor -------------------------------------------------------------------------------- monitorAddDataTransmitted :: Int -> EventStore () monitorAddDataTransmitted siz = do Env{..} <- getEnv - for_ __monitor $ \Monitoring{..}-> - liftIO $ add _dataTx (fromIntegral siz) + liftIO $ monitoringBackendAddDataTransmitted __monitor siz -------------------------------------------------------------------------------- monitorIncrForceReconnect :: EventStore () monitorIncrForceReconnect = do Env{..} <- getEnv - for_ __monitor $ \Monitoring{..}-> - liftIO $ inc _forceReco + liftIO $ monitoringBackendIncrForceReconnect __monitor -------------------------------------------------------------------------------- monitorIncrHeartbeatTimeouts :: EventStore () monitorIncrHeartbeatTimeouts = do Env{..} <- getEnv - for_ __monitor $ \Monitoring{..}-> - liftIO $ inc _heartTimeout + liftIO $ monitoringBackendIncrHeartbeatTimeouts __monitor 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 @@ -34,7 +34,7 @@ ) where -------------------------------------------------------------------------------- -import Prelude (String) +import Prelude (String, fail) import Data.Maybe -------------------------------------------------------------------------------- diff --git a/Database/EventStore/Internal/Exec.hs b/Database/EventStore/Internal/Exec.hs --- a/Database/EventStore/Internal/Exec.hs +++ b/Database/EventStore/Internal/Exec.hs @@ -108,7 +108,7 @@ hub = asHub mainBus timerService hub - connectionManager builder disc hub + connectionManager setts builder disc hub subscribe mainBus (onInit internal) subscribe mainBus (onInitFailed internal) diff --git a/Database/EventStore/Internal/Manager/Operation/Registry.hs b/Database/EventStore/Internal/Manager/Operation/Registry.hs --- a/Database/EventStore/Internal/Manager/Operation/Registry.hs +++ b/Database/EventStore/Internal/Manager/Operation/Registry.hs @@ -1,437 +1,389 @@ -{-# LANGUAGE DeriveDataTypeable #-} -{-# LANGUAGE ExistentialQuantification #-} -{-# LANGUAGE RecordWildCards #-} -{-# LANGUAGE TypeFamilies #-} -------------------------------------------------------------------------------- -- | --- Module : Database.EventStore.Internal.Manager.Operation.Registry --- Copyright : (C) 2017 Yorick Laupa --- License : (see the file LICENSE) --- --- Maintainer : Yorick Laupa --- Stability : provisional --- Portability : non-portable +-- Module : Database.EventStore.Internal.Manager.Registry +-- Copyright : (C) 2020 Yorick Laupa +-- License : (see the file LICENSE) +-- Maintainer: Yorick Laupa +-- Stability : experimental +-- Portability: non-portable -- --- Main operation bookkeeping structure. -------------------------------------------------------------------------------- -module Database.EventStore.Internal.Manager.Operation.Registry - ( Registry - , OperationMaxAttemptReached(..) - , Decision(..) - , newRegistry - , register - , handlePackage - , abortPendingRequests - , checkAndRetry - , startAwaitings - ) where +module Database.EventStore.Internal.Manager.Operation.Registry where -------------------------------------------------------------------------------- +import qualified Data.HashMap.Strict as HashMap + +-------------------------------------------------------------------------------- import Data.ProtocolBuffers import Data.Serialize -import Data.Time -import Data.UUID -import Data.UUID.V4 -import Data.Void (absurd) -------------------------------------------------------------------------------- -import Database.EventStore.Internal.Callback import Database.EventStore.Internal.Command import Database.EventStore.Internal.Control -import Database.EventStore.Internal.Connection import Database.EventStore.Internal.EndPoint import Database.EventStore.Internal.Logger -import Database.EventStore.Internal.Operation hiding (retry) import Database.EventStore.Internal.Prelude +import Database.EventStore.Internal.Operation import Database.EventStore.Internal.Stopwatch import Database.EventStore.Internal.Types -------------------------------------------------------------------------------- data Request = - Request { _requestCmd :: !Command - , _requestPayload :: !ByteString - , _requestCred :: !(Maybe Credentials) - } - --------------------------------------------------------------------------------- -data Suspend a - = Required !(Package -> Operation a) - | Optional !(Maybe Package -> Operation a) - | Resolved !(Operation a) - --------------------------------------------------------------------------------- -type SessionId = Integer -type SessionMap = HashMap SessionId Session - --------------------------------------------------------------------------------- -data Session = - forall result. - Session { sessionId :: !SessionId - , sessionOp :: !(Operation result) - , sessionStack :: !(IORef (Suspend result)) - , sessionCallback :: !(Callback result) - } - --------------------------------------------------------------------------------- -rejectSession :: Exception e => Session -> e -> EventStore () -rejectSession Session{..} = reject sessionCallback + Request + { requestOriginal :: !Package + , requestConnId :: !UUID + , requestRetries :: !Int + , requestStarted :: !NominalDiffTime + , requestMailbox :: !Mailbox + , requestLifetime :: !Lifetime + } -------------------------------------------------------------------------------- -resumeSession :: Registry -> Session -> Package -> EventStore () -resumeSession reg session@Session{..} pkg = do - atomicModifyIORef' sessionStack $ \case - Required k -> (Resolved $ k pkg, ()) - Optional k -> (Resolved $ k (Just pkg), ()) - same -> (same, ()) - - execute reg session +requestIsKeepAlive :: Request -> Bool +requestIsKeepAlive req + = case requestLifetime req of + OneTime -> False + KeepAlive _ -> True -------------------------------------------------------------------------------- -resumeNoPkgSession :: Registry -> Session -> EventStore () -resumeNoPkgSession reg session@Session{..} = do - atomicModifyIORef' sessionStack $ \case - Optional k -> (Resolved $ k Nothing, ()) - same -> (same, ()) - - execute reg session +requestToWaiting :: Request -> Waiting +requestToWaiting req = + Waiting + { waitingLifetime = requestLifetime req + , waitingPkg = requestOriginal req + , waitingMailbox = requestMailbox req + } -------------------------------------------------------------------------------- -reinitSession :: Session -> EventStore () -reinitSession Session{..} = atomicWriteIORef sessionStack (Resolved sessionOp) +waitingToRequest + :: UUID -- Connection id. + -> NominalDiffTime + -> Waiting + -> Request +waitingToRequest connId started w + = Request + { requestOriginal = waitingPkg w + , requestConnId = connId + , requestRetries = 1 + , requestStarted = started + , requestMailbox = waitingMailbox w + , requestLifetime = waitingLifetime w + } -------------------------------------------------------------------------------- -restartSession :: Registry -> Session -> EventStore () -restartSession reg session = do - reinitSession session - execute reg session +data Waiting = + Waiting + { waitingLifetime :: !Lifetime + , waitingPkg :: !Package + , waitingMailbox :: !Mailbox + } -------------------------------------------------------------------------------- -data Sessions = - Sessions { sessionsNextId :: IORef SessionId - , sessionsMap :: IORef SessionMap - } +type Requests = HashMap UUID Request -------------------------------------------------------------------------------- -createSession :: Sessions -> Operation a -> Callback a -> EventStore Session -createSession Sessions{..} op cb = do - sid <- atomicModifyIORef' sessionsNextId $ \n -> (succ n, n) - ref <- newIORef (Resolved op) - atomicModifyIORef' sessionsMap $ \m -> - let session = - Session { sessionId = sid - , sessionOp = op - , sessionStack = ref - , sessionCallback = cb - } in - (insertMap sid session m, session) +data Registry' = + Registry' + { registryRequests :: !Requests + , registryWaitings :: ![Waiting] + , registryTimeout :: !NominalDiffTime + , registryMaxRetry :: !Retry + } -------------------------------------------------------------------------------- -destroySession :: Sessions -> Session -> EventStore () -destroySession Sessions{..} s = - atomicModifyIORef' sessionsMap $ \m -> (deleteMap (sessionId s) m, ()) +registryClear :: Registry' -> Registry' +registryClear reg = + reg + { registryRequests = mempty + , registryWaitings = [] + } -------------------------------------------------------------------------------- -sessionDisposed :: Sessions -> Session -> EventStore Bool -sessionDisposed Sessions{..} s = - notMember (sessionId s) <$> readIORef sessionsMap +data Registry = + Registry + { registryState :: IORef Registry' + , registryStopwatch :: Stopwatch + } -------------------------------------------------------------------------------- -newSessions :: IO Sessions -newSessions = - Sessions <$> newIORef 0  - <*> newIORef mempty +-- | I'm bad at naming thing however, we are going to use that datastructure +-- so we could lookup and delete in one single pass. +data Blob a b = Blob a b -------------------------------------------------------------------------------- -packageOf :: Request -> UUID -> Package -packageOf Request{..} uuid = - Package { packageCmd = _requestCmd - , packageCorrelation = uuid - , packageData = _requestPayload - , packageCred = _requestCred - } +instance Functor (Blob a) where + fmap f (Blob a b) = Blob a (f b) -------------------------------------------------------------------------------- -data Pending = - Pending { _pendingRequest :: !(Maybe Request) - , _pendingSession :: !Session - , _pendingRetries :: !Int - , _pendingLastTry :: !NominalDiffTime - , _pendingConnId :: !UUID - } +registryRemoveRequest + :: UUID + -> Registry' + -> (Maybe Request, Registry') +registryRemoveRequest key reg + = let Blob result newMap + = HashMap.alterF go key (registryRequests reg) + in (result, reg { registryRequests = newMap }) + where + go Nothing = Blob Nothing Nothing + go (Just e) = Blob (Just e) Nothing -------------------------------------------------------------------------------- -destroyPendingSession :: Sessions -> Pending -> EventStore () -destroyPendingSession sessions Pending{..} = - destroySession sessions _pendingSession +registryNew :: NominalDiffTime -> Retry -> IO Registry +registryNew timeout maxRetry + = Registry + <$> newIORef state + <*> newStopwatch + where + state = + Registry' + { registryRequests = mempty + , registryWaitings = [] + , registryTimeout = timeout + , registryMaxRetry = maxRetry + } -------------------------------------------------------------------------------- -type PendingRequests = HashMap UUID Pending +registryRegister + :: Registry + -> UUID -- Connection Id. + -> Lifetime + -> Package + -> Mailbox + -> EventStore () +registryRegister reg connId lifetime pkg mailbox + = do started <- stopwatchElapsed (registryStopwatch reg) + modifyIORef' (registryState reg) (go started) + where + go started state = + let req = Request + { requestOriginal = pkg + , requestConnId = connId + , requestRetries = 1 + , requestStarted = started + , requestMailbox = mailbox + , requestLifetime = lifetime + } --------------------------------------------------------------------------------- -data Awaiting - = Awaiting !Session - | AwaitingRequest !Session !Request + correlation = packageCorrelation pkg + nextReqs = insertMap correlation req (registryRequests state) + in state { registryRequests = nextReqs } -------------------------------------------------------------------------------- -type Awaitings = [Awaiting] +registryPostpone + :: Registry + -> Mailbox + -> Lifetime + -> Package + -> EventStore () +registryPostpone reg mailbox lifetime pkg + = modifyIORef' (registryState reg) go + where + go state + = let waiting + = Waiting + { waitingLifetime = lifetime + , waitingPkg = pkg + , waitingMailbox = mailbox + } --------------------------------------------------------------------------------- -rejectPending :: Exception e => Pending -> e -> EventStore () -rejectPending Pending{..} = rejectSession _pendingSession + nextWs = waiting : registryWaitings state + in state { registryWaitings = nextWs } -------------------------------------------------------------------------------- -rejectAwaiting :: Exception e => Awaiting -> e -> EventStore () -rejectAwaiting (Awaiting s) = rejectSession s -rejectAwaiting (AwaitingRequest s _) = rejectSession s +registryHandle + :: Registry + -> Package + -> EventStore (Maybe NodeEndPoints) +registryHandle reg pkg + = do state <- readIORef (registryState reg) + case registryRemoveRequest (packageCorrelation pkg) state of + (Nothing, _) + -> do $logWarn [i|No operation associated to package: #{pkg}|] + pure Nothing + (Just req, stateWithoutReq) + -> case packageCmd pkg of + cmd | cmd == badRequestCmd + -> do let reason = packageDataAsText pkg + mailboxFail (requestMailbox req) (ServerError reason) + pure Nothing --------------------------------------------------------------------------------- -applyResponse :: Registry -> Pending -> Package -> EventStore () -applyResponse reg Pending{..} = resumeSession reg _pendingSession + | cmd == notAuthenticatedCmd + -> do mailboxFail (requestMailbox req) NotAuthenticatedOp + pure Nothing --------------------------------------------------------------------------------- -restartPending :: Registry -> Pending -> EventStore () -restartPending reg Pending{..} = restartSession reg _pendingSession + | cmd == notHandledCmd -- In all cases, we decide to postpone that command. + -> do $(logWarn) [i|Not handled response received: #{pkg}.|] + let Just msg = maybeDecodeMessage $ packageData pkg + reason = getField $ notHandledReason msg + waiting = requestToWaiting req + nextWs = waiting : registryWaitings stateWithoutReq + finalState = stateWithoutReq { registryWaitings = nextWs } + origCmd = packageCmd (requestOriginal req) + pkgId = packageCorrelation pkg --------------------------------------------------------------------------------- -data Registry = - Registry { _regConnRef :: ConnectionRef - , _regPendings :: IORef PendingRequests - , _regAwaitings :: IORef Awaitings - , _stopwatch :: Stopwatch - , _sessions :: Sessions - } + writeIORef (registryState reg) finalState --------------------------------------------------------------------------------- -newRegistry :: ConnectionRef -> IO Registry -newRegistry ref = - Registry ref <$> newIORef mempty - <*> newIORef [] - <*> newStopwatch - <*> newSessions + case reason of + N_NotMaster + -> do let Just details = getField $ notHandledAdditionalInfo msg + info = masterInfo details + node = masterInfoNodeEndPoints info --------------------------------------------------------------------------------- -scheduleAwait :: Registry -> Awaiting -> EventStore () -scheduleAwait Registry{..} aw = - atomicModifyIORef' _regAwaitings $ \stack -> - (aw : stack, ()) + $(logWarn) [i|Received a non master error on command #{origCmd} [#{pkgId}] on #{node}|] + pure (Just node) --------------------------------------------------------------------------------- -execute :: Registry -> Session -> EventStore () -execute self@Registry{..} session@Session{..} = - readIORef sessionStack >>= \case - Resolved action -> loop action - _ -> return () - where - loop stream = - case stream of - Return x -> absurd x - Effect m -> - case m of - Failed e -> do - destroySession _sessions session - rejectSession session e - Retry -> do - reinitSession session - execute self session - Proceed inner -> - loop inner - Step step -> - case step of - Stop -> destroySession _sessions session - Yield a next -> do - fulfill sessionCallback a - loop next - Await k tpe _ -> - case tpe of - NeedUUID -> loop . k =<< freshUUID - NeedRemote p -> do - let req = Request { _requestCmd = payloadCmd p - , _requestPayload = payloadData p - , _requestCred = payloadCreds p - } - atomicWriteIORef sessionStack (Required k) - maybeConnection _regConnRef >>= \case - Nothing -> scheduleAwait self (AwaitingRequest session req) - Just conn -> issueRequest self session conn req + _ -> do $(logWarn) [i|The server has either not started or is too busy. Retrying command #{origCmd} #{pkgId}|] + pure Nothing - WaitRemote uuid -> do - atomicWriteIORef sessionStack (Optional k) - let mkNewPending conn = do - let connId = connectionId conn - pending <- createPending session _stopwatch connId Nothing - insertPending self uuid pending + | otherwise + -> do let respCmd = packageCmd pkg - traverse_ mkNewPending =<< maybeConnection _regConnRef + mailboxSendPkg (requestMailbox req) pkg --------------------------------------------------------------------------------- -issueRequest :: Registry - -> Session - -> Connection - -> Request - -> EventStore () -issueRequest reg@Registry{..} session conn req = do - uuid <- liftBase nextRandom - pending <- createPending session _stopwatch (connectionId conn) (Just req) + case requestLifetime req of + OneTime + -> do writeIORef (registryState reg) stateWithoutReq + pure Nothing - insertPending reg uuid pending - enqueuePackage conn (packageOf req uuid) + KeepAlive endCmd + | endCmd == respCmd + -> do writeIORef (registryState reg) stateWithoutReq + pure Nothing + | otherwise -- Means we keep the previous state (Subscription). + -> pure Nothing -------------------------------------------------------------------------------- -createPending :: MonadBaseControl IO m - => Session - -> Stopwatch - -> UUID - -> Maybe Request - -> m Pending -createPending session stopwatch connId mReq = do - elapsed <- stopwatchElapsed stopwatch - let pending = - Pending { _pendingRequest = mReq - , _pendingSession = session - , _pendingRetries = 1 - , _pendingLastTry = elapsed - , _pendingConnId = connId - } - - return pending +data CRState = + CRState + { crsState :: !Registry' + , crsPkgs :: ![Package] + } -------------------------------------------------------------------------------- -insertPending :: Registry -> UUID -> Pending -> EventStore () -insertPending Registry{..} key pending = - atomicModifyIORef' _regPendings $ \pendings -> - (insertMap key pending pendings, ()) +crsStateNew :: Registry' -> CRState +crsStateNew reg = + CRState + { crsState = reg + , crsPkgs = [] + } -------------------------------------------------------------------------------- -register :: Registry -> Operation a -> Callback a -> EventStore () -register reg op cb = do - session <- createSession (_sessions reg) op cb - execute reg session +crsStateDeleteReq :: Request -> CRState -> CRState +crsStateDeleteReq req reg + = let state + = crsState reg + nextReqs + = deleteMap + (packageCorrelation . requestOriginal $ req) + (registryRequests state) + nextState + = state { registryRequests = nextReqs } in + reg { crsState = nextState } -------------------------------------------------------------------------------- -abortPendingRequests :: Registry -> EventStore () -abortPendingRequests Registry{..} = do - m <- atomicModifyIORef' _regPendings $ \pendings -> (mempty, pendings) - ws <- atomicModifyIORef' _regAwaitings $ \aw -> (mempty, aw) - - for_ m $ \p -> rejectPending p Aborted - for_ ws $ \a -> rejectAwaiting a Aborted +crsStateRegisterReq :: Request -> CRState -> CRState +crsStateRegisterReq req reg + = let state + = crsState reg + nextReqs + = insertMap + (packageCorrelation . requestOriginal $ req) + req + (registryRequests state) + nextState + = state { registryRequests = nextReqs } in + reg { crsState = nextState } -------------------------------------------------------------------------------- -data Decision - = Handled - | Reconnect NodeEndPoints +crsStateAddPkg :: Package -> CRState -> CRState +crsStateAddPkg pkg reg + = let nextPkgs = pkg : crsPkgs reg in + reg { crsPkgs = nextPkgs } -------------------------------------------------------------------------------- -handlePackage :: Registry -> Package -> EventStore (Maybe Decision) -handlePackage reg@Registry{..} pkg@Package{..} = do - outcome <- atomicModifyIORef' _regPendings $ \pendings -> - let uuid = packageCorrelation in - (deleteMap uuid pendings, lookup uuid pendings) - - case outcome of - Nothing -> return Nothing - Just pending -> Just <$> executePending reg pkg pending +registryCheckAndRetry + :: Registry + -> UUID -- Connection id. + -> EventStore [Package] +registryCheckAndRetry reg connId + = do state <- readIORef (registryState reg) + elapsed <- stopwatchElapsed (registryStopwatch reg) + let reqs = mapToList $ registryRequests state --------------------------------------------------------------------------------- -executePending :: Registry -> Package -> Pending -> EventStore Decision -executePending reg@Registry{..} pkg@Package{..} p@Pending{..} = - case packageCmd of - cmd | cmd == badRequestCmd -> do - let reason = packageDataAsText pkg + newState <- foldM (checking elapsed) (crsStateNew state) reqs + let newStateTemp = crsState newState + awaitings = registryWaitings newStateTemp + tempState = newStateTemp { registryWaitings = [] } + newCRState = newState { crsState = tempState } + finalState = foldl' (sending elapsed) newCRState awaitings - rejectPending p (ServerError reason) - return Handled - | cmd == notAuthenticatedCmd -> do - rejectPending p NotAuthenticatedOp - return Handled - | cmd == notHandledCmd -> do - $(logDebug) [i|Not handled response received: #{pkg}.|] - let Just msg = maybeDecodeMessage packageData - reason = getField $ notHandledReason msg - case reason of - N_NotMaster -> do - let Just details = getField $ notHandledAdditionalInfo msg - info = masterInfo details - node = masterInfoNodeEndPoints info + writeIORef (registryState reg) (crsState finalState) + pure (crsPkgs finalState) + where + checking elapsed state (_, req) + = do let maxTimeout = registryTimeout . crsState $ state + hasTimeout = elapsed - (requestStarted req) >= maxTimeout + maxRetry = registryMaxRetry . crsState $ state + if requestConnId req /= connId + then + do mailboxFail (requestMailbox req) ConnectionHasDropped + pure (crsStateDeleteReq req state) + else if not (requestIsKeepAlive req) && hasTimeout + then case maxRetry of + AtMost maxAtt + | requestRetries req + 1 > maxAtt + -> do let pkg = requestOriginal req + pkgId = packageCorrelation pkg + cmd = packageCmd pkg - restartPending reg p - return $ Reconnect node - -- In this case with just retry the operation. - _ -> Handled <$ restartPending reg p - | otherwise -> do - applyResponse reg p pkg - return Handled + $(logError) [i|Command #{cmd} [#{pkgId}] maximum retries threshold reached (#{maxAtt}), aborted!|] + mailboxFail (requestMailbox req) Aborted + pure (crsStateDeleteReq req state) + | otherwise + -> retryReq + KeepRetrying + -> retryReq + else + pure state + where + retryReq + = do let nextRetries + = requestRetries req + 1 + nextReq + = req + { requestRetries = nextRetries + , requestStarted = elapsed + } --------------------------------------------------------------------------------- --- | Occurs when an operation has been retried more than 's_operationRetry'. -data OperationMaxAttemptReached = - OperationMaxAttemptReached UUID Command - deriving (Show, Typeable) + maxAtt + = case registryMaxRetry . crsState $ state of + AtMost n -> n + KeepRetrying -> maxBound --------------------------------------------------------------------------------- -instance Exception OperationMaxAttemptReached + pkg = requestOriginal req + cmd = packageCmd pkg + pkgId = packageCorrelation pkg --------------------------------------------------------------------------------- -checkAndRetry :: Registry -> EventStore () -checkAndRetry self@Registry{..} = do - pendings <- readIORef _regPendings - elapsed <- stopwatchElapsed _stopwatch - newPendings <- foldM (checking elapsed) pendings (mapToList pendings) - atomicWriteIORef _regPendings newPendings - where - checking elapsed reg (key, p) = do - conn <- getConnection _regConnRef - setts <- getSettings - case _pendingRequest p of - Nothing - | connectionId conn == _pendingConnId p -> return reg - | otherwise -> do - resumeNoPkgSession self (_pendingSession p) - disposed <- sessionDisposed _sessions (_pendingSession p) - let newReg = if disposed then deleteMap key reg else reg - return newReg - Just req -> do - let lastTry = _pendingLastTry p - hasTimeout = elapsed - lastTry >= s_operationTimeout setts - if hasTimeout || _pendingConnId p /= connectionId conn - then do - let retry = do - uuid <- liftBase nextRandom - let pkg = packageOf req uuid - pending = - p { _pendingLastTry = elapsed - , _pendingRetries = _pendingRetries p + 1 - , _pendingConnId = connectionId conn - } - nextReg = deleteMap key $ insertMap uuid pending reg + $(logWarn) [i|Command #{cmd} [#{pkgId} has timeout. Retrying (attempt #{nextRetries}/#{maxAtt})|] - enqueuePackage conn pkg - return nextReg + pure . crsStateRegisterReq nextReq . crsStateAddPkg pkg $ state - case s_operationRetry setts of - AtMost maxAttempts - | _pendingRetries p <= maxAttempts - -> retry - | otherwise -> do - let cmd = _requestCmd req - destroyPendingSession _sessions p - rejectPending p (OperationMaxAttemptReached key cmd) - return $ deleteMap key reg - KeepRetrying -> retry - else return reg + sending elapsed state w + = let req = waitingToRequest connId elapsed w + pkg = requestOriginal req in + crsStateRegisterReq req . crsStateAddPkg pkg $ state -------------------------------------------------------------------------------- -startAwaitings :: Registry -> EventStore () -startAwaitings reg@Registry{..} = do - awaitings <- atomicModifyIORef' _regAwaitings $ \stack -> - ([], reverse stack) +registryAbort :: Registry -> EventStore () +registryAbort reg + = do state <- readIORef (registryState reg) + writeIORef (registryState reg) (registryClear state) - traverse_ starting awaitings - where - starting (Awaiting session) = execute reg session - starting (AwaitingRequest session req) = do - conn <- getConnection _regConnRef - issueRequest reg session conn req + for_ (registryRequests state) $ \req + -> mailboxFail (requestMailbox req) Aborted + + for_ (registryWaitings state) $ \w + -> mailboxFail (waitingMailbox w) Aborted -------------------------------------------------------------------------------- maybeDecodeMessage :: Decode a => ByteString -> Maybe a 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 @@ -1,404 +1,159 @@ -{-# LANGUAGE DataKinds #-} -{-# LANGUAGE DeriveDataTypeable #-} -{-# LANGUAGE GADTs #-} -{-# LANGUAGE GeneralizedNewtypeDeriving #-} -{-# LANGUAGE KindSignatures #-} -{-# LANGUAGE MultiParamTypeClasses #-} -{-# LANGUAGE Rank2Types #-} -{-# LANGUAGE StandaloneDeriving #-} +{-# LANGUAGE ExistentialQuantification #-} +{-# LANGUAGE Rank2Types #-} +{-# LANGUAGE StandaloneDeriving #-} -------------------------------------------------------------------------------- -- | --- Module : Database.EventStore.Internal.Operation --- Copyright : (C) 2015 Yorick Laupa --- License : (see the file LICENSE) --- --- Maintainer : Yorick Laupa --- Stability : provisional --- Portability : non-portable +-- Module : Database.EventStore.Internal.Operation +-- Copyright : (C) 2020 Yorick Laupa +-- License : (see the file LICENSE) +-- Maintainer: Yorick Laupa +-- Stability : experimental +-- Portability: non-portable -- -------------------------------------------------------------------------------- -module Database.EventStore.Internal.Operation - ( OpResult(..) - , OperationError(..) - , Operation - , Code - , Need(..) - , Execution(..) - , Expect(..) - , Coroutine(..) - , Payload(..) - , freshId - , failure - , retry - , send - , request - , waitForOr - , wrongVersion - , streamDeleted - , invalidTransaction - , accessDenied - , protobufDecodingError - , serverError - , invalidServerResponse - , construct - , yield - , traversing - , stop - , (<~) - , unfolding - , append - , Stream(..) - ) where +module Database.EventStore.Internal.Operation where -------------------------------------------------------------------------------- import Prelude (String) -import Control.Category - --------------------------------------------------------------------------------- import Data.ProtocolBuffers -import Data.Serialize -import Data.UUID -import Data.Void (Void, absurd) -import Streaming.Internal +import Data.Serialize (runPut, runGet) -------------------------------------------------------------------------------- import Database.EventStore.Internal.Command +import Database.EventStore.Internal.Control import Database.EventStore.Internal.Prelude hiding ((.), id) -import Database.EventStore.Internal.Settings import Database.EventStore.Internal.Stream import Database.EventStore.Internal.Types -------------------------------------------------------------------------------- -infixr 9 <~ - --------------------------------------------------------------------------------- --- | Operation result sent by the server. -data OpResult - = OP_SUCCESS - | OP_PREPARE_TIMEOUT - | OP_COMMIT_TIMEOUT - | OP_FORWARD_TIMEOUT - | OP_WRONG_EXPECTED_VERSION - | OP_STREAM_DELETED - | OP_INVALID_TRANSACTION - | OP_ACCESS_DENIED - deriving (Eq, Enum, Show) - --------------------------------------------------------------------------------- --- | Operation exception that can occurs on an operation response. -data OperationError - = WrongExpectedVersion Text ExpectedVersion -- ^ Stream and Expected Version - | StreamDeleted StreamName -- ^ Stream - | InvalidTransaction - | forall t. AccessDenied (StreamId t) -- ^ Stream - | InvalidServerResponse Command Command -- ^ Expected, Found - | ProtobufDecodingError String - | ServerError (Maybe Text) -- ^ Reason - | InvalidOperation Text - | StreamNotFound StreamName - | NotAuthenticatedOp - -- ^ Invalid operation state. If happens, it's a driver bug. - | Aborted - -- ^ Occurs when the user asked to close the connection or if the - -- connection can't reconnect anymore. - deriving Typeable - --------------------------------------------------------------------------------- -deriving instance Show OperationError - --------------------------------------------------------------------------------- -instance Exception OperationError - --------------------------------------------------------------------------------- -data Payload = - Payload - { payloadCmd :: !Command - , payloadData :: !ByteString - , payloadCreds :: !(Maybe Credentials) - } - --------------------------------------------------------------------------------- -data Coroutine k o a where - Yield :: o -> a -> Coroutine k o a - Await :: (i -> a) -> k i -> a -> Coroutine k o a - Stop :: Coroutine k o a - --------------------------------------------------------------------------------- -instance Functor (Coroutine k o) where - fmap f (Yield o a) = Yield o (f a) - fmap f (Await k n a) = Await (f . k) n (f a) - fmap _ Stop = Stop - --------------------------------------------------------------------------------- -data Is a b where - Same :: Is a a +newtype Mailbox = Mailbox (Chan (Either OperationError Package)) -------------------------------------------------------------------------------- -instance Category Is where - id = Same - Same . Same = Same +mailboxSendPkg :: MonadBase IO m => Mailbox -> Package -> m () +mailboxSendPkg (Mailbox chan) pkg = writeChan chan (Right pkg) -------------------------------------------------------------------------------- -data Need a where - NeedUUID :: Need UUID - NeedRemote :: Payload -> Need Package - WaitRemote :: UUID -> Need (Maybe Package) +mailboxFail :: MonadBase IO m => Mailbox -> OperationError -> m () +mailboxFail (Mailbox chan) e = writeChan chan (Left e) -------------------------------------------------------------------------------- -data Execution a - = Proceed a - | Retry - | Failed !OperationError +mailboxRead :: MonadBase IO m => Mailbox -> m (Either OperationError Package) +mailboxRead (Mailbox chan) = readChan chan -------------------------------------------------------------------------------- -instance Functor Execution where - fmap f (Proceed a) = Proceed (f a) - fmap _ Retry = Retry - fmap _ (Failed e) = Failed e +mailboxReadDecoded + :: (MonadBase IO m, Decode resp) + => Mailbox + -> m (Either OperationError resp) +mailboxReadDecoded (Mailbox chan) + = fmap (decodePkg =<<) $ readChan chan -------------------------------------------------------------------------------- -instance Applicative Execution where - pure = return - (<*>) = ap +mailboxNew :: MonadBase IO m => m Mailbox +mailboxNew = Mailbox <$> newChan -------------------------------------------------------------------------------- -instance Monad Execution where - return = Proceed +createPkg + :: (Encode msg, MonadIO m) + => Command + -> Maybe Credentials + -> msg + -> m Package +createPkg cmd creds msg + = do pkgId <- freshUUID + let dat = runPut $ encodeMessage msg + pkg + = Package + { packageCmd = cmd + , packageCorrelation = pkgId + , packageData = dat + , packageCred = creds + } - Proceed a >>= f = f a - Retry >>= _ = Retry - Failed e >>= _ = Failed e + pure pkg -------------------------------------------------------------------------------- -type Machine k o m r = Stream (Coroutine k o) m r -type Code output a = Machine Need output Execution a -type Operation output = Code output Void -type Process m a b = Stream (Coroutine (Is a) b) m Void +-- FIXME We could use Bifunctor but can't I am not sure it covers all the GHC +-- we support at that time. +decodePkg :: Decode msg => Package -> Either OperationError msg +decodePkg pkg + = case runGet decodeMessage (packageData pkg) of + Left e -> Left $ ProtobufDecodingError e + Right resp -> Right resp -------------------------------------------------------------------------------- -awaits :: Monad m => k i -> Machine k o m i -awaits instr = Step (Await pure instr (Step Stop)) +-- | Operation exception that can occurs on an operation response. +data OperationError + = WrongExpectedVersion !Text !ExpectedVersion -- ^ Stream and Expected Version + | StreamDeleted !StreamName -- ^ Stream + | InvalidTransaction + | forall t. AccessDenied !(StreamId t) -- ^ Stream + | InvalidServerResponse !Command !Command -- ^ Expected, Found + | ProtobufDecodingError !String + | ServerError !(Maybe Text) -- ^ Reason + | InvalidOperation !Text + | StreamNotFound !StreamName + | NotAuthenticatedOp + -- ^ Invalid operation state. If happens, it's a driver bug. + | Aborted + -- ^ Occurs when the user asked to close the connection or if the + -- connection can't reconnect anymore. + | ConnectionHasDropped + deriving Typeable -------------------------------------------------------------------------------- -await :: (Monad m, Category k) => Machine (k i) o m i -await = awaits id +deriving instance Show OperationError -------------------------------------------------------------------------------- -stop :: Machine k o m a -stop = Step Stop +instance Exception OperationError -------------------------------------------------------------------------------- -yield :: Monad m => o -> Machine k o m () -yield o = Step (Yield o (pure ())) +-- | Operation result sent by the server. +data OpResult + = OP_SUCCESS + | OP_PREPARE_TIMEOUT + | OP_COMMIT_TIMEOUT + | OP_FORWARD_TIMEOUT + | OP_WRONG_EXPECTED_VERSION + | OP_STREAM_DELETED + | OP_INVALID_TRANSACTION + | OP_ACCESS_DENIED + deriving (Eq, Enum, Show) -------------------------------------------------------------------------------- -traversing :: Monad m => (a -> m b) -> Process m a b -traversing k = repeatedly $ do - a <- await - b <- lift (k a) - yield b +data Lifetime + = OneTime + | KeepAlive !Command -------------------------------------------------------------------------------- -append :: Operation o -> Operation o -> Operation o -append start right = go start - where - go cur = - case cur of - Return x -> absurd x - Effect m -> Effect (fmap go m) - Step step -> - case step of - Yield o next -> - Step $ Yield o (append next right) - Await k instr failed -> - Step $ Await (\i -> append (k i) right) instr (append failed right) - Stop -> - right +data Loop a + = Loop + | Break !a -------------------------------------------------------------------------------- -stepping :: Operation a -> Code o (a, Operation a) -stepping = go - where - go cur = - case cur of - Return x -> absurd x - Effect m -> Effect (fmap go m) - Step step -> - case step of - Yield a next -> pure (a, next) - Await k instr failed -> Step $ Await (go . k) instr (go failed) - Stop -> stop +data LoopS s a + = LoopS !s + | BreakS !a -------------------------------------------------------------------------------- -unfolding :: (Maybe a -> Code o (Operation a)) -> Operation o -unfolding k = k Nothing >>= go +keepLooping :: Monad m => m (Loop a) -> m a +keepLooping action + = go where - go cur = do - (a, next) <- stepping cur - newState <- k (Just a) - go (append next newState) + go = do result <- action + case result of + Loop -> go + Break a -> pure a -------------------------------------------------------------------------------- -repeatedly :: Functor m => Machine k o m x -> Machine k o m r -repeatedly start = go start +keepLoopingS :: Monad m => s -> (s -> m (LoopS s a)) -> m a +keepLoopingS seed action + = go seed where - go (Return _) = go start - go (Effect m) = Effect (fmap go m) - go (Step step) = - case step of - Yield o next -> Step $ Yield o (go next) - Await k instr failed -> Step $ Await (go . k) instr (go failed) - Stop -> stop - --------------------------------------------------------------------------------- -(<~) :: Monad m => Process m a b -> Machine k a m r -> Machine k b m r -mp <~ mb = - case mp of - Return x -> absurd x - Effect m -> Effect (fmap (<~ mb) m) - Step consumer -> - case consumer of - Yield c next -> Step $ Yield c (next <~ mb) - Stop -> stop - Await k Same failed -> - case mb of - Return _ -> failed <~ stop - Effect m -> Effect (fmap (Step consumer <~) m) - Step producer -> - case producer of - Yield b next -> k b <~ next - Await kb instr kfailed -> - Step (Await ((mp <~) . kb) instr (mp <~ kfailed)) - Stop -> failed <~ stop - --------------------------------------------------------------------------------- --- | Asks for a unused 'UUID'. -freshId :: Code o UUID -freshId = Step $ Await pure NeedUUID (Step Stop) - --------------------------------------------------------------------------------- --- | Raises an 'OperationError'. -failure :: OperationError -> Code o a -failure = lift . Failed - --------------------------------------------------------------------------------- --- | Asks to resume the interpretation from the beginning. -retry :: Code o a -retry = lift Retry - --------------------------------------------------------------------------------- --- | Sends a package to the server and wait for a response. -sendRemote :: Payload -> Code o Package -sendRemote p = Step $ Await pure (NeedRemote p) (Step Stop) - --------------------------------------------------------------------------------- --- | Waits package from the server. -waitRemote :: UUID -> Code o (Maybe Package) -waitRemote c = Step $ Await pure (WaitRemote c) (Step Stop) - --------------------------------------------------------------------------------- -construct :: Code o a -> Operation o -construct m = m >> stop - --------------------------------------------------------------------------------- --- | Like 'request' except it discards the correlation id of the network --- exchange. -send :: (Encode req, Decode resp) - => Command - -> Command - -> Maybe Credentials - -> req - -> Code o resp -send reqCmd expCmd cred req = do - let dat = runPut $ encodeMessage req - payload = Payload reqCmd dat cred - pkg <- sendRemote payload - let gotCmd = packageCmd pkg - - when (gotCmd /= expCmd) - (invalidServerResponse expCmd gotCmd) - - case runGet decodeMessage (packageData pkg) of - Left e -> protobufDecodingError e - Right resp -> return resp - --------------------------------------------------------------------------------- -data Expect o where - Expect :: Decode resp => Command -> (UUID -> resp -> Code o ()) -> Expect o - --------------------------------------------------------------------------------- --- | Runs the first expection that matches. -runFirstMatch :: Package -> [Expect o] -> Code o () -runFirstMatch _ [] = invalidOperation "No expectation was fulfilled" -runFirstMatch pkg (Expect cmd k:rest) - | packageCmd pkg /= cmd = runFirstMatch pkg rest - | otherwise = - case runGet decodeMessage (packageData pkg) of - Left e -> protobufDecodingError e - Right resp -> k (packageCorrelation pkg) resp - --------------------------------------------------------------------------------- --- | Sends a message to remote server. It returns the expected deserialized --- message along with the correlation id of the network exchange. -request :: Encode req - => Command - -> Maybe Credentials - -> req - -> [Expect o] - -> Code o () -request reqCmd cred rq exps = do - let dat = runPut $ encodeMessage rq - payload = Payload reqCmd dat cred - pkg <- sendRemote payload - runFirstMatch pkg exps - --------------------------------------------------------------------------------- --- | @waitForElse uuid alternative expects@ Waits for a message from the server --- at the given /uuid/. If the connection has been reset in the meantime, it --- will use /alternative/. -waitForOr :: UUID -> Code o () -> [Expect o] -> Code o () -waitForOr pid alt exps = - waitRemote pid >>= \case - Nothing -> alt - Just pkg -> - runFirstMatch pkg exps - --------------------------------------------------------------------------------- --- | Raises 'WrongExpectedVersion' exception. -wrongVersion :: Text -> ExpectedVersion -> Code o a -wrongVersion stream ver = failure (WrongExpectedVersion stream ver) - --------------------------------------------------------------------------------- --- | Raises 'StreamDeleted' exception. -streamDeleted :: StreamName -> Code o a -streamDeleted stream = failure (StreamDeleted stream) - --------------------------------------------------------------------------------- --- | Raises 'InvalidTransaction' exception. -invalidTransaction :: Code o a -invalidTransaction = failure InvalidTransaction - --------------------------------------------------------------------------------- --- | Raises 'AccessDenied' exception. -accessDenied :: StreamId t -> Code o a -accessDenied = failure . AccessDenied - --------------------------------------------------------------------------------- --- | Raises 'ProtobufDecodingError' exception. -protobufDecodingError :: String -> Code o a -protobufDecodingError = failure . ProtobufDecodingError - --------------------------------------------------------------------------------- --- | Raises 'ServerError' exception. -serverError :: Maybe Text -> Code o a -serverError = failure . ServerError - --------------------------------------------------------------------------------- --- | Raises 'InvalidServerResponse' exception. -invalidServerResponse :: Command -> Command -> Code o a -invalidServerResponse expe got = failure $ InvalidServerResponse expe got - - --------------------------------------------------------------------------------- -invalidOperation :: Text -> Code o a -invalidOperation = failure . InvalidOperation + go cur + = do result <- action cur + case result of + LoopS next + -> go next + BreakS a + -> pure a 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 @@ -19,16 +19,20 @@ -------------------------------------------------------------------------------- import Data.Int import Data.Maybe +import Data.ProtocolBuffers -------------------------------------------------------------------------------- +import Database.EventStore.Internal.Command +import Database.EventStore.Internal.Communication (Transmit(..)) +import Database.EventStore.Internal.Control (publishWith) +import Database.EventStore.Internal.Exec (Exec) import Database.EventStore.Internal.Operation -import Database.EventStore.Internal.Operation.Read.Common -import Database.EventStore.Internal.Operation.ReadAllEvents -import Database.EventStore.Internal.Operation.ReadStreamEvents -import Database.EventStore.Internal.Operation.Volatile +import qualified Database.EventStore.Internal.Operation.ReadAllEvents.Message as ReadAll +import qualified Database.EventStore.Internal.Operation.ReadStreamEvents.Message as ReadStream import Database.EventStore.Internal.Prelude import Database.EventStore.Internal.Settings import Database.EventStore.Internal.Stream +import Database.EventStore.Internal.Subscription.Message import Database.EventStore.Internal.Subscription.Types import Database.EventStore.Internal.Types @@ -37,77 +41,212 @@ defaultBatchSize = 500 -------------------------------------------------------------------------------- -streamNotFound :: Text -> OperationError -streamNotFound stream = StreamNotFound $ StreamName stream +data State s + = Init s + | Catchup UUID UUID s + | Live UUID s -------------------------------------------------------------------------------- -fetchStream :: Settings - -> Text -- Stream name. - -> Int32 -- Batch size. - -> Bool -- Resolve link tos. - -> Maybe Credentials - -> EventNumber - -> Operation (Slice EventNumber) -fetchStream setts stream batch tos cred (EventNumber n) = - traversing go <~ readStreamEvents setts Forward stream n batch tos cred - where - go outcome = - fromReadResult stream outcome pure +createReadPkg + :: Settings + -> StreamId t + -> t + -> Int32 -- Batch size + -> Bool -- Resolve links + -> Maybe Credentials + -> IO Package +createReadPkg setts (StreamName stream) evtNum batch tos cred + = let + req = + ReadStream.newRequest + stream + (eventNumberToInt64 evtNum) + batch + tos + (s_requireMaster setts) in + createPkg readStreamEventsForwardCmd cred req +createReadPkg setts All pos batch tos cred + = let + req = + ReadAll.newRequest + (positionCommit pos) + (positionPrepare pos) + batch + tos + (s_requireMaster setts) in + createPkg readAllEventsForwardCmd cred req -------------------------------------------------------------------------------- -fetchAll :: Settings - -> Int32 -- Batch size. - -> Bool -- Resolve link tos. - -> Maybe Credentials - -> Position - -> Operation (Slice Position) -fetchAll setts batch tos cred (Position com pre) = - traversing pure <~ readAllEvents setts com pre batch tos Forward cred +catchup + :: Settings + -> Exec + -> StreamId t + -> t + -> Bool -- Resolve link tos. + -> Maybe Int32 -- Batch size. + -> Maybe Credentials + -> IO (TVar (Maybe UUID), Chan SubAction) +catchup setts exec streamId from tos batchSiz cred + = do m <- mailboxNew + subM <- newChan + var <- newTVarIO Nothing + _ <- async $ keepLoopingS (Init from) $ \case + Init pos + -> do let subReq = subscribeToStream stream tos + subPkg <- createPkg subscribeToStreamCmd cred subReq + readPkg <- createReadPkg setts streamId pos batch tos cred --------------------------------------------------------------------------------- -sourceStream :: (t -> Operation (Slice t)) - -> t - -> Operation SubAction -sourceStream fetch start = unfolding go - where - go Nothing = - pure (fetch start) - go (Just s) = do - traverse_ (yield . Submit) (sliceEvents s) + publishWith exec (Transmit m (KeepAlive subscriptionDroppedCmd) subPkg) + publishWith exec (Transmit m OneTime readPkg) + let theSubId = packageCorrelation subPkg - case sliceNext s of - Just next -> pure (fetch next) - Nothing -> stop + atomically $ writeTVar var (Just theSubId) --------------------------------------------------------------------------------- -catchup :: forall t. Settings - -> StreamId t - -> t - -> Bool -- Resolve link tos. - -> Maybe Int32 -- Batch size. - -> Maybe Credentials - -> Operation SubAction -catchup setts streamId from tos batchSiz cred = - append (sourceStream iteratee from) (volatile streamId tos cred) - where - batch = fromMaybe defaultBatchSize batchSiz + pure $ LoopS (Catchup theSubId (packageCorrelation readPkg) pos) - iteratee :: t -> Operation (Slice t) - iteratee = - case streamId of - StreamName n -> fetchStream setts n batch tos cred - All -> fetchAll setts batch tos cred + unchanged@(Catchup theSubId readId pos) + -> do outcome <- mailboxRead m + case outcome of + Left e + -> case e of + ConnectionHasDropped + -> pure $ LoopS (Init pos) + _ -> BreakS () <$ writeChan subM (Dropped SubAborted) --------------------------------------------------------------------------------- -fromReadResult :: Text - -> ReadResult EventNumber a - -> (a -> Execution x) - -> Execution x -fromReadResult stream res k = - case res of - ReadNoStream -> Failed $ streamNotFound stream - ReadStreamDeleted s -> Failed $ StreamDeleted s - ReadNotModified -> Failed $ ServerError Nothing - ReadError e -> Failed $ ServerError e - ReadAccessDenied s -> Failed $ AccessDenied s - ReadSuccess ss -> k ss + Right respPkg + | theSubId == packageCorrelation respPkg + && packageCmd respPkg == subscriptionDroppedCmd + -> let Right resp = decodePkg respPkg + reason = fromMaybe D_Unsubscribed (getField $ dropReason resp) + subReason = toSubDropReason reason in + BreakS () <$ writeChan subM (Dropped subReason) + + | theSubId == packageCorrelation respPkg + && packageCmd respPkg == subscriptionConfirmationCmd + -> let Right resp = decodePkg respPkg + lcp = getField $ subscribeLastCommitPos resp + len = getField $ subscribeLastEventNumber resp + details = + SubDetails + { subId = theSubId + , subCommitPos = lcp + , subLastEventNum = len + , subSubId = Nothing + } in + LoopS unchanged <$ writeChan subM (Confirmed details) + + | theSubId == packageCorrelation respPkg + -> pure $ LoopS unchanged + + | readId == packageCorrelation respPkg + -> case streamId of + StreamName _ + -> do let + Right resp = decodePkg respPkg + r = getField $ ReadStream._result resp + es = getField $ ReadStream._events resp + evts = fmap newResolvedEvent es + eos = getField $ ReadStream._endOfStream resp + nxt = getField $ ReadStream._nextNumber resp + case r of + ReadStream.NO_STREAM + -> pure $ LoopS (Live theSubId pos) + + ReadStream.SUCCESS + -> do traverse_ (writeChan subM . Submit) evts + if eos + then + pure $ LoopS (Live theSubId (rawEventNumber nxt)) + else + do newReadPkg <- createReadPkg setts streamId (rawEventNumber nxt) batch tos cred + let newReadId = packageCorrelation newReadPkg + + publishWith exec (Transmit m OneTime newReadPkg) + pure $ LoopS (Catchup theSubId newReadId (rawEventNumber nxt)) + + -- TODO - Do we have to close the subscription? + -- Pretty sure the subcription has failed already at + -- this point. + _ -> BreakS () <$ writeChan subM (Dropped SubAborted) + + All + -> do let + Right resp = decodePkg respPkg + r = getField $ ReadAll._Result resp + nc_pos = getField $ ReadAll._NextCommitPosition resp + np_pos = getField $ ReadAll._NextPreparePosition resp + es = getField $ ReadAll._Events resp + evts = fmap newResolvedEventFromBuf es + eos = null evts + n_pos = Position nc_pos np_pos + + case fromMaybe ReadAll.SUCCESS r of + ReadAll.SUCCESS + -> do traverse_ (writeChan subM . Submit) evts + if eos + then + pure $ LoopS (Live theSubId n_pos) + else + do newReadPkg <- createReadPkg setts streamId n_pos batch tos cred + let newReadId = packageCorrelation newReadPkg + + publishWith exec (Transmit m OneTime newReadPkg) + pure $ LoopS (Catchup theSubId newReadId n_pos) + + -- TODO - Do we have to close the subscription? + -- Pretty sure the subcription has failed already at + -- this point. + _ -> BreakS () <$ writeChan subM (Dropped SubAborted) + | otherwise + -> pure $ LoopS unchanged + + unchanged@(Live theSubId pos) + -> do outcome <- mailboxRead m + case outcome of + Left e + -> case e of + ConnectionHasDropped + -> pure $ LoopS (Init pos) + _ -> BreakS () <$ writeChan subM (Dropped SubAborted) + + Right respPkg + | theSubId == packageCorrelation respPkg + && packageCmd respPkg == subscriptionDroppedCmd + -> let Right resp = decodePkg respPkg + reason = fromMaybe D_Unsubscribed (getField $ dropReason resp) + subReason = toSubDropReason reason in + BreakS () <$ writeChan subM (Dropped subReason) + | theSubId == packageCorrelation respPkg + && packageCmd respPkg == subscriptionConfirmationCmd + -> let Right resp = decodePkg respPkg + lcp = getField $ subscribeLastCommitPos resp + len = getField $ subscribeLastEventNumber resp + details = + SubDetails + { subId = theSubId + , subCommitPos = lcp + , subLastEventNum = len + , subSubId = Nothing + } in + LoopS unchanged <$ writeChan subM (Confirmed details) + | theSubId == packageCorrelation respPkg + && packageCmd respPkg == streamEventAppearedCmd + -> let + Right resp = decodePkg respPkg + evt = newResolvedEventFromBuf $ getField $ streamResolvedEvent resp + nextState = + case streamId of + StreamName _ + -> let nxt = resolvedEventOriginalEventNumber evt + in Live theSubId (rawEventNumber nxt) + All + -> let Just nxtPos = resolvedEventPosition evt + in Live theSubId nxtPos in + LoopS nextState <$ writeChan subM (Submit evt) + | otherwise + -> pure $ LoopS unchanged + + pure (var, subM) + where + batch = fromMaybe defaultBatchSize batchSiz + stream = streamIdRaw streamId diff --git a/Database/EventStore/Internal/Operation/DeleteStream.hs b/Database/EventStore/Internal/Operation/DeleteStream.hs --- a/Database/EventStore/Internal/Operation/DeleteStream.hs +++ b/Database/EventStore/Internal/Operation/DeleteStream.hs @@ -26,6 +26,9 @@ -------------------------------------------------------------------------------- import Database.EventStore.Internal.Command +import Database.EventStore.Internal.Communication (Transmit(..)) +import Database.EventStore.Internal.Control (publishWith) +import Database.EventStore.Internal.Exec import Database.EventStore.Internal.Operation import Database.EventStore.Internal.Operation.DeleteStream.Message import Database.EventStore.Internal.Prelude @@ -37,30 +40,44 @@ -- | Returned after deleting a stream. 'Position' of the write. newtype DeleteResult = DeleteResult Position deriving (Eq, Show) --------------------------------------------------------------------------------- --- | Delete a regular stream operation. -deleteStream :: Settings - -> Text - -> ExpectedVersion - -> Maybe Bool - -> Maybe Credentials - -> Operation DeleteResult -deleteStream Settings{..} s v hard cred = construct $ do - let msg = newRequest s (expVersionInt64 v) s_requireMaster hard - resp <- send deleteStreamCmd deleteStreamCompletedCmd cred msg - let r = getField $ _result resp - com_pos = getField $ _commitPosition resp - prep_pos = getField $ _preparePosition resp - com_pos_int = fromMaybe (-1) com_pos - prep_pos_int = fromMaybe (-1) prep_pos - pos = Position com_pos_int prep_pos_int - res = DeleteResult pos - case r of - OP_SUCCESS -> yield res - OP_PREPARE_TIMEOUT -> retry - OP_FORWARD_TIMEOUT -> retry - OP_COMMIT_TIMEOUT -> retry - OP_WRONG_EXPECTED_VERSION -> wrongVersion s v - OP_STREAM_DELETED -> streamDeleted $ StreamName s - OP_INVALID_TRANSACTION -> invalidTransaction - OP_ACCESS_DENIED -> accessDenied (StreamName s) +------------------------------------------------------------------------------- +deleteStream + :: Settings + -> Exec + -> Text + -> ExpectedVersion + -> Maybe Bool + -> Maybe Credentials + -> IO (Async DeleteResult) +deleteStream setts exec stream version hard creds + = do m <- mailboxNew + async $ + do let req = newRequest stream (expVersionInt64 version) (s_requireMaster setts) hard + + pkg <- createPkg deleteStreamCmd creds req + + keepLooping $ do + publishWith exec (Transmit m OneTime pkg) + outcome <- mailboxReadDecoded m + case outcome of + Left e + -> throw e + Right resp + -> let r = getField $ _result resp + com_pos = getField $ _commitPosition resp + prep_pos = getField $ _preparePosition resp + com_pos_int = fromMaybe (-1) com_pos + prep_pos_int = fromMaybe (-1) prep_pos + pos = Position com_pos_int prep_pos_int + res = DeleteResult pos in + case r of + OP_SUCCESS -> pure $ Break res + OP_PREPARE_TIMEOUT -> pure Loop + OP_FORWARD_TIMEOUT -> pure Loop + OP_COMMIT_TIMEOUT -> pure Loop + OP_WRONG_EXPECTED_VERSION -> throw $ WrongExpectedVersion stream version + OP_STREAM_DELETED -> throw $ StreamDeleted $ StreamName stream + OP_INVALID_TRANSACTION -> throw InvalidTransaction + OP_ACCESS_DENIED -> throw $ AccessDenied (StreamName stream) + + diff --git a/Database/EventStore/Internal/Operation/Persist.hs b/Database/EventStore/Internal/Operation/Persist.hs --- a/Database/EventStore/Internal/Operation/Persist.hs +++ b/Database/EventStore/Internal/Operation/Persist.hs @@ -17,6 +17,9 @@ -------------------------------------------------------------------------------- import Database.EventStore.Internal.Command +import Database.EventStore.Internal.Communication (Transmit(..)) +import Database.EventStore.Internal.Control (publishWith) +import Database.EventStore.Internal.Exec (Exec) import Database.EventStore.Internal.Operation import Database.EventStore.Internal.Prelude import Database.EventStore.Internal.Settings @@ -25,57 +28,51 @@ import Database.EventStore.Internal.Types -------------------------------------------------------------------------------- -persist :: Text -> Text -> Int32 -> Maybe Credentials -> Operation SubAction -persist grp stream bufSize cred = - construct (issueRequest grp stream bufSize cred) - --------------------------------------------------------------------------------- -issueRequest :: Text -> Text -> Int32 -> Maybe Credentials -> Code SubAction () -issueRequest grp stream bufSize cred = do - let req = _connectToPersistentSubscription grp stream bufSize - request connectToPersistentSubscriptionCmd cred req - [ Expect subscriptionDroppedCmd $ \_ d -> - handleDropped d - , Expect persistentSubscriptionConfirmationCmd $ \sid c -> do - let lcp = getField $ pscLastCommitPos c - subSubId = getField $ pscId c - len = getField $ pscLastEvtNumber c - details = - SubDetails - { subId = sid - , subCommitPos = lcp - , subLastEventNum = len - , subSubId = Just subSubId - } - yield (Confirmed details) - live sid - ] - --------------------------------------------------------------------------------- -eventAppeared :: PersistentSubscriptionStreamEventAppeared -> Code SubAction () -eventAppeared e = do - let evt = newResolvedEvent $ getField $ psseaEvt e - yield (Submit evt) - --------------------------------------------------------------------------------- -live :: UUID -> Code SubAction () -live subscriptionId = loop - where - loop = - waitForOr subscriptionId connectionReset - [ Expect subscriptionDroppedCmd $ \_ d -> - handleDropped d - , Expect persistentSubscriptionStreamEventAppearedCmd $ \_ e -> do - eventAppeared e - loop - ] - --------------------------------------------------------------------------------- -connectionReset :: Code SubAction () -connectionReset = yield ConnectionReset +persist + :: Exec + -> Text + -> Text + -> Int32 + -> Maybe Credentials + -> IO (UUID, TVar (Maybe Text), Chan SubAction) +persist exec grp stream bufSize cred + = do m <- mailboxNew + subM <- newChan + var <- newTVarIO Nothing + let req = _connectToPersistentSubscription grp stream bufSize + pkg <- createPkg connectToPersistentSubscriptionCmd cred req + let theSubId = packageCorrelation pkg + publishWith exec (Transmit m (KeepAlive subscriptionDroppedCmd) pkg) + _ <- async $ keepLooping $ + do outcome <- mailboxRead m + case outcome of + Left _ + -> Break () <$ writeChan subM (Dropped SubAborted) + Right respPkg + | packageCmd respPkg == subscriptionDroppedCmd + -> let Right resp = decodePkg respPkg + reason = fromMaybe D_Unsubscribed (getField $ dropReason resp) + subReason = toSubDropReason reason in + Break () <$ writeChan subM (Dropped subReason) + | packageCmd respPkg == persistentSubscriptionConfirmationCmd + -> do let Right resp = decodePkg respPkg + lcp = getField $ pscLastCommitPos resp + subSubId = getField $ pscId resp + len = getField $ pscLastEvtNumber resp + details = + SubDetails + { subId = theSubId + , subCommitPos = lcp + , subLastEventNum = len + , subSubId = Just subSubId + } + atomically $ writeTVar var (Just subSubId) + Loop <$ writeChan subM (Confirmed details) + | packageCmd respPkg == persistentSubscriptionStreamEventAppearedCmd + -> let Right resp = decodePkg respPkg + evt = newResolvedEvent $ getField $ psseaEvt resp in + Loop <$ writeChan subM (Submit evt) + | otherwise + -> pure Loop --------------------------------------------------------------------------------- -handleDropped :: SubscriptionDropped -> Code SubAction () -handleDropped d = do - let reason = fromMaybe D_Unsubscribed (getField $ dropReason d) - yield (Dropped $ toSubDropReason reason) + pure (theSubId, var, subM) diff --git a/Database/EventStore/Internal/Operation/PersistOperations.hs b/Database/EventStore/Internal/Operation/PersistOperations.hs --- a/Database/EventStore/Internal/Operation/PersistOperations.hs +++ b/Database/EventStore/Internal/Operation/PersistOperations.hs @@ -20,6 +20,9 @@ -------------------------------------------------------------------------------- import Database.EventStore.Internal.Command +import Database.EventStore.Internal.Communication (Transmit(..)) +import Database.EventStore.Internal.Control (publishWith) +import Database.EventStore.Internal.Exec (Exec) import Database.EventStore.Internal.Operation import Database.EventStore.Internal.Prelude import Database.EventStore.Internal.Subscription.Message @@ -28,55 +31,76 @@ import Database.EventStore.Internal.Types -------------------------------------------------------------------------------- -persistOperation :: Text - -> Text - -> Maybe Credentials - -> PersistAction - -> Operation (Maybe PersistActionException) -persistOperation grp stream cred tpe = construct go - where - go = - case tpe of - PersistCreate ss -> do - let req = _createPersistentSubscription grp stream ss - resp <- send createPersistentSubscriptionCmd - createPersistentSubscriptionCompletedCmd cred req - let result = createRException $ getField $ cpscResult resp - yield result - PersistUpdate ss -> do - let req = _updatePersistentSubscription grp stream ss - resp <- send updatePersistentSubscriptionCmd - updatePersistentSubscriptionCompletedCmd cred req - let result = updateRException $ getField $ upscResult resp - yield result - PersistDelete -> do - let req = _deletePersistentSubscription grp stream - resp <- send deletePersistentSubscriptionCmd - deletePersistentSubscriptionCompletedCmd cred req - let result = deleteRException $ getField $ dpscResult resp - yield result +persistOperation + :: Exec + -> Text + -> Text + -> Maybe Credentials + -> PersistAction + -> IO (Async (Maybe PersistActionException)) +persistOperation exec grp stream cred tpe + = do m <- mailboxNew + async $ + case tpe of + PersistCreate ss + -> do let req = _createPersistentSubscription grp stream ss + pkg <- createPkg createPersistentSubscriptionCmd cred req + publishWith exec (Transmit m OneTime pkg) + outcome <- mailboxReadDecoded m + case outcome of + Left e + -> throw e + Right resp + -> pure $ createRException $ getField $ cpscResult resp + PersistUpdate ss + -> do let req = _updatePersistentSubscription grp stream ss + pkg <- createPkg updatePersistentSubscriptionCmd cred req + publishWith exec (Transmit m OneTime pkg) + outcome <- mailboxReadDecoded m + case outcome of + Left e + -> throw e + Right resp + -> pure $ updateRException $ getField $ upscResult resp + PersistDelete + -> do let req = _deletePersistentSubscription grp stream + pkg <- createPkg deletePersistentSubscriptionCmd cred req + publishWith exec (Transmit m OneTime pkg) + outcome <- mailboxReadDecoded m + case outcome of + Left e + -> throw e + Right resp + -> pure $ deleteRException $ getField $ dpscResult resp -------------------------------------------------------------------------------- -createPersist :: Text - -> Text - -> PersistentSubscriptionSettings - -> Maybe Credentials - -> Operation (Maybe PersistActionException) -createPersist grp stream ss cred = - persistOperation grp stream cred (PersistCreate ss) +createPersist + :: Exec + -> Text + -> Text + -> PersistentSubscriptionSettings + -> Maybe Credentials + -> IO (Async (Maybe PersistActionException)) +createPersist exec grp stream ss cred + = persistOperation exec grp stream cred (PersistCreate ss) -------------------------------------------------------------------------------- -updatePersist :: Text - -> Text - -> PersistentSubscriptionSettings - -> Maybe Credentials - -> Operation (Maybe PersistActionException) -updatePersist grp stream ss cred = - persistOperation grp stream cred (PersistUpdate ss) +updatePersist + :: Exec + -> Text + -> Text + -> PersistentSubscriptionSettings + -> Maybe Credentials + -> IO (Async (Maybe PersistActionException)) +updatePersist exec grp stream ss cred + = persistOperation exec grp stream cred (PersistUpdate ss) -------------------------------------------------------------------------------- -deletePersist :: Text - -> Text - -> Maybe Credentials - -> Operation (Maybe PersistActionException) -deletePersist grp stream cred = persistOperation grp stream cred PersistDelete +deletePersist + :: Exec + -> Text + -> Text + -> Maybe Credentials + -> IO (Async (Maybe PersistActionException)) +deletePersist exec grp stream cred + = persistOperation exec grp stream cred PersistDelete diff --git a/Database/EventStore/Internal/Operation/ReadAllEvents.hs b/Database/EventStore/Internal/Operation/ReadAllEvents.hs --- a/Database/EventStore/Internal/Operation/ReadAllEvents.hs +++ b/Database/EventStore/Internal/Operation/ReadAllEvents.hs @@ -22,6 +22,9 @@ -------------------------------------------------------------------------------- import Database.EventStore.Internal.Command +import Database.EventStore.Internal.Control (publishWith) +import Database.EventStore.Internal.Communication (Transmit(..)) +import Database.EventStore.Internal.Exec (Exec) import Database.EventStore.Internal.Operation import Database.EventStore.Internal.Operation.Read.Common import Database.EventStore.Internal.Operation.ReadAllEvents.Message @@ -32,35 +35,43 @@ -------------------------------------------------------------------------------- -- | Batch read on $all stream operation. -readAllEvents :: Settings - -> Int64 - -> Int64 - -> Int32 - -> Bool - -> ReadDirection - -> Maybe Credentials - -> Operation AllSlice -readAllEvents Settings{..} c_pos p_pos max_c tos dir cred = construct $ do - let msg = newRequest c_pos p_pos max_c tos s_requireMaster - cmd = case dir of - Forward -> readAllEventsForwardCmd - Backward -> readAllEventsBackwardCmd +readAllEvents + :: Settings + -> Exec + -> Int64 + -> Int64 + -> Int32 + -> Bool + -> ReadDirection + -> Maybe Credentials + -> IO (Async AllSlice) +readAllEvents Settings{..} exec c_pos p_pos max_c tos dir cred + = do m <- mailboxNew + async $ + do let req = newRequest c_pos p_pos max_c tos s_requireMaster + cmd = + case dir of + Forward -> readAllEventsForwardCmd + Backward -> readAllEventsBackwardCmd - resp_cmd = case dir of - Forward -> readAllEventsForwardCompletedCmd - Backward -> readAllEventsBackwardCompletedCmd - resp <- send cmd resp_cmd cred msg - let r = getField $ _Result resp - err = getField $ _Error resp - nc_pos = getField $ _NextCommitPosition resp - np_pos = getField $ _NextPreparePosition resp - es = getField $ _Events resp - evts = fmap newResolvedEventFromBuf es - eos = null evts - n_pos = Position nc_pos np_pos - slice = - if eos then SliceEndOfStream else Slice evts (Just n_pos) - case fromMaybe SUCCESS r of - ERROR -> serverError err - ACCESS_DENIED -> accessDenied All - _ -> yield slice + pkg <- createPkg cmd cred req + publishWith exec (Transmit m OneTime pkg) + outcome <- mailboxReadDecoded m + case outcome of + Left e + -> throw e + Right resp + -> let r = getField $ _Result resp + err = getField $ _Error resp + nc_pos = getField $ _NextCommitPosition resp + np_pos = getField $ _NextPreparePosition resp + es = getField $ _Events resp + evts = fmap newResolvedEventFromBuf es + eos = null evts + n_pos = Position nc_pos np_pos + slice = + if eos then SliceEndOfStream else Slice evts (Just n_pos) in + case fromMaybe SUCCESS r of + ERROR -> throw $ ServerError err + ACCESS_DENIED -> throw $ AccessDenied All + _ -> pure slice diff --git a/Database/EventStore/Internal/Operation/ReadEvent.hs b/Database/EventStore/Internal/Operation/ReadEvent.hs --- a/Database/EventStore/Internal/Operation/ReadEvent.hs +++ b/Database/EventStore/Internal/Operation/ReadEvent.hs @@ -24,6 +24,9 @@ -------------------------------------------------------------------------------- import Database.EventStore.Internal.Command +import Database.EventStore.Internal.Control (publishWith) +import Database.EventStore.Internal.Communication (Transmit(..)) +import Database.EventStore.Internal.Exec (Exec) import Database.EventStore.Internal.Operation import Database.EventStore.Internal.Operation.ReadEvent.Message import Database.EventStore.Internal.Operation.Read.Common @@ -47,24 +50,36 @@ -------------------------------------------------------------------------------- -- | Read a specific event given event number operation. -readEvent :: Settings - -> Text - -> Int64 - -> Bool - -> Maybe Credentials - -> Operation (ReadResult EventNumber ReadEvent) -readEvent Settings{..} s evtn tos cred = construct $ do - let msg = newRequest s evtn tos s_requireMaster - resp <- send readEventCmd readEventCompletedCmd cred msg - let r = getField $ _result resp - evt = newResolvedEvent $ getField $ _indexedEvent resp - err = getField $ _error resp - not_found = ReadSuccess $ ReadEventNotFound s evtn - found = ReadSuccess $ ReadEvent s evtn evt - case r of - NOT_FOUND -> yield not_found - NO_STREAM -> yield ReadNoStream - STREAM_DELETED -> yield $ ReadStreamDeleted $ StreamName s - ERROR -> yield (ReadError err) - ACCESS_DENIED -> yield $ ReadAccessDenied $ StreamName s - SUCCESS -> yield found +readEvent + :: Settings + -> Exec + -> Text + -> Int64 + -> Bool + -> Maybe Credentials + -> IO (Async (ReadResult EventNumber ReadEvent)) +readEvent Settings{..} exec stream evtn tos creds + = do m <- mailboxNew + async $ + do let req = newRequest stream evtn tos s_requireMaster + pkg <- createPkg readEventCmd creds req + publishWith exec (Transmit m OneTime pkg) + outcome <- mailboxReadDecoded m + case outcome of + Left e + -> throw e + Right resp + -> let r = getField $ _result resp + evt = newResolvedEvent $ getField $ _indexedEvent resp + err = getField $ _error resp + notFound = ReadSuccess $ ReadEventNotFound stream evtn + found = ReadSuccess $ ReadEvent stream evtn evt in + case r of + NOT_FOUND -> pure notFound + NO_STREAM -> pure ReadNoStream + STREAM_DELETED -> pure $ ReadStreamDeleted $ StreamName stream + ERROR -> pure (ReadError err) + ACCESS_DENIED -> pure $ ReadAccessDenied $ StreamName stream + SUCCESS -> pure found + + diff --git a/Database/EventStore/Internal/Operation/ReadStreamEvents.hs b/Database/EventStore/Internal/Operation/ReadStreamEvents.hs --- a/Database/EventStore/Internal/Operation/ReadStreamEvents.hs +++ b/Database/EventStore/Internal/Operation/ReadStreamEvents.hs @@ -22,6 +22,9 @@ -------------------------------------------------------------------------------- import Database.EventStore.Internal.Command +import Database.EventStore.Internal.Control (publishWith) +import Database.EventStore.Internal.Communication (Transmit(..)) +import Database.EventStore.Internal.Exec (Exec) import Database.EventStore.Internal.Operation import Database.EventStore.Internal.Operation.Read.Common import Database.EventStore.Internal.Operation.ReadStreamEvents.Message @@ -32,40 +35,46 @@ -------------------------------------------------------------------------------- -- | Batch read from a regular stream operation. -readStreamEvents :: Settings - -> ReadDirection - -> Text - -> Int64 - -> Int32 - -> Bool - -> Maybe Credentials - -> Operation (ReadResult EventNumber StreamSlice) -readStreamEvents Settings{..} dir s st cnt tos cred = construct $ do - let req_cmd = - case dir of - Forward -> readStreamEventsForwardCmd - Backward -> readStreamEventsBackwardCmd - resp_cmd = - case dir of - Forward -> readStreamEventsForwardCompletedCmd - Backward -> readStreamEventsBackwardCompletedCmd +readStreamEvents + :: Settings + -> Exec + -> ReadDirection + -> Text + -> Int64 + -> Int32 + -> Bool + -> Maybe Credentials + -> IO (Async (ReadResult EventNumber StreamSlice)) +readStreamEvents Settings{..} exec dir stream st cnt tos cred + = do m <- mailboxNew + async $ + do let reqCmd = + case dir of + Forward -> readStreamEventsForwardCmd + Backward -> readStreamEventsBackwardCmd - msg = newRequest s st cnt tos s_requireMaster - resp <- send req_cmd resp_cmd cred msg - let r = getField $ _result resp - es = getField $ _events resp - evts = fmap newResolvedEvent es - err = getField $ _error resp - eos = getField $ _endOfStream resp - nxt = getField $ _nextNumber resp - found = - if null evts && eos - then SliceEndOfStream - else Slice evts (if eos then Nothing else Just $ EventNumber nxt) - case r of - NO_STREAM -> yield ReadNoStream - STREAM_DELETED -> yield $ ReadStreamDeleted $ StreamName s - NOT_MODIFIED -> yield ReadNotModified - ERROR -> yield (ReadError err) - ACCESS_DENIED -> yield $ ReadAccessDenied $ StreamName s - SUCCESS -> yield (ReadSuccess found) + req = newRequest stream st cnt tos s_requireMaster + pkg <- createPkg reqCmd cred req + publishWith exec (Transmit m OneTime pkg) + outcome <- mailboxReadDecoded m + case outcome of + Left e + -> throw e + Right resp + -> let r = getField $ _result resp + es = getField $ _events resp + evts = fmap newResolvedEvent es + err = getField $ _error resp + eos = getField $ _endOfStream resp + nxt = getField $ _nextNumber resp + found = + if null evts && eos + then SliceEndOfStream + else Slice evts (if eos then Nothing else Just $ EventNumber nxt) in + case r of + NO_STREAM -> pure ReadNoStream + STREAM_DELETED -> pure $ ReadStreamDeleted $ StreamName stream + NOT_MODIFIED -> pure ReadNotModified + ERROR -> pure (ReadError err) + ACCESS_DENIED -> pure $ ReadAccessDenied $ StreamName stream + SUCCESS -> pure (ReadSuccess found) 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 @@ -26,6 +26,7 @@ import Data.Aeson (decode) -------------------------------------------------------------------------------- +import Database.EventStore.Internal.Exec (Exec) import Database.EventStore.Internal.Operation import Database.EventStore.Internal.Operation.Read.Common import Database.EventStore.Internal.Operation.ReadEvent @@ -41,33 +42,37 @@ -------------------------------------------------------------------------------- -- | Read stream metadata operation. -readMetaStream :: Settings - -> Text - -> Maybe Credentials - -> Operation StreamMetadataResult -readMetaStream setts s cred = construct $ do - traversing go <~ readEvent setts (metaStream s) (-1) False cred - where - go tmp = - onReadResult tmp $ \n e_num evt -> do - let bytes = recordedEventData $ resolvedEventOriginal evt - case decode $ fromStrict bytes of - Just pv -> pure $ StreamMetadataResult n e_num pv - Nothing -> Failed invalidFormat +readMetaStream + :: Settings + -> Exec + -> Text + -> Maybe Credentials + -> IO (Async StreamMetadataResult) +readMetaStream setts exec s cred + = async $ + do as <- readEvent setts exec (metaStream s) (-1) False cred + tmp <- wait as + onReadResult tmp $ \n evtNum evt -> + do let bytes = recordedEventData $ resolvedEventOriginal evt + case decode $ fromStrict bytes of + Just pv -> pure $ StreamMetadataResult n evtNum pv + Nothing -> throw invalidFormat -------------------------------------------------------------------------------- -- | Set stream metadata operation. -setMetaStream :: Settings - -> Text - -> ExpectedVersion - -> Maybe Credentials - -> StreamMetadata - -> Operation WriteResult -setMetaStream setts s v cred meta = - let stream = metaStream s +setMetaStream + :: Settings + -> Exec + -> Text + -> ExpectedVersion + -> Maybe Credentials + -> StreamMetadata + -> IO (Async WriteResult) +setMetaStream setts exec s v cred meta + = let stream = metaStream s json = streamMetadataJSON meta evt = createEvent StreamMetadataType Nothing (withJson json) in - writeEvents setts stream v cred [evt] + writeEvents setts exec stream v cred [evt] -------------------------------------------------------------------------------- invalidFormat :: OperationError @@ -79,14 +84,15 @@ -------------------------------------------------------------------------------- onReadResult :: ReadResult EventNumber ReadEvent - -> (Text -> Int64 -> ResolvedEvent -> Execution a) - -> Execution a + -> (Text -> Int64 -> ResolvedEvent -> IO a) + -> IO a onReadResult (ReadSuccess r) k = case r of ReadEvent s n e -> k s n e - _ -> Failed streamNotFound -onReadResult ReadNoStream _ = Failed streamNotFound -onReadResult (ReadStreamDeleted s) _ = Failed $ StreamDeleted s -onReadResult ReadNotModified _ = Failed $ ServerError Nothing -onReadResult (ReadError e) _ = Failed $ ServerError e -onReadResult (ReadAccessDenied s) _ = Failed $ AccessDenied s + _ -> throw streamNotFound + +onReadResult ReadNoStream _ = throw streamNotFound +onReadResult (ReadStreamDeleted s) _ = throw $ StreamDeleted s +onReadResult ReadNotModified _ = throw $ ServerError Nothing +onReadResult (ReadError e) _ = throw $ ServerError e +onReadResult (ReadAccessDenied s) _ = throw $ AccessDenied s diff --git a/Database/EventStore/Internal/Operation/Transaction.hs b/Database/EventStore/Internal/Operation/Transaction.hs --- a/Database/EventStore/Internal/Operation/Transaction.hs +++ b/Database/EventStore/Internal/Operation/Transaction.hs @@ -29,6 +29,10 @@ -------------------------------------------------------------------------------- import Database.EventStore.Internal.Command +import Database.EventStore.Internal.Control (publishWith) +import Database.EventStore.Internal.Communication (Transmit(..)) +import Database.EventStore.Internal.Exec (Exec) +import Database.EventStore.Internal.Operation (OpResult(..)) import Database.EventStore.Internal.Operation import Database.EventStore.Internal.Operation.Transaction.Message import Database.EventStore.Internal.Operation.Write.Common @@ -39,75 +43,109 @@ -------------------------------------------------------------------------------- -- | Start transaction operation. -transactionStart :: Settings - -> Text - -> ExpectedVersion - -> Maybe Credentials - -> Operation Int64 -transactionStart Settings{..} stream exp_v cred = construct $ do - let msg = newStart stream (expVersionInt64 exp_v) s_requireMaster - resp <- send transactionStartCmd transactionStartCompletedCmd cred msg - let tid = getField $ _transId resp - r = getField $ _result resp - case r of - OP_PREPARE_TIMEOUT -> retry - OP_FORWARD_TIMEOUT -> retry - OP_COMMIT_TIMEOUT -> retry - OP_WRONG_EXPECTED_VERSION -> wrongVersion stream exp_v - OP_STREAM_DELETED -> streamDeleted $ StreamName stream - OP_INVALID_TRANSACTION -> invalidTransaction - OP_ACCESS_DENIED -> accessDenied $ StreamName stream - OP_SUCCESS -> yield tid +transactionStart + :: Settings + -> Exec + -> Text + -> ExpectedVersion + -> Maybe Credentials + -> IO (Async Int64) +transactionStart Settings{..} exec stream exp_v cred + = do m <- mailboxNew + async $ + do let req = newStart stream (expVersionInt64 exp_v) s_requireMaster + pkg <- createPkg transactionStartCmd cred req + keepLooping $ + do publishWith exec (Transmit m OneTime pkg) + outcome <- mailboxReadDecoded m + case outcome of + Left e + -> throw e + Right resp + -> let tid = getField $ _transId resp + r = getField $ _result resp in + case r of + OP_PREPARE_TIMEOUT -> pure Loop + OP_FORWARD_TIMEOUT -> pure Loop + OP_COMMIT_TIMEOUT -> pure Loop + OP_WRONG_EXPECTED_VERSION -> throw $ WrongExpectedVersion stream exp_v + OP_STREAM_DELETED -> throw $ StreamDeleted $ StreamName stream + OP_INVALID_TRANSACTION -> throw InvalidTransaction + OP_ACCESS_DENIED -> throw $ AccessDenied $ StreamName stream + OP_SUCCESS -> pure $ Break tid + -------------------------------------------------------------------------------- -- | Transactional write operation. -transactionWrite :: Settings - -> Text - -> ExpectedVersion - -> Int64 - -> [Event] - -> Maybe Credentials - -> Operation () -transactionWrite Settings{..} stream exp_v trans_id evts cred = construct $ do - nevts <- traverse eventToNewEvent evts - let msg = newWrite trans_id nevts s_requireMaster - resp <- send transactionWriteCmd transactionWriteCompletedCmd cred msg - let r = getField $ _wwResult resp - case r of - OP_PREPARE_TIMEOUT -> retry - OP_FORWARD_TIMEOUT -> retry - OP_COMMIT_TIMEOUT -> retry - OP_WRONG_EXPECTED_VERSION -> wrongVersion stream exp_v - OP_STREAM_DELETED -> streamDeleted $ StreamName stream - OP_INVALID_TRANSACTION -> invalidTransaction - OP_ACCESS_DENIED -> accessDenied $ StreamName stream - OP_SUCCESS -> yield () +transactionWrite + :: Settings + -> Exec + -> Text + -> ExpectedVersion + -> Int64 + -> [Event] + -> Maybe Credentials + -> IO (Async ()) +transactionWrite Settings{..} exec stream exp_v trans_id evts cred + = do m <- mailboxNew + async $ + do nevts <- traverse eventToNewEventIO evts + let req = newWrite trans_id nevts s_requireMaster + pkg <- createPkg transactionWriteCmd cred req + keepLooping $ + do publishWith exec (Transmit m OneTime pkg) + outcome <- mailboxReadDecoded m + case outcome of + Left e + -> throw e + Right resp + -> let r = getField $ _wwResult resp in + case r of + OP_PREPARE_TIMEOUT -> pure Loop + OP_FORWARD_TIMEOUT -> pure Loop + OP_COMMIT_TIMEOUT -> pure Loop + OP_WRONG_EXPECTED_VERSION -> throw $ WrongExpectedVersion stream exp_v + OP_STREAM_DELETED -> throw $ StreamDeleted $ StreamName stream + OP_INVALID_TRANSACTION -> throw InvalidTransaction + OP_ACCESS_DENIED -> throw $ AccessDenied $ StreamName stream + OP_SUCCESS -> pure $ Break () -------------------------------------------------------------------------------- -- | Transactional commit operation. -transactionCommit :: Settings - -> Text - -> ExpectedVersion - -> Int64 - -> Maybe Credentials - -> Operation WriteResult -transactionCommit Settings{..} stream exp_v trans_id cred = construct $ do - let msg = newCommit trans_id s_requireMaster - resp <- send transactionCommitCmd transactionCommitCompletedCmd cred msg - let r = getField $ _ccResult resp - com_pos = getField $ _commitPosition resp - pre_pos = getField $ _preparePosition resp - lst_num = getField $ _lastNumber resp - p_int = fromMaybe (-1) pre_pos - c_int = fromMaybe (-1) com_pos - pos = Position c_int p_int - res = WriteResult lst_num pos - case r of - OP_PREPARE_TIMEOUT -> retry - OP_FORWARD_TIMEOUT -> retry - OP_COMMIT_TIMEOUT -> retry - OP_WRONG_EXPECTED_VERSION -> wrongVersion stream exp_v - OP_STREAM_DELETED -> streamDeleted $ StreamName stream - OP_INVALID_TRANSACTION -> invalidTransaction - OP_ACCESS_DENIED -> accessDenied $ StreamName stream - OP_SUCCESS -> yield res +transactionCommit + :: Settings + -> Exec + -> Text + -> ExpectedVersion + -> Int64 + -> Maybe Credentials + -> IO (Async WriteResult) +transactionCommit Settings{..} exec stream exp_v trans_id cred + = do m <- mailboxNew + async $ + do let req = newCommit trans_id s_requireMaster + pkg <- createPkg transactionCommitCmd cred req + keepLooping $ + do publishWith exec (Transmit m OneTime pkg) + outcome <- mailboxReadDecoded m + case outcome of + Left e + -> throw e + Right resp + -> let r = getField $ _ccResult resp + com_pos = getField $ _commitPosition resp + pre_pos = getField $ _preparePosition resp + lst_num = getField $ _lastNumber resp + p_int = fromMaybe (-1) pre_pos + c_int = fromMaybe (-1) com_pos + pos = Position c_int p_int + res = WriteResult lst_num pos in + case r of + OP_PREPARE_TIMEOUT -> pure Loop + OP_FORWARD_TIMEOUT -> pure Loop + OP_COMMIT_TIMEOUT -> pure Loop + OP_WRONG_EXPECTED_VERSION -> throw $ WrongExpectedVersion stream exp_v + OP_STREAM_DELETED -> throw $ StreamDeleted $ StreamName stream + OP_INVALID_TRANSACTION -> throw InvalidTransaction + OP_ACCESS_DENIED -> throw $ AccessDenied $ StreamName stream + OP_SUCCESS -> pure $ Break res diff --git a/Database/EventStore/Internal/Operation/Volatile.hs b/Database/EventStore/Internal/Operation/Volatile.hs --- a/Database/EventStore/Internal/Operation/Volatile.hs +++ b/Database/EventStore/Internal/Operation/Volatile.hs @@ -17,6 +17,9 @@ -------------------------------------------------------------------------------- import Database.EventStore.Internal.Command +import Database.EventStore.Internal.Communication (Transmit(..)) +import Database.EventStore.Internal.Control (publishWith) +import Database.EventStore.Internal.Exec (Exec) import Database.EventStore.Internal.Operation import Database.EventStore.Internal.Prelude import Database.EventStore.Internal.Settings @@ -26,57 +29,49 @@ import Database.EventStore.Internal.Types -------------------------------------------------------------------------------- -volatile :: StreamId t -> Bool -> Maybe Credentials -> Operation SubAction -volatile streamId tos cred = construct (issueRequest stream tos cred) - where - stream = streamIdRaw streamId - --------------------------------------------------------------------------------- -issueRequest :: Text -> Bool -> Maybe Credentials -> Code SubAction () -issueRequest stream tos cred = do - let req = subscribeToStream stream tos - request subscribeToStreamCmd cred req - [ Expect subscriptionDroppedCmd $ \_ d -> - handleDropped d - , Expect subscriptionConfirmationCmd $ \sid c -> do - let lcp = getField $ subscribeLastCommitPos c - len = getField $ subscribeLastEventNumber c - details = - SubDetails - { subId = sid - , subCommitPos = lcp - , subLastEventNum = len - , subSubId = Nothing - } - yield (Confirmed details) - live sid - ] - --------------------------------------------------------------------------------- -eventAppeared :: StreamEventAppeared -> Code SubAction () -eventAppeared e = do - let evt = newResolvedEventFromBuf $ getField $ streamResolvedEvent e - yield (Submit evt) +volatile + :: Exec + -> StreamId t + -> Bool + -> Maybe Credentials + -> IO (UUID, Chan SubAction) +volatile exec streamId tos cred + = do m <- mailboxNew + subM <- newChan + let req = subscribeToStream stream tos + pkg <- createPkg subscribeToStreamCmd cred req + let theSubId = packageCorrelation pkg + publishWith exec (Transmit m (KeepAlive subscriptionDroppedCmd) pkg) + _ <- async $ keepLooping $ + do outcome <- mailboxRead m + case outcome of + Left _ + -> Break () <$ writeChan subM (Dropped SubAborted) + Right respPkg + | packageCmd respPkg == subscriptionDroppedCmd + -> let Right resp = decodePkg respPkg + reason = fromMaybe D_Unsubscribed (getField $ dropReason resp) + subReason = toSubDropReason reason in + Break () <$ writeChan subM (Dropped subReason) + | packageCmd respPkg == subscriptionConfirmationCmd + -> let Right resp = decodePkg respPkg + lcp = getField $ subscribeLastCommitPos resp + len = getField $ subscribeLastEventNumber resp + details = + SubDetails + { subId = theSubId + , subCommitPos = lcp + , subLastEventNum = len + , subSubId = Nothing + } in + Loop <$ writeChan subM (Confirmed details) + | packageCmd respPkg == streamEventAppearedCmd + -> let Right resp = decodePkg respPkg + evt = newResolvedEventFromBuf $ getField $ streamResolvedEvent resp in + Loop <$ writeChan subM (Submit evt) + | otherwise + -> pure Loop --------------------------------------------------------------------------------- -live :: UUID -> Code SubAction () -live subscriptionId = loop + pure (theSubId, subM) where - loop = - waitForOr subscriptionId connectionReset - [ Expect subscriptionDroppedCmd $ \_ d -> - handleDropped d - , Expect streamEventAppearedCmd $ \_ e -> do - eventAppeared e - loop - ] - --------------------------------------------------------------------------------- -connectionReset :: Code SubAction () -connectionReset = yield ConnectionReset - --------------------------------------------------------------------------------- -handleDropped :: SubscriptionDropped -> Code SubAction () -handleDropped d = do - let reason = fromMaybe D_Unsubscribed (getField $ dropReason d) - yield (Dropped $ toSubDropReason reason) + stream = streamIdRaw streamId 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 @@ -15,7 +15,7 @@ import Data.Int -------------------------------------------------------------------------------- -import Database.EventStore.Internal.Operation +import Database.EventStore.Internal.Control (freshUUID) import Database.EventStore.Internal.Prelude import Database.EventStore.Internal.Types @@ -32,9 +32,9 @@ -------------------------------------------------------------------------------- -- | Constructs a 'NewEvent' from an 'Event'. -eventToNewEvent :: Event -> Code o NewEvent -eventToNewEvent evt = do - uuid <- maybe freshId return evt_id +eventToNewEventIO :: Event -> IO NewEvent +eventToNewEventIO evt = do + uuid <- maybe freshUUID pure evt_id return $ newEvent evt_type uuid evt_data_type diff --git a/Database/EventStore/Internal/Operation/WriteEvents.hs b/Database/EventStore/Internal/Operation/WriteEvents.hs --- a/Database/EventStore/Internal/Operation/WriteEvents.hs +++ b/Database/EventStore/Internal/Operation/WriteEvents.hs @@ -21,6 +21,10 @@ -------------------------------------------------------------------------------- import Database.EventStore.Internal.Command +import Database.EventStore.Internal.Communication (Transmit(..)) +import Database.EventStore.Internal.Control (publishWith) +import Database.EventStore.Internal.Exec +import Database.EventStore.Internal.Operation (OpResult(..)) import Database.EventStore.Internal.Operation import Database.EventStore.Internal.Operation.Write.Common import Database.EventStore.Internal.Operation.WriteEvents.Message @@ -29,32 +33,45 @@ import Database.EventStore.Internal.Stream import Database.EventStore.Internal.Types --------------------------------------------------------------------------------- --- | Write events operation. -writeEvents :: Settings - -> Text - -> ExpectedVersion - -> Maybe Credentials - -> [Event] - -> Operation WriteResult -writeEvents Settings{..} s v cred evts = construct $ do - nevts <- traverse eventToNewEvent evts - let msg = newRequest s (expVersionInt64 v) nevts s_requireMaster - resp <- send writeEventsCmd writeEventsCompletedCmd cred msg - let r = getField $ _result resp - com_pos = getField $ _commitPosition resp - prep_pos = getField $ _preparePosition resp - lst_num = getField $ _lastNumber resp - com_pos_int = fromMaybe (-1) com_pos - prep_pos_int = fromMaybe (-1) prep_pos - pos = Position com_pos_int prep_pos_int - res = WriteResult lst_num pos - case r of - OP_SUCCESS -> yield res - OP_PREPARE_TIMEOUT -> retry - OP_FORWARD_TIMEOUT -> retry - OP_COMMIT_TIMEOUT -> retry - OP_WRONG_EXPECTED_VERSION -> wrongVersion s v - OP_STREAM_DELETED -> streamDeleted $ StreamName s - OP_INVALID_TRANSACTION -> invalidTransaction - OP_ACCESS_DENIED -> accessDenied (StreamName s) +------------------------------------------------------------------------------- +writeEvents + :: Settings + -> Exec + -> Text + -> ExpectedVersion + -> Maybe Credentials + -> [Event] + -> IO (Async WriteResult) +writeEvents setts exec stream version creds evts + = do m <- mailboxNew + async $ + do nevts <- traverse eventToNewEventIO evts + let req = newRequest stream (expVersionInt64 version) nevts (s_requireMaster setts) + + pkg <- createPkg writeEventsCmd creds req + + keepLooping $ do + publishWith exec (Transmit m OneTime pkg) + outcome <- mailboxReadDecoded m + case outcome of + Left e + -> throw e + Right resp + -> let r = getField $ _result resp + com_pos = getField $ _commitPosition resp + prep_pos = getField $ _preparePosition resp + lst_num = getField $ _lastNumber resp + com_pos_int = fromMaybe (-1) com_pos + prep_pos_int = fromMaybe (-1) prep_pos + pos = Position com_pos_int prep_pos_int + res = WriteResult lst_num pos in + case r of + OP_SUCCESS -> pure $ Break res + OP_PREPARE_TIMEOUT -> pure Loop + OP_FORWARD_TIMEOUT -> pure Loop + OP_COMMIT_TIMEOUT -> pure Loop + OP_WRONG_EXPECTED_VERSION -> throw $ WrongExpectedVersion stream version + OP_STREAM_DELETED -> throw $ StreamDeleted $ StreamName stream + OP_INVALID_TRANSACTION -> throw InvalidTransaction + OP_ACCESS_DENIED -> throw $ AccessDenied (StreamName stream) + diff --git a/Database/EventStore/Internal/OperationManager.hs b/Database/EventStore/Internal/OperationManager.hs deleted file mode 100644 --- a/Database/EventStore/Internal/OperationManager.hs +++ /dev/null @@ -1,60 +0,0 @@ -{-# LANGUAGE OverloadedStrings #-} -{-# LANGUAGE RecordWildCards #-} --------------------------------------------------------------------------------- --- | --- Module : Database.EventStore.Internal.OperationManager --- Copyright : (C) 2017 Yorick Laupa --- License : (see the file LICENSE) --- --- Maintainer : Yorick Laupa --- Stability : provisional --- Portability : non-portable --- --------------------------------------------------------------------------------- -module Database.EventStore.Internal.OperationManager - ( Manager - , Decision(..) - , new - , submit - , handle - , cleanup - , check - ) where - --------------------------------------------------------------------------------- -import Database.EventStore.Internal.Callback -import Database.EventStore.Internal.Connection -import Database.EventStore.Internal.Control -import Database.EventStore.Internal.Logger -import Database.EventStore.Internal.Manager.Operation.Registry -import Database.EventStore.Internal.Operation -import Database.EventStore.Internal.Prelude -import Database.EventStore.Internal.Types - --------------------------------------------------------------------------------- -data Manager = Manager { _reg :: Registry } - --------------------------------------------------------------------------------- -new :: ConnectionRef -> IO Manager -new = fmap Manager . newRegistry - --------------------------------------------------------------------------------- -submit :: Manager -> Operation a -> Callback a -> EventStore () -submit Manager{..} op cb = register _reg op cb - --------------------------------------------------------------------------------- -handle :: Manager -> Package -> EventStore (Maybe Decision) -handle Manager{..} pkg = handlePackage _reg pkg - --------------------------------------------------------------------------------- -cleanup :: Manager -> EventStore () -cleanup Manager{..} = do - $(logInfo) "Cleaning up pending requests..." - abortPendingRequests _reg - $(logInfo) "Cleanup done successfully." - --------------------------------------------------------------------------------- -check :: Manager -> EventStore () -check Manager{..} = do - checkAndRetry _reg - startAwaitings _reg diff --git a/Database/EventStore/Internal/Settings.hs b/Database/EventStore/Internal/Settings.hs --- a/Database/EventStore/Internal/Settings.hs +++ b/Database/EventStore/Internal/Settings.hs @@ -13,7 +13,6 @@ -------------------------------------------------------------------------------- import Network.Connection (TLSSettings) -import System.Metrics (Store) -------------------------------------------------------------------------------- import Database.EventStore.Internal.Logger @@ -99,8 +98,10 @@ -- ^ Delay in which an operation will be retried if no response arrived. , s_operationRetry :: Retry -- ^ Retry strategy when an operation timeout. - , s_monitoring :: Maybe Store - -- ^ EKG metric store. + , s_monitoring :: MonitoringBackend + -- ^ Monitoring backend abstraction. You could implement one targetting + -- `ekg-core` for example. We will expose an `ekg-core` implementation + -- as soon as `ekg-core` supports GHC 8.8.*. , s_defaultConnectionName :: Maybe Text -- ^ Default connection name. , s_defaultUserCredentials :: Maybe Credentials @@ -122,7 +123,7 @@ -- * 's_loggerDetailed' = 'False' -- * 's_operationTimeout' = 10 seconds -- * 's_operationRetry' = 'atMost' 3 --- * 's_monitoring' = 'Nothing' +-- * 's_monitoring' = 'noopMonitoringBackend' -- * 's_defaultConnectionName' = 'Nothing' -- * 's_defaultUserCredentials' = 'Nothing' defaultSettings :: Settings @@ -138,7 +139,7 @@ , s_loggerDetailed = False , s_operationTimeout = 10 -- secs , s_operationRetry = atMost 3 - , s_monitoring = Nothing + , s_monitoring = noopMonitoringBackend , s_defaultConnectionName = Nothing , s_defaultUserCredentials = Nothing } @@ -152,3 +153,40 @@ -- | Millisecond timespan msDiffTime :: Float -> NominalDiffTime msDiffTime n = fromRational $ toRational (n / 1000) + +-------------------------------------------------------------------------------- +-- | Monitoring backend abstraction. Gathers all the metrics currently tracked +-- by the client. Used only by the TCP interface. Be careful as +-- 'MonitoringBackend' is used in a very tight loop. Each +-- function must not throw any exception or the client will end in a broken +-- state. +data MonitoringBackend = + MonitoringBackend + { monitoringBackendIncrPkgCount :: IO () + -- ^ Called every time a TCP package is sent. We mean high-level TCP + -- package, used in EventStore TCP protocol. + , monitoringBackendIncrConnectionDrop :: IO () + -- ^ Called every time the client has lost the connection. + , monitoringBackendAddDataTransmitted :: Int -> IO () + -- ^ When the client sends a TCP package, it calls that function by + -- passing the size of the payload. The goal is to have a distrubtion + -- of the amount of data exchanged with the server. + , monitoringBackendIncrForceReconnect :: IO () + -- ^ Called every time the client is asked by a node to connect to + -- another node. It happens only in cluster connection setting. + , monitoringBackendIncrHeartbeatTimeouts :: IO () + -- ^ Called every time the client detects a heartbeat timeout from the + -- server. + } + +-------------------------------------------------------------------------------- +-- | A 'MonitoringBackend' that does nothing. +noopMonitoringBackend :: MonitoringBackend +noopMonitoringBackend = + MonitoringBackend + { monitoringBackendIncrPkgCount = pure () + , monitoringBackendIncrConnectionDrop = pure () + , monitoringBackendAddDataTransmitted = const (pure ()) + , monitoringBackendIncrForceReconnect = pure () + , monitoringBackendIncrHeartbeatTimeouts = pure () + } diff --git a/Database/EventStore/Internal/Subscription/Api.hs b/Database/EventStore/Internal/Subscription/Api.hs --- a/Database/EventStore/Internal/Subscription/Api.hs +++ b/Database/EventStore/Internal/Subscription/Api.hs @@ -16,38 +16,23 @@ module Database.EventStore.Internal.Subscription.Api where -------------------------------------------------------------------------------- -import Database.EventStore.Internal.Callback -import Database.EventStore.Internal.Communication -import Database.EventStore.Internal.Control +import Streaming +import qualified Streaming.Prelude as Streaming + +-------------------------------------------------------------------------------- import Database.EventStore.Internal.Types import Database.EventStore.Internal.Prelude import Database.EventStore.Internal.Stream -import Database.EventStore.Internal.Subscription.Packages import Database.EventStore.Internal.Subscription.Types -------------------------------------------------------------------------------- -submit :: Callback SubAction -> ResolvedEvent -> IO () -submit s xs = fulfill s (Submit xs) - --------------------------------------------------------------------------------- -dropped :: Callback SubAction -> SubDropReason -> IO () -dropped s r = fulfill s (Dropped r) - --------------------------------------------------------------------------------- -confirmed :: Callback SubAction -> SubDetails -> IO () -confirmed s d = fulfill s (Confirmed d) - --------------------------------------------------------------------------------- -- | Common operations supported by a subscription. class Subscription s where - -- | Asks for the next incoming event like 'nextEventMaybe' while still being - -- in the the 'STM'. - nextEventMaybeSTM :: s -> STM (Maybe ResolvedEvent) - - -- | Returns the runtime details of a subscription. - getSubscriptionDetailsSTM :: s -> STM SubDetails + -- | Asks for the next subcription event. If that function is called after + -- a SubDropped event, expect it to hang indefinitely. + nextSubEvent :: s -> IO SubAction - -- | Asynchronously unsubscribe from the the stream. + -- | Asynchronously unsubscribe from a subscription. unsubscribe :: s -> IO () -------------------------------------------------------------------------------- @@ -56,62 +41,22 @@ subscriptionStream :: s -> StreamId t -------------------------------------------------------------------------------- --- | Awaits for the next event. -nextEvent :: Subscription s => s -> IO ResolvedEvent -nextEvent s = atomically $ do - outcome <- nextEventMaybeSTM s - case outcome of - Just e -> return e - Nothing -> retrySTM - --------------------------------------------------------------------------------- --- | Non blocking version of 'nextEvent'. -nextEventMaybe :: Subscription s => s -> IO (Maybe ResolvedEvent) -nextEventMaybe = atomically . nextEventMaybeSTM - --------------------------------------------------------------------------------- --- | Waits until the `Subscription` has been confirmed. -waitConfirmation :: Subscription s => s -> IO () -waitConfirmation s = atomically $ do - _ <- getSubscriptionDetailsSTM s - return () - --------------------------------------------------------------------------------- --- | Like 'unsubscribeConfirmed' but lives in 'STM' monad. -unsubscribeConfirmedSTM :: Subscription s => s -> STM Bool -unsubscribeConfirmedSTM s = do - let action = do - _ <- getSubscriptionDetailsSTM s - return False - catchSTM action $ \(_ :: SomeException) -> return True - --------------------------------------------------------------------------------- --- | Non blocking version of `waitUnsubscribeConfirmed`. -unsubscribeConfirmed :: Subscription s => s -> IO Bool -unsubscribeConfirmed = atomically . unsubscribeConfirmedSTM - --------------------------------------------------------------------------------- --- | Wait until unsubscription has been confirmed by the server. -waitUnsubscribeConfirmed :: Subscription s => s -> IO () -waitUnsubscribeConfirmed s = atomically $ - unlessM (unsubscribeConfirmedSTM s) retrySTM +-- | Streams a subscription events. The stream will end when hitting `Dropped` +-- event but will still emit it. +streamSubEvents :: Subscription s => s -> Stream (Of SubAction) IO () +streamSubEvents s + = do rest <- Streaming.span predicate $ Streaming.repeatM (nextSubEvent s) + outcome <- lift $ Streaming.uncons rest + for_ outcome $ \(dropped, _) -> Streaming.yield dropped + where + predicate (Dropped _) = False + predicate _ = True -------------------------------------------------------------------------------- -subUnsubscribe :: (Pub pub, Subscription s) => pub -> s -> IO () -subUnsubscribe pub s = do - outcome <- atomically $ do - unsubscribed <- unsubscribeConfirmedSTM s - if unsubscribed - then return Nothing - else Just <$> getSubscriptionDetailsSTM s - - for_ outcome $ \details -> do - let pkg = createUnsubscribePackage (subId details) - publishWith pub (SendPackage pkg) +-- | Like `streamSubEvent` but will only emit `ResolvedEvent`. +streamSubResolvedEvents :: Subscription s => s -> Stream (Of ResolvedEvent) IO () +streamSubResolvedEvents = Streaming.mapMaybe go . streamSubEvents + where + go (Submit e) = Just e + go _ = Nothing --------------------------------------------------------------------------------- --- | Gets the ID of the subscription. -getSubscriptionId :: Subscription s => s -> IO SubscriptionId -getSubscriptionId s = atomically $ do - details <- getSubscriptionDetailsSTM s - return (SubscriptionId $ subId details) diff --git a/Database/EventStore/Internal/Subscription/Catchup.hs b/Database/EventStore/Internal/Subscription/Catchup.hs --- a/Database/EventStore/Internal/Subscription/Catchup.hs +++ b/Database/EventStore/Internal/Subscription/Catchup.hs @@ -18,30 +18,21 @@ module Database.EventStore.Internal.Subscription.Catchup where -------------------------------------------------------------------------------- -import Control.Monad.Fix import Safe (fromJustNote) -------------------------------------------------------------------------------- -import Database.EventStore.Internal.Callback import Database.EventStore.Internal.Communication import Database.EventStore.Internal.Control import Database.EventStore.Internal.Exec -import Database.EventStore.Internal.Operation import Database.EventStore.Internal.Operation.Catchup -import Database.EventStore.Internal.Operation.Volatile import Database.EventStore.Internal.Prelude import Database.EventStore.Internal.Stream import Database.EventStore.Internal.Subscription.Api import Database.EventStore.Internal.Subscription.Types +import Database.EventStore.Internal.Subscription.Packages import Database.EventStore.Internal.Types -------------------------------------------------------------------------------- -data Phase - = CatchingUp - | Running SubDetails - | Closed (Either SomeException SubDropReason) - --------------------------------------------------------------------------------- receivedAlready :: StreamId t -> t -> ResolvedEvent -> Bool receivedAlready StreamName{} old e = EventNumber (resolvedEventOriginalEventNumber e) < old @@ -72,108 +63,48 @@ -- then any events subsequently written until such time as the subscription is -- dropped or closed. data CatchupSubscription t = - CatchupSubscription { _catchupExec :: Exec - , _catchupStream :: StreamId t - , _catchupPhase :: TVar Phase - , _catchupTrack :: TVar t - , _catchupNext :: STM (Maybe ResolvedEvent) - } + CatchupSubscription + { _catchupExec :: Exec + , _catchupStream :: StreamId t + , _catchupSub :: TVar (Maybe UUID) + , _catchupChan :: Chan SubAction + } -------------------------------------------------------------------------------- -instance Subscription (CatchupSubscription t) where - nextEventMaybeSTM = _catchupNext +instance Subscription (CatchupSubscription s) where + nextSubEvent s = readChan (_catchupChan s) - getSubscriptionDetailsSTM s = do - p <- readTVar (_catchupPhase s) - case p of - Running details -> return details - Closed r -> throwClosed r - _ -> retrySTM + unsubscribe s + = do subId <- atomically $ + do idMay <- readTVar (_catchupSub s) + case idMay of + Nothing -> retrySTM + Just sid -> pure sid - unsubscribe s = subUnsubscribe (_catchupExec s) s + let pkg = createUnsubscribePackage subId + publishWith (_catchupExec s) (SendPackage pkg) -------------------------------------------------------------------------------- instance SubscriptionStream (CatchupSubscription t) t where subscriptionStream = _catchupStream -------------------------------------------------------------------------------- -newCatchupSubscription :: Exec - -> Bool - -> Maybe Int32 - -> Maybe Credentials - -> StreamId t - -> t - -> IO (CatchupSubscription t) -newCatchupSubscription exec tos batch cred streamId seed = do - phaseVar <- newTVarIO CatchingUp - queue <- newTQueueIO - track <- newTVarIO seed - - let sub = CatchupSubscription exec streamId phaseVar track $ do - p <- readTVar phaseVar - isEmpty <- isEmptyTQueue queue - if isEmpty - then - case p of - Closed r -> throwClosed r - _ -> return Nothing - else Just <$> readTQueue queue - - callback cb (Left e) = - case fromException e of - Just opE -> - case opE of - StreamNotFound{} -> do - let op = volatile streamId tos cred - publishWith exec (SubmitOperation cb op) - _ -> atomically $ writeTVar phaseVar (Closed $ Left e) - _ -> atomically $ writeTVar phaseVar (Closed $ Left e) - callback _ (Right action) = - case action of - Confirmed details -> atomically $ writeTVar phaseVar (Running details) - Dropped r -> - atomically $ writeTVar phaseVar (Closed $ Right r) - Submit e -> atomically $ do - tracker <- readTVar track - unless (receivedAlready streamId tracker e) $ do - writeTVar track (nextTarget streamId e) - writeTQueue queue e - ConnectionReset -> do - chk <- readTVarIO track - let newOp = catchup (execSettings exec) streamId chk tos batch cred - - newCb <- mfix $ \self -> newCallback (callback self) - publishWith exec (SubmitOperation newCb newOp) - - cb <- mfix $ \self -> newCallback (callback self) - let op = catchup (execSettings exec) streamId seed tos batch cred - publishWith exec (SubmitOperation cb op) - return sub - --------------------------------------------------------------------------------- -throwClosed :: Either SomeException SubDropReason -> STM a -throwClosed (Left e) = throwSTM e -throwClosed (Right r) = throwSTM (SubscriptionClosed $ Just r) - --------------------------------------------------------------------------------- --- | Non blocking version of `waitTillCatchup`. -hasCaughtUp :: CatchupSubscription t -> IO Bool -hasCaughtUp sub = atomically $ hasCaughtUpSTM sub - --------------------------------------------------------------------------------- --- | Waits until 'CatchupSubscription' subscription catch-up its stream. -waitTillCatchup :: CatchupSubscription t -> IO () -waitTillCatchup sub = atomically $ unlessM (hasCaughtUpSTM sub) retrySTM +newCatchupSubscription + :: Exec + -> Bool + -> Maybe Int32 + -> Maybe Credentials + -> StreamId t + -> t + -> IO (CatchupSubscription t) +newCatchupSubscription exec tos batch cred streamId seed + = do (var, chan) <- catchup (execSettings exec) exec streamId seed tos batch cred + let sub = + CatchupSubscription + { _catchupExec = exec + , _catchupStream = streamId + , _catchupSub = var + , _catchupChan = chan + } --------------------------------------------------------------------------------- --- | Like 'hasCaughtUp' but lives in 'STM' monad. -hasCaughtUpSTM :: CatchupSubscription t -> STM Bool -hasCaughtUpSTM CatchupSubscription{..} = do - p <- readTVar _catchupPhase - case p of - CatchingUp -> return False - Running{} -> return True - Closed tpe -> - case tpe of - Left e -> throwSTM e - Right r -> throwSTM (SubscriptionClosed $ Just r) + pure sub diff --git a/Database/EventStore/Internal/Subscription/Persistent.hs b/Database/EventStore/Internal/Subscription/Persistent.hs --- a/Database/EventStore/Internal/Subscription/Persistent.hs +++ b/Database/EventStore/Internal/Subscription/Persistent.hs @@ -18,7 +18,6 @@ import Data.UUID -------------------------------------------------------------------------------- -import Database.EventStore.Internal.Callback import Database.EventStore.Internal.Communication import Database.EventStore.Internal.Control import Database.EventStore.Internal.Exec @@ -32,106 +31,78 @@ import Database.EventStore.Internal.Types -------------------------------------------------------------------------------- -data Phase - = Pending - | Running SubDetails - | Closed (Either SomeException SubDropReason) - --------------------------------------------------------------------------------- -- | 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 PersistentSubscription = - PersistentSubscription { _perExec :: Exec - , _perStream :: StreamName - , _perCred :: Maybe Credentials - , _perPhase :: TVar Phase - , _perNext :: STM (Maybe ResolvedEvent) - } + PersistentSubscription + { _perExec :: Exec + , _perSubId :: UUID + , _perStream :: StreamName + , _perCred :: Maybe Credentials + , _perSubKey :: TVar (Maybe Text) + , _perChan :: Chan SubAction + } -------------------------------------------------------------------------------- instance Subscription PersistentSubscription where - nextEventMaybeSTM = _perNext - - getSubscriptionDetailsSTM s = do - p <- readTVar (_perPhase s) - case p of - Pending -> retrySTM - Running details -> return details - Closed outcome -> - case outcome of - Right r -> throwSTM (SubscriptionClosed $ Just r) - Left e -> throwSTM e + nextSubEvent s = readChan (_perChan s) - unsubscribe s = subUnsubscribe (_perExec s) s + unsubscribe s = publishWith (_perExec s) (SendPackage pkg) + where + pkg = createUnsubscribePackage (_perSubId s) -------------------------------------------------------------------------------- instance SubscriptionStream PersistentSubscription EventNumber where subscriptionStream = _perStream -------------------------------------------------------------------------------- -newPersistentSubscription :: Exec - -> Text - -> StreamName - -> Int32 - -> Maybe Credentials - -> IO PersistentSubscription -newPersistentSubscription exec grp stream bufSize cred = do - phaseVar <- newTVarIO Pending - queue <- newTQueueIO - - let name = streamIdRaw stream - sub = PersistentSubscription exec stream cred phaseVar $ do - p <- readTVar phaseVar - isEmpty <- isEmptyTQueue queue - if isEmpty - then - case p of - Closed outcome -> - case outcome of - Right r -> throwSTM (SubscriptionClosed $ Just r) - Left e -> throwSTM e - _ -> return Nothing - else Just <$> readTQueue queue +newPersistentSubscription + :: Exec + -> Text + -> StreamName + -> Int32 + -> Maybe Credentials + -> IO PersistentSubscription +newPersistentSubscription exec grp (StreamName stream) bufSize cred + = do (subId, varSubKey, chan) <- persist exec grp stream bufSize cred + let sub = + PersistentSubscription + { _perExec = exec + , _perSubId = subId + , _perCred = cred + , _perSubKey = varSubKey + , _perChan = chan + , _perStream = StreamName stream + } - callback (Left e) = atomically $ - writeTVar phaseVar (Closed $ Left e) - callback (Right action) = - case action of - Confirmed details -> atomically $ - writeTVar phaseVar (Running details) - Dropped r -> atomically $ - writeTVar phaseVar (Closed $ Right r) - Submit e -> atomically $ do - readTVar phaseVar >>= \case - Running{} -> writeTQueue queue e - _ -> return () - ConnectionReset -> atomically $ - writeTVar phaseVar (Closed $ Right SubAborted) + pure sub - cb <- newCallback callback - publishWith exec (SubmitOperation cb (persist grp name bufSize cred)) - return sub +-------------------------------------------------------------------------------- +persistentGetSubKey + :: PersistentSubscription + -> IO Text +persistentGetSubKey sub + = atomically $ + do subKeyMay <- readTVar (_perSubKey sub) + case subKeyMay of + Just key + -> pure key + Nothing + -> retrySTM -------------------------------------------------------------------------------- -- | Acknowledges those event ids have been successfully processed. -notifyEventsProcessed :: PersistentSubscription -> [UUID] -> IO () -notifyEventsProcessed PersistentSubscription{..} evts = do - details <- atomically $ do - p <- readTVar _perPhase - case p of - Closed outcome -> - case outcome of - Right r -> throwSTM (SubscriptionClosed $ Just r) - Left e -> throwSTM e - Pending -> retrySTM - Running d -> return d - - let uuid = subId details - Just sid = subSubId details - pkg = createAckPackage _perCred uuid sid evts - publishWith _perExec (SendPackage pkg) +notifyEventsProcessed + :: PersistentSubscription + -> [UUID] + -> IO () +notifyEventsProcessed sub evts + = do subKey <- persistentGetSubKey sub + let uuid = _perSubId sub + pkg = createAckPackage (_perCred sub) uuid subKey evts + publishWith (_perExec sub) (SendPackage pkg) -------------------------------------------------------------------------------- -- | Acknowledges that 'ResolvedEvent' has been successfully processed. @@ -166,23 +137,14 @@ -------------------------------------------------------------------------------- -- | Acknowledges those event ids have failed to be processed successfully. -notifyEventsFailed :: PersistentSubscription - -> NakAction - -> Maybe Text - -> [UUID] - -> IO () -notifyEventsFailed PersistentSubscription{..} act res evts = do - details <- atomically $ do - p <- readTVar _perPhase - case p of - Closed outcome -> - case outcome of - Right r -> throwSTM (SubscriptionClosed $ Just r) - Left e -> throwSTM e - Pending -> retrySTM - Running d -> return d - - let uuid = subId details - Just sid = subSubId details - pkg = createNakPackage _perCred uuid sid act res evts - publishWith _perExec (SendPackage pkg) +notifyEventsFailed + :: PersistentSubscription + -> NakAction + -> Maybe Text -- Reason + -> [UUID] + -> IO () +notifyEventsFailed sub act res evts + = do subKey <- persistentGetSubKey sub + let uuid = _perSubId sub + pkg = createNakPackage (_perCred sub) uuid subKey act res evts + publishWith (_perExec sub) (SendPackage pkg) diff --git a/Database/EventStore/Internal/Subscription/Regular.hs b/Database/EventStore/Internal/Subscription/Regular.hs --- a/Database/EventStore/Internal/Subscription/Regular.hs +++ b/Database/EventStore/Internal/Subscription/Regular.hs @@ -14,7 +14,6 @@ module Database.EventStore.Internal.Subscription.Regular where -------------------------------------------------------------------------------- -import Database.EventStore.Internal.Callback import Database.EventStore.Internal.Communication import Database.EventStore.Internal.Control import Database.EventStore.Internal.Exec @@ -23,13 +22,7 @@ import Database.EventStore.Internal.Stream import Database.EventStore.Internal.Subscription.Api import Database.EventStore.Internal.Subscription.Types -import Database.EventStore.Internal.Types - --------------------------------------------------------------------------------- -data Phase - = Pending - | Running SubDetails - | Closed (Either SomeException SubDropReason) +import Database.EventStore.Internal.Subscription.Packages -------------------------------------------------------------------------------- -- | Also referred as volatile subscription. For example, if a stream has 100 @@ -37,70 +30,41 @@ -- event number 101 onwards until the time the subscription is closed or -- dropped. data RegularSubscription t = - RegularSubscription { _regExec :: Exec - , _regStream :: StreamId t - , _regPhase :: TVar Phase - , _regNext :: STM (Maybe ResolvedEvent) - } + RegularSubscription + { _regExec :: Exec + , _regSubId :: UUID + , _regStream :: StreamId t + , _regChan :: Chan SubAction + } -------------------------------------------------------------------------------- -instance Subscription (RegularSubscription t) where - nextEventMaybeSTM = _regNext - - getSubscriptionDetailsSTM s = do - p <- readTVar (_regPhase s) - case p of - Pending -> retrySTM - Running details -> return details - Closed outcome -> - case outcome of - Right r -> throwSTM (SubscriptionClosed $ Just r) - Left e -> throwSTM e +instance Subscription (RegularSubscription s) where + nextSubEvent s = readChan (_regChan s) - unsubscribe s = subUnsubscribe (_regExec s) s + unsubscribe s = publishWith (_regExec s) (SendPackage pkg) + where + pkg = createUnsubscribePackage (_regSubId s) -------------------------------------------------------------------------------- instance SubscriptionStream (RegularSubscription t) t where subscriptionStream = _regStream -------------------------------------------------------------------------------- -newRegularSubscription :: Exec - -> StreamId t - -> Bool - -> Maybe Credentials - -> IO (RegularSubscription t) -newRegularSubscription exec streamId tos cred = do - phaseVar <- newTVarIO Pending - queue <- newTQueueIO - - let sub = RegularSubscription exec streamId phaseVar $ do - p <- readTVar phaseVar - isEmpty <- isEmptyTQueue queue - if isEmpty - then - case p of - Closed outcome -> - case outcome of - Right r -> throwSTM (SubscriptionClosed $ Just r) - Left e -> throwSTM e - _ -> return Nothing - else Just <$> readTQueue queue +newRegularSubscription + :: Exec + -> StreamId t + -> Bool + -> Maybe Credentials + -> IO (RegularSubscription t) +newRegularSubscription exec streamId tos cred + = do (subId, chan) <- volatile exec streamId tos cred + let sub = + RegularSubscription + { _regExec = exec + , _regSubId = subId + , _regStream = streamId + , _regChan = chan + } - callback (Left e) = atomically $ - writeTVar phaseVar (Closed $ Left e) - callback (Right action) = - case action of - Confirmed details -> atomically $ - writeTVar phaseVar (Running details) - Dropped r -> atomically $ - writeTVar phaseVar (Closed $ Right r) - Submit e -> atomically $ do - readTVar phaseVar >>= \case - Running{} -> writeTQueue queue e - _ -> return () - ConnectionReset -> atomically $ - writeTVar phaseVar (Closed $ Right SubAborted) + pure sub - cb <- newCallback callback - publishWith exec (SubmitOperation cb (volatile streamId tos cred)) - return sub diff --git a/Database/EventStore/Internal/Subscription/Types.hs b/Database/EventStore/Internal/Subscription/Types.hs --- a/Database/EventStore/Internal/Subscription/Types.hs +++ b/Database/EventStore/Internal/Subscription/Types.hs @@ -135,4 +135,3 @@ = Submit ResolvedEvent | Dropped SubDropReason | Confirmed SubDetails - | ConnectionReset diff --git a/Database/EventStore/Internal/Test.hs b/Database/EventStore/Internal/Test.hs --- a/Database/EventStore/Internal/Test.hs +++ b/Database/EventStore/Internal/Test.hs @@ -12,8 +12,7 @@ -- Re-exports several modules to ease internal testing. -------------------------------------------------------------------------------- module Database.EventStore.Internal.Test - ( module Database.EventStore.Internal.Callback - , module Database.EventStore.Internal.Command + ( module Database.EventStore.Internal.Command , module Database.EventStore.Internal.Communication , module Database.EventStore.Internal.Connection , module Database.EventStore.Internal.Control @@ -27,7 +26,6 @@ , module Database.EventStore.Internal.Types ) where -import Database.EventStore.Internal.Callback import Database.EventStore.Internal.Command import Database.EventStore.Internal.Communication import Database.EventStore.Internal.Connection diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -11,7 +11,7 @@ * 64bits system * GHC >= 8.0.3 * Cabal >= 1.18 - * EventStore >= 4 + * EventStore >= 4 (Doesn't support EventStore 2020 Preview yet, previously named version 6). *Note: If you use this client version >= to `1.1`, it will only supports EventStore >= 4.0.0.* @@ -26,10 +26,8 @@ * From source ``` -$ git clone https://gitlab.com/YoEight/eventstore-hs.git +$ git clone https://github.com/YoEight/eventstore.git $ cd eventstore -$ cabal install --only-dependencies -$ cabal configure $ cabal install ``` @@ -37,8 +35,6 @@ =========== Tests are available. Those assume a server is running on `127.0.0.1` and `1113` port. ``` -$ cabal install --only-dependencies --enable-tests -$ cabal configure --enable-tests $ cabal test ``` diff --git a/eventstore.cabal b/eventstore.cabal --- a/eventstore.cabal +++ b/eventstore.cabal @@ -1,13 +1,13 @@ cabal-version: 1.12 --- This file has been generated from package.yaml by hpack version 0.31.2. +-- This file has been generated from package.yaml by hpack version 0.33.0. -- -- see: https://github.com/sol/hpack -- --- hash: bf11e2eb3f56e21967cbcbcdbafbb405a8955a6942e4b7f12ddaae7597d57737 +-- hash: ddd7a1bceafaa54e7b8bf636d520fca1eaa6569f66fe19640603f77e065c9469 name: eventstore -version: 1.3.3 +version: 1.4.0 synopsis: EventStore TCP Client description: EventStore TCP Client category: Database @@ -17,7 +17,7 @@ copyright: Yorick Laupa license: BSD3 license-file: LICENSE -tested-with: GHC >= 7.8 && <= 8.6 +tested-with: GHC >= 8.0 && <= 8.8 build-type: Simple extra-source-files: README.md @@ -34,7 +34,6 @@ Database.EventStore.Internal.Test other-modules: Database.EventStore.Internal - Database.EventStore.Internal.Callback Database.EventStore.Internal.Command Database.EventStore.Internal.Communication Database.EventStore.Internal.Connection @@ -67,7 +66,6 @@ Database.EventStore.Internal.Operation.Write.Common Database.EventStore.Internal.Operation.WriteEvents Database.EventStore.Internal.Operation.WriteEvents.Message - Database.EventStore.Internal.OperationManager Database.EventStore.Internal.Operations Database.EventStore.Internal.Prelude Database.EventStore.Internal.Settings @@ -100,7 +98,6 @@ , containers , dns >=3.0.1 , dotnet-timespan - , ekg-core , exceptions , fast-logger , hashable diff --git a/tests/Test/Integration/Tests.hs b/tests/Test/Integration/Tests.hs --- a/tests/Test/Integration/Tests.hs +++ b/tests/Test/Integration/Tests.hs @@ -17,6 +17,7 @@ module Test.Integration.Tests (spec) where -------------------------------------------------------------------------------- +import Prelude (fail) import Control.Concurrent.Async (wait) import Data.Aeson import Data.DotNet.TimeSpan @@ -38,22 +39,20 @@ , Slice(..) , ConnectionType(..) , Connection + , SubAction(..) + , acknowledge , getStreamMetadata , sendEvent , sendEvents , setStreamMetadata - , nextEvent - , notifyEventsProcessed - , waitConfirmation + , nextSubEvent + , streamSubEvents , connectToPersistentSubscription , unsubscribe , createPersistentSubscription , deletePersistentSubscription , updatePersistentSubscription - , nextEventMaybe - , waitUnsubscribeConfirmed , subscribeFrom - , waitTillCatchup , readEventsBackward , readEventsForward , subscribe @@ -78,7 +77,9 @@ , transactionCommit , i ) +import Data.Maybe (fromJust) import Database.EventStore.Streaming +import qualified Streaming.Prelude as Streaming import Test.Common -------------------------------------------------------------------------------- @@ -260,22 +261,35 @@ ] evts = fmap (createEvent "foo" Nothing . withJson) jss sub <- subscribe conn stream NoResolveLink Nothing - _ <- waitConfirmation sub - _ <- sendEvents conn stream anyVersion evts Nothing >>= wait - let loop 3 = return [] - loop i = do - e <- nextEvent sub - fmap (resolvedEventDataAsJson e:) $ loop (i+1) + confirmChan <- newChan + resultChan <- newChan - nxt_js <- loop (0 :: Int) - assertEqual "Events should be equal" jss (catMaybes nxt_js) - unsubscribe sub - let action = do - _ <- nextEvent sub - return False - res <- catch action $ \(_ :: SubscriptionClosed) -> return True - assertBool "Should have raised an exception" res + fork $ + do let loop state@(acc, cnt) + | cnt == 3 + = writeChan resultChan (reverse acc) + + | otherwise + = do tpe <- nextSubEvent sub + case tpe of + Confirmed _ + -> do writeChan confirmChan () + loop state + + Submit evt + -> loop (evt:acc, cnt+1) + + Dropped _ + -> writeChan resultChan (reverse acc) + loop ([],0) + + readChan confirmChan + _ <- sendEvents conn stream anyVersion evts Nothing >>= wait + result <- fmap (fromJust . resolvedEventDataAsJson) <$> readChan resultChan + + assertEqual "Events should match" jss result + -------------------------------------------------------------------------------- subscribeFromTest :: Connection -> IO () subscribeFromTest conn = do @@ -294,74 +308,76 @@ evts2 = fmap (createEvent "foo" Nothing . withJson) jss2 _ <- sendEvents conn stream anyVersion evts Nothing >>= wait sub <- subscribeFrom conn stream NoResolveLink Nothing (Just 1) Nothing - _ <- waitConfirmation sub - _ <- sendEvents conn stream anyVersion evts2 Nothing >>= wait + confirmChan <- newChan + resultChan <- newChan - let loop [] = do - m <- nextEventMaybe sub - case m of - Just _ -> fail "should not have more events at the point." - Nothing -> return () - loop (x:xs) = do - evt <- nextEvent sub - case recordedEventDataAsJson $ resolvedEventOriginal evt of - Just e | e == x -> loop xs - | otherwise -> fail "Out of order event's appeared." - _ -> fail "Can't deserialized event" - loop alljss - unsubscribe sub - waitUnsubscribeConfirmed sub - let action = do - _ <- nextEvent sub - return False - res <- catch action $ \(_ :: SubscriptionClosed) -> return True - assertBool "Should have raised an exception" res + fork $ + do let loop state@(acc, cnt) + | cnt == 6 + = writeChan resultChan (reverse acc) --------------------------------------------------------------------------------- -data SubNoStreamTest - = SubNoStreamTestSuccess - | SubNoStreamTestTimeout - deriving (Eq, Show) + | otherwise + = do tpe <- nextSubEvent sub + case tpe of + Confirmed _ + -> do writeChan confirmChan () + loop state + Submit evt + -> do loop (evt:acc, cnt+1) + + Dropped _ + -> writeChan resultChan (reverse acc) + loop ([],0) + + readChan confirmChan + _ <- sendEvents conn stream anyVersion evts2 Nothing >>= wait + result <- fmap (fromJust . resolvedEventDataAsJson) <$> readChan resultChan + + assertEqual "Events should match" alljss result + -------------------------------------------------------------------------------- subscribeFromNoStreamTest :: Connection -> IO () subscribeFromNoStreamTest conn = do stream <- freshStreamId sub <- subscribeFrom conn stream NoResolveLink Nothing Nothing Nothing - let loop [] = do - m <- nextEventMaybe sub - case m of - Just _ -> fail "should not have more events at the point." - Nothing -> return () - loop (x:xs) = do - evt <- nextEvent sub - case recordedEventDataAsJson $ resolvedEventOriginal evt of - Just e | e == x -> loop xs - | otherwise -> fail "Out of order event's appeared." - _ -> fail "Can't deserialized event" + confirmChan <- newChan + resultChan <- newChan - subAction = do - waitTillCatchup sub - let jss = [ object [ "1" .= (1 :: Int)] - , object [ "2" .= (2 :: Int)] - , object [ "3" .= (3 :: Int)] - ] - evts = fmap (createEvent "foo" Nothing . withJson) jss + fork $ + do let loop state@(acc, cnt) + | cnt == 3 + = writeChan resultChan (reverse acc) - _ <- sendEvents conn stream anyVersion evts Nothing >>= wait - loop jss - return SubNoStreamTestSuccess - timeout = do - threadDelay (12 * secs) - return SubNoStreamTestTimeout + | otherwise + = do tpe <- nextSubEvent sub + case tpe of + Confirmed _ + -> do writeChan confirmChan () + loop state - res <- race subAction timeout - case res of - Left r -> assertEqual "Wrong test result" SubNoStreamTestSuccess r - Right r -> assertEqual "Wrong test result" SubNoStreamTestSuccess r + Submit evt + -> loop (evt:acc, cnt+1) + Dropped _ + -> writeChan resultChan (reverse acc) + loop ([],0) + + readChan confirmChan + let jss = [ object [ "1" .= (1 :: Int)] + , object [ "2" .= (2 :: Int)] + , object [ "3" .= (3 :: Int)] + ] + + evts = fmap (createEvent "foo" Nothing . withJson) jss + + _ <- sendEvents conn stream anyVersion evts Nothing >>= wait + result <- fmap (fromJust . resolvedEventDataAsJson) <$> readChan resultChan + + assertEqual "Events should match" jss result + -------------------------------------------------------------------------------- setStreamMetadataTest :: Connection -> IO () setStreamMetadataTest conn = do @@ -483,29 +499,35 @@ evts = fmap (createEvent "foo" Nothing . withJson) jss stream <- freshStreamId _ <- createPersistentSubscription conn "group" stream def Nothing >>= wait - _ <- sendEvents conn stream anyVersion evts Nothing >>= wait sub <- connectToPersistentSubscription conn "group" stream 1 Nothing - _ <- waitConfirmation sub - r <- nextEvent sub - case resolvedEventDataAsJson r of - Just js_evt -> assertEqual "event 1 should match" js1 js_evt - _ -> fail "Deserialization error" + confirmChan <- newChan + resultChan <- newChan - notifyEventsProcessed sub [resolvedEventOriginalId r] + fork $ + do let loop state@(acc, cnt) + | cnt == 2 + = writeChan resultChan (reverse acc) - r2 <- nextEvent sub - case resolvedEventDataAsJson r2 of - Just js_evt -> assertEqual "event 2 should match" js2 js_evt - _ -> fail "Deserialization error" + | otherwise + = do tpe <- nextSubEvent sub + case tpe of + Confirmed _ + -> do writeChan confirmChan () + loop state - notifyEventsProcessed sub [resolvedEventOriginalId r2] + Submit evt + -> do acknowledge sub evt + loop (evt:acc, cnt+1) - unsubscribe sub - let action = do - _ <- nextEvent sub - return False - res <- catch action $ \(_ :: SubscriptionClosed) -> return True - assertBool "Should have raised an exception" res + Dropped _ + -> writeChan resultChan (reverse acc) + loop ([],0) + + readChan confirmChan + _ <- sendEvents conn stream anyVersion evts Nothing >>= wait + result <- fmap (fromJust . resolvedEventDataAsJson) <$> readChan resultChan + + assertEqual "Both events should match" [js1,js2] result -------------------------------------------------------------------------------- maxAgeTest :: Connection -> IO () diff --git a/tests/Test/Operation.hs b/tests/Test/Operation.hs --- a/tests/Test/Operation.hs +++ b/tests/Test/Operation.hs @@ -57,10 +57,7 @@ exec <- newExec testSettings bus builder testDisc - p <- newPromise - let op = readEvent testSettings "foo" 1 True Nothing - publishWith exec (SubmitOperation p op) - + readEvent testSettings exec "foo" 1 True Nothing res <- takeMVar var publishWith exec SystemShutdown