eventstore 1.3.3 → 1.4.0

raw patch · 35 files changed

+1545/−1968 lines, 35 filesdep −ekg-corePVP ok

version bump matches the API change (PVP)

Dependencies removed: ekg-core

API changes (from Hackage documentation)

Files

• CHANGELOG.markdown +7/−0
• Database/EventStore.hs +7/−16
• Database/EventStore/Internal.hs +29/−72
• Database/EventStore/Internal/Callback.hs +0/−111
• Database/EventStore/Internal/Communication.hs +4/−7
• Database/EventStore/Internal/ConnectionManager.hs +28/−31
• Database/EventStore/Internal/Control.hs +12/−37
• Database/EventStore/Internal/Discovery.hs +1/−1
• Database/EventStore/Internal/Exec.hs +1/−1
• Database/EventStore/Internal/Manager/Operation/Registry.hs +301/−349
• Database/EventStore/Internal/Operation.hs +108/−353
• Database/EventStore/Internal/Operation/Catchup.hs +208/−69
• Database/EventStore/Internal/Operation/DeleteStream.hs +44/−27
• Database/EventStore/Internal/Operation/Persist.hs +50/−53
• Database/EventStore/Internal/Operation/PersistOperations.hs +70/−46
• Database/EventStore/Internal/Operation/ReadAllEvents.hs +42/−31
• Database/EventStore/Internal/Operation/ReadEvent.hs +36/−21
• Database/EventStore/Internal/Operation/ReadStreamEvents.hs +45/−36
• Database/EventStore/Internal/Operation/StreamMetadata.hs +36/−30
• Database/EventStore/Internal/Operation/Transaction.hs +104/−66
• Database/EventStore/Internal/Operation/Volatile.hs +47/−52
• Database/EventStore/Internal/Operation/Write/Common.hs +4/−4
• Database/EventStore/Internal/Operation/WriteEvents.hs +46/−29
• Database/EventStore/Internal/OperationManager.hs +0/−60
• Database/EventStore/Internal/Settings.hs +43/−5
• Database/EventStore/Internal/Subscription/Api.hs +24/−79
• Database/EventStore/Internal/Subscription/Catchup.hs +35/−104
• Database/EventStore/Internal/Subscription/Persistent.hs +63/−101
• Database/EventStore/Internal/Subscription/Regular.hs +28/−64
• Database/EventStore/Internal/Subscription/Types.hs +0/−1
• Database/EventStore/Internal/Test.hs +1/−3
• README.md +2/−6
• eventstore.cabal +4/−7
• tests/Test/Integration/Tests.hs +114/−92
• tests/Test/Operation.hs +1/−4

@@ -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.

@@ -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

@@ -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

@@ -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 <yo.eight@gmail.com>--- 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

@@ -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

@@ -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 ()

@@ -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

@@ -34,7 +34,7 @@     ) where  ---------------------------------------------------------------------------------import Prelude (String)+import Prelude (String, fail) import Data.Maybe  --------------------------------------------------------------------------------

@@ -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)

@@ -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 <yo.eight@gmail.com>--- Stability : provisional--- Portability : non-portable+-- Module    :  Database.EventStore.Internal.Manager.Registry+-- Copyright :  (C) 2020 Yorick Laupa+-- License   :  (see the file LICENSE)+-- Maintainer:  Yorick Laupa <yo.eight@gmail.com>+-- 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

@@ -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 <yo.eight@gmail.com>--- Stability : provisional--- Portability : non-portable+-- Module    :  Database.EventStore.Internal.Operation+-- Copyright :  (C) 2020 Yorick Laupa+-- License   :  (see the file LICENSE)+-- Maintainer:  Yorick Laupa <yo.eight@gmail.com>+-- 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

@@ -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

@@ -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)++

@@ -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)

@@ -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

@@ -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

@@ -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++

@@ -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)

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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)+

@@ -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 <yo.eight@gmail.com>--- 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

@@ -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 ()+    }

@@ -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)

@@ -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

@@ -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)

@@ -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

@@ -135,4 +135,3 @@   = Submit ResolvedEvent   | Dropped SubDropReason   | Confirmed SubDetails-  | ConnectionReset

@@ -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

@@ -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 ``` 

@@ -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 <https://eventstore.org> 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

@@ -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 ()

@@ -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