eventstore 0.3.0.0 → 0.3.1.0

raw patch · 10 files changed

+240/−66 lines, 10 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• Database/EventStore.hs +19/−0
• Database/EventStore/Catchup.hs +156/−0
• Database/EventStore/Internal/Manager/Subscription.hs +0/−1
• Database/EventStore/Internal/Operation/DeleteStreamOperation.hs +15/−15
• Database/EventStore/Internal/Operation/ReadStreamEventsOperation.hs +24/−24
• Database/EventStore/Internal/Operation/WriteEventsOperation.hs +18/−18
• Database/EventStore/Internal/Processor.hs +0/−1
• Database/EventStore/Internal/Reader.hs +0/−4
• README.md +2/−1
• eventstore.cabal +6/−2

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

@@ -0,0 +1,156 @@+{-# LANGUAGE DeriveDataTypeable #-}+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Catchup+-- Copyright : (C) 2014 Yorick Laupa+-- License : (see the file LICENSE)+--+-- Maintainer : Yorick Laupa <yo.eight@gmail.com>+-- 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

@@ -23,7 +23,6 @@  -------------------------------------------------------------------------------- import           Control.Concurrent-import           Control.Monad import           Control.Monad.Fix import           Data.ByteString (ByteString) import           Data.Int

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

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

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

@@ -32,7 +32,6 @@  -------------------------------------------------------------------------------- import Control.Concurrent.Async-import Data.Text (Text) import Data.UUID import FRP.Sodium import Network

@@ -47,10 +47,6 @@ getLengthPrefix = fmap fromIntegral getWord32le  ---------------------------------------------------------------------------------authOffset :: Int-authOffset = 18---------------------------------------------------------------------------------- getPackage :: Get Package getPackage = do     cmd  <- getWord8

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

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