eventstore 1.3.0 → 1.3.1

raw patch · 5 files changed

+104/−36 lines, 5 filesPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

API changes (from Hackage documentation)

Files

• CHANGELOG.markdown +9/−0
• Database/EventStore/Internal/ConnectionManager.hs +36/−5
• Database/EventStore/Internal/Discovery.hs +1/−2
• Database/EventStore/Streaming.hs +54/−25
• eventstore.cabal +4/−4

@@ -1,3 +1,12 @@+1.4.0+-----+* Better encoding of streaming interface ReadError.+* Expose more internal functions of the streaming interface.+* Internal: Reset heartbeat tracking on force reconnect request.+* Bug fix: Increment package number unless connection is init or closed (prevent false heartbeat timeout).+* Bug fix: Was also using `s_heartbeatInterval` value for `s_heartbeatTimeout` causing false heartbeat timeout exceptions.+* Bug fix: Stop trying to reconnect to the most recent node that failed (cluster connection mode).+ 1.3.0 ----- * Discard `machines` dependency.

@@ -65,6 +65,18 @@   deriving Show  --------------------------------------------------------------------------------+atLeastEstablishingState :: Stage -> Bool+atLeastEstablishingState = \case+  Init -> False+  Connected{} -> True+  Closed -> True+  Connecting a s ->+    case s of+      Reconnecting -> False+      EndpointDiscovery -> False+      _ -> True++-------------------------------------------------------------------------------- data Attempts =   Attempts { attemptCount     :: !Int            , attemptLastStart :: !NominalDiffTime@@ -366,6 +378,10 @@     monitorIncrForceReconnect     closeTcpConnection self (ForceReconnect ept) conn     att <- freshAttempt _stopwatch++    -- We update the last pkg number to nullify the current heartbeat tracking.+    atomicModifyIORef' _lastPkgNum $ \cur -> (cur + 1, ())+     atomicWriteIORef _stage (Connecting att EndpointDiscovery)     $logInfo [i|#{conn}: going to reconnect to #{ept}.|]     establish self ept@@ -415,9 +431,13 @@         Operation.check _opMgr         atomicWriteIORef _lastCheck elapsed -      manageHeartbeats self-     _ -> return ()++  -- FIXME - This `readIORef` call can be merged into the previous one.+  -- Done in 2019's refactoring code.+  stage <- readIORef _stage+  when (atLeastEstablishingState stage) $+    manageHeartbeats self   where     onGoingConnection (Connecting att Reconnecting)             = Just att     onGoingConnection (Connecting att ConnectionEstablishing{}) = Just att@@ -461,7 +481,7 @@       setts   <- getSettings        let interval    = s_heartbeatInterval setts-          timeout     = s_heartbeatInterval setts+          timeout     = s_heartbeatTimeout setts           initTracker = tracker                         { _heartbeatStage = Interval                         , _startedSince   = elapsed@@ -493,7 +513,14 @@  -------------------------------------------------------------------------------- onArrived :: Internal -> PackageArrived -> EventStore ()-onArrived self@Internal{..} (PackageArrived conn pkg@Package{..}) =+onArrived self@Internal{..} (PackageArrived conn pkg@Package{..}) = do+  -- FIXME - We can merge those two `readIORef` calls into one.+  -- It's done on 2019's refactoring.+  cur <- readIORef _stage+  unless (closedOrInit cur) $+    incrPackageNumber self+  -- /FIXME+   readIORef _stage >>= \case     (onAuthentication -> Just att) -> do       when (packageCmd == notAuthenticatedCmd) $@@ -506,7 +533,6 @@      (runningConnection -> True) -> do       $logDebug [i|Package received:  #{pkg}.|]-      incrPackageNumber self       handlePackage      _ -> $logDebug [i|Package IGNORED: #{pkg}.|]@@ -541,6 +567,11 @@             case decision of               Operation.Handled        -> return ()               Operation.Reconnect node -> forceReconnect self node++    closedOrInit = \case+      Init -> True+      Closed -> True+      _ -> False  -------------------------------------------------------------------------------- isSameConnection :: Internal -> Connection -> EventStore Bool

@@ -361,8 +361,7 @@         case fend_m of             Nothing   -> oldGossip             Just fend -> [ c | c <- oldGossip-                             , _externalTcpPort c == endPointPort fend-                             , _externalTcpIp c   == endPointIp fend+                             , EndPoint (_externalTcpIp c) (_externalTcpPort c) /= fend                              ]  --------------------------------------------------------------------------------

@@ -1,4 +1,6 @@-{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE GADTs              #-}+{-# LANGUAGE LambdaCase         #-}+{-# LANGUAGE StandaloneDeriving #-} -------------------------------------------------------------------------------- -- | -- Module    :  Database.EventStore.Streaming@@ -11,9 +13,14 @@ -------------------------------------------------------------------------------- module Database.EventStore.Streaming     ( ReadError(..)+    , Fetch(..)+    , ReadResultHandler(..)     , readThroughForward     , readThroughBackward     , throwOnError+    , defaultReadResultHandler+    , onRegularStream+    , readThrough     ) where  --------------------------------------------------------------------------------@@ -34,33 +41,51 @@ import qualified Database.EventStore as ES  ---------------------------------------------------------------------------------data ReadError t-    = StreamDeleted !ES.StreamName-    | ReadError !(Maybe Text)-    | AccessDenied !(ES.StreamId t)-    deriving Show+data ReadError t where+    StreamDeleted :: ES.StreamName -> ReadError ES.EventNumber+    ReadError :: Maybe Text -> ReadError t+    AccessDenied :: ES.StreamId t -> ReadError t+    NoStream :: ReadError ES.EventNumber  --------------------------------------------------------------------------------+deriving instance Show (ReadError t)++-------------------------------------------------------------------------------- instance (Show t, Typeable t) => Exception (ReadError t)  -------------------------------------------------------------------------------- data Fetch t = FetchError !(ReadError t) | Fetch !(ES.Slice t)  ---------------------------------------------------------------------------------toFetch :: ES.ReadResult t (ES.Slice t) -> Fetch t-toFetch ES.ReadNoStream          = Fetch ES.emptySlice-toFetch ES.ReadNotModified       = Fetch ES.emptySlice-toFetch (ES.ReadStreamDeleted n) = FetchError (StreamDeleted n)-toFetch (ES.ReadError e)         = FetchError (ReadError e)-toFetch (ES.ReadAccessDenied n)  = FetchError (AccessDenied n)-toFetch (ES.ReadSuccess s)       = Fetch s+newtype ReadResultHandler =+  ReadResultHandler+  { runReadResultHandler :: forall t. ES.StreamId t -> ES.BatchResult t -> Fetch t }  ---------------------------------------------------------------------------------handleBatchResult :: ES.StreamId t -> ES.BatchResult t -> Fetch t-handleBatchResult ES.StreamName{} = toFetch-handleBatchResult ES.All          = Fetch+defaultReadResultHandler :: ReadResultHandler+defaultReadResultHandler = ReadResultHandler go+  where+    go :: ES.StreamId t -> ES.BatchResult t -> Fetch t+    go ES.StreamName{} = toFetch+    go ES.All          = Fetch +    toFetch ES.ReadNoStream          = Fetch ES.emptySlice+    toFetch ES.ReadNotModified       = Fetch ES.emptySlice+    toFetch (ES.ReadStreamDeleted n) = FetchError (StreamDeleted n)+    toFetch (ES.ReadError e)         = FetchError (ReadError e)+    toFetch (ES.ReadAccessDenied n)  = FetchError (AccessDenied n)+    toFetch (ES.ReadSuccess s)       = Fetch s+ --------------------------------------------------------------------------------+onRegularStream :: (ES.ReadResult ES.EventNumber (ES.Slice ES.EventNumber) -> Fetch ES.EventNumber)+                -> ReadResultHandler+onRegularStream callback = ReadResultHandler go+  where+    go :: ES.StreamId t -> ES.BatchResult t -> Fetch t+    go ES.StreamName{} = callback+    go ES.All          = Fetch++-------------------------------------------------------------------------------- data State t = Need t | Fetched ![ES.ResolvedEvent] !(Maybe t)  --------------------------------------------------------------------------------@@ -93,7 +118,7 @@                    -> Maybe Int32                    -> Maybe ES.Credentials                    -> Stream (Of ES.ResolvedEvent) (ExceptT (ReadError t) IO) ()-readThroughForward conn = readThrough conn ES.Forward+readThroughForward conn = readThrough conn defaultReadResultHandler ES.Forward  -------------------------------------------------------------------------------- -- | Returns an iterator able to consume a stream entirely. When reading backward,@@ -105,7 +130,7 @@                     -> Maybe Int32                     -> Maybe ES.Credentials                     -> Stream (Of ES.ResolvedEvent) (ExceptT (ReadError t) IO) ()-readThroughBackward conn = readThrough conn ES.Backward+readThroughBackward conn = readThrough conn defaultReadResultHandler ES.Backward  -------------------------------------------------------------------------------- -- | Throws an exception in case 'ExceptT' is a 'Left'.@@ -120,7 +145,9 @@             Right a -> pure a  --------------------------------------------------------------------------------+-- | Returns an iterator able to consume a stream entirely. readThrough :: ES.Connection+            -> ReadResultHandler             -> ES.ReadDirection             -> ES.StreamId t             -> ES.ResolveLink@@ -128,35 +155,37 @@             -> Maybe Int32             -> Maybe ES.Credentials             -> Stream (Of ES.ResolvedEvent) (ExceptT (ReadError t) IO) ()-readThrough conn dir streamId lnk from sizMay cred = streaming iteratee from+readThrough conn handler dir streamId lnk from sizMay cred = streaming iteratee from   where     batchSize = fromMaybe 500 sizMay      iteratee =         case dir of-            ES.Forward  -> readForward conn streamId batchSize lnk cred-            ES.Backward -> readBackward conn streamId batchSize lnk cred+            ES.Forward  -> readForward conn handler streamId batchSize lnk cred+            ES.Backward -> readBackward conn handler streamId batchSize lnk cred  -------------------------------------------------------------------------------- readForward :: ES.Connection+            -> ReadResultHandler             -> ES.StreamId t             -> Int32             -> ES.ResolveLink             -> Maybe ES.Credentials             -> t             -> IO (Fetch t)-readForward conn streamId siz lnk creds start =-    fmap (handleBatchResult streamId) . wait =<<+readForward conn handler streamId siz lnk creds start =+    fmap (runReadResultHandler handler streamId) . wait =<<         ES.readEventsForward conn streamId start siz lnk creds  -------------------------------------------------------------------------------- readBackward :: ES.Connection+             -> ReadResultHandler              -> ES.StreamId t              -> Int32              -> ES.ResolveLink              -> Maybe ES.Credentials              -> t              -> IO (Fetch t)-readBackward conn streamId siz lnk creds start =-    fmap (handleBatchResult streamId) . wait =<<+readBackward conn handler streamId siz lnk creds start =+    fmap (runReadResultHandler handler streamId) . wait =<<         ES.readEventsBackward conn streamId start siz lnk creds

@@ -4,14 +4,14 @@ -- -- see: https://github.com/sol/hpack ----- hash: 663b08f5bc970725b2a57c47aa571a768636e650f33cdb9b1d1a1b3d0d351dd4+-- hash: 1183167359a5b717d35d6bbd4ca75dcce93e5223d77d575319cd5fe83124a473  name:           eventstore-version:        1.3.0+version:        1.3.1 synopsis:       EventStore TCP Client description:    EventStore TCP Client <https://eventstore.org> category:       Database-homepage:       https://gitlab.com/YoEight/eventstore-hs+homepage:       https://github.com/YoEight/eventstore author:         Yorick Laupa maintainer:     yo.eight@gmail.com copyright:      Yorick Laupa@@ -25,7 +25,7 @@  source-repository head   type: git-  location: https://gitlab.com/YoEight/eventstore-hs.git+  location: https://github.com/YoEight/eventstore.git  library   exposed-modules: