diff --git a/Database/EventStore.hs b/Database/EventStore.hs --- a/Database/EventStore.hs +++ b/Database/EventStore.hs @@ -20,6 +20,8 @@ , Port , Settings(..) , Subscription(..) + , Catchup(..) + , CatchupError(..) , credentials -- * Result , AllEventsSlice(..) @@ -56,6 +58,7 @@ , shutdown , transactionStart , subscribe + , subscribeFrom -- * Transaction , Transaction , transactionCommit @@ -75,6 +78,7 @@ import Data.Text -------------------------------------------------------------------------------- +import Database.EventStore.Catchup import Database.EventStore.Internal.Processor import Database.EventStore.Internal.Types import Database.EventStore.Internal.Operation.DeleteStreamOperation @@ -285,6 +289,21 @@ stream_id res_lnk_tos async $ readMVar tmp + +-------------------------------------------------------------------------------- +subscribeFrom :: Connection + -> Text + -> Bool + -> Maybe Int32 + -> Maybe Int32 + -> IO Catchup +subscribeFrom conn stream_id res_lnk_tos last_chk_pt batch_m = do + catchStart evts_fwd get_sub stream_id batch_m last_chk_pt + where + evts_fwd cur_num batch_size = + readStreamEventsForward conn stream_id cur_num batch_size res_lnk_tos + + get_sub = subscribe conn stream_id res_lnk_tos -------------------------------------------------------------------------------- createAsync :: IO (Async a, MVar (OperationExceptional a)) diff --git a/Database/EventStore/Catchup.hs b/Database/EventStore/Catchup.hs new file mode 100644 --- /dev/null +++ b/Database/EventStore/Catchup.hs @@ -0,0 +1,156 @@ +{-# LANGUAGE DeriveDataTypeable #-} +-------------------------------------------------------------------------------- +-- | +-- Module : Database.EventStore.Catchup +-- Copyright : (C) 2014 Yorick Laupa +-- License : (see the file LICENSE) +-- +-- Maintainer : Yorick Laupa +-- Stability : provisional +-- Portability : non-portable +-- +-------------------------------------------------------------------------------- +module Database.EventStore.Catchup + ( Catchup(..) + , CatchupError(..) + , catchStart + ) where + +-------------------------------------------------------------------------------- +import Control.Concurrent +import Control.Exception +import Control.Monad +import Data.Foldable (traverse_) +import Data.Int +import Data.Maybe +import Data.Typeable + +-------------------------------------------------------------------------------- +import Control.Concurrent.Async +import Data.Text + +-------------------------------------------------------------------------------- +import Database.EventStore.Internal.Manager.Subscription +import Database.EventStore.Internal.Operation.ReadStreamEventsOperation +import Database.EventStore.Internal.Types + +-------------------------------------------------------------------------------- +data CatchupError + = CatchupStreamDisappeared Text + | CatchupStreamDeleted Text + | CatchupUnexpectedStreamStatus Text ReadStreamResult + | CatchupSubscriptionDropReason Text DropReason + deriving (Show, Typeable) + +-------------------------------------------------------------------------------- +instance Exception CatchupError + +-------------------------------------------------------------------------------- +data Catchup + = Catchup + { catchupStream :: Text + , catchupChan :: Chan (Either CatchupError ResolvedEvent) + , catchupUnsubscribe :: IO () + } + +-------------------------------------------------------------------------------- +defaultBatchSize :: Int32 +defaultBatchSize = 500 + +-------------------------------------------------------------------------------- +secs :: Int +secs = 1000000 + +-------------------------------------------------------------------------------- +catchStart :: (Int32 -> Int32 -> IO (Async StreamEventsSlice)) + -> IO (Async Subscription) + -> Text + -> Maybe Int32 + -> Maybe Int32 + -> IO Catchup +catchStart evt_fwd get_sub stream_id batch_size_m last_m = do + chan <- newChan + var <- newEmptyMVar + let batch_size = fromMaybe defaultBatchSize batch_size_m + nxt_read_evt = fromMaybe 0 last_m + + as <- async $ do + res_m <- readEventsTill evt_fwd + (writeChan chan) + stream_id + nxt_read_evt + batch_size + last_m + case res_m of + Just e -> throwIO e + _ -> return () + action <- get_sub + sub <- wait action + putMVar var sub + forever $ do + evt_e <- readChan $ subChan 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 + + let catchup = Catchup + { catchupStream = stream_id + , 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 = + loop False start + where + loop done cur_evt_num + | done = threadDelay (1 * secs) >> 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 + 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_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 + + reportError e = do + proc_evt $ Left e + return $ Just e diff --git a/Database/EventStore/Internal/Manager/Subscription.hs b/Database/EventStore/Internal/Manager/Subscription.hs --- a/Database/EventStore/Internal/Manager/Subscription.hs +++ b/Database/EventStore/Internal/Manager/Subscription.hs @@ -23,7 +23,6 @@ -------------------------------------------------------------------------------- import Control.Concurrent -import Control.Monad import Control.Monad.Fix import Data.ByteString (ByteString) import Data.Int diff --git a/Database/EventStore/Internal/Operation/DeleteStreamOperation.hs b/Database/EventStore/Internal/Operation/DeleteStreamOperation.hs --- a/Database/EventStore/Internal/Operation/DeleteStreamOperation.hs +++ b/Database/EventStore/Internal/Operation/DeleteStreamOperation.hs @@ -31,10 +31,10 @@ -------------------------------------------------------------------------------- data DeleteStream = DeleteStream - { deleteStreamId :: Required 1 (Value Text) - , deleteStreamExpectedVersion :: Required 2 (Value Int32) - , deleteStreamRequireMaster :: Required 3 (Value Bool) - , deleteStreamHardDelete :: Optional 4 (Value Bool) + { _deleteStreamId :: Required 1 (Value Text) + , _deleteStreamExpectedVersion :: Required 2 (Value Int32) + , _deleteStreamRequireMaster :: Required 3 (Value Bool) + , _deleteStreamHardDelete :: Optional 4 (Value Bool) } deriving (Generic, Show) @@ -49,19 +49,19 @@ -> DeleteStream newDeleteStream stream_id exp_ver req_master hard_delete = DeleteStream - { deleteStreamId = putField stream_id - , deleteStreamExpectedVersion = putField exp_ver - , deleteStreamRequireMaster = putField req_master - , deleteStreamHardDelete = putField hard_delete + { _deleteStreamId = putField stream_id + , _deleteStreamExpectedVersion = putField exp_ver + , _deleteStreamRequireMaster = putField req_master + , _deleteStreamHardDelete = putField hard_delete } -------------------------------------------------------------------------------- data DeleteStreamCompleted = DeleteStreamCompleted - { deleteCompletedResult :: Required 1 (Enumeration OpResult) - , deleteCompletedMessage :: Optional 2 (Value Text) - , deleteCompletedPreparePosition :: Optional 3 (Value Int64) - , deleteCompletedCommitPosition :: Optional 4 (Value Int64) + { _deleteCompletedResult :: Required 1 (Enumeration OpResult) + , _deleteCompletedMessage :: Optional 2 (Value Text) + , _deleteCompletedPreparePosition :: Optional 3 (Value Int64) + , _deleteCompletedCommitPosition :: Optional 4 (Value Int64) } deriving (Generic, Show) @@ -100,7 +100,7 @@ -> ExpectedVersion -> DeleteStreamCompleted -> IO Decision -inspect mvar stream exp_ver dsc = go (getField $ deleteCompletedResult dsc) +inspect mvar stream exp_ver dsc = go (getField $ _deleteCompletedResult dsc) where go OP_SUCCESS = succeed mvar dsc go OP_PREPARE_TIMEOUT = return Retry @@ -121,8 +121,8 @@ putMVar mvar (Right wr) return EndOperation where - com_pos = getField $ deleteCompletedCommitPosition wec - pre_pos = getField $ deleteCompletedPreparePosition wec + com_pos = getField $ _deleteCompletedCommitPosition wec + pre_pos = getField $ _deleteCompletedPreparePosition wec com_pos_int = fromMaybe (-1) com_pos pre_pos_int = fromMaybe (-1) pre_pos pos = Position com_pos_int pre_pos_int diff --git a/Database/EventStore/Internal/Operation/ReadStreamEventsOperation.hs b/Database/EventStore/Internal/Operation/ReadStreamEventsOperation.hs --- a/Database/EventStore/Internal/Operation/ReadStreamEventsOperation.hs +++ b/Database/EventStore/Internal/Operation/ReadStreamEventsOperation.hs @@ -33,11 +33,11 @@ -------------------------------------------------------------------------------- data ReadStreamEvents = ReadStreamEvents - { readStreamId :: Required 1 (Value Text) - , readStreamEventNumber :: Required 2 (Value Int32) - , readStreamMaxCount :: Required 3 (Value Int32) - , readStreamResolveLinkTos :: Required 4 (Value Bool) - , readStreamRequireMaster :: Required 5 (Value Bool) + { _readStreamId :: Required 1 (Value Text) + , _readStreamEventNumber :: Required 2 (Value Int32) + , _readStreamMaxCount :: Required 3 (Value Int32) + , _readStreamResolveLinkTos :: Required 4 (Value Bool) + , _readStreamRequireMaster :: Required 5 (Value Bool) } deriving (Generic, Show) @@ -50,11 +50,11 @@ -> ReadStreamEvents newReadStreamEvents stream_id evt_num max_c res_link_tos req_master = ReadStreamEvents - { readStreamId = putField stream_id - , readStreamEventNumber = putField evt_num - , readStreamMaxCount = putField max_c - , readStreamResolveLinkTos = putField res_link_tos - , readStreamRequireMaster = putField req_master + { _readStreamId = putField stream_id + , _readStreamEventNumber = putField evt_num + , _readStreamMaxCount = putField max_c + , _readStreamResolveLinkTos = putField res_link_tos + , _readStreamRequireMaster = putField req_master } -------------------------------------------------------------------------------- @@ -73,13 +73,13 @@ -------------------------------------------------------------------------------- data ReadStreamEventsCompleted = ReadStreamEventsCompleted - { readSECEvents :: Repeated 1 (Message ResolvedIndexedEvent) - , readSECResult :: Required 2 (Enumeration ReadStreamResult) - , readSECNextNumber :: Required 3 (Value Int32) - , readSECLastNumber :: Required 4 (Value Int32) - , readSECEndOfStream :: Required 5 (Value Bool) - , readSECLastCommitPosition :: Required 6 (Value Int64) - , readSECError :: Optional 7 (Value Text) + { _readSECEvents :: Repeated 1 (Message ResolvedIndexedEvent) + , _readSECResult :: Required 2 (Enumeration ReadStreamResult) + , _readSECNextNumber :: Required 3 (Value Int32) + , _readSECLastNumber :: Required 4 (Value Int32) + , _readSECEndOfStream :: Required 5 (Value Bool) + , _readSECLastCommitPosition :: Required 6 (Value Int64) + , _readSECError :: Optional 7 (Value Text) } deriving (Generic, Show) @@ -108,15 +108,15 @@ -> StreamEventsSlice newStreamEventsSlice stream_id start dir reco = ses where - evts = getField $ readSECEvents reco + evts = getField $ _readSECEvents reco ses = StreamEventsSlice - { streamEventsSliceResult = getField $ readSECResult reco + { streamEventsSliceResult = getField $ _readSECResult reco , streamEventsSliceStreamId = stream_id , streamEventsSliceStart = start - , streamEventsSliceNext = getField $ readSECNextNumber reco - , streamEventsSliceLast = getField $ readSECLastNumber reco - , streamEventsSliceIsEOS = getField $ readSECEndOfStream reco + , streamEventsSliceNext = getField $ _readSECNextNumber reco + , streamEventsSliceLast = getField $ _readSECLastNumber reco + , streamEventsSliceIsEOS = getField $ _readSECEndOfStream reco , streamEventsSliceEvents = fmap newResolvedEvent evts , streamEventsSliceDirection = dir } @@ -164,9 +164,9 @@ -> Int32 -> ReadStreamEventsCompleted -> IO Decision -inspect mvar dir stream_id start rsec = go (getField $ readSECResult rsec) +inspect mvar dir stream_id start rsec = go (getField $ _readSECResult rsec) where - may_err = getField $ readSECError rsec + may_err = getField $ _readSECError rsec go RS_ERROR = failed mvar (ServerError may_err) go RS_ACCESS_DENIED = failed mvar (AccessDenied stream_id) diff --git a/Database/EventStore/Internal/Operation/WriteEventsOperation.hs b/Database/EventStore/Internal/Operation/WriteEventsOperation.hs --- a/Database/EventStore/Internal/Operation/WriteEventsOperation.hs +++ b/Database/EventStore/Internal/Operation/WriteEventsOperation.hs @@ -32,10 +32,10 @@ -------------------------------------------------------------------------------- data WriteEvents = WriteEvents - { writeStreamId :: Required 1 (Value Text) - , writeExpectedVersion :: Required 2 (Value Int32) - , writeEvents :: Repeated 3 (Message NewEvent) - , writeRequireMaster :: Required 4 (Value Bool) + { _writeStreamId :: Required 1 (Value Text) + , _writeExpectedVersion :: Required 2 (Value Int32) + , _writeEvents :: Repeated 3 (Message NewEvent) + , _writeRequireMaster :: Required 4 (Value Bool) } deriving (Generic, Show) @@ -50,21 +50,21 @@ -> WriteEvents newWriteEvents stream_id exp_ver evts req_master = WriteEvents - { writeStreamId = putField stream_id - , writeExpectedVersion = putField exp_ver - , writeEvents = putField evts - , writeRequireMaster = putField req_master + { _writeStreamId = putField stream_id + , _writeExpectedVersion = putField exp_ver + , _writeEvents = putField evts + , _writeRequireMaster = putField req_master } -------------------------------------------------------------------------------- data WriteEventsCompleted = WriteEventsCompleted - { writeCompletedResult :: Required 1 (Enumeration OpResult) - , writeCompletedMessage :: Optional 2 (Value Text) - , writeCompletedFirstNumber :: Required 3 (Value Int32) - , writeCompletedLastNumber :: Required 4 (Value Int32) - , writeCompletedPreparePosition :: Optional 5 (Value Int64) - , writeCompletedCommitPosition :: Optional 6 (Value Int64) + { _writeCompletedResult :: Required 1 (Enumeration OpResult) + , _writeCompletedMessage :: Optional 2 (Value Text) + , _writeCompletedFirstNumber :: Required 3 (Value Int32) + , _writeCompletedLastNumber :: Required 4 (Value Int32) + , _writeCompletedPreparePosition :: Optional 5 (Value Int64) + , _writeCompletedCommitPosition :: Optional 6 (Value Int64) } deriving (Generic, Show) @@ -105,7 +105,7 @@ -> ExpectedVersion -> WriteEventsCompleted -> IO Decision -inspect mvar stream exp_ver wec = go (getField $ writeCompletedResult wec) +inspect mvar stream exp_ver wec = go (getField $ _writeCompletedResult wec) where go OP_SUCCESS = succeed mvar wec go OP_PREPARE_TIMEOUT = return Retry @@ -126,9 +126,9 @@ putMVar mvar (Right wr) return EndOperation where - last_evt_num = getField $ writeCompletedLastNumber wec - com_pos = getField $ writeCompletedCommitPosition wec - pre_pos = getField $ writeCompletedPreparePosition wec + last_evt_num = getField $ _writeCompletedLastNumber wec + com_pos = getField $ _writeCompletedCommitPosition wec + pre_pos = getField $ _writeCompletedPreparePosition wec com_pos_int = fromMaybe (-1) com_pos pre_pos_int = fromMaybe (-1) pre_pos pos = Position com_pos_int pre_pos_int diff --git a/Database/EventStore/Internal/Processor.hs b/Database/EventStore/Internal/Processor.hs --- a/Database/EventStore/Internal/Processor.hs +++ b/Database/EventStore/Internal/Processor.hs @@ -32,7 +32,6 @@ -------------------------------------------------------------------------------- import Control.Concurrent.Async -import Data.Text (Text) import Data.UUID import FRP.Sodium import Network diff --git a/Database/EventStore/Internal/Reader.hs b/Database/EventStore/Internal/Reader.hs --- a/Database/EventStore/Internal/Reader.hs +++ b/Database/EventStore/Internal/Reader.hs @@ -47,10 +47,6 @@ getLengthPrefix = fmap fromIntegral getWord32le -------------------------------------------------------------------------------- -authOffset :: Int -authOffset = 18 - --------------------------------------------------------------------------------- getPackage :: Get Package getPackage = do cmd <- getWord8 diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -11,11 +11,12 @@ 6. ReadAllEvents (Forward and Backward) 7. Volatile subscriptions 8. Authentication + 9. Catchup subscriptions TODO ==== - 1. All kind of Subscriptions + 1. Persistent Subscriptions 2. SSL Requirements diff --git a/eventstore.cabal b/eventstore.cabal --- a/eventstore.cabal +++ b/eventstore.cabal @@ -10,7 +10,7 @@ -- PVP summary: +-+------- breaking API changes -- | | +----- non-breaking API additions -- | | | +--- code changes with no API change -version: 0.3.0.0 +version: 0.3.1.0 -- A short (one-line) description of the package. synopsis: EventStore Haskell TCP Client @@ -34,6 +34,7 @@ -- A copyright notice. -- copyright: +homepage: http://github.com/YoEight/eventstore category: Database build-type: Simple @@ -54,7 +55,8 @@ exposed-modules: Database.EventStore -- Modules included in this library but not exported. - other-modules: Database.EventStore.Internal.Packages + other-modules: Database.EventStore.Catchup + Database.EventStore.Internal.Packages Database.EventStore.Internal.Processor Database.EventStore.Internal.Reader Database.EventStore.Internal.Types @@ -91,4 +93,6 @@ -- hs-source-dirs: -- Base language which the package is written in. + ghc-options: -Wall + default-language: Haskell2010