eventstore 1.2.4 → 1.3.0
raw patch · 15 files changed
+262/−116 lines, 15 filesdep −machinesdep ~connectionPVP ok
version bump matches the API change (PVP)
Dependencies removed: machines
Dependency ranges changed: connection
API changes (from Hackage documentation)
Files
- CHANGELOG.markdown +5/−0
- Database/EventStore.hs +2/−3
- Database/EventStore/Internal.hs +0/−10
- Database/EventStore/Internal/Discovery.hs +9/−9
- Database/EventStore/Internal/Manager/Operation/Registry.hs +20/−14
- Database/EventStore/Internal/Operation.hs +173/−24
- Database/EventStore/Internal/Operation/Catchup.hs +25/−26
- Database/EventStore/Internal/Operation/Read/Common.hs +1/−2
- Database/EventStore/Internal/Operation/StreamMetadata.hs +16/−16
- Database/EventStore/Internal/Subscription/Catchup.hs +2/−2
- Database/EventStore/Internal/Subscription/Persistent.hs +1/−0
- Database/EventStore/Internal/Subscription/Regular.hs +1/−0
- Database/EventStore/Internal/Types.hs +2/−2
- Database/EventStore/Streaming.hs +0/−2
- eventstore.cabal +5/−6
CHANGELOG.markdown view
@@ -1,3 +1,8 @@+1.3.0+-----+* Discard `machines` dependency.+* Remove `connection` upper bound version.+ 1.2.4 ----- * Fix possible connection issues if Authentication or Identification processes takes too long to complete.
Database/EventStore.hs view
@@ -56,7 +56,7 @@ , OperationMaxAttemptReached(..) -- * Read Operations , StreamMetadataResult(..)- , BatchResult(..)+ , BatchResult , ResolveLink(..) , readEvent , readEventsBackward@@ -169,7 +169,7 @@ , ResolvedEvent(..) , OperationError(..) , StreamId(..)- , StreamName(..)+ , StreamName , isAllStream , isEventResolvedLink , resolvedEventOriginal@@ -210,7 +210,6 @@ -------------------------------------------------------------------------------- import Database.EventStore.Internal-import Database.EventStore.Internal.Callback import Database.EventStore.Internal.Command import Database.EventStore.Internal.Discovery import Database.EventStore.Internal.Logger
Database/EventStore/Internal.hs view
@@ -21,29 +21,19 @@ import Prelude (String) import Data.Int import Data.Maybe-import Data.Time (NominalDiffTime) ---------------------------------------------------------------------------------import Data.List.NonEmpty(NonEmpty(..), nonEmpty)-import Network.Connection (TLSSettings)---------------------------------------------------------------------------------- import Database.EventStore.Internal.Callback-import Database.EventStore.Internal.Command import Database.EventStore.Internal.Communication import Database.EventStore.Internal.Connection (connectionBuilder) import Database.EventStore.Internal.Control hiding (subscribe) import Database.EventStore.Internal.Discovery import Database.EventStore.Internal.Exec-import Database.EventStore.Internal.Subscription.Api import Database.EventStore.Internal.Subscription.Catchup-import Database.EventStore.Internal.Subscription.Message import Database.EventStore.Internal.Subscription.Persistent import Database.EventStore.Internal.Subscription.Types import Database.EventStore.Internal.Subscription.Regular import Database.EventStore.Internal.Logger-import Database.EventStore.Internal.Manager.Operation.Registry-import Database.EventStore.Internal.Operation (OperationError(..)) import qualified Database.EventStore.Internal.Operations as Op import Database.EventStore.Internal.Operation.Read.Common import Database.EventStore.Internal.Operation.Write.Common
Database/EventStore/Internal/Discovery.hs view
@@ -38,7 +38,7 @@ import Data.Maybe ---------------------------------------------------------------------------------import Control.Exception.Safe (SomeException, tryAny)+import Control.Exception.Safe (tryAny) import Data.Aeson import Data.Aeson.Types import Data.Array.IO@@ -302,8 +302,8 @@ candidates <- case old_m of Nothing -> gossipCandidatesFromDns settings Just old -> liftIO $ gossipCandidatesFromOldGossip fend old- forArrayFirst candidates $ \i -> do- c <- liftIO $ readArray candidates i+ forArrayFirst candidates $ \idx -> do+ c <- liftIO $ readArray candidates idx res <- tryGetGossipFrom settings mgr c let fin_end = do info <- res@@ -371,10 +371,10 @@ -------------------------------------------------------------------------------- arrangeGossipCandidates :: [MemberInfo] -> IO (IOArray Int GossipSeed) arrangeGossipCandidates members = do- arr <- newArray (0, len) emptyGossipSeed- AState i j <- foldM (go arr) (AState (-1) len) members+ arr <- newArray (0, len) emptyGossipSeed+ AState idx j <- foldM (go arr) (AState (-1) len) members - shuffle arr 0 i -- shuffle nodes+ shuffle arr 0 idx -- shuffle nodes shuffle arr j (len - 1) -- shuffle managers return arr@@ -382,14 +382,14 @@ len = length members go :: IOArray Int GossipSeed -> AState -> MemberInfo -> IO AState- go arr (AState i j) m =+ go arr (AState idx j) m = case _state m of Manager -> do let new_j = j - 1 writeArray arr new_j seed- return (AState i new_j)+ return (AState idx new_j) _ -> do- let new_i = i + 1+ let new_i = idx + 1 writeArray arr new_i seed return (AState new_i j) where
Database/EventStore/Internal/Manager/Operation/Registry.hs view
@@ -32,6 +32,7 @@ import Data.Time import Data.UUID import Data.UUID.V4+import Data.Void (absurd) -------------------------------------------------------------------------------- import Database.EventStore.Internal.Callback@@ -221,16 +222,21 @@ Resolved action -> loop action _ -> return () where- loop (MachineT m) =- case m of- Failed e -> do- destroySession _sessions session- rejectSession session e- Retry -> do- reinitSession session- execute self session- Proceed s ->- case s of+ 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@@ -238,10 +244,10 @@ Await k tpe _ -> case tpe of NeedUUID -> loop . k =<< freshUUID- NeedRemote cmd payload cred -> do- let req = Request { _requestCmd = cmd- , _requestPayload = payload- , _requestCred = cred+ NeedRemote p -> do+ let req = Request { _requestCmd = payloadCmd p+ , _requestPayload = payloadData p+ , _requestCred = payloadCreds p } atomicWriteIORef sessionStack (Required k) maybeConnection _regConnRef >>= \case
Database/EventStore/Internal/Operation.hs view
@@ -21,16 +21,17 @@ ( OpResult(..) , OperationError(..) , Operation- , Need(..) , Code+ , Need(..) , Execution(..) , Expect(..)+ , Coroutine(..)+ , Payload(..) , freshId , failure , retry , send , request- , waitFor , waitForOr , wrongVersion , streamDeleted@@ -39,26 +40,38 @@ , protobufDecodingError , serverError , invalidServerResponse- , module Data.Machine+ , construct+ , yield+ , traversing+ , stop+ , (<~)+ , unfolding+ , append+ , Stream(..) ) where -------------------------------------------------------------------------------- import Prelude (String)+import Control.Category ---------------------------------------------------------------------------------import Data.Machine import Data.ProtocolBuffers import Data.Serialize import Data.UUID+import Data.Void (Void, absurd)+import Streaming.Internal -------------------------------------------------------------------------------- import Database.EventStore.Internal.Command-import Database.EventStore.Internal.Prelude+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@@ -97,6 +110,41 @@ 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++--------------------------------------------------------------------------------+instance Category Is where+ id = Same+ Same . Same = Same++--------------------------------------------------------------------------------+data Need a where+ NeedUUID :: Need UUID+ NeedRemote :: Payload -> Need Package+ WaitRemote :: UUID -> Need (Maybe Package)++-------------------------------------------------------------------------------- data Execution a = Proceed a | Retry@@ -122,22 +170,111 @@ Failed e >>= _ = Failed e ---------------------------------------------------------------------------------type Operation output = MachineT Execution Need output+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 ---------------------------------------------------------------------------------data Need a where- NeedUUID :: Need UUID- NeedRemote :: Command -> ByteString -> Maybe Credentials -> Need Package- WaitRemote :: UUID -> Need (Maybe Package)+awaits :: Monad m => k i -> Machine k o m i+awaits instr = Step (Await pure instr (Step Stop)) ----------------------------------------------------------------------------------- | Instruction that composed an 'Operation'.-type Code o a = PlanT Need o Execution a+await :: (Monad m, Category k) => Machine (k i) o m i+await = awaits id --------------------------------------------------------------------------------+stop :: Machine k o m a+stop = Step Stop++--------------------------------------------------------------------------------+yield :: Monad m => o -> Machine k o m ()+yield o = Step (Yield o (pure ()))++--------------------------------------------------------------------------------+traversing :: Monad m => (a -> m b) -> Process m a b+traversing k = repeatedly $ do+ a <- await+ b <- lift (k a)+ yield b++--------------------------------------------------------------------------------+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++--------------------------------------------------------------------------------+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++--------------------------------------------------------------------------------+unfolding :: (Maybe a -> Code o (Operation a)) -> Operation o+unfolding k = k Nothing >>= go+ where+ go cur = do+ (a, next) <- stepping cur+ newState <- k (Just a)+ go (append next newState)++--------------------------------------------------------------------------------+repeatedly :: Functor m => Machine k o m x -> Machine k o m r+repeatedly start = go start+ 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 = awaits NeedUUID+freshId = Step $ Await pure NeedUUID (Step Stop) -------------------------------------------------------------------------------- -- | Raises an 'OperationError'.@@ -150,6 +287,20 @@ 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)@@ -159,8 +310,9 @@ -> req -> Code o resp send reqCmd expCmd cred req = do- let payload = runPut $ encodeMessage req- pkg <- awaits $ NeedRemote reqCmd payload cred+ let dat = runPut $ encodeMessage req+ payload = Payload reqCmd dat cred+ pkg <- sendRemote payload let gotCmd = packageCmd pkg when (gotCmd /= expCmd)@@ -195,22 +347,18 @@ -> [Expect o] -> Code o () request reqCmd cred rq exps = do- let payload = runPut $ encodeMessage rq- pkg <- awaits $ NeedRemote reqCmd payload cred+ let dat = runPut $ encodeMessage rq+ payload = Payload reqCmd dat cred+ pkg <- sendRemote payload runFirstMatch pkg exps ----------------------------------------------------------------------------------- | Like 'waitForOr' but will 'stop' if the connection reset.-waitFor :: UUID -> [Expect o] -> Code o ()-waitFor pid exps = waitForOr pid stop 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 :: UUID -> Code o () -> [Expect o] -> Code o () waitForOr pid alt exps =- awaits (WaitRemote pid) >>= \case+ waitRemote pid >>= \case Nothing -> alt Just pkg -> runFirstMatch pkg exps@@ -249,6 +397,7 @@ -- | Raises 'InvalidServerResponse' exception. invalidServerResponse :: Command -> Command -> Code o a invalidServerResponse expe got = failure $ InvalidServerResponse expe got+ -------------------------------------------------------------------------------- invalidOperation :: Text -> Code o a
Database/EventStore/Internal/Operation/Catchup.hs view
@@ -47,13 +47,12 @@ -> Bool -- Resolve link tos. -> Maybe Credentials -> EventNumber- -> Code o (Slice EventNumber)-fetchStream setts stream batch tos cred (EventNumber n) = do- outcome <-- deconstruct $ fmap Left $- readStreamEvents setts Forward stream n batch tos cred-- fromReadResult stream outcome pure+ -> 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 -------------------------------------------------------------------------------- fetchAll :: Settings@@ -61,24 +60,24 @@ -> Bool -- Resolve link tos. -> Maybe Credentials -> Position- -> Code o (Slice Position)+ -> Operation (Slice Position) fetchAll setts batch tos cred (Position com pre) =- deconstruct $ fmap Left $- readAllEvents setts com pre batch tos Forward cred+ traversing pure <~ readAllEvents setts com pre batch tos Forward cred ---------------------------------------------------------------------------------sourceStream :: t- -> (forall o. t -> Code o (Slice t))+sourceStream :: (t -> Operation (Slice t))+ -> t -> Operation SubAction-sourceStream seed iteratee = unfoldPlan seed go+sourceStream fetch start = unfolding go where- go state = do- s <- iteratee state+ go Nothing =+ pure (fetch start)+ go (Just s) = do traverse_ (yield . Submit) (sliceEvents s) case sliceNext s of- Just newState -> pure newState- Nothing -> stop+ Just next -> pure (fetch next)+ Nothing -> stop -------------------------------------------------------------------------------- catchup :: forall t. Settings@@ -89,11 +88,11 @@ -> Maybe Credentials -> Operation SubAction catchup setts streamId from tos batchSiz cred =- sourceStream from iteratee <> volatile streamId tos cred+ append (sourceStream iteratee from) (volatile streamId tos cred) where batch = fromMaybe defaultBatchSize batchSiz - iteratee :: t -> Code o (Slice t)+ iteratee :: t -> Operation (Slice t) iteratee = case streamId of StreamName n -> fetchStream setts n batch tos cred@@ -102,13 +101,13 @@ -------------------------------------------------------------------------------- fromReadResult :: Text -> ReadResult EventNumber a- -> (a -> Code o x)- -> Code o x+ -> (a -> Execution x)+ -> Execution x fromReadResult stream res k = case res of- ReadNoStream -> failure $ streamNotFound stream- ReadStreamDeleted s -> failure $ StreamDeleted s- ReadNotModified -> failure $ ServerError Nothing- ReadError e -> failure $ ServerError e- ReadAccessDenied s -> failure $ AccessDenied s+ 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
Database/EventStore/Internal/Operation/Read/Common.hs view
@@ -21,7 +21,6 @@ import Data.Maybe (isNothing) import Data.Monoid import Data.Traversable-import Data.Int -------------------------------------------------------------------------------- import Database.EventStore.Internal.Prelude@@ -97,7 +96,7 @@ -------------------------------------------------------------------------------- instance Functor Slice where- fmap f SliceEndOfStream = SliceEndOfStream+ fmap _ SliceEndOfStream = SliceEndOfStream fmap f (Slice xs next) = Slice xs (fmap f next) --------------------------------------------------------------------------------
Database/EventStore/Internal/Operation/StreamMetadata.hs view
@@ -33,7 +33,6 @@ import Database.EventStore.Internal.Operation.WriteEvents import Database.EventStore.Internal.Prelude import Database.EventStore.Internal.Settings-import Database.EventStore.Internal.Stream import Database.EventStore.Internal.Types --------------------------------------------------------------------------------@@ -47,13 +46,14 @@ -> Maybe Credentials -> Operation StreamMetadataResult readMetaStream setts s cred = construct $ do- let op = readEvent setts (metaStream s) (-1) False cred- tmp <- deconstruct (fmap Left op)- onReadResult tmp $ \n e_num evt -> do- let bytes = recordedEventData $ resolvedEventOriginal evt- case decode $ fromStrict bytes of- Just pv -> yield $ StreamMetadataResult n e_num pv- Nothing -> failure invalidFormat+ 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 -------------------------------------------------------------------------------- -- | Set stream metadata operation.@@ -79,14 +79,14 @@ -------------------------------------------------------------------------------- onReadResult :: ReadResult EventNumber ReadEvent- -> (Text -> Int64 -> ResolvedEvent -> Code o a)- -> Code o a+ -> (Text -> Int64 -> ResolvedEvent -> Execution a)+ -> Execution a onReadResult (ReadSuccess r) k = case r of ReadEvent s n e -> k s n e- _ -> failure streamNotFound-onReadResult ReadNoStream _ = failure streamNotFound-onReadResult (ReadStreamDeleted s) _ = failure $ StreamDeleted s-onReadResult ReadNotModified _ = failure $ ServerError Nothing-onReadResult (ReadError e) _ = failure $ ServerError e-onReadResult (ReadAccessDenied s) _ = failure $ AccessDenied s+ _ -> 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
Database/EventStore/Internal/Subscription/Catchup.hs view
@@ -3,6 +3,7 @@ {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RecordWildCards #-} {-# LANGUAGE ScopedTypeVariables #-}+{-# OPTIONS -Wno-orphans #-} -------------------------------------------------------------------------------- -- | -- Module : Database.EventStore.Internal.Subscription.Catchup@@ -108,8 +109,7 @@ queue <- newTQueueIO track <- newTVarIO seed - let stream = streamIdRaw streamId- sub = CatchupSubscription exec streamId phaseVar track $ do+ let sub = CatchupSubscription exec streamId phaseVar track $ do p <- readTVar phaseVar isEmpty <- isEmptyTQueue queue if isEmpty
Database/EventStore/Internal/Subscription/Persistent.hs view
@@ -1,5 +1,6 @@ {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE RecordWildCards #-}+{-# OPTIONS -Wno-orphans #-} -------------------------------------------------------------------------------- -- | -- Module : Database.EventStore.Internal.Subscription.Persistent
Database/EventStore/Internal/Subscription/Regular.hs view
@@ -1,4 +1,5 @@ {-# LANGUAGE FlexibleInstances #-}+{-# OPTIONS -Wno-orphans #-} -------------------------------------------------------------------------------- -- | -- Module : Database.EventStore.Internal.Subscription.Regular
Database/EventStore/Internal/Types.hs view
@@ -774,8 +774,8 @@ go (_, A.Null) = [] go (name, obj) = [(name, deeper obj)] - deeper cur@(A.Array xs)- | Vector.length xs == 1 = Vector.head xs+ deeper cur@(A.Array as)+ | Vector.length as == 1 = Vector.head as | otherwise = cur deeper cur = cur
Database/EventStore/Streaming.hs view
@@ -32,8 +32,6 @@ -------------------------------------------------------------------------------- import qualified Database.EventStore as ES-import Database.EventStore.Internal.Operation.Read.Common (emptySlice)-import Database.EventStore.Internal.Types (EventNumber(..)) -------------------------------------------------------------------------------- data ReadError t
eventstore.cabal view
@@ -1,13 +1,13 @@ cabal-version: 1.12 --- This file has been generated from package.yaml by hpack version 0.31.1.+-- This file has been generated from package.yaml by hpack version 0.31.2. -- -- see: https://github.com/sol/hpack ----- hash: c9911a422246a5be440ff6488ce51535e4428fcd8bb1514a94970f10f9ef698d+-- hash: 663b08f5bc970725b2a57c47aa571a768636e650f33cdb9b1d1a1b3d0d351dd4 name: eventstore-version: 1.2.4+version: 1.3.0 synopsis: EventStore TCP Client description: EventStore TCP Client <https://eventstore.org> category: Database@@ -96,7 +96,7 @@ , bytestring , cereal >=0.4 && <0.6 , clock- , connection ==0.2.*+ , connection >=0.2 , containers , dns >=3.0.1 , dotnet-timespan@@ -108,7 +108,6 @@ , interpolate , lifted-async , lifted-base- , machines >=0.6 , monad-control , monad-logger >=0.3.20 , mono-traversable ==1.*@@ -151,7 +150,7 @@ , base >=4.7 && <5 , bytestring , cereal- , connection ==0.2.*+ , connection >=0.2 , containers , dotnet-timespan , eventstore