eventstore 0.4.0.0 → 0.5.0.0

raw patch · 4 files changed

+135/−46 lines, 4 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• Database/EventStore.hs +40/−8
• Database/EventStore/Catchup.hs +82/−35
• Database/EventStore/Internal/Types.hs +10/−0
• eventstore.cabal +3/−3

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

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

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

@@ -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 <http://geteventstore.com>  -- The license under which the package is released. license:             BSD3