diff --git a/Database/EventStore.hs b/Database/EventStore.hs --- a/Database/EventStore.hs +++ b/Database/EventStore.hs @@ -1,4 +1,5 @@ -{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE RecordWildCards #-} -------------------------------------------------------------------------------- -- | -- Module : Database.EventStore @@ -45,6 +46,7 @@ , DropReason(..) , Subscription , subscribe + , subscribeToAll , subAwait , subId , subStream @@ -56,6 +58,7 @@ , Catchup , CatchupError(..) , subscribeFrom + , subscribeToAllFrom , catchupAwait , catchupStream , catchupUnsubscribe @@ -76,6 +79,8 @@ , eventResolved , resolvedEventOriginal , resolvedEventOriginalStreamId + , positionStart + , positionEnd -- * Misc , ExpectedVersion(..) -- * Re-export @@ -260,8 +265,7 @@ -------------------------------------------------------------------------------- -- | Reads events from the $all stream forward. readAllEventsForward :: Connection - -> Int64 -- ^ Commit position - -> Int64 -- ^ Prepare position + -> Position -> Int32 -- ^ Batch size -> Bool -- ^ Resolve Link Tos -> IO (Async AllEventsSlice) @@ -271,8 +275,7 @@ -------------------------------------------------------------------------------- -- | Reads events from the $all stream backward readAllEventsBackward :: Connection - -> Int64 -- ^ Commit position - -> Int64 -- ^ Prepare position + -> Position -> Int32 -- ^ Batch size -> Bool -- ^ Resolve Link Tos -> IO (Async AllEventsSlice) @@ -282,12 +285,11 @@ -------------------------------------------------------------------------------- readAllEventsCommon :: Connection -> ReadDirection - -> Int64 - -> Int64 + -> Position -> Int32 -> Bool -> IO (Async AllEventsSlice) -readAllEventsCommon Connection{..} dir c_pos p_pos max_c res_link_tos = do +readAllEventsCommon Connection{..} dir pos max_c res_link_tos = do (as, mvar) <- createAsync let op = readAllEventsOperation conSettings @@ -300,6 +302,8 @@ processorNewOperation conProcessor op return as + where + Position c_pos p_pos = pos -------------------------------------------------------------------------------- -- | Subcribes to given stream. @@ -316,6 +320,19 @@ async $ readMVar tmp -------------------------------------------------------------------------------- +-- | Subcribes to $all stream. +subscribeToAll :: Connection + -> Bool -- ^ Resolve Link Tos + -> IO (Async Subscription) +subscribeToAll Connection{..} res_lnk_tos = do + tmp <- newEmptyMVar + processorNewSubcription conProcessor + (putMVar tmp) + "" + res_lnk_tos + async $ readMVar tmp + +-------------------------------------------------------------------------------- -- | Subscribes to given stream. If last checkpoint is defined, this will -- 'readStreamEventsForward' from that event number, otherwise from the -- beginning. Once last stream event reached up, a subscription request will @@ -333,6 +350,21 @@ readStreamEventsForward conn stream_id cur_num batch_size res_lnk_tos get_sub = subscribe conn stream_id res_lnk_tos + +-------------------------------------------------------------------------------- +-- | Same as 'subscribeFrom' but applied to $all stream. +subscribeToAllFrom :: Connection + -> Bool -- ^ Resolve Link Tos + -> Maybe Position -- ^ Last checkpoint + -> Maybe Int32 -- ^ Batch size + -> IO Catchup +subscribeToAllFrom conn res_lnk_tos last_chk_pt batch_m = do + catchupAllStart evts_fwd get_sub last_chk_pt batch_m + where + evts_fwd pos batch_size = + readAllEventsForward conn pos batch_size res_lnk_tos + + get_sub = subscribeToAll conn res_lnk_tos -------------------------------------------------------------------------------- createAsync :: IO (Async a, MVar (OperationExceptional a)) diff --git a/Database/EventStore/Catchup.hs b/Database/EventStore/Catchup.hs --- a/Database/EventStore/Catchup.hs +++ b/Database/EventStore/Catchup.hs @@ -1,4 +1,5 @@ {-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE OverloadedStrings #-} -------------------------------------------------------------------------------- -- | -- Module : Database.EventStore.Catchup @@ -15,6 +16,7 @@ , CatchupError(..) , catchupAwait , catchupStart + , catchupAllStart , catchupStream , catchupUnsubscribe ) where @@ -35,14 +37,14 @@ -------------------------------------------------------------------------------- import Database.EventStore.Internal.Manager.Subscription import Database.EventStore.Internal.Operation.ReadStreamEventsOperation +import Database.EventStore.Internal.Operation.ReadAllEventsOperation import Database.EventStore.Internal.Types -------------------------------------------------------------------------------- -- | Errors that could arise during a catch-up subscription. 'Text' value -- represents the stream name. data CatchupError - = CatchupStreamDisappeared Text - | CatchupStreamDeleted Text + = CatchupStreamDeleted Text | CatchupUnexpectedStreamStatus Text ReadStreamResult | CatchupSubscriptionDropReason Text DropReason deriving (Show, Typeable) @@ -71,10 +73,6 @@ defaultBatchSize = 500 -------------------------------------------------------------------------------- -secs :: Int -secs = 1000000 - --------------------------------------------------------------------------------- catchupStart :: (Int32 -> Int32 -> IO (Async StreamEventsSlice)) -> IO (Async Subscription) -> Text @@ -93,22 +91,12 @@ stream_id nxt_read_evt batch_size - last_m - case res_m of - Just e -> throwIO e - _ -> return () + + maybe (return ()) throwIO res_m action <- get_sub sub <- wait action putMVar var sub - forever $ do - evt_e <- subAwait sub - case evt_e of - Right evt -> writeChan chan (Right evt) - Left r -> do - let e = CatchupSubscriptionDropReason stream_id r - - writeChan chan (Left e) - throwIO e + keepAwaitingSubEvent stream_id chan sub let catchup = Catchup { catchupStream = stream_id @@ -122,44 +110,67 @@ return catchup -------------------------------------------------------------------------------- +catchupAllStart :: ( Position -> Int32 -> IO (Async AllEventsSlice)) + -> IO (Async Subscription) + -> Maybe Position + -> Maybe Int32 + -> IO Catchup +catchupAllStart evt_fwd get_sub last_chk_pt_m batch_size_m = do + chan <- newChan + var <- newEmptyMVar + let batch_size = fromMaybe defaultBatchSize batch_size_m + start_pos = fromMaybe positionStart last_chk_pt_m + + as <- async $ do + res_m <- readAllTill evt_fwd + (writeChan chan) + start_pos + batch_size + + maybe (return ()) throwIO res_m + action <- get_sub + sub <- wait action + putMVar var sub + keepAwaitingSubEvent "" chan sub + + let catchup = Catchup + { catchupStream = "" + , catchupChan = chan + , catchupUnsubscribe = do + cancel as + sub_m <- tryTakeMVar var + traverse_ subUnsubscribe sub_m + } + + return catchup + +-------------------------------------------------------------------------------- readEventsTill :: (Int32 -> Int32 -> IO (Async StreamEventsSlice)) -> (Either CatchupError ResolvedEvent -> IO ()) -> Text -> Int32 -> Int32 - -> Maybe Int32 -> IO (Maybe CatchupError) -readEventsTill evts_fwd proc_evt stream_id start batch_size last_m = +readEventsTill evts_fwd proc_evt stream_id start batch_size = loop False start where loop done cur_evt_num - | done = threadDelay (1 * secs) >> return Nothing + | done = return Nothing | otherwise = do action <- evts_fwd cur_evt_num batch_size slice <- wait action case streamEventsSliceResult slice of RS_SUCCESS -> do let nxt = streamEventsSliceNext slice - eos = streamEventsSliceIsEOS slice - n_done = maybe eos (< nxt) last_m + n_done = streamEventsSliceIsEOS slice evts = streamEventsSliceEvents slice traverse_ (proc_evt . Right) evts loop n_done nxt - RS_NO_STREAM -> - if desappeared - then reportError desappearError - else loop True cur_evt_num + RS_NO_STREAM -> loop True cur_evt_num RS_STREAM_DELETED -> reportError deletedError s -> reportError $ unexpectedError s - desappeared = - case last_m of - Just i -> i /= (-1) - _ -> False - - desappearError = CatchupStreamDisappeared stream_id - deletedError = CatchupStreamDeleted stream_id unexpectedError s = CatchupUnexpectedStreamStatus stream_id s @@ -167,3 +178,39 @@ reportError e = do proc_evt $ Left e return $ Just e + +-------------------------------------------------------------------------------- +readAllTill :: (Position -> Int32 -> IO (Async AllEventsSlice)) + -> (Either CatchupError ResolvedEvent -> IO ()) + -> Position + -> Int32 + -> IO (Maybe CatchupError) +readAllTill evts_fwd proc_evt start batch_size = + loop False start + where + loop done pos + | done = return Nothing + | otherwise = do + action <- evts_fwd pos batch_size + slice <- wait action + let evts = allEventsSliceEvents slice + nxt = allEventsSliceNext slice + n_done = allEventsSliceIsEOS slice + + traverse_ (proc_evt . Right) evts + loop n_done nxt + +-------------------------------------------------------------------------------- +keepAwaitingSubEvent :: Text + -> Chan (Either CatchupError ResolvedEvent) + -> Subscription + -> IO () +keepAwaitingSubEvent stream_id chan sub = forever $ do + evt_e <- subAwait sub + case evt_e of + Right evt -> writeChan chan (Right evt) + Left r -> do + let e = CatchupSubscriptionDropReason stream_id r + + writeChan chan (Left e) + throwIO e diff --git a/Database/EventStore/Internal/Types.hs b/Database/EventStore/Internal/Types.hs --- a/Database/EventStore/Internal/Types.hs +++ b/Database/EventStore/Internal/Types.hs @@ -350,6 +350,16 @@ deriving Show -------------------------------------------------------------------------------- +-- | Representing the start of the transaction file. +positionStart :: Position +positionStart = Position 0 0 + +-------------------------------------------------------------------------------- +-- | Representing the end of the transaction file. +positionEnd :: Position +positionEnd = Position (-1) (-1) + +-------------------------------------------------------------------------------- -- | Returned after writing to a stream. data WriteResult = WriteResult diff --git a/eventstore.cabal b/eventstore.cabal --- a/eventstore.cabal +++ b/eventstore.cabal @@ -10,13 +10,13 @@ -- PVP summary: +-+------- breaking API changes -- | | +----- non-breaking API additions -- | | | +--- code changes with no API change -version: 0.4.0.0 +version: 0.5.0.0 -- A short (one-line) description of the package. -synopsis: EventStore Haskell TCP Client +synopsis: EventStore TCP Client -- A longer description of the package. -description: EventStore Haskell TCP Client +description: EventStore TCP Client -- The license under which the package is released. license: BSD3