eventstore 0.8.0.0 → 0.9.0.0

raw patch · 49 files changed

+5196/−3536 lines, 49 filesdep +eventstoredep +tastydep +tasty-hunitdep −sodiumdep ~aesondep ~basedep ~textPVP ok

version bump matches the API change (PVP)

Dependencies added: eventstore, tasty, tasty-hunit

Dependencies removed: sodium

Dependency ranges changed: aeson, base, text, time

API changes (from Hackage documentation)

Files

• CHANGELOG.markdown +9/−0
• Database/EventStore.hs +434/−197
• Database/EventStore/Catchup.hs +0/−186
• Database/EventStore/Internal/Connection.hs +125/−30
• Database/EventStore/Internal/Execution/Production.hs +636/−0
• Database/EventStore/Internal/Generator.hs +42/−0
• Database/EventStore/Internal/Manager/Operation.hs +0/−197
• Database/EventStore/Internal/Manager/Operation/Model.hs +178/−0
• Database/EventStore/Internal/Manager/Subscription.hs +220/−1112
• Database/EventStore/Internal/Manager/Subscription/Driver.hs +481/−0
• Database/EventStore/Internal/Manager/Subscription/Message.hs +395/−0
• Database/EventStore/Internal/Manager/Subscription/Model.hs +318/−0
• Database/EventStore/Internal/Manager/Subscription/Packages.hs +137/−0
• Database/EventStore/Internal/Operation.hs +193/−0
• Database/EventStore/Internal/Operation/Catchup.hs +102/−0
• Database/EventStore/Internal/Operation/DeleteStream.hs +63/−0
• Database/EventStore/Internal/Operation/DeleteStream/Message.hs +64/−0
• Database/EventStore/Internal/Operation/DeleteStreamOperation.hs +0/−137
• Database/EventStore/Internal/Operation/Read/Common.hs +145/−0
• Database/EventStore/Internal/Operation/ReadAllEvents.hs +62/−0
• Database/EventStore/Internal/Operation/ReadAllEvents/Message.hs +84/−0
• Database/EventStore/Internal/Operation/ReadAllEventsOperation.hs +0/−200
• Database/EventStore/Internal/Operation/ReadEvent.hs +67/−0
• Database/EventStore/Internal/Operation/ReadEvent/Message.hs +75/−0
• Database/EventStore/Internal/Operation/ReadEventOperation.hs +0/−170
• Database/EventStore/Internal/Operation/ReadStreamEvents.hs +66/−0
• Database/EventStore/Internal/Operation/ReadStreamEvents/Message.hs +80/−0
• Database/EventStore/Internal/Operation/ReadStreamEventsOperation.hs +0/−205
• Database/EventStore/Internal/Operation/StreamMetadata.hs +89/−0
• Database/EventStore/Internal/Operation/Transaction.hs +109/−0
• Database/EventStore/Internal/Operation/Transaction/Message.hs +136/−0
• Database/EventStore/Internal/Operation/TransactionStartOperation.hs +0/−283
• Database/EventStore/Internal/Operation/Write/Common.hs +49/−0
• Database/EventStore/Internal/Operation/WriteEvents.hs +61/−0
• Database/EventStore/Internal/Operation/WriteEvents/Message.hs +71/−0
• Database/EventStore/Internal/Operation/WriteEventsOperation.hs +0/−161
• Database/EventStore/Internal/Operations.hs +32/−0
• Database/EventStore/Internal/Packages.hs +5/−16
• Database/EventStore/Internal/Processor.hs +256/−216
• Database/EventStore/Internal/Reader.hs +0/−95
• Database/EventStore/Internal/Stream.hs +31/−0
• Database/EventStore/Internal/TimeSpan.hs +33/−3
• Database/EventStore/Internal/Types.hs +139/−222
• Database/EventStore/Internal/Util/Sodium.hs +0/−35
• Database/EventStore/Internal/Writer.hs +0/−31
• Database/EventStore/Logging.hs +10/−8
• README.md +95/−18
• eventstore.cabal +45/−14
• tests/integration.hs +59/−0

@@ -1,3 +1,12 @@+0.9.0.0+-------+* Rewrite entirely the internals.+* Implement integration tests.+* Rename every `ExpectedVersion` smart constructors.+* Improve internal and public documentation.+* Improve failure reports when the connection dropped.+* Implement more robust internal connection.+ 0.8.0.0 ------- * Implement competing consumers.

@@ -1,5 +1,7 @@-{-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE RecordWildCards   #-}+{-# LANGUAGE DataKinds          #-}+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE OverloadedStrings  #-}+{-# LANGUAGE RecordWildCards    #-} -------------------------------------------------------------------------------- -- | -- Module : Database.EventStore@@ -12,15 +14,10 @@ -- -------------------------------------------------------------------------------- module Database.EventStore-    ( -- * Event-      Event-    , EventData-    , createEvent-    , withJson-    , withJsonAndMetadata-      -- * Connection-    , Connection+    ( -- * Connection+      Connection     , ConnectionException(..)+    , ServerConnectionError(..)     , Credentials     , Settings(..)     , Retry@@ -30,6 +27,13 @@     , defaultSettings     , connect     , shutdown+    , waitTillClosed+      -- * Event+    , Event+    , EventData+    , createEvent+    , withJson+    , withJsonAndMetadata      -- * Read Operations     , StreamMetadataResult(..)     , readEvent@@ -41,8 +45,8 @@       -- * Write Operations     , StreamACL(..)     , StreamMetadata(..)-    , streamMetadataGetCustomPropertyValue-    , streamMetadataGetCustomProperty+    , getCustomPropertyValue+    , getCustomProperty     , emptyStreamACL     , emptyStreamMetadata     , deleteStream@@ -95,38 +99,43 @@     , timeSpanTotalMillis       -- * Transaction     , Transaction-    , transactionStart+    , TransactionId+    , startTransaction+    , transactionId     , transactionCommit     , transactionRollback-    , transactionSendEvents-      -- * Volatile Subscription-    , DropReason(..)-    , Identifiable+    , transactionWrite+      -- * Subscription+    , SubscriptionClosed(..)+    , SubscriptionId     , Subscription-    , NextEvent-    , Regular-    , Catchup-    , Persistent+    , S.Running(..)+    , S.SubDropReason(..)+      -- * Volatile Subscription+    , S.Regular     , subscribe     , subscribeToAll-    , subNextEvent-    , subId-    , subStreamId-    , subIsSubscribedToAll-    , subResolveLinkTos-    , subLastCommitPos-    , subLastEventNumber-    , subUnsubscribe+    , getSubId+    , getSubStream+    , isSubscribedToAll+    , unsubscribe+    , nextEvent+    , nextEventMaybe+    , getSubResolveLinkTos+    , getSubLastCommitPos+    , getSubLastEventNumber       -- * Catch-up Subscription-    , CatchupError(..)+    , S.Catchup     , subscribeFrom     , subscribeToAllFrom     , waitTillCatchup     , hasCaughtUp      -- * Persistent Subscription+    , S.Persistent     , PersistentSubscriptionSettings(..)     , SystemConsumerStrategy(..)     , NakAction(..)+    , S.PersistActionException(..)     , notifyEventsProcessed     , notifyEventsFailed     , defaultPersistentSubscriptionSettings@@ -135,31 +144,35 @@     , deletePersistentSubscription     , connectToPersistentSubscription      -- * Results-    , AllEventsSlice(..)-    , DeleteResult(..)+    , Slice(..)+    , AllSlice+    , Op.DeleteResult(..)     , WriteResult(..)     , ReadResult(..)     , RecordedEvent(..)-    , StreamEventsSlice(..)+    , Op.ReadEvent(..)+    , StreamType(..)+    , StreamSlice     , Position(..)     , ReadDirection(..)-    , ReadAllResult(..)-    , ReadEventResult(..)     , ResolvedEvent(..)-    , ReadStreamResult(..)-    , OperationException(..)-    , eventResolved+    , OperationError(..)+    , StreamName(..)+    , isEventResolvedLink     , resolvedEventOriginal+    , resolvedEventDataAsJson     , resolvedEventOriginalStreamId     , resolvedEventOriginalId+    , recordedEventDataAsJson     , positionStart     , positionEnd       -- * Misc+    , DropReason(..)     , ExpectedVersion-    , anyStream-    , noStream-    , emptyStream-    , exactStream+    , anyVersion+    , noStreamVersion+    , emptyStreamVersion+    , exactEventVersion       -- * Re-export     , module Control.Concurrent.Async     , (<>)@@ -167,29 +180,31 @@  -------------------------------------------------------------------------------- import Control.Concurrent-import Control.Concurrent.STM (atomically)+import Control.Concurrent.STM import Control.Exception-import Data.ByteString.Lazy (fromStrict)+import Control.Monad (when) import Data.Int+import Data.Maybe import Data.Monoid ((<>))+import Data.Typeable  -------------------------------------------------------------------------------- import Control.Concurrent.Async-import Data.Aeson (decode) import Data.Text hiding (group)+import Data.UUID  ---------------------------------------------------------------------------------import Database.EventStore.Catchup-import Database.EventStore.Internal.Manager.Subscription-import Database.EventStore.Internal.Operation.DeleteStreamOperation-import Database.EventStore.Internal.Operation.ReadAllEventsOperation-import Database.EventStore.Internal.Operation.ReadEventOperation-import Database.EventStore.Internal.Operation.ReadStreamEventsOperation-import Database.EventStore.Internal.Operation.TransactionStartOperation-import Database.EventStore.Internal.Operation.WriteEventsOperation-import Database.EventStore.Internal.Processor-import Database.EventStore.Internal.TimeSpan-import Database.EventStore.Internal.Types+import           Database.EventStore.Internal.Connection hiding (Connection)+import qualified Database.EventStore.Internal.Manager.Subscription as S+import           Database.EventStore.Internal.Manager.Subscription.Message+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+import           Database.EventStore.Internal.Stream+import           Database.EventStore.Internal.TimeSpan+import           Database.EventStore.Internal.Types+import           Database.EventStore.Internal.Execution.Production  -------------------------------------------------------------------------------- -- Connection@@ -197,37 +212,40 @@ -- | Represents a connection to a single EventStore node. data Connection     = Connection-      { _runCmd   :: Cmd -> IO ()+      { _prod     :: Production       , _settings :: Settings       }  -------------------------------------------------------------------------------- -- | Creates a new 'Connection' to a single node. It maintains a full duplex --   connection to the EventStore. An EventStore 'Connection' operates quite---   differently than say a SQL connection. Normally when you use a SQL+--   differently than say a SQL connection. Normally when you use an EventStore --   connection you want to keep the connection open for a much longer of time --   than when you use a SQL connection. -----   Another difference  is that with the EventStore 'Connection' all operation+--   Another difference  is that with the EventStore 'Connection' all operations --   are handled in a full async manner (even if you call the synchronous --   behaviors). Many threads can use an EvenStore 'Connection' at the same time --   or a single thread can make many asynchronous requests. To get the most---   performance out of the connection it is generally recommend to use it in+--   performance out of the connection it is generally recommended to use it in --   this way. connect :: Settings         -> String   -- ^ HostName         -> Int      -- ^ Port         -> IO Connection connect settings host port = do-    processor <- newProcessor settings-    processor (DoConnect host port)+    prod <- newExecutionModel settings host port+    return $ Connection prod settings -    return $ Connection processor settings+--------------------------------------------------------------------------------+-- | Waits the 'Connection' to be closed.+waitTillClosed :: Connection -> IO ()+waitTillClosed Connection{..} = prodWaitTillClosed _prod  -------------------------------------------------------------------------------- -- | Asynchronously closes the 'Connection'. shutdown :: Connection -> IO ()-shutdown Connection{..} = _runCmd DoShutdown+shutdown Connection{..} = shutdownExecutionModel _prod  -------------------------------------------------------------------------------- -- | Sends a single 'Event' to given stream.@@ -240,6 +258,163 @@     sendEvents mgr evt_stream exp_ver [evt]  --------------------------------------------------------------------------------+-- | Represents a subscription id.+newtype SubscriptionId = SubId UUID deriving (Eq, Ord, Show)++--------------------------------------------------------------------------------+-- | Determines whether or not any link events encontered in the stream will be+--   resolved.+getSubResolveLinkTos :: Subscription S.Regular -> Bool+getSubResolveLinkTos = S._subTos . _subInner++--------------------------------------------------------------------------------+-- | Non blocking version of `waitTillCatchup`.+hasCaughtUp :: Subscription S.Catchup -> IO Bool+hasCaughtUp sub = atomically $ _hasCaughtUp sub++--------------------------------------------------------------------------------+-- | Waits until 'CatchupSubscription' subscription catch-up its stream.+waitTillCatchup :: Subscription S.Catchup -> IO ()+waitTillCatchup sub = atomically $ do+    caughtUp <- _hasCaughtUp sub+    when (not caughtUp) retry++--------------------------------------------------------------------------------+_hasCaughtUp :: Subscription S.Catchup -> STM Bool+_hasCaughtUp Subscription{..} = do+    SubState sm _ <- readTVar _subVar+    return $ S.hasCaughtUp sm++--------------------------------------------------------------------------------+-- | Tracks a 'Subcription' lifecycle. It holds a 'Subscription' state machine+--   and `SubDropReason` if any.+data SubState a = SubState (S.Subscription a) (Maybe S.SubDropReason)++--------------------------------------------------------------------------------+-- | It's possible to subscribe to a stream and be notified when new events are+--   written to that stream. There are three types of subscription which are+--   available, all of which can be useful in different situations.+--+--     * 'S.Regular'+--+--     * 'S.Catchup'+--+--     * 'S.Persistent'+data Subscription a =+    Subscription+    { _subVar    :: TVar (SubState a)+    , _subRun    :: TMVar S.Running+    , _subStream :: Text+    , _subProd   :: Production+    , _subInner  :: a+    }++--------------------------------------------------------------------------------+-- | Gets the ID of the subscription.+getSubId :: Subscription a -> IO SubscriptionId+getSubId Subscription{..} = atomically $ do+    run <- readTMVar _subRun+    return $ SubId $ S.runningUUID run++--------------------------------------------------------------------------------+-- | Gets the subscription stream name.+getSubStream :: Subscription a -> Text+getSubStream = _subStream++--------------------------------------------------------------------------------+-- | Asynchronously unsubscribe from the the stream.+unsubscribe :: Subscription a -> IO ()+unsubscribe Subscription{..} = do+    run <- atomically $ readTMVar _subRun+    pushUnsubscribe _subProd run++--------------------------------------------------------------------------------+-- | If the subscription is on the $all stream.+isSubscribedToAll :: Subscription a -> Bool+isSubscribedToAll = (== "") . getSubStream++--------------------------------------------------------------------------------+-- | The last commit position seen on the subscription (if this a subscription+--   to $all stream).+getSubLastCommitPos :: Subscription a -> IO Int64+getSubLastCommitPos Subscription{..} = atomically $ do+    run <- readTMVar _subRun+    return $ S.runningLastCommitPosition run++--------------------------------------------------------------------------------+-- | The last event number seen on the subscription (if this is a subscription+--   to a single stream).+getSubLastEventNumber :: Subscription a -> IO (Maybe Int32)+getSubLastEventNumber Subscription{..} = atomically $ do+    run <- readTMVar _subRun+    return $ S.runningLastEventNumber run++--------------------------------------------------------------------------------+-- | Awaits for the next event.+nextEvent :: Subscription a -> IO ResolvedEvent+nextEvent sub = atomically $ do+    m <- _nextEventMaybe sub+    case m of+        Nothing -> retry+        Just e  -> return e++--------------------------------------------------------------------------------+-- | Non blocking version of 'nextEvent'.+nextEventMaybe :: Subscription a -> IO (Maybe ResolvedEvent)+nextEventMaybe = atomically . _nextEventMaybe++--------------------------------------------------------------------------------+_nextEventMaybe :: Subscription a -> STM (Maybe ResolvedEvent)+_nextEventMaybe Subscription{..} = do+    SubState sub close <- readTVar _subVar+    run                <- readTMVar _subRun+    let (res, nxt) = S.readNext sub+    case res of+        Nothing -> do+            case close of+                Nothing  -> return Nothing+                Just err -> throwSTM $ SubscriptionClosed run err+        Just e -> do+            writeTVar _subVar $ SubState nxt close+            return $ Just e++--------------------------------------------------------------------------------+-- | Acknowledges those event ids have been successfully processed.+notifyEventsProcessed :: Subscription S.Persistent -> [UUID] -> IO ()+notifyEventsProcessed Subscription{..} evts = do+    run <- atomically $ readTMVar _subRun+    pushAckPersist _subProd run evts++--------------------------------------------------------------------------------+-- | Acknowledges those event ids have failed to be processed successfully.+notifyEventsFailed :: Subscription S.Persistent+                   -> NakAction+                   -> Maybe Text+                   -> [UUID]+                   -> IO ()+notifyEventsFailed Subscription{..} act res evts = do+    run <- atomically $ readTMVar _subRun+    pushNakPersist _subProd run act res evts++--------------------------------------------------------------------------------+-- | Modifies 'SubState' internal state machine, letting any 'SubDropReason'+--   untouched.+modifySubSM :: (S.Subscription a -> S.Subscription a)+            -> SubState a+            -> SubState a+modifySubSM k (SubState sm r) = SubState (k sm) r++--------------------------------------------------------------------------------+-- | This exception is raised when the user tries to get the next event from a+--   'Subscription' that is already closed.+data SubscriptionClosed =+    SubscriptionClosed S.Running S.SubDropReason+    deriving (Show, Typeable)++--------------------------------------------------------------------------------++instance Exception SubscriptionClosed+-------------------------------------------------------------------------------- -- | Sends a list of 'Event' to given stream. sendEvents :: Connection            -> Text             -- ^ Stream name@@ -247,11 +422,9 @@            -> [Event]            -> IO (Async WriteResult) sendEvents Connection{..} evt_stream exp_ver evts = do-    (as, mvar) <- createAsync--    let op = writeEventsOperation _settings mvar evt_stream exp_ver evts--    _runCmd (NewOperation op)+    (k, as)  <- createOpAsync+    let op = Op.writeEvents _settings evt_stream exp_ver evts+    pushOperation _prod k op     return as  --------------------------------------------------------------------------------@@ -260,46 +433,92 @@              -> Text             -- ^ Stream name              -> ExpectedVersion              -> Maybe Bool       -- ^ Hard delete-             -> IO (Async DeleteResult)+             -> IO (Async Op.DeleteResult) deleteStream Connection{..} evt_stream exp_ver hard_del = do-    (as, mvar) <- createAsync+    (k, as) <- createOpAsync+    let op = Op.deleteStream _settings evt_stream exp_ver hard_del+    pushOperation _prod k op+    return as -    let op = deleteStreamOperation _settings mvar evt_stream exp_ver hard_del+--------------------------------------------------------------------------------+-- | Represents a multi-request transaction with the EventStore.+data Transaction =+    Transaction+    { _tStream  :: Text+    , _tTransId :: TransactionId+    , _tExpVer  :: ExpectedVersion+    , _tConn    :: Connection+    } -    _runCmd (NewOperation op)-    return as+--------------------------------------------------------------------------------+-- | The id of a 'Transaction'.+newtype TransactionId =+    TransactionId { _unTransId :: Int64 }+    deriving (Eq, Ord, Show)  --------------------------------------------------------------------------------+-- | Gets the id of a 'Transaction'.+transactionId :: Transaction -> TransactionId+transactionId = _tTransId++-------------------------------------------------------------------------------- -- | Starts a transaction on given stream.-transactionStart :: Connection+startTransaction :: Connection                  -> Text            -- ^ Stream name                  -> ExpectedVersion                  -> IO (Async Transaction)-transactionStart Connection{..} evt_stream exp_ver = do-    (as, mvar) <- createAsync+startTransaction conn@Connection{..} evt_stream exp_ver = do+    (k, as) <- createOpAsync+    let op = Op.transactionStart _settings evt_stream exp_ver+    pushOperation _prod k op+    let _F trans_id =+            Transaction+            { _tStream  = evt_stream+            , _tTransId = TransactionId trans_id+            , _tExpVer  = exp_ver+            , _tConn    = conn+            }+    return $ fmap _F as -    let op = transactionStartOperation _settings-                                       _runCmd-                                       mvar-                                       evt_stream-                                       exp_ver+--------------------------------------------------------------------------------+-- | Asynchronously writes to a transaction in the EventStore.+transactionWrite :: Transaction -> [Event] -> IO (Async ())+transactionWrite Transaction{..} evts = do+    (k, as) <- createOpAsync+    let Connection{..} = _tConn+        raw_id = _unTransId _tTransId+        op     = Op.transactionWrite _settings _tStream _tExpVer raw_id evts+    pushOperation _prod k op+    return as -    _runCmd (NewOperation op)+--------------------------------------------------------------------------------+-- | Asynchronously commits this transaction.+transactionCommit :: Transaction -> IO (Async WriteResult)+transactionCommit Transaction{..} = do+    (k, as) <- createOpAsync+    let Connection{..} = _tConn+        raw_id = _unTransId _tTransId+        op     = Op.transactionCommit _settings _tStream _tExpVer raw_id+    pushOperation _prod k op     return as  --------------------------------------------------------------------------------+-- | There isn't such of thing in EventStore parlance. Basically, if you want to+--   rollback, you just have to not 'transactionCommit' a 'Transaction'.+transactionRollback :: Transaction -> IO ()+transactionRollback _ = return ()++-------------------------------------------------------------------------------- -- | Reads a single event from given stream. readEvent :: Connection           -> Text       -- ^ Stream name           -> Int32      -- ^ Event number           -> Bool       -- ^ Resolve Link Tos-          -> IO (Async ReadResult)+          -> IO (Async (ReadResult 'RegularStream Op.ReadEvent)) readEvent Connection{..} stream_id evt_num res_link_tos = do-    (as, mvar) <- createAsync--    let op = readEventOperation _settings mvar stream_id evt_num res_link_tos--    _runCmd (NewOperation op)+    (k, as) <- createOpAsync+    let op = Op.readEvent _settings stream_id evt_num res_link_tos+    pushOperation _prod k op     return as  --------------------------------------------------------------------------------@@ -309,7 +528,7 @@                         -> Int32      -- ^ From event number                         -> Int32      -- ^ Batch size                         -> Bool       -- ^ Resolve Link Tos-                        -> IO (Async StreamEventsSlice)+                        -> IO (Async (ReadResult 'RegularStream StreamSlice)) readStreamEventsForward mgr =     readStreamEventsCommon mgr Forward @@ -320,7 +539,7 @@                          -> Int32      -- ^ From event number                          -> Int32      -- ^ Batch size                          -> Bool       -- ^ Resolve Link Tos-                         -> IO (Async StreamEventsSlice)+                         -> IO (Async (ReadResult 'RegularStream StreamSlice)) readStreamEventsBackward mgr =     readStreamEventsCommon mgr Backward @@ -331,19 +550,11 @@                        -> Int32                        -> Int32                        -> Bool-                       -> IO (Async StreamEventsSlice)+                       -> IO (Async (ReadResult 'RegularStream StreamSlice)) readStreamEventsCommon Connection{..} dir stream_id start cnt res_link_tos = do-    (as, mvar) <- createAsync--    let op = readStreamEventsOperation _settings-                                       dir-                                       mvar-                                       stream_id-                                       start-                                       cnt-                                       res_link_tos--    _runCmd (NewOperation op)+    (k, as) <- createOpAsync+    let op = Op.readStreamEvents _settings dir stream_id start cnt res_link_tos+    pushOperation _prod k op     return as  --------------------------------------------------------------------------------@@ -352,7 +563,7 @@                      -> Position                      -> Int32      -- ^ Batch size                      -> Bool       -- ^ Resolve Link Tos-                     -> IO (Async AllEventsSlice)+                     -> IO (Async AllSlice) readAllEventsForward mgr =     readAllEventsCommon mgr Forward @@ -362,7 +573,7 @@                       -> Position                       -> Int32      -- ^ Batch size                       -> Bool       -- ^ Resolve Link Tos-                      -> IO (Async AllEventsSlice)+                      -> IO (Async AllSlice) readAllEventsBackward mgr =     readAllEventsCommon mgr Backward @@ -372,19 +583,11 @@                     -> Position                     -> Int32                     -> Bool-                    -> IO (Async AllEventsSlice)+                    -> IO (Async AllSlice) readAllEventsCommon Connection{..} dir pos max_c res_link_tos = do-    (as, mvar) <- createAsync--    let op = readAllEventsOperation _settings-                                    dir-                                    mvar-                                    c_pos-                                    p_pos-                                    max_c-                                    res_link_tos--    _runCmd (NewOperation op)+    (k, as) <- createOpAsync+    let op = Op.readAllEvents _settings c_pos p_pos max_c res_link_tos dir+    pushOperation _prod k op     return as   where     Position c_pos p_pos = pos@@ -394,21 +597,26 @@ subscribe :: Connection           -> Text       -- ^ Stream name           -> Bool       -- ^ Resolve Link Tos-          -> IO (Async (Subscription Regular))+          -> IO (Subscription S.Regular) subscribe Connection{..} stream_id res_lnk_tos = do-    tmp <- newEmptyMVar-    _runCmd (NewSub stream_id res_lnk_tos (putMVar tmp))-    async $ readMVar tmp+    mvar <- newEmptyTMVarIO+    var  <- newTVarIO $ SubState S.regularSubscription Nothing+    let mk r = putTMVar mvar r+        recv = readTVar var+        send = writeTVar var+        dropped r = do+            SubState sm _ <- readTVar var+            writeTVar var $ SubState sm (Just r)+        cb = createSubAsync mk recv send dropped+    pushConnectStream _prod cb stream_id res_lnk_tos+    return $ Subscription var mvar stream_id _prod (S.Regular res_lnk_tos)  -------------------------------------------------------------------------------- -- | Subcribes to $all stream. subscribeToAll :: Connection                -> Bool       -- ^ Resolve Link Tos-               -> IO (Async (Subscription Regular))-subscribeToAll Connection{..} res_lnk_tos = do-    tmp <- newEmptyMVar-    _runCmd (NewSub "" res_lnk_tos (putMVar tmp))-    async $ readMVar tmp+               -> IO (Subscription S.Regular)+subscribeToAll conn res_lnk_tos = subscribe conn "" res_lnk_tos  -------------------------------------------------------------------------------- -- | Subscribes to given stream. If last checkpoint is defined, this will@@ -420,14 +628,11 @@               -> Bool        -- ^ Resolve Link Tos               -> Maybe Int32 -- ^ Last checkpoint               -> Maybe Int32 -- ^ Batch size-              -> IO (Subscription Catchup)-subscribeFrom conn stream_id res_lnk_tos last_chk_pt batch_m = do-    catchupStart evts_fwd get_sub stream_id batch_m last_chk_pt+              -> IO (Subscription S.Catchup)+subscribeFrom conn stream_id res_lnk_tos last_chk_pt batch_m =+    subscribeFromCommon conn stream_id res_lnk_tos batch_m tpe   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+    tpe = Op.RegularCatchup stream_id (fromMaybe 0 last_chk_pt)  -------------------------------------------------------------------------------- -- | Same as 'subscribeFrom' but applied to $all stream.@@ -435,15 +640,43 @@                    -> Bool           -- ^ Resolve Link Tos                    -> Maybe Position -- ^ Last checkpoint                    -> Maybe Int32    -- ^ Batch size-                   -> IO (Subscription Catchup)-subscribeToAllFrom conn res_lnk_tos last_chk_pt batch_m = do-    catchupAllStart evts_fwd get_sub last_chk_pt batch_m+                   -> IO (Subscription S.Catchup)+subscribeToAllFrom conn res_lnk_tos last_chk_pt batch_m =+    subscribeFromCommon conn "" res_lnk_tos batch_m tpe   where-    evts_fwd pos batch_size =-        readAllEventsForward conn pos batch_size res_lnk_tos+    Position c_pos p_pos = fromMaybe positionStart last_chk_pt+    tpe = Op.AllCatchup c_pos p_pos -    get_sub = subscribeToAll conn res_lnk_tos+--------------------------------------------------------------------------------+subscribeFromCommon :: Connection+                    -> Text+                    -> Bool+                    -> Maybe Int32+                    -> Op.CatchupState+                    -> IO (Subscription S.Catchup)+subscribeFromCommon Connection{..} stream_id res_lnk_tos batch_m tpe = do+    mvar <- newEmptyTMVarIO+    var  <- newTVarIO $ SubState S.catchupSubscription Nothing+    let readFrom res =+            case res of+                Left _ -> return ()+                Right (xs, eos, chk) -> atomically $ do+                    s <- readTVar var+                    let nxt_s = modifySubSM (S.batchRead xs eos chk) s+                    writeTVar var nxt_s+        mk   = putTMVar mvar+        rcv  = readTVar var+        send = writeTVar var+        dropped r = do+            SubState sm _ <- readTVar var+            writeTVar var $ SubState sm (Just r)+        op  = Op.catchup _settings tpe res_lnk_tos batch_m+        cb  = createSubAsync mk rcv send dropped +    pushOperation _prod readFrom op+    pushConnectStream _prod cb stream_id res_lnk_tos+    return $ Subscription var mvar stream_id _prod S.Catchup+ -------------------------------------------------------------------------------- -- | Asynchronously sets the metadata for a stream. setStreamMetadata :: Connection@@ -451,46 +684,20 @@                   -> ExpectedVersion                   -> StreamMetadata                   -> IO (Async WriteResult)-setStreamMetadata conn evt_stream exp_ver metadata =-    let dat = withJson $ streamMetadataJSON metadata-        evt = createEvent "$metadata" Nothing dat in-    sendEvent conn (metaStreamOf evt_stream) exp_ver evt+setStreamMetadata Connection{..} evt_stream exp_ver metadata = do+    (k, as) <- createOpAsync+    let op = Op.setMetaStream _settings evt_stream exp_ver metadata+    pushOperation _prod k op+    return as  -------------------------------------------------------------------------------- -- | Asynchronously gets the metadata of a stream. getStreamMetadata :: Connection -> Text -> IO (Async StreamMetadataResult)-getStreamMetadata conn evt_stream = do-    as <- readEvent conn (metaStreamOf evt_stream) (-1) False-    async $ atomically $ waitSTM as >>= extractStreamMetadataResult evt_stream-----------------------------------------------------------------------------------extractStreamMetadataResult :: Monad m-                            => Text-                            -> ReadResult-                            -> m StreamMetadataResult-extractStreamMetadataResult stream rres =-    case readResultStatus rres of-        RE_SUCCESS ->-            case action of-                Just orig ->-                    case decode $ fromStrict $ recordedEventData orig of-                        Just s ->-                            let res = StreamMetadataResult-                                      { streamMetaResultStream  = stream-                                      , streamMetaResultVersion = evt_number-                                      , streamMetaResultData    = s-                                      } in-                            return res-                        Nothing -> fail "StreamMetadata: wrong format."-                Nothing -> fail "impossible: extractStreamMetadataResult"-        RE_STREAM_DELETED -> return $ DeletedStreamMetadataResult stream-        RE_NOT_FOUND      -> return $ NotFoundStreamMetadataResult stream-        RE_NO_STREAM      -> return $ NotFoundStreamMetadataResult stream-        _                 -> fail "unexpected ReadEventResult"--  where-    action     = readResultResolvedEvent rres >>= resolvedEventOriginal-    evt_number = readResultEventNumber rres+getStreamMetadata Connection{..} evt_stream = do+    (k, as) <- createOpAsync+    let op = Op.readMetaStream _settings evt_stream+    pushOperation _prod k op+    return as  -------------------------------------------------------------------------------- -- | Asynchronously create a persistent subscription group on a stream.@@ -498,11 +705,15 @@                              -> Text                              -> Text                              -> PersistentSubscriptionSettings-                             -> IO (Async ())+                             -> IO (Async (Maybe S.PersistActionException)) createPersistentSubscription Connection{..} group stream sett = do-    (as, mvar) <- createAsync-    _runCmd (CreatePersist group stream sett (putMVar mvar))-    return as+    mvar <- newEmptyTMVarIO+    let _F res = atomically $+            case res of+                Left e -> putTMVar mvar (Just e)+                _      -> putTMVar mvar Nothing+    pushCreatePersist _prod _F group stream sett+    async $ atomically $ readTMVar mvar  -------------------------------------------------------------------------------- -- | Asynchronously update a persistent subscription group on a stream.@@ -510,22 +721,30 @@                              -> Text                              -> Text                              -> PersistentSubscriptionSettings-                             -> IO (Async ())+                             -> IO (Async (Maybe S.PersistActionException)) updatePersistentSubscription Connection{..} group stream sett = do-    (as, mvar) <- createAsync-    _runCmd (UpdatePersist group stream sett (putMVar mvar))-    return as+    mvar <- newEmptyTMVarIO+    let _F res = atomically $+            case res of+                Left e -> putTMVar mvar (Just e)+                _      -> putTMVar mvar Nothing+    pushUpdatePersist _prod _F group stream sett+    async $ atomically $ readTMVar mvar  -------------------------------------------------------------------------------- -- | Asynchronously delete a persistent subscription group on a stream. deletePersistentSubscription :: Connection                              -> Text                              -> Text-                             -> IO (Async ())+                             -> IO (Async (Maybe S.PersistActionException)) deletePersistentSubscription Connection{..} group stream = do-    (as, mvar) <- createAsync-    _runCmd (DeletePersist group stream (putMVar mvar))-    return as+    mvar <- newEmptyTMVarIO+    let _F res = atomically $+            case res of+                Left e -> putTMVar mvar (Just e)+                _      -> putTMVar mvar Nothing+    pushDeletePersist _prod _F group stream+    async $ atomically $ readTMVar mvar  -------------------------------------------------------------------------------- -- | Asynchronously connect to a persistent subscription given a group on a@@ -534,22 +753,40 @@                                 -> Text                                 -> Text                                 -> Int32-                                -> IO (Async (Subscription Persistent))+                                -> IO (Subscription S.Persistent) connectToPersistentSubscription Connection{..} group stream bufSize = do-    mvar <- newEmptyMVar-    _runCmd (ConnectPersist group stream bufSize (putMVar mvar))-    async $ readMVar mvar+    mvar <- newEmptyTMVarIO+    var  <- newTVarIO $ SubState S.persistentSubscription Nothing+    let mk r = putTMVar mvar r+        recv = readTVar var+        send = writeTVar var+        dropped r = do+            SubState sm _ <- readTVar var+            writeTVar var $ SubState sm (Just r)+        cb = createSubAsync mk recv send dropped+    pushConnectPersist _prod cb group stream bufSize+    return $ Subscription var mvar stream _prod (S.Persistent group)  ---------------------------------------------------------------------------------createAsync :: IO (Async a, MVar (OperationExceptional a))-createAsync = do+createOpAsync :: IO (Either OperationError a -> IO (), Async a)+createOpAsync = do     mvar <- newEmptyMVar     as   <- async $ do         res <- readMVar mvar         either throwIO return res--    return (as, mvar)+    return (putMVar mvar, as)  ---------------------------------------------------------------------------------metaStreamOf :: Text -> Text-metaStreamOf s = "$$" <> s+createSubAsync :: (S.Running -> STM ())+               -> STM (SubState a)+               -> (SubState a -> STM ())+               -> (S.SubDropReason -> STM ())+               -> (S.SubConnectEvent -> IO ())+createSubAsync mk rcv send quit = go+  where+    go (S.SubConfirmed run) = atomically $ mk run+    go (S.EventAppeared e) = atomically $ do+        SubState sm close <- rcv+        let nxt = S.eventArrived e sm+        send $ SubState nxt close+    go (S.Dropped r) = atomically $ quit r

@@ -1,186 +0,0 @@-{-# LANGUAGE DeriveDataTypeable #-}-{-# LANGUAGE OverloadedStrings  #-}-{-# LANGUAGE TypeFamilies       #-}------------------------------------------------------------------------------------ |--- 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 where-----------------------------------------------------------------------------------import Control.Concurrent-import Control.Exception-import Control.Monad-import Data.Foldable (traverse_)-import Data.Int-import Data.Maybe-----------------------------------------------------------------------------------import Control.Concurrent.Async-import Data.Text-----------------------------------------------------------------------------------import Database.EventStore.Internal.Manager.Subscription-import Database.EventStore.Internal.Operation.ReadStreamEventsOperation-import Database.EventStore.Internal.Operation.ReadAllEventsOperation-import Database.EventStore.Internal.Types-----------------------------------------------------------------------------------defaultBatchSize :: Int32-defaultBatchSize = 500-----------------------------------------------------------------------------------catchupStart :: (Int32 -> Int32 -> IO (Async StreamEventsSlice))-             -> IO (Async (Subscription Regular))-             -> Text-             -> Maybe Int32-             -> Maybe Int32-             -> IO (Subscription Catchup)-catchupStart 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--        maybe (return ()) throwIO res_m-        action <- get_sub-        sub    <- wait action-        putMVar var sub-        keepAwaitingSubEvent stream_id chan sub--    let catchup = Catchup var-        sub     = Subscription-                  { subStreamId = stream_id-                  , subNextEvent = readChan chan-                  , subIsSubscribedToAll = stream_id == ""-                  , subUnsubscribe = do-                      cancel as-                      sub_m <- tryTakeMVar var-                      traverse_ subUnsubscribe sub_m-                  , _subInternal = catchup-                  }--    return sub-----------------------------------------------------------------------------------catchupAllStart :: ( Position -> Int32 -> IO (Async AllEventsSlice))-                -> IO (Async (Subscription Regular))-                -> Maybe Position-                -> Maybe Int32-                -> IO (Subscription 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 var-        sub = Subscription-              { subStreamId = ""-              , subNextEvent = readChan chan-              , subIsSubscribedToAll = True-              , subUnsubscribe = do-                  cancel as-                  sub_m <- tryTakeMVar var-                  traverse_ subUnsubscribe sub_m-              , _subInternal = catchup-              }--    return sub-----------------------------------------------------------------------------------readEventsTill :: (Int32 -> Int32 -> IO (Async StreamEventsSlice))-               -> (Either CatchupError ResolvedEvent -> IO ())-               -> Text-               -> Int32-               -> Int32-               -> IO (Maybe CatchupError)-readEventsTill evts_fwd proc_evt stream_id start batch_size =-    loop False start-  where-    loop done cur_evt_num-        | 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-                          n_done = streamEventsSliceIsEOS slice-                          evts   = streamEventsSliceEvents slice--                      traverse_ (proc_evt . Right) evts-                      loop n_done nxt-                  RS_NO_STREAM      -> loop True cur_evt_num-                  RS_STREAM_DELETED -> reportError deletedError-                  s -> reportError $ unexpectedError s--    deletedError = CatchupStreamDeleted stream_id--    unexpectedError s = CatchupUnexpectedStreamStatus stream_id s--    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 Regular-                     -> IO ()-keepAwaitingSubEvent stream_id chan sub = forever $ do-    evt_e <- subNextEvent 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

@@ -1,4 +1,6 @@ {-# LANGUAGE DeriveDataTypeable  #-}+{-# LANGUAGE GADTs               #-}+{-# LANGUAGE RecordWildCards     #-} {-# LANGUAGE ScopedTypeVariables #-} -------------------------------------------------------------------------------- -- |@@ -14,16 +16,18 @@ module Database.EventStore.Internal.Connection     ( Connection     , ConnectionException(..)+    , HostName     , connUUID     , connClose-    , connFlush     , connSend     , connRecv+    , connIsClosed     , newConnection     ) where  -------------------------------------------------------------------------------- import           Control.Concurrent+import           Control.Concurrent.STM import           Control.Exception import qualified Data.ByteString as B import           Data.Typeable@@ -31,79 +35,170 @@  -------------------------------------------------------------------------------- import Data.UUID+import Data.UUID.V4 import Network-import System.Random  -------------------------------------------------------------------------------- import Database.EventStore.Internal.Types import Database.EventStore.Logging  ---------------------------------------------------------------------------------data ConnectionException =-    MaxAttempt HostName Int Int -- ^ HostName Port MaxAttempt's value+-- | Type of connection issue that can arise during the communication with the+--   server.+data ConnectionException+    = MaxAttemptConnectionReached HostName Int Int+      -- ^ The max reconnection attempt threshold has been reached. Holds a+      --   'HostName', the port used and the given threshold.+    | ClosedConnection+      -- ^ Use of a close 'Connection'.     deriving (Show, Typeable)  -------------------------------------------------------------------------------- instance Exception ConnectionException  --------------------------------------------------------------------------------+data In a where+    Id    :: In UUID+    Close :: In ()+    Send  :: B.ByteString -> In ()+    Recv  :: Int -> In B.ByteString++--------------------------------------------------------------------------------+-- | Internal representation of a connection with the server. data Connection =     Connection-    { connUUID  :: UUID-    , connClose :: IO ()-    , connFlush :: IO ()-    , connSend  :: B.ByteString -> IO ()-    , connRecv  :: Int -> IO B.ByteString+    { _var   :: TMVar State+    , _host  :: HostName+    , _port  :: Int+    , _setts :: Settings     }  --------------------------------------------------------------------------------+data State+    = Offline+    | Online !UUID !Handle+    | Closed++--------------------------------------------------------------------------------+-- | Creates a new 'Connection'. newConnection :: Settings -> HostName -> Int -> IO Connection-newConnection sett host port =+newConnection setts host port = do+    var <- newTMVarIO Offline+    return $ Connection var host port setts++--------------------------------------------------------------------------------+-- | Gets current 'Connection' 'UUID'.+connUUID :: Connection -> IO UUID+connUUID conn = execute conn Id++--------------------------------------------------------------------------------+-- | Closes the 'Connection'. It will not retry to reconnect after that call. it+--   means a new 'Connection' has to be created. 'ClosedConnection' exception+--   will be raised if the same 'Connection' object is used after a 'connClose'+--   call.+connClose :: Connection -> IO ()+connClose conn = execute conn Close++--------------------------------------------------------------------------------+-- | Writes 'ByteString' into the buffer.+connSend :: Connection -> B.ByteString -> IO ()+connSend conn b = execute conn (Send b)++--------------------------------------------------------------------------------+-- | Asks the requested amount of bytes from the 'handle'.+connRecv :: Connection -> Int -> IO B.ByteString+connRecv conn i = execute conn (Recv i)++--------------------------------------------------------------------------------+-- | Returns True if the connection is in closed state.+connIsClosed :: Connection -> STM Bool+connIsClosed Connection{..} = do+    r <- readTMVar _var+    case r of+        Closed -> return True+        _      -> return False++--------------------------------------------------------------------------------+-- | Main connection logic. It will automatically reconnect to the server when+--   a exception occured while the 'Handle' is accessed.+execute :: forall a. Connection -> In a -> IO a+execute Connection{..} i = do+    res <- atomically $ do+        s <- takeTMVar _var+        case s of+            Offline      -> return $ Right Nothing+            Online u hdl -> return $ Right $ Just (u, hdl)+            Closed       -> return $ Left ClosedConnection+    case i of+        Close ->+            case res of+                Left _ -> atomically $ putTMVar _var Closed+                Right Nothing       -> atomically $ putTMVar _var Closed+                Right (Just (_, h)) -> do+                    hClose h+                    atomically $ putTMVar _var Closed+        other ->+            case res of+                Left e -> do+                    atomically $ putTMVar _var Closed+                    throwIO e+                Right alt -> do+                    sres <- case alt of+                        Nothing     -> newState _setts _host _port+                        Just (u, h) -> return $ Right $ Online u h+                    case sres of+                        Left e -> do+                            atomically $ putTMVar _var Closed+                            throwIO e+                        Right s -> do+                            atomically $ putTMVar _var s+                            let Online u h = s+                            case other of+                                Id       -> return u+                                Send b   -> B.hPut h b >> hFlush h+                                Recv siz -> B.hGet h siz+                                Close    -> error "impossible execute"++--------------------------------------------------------------------------------+newState :: Settings -> HostName -> Int -> IO (Either ConnectionException State)+newState sett host port =     case s_retry sett of         AtMost n ->             let loop i = do                     _settingsLog sett (Info $ Connecting i)-                    catch (connect sett host port) $ \(_ :: SomeException) -> do+                    let action = fmap Right $ connect sett host port+                    catch action $ \(_ :: SomeException) -> do                         threadDelay delay                         if n <= i-                            then do-                                _settingsLog sett-                                             $ Error-                                             $ MaxAttemptConnectionReached i-                                throwIO $ MaxAttempt host port n+                            then return $+                                   Left $+                                   MaxAttemptConnectionReached host port n                             else loop (i + 1) in              loop 1         KeepRetrying ->             let endlessly i = do                     _settingsLog sett (Info $ Connecting i)-                    catch (connect sett host port) $ \(_ :: SomeException) -> do+                    let action = fmap Right $ connect sett host port+                    catch action $ \(_ :: SomeException) ->                         threadDelay delay >> endlessly (i + 1) in              endlessly (1 :: Int)   where-    delay = (s_reconnect_delay_secs sett) * secs+    delay = s_reconnect_delay_secs sett * secs  -------------------------------------------------------------------------------- secs :: Int secs = 1000000  ---------------------------------------------------------------------------------connect :: Settings -> HostName -> Int -> IO Connection+connect :: Settings -> HostName -> Int -> IO State connect sett host port = do     hdl <- connectTo host (PortNumber $ fromIntegral port)     hSetBuffering hdl NoBuffering-    uuid <- randomIO+    uuid <- nextRandom     regularConnection sett hdl uuid  ---------------------------------------------------------------------------------regularConnection :: Settings -> Handle -> UUID -> IO Connection+regularConnection :: Settings -> Handle -> UUID -> IO State regularConnection sett h uuid = do     _settingsLog sett (Info $ Connected uuid)-    return Connection-           { connUUID  = uuid-           , connClose = do-               _settingsLog sett (Info $ ConnectionClosed uuid)-               hClose h-           , connFlush = hFlush h-           , connSend  = B.hPut h-           , connRecv  = B.hGet h-           }+    return $ Online uuid h

@@ -0,0 +1,636 @@+{-# LANGUAGE DataKinds                 #-}+{-# LANGUAGE DeriveDataTypeable        #-}+{-# LANGUAGE ExistentialQuantification #-}+{-# LANGUAGE RecordWildCards           #-}+{-# LANGUAGE ScopedTypeVariables       #-}+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Internal.Execution.Production+-- Copyright : (C) 2015 Yorick Laupa+-- License : (see the file LICENSE)+--+-- Maintainer : Yorick Laupa <yo.eight@gmail.com>+-- Stability : provisional+-- Portability : non-portable+--+-- Production execution model. It's striving for robustness. The model consists+-- on 4 threads. The Reader thread that reads 'Package' from the connection, the+-- Runner thread which executes finalizers submitted by the user (typically what+-- to do on operation completion or when a event has arrived for a subscription)+-- , the writer thread that sends 'Package' to the server, and the Manager+-- thread that handles requests coming both from the user and the Reader+-- thread. If the Reader or Runner threads die, it will be restarted by the+-- Manager thread if the connection hasn't been closed by user in the meantime.+--------------------------------------------------------------------------------+module Database.EventStore.Internal.Execution.Production+    ( Production+    , ServerConnectionError(..)+    , newExecutionModel+    , pushOperation+    , shutdownExecutionModel+    , pushConnectStream+    , pushConnectPersist+    , pushCreatePersist+    , pushUpdatePersist+    , pushDeletePersist+    , pushAckPersist+    , pushNakPersist+    , pushUnsubscribe+    , prodWaitTillClosed+    ) where++--------------------------------------------------------------------------------+import Prelude hiding (take)+import Control.Concurrent+import Control.Concurrent.STM+import Control.Exception+import Control.Monad+import Control.Monad.Fix+import Data.IORef+import Data.Int+import Data.Foldable+import Data.Typeable+import Text.Printf++--------------------------------------------------------------------------------+import Data.Serialize.Get hiding (Done)+import Data.Serialize.Put+import Data.Text+import Data.UUID++--------------------------------------------------------------------------------+import Database.EventStore.Internal.Connection+import Database.EventStore.Internal.Generator+import Database.EventStore.Internal.Manager.Subscription hiding+    ( submitPackage+    , unsubscribe+    , ackPersist+    , nakPersist+    , abort+    )+import Database.EventStore.Internal.Operation hiding (retry)+import Database.EventStore.Internal.Packages+import Database.EventStore.Internal.Processor+import Database.EventStore.Internal.Types+import Database.EventStore.Logging++--------------------------------------------------------------------------------+data Worker+    = Reader ThreadId+    | Runner ThreadId+    | Writer ThreadId+    deriving Show++--------------------------------------------------------------------------------+-- | Raised when the server responded in an unexpected way.+data ServerConnectionError+    = WrongPackageFraming+      -- ^ TCP package sent by the server had a wrong framing.+    | PackageParsingError String+      -- ^ Server sent a malformed TCP package.+    deriving (Show, Typeable)++--------------------------------------------------------------------------------+instance Exception ServerConnectionError++--------------------------------------------------------------------------------+-- | Used to determine if we hit the end of the queue.+data Slot a = Slot !a | End++--------------------------------------------------------------------------------+-- | A 'TQueue' that can be cycled.+newtype CycleQueue a = CycleQueue (TQueue (Slot a))++--------------------------------------------------------------------------------+-- | Creates an empty 'CycleQueue'.+newCycleQueue :: IO (CycleQueue a)+newCycleQueue = fmap CycleQueue newTQueueIO++--------------------------------------------------------------------------------+-- | Gets an element from the 'CycleQueue'.+readCycleQueue :: CycleQueue a -> STM a+readCycleQueue (CycleQueue q) = do+    Slot a <- readTQueue q+    return a++--------------------------------------------------------------------------------+-- | Writes an element to the 'CycleQueue'.+writeCycleQueue :: CycleQueue a -> a -> STM ()+writeCycleQueue (CycleQueue q) a = writeTQueue q (Slot a)++--------------------------------------------------------------------------------+-- | Empties a 'CycleQueue'.+emptyCycleQueue :: CycleQueue a -> STM ()+emptyCycleQueue (CycleQueue q) = writeTQueue q End >> go+  where+    go = do+        s <- readTQueue q+        case s of+            End -> return ()+            _   -> go++--------------------------------------------------------------------------------+-- | Updates a 'CycleQueue'.+updateCycleQueue :: CycleQueue a -> (a -> STM (Maybe a)) -> STM ()+updateCycleQueue (CycleQueue q) k = writeTQueue q End >> go+  where+    go = do+        s <- readTQueue q+        case s of+            End    -> return ()+            Slot a -> do+                r <- k a+                case r of+                    Nothing -> go+                    Just a' -> writeTQueue q (Slot a') >> go++--------------------------------------------------------------------------------+-- | Indicates if a 'CycleQueue' is empty.+isEmptyCycleQueue :: CycleQueue a -> STM Bool+isEmptyCycleQueue (CycleQueue q) = isEmptyTQueue q++--------------------------------------------------------------------------------+wkUpdState :: Worker -> State -> State+wkUpdState (Reader tid) s = s { _reader = Just tid }+wkUpdState (Runner tid) s = s { _runner = Just tid }+wkUpdState (Writer tid) s = s { _writer = Just tid }++--------------------------------------------------------------------------------+-- | Holds the execution model state.+data Production =+    Prod+    { _submit :: TVar (Msg -> IO ())+      -- ^ The action to call when pushing new command.+    , _waitClosed :: STM ()+      -- ^ Action that attests the execution model has been closed successfully.+      --   It doesn't mean the execution model hasn't been shutdown because of+      --   some random exception.+    }++--------------------------------------------------------------------------------+-- | Main execution environment used among different transitions.+data Env =+    Env+    { _setts :: Settings+      -- ^ Global settings reference.+    , _queue :: CycleQueue Msg+      -- ^ That queue ties the user, the reader thread and the manager thread.+      --   The user and the reader push new messages onto the queue while the+      --   manager dequeue and handles one message at the time.+    , _pkgQueue :: CycleQueue Package+      -- ^ That queue ties the writer thread with the manager thread. The writer+      --   dequeue packages from that queue and sends those to the server. While+      --   the manager pushes new packages on every new submitted operation.+    , _jobQueue :: CycleQueue Job+      -- ^ That queue ties the runner thread with the manager thread. The runner+      --   dequeues IO action from it while the manager pushes new command+      --   finalizers as those arrived.+    , _state :: TVar State+      -- ^ Holds manager thread state.+    , _nextSubmit :: TVar (Msg -> IO ())+      -- ^ Indicates the action to call in order to push new commands.+    , _connRef :: IORef Connection+      -- ^ Connection to the server.+    , _disposed :: TMVar ()+      -- ^ Indicates when the production execution model has been shutdown and+      --   disposed any ongoing operations.+    }++--------------------------------------------------------------------------------+data Msg+    = Stopped Worker SomeException+    | Arrived Package+    | Shutdown+    | forall a.+      NewOperation (Either OperationError a -> IO ()) (Operation a)+    | ConnectStream (SubConnectEvent -> IO ()) Text Bool+    | ConnectPersist (SubConnectEvent -> IO ()) Text Text Int32+    | Unsubscribe Running+    | CreatePersist (Either PersistActionException ConfirmedAction -> IO ())+          Text Text PersistentSubscriptionSettings+    | UpdatePersist (Either PersistActionException ConfirmedAction -> IO ())+          Text Text PersistentSubscriptionSettings+    | DeletePersist (Either PersistActionException ConfirmedAction -> IO ())+          Text Text+    | AckPersist Running [UUID]+    | NakPersist Running NakAction (Maybe Text) [UUID]++--------------------------------------------------------------------------------+pushCmd :: Production -> Msg -> IO ()+pushCmd (Prod _sender _) msg = do+    push <- readTVarIO _sender+    push msg++--------------------------------------------------------------------------------+-- | Asks to shutdown the connection to the server asynchronously.+shutdownExecutionModel :: Production -> IO ()+shutdownExecutionModel prod = pushCmd prod Shutdown++--------------------------------------------------------------------------------+-- | Pushes a new 'Operation' asynchronously.+pushOperation :: Production+             -> (Either OperationError a -> IO ())+             -> Operation a+             -> IO ()+pushOperation prod k op = pushCmd prod (NewOperation k op)++--------------------------------------------------------------------------------+-- | Subscribes to a regular stream.+pushConnectStream :: Production+                  -> (SubConnectEvent -> IO ())+                  -> Text+                  -> Bool+                  -> IO ()+pushConnectStream prod k n tos = pushCmd prod (ConnectStream k n tos)++--------------------------------------------------------------------------------+-- | Subscribes to a persistent subscription.+pushConnectPersist :: Production+                   -> (SubConnectEvent -> IO ())+                   -> Text+                   -> Text+                   -> Int32+                   -> IO ()+pushConnectPersist prod k g n buf = pushCmd prod (ConnectPersist k g n buf)++--------------------------------------------------------------------------------+-- | Creates a persistent subscription.+pushCreatePersist :: Production+                  -> (Either PersistActionException ConfirmedAction -> IO ())+                  -> Text+                  -> Text+                  -> PersistentSubscriptionSettings+                  -> IO ()+pushCreatePersist prod k g n setts = pushCmd prod (CreatePersist k g n setts)++--------------------------------------------------------------------------------+-- | Updates a persistent subscription.+pushUpdatePersist :: Production+                  -> (Either PersistActionException ConfirmedAction -> IO ())+                  -> Text+                  -> Text+                  -> PersistentSubscriptionSettings+                  -> IO ()+pushUpdatePersist prod k g n setts = pushCmd prod (UpdatePersist k g n setts)++--------------------------------------------------------------------------------+-- | Deletes a persistent subscription.+pushDeletePersist :: Production+                  -> (Either PersistActionException ConfirmedAction -> IO ())+                  -> Text+                  -> Text+                  -> IO ()+pushDeletePersist prod k g n = pushCmd prod (DeletePersist k g n)++--------------------------------------------------------------------------------+-- | Acknowledges a set of events has been successfully handled.+pushAckPersist :: Production -> Running -> [UUID] -> IO ()+pushAckPersist prod run evts = pushCmd prod (AckPersist run evts)++--------------------------------------------------------------------------------+-- | Acknowledges a set of events hasn't been handled successfully.+pushNakPersist :: Production+               -> Running+               -> NakAction+               -> Maybe Text+               -> [UUID]+               -> IO ()+pushNakPersist prod run act res evts =+    pushCmd prod (NakPersist run act res evts)++--------------------------------------------------------------------------------+-- | Unsubscribe from a subscription.+pushUnsubscribe :: Production -> Running -> IO ()+pushUnsubscribe prod r = pushCmd prod (Unsubscribe r)++--------------------------------------------------------------------------------+-- | Waits the execution model to close properly.+prodWaitTillClosed :: Production -> IO ()+prodWaitTillClosed (Prod _ disposed) = atomically disposed++--------------------------------------------------------------------------------+newtype Job = Job (IO ())++--------------------------------------------------------------------------------+-- Internal Production state.+--------------------------------------------------------------------------------+data State =+    State+    { _proc   :: !(Processor (IO ()))+    , _reader :: !(Maybe ThreadId)+    , _runner :: !(Maybe ThreadId)+    , _writer :: !(Maybe ThreadId)+    }++--------------------------------------------------------------------------------+emptyState :: Settings -> Generator -> State+emptyState setts gen = State (newProcessor setts gen) Nothing Nothing Nothing++--------------------------------------------------------------------------------+updateProc :: Processor (IO ()) -> State -> State+updateProc p s = s { _proc = p }++--------------------------------------------------------------------------------+-- | Reader thread. Keeps reading 'Package' from the connection.+--------------------------------------------------------------------------------+reader :: Settings -> CycleQueue Msg -> Connection -> IO ()+reader sett queue c = forever $ do+    header_bs <- connRecv c 4+    case runGet getLengthPrefix header_bs of+        Left _              -> throwIO WrongPackageFraming+        Right length_prefix -> connRecv c length_prefix >>= parsePackage+  where+    parsePackage bs =+        case runGet getPackage bs of+            Left e    -> throwIO $ PackageParsingError e+            Right pkg -> do+                atomically $ writeCycleQueue queue (Arrived pkg)+                let cmd  = packageCmd pkg+                    uuid = packageCorrelation pkg+                _settingsLog sett $ Info $ PackageReceived cmd uuid++--------------------------------------------------------------------------------+-- | Writer thread, writes incoming 'Package's+--------------------------------------------------------------------------------+writer :: Settings -> CycleQueue Package -> Connection -> IO ()+writer setts pkg_queue conn = forever $ do+    pkg <- atomically $ readCycleQueue pkg_queue+    connSend conn $ runPut $ putPackage pkg+    let cmd  = packageCmd pkg+        uuid = packageCorrelation pkg+    _settingsLog setts $ Info $ PackageSent cmd uuid++--------------------------------------------------------------------------------+getLengthPrefix :: Get Int+getLengthPrefix = fmap fromIntegral getWord32le++--------------------------------------------------------------------------------+getPackage :: Get Package+getPackage = do+    cmd  <- getWord8+    flg  <- getFlag+    col  <- getUUID+    cred <- getCredentials flg+    rest <- remaining+    dta  <- getBytes rest++    let pkg = Package+              { packageCmd         = cmd+              , packageCorrelation = col+              , packageData        = dta+              , packageCred        = cred+              }++    return pkg++--------------------------------------------------------------------------------+getFlag :: Get Flag+getFlag = do+    wd <- getWord8+    case wd of+        0x00 -> return None+        0x01 -> return Authenticated+        _    -> fail $ printf "TCP: Unhandled flag value 0x%x" wd++--------------------------------------------------------------------------------+getCredEntryLength :: Get Int+getCredEntryLength = fmap fromIntegral getWord8++--------------------------------------------------------------------------------+getCredentials :: Flag -> Get (Maybe Credentials)+getCredentials None = return Nothing+getCredentials _ = do+    loginLen <- getCredEntryLength+    login    <- getBytes loginLen+    passwLen <- getCredEntryLength+    passw    <- getBytes passwLen+    return $ Just $ credentials login passw++--------------------------------------------------------------------------------+getUUID :: Get UUID+getUUID = do+    bs <- getLazyByteString 16+    case fromByteString bs of+        Just uuid -> return uuid+        _         -> fail "TCP: Wrong UUID format"++--------------------------------------------------------------------------------+-- Runner thread. Keeps running job comming from the Manager thread.+--------------------------------------------------------------------------------+runner :: CycleQueue Job -> IO ()+runner job_queue = forever $ do+    Job j <- atomically $ readCycleQueue job_queue+    j++--------------------------------------------------------------------------------+-- | Spawns a new thread worker.+spawn :: Env -> (ThreadId -> Worker) -> IO Worker+spawn Env{..} mk = do+    conn <- readIORef _connRef+    tid  <- mfix $ \tid ->+        let worker = mk tid+            action =+                case worker of+                    Reader _ -> reader _setts _queue conn+                    Runner _ -> runner _jobQueue+                    Writer _ -> writer _setts _pkgQueue conn in+        forkFinally action $ \r ->+            case r of+                Left e ->+                    case asyncExceptionFromException e of+                        Just ThreadKilled -> return ()+                        _ ->  atomically $ writeCycleQueue _queue+                                         $ Stopped worker e+                _      -> return ()+    return $ mk tid++--------------------------------------------------------------------------------+-- | Loops over a 'Processor''s 'Transition' state machine, returning an updated+--   'Processor' model at the end.+runTransition :: Env -> Transition (IO ()) -> STM (Processor (IO ()))+runTransition Env{..} = go+  where+    go (Produce j nxt) = do+        let job = Job j+        writeCycleQueue _jobQueue job+        go nxt+    go (Transmit pkg nxt) = do+        writeCycleQueue _pkgQueue pkg+        go nxt+    go (Await new_proc) = return new_proc++--------------------------------------------------------------------------------+-- | First execution mode. It spawns initial reader, runner and writer threads.+--   Then it switches to 'cruising' mode.+bootstrap :: Env -> IO ()+bootstrap env@Env{..} = do+    rew <- spawn env Reader+    ruw <- spawn env Runner+    wrw <- spawn env Writer+    let _F = wkUpdState rew .+             wkUpdState ruw .+             wkUpdState wrw+    atomically $ modifyTVar' _state _F+    cruising env++--------------------------------------------------------------------------------+-- | Crusing execution mode. Reads and handle message coming from the channel as+--   those are arrived. That mode is used when the connection to the server is+--   still live. We might have deconnection once in a while but at the end, if+--   we managed to reconnect to it, we consider everything is fine.+cruising :: Env -> IO ()+cruising env@Env{..} = do+    msg <- atomically $ readCycleQueue _queue+    s   <- readTVarIO _state+    case msg of+        Stopped _ e -> throwIO e+        Arrived pkg -> do+            let sm = submitPackage pkg $ _proc s+            atomically $ do+                new_proc <- runTransition env sm+                modifyTVar' _state $ updateProc new_proc+            cruising env+        Shutdown -> throwIO ClosedConnection+        NewOperation k op -> do+            let sm = newOperation k op $ _proc s+            atomically $ do+                new_proc <- runTransition env sm+                modifyTVar' _state $ updateProc new_proc+            cruising env+        ConnectStream k n tos -> do+            let sm = connectRegularStream k n tos $ _proc s+            atomically $ do+                new_proc <- runTransition env sm+                modifyTVar' _state $ updateProc new_proc+            cruising env+        ConnectPersist k g n b -> do+            let sm = connectPersistent k g n b $ _proc s+            atomically $ do+                new_proc <- runTransition env sm+                modifyTVar' _state $ updateProc new_proc+            cruising env+        Unsubscribe r -> do+            let sm = unsubscribe r $ _proc s+            atomically $ do+                new_proc <- runTransition env sm+                modifyTVar' _state $ updateProc new_proc+            cruising env+        CreatePersist k g n psetts -> do+            let sm = createPersistent k g n psetts $ _proc s+            atomically $ do+                new_proc <- runTransition env sm+                modifyTVar' _state $ updateProc new_proc+            cruising env+        UpdatePersist k g n psetts -> do+            let sm = updatePersistent k g n psetts $ _proc s+            atomically $ do+                new_proc <- runTransition env sm+                modifyTVar' _state $ updateProc new_proc+            cruising env+        DeletePersist k g n -> do+            let sm = deletePersistent k g n $ _proc s+            atomically $ do+                new_proc <- runTransition env sm+                modifyTVar' _state $ updateProc new_proc+            cruising env+        AckPersist run evts -> do+            let sm = ackPersist (return ()) run evts $ _proc s+            atomically $ do+                new_proc <- runTransition env sm+                modifyTVar' _state $ updateProc new_proc+            cruising env+        NakPersist run act res evts -> do+            let sm = nakPersist (return ()) run act res evts $ _proc s+            atomically $ do+                new_proc <- runTransition env sm+                modifyTVar' _state $ updateProc new_proc+            cruising env++--------------------------------------------------------------------------------+-- | That mode is triggered either because the user asks to shutdown the+--   connection or because the connection to server has been dropped and we+--   can't reconnect.+closing :: Env -> IO ()+closing env@Env{..} = do+    State _ retid rutid  wutid <- readTVarIO _state+    -- We kill reader and writer threads to avoid in fly package in the cleaning+    -- phase.+    traverse_ killThread retid+    traverse_ killThread wutid++    -- Discards every 'Package' that was about to be sent.+    atomically $ emptyCycleQueue _pkgQueue++    -- Takes care of 'Package's that have already arrived. Just in case those+    -- are completing or moving forward ongoing operations. Every ongoing+    -- request is kept for later reconnection. Some transient operations like+    -- Ack, Nak or Unsubscribe are just discard.+    atomically $ updateCycleQueue _queue $ \nxt ->+        case nxt of+            Arrived pkg -> do+                s <- readTVar _state+                let sm = submitPackage pkg $ _proc s+                nxt_proc <- runTransition env sm+                modifyTVar' _state $ updateProc nxt_proc+                return Nothing+            Shutdown -> return Nothing+            AckPersist _ _ -> return Nothing+            NakPersist _ _ _ _ -> return Nothing+            Unsubscribe _ -> return Nothing+            Stopped _ _ -> return Nothing+            _ -> return $ Just nxt++    -- If the connection is already closed, it will throw an exception. We just+    -- make sure it doesn't interfere with the cleaning process.+    conn <- readIORef _connRef+    _    <- try $ connClose conn :: (IO (Either ConnectionException ()))+    atomically $ do+        s <- readTVar _state+        _ <- runTransition env $ abort $ _proc s+        return ()++    -- Waits the runner thread to deal with its jobs list.+    atomically $ do+        end <- isEmptyCycleQueue _jobQueue+        unless end retry++    traverse_ killThread rutid++--------------------------------------------------------------------------------+raiseException :: Exception e => e -> Msg -> IO ()+raiseException e _ = throwIO e++--------------------------------------------------------------------------------+-- | Main Production execution model entry point.+newExecutionModel :: Settings -> HostName -> Int -> IO Production+newExecutionModel setts host port = do+    gen       <- newGenerator+    queue     <- newCycleQueue+    pkg_queue <- newCycleQueue+    job_queue <- newCycleQueue+    conn      <- newConnection setts host port+    conn_ref  <- newIORef conn+    var       <- newTVarIO $ emptyState setts gen+    nxt_sub   <- newTVarIO (atomically . writeCycleQueue queue)+    disposed  <- newEmptyTMVarIO+    let env = Env setts queue pkg_queue job_queue var nxt_sub conn_ref disposed+        handler res = do+            closing env+            case res of+                Left e -> do+                    _settingsLog setts (Error $ UnexpectedException e)+                    case fromException e of+                        Just (_ :: ConnectionException) -> atomically $ do+                            writeTVar nxt_sub (raiseException e)+                            putTMVar disposed ()+                        _ -> do new_conn <- newConnection setts host port+                                writeIORef conn_ref new_conn+                                _ <- forkFinally (bootstrap env) handler+                                return ()++                _ -> atomically $ putTMVar disposed ()+    _ <- forkFinally (bootstrap env) handler+    return $ Prod nxt_sub $ do+        closed <- connIsClosed conn+        unless closed retry+        readTMVar disposed

@@ -0,0 +1,42 @@+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Internal.Generator+-- Copyright : (C) 2015 Yorick Laupa+-- License : (see the file LICENSE)+--+-- Maintainer : Yorick Laupa <yo.eight@gmail.com>+-- Stability : provisional+-- Portability : non-portable+--+-- Pure UUID generator.+--------------------------------------------------------------------------------+module Database.EventStore.Internal.Generator+    ( Generator+    , nextUUID+    , newGenerator+    , splitGenerator+    ) where++--------------------------------------------------------------------------------+import Data.UUID+import System.Random++--------------------------------------------------------------------------------+-- | Pure 'UUID' generator.+newtype Generator = Generator StdGen++--------------------------------------------------------------------------------+-- | Gets the next fresh 'UUID'.+nextUUID :: Generator -> (UUID, Generator)+nextUUID (Generator g) =  let (u, nxt) = random g in (u, Generator nxt)++--------------------------------------------------------------------------------+-- | Builds 2 new 'Generator's out of one.+splitGenerator :: Generator -> (Generator, Generator)+splitGenerator (Generator g) =+    let (g1, g2) = split g in (Generator g1, Generator g2)++--------------------------------------------------------------------------------+-- | Creates a new 'Generator'.+newGenerator :: IO Generator+newGenerator = fmap Generator getStdGen

@@ -1,197 +0,0 @@-{-# LANGUAGE RecordWildCards           #-}-{-# LANGUAGE ExistentialQuantification #-}------------------------------------------------------------------------------------ |--- Module : Database.EventStore.Internal.Manager.Operation--- 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.Internal.Manager.Operation-    ( Decision(..)-    , OperationParams(..)-    , operationNetwork-    ) where-----------------------------------------------------------------------------------import qualified Data.Map.Strict as M-import           Data.Monoid ((<>))-import           Data.Word-----------------------------------------------------------------------------------import Data.ProtocolBuffers-import Data.Serialize-import Data.UUID-import FRP.Sodium-import System.Random-----------------------------------------------------------------------------------import Database.EventStore.Internal.Types hiding (Event, newEvent)-import Database.EventStore.Internal.Util.Sodium-----------------------------------------------------------------------------------newtype Manager = Manager (M.Map UUID Operation)-----------------------------------------------------------------------------------initManager :: Manager-initManager = Manager M.empty------------------------------------------------------------------------------------- Operation----------------------------------------------------------------------------------data Decision-    = DoNothing-    | EndOperation-    | Retry-    | Reconnection-    | Subscribed-----------------------------------------------------------------------------------data Operation-    = Operation-      { operationCreatePackage :: UUID    -> IO Package-      , operationInspect       :: Package -> IO Decision-      }-----------------------------------------------------------------------------------data OperationParams-    = forall req resp. (Encode req, Decode resp) =>-      OperationParams-      { opSettings    :: !Settings-      , opRequestCmd  :: !Word8-      , opResponseCmd :: !Word8--      , opRequest     :: IO req-      , opSuccess     :: resp -> IO Decision-      , opFailure     :: OperationException -> IO Decision-      }-----------------------------------------------------------------------------------createOperation :: Settings -> OperationParams -> Operation-createOperation sett params =-    Operation-    { operationCreatePackage = createPackage sett params-    , operationInspect       = inspection params-    }-----------------------------------------------------------------------------------createPackage :: Settings -> OperationParams -> UUID -> IO Package-createPackage Settings{..} OperationParams{..} uuid = do-    req <- opRequest--    let pack = Package-               { packageCmd         = opRequestCmd-               , packageCorrelation = uuid-               , packageData        = runPut $ encodeMessage req-               , packageCred        = s_credentials-               }--    return pack-----------------------------------------------------------------------------------inspection :: OperationParams -> Package -> IO Decision-inspection params@OperationParams{..} pack-    | found == exp_v = deeperInspection params pack-    | otherwise      = failed (InvalidServerResponse exp_v found)-  where-    exp_v  = opResponseCmd-    failed = opFailure-    found  = packageCmd pack-----------------------------------------------------------------------------------deeperInspection :: OperationParams -> Package -> IO Decision-deeperInspection OperationParams{..} pack =-    case runGet decodeMessage bytes of-        Left e    -> failed (ProtobufDecodingError e)-        Right msg -> succeed msg-  where-    failed  = opFailure-    succeed = opSuccess-    bytes   = packageData pack------------------------------------------------------------------------------------- Event----------------------------------------------------------------------------------data Register = Register UUID Operation--newtype Remove = Remove UUID--data Response = Response !Package !Operation-----------------------------------------------------------------------------------operationNetwork :: Settings-                 -> (Package -> Reactive ())-                 -> Reactive ()-                 -> Event Package-                 -> Reactive (OperationParams -> Reactive ())-operationNetwork sett push_pkg push_reco e_pkg = do-    (on_new, push_new) <- newEvent-    (on_reg, push_reg) <- newEvent-    (on_rem, push_rem) <- newEvent-    (on_ret, push_ret) <- newEvent--    let mgr_e = fmap register on_reg <>-                fmap remove on_rem--    mgr_b <- accum initManager mgr_e--    let resp_e = filterJust $ snapshot response e_pkg mgr_b--        on_new_op = fmap (createOperation sett) on_new <> on_ret--        push_reg_io   = pushAsync2 $ \uuid op -> push_reg $ Register uuid op-        push_rem_io   = pushAsync (push_rem . Remove)-        push_retry_io = pushAsync2 $ \uuid op -> do-            push_rem $ Remove uuid-            push_ret op-        push_reco_io  = pushAsync2 $ \uuid op -> do-            push_reco-            push_rem $ Remove uuid-            push_ret op-        push_send_io  = pushAsync push_pkg--    _ <- listen on_new_op $ \op -> do-             uuid <- randomIO-             push_reg_io uuid op--    _ <- listen resp_e $ \(Response pkg op) -> do-             decision <- operationInspect op pkg-             let corr_id = packageCorrelation pkg--             case decision of-                 DoNothing    -> return ()-                 EndOperation -> push_rem_io corr_id-                 Retry        -> push_retry_io corr_id op-                 Reconnection -> push_reco_io corr_id op-                 _            -> fail unexpectedDecision--    _ <- listen on_reg $ \(Register uuid op) ->-             operationCreatePackage op uuid >>= push_send_io--    return push_new-----------------------------------------------------------------------------------unexpectedDecision :: String-unexpectedDecision = "Unexpected decision Processor.handlingOperation"------------------------------------------------------------------------------------- Model----------------------------------------------------------------------------------register :: Register -> Manager -> Manager-register (Register uuid op) (Manager m) = Manager $ M.insert uuid op m-----------------------------------------------------------------------------------remove :: Remove -> Manager -> Manager-remove (Remove uuid) (Manager m) = Manager $ M.delete uuid m------------------------------------------------------------------------------------- Snapshot----------------------------------------------------------------------------------response :: Package -> Manager -> Maybe Response-response pkg (Manager m) = fmap (Response pkg) $ M.lookup corr_id m-  where-    corr_id = packageCorrelation pkg

@@ -0,0 +1,178 @@+{-# LANGUAGE ExistentialQuantification #-}+{-# LANGUAGE RecordWildCards           #-}+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Internal.Manager.Operation.Model+-- Copyright : (C) 2015 Yorick Laupa+-- License : (see the file LICENSE)+--+-- Maintainer : Yorick Laupa <yo.eight@gmail.com>+-- Stability : provisional+-- Portability : non-portable+--+-- Main operation bookkeeping structure.+--------------------------------------------------------------------------------+module Database.EventStore.Internal.Manager.Operation.Model+    ( Model+    , Transition(..)+    , newModel+    , pushOperation+    , submitPackage+    , abort+    ) where++--------------------------------------------------------------------------------+import Data.Word++--------------------------------------------------------------------------------+import qualified Data.HashMap.Strict  as H+import           Data.ProtocolBuffers+import           Data.Serialize+import           Data.UUID++--------------------------------------------------------------------------------+import Database.EventStore.Internal.Generator+import Database.EventStore.Internal.Operation+import Database.EventStore.Internal.Types++--------------------------------------------------------------------------------+-- | Entry of a running 'Operation'.+data Elem r =+    forall a resp. Decode resp =>+    Elem+    { _opOp   :: Operation a+    , _opCmd  :: Word8+    , _opCont :: resp -> SM a ()+    , _opCb   :: Either OperationError a -> r+    }++--------------------------------------------------------------------------------+-- | Operation internal state.+data State r =+    State+    { _gen :: Generator+      -- ^ 'UUID' generator.+    , _pending :: H.HashMap UUID (Elem r)+      -- ^ Contains all running 'Operation's.+    }++--------------------------------------------------------------------------------+initState :: Generator -> State r+initState g = State g H.empty++--------------------------------------------------------------------------------+-- | Type of requests handled by the model.+data Request r+    = forall a. New (Operation a) (Either OperationError a -> r)+      -- ^ Register a new 'Operation'.+    | Pkg Package+      -- ^ Submit a package.+    | Abort+      -- ^ Aborts every pending operation.++--------------------------------------------------------------------------------+-- | Output produces by the interpretation of an 'Operation'.+data Transition r+    = Produce r (Transition r)+      -- ^ Produces an intermediary value.+    | Transmit Package (Transition r)+      -- ^ Asks for sending the given 'Package'.+    | Await (Model r)+      -- ^ waits for more input.++--------------------------------------------------------------------------------+-- | Main 'Operation' bookkeeping state machine.+newtype Model r = Model (Request r -> Maybe (Transition r))++--------------------------------------------------------------------------------+-- | Pushes a new 'Operation' to model. The given 'Operation' state-machine is+--   initialized and produces a 'Package'.+pushOperation :: (Either OperationError a -> r)+              -> Operation a+              -> Model r+              -> Transition r+pushOperation cb op (Model k) = let Just t = k (New op cb) in t++--------------------------------------------------------------------------------+-- | Submits a 'Package' to the model. If the model isn't concerned by the+--   'Package', it will returns 'Nothing'. Because 'Operation' can implement+--   complex logic (retry for instance), it returns a 'Step'.+submitPackage :: Package -> Model r -> Maybe (Transition r)+submitPackage pkg (Model k) = k (Pkg pkg)++--------------------------------------------------------------------------------+-- | Aborts every pending operation.+abort :: Model r -> Transition r+abort (Model k) = let Just t = k Abort in t++--------------------------------------------------------------------------------+runOperation :: Settings+             -> (Either OperationError a -> r)+             -> Operation a+             -> SM a ()+             -> State r+             -> Transition r+runOperation setts cb op start init_st = go init_st start+  where+    go st (Return _) = Await $ Model $ handle setts st+    go st (Yield a n) = Produce (cb $ Right a) (go st n)+    go st (FreshId k) =+        let (new_id, nxt_gen) = nextUUID $ _gen st+            nxt_st            = st { _gen = nxt_gen } in+        go nxt_st $ k new_id+    go st (SendPkg ci co rq k) =+        let (new_uuid, nxt_gen) = nextUUID $ _gen st+            pkg = Package+                  { packageCmd         = ci+                  , packageCorrelation = new_uuid+                  , packageData        = runPut $ encodeMessage rq+                  , packageCred        = s_credentials setts+                  }+            elm    = Elem op co k cb+            ps     = H.insert new_uuid elm $ _pending st+            nxt_st = st { _pending = ps+                        , _gen     = nxt_gen+                        } in+        Transmit pkg (Await $ Model $ handle setts nxt_st)+    go st (Failure m) =+        case m of+            Just e -> Produce (cb $ Left e) (Await $ Model $ handle setts st)+            _      -> runOperation setts cb op op st++--------------------------------------------------------------------------------+runPackage :: Settings -> State r -> Package -> Maybe (Transition r)+runPackage setts st Package{..} = do+    Elem op resp_cmd cont cb <- H.lookup packageCorrelation $ _pending st+    let nxt_ps = H.delete packageCorrelation $ _pending st+        nxt_st = st { _pending = nxt_ps }+    if resp_cmd /= packageCmd+        then+            let r = cb $ Left $ InvalidServerResponse resp_cmd packageCmd in+            return $ Produce r (Await $ Model $ handle setts nxt_st)+        else+            case runGet decodeMessage packageData of+                Left e  ->+                    let r = cb $ Left $ ProtobufDecodingError e in+                    return $ Produce r (Await $ Model $ handle setts nxt_st)+                Right m -> return $ runOperation setts cb op (cont m) nxt_st++--------------------------------------------------------------------------------+abortOperations :: Settings -> State r -> Transition r+abortOperations setts init_st = go init_st $ H.toList $ _pending init_st+  where+    go st ((key, Elem _ _ _ k):xs) =+        let ps     = H.delete key $ _pending st+            nxt_st = st { _pending = ps } in+        Produce (k $ Left Aborted) $ go nxt_st xs+    go st [] = Await $ Model $ handle setts st++--------------------------------------------------------------------------------+-- | Creates a new 'Operation' model state-machine.+newModel :: Settings -> Generator -> Model r+newModel setts g = Model $ handle setts $ initState g++--------------------------------------------------------------------------------+handle :: Settings -> State r -> Request r -> Maybe (Transition r)+handle setts st (New op cb) = Just $ runOperation setts cb op op st+handle setts st (Pkg pkg)   = runPackage setts st pkg+handle setts st Abort       = Just $ abortOperations setts st

@@ -1,1112 +1,220 @@-{-# LANGUAGE BangPatterns              #-}-{-# LANGUAGE DeriveGeneric             #-}-{-# LANGUAGE DataKinds                 #-}-{-# LANGUAGE ExistentialQuantification #-}-{-# LANGUAGE DeriveDataTypeable        #-}-{-# LANGUAGE GADTs                     #-}-{-# LANGUAGE MultiWayIf                #-}-{-# LANGUAGE OverloadedStrings         #-}-{-# LANGUAGE RecordWildCards           #-}-{-# LANGUAGE TypeFamilies              #-}-{-# OPTIONS_GHC -fcontext-stack=26     #-}------------------------------------------------------------------------------------ |--- Module : Database.EventStore.Internal.Manager.Subscription--- 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.Internal.Manager.Subscription where-----------------------------------------------------------------------------------import           Control.Concurrent-import           Control.Exception-import           Control.Monad.Fix-import           Data.ByteString (ByteString)-import           Data.ByteString.Lazy (toStrict)-import           Data.Foldable-import           Data.Functor-import           Data.Int-import qualified Data.Map.Strict as M-import           Data.Maybe-import           Data.Monoid ((<>))-import           Data.Typeable-import           GHC.Generics (Generic)-import           Prelude-----------------------------------------------------------------------------------import Data.ProtocolBuffers-import Data.Serialize-import Data.Text hiding (group)-import Data.UUID-import FRP.Sodium-import System.Random-----------------------------------------------------------------------------------import Database.EventStore.Internal.Operation.ReadStreamEventsOperation-import Database.EventStore.Internal.TimeSpan-import Database.EventStore.Internal.Types hiding (Event, newEvent)-import Database.EventStore.Internal.Util.Sodium-----------------------------------------------------------------------------------data SubscribeToStream-    = SubscribeToStream-      { subscribeStreamId       :: Required 1 (Value Text)-      , subscribeResolveLinkTos :: Required 2 (Value Bool)-      }-    deriving (Generic, Show)-----------------------------------------------------------------------------------instance Encode SubscribeToStream-----------------------------------------------------------------------------------subscribeToStream :: Text -> Bool -> SubscribeToStream-subscribeToStream stream_id res_link_tos =-    SubscribeToStream-    { subscribeStreamId       = putField stream_id-    , subscribeResolveLinkTos = putField res_link_tos-    }-----------------------------------------------------------------------------------data SubscriptionConfirmation-    = SubscriptionConfirmation-      { subscribeLastCommitPos   :: Required 1 (Value Int64)-      , subscribeLastEventNumber :: Optional 2 (Value Int32)-      }-    deriving (Generic, Show)-----------------------------------------------------------------------------------instance Decode SubscriptionConfirmation-----------------------------------------------------------------------------------data StreamEventAppeared-    = StreamEventAppeared-      { streamResolvedEvent :: Required 1 (Message ResolvedEventBuf) }-    deriving (Generic, Show)-----------------------------------------------------------------------------------instance Decode StreamEventAppeared------------------------------------------------------------------------------------- | Represents the reason subscription drop happened.-data DropReason-    = D_Unsubscribed-    | D_AccessDenied-    | D_NotFound-    | D_PersistentSubscriptionDeleted-    deriving (Enum, Eq, Show)-----------------------------------------------------------------------------------data SubscriptionDropped-    = SubscriptionDropped-      { dropReason :: Optional 1 (Enumeration DropReason) }-    deriving (Generic, Show)-----------------------------------------------------------------------------------instance Decode SubscriptionDropped-----------------------------------------------------------------------------------data UnsubscribeFromStream = UnsubscribeFromStream deriving (Generic, Show)-----------------------------------------------------------------------------------instance Encode UnsubscribeFromStream-----------------------------------------------------------------------------------data CreatePersistentSubscription =-    CreatePersistentSubscription-    { cpsGroupName         :: Required 1  (Value Text)-    , cpsStreamId          :: Required 2  (Value Text)-    , cpsResolveLinkTos    :: Required 3  (Value Bool)-    , cpsStartFrom         :: Required 4  (Value Int32)-    , cpsMsgTimeout        :: Required 5  (Value Int32)-    , cpsRecordStats       :: Required 6  (Value Bool)-    , cpsLiveBufSize       :: Required 7  (Value Int32)-    , cpsReadBatchSize     :: Required 8  (Value Int32)-    , cpsBufSize           :: Required 9  (Value Int32)-    , cpsMaxRetryCount     :: Required 10 (Value Int32)-    , cpsPreferRoundRobin  :: Required 11 (Value Bool)-    , cpsChkPtAfterTime    :: Required 12 (Value Int32)-    , cpsChkPtMaxCount     :: Required 13 (Value Int32)-    , cpsChkPtMinCount     :: Required 14 (Value Int32)-    , cpsSubMaxCount       :: Required 15 (Value Int32)-    , cpsNamedConsStrategy :: Optional 16 (Value Text)-    } deriving (Generic, Show)-----------------------------------------------------------------------------------_createPersistentSubscription :: Text-                              -> Text-                              -> PersistentSubscriptionSettings-                              -> CreatePersistentSubscription-_createPersistentSubscription group stream sett =-    CreatePersistentSubscription-    { cpsGroupName         = putField group-    , cpsStreamId          = putField stream-    , cpsResolveLinkTos    = putField $ psSettingsResolveLinkTos sett-    , cpsStartFrom         = putField $ psSettingsStartFrom sett-    , cpsMsgTimeout        = putField $ ms $ psSettingsMsgTimeout sett-    , cpsRecordStats       = putField $ psSettingsExtraStats sett-    , cpsLiveBufSize       = putField $ psSettingsLiveBufSize sett-    , cpsReadBatchSize     = putField $ psSettingsReadBatchSize sett-    , cpsBufSize           = putField $ psSettingsHistoryBufSize sett-    , cpsMaxRetryCount     = putField $ psSettingsMaxRetryCount sett-    , cpsPreferRoundRobin  = putField False-    , cpsChkPtAfterTime    = putField $ ms $ psSettingsCheckPointAfter sett-    , cpsChkPtMaxCount     = putField $ psSettingsMaxCheckPointCount sett-    , cpsChkPtMinCount     = putField $ psSettingsMinCheckPointCount sett-    , cpsSubMaxCount       = putField $ psSettingsMaxSubsCount sett-    , cpsNamedConsStrategy = putField $ Just strText-    }-  where-    strText = strategyText $ psSettingsNamedConsumerStrategy sett-    ms      = fromIntegral . timeSpanTotalMillis-----------------------------------------------------------------------------------instance Encode CreatePersistentSubscription-----------------------------------------------------------------------------------data CreatePersistentSubscriptionResult-    = CPS_Success-    | CPS_AlreadyExists-    | CPS_Fail-    | CPS_AccessDenied-    deriving (Enum, Eq, Show)-----------------------------------------------------------------------------------data CreatePersistentSubscriptionCompleted =-    CreatePersistentSubscriptionCompleted-    { cpscResult :: Required 1 (Enumeration CreatePersistentSubscriptionResult)-    , cpscReason :: Optional 2 (Value Text)-    } deriving (Generic, Show)-----------------------------------------------------------------------------------instance Decode CreatePersistentSubscriptionCompleted-----------------------------------------------------------------------------------data DeletePersistentSubscription =-    DeletePersistentSubscription-    { dpsGroupName :: Required 1 (Value Text)-    , dpsStreamId  :: Required 2 (Value Text)-    } deriving (Generic, Show)-----------------------------------------------------------------------------------instance Encode DeletePersistentSubscription-----------------------------------------------------------------------------------_deletePersistentSubscription :: Text -> Text -> DeletePersistentSubscription-_deletePersistentSubscription group_name stream_id =-    DeletePersistentSubscription-    { dpsGroupName = putField group_name-    , dpsStreamId  = putField stream_id-    }-----------------------------------------------------------------------------------data DeletePersistentSubscriptionResult-    = DPS_Success-    | DPS_DoesNotExist-    | DPS_Fail-    | DPS_AccessDenied-    deriving (Enum, Eq, Show)-----------------------------------------------------------------------------------data DeletePersistentSubscriptionCompleted =-    DeletePersistentSubscriptionCompleted-    { dpscResult :: Required 1 (Enumeration DeletePersistentSubscriptionResult)-    , dpscReason :: Optional 2 (Value Text)-    } deriving (Generic, Show)-----------------------------------------------------------------------------------instance Decode DeletePersistentSubscriptionCompleted-----------------------------------------------------------------------------------data UpdatePersistentSubscription =-    UpdatePersistentSubscription-    { upsGroupName         :: Required 1  (Value Text)-    , upsStreamId          :: Required 2  (Value Text)-    , upsResolveLinkTos    :: Required 3  (Value Bool)-    , upsStartFrom         :: Required 4  (Value Int32)-    , upsMsgTimeout        :: Required 5  (Value Int32)-    , upsRecordStats       :: Required 6  (Value Bool)-    , upsLiveBufSize       :: Required 7  (Value Int32)-    , upsReadBatchSize     :: Required 8  (Value Int32)-    , upsBufSize           :: Required 9  (Value Int32)-    , upsMaxRetryCount     :: Required 10 (Value Int32)-    , upsPreferRoundRobin  :: Required 11 (Value Bool)-    , upsChkPtAfterTime    :: Required 12 (Value Int32)-    , upsChkPtMaxCount     :: Required 13 (Value Int32)-    , upsChkPtMinCount     :: Required 14 (Value Int32)-    , upsSubMaxCount       :: Required 15 (Value Int32)-    , upsNamedConsStrategy :: Optional 16 (Value Text)-    } deriving (Generic, Show)-----------------------------------------------------------------------------------_updatePersistentSubscription :: Text-                              -> Text-                              -> PersistentSubscriptionSettings-                              -> UpdatePersistentSubscription-_updatePersistentSubscription group stream sett =-    UpdatePersistentSubscription-    { upsGroupName         = putField group-    , upsStreamId          = putField stream-    , upsResolveLinkTos    = putField $ psSettingsResolveLinkTos sett-    , upsStartFrom         = putField $ psSettingsStartFrom sett-    , upsMsgTimeout        = putField $ ms $ psSettingsMsgTimeout sett-    , upsRecordStats       = putField $ psSettingsExtraStats sett-    , upsLiveBufSize       = putField $ psSettingsLiveBufSize sett-    , upsReadBatchSize     = putField $ psSettingsReadBatchSize sett-    , upsBufSize           = putField $ psSettingsHistoryBufSize sett-    , upsMaxRetryCount     = putField $ psSettingsMaxRetryCount sett-    , upsPreferRoundRobin  = putField False-    , upsChkPtAfterTime    = putField $ ms $ psSettingsCheckPointAfter sett-    , upsChkPtMaxCount     = putField $ psSettingsMaxCheckPointCount sett-    , upsChkPtMinCount     = putField $ psSettingsMinCheckPointCount sett-    , upsSubMaxCount       = putField $ psSettingsMaxSubsCount sett-    , upsNamedConsStrategy = putField $ Just strText-    }-  where-    strText = strategyText $ psSettingsNamedConsumerStrategy sett-    ms      = fromIntegral . timeSpanTotalMillis-----------------------------------------------------------------------------------instance Encode UpdatePersistentSubscription-----------------------------------------------------------------------------------data UpdatePersistentSubscriptionResult-    = UPS_Success-    | UPS_DoesNotExist-    | UPS_Fail-    | UPS_AccessDenied-    deriving (Enum, Eq, Show)-----------------------------------------------------------------------------------data UpdatePersistentSubscriptionCompleted =-    UpdatePersistentSubscriptionCompleted-    { upscResult :: Required 1 (Enumeration UpdatePersistentSubscriptionResult)-    , upscReason :: Optional 2 (Value Text)-    } deriving (Generic, Show)-----------------------------------------------------------------------------------instance Decode UpdatePersistentSubscriptionCompleted-----------------------------------------------------------------------------------data ConnectToPersistentSubscription =-    ConnectToPersistentSubscription-    { ctsId                  :: Required 1 (Value Text)-    , ctsStreamId            :: Required 2 (Value Text)-    , ctsAllowedInFlightMsgs :: Required 3 (Value Int32)-    } deriving (Generic, Show)-----------------------------------------------------------------------------------instance Encode ConnectToPersistentSubscription-----------------------------------------------------------------------------------_connectToPersistentSubscription :: Text-                                 -> Text-                                 -> Int32-                                 -> ConnectToPersistentSubscription-_connectToPersistentSubscription sub_id stream_id all_fly_msgs =-    ConnectToPersistentSubscription-    { ctsId                  = putField sub_id-    , ctsStreamId            = putField stream_id-    , ctsAllowedInFlightMsgs = putField all_fly_msgs-    }-----------------------------------------------------------------------------------data PersistentSubscriptionAckEvents =-    PersistentSubscriptionAckEvents-    { psaeId              :: Required 1 (Value Text)-    , psaeProcessedEvtIds :: Required 2 (Value ByteString)-    } deriving (Generic, Show)-----------------------------------------------------------------------------------instance Encode PersistentSubscriptionAckEvents-----------------------------------------------------------------------------------persistentSubscriptionAckEvents :: Text-                                -> ByteString-                                -> PersistentSubscriptionAckEvents-persistentSubscriptionAckEvents sub_id evt_ids =-    PersistentSubscriptionAckEvents-    { psaeId              = putField sub_id-    , psaeProcessedEvtIds = putField evt_ids-    }-----------------------------------------------------------------------------------data NakAction-    = NA_Unknown-    | NA_Park-    | NA_Retry-    | NA_Skip-    | NA_Stop-    deriving (Enum, Eq, Show)-----------------------------------------------------------------------------------data PersistentSubscriptionNakEvents =-    PersistentSubscriptionNakEvents-    { psneId              :: Required 1 (Value Text)-    , psneProcessedEvtIds :: Required 2 (Value ByteString)-    , psneMsg             :: Optional 3 (Value Text)-    , psneAction          :: Required 4 (Enumeration NakAction)-    } deriving (Generic, Show)-----------------------------------------------------------------------------------instance Encode PersistentSubscriptionNakEvents-----------------------------------------------------------------------------------persistentSubscriptionNakEvents :: Text-                                -> ByteString-                                -> Maybe Text-                                -> NakAction-                                -> PersistentSubscriptionNakEvents-persistentSubscriptionNakEvents sub_id evt_ids msg action =-    PersistentSubscriptionNakEvents-    { psneId              = putField sub_id-    , psneProcessedEvtIds = putField evt_ids-    , psneMsg             = putField msg-    , psneAction          = putField action-    }-----------------------------------------------------------------------------------data PersistentSubscriptionConfirmation =-    PersistentSubscriptionConfirmation-    { pscLastCommitPos :: Required 1 (Value Int64)-    , pscId            :: Required 2 (Value Text)-    , pscLastEvtNumber :: Optional 3 (Value Int32)-    } deriving (Generic, Show)-----------------------------------------------------------------------------------instance Decode PersistentSubscriptionConfirmation-----------------------------------------------------------------------------------data PersistentSubscriptionStreamEventAppeared =-    PersistentSubscriptionStreamEventAppeared-    { psseaEvt :: Required 1 (Message ResolvedIndexedEvent) }-    deriving (Generic, Show)-----------------------------------------------------------------------------------instance Decode PersistentSubscriptionStreamEventAppeared-----------------------------------------------------------------------------------data Sub a where-    RegularSub    :: Text -> Bool -> Sub Regular-    PersistentSub :: Text -> Text -> Int32 -> Sub Persistent-----------------------------------------------------------------------------------data Pending =-    forall s. Push s =>-    Pending-    { _penId  :: !UUID-    , _penSub :: !(Sub s)-    , _penCb  :: Subscription s -> IO ()-    }-----------------------------------------------------------------------------------data Confirmed =-    forall s. Push s =>-    Confirmed-    { _conId  :: !UUID-    , _conTyp :: !(Sub s)-    , _conCB  :: Subscription s -> IO ()-    , _conSub :: !(IO (Subscription s))-    }-----------------------------------------------------------------------------------data OnGoing =-    forall s. Push s =>-    OnGoing-    { _ongTyp :: !(Sub s)-    , _ongSub :: !(Subscription s)-    }------------------------------------------------------------------------------------- | Value's type returned when calling 'subNextEvent'-type family NextEvent a :: * where-    NextEvent Regular    = Either DropReason ResolvedEvent-    NextEvent Persistent = Either DropReason ResolvedEvent-    NextEvent Catchup    = Either CatchupError ResolvedEvent------------------------------------------------------------------------------------- | Represents a subscription to a stream.-data Subscription a =-    Subscription-    { subStreamId :: !Text-      -- ^ The name of the stream to which the subscription is subscribed.-    , subUnsubscribe :: !(IO ())-      -- ^ Asynchronously unsubscribe from the the stream.-    , subNextEvent :: !(IO (NextEvent a))-      -- ^ Awaits for the next event.-    , subIsSubscribedToAll :: !Bool-      -- ^ True if this subscription is to $all stream.-    , _subInternal :: !a-    }------------------------------------------------------------------------------------- | Internal use only because we all know that lawless type-classes are bad.---   But Haskell clearly lacks of a proper module system. So meanwhile, we use---   type-class as a (hacky) way to have a bit of modularity.-class Push a where-    _pushEvt :: a -> Either DropReason ResolvedEvent -> IO ()-----------------------------------------------------------------------------------_subPushEvt :: Push a-            => Subscription a-            -> Either DropReason ResolvedEvent-            -> IO ()-_subPushEvt = _pushEvt . _subInternal------------------------------------------------------------------------------------- | Represents a subscription that is directly identifiable. 'Regular' and---   'Persistent' fit that description while 'Catchup' doesn't. Because---   'Catchup' reads all events from a particular checkpoint and when it's---   finished, it issues a subscription request.-class Identifiable a where-    _getId              :: a -> UUID-    _getLastCommitPos   :: a -> Int64-    _getLastEventNumber :: a -> Maybe Int32------------------------------------------------------------------------------------- | Gets the ID of the subscription.-subId :: Identifiable a => Subscription a -> UUID-subId = _getId . _subInternal------------------------------------------------------------------------------------- | The last commit position seen on the subscription (if this a subscription---   to $all stream).-subLastCommitPos :: Identifiable a => Subscription a -> Int64-subLastCommitPos = _getLastCommitPos . _subInternal------------------------------------------------------------------------------------- | The last event number seen on the subscription (if this is a subscription---   to a single stream).-subLastEventNumber :: Identifiable a => Subscription a -> Maybe Int32-subLastEventNumber = _getLastEventNumber . _subInternal------------------------------------------------------------------------------------- | Represents a subscription to a single stream or $all stream in the---   EventStore.-data Regular =-    Regular-    { _regId              :: !UUID-    , _regResolveLinkTos  :: !Bool-    , _regLastCommitPos   :: !Int64-    , _regLastEventNumber :: !(Maybe Int32)-    , _regChan            :: !(Chan (Either DropReason ResolvedEvent))-    }-----------------------------------------------------------------------------------instance Identifiable Regular where-    _getId              = _regId-    _getLastCommitPos   = _regLastCommitPos-    _getLastEventNumber = _regLastEventNumber-----------------------------------------------------------------------------------instance Push Regular where-    _pushEvt reg = writeChan (_regChan reg)------------------------------------------------------------------------------------- | Determines whether or not any link events encontered in the stream will be---   resolved.-subResolveLinkTos :: Subscription Regular -> Bool-subResolveLinkTos Subscription { _subInternal = reg } = _regResolveLinkTos reg------------------------------------------------------------------------------------- | Errors that could arise during a catch-up subscription. 'Text' value---   represents the stream name.-data CatchupError-    = CatchupStreamDeleted Text-    | CatchupUnexpectedStreamStatus Text ReadStreamResult-    | CatchupSubscriptionDropReason Text DropReason-    deriving (Show, Typeable)-----------------------------------------------------------------------------------instance Exception CatchupError------------------------------------------------------------------------------------- | Represents catch-up subscription.-data Catchup = Catchup { _catchupSub :: MVar (Subscription Regular) }------------------------------------------------------------------------------------- | Waits until 'Catchup' subscription catch-up its stream.-waitTillCatchup :: Subscription Catchup -> IO ()-waitTillCatchup Subscription { _subInternal = Catchup mvar } = do-    _ <- readMVar mvar-    return ()------------------------------------------------------------------------------------- | Non blocking version of `waitTillCatchup`.-hasCaughtUp :: Subscription Catchup -> IO Bool-hasCaughtUp Subscription { _subInternal = Catchup mvar } =-    fmap isJust $ tryReadMVar mvar------------------------------------------------------------------------------------- | Represents a persistent subscription.-data Persistent =-    Persistent-    { _persistId       :: !UUID-    , _persistChan     :: !(Chan (Either DropReason ResolvedEvent))-    , _persistSubId    :: !Text-    , _persistGroup    :: !Text-    , _persistLastCPos :: !Int64-    , _persistLastENum :: !(Maybe Int32)-    , _persistAckCmd   :: AckCmd -> IO ()-    }-----------------------------------------------------------------------------------instance Identifiable Persistent where-    _getId              = _persistId-    _getLastCommitPos   = _persistLastCPos-    _getLastEventNumber = _persistLastENum-----------------------------------------------------------------------------------instance Push Persistent where-    _pushEvt p = writeChan (_persistChan p)------------------------------------------------------------------------------------- | Acknowledges those event ids have been successfully processed.-notifyEventsProcessed :: Subscription Persistent -> [UUID] -> IO ()-notifyEventsProcessed sub eids = _persistAckCmd p (AckCmd eids)-  where-    p = _subInternal sub------------------------------------------------------------------------------------- | Acknowledges those event ids have failed to be processed successfully.-notifyEventsFailed :: Subscription Persistent-                   -> NakAction-                   -> Maybe Text-                   -> [UUID]-                   -> IO ()-notifyEventsFailed sub act msg eids = _persistAckCmd p (NakCmd act msg eids)-  where-    p = _subInternal sub-----------------------------------------------------------------------------------data PersistAction-    = PersistCreate PersistentSubscriptionSettings-    | PersistUpdate PersistentSubscriptionSettings-    | PersistDelete-----------------------------------------------------------------------------------data AckCmd-    = AckCmd [UUID]-    | NakCmd NakAction (Maybe Text) [UUID]-----------------------------------------------------------------------------------data PendingPersistAction =-    PendingPersistAction-    { _ppaId     :: !UUID-    , _ppaGroup  :: !Text-    , _ppaStream :: !Text-    , _ppaTyp    :: !PersistAction-    , _ppaCB     :: Either OperationException () -> IO ()-    }-----------------------------------------------------------------------------------data PersistActionConfirmed =-    PersistActionConfirmed-    { _pacId     :: !UUID-    , _pacResult :: !(Either OperationException ())-    , _pacCB     :: Either OperationException () -> IO ()-    }-----------------------------------------------------------------------------------data Manager-    = Manager-      { _pendings              :: !(M.Map UUID Pending)-      , _ongoings              :: !(M.Map UUID OnGoing)-      , _pendingPersistActions :: !(M.Map UUID PendingPersistAction)-      }-----------------------------------------------------------------------------------initManager :: Manager-initManager =-    Manager-    { _pendings              = M.empty-    , _ongoings              = M.empty-    , _pendingPersistActions = M.empty-    }--------------------------------------------------------------------------------------------------------------------------------------------------------------------maybeDecodeMessage :: Decode a => ByteString -> Maybe a-maybeDecodeMessage bytes =-    case runGet decodeMessage bytes of-        Right a -> Just a-        _       -> Nothing-----------------------------------------------------------------------------------unsafeDecodeMessage :: Decode a => ByteString -> a-unsafeDecodeMessage bytes =-    case runGet decodeMessage bytes of-        Right a -> a-        Left  e -> error $ "decoding error: " ++ e-----------------------------------------------------------------------------------data Appeared =-    forall s. Push s =>-    Appeared-    { _appSub :: !(Subscription s)-    , _appEvt :: !ResolvedEvent-    }-----------------------------------------------------------------------------------onEventAppeared :: Package -> Manager -> Maybe Appeared-onEventAppeared Package{..} Manager{..} =-    case M.lookup packageCorrelation _ongoings of-        Just (OnGoing typ sub) ->-            case (packageCmd, typ) of-                (0xC2, RegularSub _ _ ) ->-                    let msg = unsafeDecodeMessage packageData-                        evt = getField $ streamResolvedEvent msg in-                     Just $ Appeared sub (newResolvedEventFromBuf evt)-                (0xC7, PersistentSub _ _ _) ->-                    let msg = unsafeDecodeMessage packageData-                        evt = getField $ psseaEvt msg in-                    Just $ Appeared sub (newResolvedEvent evt)-                _ -> Nothing-        _ -> Nothing-----------------------------------------------------------------------------------confirmSub :: (UUID -> IO ())-           -> (UUID -> Text -> AckCmd -> IO ())-           -> Package-           -> Manager-           -> Maybe Confirmed-confirmSub unsub ackF Package{..} Manager{..} =-    case M.lookup packageCorrelation _pendings of-        Just (Pending _ typ cb) ->-            case (packageCmd, typ) of-                (0xC1, RegularSub stream tos) ->-                    let !msg = unsafeDecodeMessage packageData-                        lcp  = getField $ subscribeLastCommitPos msg-                        len  = getField $ subscribeLastEventNumber msg in-                    Just $ Confirmed packageCorrelation typ cb $ do-                        chan <- newChan-                        let reg = Regular-                                  { _regId              = packageCorrelation-                                  , _regResolveLinkTos  = tos-                                  , _regLastCommitPos   = lcp-                                  , _regLastEventNumber = len-                                  , _regChan            = chan-                                  }--                        return Subscription-                               { subStreamId          = stream-                               , subUnsubscribe       = unsub packageCorrelation-                               , subNextEvent         = readChan chan-                               , subIsSubscribedToAll = stream == ""-                               , _subInternal         = reg-                               }-                (0xC6, PersistentSub grp stream _) ->-                    let !msg = unsafeDecodeMessage packageData-                        lcp  = getField $ pscLastCommitPos msg-                        sid  = getField $ pscId msg-                        len  = getField $ pscLastEvtNumber msg in-                    Just $ Confirmed packageCorrelation typ cb $ do-                        chan <- newChan-                        let pes = Persistent-                                  { _persistId       = packageCorrelation-                                  , _persistChan     = chan-                                  , _persistSubId    = sid-                                  , _persistGroup    = grp-                                  , _persistLastCPos = lcp-                                  , _persistLastENum = len-                                  , _persistAckCmd   = \cmd ->-                                    ackF packageCorrelation sid cmd-                                  }--                        return Subscription-                               { subStreamId          = stream-                               , subUnsubscribe       = unsub packageCorrelation-                               , subNextEvent         = readChan chan-                               , subIsSubscribedToAll = False-                               , _subInternal         = pes-                               }-                _ -> Nothing-        _ -> Nothing------------------------------------------------------------------------------------- Events----------------------------------------------------------------------------------data Subscribe-    = Subscribe-      { _subId             :: !UUID-      , _subCallback       :: Subscription Regular -> IO ()-      , _subStream         :: !Text-      , _subResolveLinkTos :: !Bool-      }-----------------------------------------------------------------------------------data RegisterSub = forall s. Push s => RegisterSub UUID (Sub s) (Subscription s)------------------------------------------------------------------------------------- Commands----------------------------------------------------------------------------------data SubCommand-    = forall s. Push s => SubscribeTo (Sub s) (Subscription s -> IO ())-    | SubmitPersistAction Text-                          Text-                          PersistAction-                          (Either OperationException () -> IO ())-----------------------------------------------------------------------------------subscriptionNetwork :: Settings-                    -> (Package -> Reactive ())-                    -> Event Package-                    -> Reactive (SubCommand -> IO ())-subscriptionNetwork sett push_pkg e_pkg = do-    -- When a subscription request has been submitted by the user.-    (on_sub, push_sub) <- newEvent--    -- When a subscription has been confirmed by EventStore and we succesfully-    -- create a `forall s. Push s => Subscription s` object.-    (on_reg_sub, push_reg_sub) <- newEvent--    -- When a persist action has been emitted by the user.-    (on_persist_action, push_persist_action) <- newEvent--    let push_pkg_io = pushAsync push_pkg-        push_ack_cmd uuid sid cmd =-            push_pkg_io $ createAckCmdPackage sett uuid sid cmd-    mgr_b <- mfix $ \mgr_b -> do-        let send_unsub = push_pkg_io . createUnsubscribePackage sett--            on_con_sub = filterJust $ snapshot (confirmSub send_unsub-                                                           push_ack_cmd)-                                               e_pkg mgr_b--            on_drop = filterJust $ snapshot dropError e_pkg mgr_b--            on_persist_action_cfrm =-                filterJust $ snapshot onPersistActionConfirmed e_pkg mgr_b--            mgr_e = fmap confirmed on_reg_sub               <>-                    fmap subscribeRequest on_sub            <>-                    fmap newPersistAction on_persist_action <>-                    fmap dropped on_drop                    <>-                    fmap persistActionConfirmed on_persist_action_cfrm--        _ <- listen on_drop $ \(Dropped reason sub _) ->-          _subPushEvt sub (Left reason)--        _ <- listen on_persist_action_cfrm $ \(PersistActionConfirmed _ res k) ->-          k res--        _ <- listen on_con_sub $ \(Confirmed uuid typ cb action) -> do-          sub <- action-          _   <- forkIO $ sync $ push_reg_sub (RegisterSub uuid typ sub)-          cb sub--        accum initManager mgr_e--    let on_app  = filterJust $ snapshot onEventAppeared e_pkg mgr_b---        runSubCommand (SubscribeTo typ cb) = do-            uuid <- randomIO-            let sub = Pending-                      { _penId  = uuid-                      , _penSub = typ-                      , _penCb  = cb-                      }-            void $ forkIO $ sync $ push_sub sub-        runSubCommand (SubmitPersistAction group stream typ cb) = do-            uuid <- randomIO-            let action = PendingPersistAction-                         { _ppaId     = uuid-                         , _ppaGroup  = group-                         , _ppaStream = stream-                         , _ppaTyp    = typ-                         , _ppaCB     = cb-                         }-            void $ forkIO $ sync $ push_persist_action action-    _ <- listen on_sub (push_pkg_io . createSubscriptionPackage sett)--    _ <- listen on_persist_action (push_pkg_io . createPersistActionPkg sett)--    _ <- listen on_app $ \(Appeared sub evt) ->-        _subPushEvt sub (Right evt)---    return runSubCommand-----------------------------------------------------------------------------------createSubscriptionPackage :: Settings -> Pending -> Package-createSubscriptionPackage sett (Pending uuid typ _) =-    case typ of-        RegularSub stream tos ->-            createConnectRegularPackage sett uuid stream tos-        PersistentSub grp str bufSize ->-            createConnectPersistPackage sett uuid grp str bufSize-----------------------------------------------------------------------------------createConnectRegularPackage :: Settings -> UUID -> Text -> Bool -> Package-createConnectRegularPackage Settings{..} uuid stream tos =-    Package-    { packageCmd         = 0xC0-    , packageCorrelation = uuid-    , packageData        = runPut $ encodeMessage msg-    , packageCred        = s_credentials-    }-  where-    msg = subscribeToStream stream tos-----------------------------------------------------------------------------------createConnectPersistPackage :: Settings-                            -> UUID-                            -> Text-                            -> Text-                            -> Int32-                            -> Package-createConnectPersistPackage Settings{..} uuid group stream bufSize =-    Package-    { packageCmd         = 0xC5-    , packageCorrelation = uuid-    , packageData        = runPut $ encodeMessage msg-    , packageCred        = s_credentials-    }-  where-    msg = _connectToPersistentSubscription group stream bufSize-----------------------------------------------------------------------------------createAckCmdPackage :: Settings -> UUID -> Text -> AckCmd -> Package-createAckCmdPackage sett uuid sid cmd =-    case cmd of-        AckCmd eids         -> createAckPackage sett uuid sid eids-        NakCmd act msg eids -> createNakPackage sett uuid sid act msg eids-----------------------------------------------------------------------------------createAckPackage :: Settings -> UUID -> Text -> [UUID] -> Package-createAckPackage Settings{..} corr sid eids =-    Package-    { packageCmd         = 0xCC-    , packageCorrelation = corr-    , packageData        = runPut $ encodeMessage msg-    , packageCred        = s_credentials-    }-  where-    bytes = toStrict $ foldMap toByteString eids-    msg   = persistentSubscriptionAckEvents sid bytes-----------------------------------------------------------------------------------createNakPackage :: Settings-                 -> UUID-                 -> Text-                 -> NakAction-                 -> Maybe Text-                 -> [UUID]-                 -> Package-createNakPackage Settings{..} corr sid act txt eids =-    Package-    { packageCmd         = 0xCD-    , packageCorrelation = corr-    , packageData        = runPut $ encodeMessage msg-    , packageCred        = s_credentials-    }-  where-    bytes = toStrict $ foldMap toByteString eids-    msg   = persistentSubscriptionNakEvents sid bytes txt act-----------------------------------------------------------------------------------createUnsubscribePackage :: Settings -> UUID -> Package-createUnsubscribePackage Settings{..} uuid =-    Package-    { packageCmd         = 0xC3-    , packageCorrelation = uuid-    , packageData        = runPut $ encodeMessage UnsubscribeFromStream-    , packageCred        = s_credentials-    }-----------------------------------------------------------------------------------createPersistActionPkg :: Settings -> PendingPersistAction -> Package-createPersistActionPkg Settings{..} (PendingPersistAction aId grp strm typ _) =-    Package-    { packageCmd         = cmd-    , packageCorrelation = aId-    , packageData        = runPut msg-    , packageCred        = s_credentials-    }-  where-    msg =-        case typ of-            PersistCreate sett ->-                encodeMessage $ _createPersistentSubscription grp strm sett-            PersistUpdate sett ->-                encodeMessage $ _updatePersistentSubscription grp strm sett-            PersistDelete ->-                encodeMessage $ _deletePersistentSubscription grp strm-    cmd =-        case typ of-            PersistCreate _  -> 0xC8-            PersistUpdate _  -> 0xCE-            PersistDelete    -> 0xCA-----------------------------------------------------------------------------------data Dropped =-    forall s. Push s =>-    Dropped-    { droppedReason :: !DropReason-    , droppedSub    :: !(Subscription s)-    , droppedId     :: !UUID-    }-----------------------------------------------------------------------------------dropError :: Package -> Manager -> Maybe Dropped-dropError Package{..} Manager{..}-    | packageCmd == 0xC4 = do-         OnGoing _ sub <- M.lookup packageCorrelation _ongoings-         msg           <- maybeDecodeMessage packageData-         let reason = fromMaybe D_Unsubscribed $ getField $ dropReason msg--         return Dropped-                { droppedReason = reason-                , droppedSub    = sub-                , droppedId     = packageCorrelation-                }-    | otherwise = Nothing-----------------------------------------------------------------------------------nonEmptyText :: Text -> Maybe Text-nonEmptyText "" = Nothing-nonEmptyText t  = Just t-----------------------------------------------------------------------------------onPersistActionConfirmed :: Package -> Manager -> Maybe PersistActionConfirmed-onPersistActionConfirmed Package{..} Manager{..} =-    case M.lookup packageCorrelation _pendingPersistActions of-        Just (PendingPersistAction _ grp stream typ cb) ->-            case (packageCmd, typ) of-                (0xC9, PersistCreate _) -> do-                    msg <- maybeDecodeMessage packageData-                    let res    = getField $ cpscResult msg-                        reason = nonEmptyText =<< getField (cpscReason msg)-                        ret =-                            case res of-                                CPS_Success ->-                                    Right ()-                                CPS_Fail ->-                                    Left $ peristentCreationFailure grp-                                                                    stream-                                                                    reason-                                CPS_AlreadyExists ->-                                    Left $ persistentCreationExists grp-                                                                    stream-                                CPS_AccessDenied ->-                                    Left $ persistentAccessDenied stream-                        pac = PersistActionConfirmed-                              { _pacId     = packageCorrelation-                              , _pacResult = ret-                              , _pacCB     = cb-                              }-                    return pac-                (0xCF, PersistUpdate _) -> do-                    msg <- maybeDecodeMessage packageData-                    let res    = getField $ upscResult msg-                        reason = nonEmptyText =<< getField (upscReason msg)-                        ret =-                            case res of-                                UPS_Success ->-                                    Right ()-                                UPS_Fail ->-                                    Left $ peristentCreationFailure grp-                                                                    stream-                                                                    reason-                                UPS_DoesNotExist ->-                                    Left $ persistentDoesNotExist grp-                                                                  stream-                                UPS_AccessDenied ->-                                    Left $ persistentAccessDenied stream-                        pac = PersistActionConfirmed-                              { _pacId     = packageCorrelation-                              , _pacResult = ret-                              , _pacCB     = cb-                              }-                    return pac-                (0xCB, PersistDelete) -> do-                    msg <- maybeDecodeMessage packageData-                    let res    = getField $ dpscResult msg-                        reason = nonEmptyText =<< getField (dpscReason msg)-                        ret =-                            case res of-                                DPS_Success ->-                                    Right ()-                                DPS_Fail ->-                                    Left $ peristentCreationFailure grp-                                                                    stream-                                                                    reason-                                DPS_DoesNotExist ->-                                    Left $ persistentDoesNotExist grp-                                                                  stream-                                DPS_AccessDenied ->-                                    Left $ persistentAccessDenied stream-                        pac = PersistActionConfirmed-                              { _pacId     = packageCorrelation-                              , _pacResult = ret-                              , _pacCB     = cb-                              }-                    return pac-                _ -> Nothing-        _ -> Nothing------------------------------------------------------------------------------------- Model----------------------------------------------------------------------------------persistActionConfirmed :: PersistActionConfirmed -> Manager -> Manager-persistActionConfirmed pc s@Manager{..} =-    s { _pendingPersistActions = M.delete (_pacId pc) _pendingPersistActions }-----------------------------------------------------------------------------------subscribeRequest :: Pending -> Manager -> Manager-subscribeRequest p@(Pending uuid _ _) s@Manager{..} =-    s { _pendings = M.insert uuid p _pendings }-----------------------------------------------------------------------------------dropped :: Dropped -> Manager -> Manager-dropped d s@Manager{..} = s { _ongoings = M.delete (droppedId d) _ongoings }-----------------------------------------------------------------------------------confirmed :: RegisterSub -> Manager -> Manager-confirmed (RegisterSub uuid typ sub) s@Manager{..} =-    s { _pendings = M.delete uuid _pendings-      , _ongoings = M.insert uuid (OnGoing typ sub) _ongoings-      }-----------------------------------------------------------------------------------newPersistAction :: PendingPersistAction -> Manager -> Manager-newPersistAction ppa@PendingPersistAction{..} s@Manager{..} =-    s { _pendingPersistActions = M.insert _ppaId ppa _pendingPersistActions }--------------------------------------------------------------------------------------------------------------------------------------------------------------------peristentCreationFailure :: Text-                         -> Text-                         -> Maybe Text-                         -> OperationException-peristentCreationFailure group stream m_reason = InvalidOperation msg-  where-    msg = "Subscription group " <> group <> " on stream " <> stream <>-          " failed" <> reasonTxt--    reasonTxt = foldMap (" reason: " <>) m_reason-----------------------------------------------------------------------------------persistentCreationExists :: Text -> Text -> OperationException-persistentCreationExists group stream = InvalidOperation msg-  where-    msg = "Subscription group " <> group <> " on stream " <> stream <>-          " already exists."-----------------------------------------------------------------------------------persistentDoesNotExist :: Text -> Text -> OperationException-persistentDoesNotExist group stream = InvalidOperation msg-  where-    msg = "Subscription group " <> group <> " on stream " <> stream <>-          " doesn't exist."-----------------------------------------------------------------------------------persistentAccessDenied :: Text -> OperationException-persistentAccessDenied stream = AccessDenied msg-  where-    msg = "Write access denied for stream " <> stream+{-# LANGUAGE GADTs               #-}+{-# LANGUAGE Rank2Types          #-}+{-# LANGUAGE ScopedTypeVariables #-}+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Internal.Manager.Subscription+-- Copyright : (C) 2015 Yorick Laupa+-- License : (see the file LICENSE)+--+-- Maintainer : Yorick Laupa <yo.eight@gmail.com>+-- Stability : provisional+-- Portability : non-portable+--+-- Main subscription state machine declaration module. It also declares every+-- functions required to drive a 'Subscription'.+--------------------------------------------------------------------------------+module Database.EventStore.Internal.Manager.Subscription+    ( module Database.EventStore.Internal.Manager.Subscription.Driver+    , Regular(..)+    , Persistent(..)+    , Catchup(..)+    , Subscription+    , Running(..)+    , Checkpoint(..)+    , regularSubscription+    , catchupSubscription+    , persistentSubscription+    , eventArrived+    , readNext+    , batchRead+    , hasCaughtUp+    , runningUUID+    , runningLastEventNumber+    , runningLastCommitPosition+    ) where++--------------------------------------------------------------------------------+import Data.Int++--------------------------------------------------------------------------------+import Data.Sequence+import Data.Text (Text)++--------------------------------------------------------------------------------+import Database.EventStore.Internal.Manager.Subscription.Driver+import Database.EventStore.Internal.Manager.Subscription.Model+import Database.EventStore.Internal.Types++--------------------------------------------------------------------------------+-- | Also referred as volatile subscription. For example, if a stream has 100+--   events in it when a subscriber connects, the subscriber can expect to see+--   event number 101 onwards until the time the subscription is closed or+--   dropped.+data Regular = Regular { _subTos :: Bool }++--------------------------------------------------------------------------------+-- | This kind of subscription specifies a starting point, in the form of an+--   event number or transaction file position. The given function will be+--   called for events from the starting point until the end of the stream, and+--   then for subsequently written events.+--+--   For example, if a starting point of 50 is specified when a stream has 100+--   events in it, the subscriber can expect to see events 51 through 100, and+--   then any events subsequently written until such time as the subscription is+--   dropped or closed.+data Catchup = Catchup++--------------------------------------------------------------------------------+-- | The server remembers the state of the subscription. This allows for many+--   different modes of operations compared to a regular or catchup subscription+--   where the client holds the subscription state.+--   (Need EventStore >= v3.1.0).+data Persistent = Persistent { _perGroup  :: Text }++--------------------------------------------------------------------------------+-- | Represents the different type of inputs a subscription state-machine can+--   handle.+data Input t a where+    -- A event has written to the stream. Subscription state machine should+    -- store that event withing its state.+    Arrived :: ResolvedEvent -> Input t (Subscription t)+    -- The user asks for the next event coming from the server.+    ReadNext :: Input t (Maybe ResolvedEvent, Subscription t)+    -- A batch read has been made. It's only use for 'Catchup' subscription+    -- type. It gives the list of read events and indicates if it reaches the+    -- end of the stream along with the next checkpoint to point at.+    BatchRead :: [ResolvedEvent]+              -> Bool+              -> Checkpoint+              -> Input Catchup (Subscription Catchup)+    -- Used only for 'Catchup' subscription type. Asks if the subscription+    -- read every events up to the checkpoint given by the user.+    CaughtUp :: Input Catchup Bool++--------------------------------------------------------------------------------+-- | Main subscription state machine.+newtype Subscription t = Subscription (forall a. Input t a -> a)++--------------------------------------------------------------------------------+-- | Submit a new event to the subscription state machine. Internally,+--   that event should be stored into the subscription buffer.+eventArrived :: ResolvedEvent -> Subscription t -> Subscription t+eventArrived e (Subscription k) = k (Arrived e)++--------------------------------------------------------------------------------+-- | Reads the next available event. Returns 'Nothing' it there is any. When+--   returning an event, it will be removed from the subscription buffer.+readNext :: Subscription t -> (Maybe ResolvedEvent, Subscription t)+readNext (Subscription k) = k ReadNext++--------------------------------------------------------------------------------+-- | Submits a list of events read from a stream. It's only used by a 'Catchup'+--   subscription.+batchRead :: [ResolvedEvent]+          -> Bool -- ^ If it reaches the end of the stream.+          -> Checkpoint+          -> Subscription Catchup+          -> Subscription Catchup+batchRead es eos nxt (Subscription k) = k (BatchRead es eos nxt)++--------------------------------------------------------------------------------+-- | Indicates if the subscription caught up the end of the stream, meaning the+--   subscription is actually live. Only used by 'Catchup' subscription.+hasCaughtUp :: Subscription Catchup -> Bool+hasCaughtUp (Subscription k) = k CaughtUp++--------------------------------------------------------------------------------+-- | Main 'Regular' subscription state machine.+regularSubscription :: Subscription Regular+regularSubscription = baseSubscription++--------------------------------------------------------------------------------+-- | Main 'Persistent' subscription state machine.+persistentSubscription :: Subscription Persistent+persistentSubscription = baseSubscription++--------------------------------------------------------------------------------+-- | Represents the next checkpoint to reach on a catchup subscription. Wheither+--   it's a regular stream or the $all stream, it either point to an 'Int32' or+--   a 'Position'.+data Checkpoint = CheckpointNumber Int32 | CheckpointPosition Position++--------------------------------------------------------------------------------+-- | Depending either if the subscription concerns a regular stream or $all,+--  indicates if an event number (or 'Position') is lesser that the current the+--  given 'CheckPoint'.+beforeChk :: Checkpoint -> ResolvedEvent -> Bool+beforeChk (CheckpointNumber num) re =+    recordedEventNumber (resolvedEventOriginal re) < num+beforeChk (CheckpointPosition pos) re =+    maybe False (< pos) $ resolvedEventPosition re++--------------------------------------------------------------------------------+-- | That subscription state machine accumulates events coming from batch read+--   and any real time change made on a stream. That state machine will not+--   served any recent change made on the stream until it reaches the end of the+--   stream. On every batch read, it makes sure events contained in that batch+--   are deleted from the subscription buffer in order to avoid duplicates. That+--   implemention has been chosen to avoid potential message lost between the+--   moment with reach the end of the stream and the delay required by asking+--   for a subscription.+catchupSubscription :: Subscription Catchup+catchupSubscription = Subscription $ catchingUp empty empty+  where+    catchingUp :: forall a. Seq ResolvedEvent+               -> Seq ResolvedEvent+               -> Input Catchup a+               -> a+    catchingUp b s (Arrived e) =+        Subscription $ catchingUp b (s |> e)+    catchingUp b s ReadNext =+        case viewl b of+            EmptyL    -> (Nothing, Subscription $ catchingUp b s)+            e :< rest -> (Just e, Subscription $ catchingUp rest s)+    catchingUp b s (BatchRead es eos nxt_pt) =+        let nxt_b = foldl (|>) b es+            nxt_s = dropWhileL (beforeChk nxt_pt) s+            nxt   = if eos+                    then Subscription $ caughtUp nxt_b nxt_s+                    else Subscription $ catchingUp nxt_b nxt_s in+        nxt+    catchingUp _ _ CaughtUp = False++    caughtUp :: forall a. Seq ResolvedEvent+             -> Seq ResolvedEvent+             -> Input Catchup a+             -> a+    caughtUp  b s (Arrived e) = Subscription $ caughtUp b (s |> e)+    caughtUp b s  ReadNext =+        case viewl b of+            EmptyL -> live s ReadNext+            e :< rest ->+                case viewl rest of+                    EmptyL -> (Just e, Subscription $ live s)+                    _      -> (Just e, Subscription $ caughtUp rest s)+    caughtUp b s (BatchRead _ _ _) = Subscription $ caughtUp b s+    caughtUp _ _ CaughtUp = False++    live :: forall a. Seq ResolvedEvent -> Input Catchup a -> a+    live s (Arrived e) = Subscription $ live (s |> e)+    live s ReadNext =+        case viewl s of+            EmptyL    -> (Nothing, Subscription $ live s)+            e :< rest -> (Just e, Subscription $ live rest)+    live s (BatchRead _ _ _) = Subscription $ live s+    live _ CaughtUp = True++--------------------------------------------------------------------------------+-- | Base subscription used for 'Regular' or 'Persistent' subscription.+baseSubscription :: forall t. Subscription t+baseSubscription = Subscription $ go empty+  where+    go :: forall a. Seq ResolvedEvent -> Input t a -> a+    go s (Arrived e) = Subscription $ go (s |> e)+    go s ReadNext =+        case viewl s of+            EmptyL    -> (Nothing, Subscription $ go s)+            e :< rest -> (Just e, Subscription $ go rest)+    go _ BatchRead{} = error "impossible: base subscription"+    go _ CaughtUp    = error "impossible: base subscription"

@@ -0,0 +1,481 @@+{-# LANGUAGE DeriveDataTypeable  #-}+{-# LANGUAGE GADTs               #-}+{-# LANGUAGE Rank2Types          #-}+{-# LANGUAGE RecordWildCards     #-}+{-# LANGUAGE ScopedTypeVariables #-}+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Internal.Manager.Subscription.Driver+-- Copyright : (C) 2015 Yorick Laupa+-- License : (see the file LICENSE)+--+-- Maintainer : Yorick Laupa <yo.eight@gmail.com>+-- Stability : provisional+-- Portability : non-portable+--+-- Subscription model driver. It drivers the model accordingly depending on the+-- 'Package' or commands submitted to it.+--------------------------------------------------------------------------------+module Database.EventStore.Internal.Manager.Subscription.Driver+    ( SubDropReason(..)+    , SubConnectEvent(..)+    , PersistActionException(..)+    , ConfirmedAction(..)+    , NakAction(..)+    , Driver+    , newDriver+    , submitPackage+    , connectToStream+    , connectToPersist+    , createPersist+    , updatePersist+    , deletePersist+    , ackPersist+    , nakPersist+    , unsubscribe+    , abort+    ) where++--------------------------------------------------------------------------------+import Control.Exception+import Data.Int+import Data.Maybe+import Data.Typeable++--------------------------------------------------------------------------------+import           Data.ByteString+import qualified Data.HashMap.Strict as H+import           Data.Serialize+import           Data.ProtocolBuffers+import           Data.Text+import           Data.UUID++--------------------------------------------------------------------------------+import Database.EventStore.Internal.Generator+import Database.EventStore.Internal.Manager.Subscription.Message+import Database.EventStore.Internal.Manager.Subscription.Model+import Database.EventStore.Internal.Manager.Subscription.Packages+import Database.EventStore.Internal.Types++--------------------------------------------------------------------------------+-- | Set of events that can occurs during a subscription lifetime.+data SubConnectEvent+    = EventAppeared ResolvedEvent+      -- ^ A wild event appeared !+    | Dropped SubDropReason+      -- ^ The subscription connection dropped.+    | SubConfirmed Running+      -- ^ Subscription connection is confirmed. It means that subscription can+      --   receive events from the server.++--------------------------------------------------------------------------------+-- | Indicates why a subscription has been dropped.+data SubDropReason+    = SubUnsubscribed+      -- ^ Subscription connection has been closed by the user.+    | SubAccessDenied+      -- ^ The current user is not allowed to operate on the supplied stream.+    | SubNotFound+      -- ^ Given stream name doesn't exist.+    | SubPersistDeleted+      -- ^ Given stream is deleted.+    | SubAborted+      -- ^ Occurs when the user shutdown the connection from the server or if+      -- the connection to the server is no longer possible.+    deriving (Show, Eq)++--------------------------------------------------------------------------------+-- | Enumerates all persistent action exceptions.+data PersistActionException+    = PersistActionFail+      -- ^ The action failed.+    | PersistActionAlreadyExist+      -- ^ Happens when creating a persistent subscription on a stream with a+      --   group name already taken.+    | PersistActionDoesNotExist+      -- ^ An operation tried to do something on a persistent subscription or a+      --   stream that don't exist.+    | PersistActionAccessDenied+      -- ^ The current user is not allowed to operate on the supplied stream or+      --   persistent subscription.+    | PersistActionAborted+      -- ^ That action has been aborted because the user shutdown the connection+      --   to the server or the connection to the server is no longer possible.+    deriving (Show, Typeable)++--------------------------------------------------------------------------------+instance Exception PersistActionException++--------------------------------------------------------------------------------+-- | Emitted when a persistent action has been carried out successfully.+data ConfirmedAction =+    ConfirmedAction+    { caId     :: !UUID+      -- ^ Action id.+    , caGroup  :: !Text+      -- ^ Subscription group name.+    , caStream :: !Text+      -- ^ Stream name.+    , caAction :: !PersistAction+      -- ^ Persistent action type.+    }++--------------------------------------------------------------------------------+-- | Submits a 'Package' to a subscription driver. If the 'Package' was+--   processed by the driver, it will return a final value and a new driver with+--   its internal state updated accordingly.+submitPackage :: Package -> Driver r -> Maybe (r, Driver r)+submitPackage pkg (Driver k) = k (Pkg pkg)++--------------------------------------------------------------------------------+-- | Starts a regular subscription connection. It returns the associated+--   'Package' and updates driver internal state.+connectToStream :: (SubConnectEvent -> r)+                -> Text -- ^ Stream name.+                -> Bool -- ^ Resolve Link TOS+                -> Driver r+                -> (Package, Driver r)+connectToStream c s t (Driver k) = k (Cmd $ ConnectReg c s t)++--------------------------------------------------------------------------------+-- | Starts a persistent subscription connection. It returns the associated+--   'Package' and updates driver internal state.+connectToPersist :: (SubConnectEvent -> r)+                 -> Text  -- ^ Group name.+                 -> Text  -- ^ Stream name.+                 -> Int32 -- ^ Buffer size.+                 -> Driver r+                 -> (Package, Driver r)+connectToPersist c g s b (Driver k) = k (Cmd $ ConnectPersist c g s b)++--------------------------------------------------------------------------------+-- | Creates a persistent subscription. It returns the associated 'Package' and+--   updates driver internal state.+createPersist :: (Either PersistActionException ConfirmedAction -> r)+              -> Text -- ^ Group name.+              -> Text -- ^ Stream name.+              -> PersistentSubscriptionSettings+              -> Driver r+              -> (Package, Driver r)+createPersist c g s ss (Driver k) =+    k (Cmd $ ApplyPersistAction c g s (PersistCreate ss))++--------------------------------------------------------------------------------+-- | Updates a persistent subscription. It returns the associated 'Package' and+--   updates driver internal state.+updatePersist :: (Either PersistActionException ConfirmedAction -> r)+              -> Text -- ^ Group name.+              -> Text -- ^ Stream name.+              -> PersistentSubscriptionSettings+              -> Driver r+              -> (Package, Driver r)+updatePersist c g s ss (Driver k) =+    k (Cmd $ ApplyPersistAction c g s (PersistUpdate ss))++--------------------------------------------------------------------------------+-- | Deletes a persistent subscription. It returns the associated 'Package' and+-- updates driver internal state.+deletePersist :: (Either PersistActionException ConfirmedAction -> r)+              -> Text -- ^ Group name.+              -> Text -- ^ Stream name.+              -> Driver r+              -> (Package, Driver r)+deletePersist c g s (Driver k) =+    k (Cmd $ ApplyPersistAction c g s PersistDelete)++--------------------------------------------------------------------------------+-- | Given a persistent subscription, acknowledges a set of events have been+--   successfully processed. It returns the associated 'Package' and updates+--   driver internal state.+ackPersist :: r+           -> Running+           -> [UUID] -- ^ Event ids.+           -> Driver r+           -> (Package, Driver r)+ackPersist r i evts (Driver k) = k (Cmd $ PersistAck r i evts)++--------------------------------------------------------------------------------+-- | Given a persistent subscription, indicates a set of event haven't been+--   processed correctly. It returns the associated 'Package' and updates driver+--   internal state.+nakPersist :: r+           -> Running+           -> NakAction+           -> Maybe Text -- ^ Reason.+           -> [UUID]     -- ^ Event ids.+           -> Driver r+           -> (Package, Driver r)+nakPersist r i na mt evts (Driver k) =+    k (Cmd $ PersistNak r i na mt evts)++--------------------------------------------------------------------------------+-- | Unsubscribe from a subscription.+unsubscribe :: Running -> Driver r -> (Package, Driver r)+unsubscribe r (Driver k) = k (Cmd $ Unsubscribe r)++--------------------------------------------------------------------------------+-- | Aborts every pending action.+abort :: Driver r -> [r]+abort (Driver k) = k Abort++--------------------------------------------------------------------------------+-- EventStore result mappers:+-- =========================+-- EventStore protocol has several values that means the exact same thing. Those+-- functions convert a specific EventStore to uniform result type common to all+-- persistent actions.+--------------------------------------------------------------------------------+createRException :: CreatePersistentSubscriptionResult+                 -> Maybe PersistActionException+createRException CPS_Success       = Nothing+createRException CPS_AlreadyExists = Just PersistActionAlreadyExist+createRException CPS_Fail          = Just PersistActionFail+createRException CPS_AccessDenied  = Just PersistActionAccessDenied++--------------------------------------------------------------------------------+deleteRException :: DeletePersistentSubscriptionResult+                 -> Maybe PersistActionException+deleteRException DPS_Success      = Nothing+deleteRException DPS_DoesNotExist = Just PersistActionDoesNotExist+deleteRException DPS_Fail         = Just PersistActionFail+deleteRException DPS_AccessDenied = Just PersistActionAccessDenied++--------------------------------------------------------------------------------+updateRException :: UpdatePersistentSubscriptionResult+                 -> Maybe PersistActionException+updateRException UPS_Success      = Nothing+updateRException UPS_DoesNotExist = Just PersistActionDoesNotExist+updateRException UPS_Fail         = Just PersistActionFail+updateRException UPS_AccessDenied = Just PersistActionAccessDenied++--------------------------------------------------------------------------------+toSubDropReason :: DropReason -> SubDropReason+toSubDropReason D_Unsubscribed                  = SubUnsubscribed+toSubDropReason D_NotFound                      = SubNotFound+toSubDropReason D_AccessDenied                  = SubAccessDenied+toSubDropReason D_PersistentSubscriptionDeleted = SubPersistDeleted++--------------------------------------------------------------------------------+--------------------------------------------------------------------------------+-- | Type of inputs handled by the 'Subscription' driver.+data In r a where+    -- A command consists of receiving some parameters, updating the+    -- 'Subscription' model accordingly and thus modifying driver internal+    -- state.+    Cmd :: Cmd r -> In r (Package, Driver r)+    -- A 'Package' has been submitted to the 'Subscription' driver. If the+    -- driver recognize that 'Package', it returns a final value and update+    -- the driver internal state.+    Pkg :: Package -> In r (Maybe (r, Driver r))+    -- Aborts every pending action.+    Abort :: In r [r]++--------------------------------------------------------------------------------+-- | Set of commands handled by the driver.+data Cmd r+    = ConnectReg (SubConnectEvent -> r) Text Bool+      -- ^ Creates a regular 'Subscription' connection. When a 'SubConnectEvent'+      --   has arrived, the driver will use the provided callback and emit a+      --   final value. It holds a stream name and `Resolve Link TOS` setting.+    | ConnectPersist (SubConnectEvent -> r)+                     Text+                     Text+                     Int32+      -- ^ Creates a persistent 'Subscription' connection. When a+      --   'SubConnectEvent' has arrived, the driver will use the provided+      --   callback  and emit a final value. It holds a group name, stream+      --   name and a buffer size.++    | Unsubscribe Running+      -- ^ Unsubscribe from a subscription.++    | ApplyPersistAction (Either PersistActionException ConfirmedAction -> r)+                         Text+                         Text+                         PersistAction+      -- ^ Creates a persistent action. Depending of the failure or the success+      --   of that action, the driver will use the provided callback to emit a+      --   final value. It hols a group name, a stream name and a persistent+      --   action.++    | PersistAck r Running [UUID]+      -- ^ Acks a set of Event 'UUID' to notify those events have been correctly+      --   handled. It holds a 'Running' subscription and a set of `UUID`+      --   representing event id. When the ack would be confirmed the driver+      --   will return the supplied final value.+    | PersistNak r+                 Running+                 NakAction+                 (Maybe Text)+                 [UUID]+      -- ^ Naks a set of Event 'UUID' to notify those events haven't been+      --   handled correctly. it holds a 'Running' subscription, a 'NakAction',+      --   an optional reason and a set of event ids. When the nak would be+      --   confirmed, the driver will return the provided final value.++--------------------------------------------------------------------------------+-- | Driver internal state.+data State r =+    State+    { _model :: !Model+      -- ^ Subscription model.+    , _gen :: !Generator+      -- ^ 'UUID' generator.+    , _reg :: !(H.HashMap UUID (Cmd r))+      -- ^ Holds ongoing commands. When stored, it means an action hasn't been+      --   confirmed yet.+    }++--------------------------------------------------------------------------------+initState :: Generator -> State r+initState gen = State newModel gen H.empty++--------------------------------------------------------------------------------+-- | Subscription driver state machine.+newtype Driver r = Driver (forall a. In r a -> a)++--------------------------------------------------------------------------------+-- | Creates a new subscription 'Driver' state machine.+newDriver :: forall r. Settings -> Generator -> Driver r+newDriver setts gen = Driver $ go (initState gen)+  where+    go :: forall a. State r -> In r a -> a+    go st@State{..} (Pkg Package{..}) = do+        elm <- H.lookup packageCorrelation _reg+        case packageCmd of+            0xC2 -> do+                _   <- querySubscription packageCorrelation _model+                msg <- maybeDecodeMessage packageData+                let e   = getField $ streamResolvedEvent msg+                    evt = newResolvedEventFromBuf e+                    app = EventAppeared evt+                    ConnectReg k _ _ = elm+                return (k app, Driver $ go st)++            0xC7 -> do+                _   <- querySubscription packageCorrelation _model+                msg <- maybeDecodeMessage packageData+                let e   = getField $ psseaEvt msg+                    evt = newResolvedEvent e+                    app = EventAppeared evt+                    ConnectPersist k _ _ _ = elm+                return (k app, Driver $ go st)++            0xC1 -> do+                msg <- maybeDecodeMessage packageData+                let lcp  = getField $ subscribeLastCommitPos msg+                    len  = getField $ subscribeLastEventNumber msg+                    meta = RegularMeta lcp len+                    ConnectReg k _ _ = elm+                    nxt_m = confirmedSubscription packageCorrelation meta _model+                run   <- querySubscription packageCorrelation nxt_m+                let nxt_st = st { _model = nxt_m }+                    evt    = SubConfirmed run+                return (k evt, Driver $ go nxt_st)++            0xC6 -> do+                msg <- maybeDecodeMessage packageData+                let lcp  = getField $ pscLastCommitPos msg+                    sid  = getField $ pscId msg+                    len  = getField $ pscLastEvtNumber msg+                    meta = PersistMeta sid lcp len+                    ConnectPersist k _ _ _ = elm+                    nxt_m = confirmedSubscription packageCorrelation meta _model+                run   <- querySubscription packageCorrelation nxt_m+                let nxt_st = st { _model = nxt_m }+                    evt    = SubConfirmed run+                return (k evt, Driver $ go nxt_st)++            0xC9 -> confirmPAction elm (getField . cpscResult) createRException+            0xCF -> confirmPAction elm (getField . upscResult) updateRException+            0xCB -> confirmPAction elm (getField . dpscResult) deleteRException++            0xC4 -> do+                run <- querySubscription packageCorrelation _model+                msg <- maybeDecodeMessage packageData+                let reason  = fromMaybe D_Unsubscribed $ getField+                                                       $ dropReason msg+                    nxt_m   = unsubscribed run _model+                    dreason = toSubDropReason reason+                    evt     = Dropped dreason+                    nxt_reg = H.delete packageCorrelation _reg+                    nxt_st  = st { _model = nxt_m+                                 , _reg   = nxt_reg }+                case elm of+                  ConnectReg k _ _       -> return (k evt, Driver $ go nxt_st)+                  ConnectPersist k _ _ _ -> return (k evt, Driver $ go nxt_st)+                  _                      -> Nothing++            _ -> Nothing+      where+        confirmPAction :: Decode m+                       => Cmd r+                       -> (m -> e)+                       -> (e -> Maybe PersistActionException)+                       -> Maybe (r, Driver r)+        confirmPAction (ApplyPersistAction k g n c) fd em = do+            msg <- maybeDecodeMessage packageData+            _   <- queryPersistentAction packageCorrelation _model+            let nxt_m  = confirmedAction packageCorrelation _model+                nxt_rg = H.delete packageCorrelation _reg+                nxt_st = st { _model = nxt_m+                            , _reg   = nxt_rg+                            }+                evt    = ConfirmedAction packageCorrelation g n c+            case em $ fd msg of+                Just e  -> return (k $ Left e, Driver $ go nxt_st)+                Nothing -> return (k $ Right evt, Driver $ go nxt_st)+        confirmPAction _ _ _ = Nothing++    go st@State{..} (Cmd cmd) =+        case cmd of+            ConnectReg _ s tos ->+                let (u, nxt_g) = nextUUID _gen+                    pkg        = createConnectRegularPackage setts u s tos+                    nxt_m      = connectReg s tos u _model+                    nxt_st     = st { _model = nxt_m+                                    , _gen   = nxt_g+                                    , _reg   = H.insert u cmd _reg } in+                (pkg, Driver $ go nxt_st)+            ConnectPersist _ gn n b ->+                let (u, nxt_g) = nextUUID _gen+                    pkg        = createConnectPersistPackage setts u gn n b+                    nxt_m      = connectPersist gn n b u _model+                    nxt_st     = st { _model = nxt_m+                                    , _gen   = nxt_g+                                    , _reg   = H.insert u cmd _reg } in+                (pkg, Driver $ go nxt_st)+            Unsubscribe r ->+                let pkg    = createUnsubscribePackage setts $ runningUUID r in+                (pkg, Driver $ go st)+            ApplyPersistAction _ gn n a ->+                let (u, nxt_g) = nextUUID _gen+                    pkg        = createPersistActionPackage setts u gn n a+                    nxt_m      = persistAction gn n u a _model+                    nxt_st     = st { _model = nxt_m+                                    , _gen   = nxt_g+                                    , _reg   = H.insert u cmd _reg } in+                (pkg, Driver $ go nxt_st)+            PersistAck _ run evts ->+                let RunningPersist _ _ _ _ sid _ _ = run+                    u   = runningUUID run+                    pkg = createAckPackage setts u sid evts in+                (pkg, Driver $ go st)+            PersistNak _ run na r evts ->+                let RunningPersist _ _ _ _ sid _ _ = run+                    u   = runningUUID run+                    pkg = createNakPackage setts u sid na r evts  in+                (pkg, Driver $ go st)+    go st Abort = (H.elems $ _reg st) >>= _F+      where+        _F (ConnectReg k _ _)           = [k $ Dropped SubAborted]+        _F (ConnectPersist k _ _ _)     = [k $ Dropped SubAborted]+        _F (ApplyPersistAction k _ _ _) = [k $ Left PersistActionAborted]+        _F _                            = []++--------------------------------------------------------------------------------+maybeDecodeMessage :: Decode a => ByteString -> Maybe a+maybeDecodeMessage bytes =+    case runGet decodeMessage bytes of+        Right a -> Just a+        _       -> Nothing

@@ -0,0 +1,395 @@+{-# LANGUAGE DataKinds                 #-}+{-# LANGUAGE DeriveGeneric             #-}+{-# OPTIONS_GHC -fcontext-stack=26     #-}+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Internal.Manager.Subscription.Message+-- Copyright : (C) 2015 Yorick Laupa+-- License : (see the file LICENSE)+--+-- Maintainer : Yorick Laupa <yo.eight@gmail.com>+-- Stability : provisional+-- Portability : non-portable+--+--------------------------------------------------------------------------------+module Database.EventStore.Internal.Manager.Subscription.Message where++--------------------------------------------------------------------------------+import Data.ByteString (ByteString)+import Data.Int+import GHC.Generics (Generic)++--------------------------------------------------------------------------------+import Data.ProtocolBuffers+import Data.Text (Text)++--------------------------------------------------------------------------------+import Database.EventStore.Internal.TimeSpan+import Database.EventStore.Internal.Types++--------------------------------------------------------------------------------+-- | Stream subscription connection request.+data SubscribeToStream+    = SubscribeToStream+      { subscribeStreamId       :: Required 1 (Value Text)+      , subscribeResolveLinkTos :: Required 2 (Value Bool)+      }+    deriving (Generic, Show)++--------------------------------------------------------------------------------+instance Encode SubscribeToStream++--------------------------------------------------------------------------------+-- | 'SubscribeToStream' smart constructor.+subscribeToStream :: Text -> Bool -> SubscribeToStream+subscribeToStream stream_id res_link_tos =+    SubscribeToStream+    { subscribeStreamId       = putField stream_id+    , subscribeResolveLinkTos = putField res_link_tos+    }++--------------------------------------------------------------------------------+-- | Stream subscription connection response.+data SubscriptionConfirmation+    = SubscriptionConfirmation+      { subscribeLastCommitPos   :: Required 1 (Value Int64)+      , subscribeLastEventNumber :: Optional 2 (Value Int32)+      }+    deriving (Generic, Show)++--------------------------------------------------------------------------------+instance Decode SubscriptionConfirmation++--------------------------------------------------------------------------------+-- | Serialized event sent by the server when a new event has been appended to a+--   stream.+data StreamEventAppeared+    = StreamEventAppeared+      { streamResolvedEvent :: Required 1 (Message ResolvedEventBuf) }+    deriving (Generic, Show)++--------------------------------------------------------------------------------+instance Decode StreamEventAppeared++--------------------------------------------------------------------------------+-- | Represents the reason subscription drop happened.+data DropReason+    = D_Unsubscribed+    | D_AccessDenied+    | D_NotFound+    | D_PersistentSubscriptionDeleted+    deriving (Enum, Eq, Show)++--------------------------------------------------------------------------------+-- | A message sent by the server when a subscription has been dropped.+data SubscriptionDropped+    = SubscriptionDropped+      { dropReason :: Optional 1 (Enumeration DropReason) }+    deriving (Generic, Show)++--------------------------------------------------------------------------------+instance Decode SubscriptionDropped++--------------------------------------------------------------------------------+-- | A message sent to the server to indicate the user asked to end a+--   subscription.+data UnsubscribeFromStream = UnsubscribeFromStream deriving (Generic, Show)++--------------------------------------------------------------------------------+instance Encode UnsubscribeFromStream++--------------------------------------------------------------------------------+-- | Create persistent subscription request.+data CreatePersistentSubscription =+    CreatePersistentSubscription+    { cpsGroupName         :: Required 1  (Value Text)+    , cpsStreamId          :: Required 2  (Value Text)+    , cpsResolveLinkTos    :: Required 3  (Value Bool)+    , cpsStartFrom         :: Required 4  (Value Int32)+    , cpsMsgTimeout        :: Required 5  (Value Int32)+    , cpsRecordStats       :: Required 6  (Value Bool)+    , cpsLiveBufSize       :: Required 7  (Value Int32)+    , cpsReadBatchSize     :: Required 8  (Value Int32)+    , cpsBufSize           :: Required 9  (Value Int32)+    , cpsMaxRetryCount     :: Required 10 (Value Int32)+    , cpsPreferRoundRobin  :: Required 11 (Value Bool)+    , cpsChkPtAfterTime    :: Required 12 (Value Int32)+    , cpsChkPtMaxCount     :: Required 13 (Value Int32)+    , cpsChkPtMinCount     :: Required 14 (Value Int32)+    , cpsSubMaxCount       :: Required 15 (Value Int32)+    , cpsNamedConsStrategy :: Optional 16 (Value Text)+    } deriving (Generic, Show)++--------------------------------------------------------------------------------+-- | 'CreatePersistentSubscription' smart constructor.+_createPersistentSubscription :: Text+                              -> Text+                              -> PersistentSubscriptionSettings+                              -> CreatePersistentSubscription+_createPersistentSubscription group stream sett =+    CreatePersistentSubscription+    { cpsGroupName         = putField group+    , cpsStreamId          = putField stream+    , cpsResolveLinkTos    = putField $ psSettingsResolveLinkTos sett+    , cpsStartFrom         = putField $ psSettingsStartFrom sett+    , cpsMsgTimeout        = putField $ ms $ psSettingsMsgTimeout sett+    , cpsRecordStats       = putField $ psSettingsExtraStats sett+    , cpsLiveBufSize       = putField $ psSettingsLiveBufSize sett+    , cpsReadBatchSize     = putField $ psSettingsReadBatchSize sett+    , cpsBufSize           = putField $ psSettingsHistoryBufSize sett+    , cpsMaxRetryCount     = putField $ psSettingsMaxRetryCount sett+    , cpsPreferRoundRobin  = putField False+    , cpsChkPtAfterTime    = putField $ ms $ psSettingsCheckPointAfter sett+    , cpsChkPtMaxCount     = putField $ psSettingsMaxCheckPointCount sett+    , cpsChkPtMinCount     = putField $ psSettingsMinCheckPointCount sett+    , cpsSubMaxCount       = putField $ psSettingsMaxSubsCount sett+    , cpsNamedConsStrategy = putField $ Just strText+    }+  where+    strText = strategyText $ psSettingsNamedConsumerStrategy sett+    ms      = fromIntegral . timeSpanTotalMillis++--------------------------------------------------------------------------------+instance Encode CreatePersistentSubscription++--------------------------------------------------------------------------------+-- | Create persistent subscription outcome.+data CreatePersistentSubscriptionResult+    = CPS_Success+    | CPS_AlreadyExists+    | CPS_Fail+    | CPS_AccessDenied+    deriving (Enum, Eq, Show)++--------------------------------------------------------------------------------+-- | Create persistent subscription response.+data CreatePersistentSubscriptionCompleted =+    CreatePersistentSubscriptionCompleted+    { cpscResult :: Required 1 (Enumeration CreatePersistentSubscriptionResult)+    , cpscReason :: Optional 2 (Value Text)+    } deriving (Generic, Show)++--------------------------------------------------------------------------------+instance Decode CreatePersistentSubscriptionCompleted++--------------------------------------------------------------------------------+-- | Delete persistent subscription request.+data DeletePersistentSubscription =+    DeletePersistentSubscription+    { dpsGroupName :: Required 1 (Value Text)+    , dpsStreamId  :: Required 2 (Value Text)+    } deriving (Generic, Show)++--------------------------------------------------------------------------------+instance Encode DeletePersistentSubscription++--------------------------------------------------------------------------------+-- | 'DeletePersistentSubscription' smart construction.+_deletePersistentSubscription :: Text -> Text -> DeletePersistentSubscription+_deletePersistentSubscription group_name stream_id =+    DeletePersistentSubscription+    { dpsGroupName = putField group_name+    , dpsStreamId  = putField stream_id+    }++--------------------------------------------------------------------------------+-- | Delete persistent subscription outcome.+data DeletePersistentSubscriptionResult+    = DPS_Success+    | DPS_DoesNotExist+    | DPS_Fail+    | DPS_AccessDenied+    deriving (Enum, Eq, Show)++--------------------------------------------------------------------------------+-- | Delete persistent subscription response.+data DeletePersistentSubscriptionCompleted =+    DeletePersistentSubscriptionCompleted+    { dpscResult :: Required 1 (Enumeration DeletePersistentSubscriptionResult)+    , dpscReason :: Optional 2 (Value Text)+    } deriving (Generic, Show)++--------------------------------------------------------------------------------+instance Decode DeletePersistentSubscriptionCompleted++--------------------------------------------------------------------------------+-- | Update persistent subscription request.+data UpdatePersistentSubscription =+    UpdatePersistentSubscription+    { upsGroupName         :: Required 1  (Value Text)+    , upsStreamId          :: Required 2  (Value Text)+    , upsResolveLinkTos    :: Required 3  (Value Bool)+    , upsStartFrom         :: Required 4  (Value Int32)+    , upsMsgTimeout        :: Required 5  (Value Int32)+    , upsRecordStats       :: Required 6  (Value Bool)+    , upsLiveBufSize       :: Required 7  (Value Int32)+    , upsReadBatchSize     :: Required 8  (Value Int32)+    , upsBufSize           :: Required 9  (Value Int32)+    , upsMaxRetryCount     :: Required 10 (Value Int32)+    , upsPreferRoundRobin  :: Required 11 (Value Bool)+    , upsChkPtAfterTime    :: Required 12 (Value Int32)+    , upsChkPtMaxCount     :: Required 13 (Value Int32)+    , upsChkPtMinCount     :: Required 14 (Value Int32)+    , upsSubMaxCount       :: Required 15 (Value Int32)+    , upsNamedConsStrategy :: Optional 16 (Value Text)+    } deriving (Generic, Show)++--------------------------------------------------------------------------------+-- | 'UpdatePersistentSubscription' smart constructor.+_updatePersistentSubscription :: Text+                              -> Text+                              -> PersistentSubscriptionSettings+                              -> UpdatePersistentSubscription+_updatePersistentSubscription group stream sett =+    UpdatePersistentSubscription+    { upsGroupName         = putField group+    , upsStreamId          = putField stream+    , upsResolveLinkTos    = putField $ psSettingsResolveLinkTos sett+    , upsStartFrom         = putField $ psSettingsStartFrom sett+    , upsMsgTimeout        = putField $ ms $ psSettingsMsgTimeout sett+    , upsRecordStats       = putField $ psSettingsExtraStats sett+    , upsLiveBufSize       = putField $ psSettingsLiveBufSize sett+    , upsReadBatchSize     = putField $ psSettingsReadBatchSize sett+    , upsBufSize           = putField $ psSettingsHistoryBufSize sett+    , upsMaxRetryCount     = putField $ psSettingsMaxRetryCount sett+    , upsPreferRoundRobin  = putField False+    , upsChkPtAfterTime    = putField $ ms $ psSettingsCheckPointAfter sett+    , upsChkPtMaxCount     = putField $ psSettingsMaxCheckPointCount sett+    , upsChkPtMinCount     = putField $ psSettingsMinCheckPointCount sett+    , upsSubMaxCount       = putField $ psSettingsMaxSubsCount sett+    , upsNamedConsStrategy = putField $ Just strText+    }+  where+    strText = strategyText $ psSettingsNamedConsumerStrategy sett+    ms      = fromIntegral . timeSpanTotalMillis++--------------------------------------------------------------------------------+instance Encode UpdatePersistentSubscription++--------------------------------------------------------------------------------+-- | Update persistent subscription outcome.+data UpdatePersistentSubscriptionResult+    = UPS_Success+    | UPS_DoesNotExist+    | UPS_Fail+    | UPS_AccessDenied+    deriving (Enum, Eq, Show)++--------------------------------------------------------------------------------+-- | Update persistent subscription response.+data UpdatePersistentSubscriptionCompleted =+    UpdatePersistentSubscriptionCompleted+    { upscResult :: Required 1 (Enumeration UpdatePersistentSubscriptionResult)+    , upscReason :: Optional 2 (Value Text)+    } deriving (Generic, Show)++--------------------------------------------------------------------------------+instance Decode UpdatePersistentSubscriptionCompleted++--------------------------------------------------------------------------------+-- | Connect to a persistent subscription request.+data ConnectToPersistentSubscription =+    ConnectToPersistentSubscription+    { ctsId                  :: Required 1 (Value Text)+    , ctsStreamId            :: Required 2 (Value Text)+    , ctsAllowedInFlightMsgs :: Required 3 (Value Int32)+    } deriving (Generic, Show)++--------------------------------------------------------------------------------+instance Encode ConnectToPersistentSubscription++--------------------------------------------------------------------------------+-- | 'ConnectToPersistentSubscription' smart constructor.+_connectToPersistentSubscription :: Text+                                 -> Text+                                 -> Int32+                                 -> ConnectToPersistentSubscription+_connectToPersistentSubscription sub_id stream_id all_fly_msgs =+    ConnectToPersistentSubscription+    { ctsId                  = putField sub_id+    , ctsStreamId            = putField stream_id+    , ctsAllowedInFlightMsgs = putField all_fly_msgs+    }++--------------------------------------------------------------------------------+-- | Ack processed events request.+data PersistentSubscriptionAckEvents =+    PersistentSubscriptionAckEvents+    { psaeId              :: Required 1 (Value Text)+    , psaeProcessedEvtIds :: Repeated 2 (Value ByteString)+    } deriving (Generic, Show)++--------------------------------------------------------------------------------+instance Encode PersistentSubscriptionAckEvents++--------------------------------------------------------------------------------+-- | 'PersistentSubscriptionAckEvents' smart constructor.+persistentSubscriptionAckEvents :: Text+                                -> [ByteString]+                                -> PersistentSubscriptionAckEvents+persistentSubscriptionAckEvents sub_id evt_ids =+    PersistentSubscriptionAckEvents+    { psaeId              = putField sub_id+    , psaeProcessedEvtIds = putField evt_ids+    }++--------------------------------------------------------------------------------+-- | Gathers every possible Nak actions.+data NakAction+    = NA_Unknown+    | NA_Park+    | NA_Retry+    | NA_Skip+    | NA_Stop+    deriving (Enum, Eq, Show)++--------------------------------------------------------------------------------+-- | Nak processed events request.+data PersistentSubscriptionNakEvents =+    PersistentSubscriptionNakEvents+    { psneId              :: Required 1 (Value Text)+    , psneProcessedEvtIds :: Repeated 2 (Value ByteString)+    , psneMsg             :: Optional 3 (Value Text)+    , psneAction          :: Required 4 (Enumeration NakAction)+    } deriving (Generic, Show)++--------------------------------------------------------------------------------+instance Encode PersistentSubscriptionNakEvents++--------------------------------------------------------------------------------+-- | 'PersistentSubscriptionNakEvents' smart constructor.+persistentSubscriptionNakEvents :: Text+                                -> [ByteString]+                                -> Maybe Text+                                -> NakAction+                                -> PersistentSubscriptionNakEvents+persistentSubscriptionNakEvents sub_id evt_ids msg action =+    PersistentSubscriptionNakEvents+    { psneId              = putField sub_id+    , psneProcessedEvtIds = putField evt_ids+    , psneMsg             = putField msg+    , psneAction          = putField action+    }++--------------------------------------------------------------------------------+-- | Connection to persistent subscription response.+data PersistentSubscriptionConfirmation =+    PersistentSubscriptionConfirmation+    { pscLastCommitPos :: Required 1 (Value Int64)+    , pscId            :: Required 2 (Value Text)+    , pscLastEvtNumber :: Optional 3 (Value Int32)+    } deriving (Generic, Show)++--------------------------------------------------------------------------------+instance Decode PersistentSubscriptionConfirmation++--------------------------------------------------------------------------------+-- | Avalaible event sent by the server in the context of a persistent+--   subscription..+data PersistentSubscriptionStreamEventAppeared =+    PersistentSubscriptionStreamEventAppeared+    { psseaEvt :: Required 1 (Message ResolvedIndexedEvent) }+    deriving (Generic, Show)++--------------------------------------------------------------------------------+instance Decode PersistentSubscriptionStreamEventAppeared

@@ -0,0 +1,318 @@+{-# LANGUAGE GADTs           #-}+{-# LANGUAGE Rank2Types      #-}+{-# LANGUAGE RecordWildCards #-}+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Internal.Manager.Subscription.Model+-- Copyright : (C) 2015 Yorick Laupa+-- License : (see the file LICENSE)+--+-- Maintainer : Yorick Laupa <yo.eight@gmail.com>+-- Stability : provisional+-- Portability : non-portable+--+-- Main Subscription bookkeeping structure.+--------------------------------------------------------------------------------+module Database.EventStore.Internal.Manager.Subscription.Model+    ( PersistAction(..)+    , PendingAction(..)+    , Running(..)+    , Meta(..)+    , Model+    , runningUUID+    , runningLastEventNumber+    , runningLastCommitPosition+    , querySubscription+    , queryPersistentAction+    , confirmedSubscription+    , confirmedAction+    , newModel+    , unsubscribed+    , connectReg+    , connectPersist+    , persistAction+    ) where++--------------------------------------------------------------------------------+import Data.Int++--------------------------------------------------------------------------------+import qualified Data.HashMap.Strict as H+import           Data.Text+import           Data.UUID++--------------------------------------------------------------------------------+import Database.EventStore.Internal.Types++--------------------------------------------------------------------------------+-- | Type of persistent action.+data PersistAction+    = PersistCreate PersistentSubscriptionSettings+    | PersistUpdate PersistentSubscriptionSettings+    | PersistDelete++--------------------------------------------------------------------------------+-- | Represents an persistent action that hasn't been completed yet.+data PendingAction =+    PendingAction+    { _paGroup  :: !Text+    , _paStream :: !Text+    , _paTpe    :: !PersistAction+    }++--------------------------------------------------------------------------------+type Register a = H.HashMap UUID a++--------------------------------------------------------------------------------+-- | Represents a 'Subscription' which is about to be confirmed.+data Pending+    = PendingReg Text Bool+      -- ^ Related to regular subscription. In order of appearance:+      --+      --   * Stream name.+      --+      --   * Resolve Link TOS.+    | PendingPersist Text Text Int32+      -- ^ Related to persistent subscription. In order of appearance:+      --+      --   * Group name.+      --+      --   * Stream name.+      --+      --   * Buffer size.+      deriving Show++--------------------------------------------------------------------------------+-- | Represents a running subscription. Gathers useful information.+data Running+    = RunningReg UUID Text Bool Int64 (Maybe Int32)+      -- ^ Related regular subscription. In order of appearance:+      --+      --   * Subscription id.+      --+      --   * Stream name.+      --+      --   * Resolve Link TOS.+      --+      --   * Last commit position.+      --+      --   * Last event number.+    | RunningPersist UUID Text Text Int32 Text Int64 (Maybe Int32)+      -- ^ Related to persistent subscription. In order of appearance:+      --+      --   * Subscription id.+      --+      --   * Group name.+      --+      --   * Stream name.+      --+      --   * Buffer size.+      --+      --   * Persistence subscription id.+      --+      --   * Last commit position.+      --+      --   * Last event number.+      deriving Show++--------------------------------------------------------------------------------+-- | Gets the event number of a running subscription.+runningLastEventNumber :: Running -> Maybe Int32+runningLastEventNumber (RunningReg _ _ _ _ i) = i+runningLastEventNumber (RunningPersist _ _ _ _ _ _ i) = i++--------------------------------------------------------------------------------+-- | Gets the commit position of a running subscription.+runningLastCommitPosition :: Running -> Int64+runningLastCommitPosition (RunningReg _ _ _ i _) = i+runningLastCommitPosition (RunningPersist _ _ _ _ _ i _) = i++--------------------------------------------------------------------------------+-- | Gets the 'UUID' of a running subscription.+runningUUID :: Running -> UUID+runningUUID (RunningReg i _ _ _ _)         = i+runningUUID (RunningPersist i _ _ _ _ _ _) = i++--------------------------------------------------------------------------------+-- | Type of requests handled by the model.+data Request a where+    -- Read request.+    Query :: Query a -> Request a+    -- Write request.+    Execute :: Action -> Request Model++--------------------------------------------------------------------------------+-- | Set of a piece of information we can query from the 'Subscription' model.+data Query a where+    -- Query a running 'Subscription'.+    QuerySub :: UUID -> Query (Maybe Running)+    -- Query a pending persistent action.+    QueryAction :: UUID -> Query (Maybe PendingAction)++--------------------------------------------------------------------------------+-- | Set of actions handled by the 'Subscription' model.+data Action+    = Connect UUID Connect+      -- ^ Subscription connection.+    | Confirmed Confirmed+      -- ^ Subscription action confirmation.+    | Unsubscribed UUID+      -- ^ Subscription no longer exist.+    | PersistAction Text Text UUID PersistAction+      -- ^ Add a new persist action.++--------------------------------------------------------------------------------+-- | Subscription connection information.+data Connect+    = ConnectReg Text Bool+      --         |    |---- Resolve TOS link.+      --         |--------- Stream name.+    | ConnectPersist Text Text Int32+      --             |    |    |---- Buffer size.+      --             |    |--------- Stream name.+      --             |-------------- Group name.++--------------------------------------------------------------------------------+-- | Information related to a confirmed 'Subscription'.+data Meta+    = RegularMeta Int64 (Maybe Int32)+      --          |     |------------- Last commit position.+      --          |------------------- Last event number.+    | PersistMeta Text Int64 (Maybe Int32)+      --          |    |     |------------- Subscription Id.+      --          |    |------------------- Last commit position.+      --          |------------------------ Last event number.++--------------------------------------------------------------------------------+-- | Subscription action confirmation.+data Confirmed+    = ConfirmedConnection UUID Meta+      -- ^ Confirms a 'Subscription' connection has handled successfully.+    | ConfirmedPersistAction UUID+      -- ^ Confirms a persist action has been handled successfully.++--------------------------------------------------------------------------------+-- | Retrieves a running 'Subscription'.+querySubscription :: UUID -> Model -> Maybe Running+querySubscription u (Model k) = k $ Query $ QuerySub u++--------------------------------------------------------------------------------+-- | Retrieves an ongoing persistent action.+queryPersistentAction :: UUID -> Model -> Maybe PendingAction+queryPersistentAction u (Model k) = k $ Query $ QueryAction u++--------------------------------------------------------------------------------+-- | Registers a regular 'Subscription' request.+connectReg :: Text -> Bool -> UUID -> Model -> Model+connectReg n t u (Model k) = k $ Execute $ Connect u (ConnectReg n t)++--------------------------------------------------------------------------------+-- | Registers a persistent 'Subscription' request.+connectPersist :: Text -> Text -> Int32 -> UUID -> Model -> Model+connectPersist g n b u (Model k) =+    k $ Execute $ Connect u (ConnectPersist g n b)++--------------------------------------------------------------------------------+-- | Registers a persistent action.+persistAction :: Text -> Text -> UUID -> PersistAction -> Model -> Model+persistAction g n u a (Model k) = k $ Execute $ PersistAction g n u a++--------------------------------------------------------------------------------+-- | Confirms a subscription.+confirmedSubscription :: UUID -> Meta -> Model -> Model+confirmedSubscription u m (Model k) =+    k $ Execute $ Confirmed $ ConfirmedConnection u m++--------------------------------------------------------------------------------+-- | Confirms a persistent action. It doesn't assume if the action went well.+confirmedAction :: UUID -> Model -> Model+confirmedAction u (Model k) = k $ Execute $ Confirmed $ ConfirmedPersistAction u++--------------------------------------------------------------------------------+-- | Remove a 'Subscription'.+unsubscribed :: Running -> Model -> Model+unsubscribed r (Model k) = k $ Execute $ Unsubscribed $ runningUUID r++--------------------------------------------------------------------------------+-- | 'Subscription' model internal state.+data State =+    State+    { _stPending :: !(Register Pending)+      -- ^ Holds all pending 'Subscription's+    , _stRunning :: !(Register Running)+      -- ^ Holds all 'Subscription's that are currently running.+    , _stAction  :: !(Register PendingAction)+      -- ^ Holds all pending persistent actions.+    }++--------------------------------------------------------------------------------+emptyState :: State+emptyState = State H.empty H.empty H.empty++--------------------------------------------------------------------------------+-- | Subscription operations state machine. Keeps every information related to+--   subscription updated.+newtype Model = Model (forall a. Request a -> a)++--------------------------------------------------------------------------------+-- | Creates a new 'Subscription' model.+newModel :: Model+newModel = Model $ modelHandle emptyState++--------------------------------------------------------------------------------+-- | Main model handler.+modelHandle :: State -> Request a -> a+modelHandle s (Execute e) =+    case e of+        Connect u c ->+            case c of+                ConnectReg n tos ->+                    let p      = PendingReg n tos+                        nxt_ps = H.insert u p $ _stPending s+                        nxt_s  = s { _stPending = nxt_ps } in+                    Model $ modelHandle nxt_s+                ConnectPersist g n b ->+                    let p      = PendingPersist g n b+                        nxt_ps = H.insert u p $ _stPending s+                        nxt_s  = s { _stPending = nxt_ps } in+                    Model $ modelHandle nxt_s+        Confirmed c ->+            case c of+                ConfirmedConnection u tpe ->+                    case tpe of+                        RegularMeta lc le ->+                            case H.lookup u $ _stPending s of+                              Just (PendingReg n tos) ->+                                  let r      = RunningReg u n tos lc le+                                      nxt_rs = H.insert u r $ _stRunning s+                                      nxt_s  = s { _stRunning = nxt_rs } in+                                  Model $ modelHandle nxt_s+                              _ -> Model $ modelHandle s+                        PersistMeta sb lc le ->+                            case H.lookup u $ _stPending s of+                                Just (PendingPersist g n b) ->+                                    let r      = RunningPersist u g n b sb lc le+                                        nxt_rs = H.insert u r $ _stRunning s+                                        nxt_s  = s { _stRunning = nxt_rs } in+                                    Model $ modelHandle nxt_s+                                _ -> Model $ modelHandle s+                ConfirmedPersistAction u ->+                    case H.lookup u $ _stAction s of+                        Just (PendingAction{}) ->+                            let nxt_as = H.delete u $ _stAction s+                                nxt_s  = s { _stAction = nxt_as } in+                            Model $ modelHandle nxt_s+                        _ -> Model $ modelHandle s+        Unsubscribed u ->+            let nxt_ps = H.delete u $ _stRunning s+                nxt_s  = s { _stRunning = nxt_ps } in+            Model $ modelHandle nxt_s+        PersistAction g n u t ->+            let a      = PendingAction g n t+                nxt_as = H.insert u a $ _stAction s+                nxt_s  = s { _stAction = nxt_as } in+            Model $ modelHandle nxt_s+modelHandle s (Query q) =+    case q of+        QuerySub u    -> H.lookup u $ _stRunning s+        QueryAction u -> H.lookup u $ _stAction s

@@ -0,0 +1,137 @@+{-# LANGUAGE RecordWildCards #-}+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Internal.Manager.Subscription.Packages+-- Copyright : (C) 2015 Yorick Laupa+-- License : (see the file LICENSE)+--+-- Maintainer : Yorick Laupa <yo.eight@gmail.com>+-- Stability : provisional+-- Portability : non-portable+--+--------------------------------------------------------------------------------+module Database.EventStore.Internal.Manager.Subscription.Packages where++--------------------------------------------------------------------------------+import Data.Int++--------------------------------------------------------------------------------+import Data.ByteString.Lazy (toStrict)+import Data.ProtocolBuffers+import Data.Serialize+import Data.Text+import Data.UUID++--------------------------------------------------------------------------------+import Database.EventStore.Internal.Manager.Subscription.Message+import Database.EventStore.Internal.Manager.Subscription.Model+import Database.EventStore.Internal.Types++--------------------------------------------------------------------------------+import Prelude++--------------------------------------------------------------------------------+-- | Creates a regular subscription connection 'Package'.+createConnectRegularPackage :: Settings -> UUID -> Text -> Bool -> Package+createConnectRegularPackage Settings{..} uuid stream tos =+    Package+    { packageCmd         = 0xC0+    , packageCorrelation = uuid+    , packageData        = runPut $ encodeMessage msg+    , packageCred        = s_credentials+    }+  where+    msg = subscribeToStream stream tos++--------------------------------------------------------------------------------+-- | Creates a persistent subscription connection 'Package'.+createConnectPersistPackage :: Settings+                            -> UUID+                            -> Text+                            -> Text+                            -> Int32+                            -> Package+createConnectPersistPackage Settings{..} uuid grp stream bufSize =+    Package+    { packageCmd         = 0xC5+    , packageCorrelation = uuid+    , packageData        = runPut $ encodeMessage msg+    , packageCred        = s_credentials+    }+  where+    msg = _connectToPersistentSubscription grp stream bufSize++--------------------------------------------------------------------------------+-- | Creates a persistent subscription 'Package'.+createPersistActionPackage :: Settings+                           -> UUID+                           -> Text+                           -> Text+                           -> PersistAction+                           -> Package+createPersistActionPackage Settings{..} u grp strm tpe =+    Package+    { packageCmd         = cmd+    , packageCorrelation = u+    , packageData        = runPut msg+    , packageCred        = s_credentials+    }+  where+    msg =+        case tpe of+            PersistCreate sett ->+                encodeMessage $ _createPersistentSubscription grp strm sett+            PersistUpdate sett ->+                encodeMessage $ _updatePersistentSubscription grp strm sett+            PersistDelete ->+                encodeMessage $ _deletePersistentSubscription grp strm+    cmd =+        case tpe of+            PersistCreate _  -> 0xC8+            PersistUpdate _  -> 0xCE+            PersistDelete    -> 0xCA++--------------------------------------------------------------------------------+-- | Creates Ack 'Package'.+createAckPackage :: Settings -> UUID -> Text -> [UUID] -> Package+createAckPackage Settings{..} corr sid eids =+    Package+    { packageCmd         = 0xCC+    , packageCorrelation = corr+    , packageData        = runPut $ encodeMessage msg+    , packageCred        = s_credentials+    }+  where+    bytes = fmap (toStrict . toByteString) eids+    msg   = persistentSubscriptionAckEvents sid bytes++--------------------------------------------------------------------------------+-- | Create Nak 'Package'.+createNakPackage :: Settings+                 -> UUID+                 -> Text+                 -> NakAction+                 -> Maybe Text+                 -> [UUID]+                 -> Package+createNakPackage Settings{..} corr sid act txt eids =+    Package+    { packageCmd         = 0xCD+    , packageCorrelation = corr+    , packageData        = runPut $ encodeMessage msg+    , packageCred        = s_credentials+    }+  where+    bytes = fmap (toStrict . toByteString) eids+    msg   = persistentSubscriptionNakEvents sid bytes txt act++--------------------------------------------------------------------------------+-- | Create an unsubscribe 'Package'.+createUnsubscribePackage :: Settings -> UUID -> Package+createUnsubscribePackage Settings{..} uuid =+    Package+    { packageCmd         = 0xC3+    , packageCorrelation = uuid+    , packageData        = runPut $ encodeMessage UnsubscribeFromStream+    , packageCred        = s_credentials+    }

@@ -0,0 +1,193 @@+{-# LANGUAGE DataKinds          #-}+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE GADTs              #-}+{-# LANGUAGE KindSignatures     #-}+{-# LANGUAGE Rank2Types         #-}+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Internal.Operation+-- Copyright : (C) 2015 Yorick Laupa+-- License : (see the file LICENSE)+--+-- Maintainer : Yorick Laupa <yo.eight@gmail.com>+-- Stability : provisional+-- Portability : non-portable+--+--------------------------------------------------------------------------------+module Database.EventStore.Internal.Operation where++--------------------------------------------------------------------------------+import Control.Applicative+import Control.Exception+import Control.Monad+import Data.Typeable++--------------------------------------------------------------------------------+import Data.ProtocolBuffers+import Data.Text+import Data.UUID+import Data.Word++--------------------------------------------------------------------------------+import Database.EventStore.Internal.Stream+import Database.EventStore.Internal.Types++--------------------------------------------------------------------------------+import Prelude++--------------------------------------------------------------------------------+-- | Operation result sent by the server.+data OpResult+    = OP_SUCCESS+    | OP_PREPARE_TIMEOUT+    | OP_COMMIT_TIMEOUT+    | OP_FORWARD_TIMEOUT+    | OP_WRONG_EXPECTED_VERSION+    | OP_STREAM_DELETED+    | OP_INVALID_TRANSACTION+    | OP_ACCESS_DENIED+    deriving (Eq, Enum, Show)++--------------------------------------------------------------------------------+-- | Operation exception that can occurs on an operation response.+data OperationError+    = WrongExpectedVersion Text ExpectedVersion -- ^ Stream and Expected Version+    | StreamDeleted Text                        -- ^ Stream+    | InvalidTransaction+    | AccessDenied StreamName                   -- ^ Stream+    | InvalidServerResponse Word8 Word8         -- ^ Expected, Found+    | ProtobufDecodingError String+    | ServerError (Maybe Text)                  -- ^ Reason+    | InvalidOperation Text+      -- ^ Invalid operation state. If happens, it's a driver bug.+    | Aborted+      -- ^ Occurs when the user asked to close the connection or if the+      --   connection can't reconnect anymore.+    deriving (Show, Typeable)++--------------------------------------------------------------------------------+instance Exception OperationError++--------------------------------------------------------------------------------+-- | Main operation state machine instruction.+data SM o a+    = Return a+      -- ^ Lifts a pure value into the intruction tree. Also marks the end of+      --   an instruction tree.+    | Yield o (SM o a)+      -- ^ Emits an operation return value.+    | FreshId (UUID -> SM o a)+      -- ^ Asks for an unused 'UUID'.+    | forall rq rp. (Encode rq, Decode rp) =>+      SendPkg Word8 Word8 rq (rp -> SM o a)+      -- ^ Send a request message given a command and an expected command.+      --   response. It also carries a callback to call when response comes in.+    | Failure (Maybe OperationError)+      -- ^ Ends the instruction interpretation. If holds Nothing, the+      --   interpretation should resume from the beginning. Otherwise it ends+      --   by indicating what went wrong.++--------------------------------------------------------------------------------+instance Functor (SM o) where+    fmap f (Return a)          = Return (f a)+    fmap f (Yield o n)         = Yield o (fmap f n)+    fmap f (FreshId k)         = FreshId (fmap f . k)+    fmap f (SendPkg ci co p k) = SendPkg ci co p (fmap f . k)+    fmap _ (Failure e)         = Failure e++--------------------------------------------------------------------------------+instance Applicative (SM o) where+    pure = return+    (<*>) = ap++--------------------------------------------------------------------------------+instance Monad (SM o) where+    return = Return++    Return a          >>= f = f a+    Yield o n         >>= f = Yield o (n >>= f)+    FreshId k         >>= f = FreshId ((f =<<) . k)+    SendPkg ci co p k >>= f = SendPkg ci co p ((f =<<) . k)+    Failure e         >>= _ = Failure e++--------------------------------------------------------------------------------+-- | Asks for a unused 'UUID'.+freshId :: SM o UUID+freshId = FreshId Return++--------------------------------------------------------------------------------+-- | Raises an 'OperationError'.+failure :: OperationError -> SM o a+failure e = Failure $ Just e++--------------------------------------------------------------------------------+-- | Asks to resume the interpretation from the beginning.+retry :: SM o a+retry = Failure Nothing++--------------------------------------------------------------------------------+-- | Sends a request to the server given a command request and response. It+--   returns the expected deserialized message.+send :: (Encode rq, Decode rp) => Word8 -> Word8 -> rq -> SM o rp+send ci co rq = SendPkg ci co rq Return++--------------------------------------------------------------------------------+-- | Emits operation return value.+yield :: o -> SM o ()+yield o = Yield o (Return ())++--------------------------------------------------------------------------------+-- | Replaces every emitted value, via 'yield' function by calling the given+--   callback.+foreach :: SM a x -> (a -> SM b x) -> SM b x+foreach start k = go start+  where+    go (Return x)           = Return x+    go (Yield a n)          = k a >> go n+    go (FreshId ki)         = FreshId (go . ki)+    go (SendPkg ci co p kp) = SendPkg ci co p (go . kp)+    go (Failure e)          = Failure e++--------------------------------------------------------------------------------+-- | Maps every emitted value, via 'yield', using given function.+mapOp :: (a -> b) -> SM a () -> SM b ()+mapOp k sm = foreach sm (yield . k)++--------------------------------------------------------------------------------+-- | An operation is just a 'SM' tree.+type Operation a = SM a ()++--------------------------------------------------------------------------------+-- | Raises 'WrongExpectedVersion' exception.+wrongVersion :: Text -> ExpectedVersion -> SM o a+wrongVersion stream ver = failure (WrongExpectedVersion stream ver)++--------------------------------------------------------------------------------+-- | Raises 'StreamDeleted' exception.+streamDeleted :: Text -> SM o a+streamDeleted stream = failure (StreamDeleted stream)++--------------------------------------------------------------------------------+-- | Raises 'InvalidTransaction' exception.+invalidTransaction :: SM o a+invalidTransaction = failure InvalidTransaction++--------------------------------------------------------------------------------+-- | Raises 'AccessDenied' exception.+accessDenied :: StreamName -> SM o a+accessDenied = failure . AccessDenied++--------------------------------------------------------------------------------+-- | Raises 'ProtobufDecodingError' exception.+protobufDecodingError :: String -> SM o a+protobufDecodingError = failure . ProtobufDecodingError++--------------------------------------------------------------------------------+-- | Raises 'ServerError' exception.+serverError :: Maybe Text -> SM o a+serverError = failure . ServerError++--------------------------------------------------------------------------------+-- | Raises 'InvalidServerResponse' exception.+invalidServerResponse :: Word8 -> Word8 -> SM o a+invalidServerResponse expe got = failure $ InvalidServerResponse expe got

@@ -0,0 +1,102 @@+{-# LANGUAGE DataKinds         #-}+{-# LANGUAGE GADTs             #-}+{-# LANGUAGE OverloadedStrings #-}+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Internal.Operation.Catchup+-- Copyright : (C) 2015 Yorick Laupa+-- License : (see the file LICENSE)+--+-- Maintainer : Yorick Laupa <yo.eight@gmail.com>+-- Stability : provisional+-- Portability : non-portable+--+--------------------------------------------------------------------------------+module Database.EventStore.Internal.Operation.Catchup+    ( CatchupState(..)+    , catchup+    ) where++--------------------------------------------------------------------------------+import Control.Monad+import Data.Int+import Data.Maybe++--------------------------------------------------------------------------------+import Data.Text (Text)++--------------------------------------------------------------------------------+import Database.EventStore.Internal.Manager.Subscription (Checkpoint(..))+import Database.EventStore.Internal.Operation+import Database.EventStore.Internal.Operation.Read.Common+import Database.EventStore.Internal.Operation.ReadAllEvents+import Database.EventStore.Internal.Operation.ReadStreamEvents+import Database.EventStore.Internal.Stream+import Database.EventStore.Internal.Types++--------------------------------------------------------------------------------+defaultBatchSize :: Int32+defaultBatchSize = 500++--------------------------------------------------------------------------------+streamNotFound :: OperationError+streamNotFound = InvalidOperation "Catchup. inexistant stream"++--------------------------------------------------------------------------------+-- | Catchup operation state.+data CatchupState+    = RegularCatchup Text Int32+      -- ^ Indicates the stream name and the next event number to start from.+    | AllCatchup Int64 Int64+      -- ^ Indicates the commit and prepare position. Used when catching up from+      --   the $all stream.++--------------------------------------------------------------------------------+-- | Stream catching up operation.+catchup :: Settings+        -> CatchupState+        -> Bool+        -> Maybe Int32+        -> Operation ([ResolvedEvent], Bool, Checkpoint)+catchup setts init_tpe tos bat_siz = go init_tpe+  where+    batch = fromMaybe defaultBatchSize bat_siz+    go tpe = do+        let action =+                case tpe of+                    RegularCatchup stream cur_evt ->+                        let op = readStreamEvents setts Forward stream cur_evt+                                 batch tos in+                        mapOp Left op+                    AllCatchup c_pos p_pos ->+                        let op = readAllEvents setts c_pos p_pos batch+                                 tos Forward in+                        mapOp Right op++        foreach action $ \res -> do+            (eos, evts, nchk, nxt_tpe) <- case res of+                Right as -> do+                    let Position nxt_c nxt_p = sliceNext as+                        tmp_tpe = AllCatchup nxt_c nxt_p+                        chk     = CheckpointPosition $ sliceNext as+                    return (sliceEOS as, sliceEvents as, chk, tmp_tpe)+                Left rr -> fromReadResult rr $ \as ->+                    let RegularCatchup s _ = tpe+                        nxt = sliceNext as+                        tmp_tpe = RegularCatchup s nxt+                        chk = CheckpointNumber nxt in+                    return (sliceEOS as, sliceEvents as, chk, tmp_tpe)++            yield (evts, eos, nchk)+            when (not eos) $ go nxt_tpe++--------------------------------------------------------------------------------+fromReadResult :: ReadResult 'RegularStream a -> (a -> SM b x) -> SM b x+fromReadResult res k =+    case res of+        ReadNoStream        -> failure streamNotFound+        ReadStreamDeleted s -> failure $ StreamDeleted s+        ReadNotModified     -> failure $ ServerError Nothing+        ReadError e         -> failure $ ServerError e+        ReadAccessDenied s  -> failure $ AccessDenied s+        ReadSuccess ss      -> k ss

@@ -0,0 +1,63 @@+{-# LANGUAGE DataKinds       #-}+{-# LANGUAGE GADTs           #-}+{-# LANGUAGE Rank2Types      #-}+{-# LANGUAGE RecordWildCards #-}+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Internal.Operation.DeleteStream+-- Copyright : (C) 2015 Yorick Laupa+-- License : (see the file LICENSE)+--+-- Maintainer : Yorick Laupa <yo.eight@gmail.com>+-- Stability : provisional+-- Portability : non-portable+--+--------------------------------------------------------------------------------+module Database.EventStore.Internal.Operation.DeleteStream+    ( DeleteResult(..)+    , deleteStream+    ) where++--------------------------------------------------------------------------------+import Data.Maybe++--------------------------------------------------------------------------------+import Data.ProtocolBuffers+import Data.Text++--------------------------------------------------------------------------------+import Database.EventStore.Internal.Operation+import Database.EventStore.Internal.Operation.DeleteStream.Message+import Database.EventStore.Internal.Stream+import Database.EventStore.Internal.Types++--------------------------------------------------------------------------------+-- | Returned after deleting a stream. 'Position' of the write.+newtype DeleteResult = DeleteResult Position deriving (Eq, Show)++--------------------------------------------------------------------------------+-- | Delete a regular stream operation.+deleteStream :: Settings+             -> Text+             -> ExpectedVersion+             -> Maybe Bool+             -> Operation DeleteResult+deleteStream Settings{..} s v hard = do+    let msg = newRequest s (expVersionInt32 v) s_requireMaster hard+    resp <- send 0x8A 0x8B msg+    let r            = getField $ _result resp+        com_pos      = getField $ _commitPosition resp+        prep_pos     = getField $ _preparePosition resp+        com_pos_int  = fromMaybe (-1) com_pos+        prep_pos_int = fromMaybe (-1) prep_pos+        pos          = Position com_pos_int prep_pos_int+        res          = DeleteResult pos+    case r of+        OP_SUCCESS                -> yield res+        OP_PREPARE_TIMEOUT        -> retry+        OP_FORWARD_TIMEOUT        -> retry+        OP_COMMIT_TIMEOUT         -> retry+        OP_WRONG_EXPECTED_VERSION -> wrongVersion s v+        OP_STREAM_DELETED         -> streamDeleted s+        OP_INVALID_TRANSACTION    -> invalidTransaction+        OP_ACCESS_DENIED          -> accessDenied (StreamName s)

@@ -0,0 +1,64 @@+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DataKinds     #-}+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Internal.Operation.DeleteStream.Message+-- Copyright : (C) 2015 Yorick Laupa+-- License : (see the file LICENSE)+--+-- Maintainer : Yorick Laupa <yo.eight@gmail.com>+-- Stability : provisional+-- Portability : non-portable+--+--------------------------------------------------------------------------------+module Database.EventStore.Internal.Operation.DeleteStream.Message where++--------------------------------------------------------------------------------+import Data.Int+import GHC.Generics++--------------------------------------------------------------------------------+import Data.ProtocolBuffers+import Data.Text++--------------------------------------------------------------------------------+import Database.EventStore.Internal.Operation++--------------------------------------------------------------------------------+-- | Delete stream request.+data Request+    = Request+      { _streamId        :: Required 1 (Value Text)+      , _expectedVersion :: Required 2 (Value Int32)+      , _requireMaster   :: Required 3 (Value Bool)+      , _hardDelete      :: Optional 4 (Value Bool)+      }+    deriving (Generic, Show)++--------------------------------------------------------------------------------+instance Encode Request++--------------------------------------------------------------------------------+-- | 'Request' smart constructor.+newRequest :: Text -> Int32 -> Bool -> Maybe Bool -> Request+newRequest stream_id exp_ver req_master hard_delete =+    Request+    { _streamId        = putField stream_id+    , _expectedVersion = putField exp_ver+    , _requireMaster   = putField req_master+    , _hardDelete      = putField hard_delete+    }++--------------------------------------------------------------------------------+-- | Delete stream response.+data Response+    = Response+      { _result          :: Required 1 (Enumeration OpResult)+      , _message         :: Optional 2 (Value Text)+      , _preparePosition :: Optional 3 (Value Int64)+      , _commitPosition  :: Optional 4 (Value Int64)+      }+    deriving (Generic, Show)++--------------------------------------------------------------------------------+instance Decode Response

@@ -1,137 +0,0 @@-{-# LANGUAGE DeriveGeneric #-}-{-# LANGUAGE DataKinds     #-}------------------------------------------------------------------------------------ |--- Module : Database.EventStore.Internal.Operation.DeleteStreamOperation--- 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.Internal.Operation.DeleteStreamOperation-    ( deleteStreamOperation ) where-----------------------------------------------------------------------------------import Control.Concurrent-import Data.Int-import Data.Maybe-import GHC.Generics (Generic)-----------------------------------------------------------------------------------import Data.ProtocolBuffers-import Data.Text-----------------------------------------------------------------------------------import Database.EventStore.Internal.Manager.Operation-import Database.EventStore.Internal.Types-----------------------------------------------------------------------------------data DeleteStream-    = DeleteStream-      { _deleteStreamId              :: Required 1 (Value Text)-      , _deleteStreamExpectedVersion :: Required 2 (Value Int32)-      , _deleteStreamRequireMaster   :: Required 3 (Value Bool)-      , _deleteStreamHardDelete      :: Optional 4 (Value Bool)-      }-    deriving (Generic, Show)-----------------------------------------------------------------------------------instance Encode DeleteStream-----------------------------------------------------------------------------------newDeleteStream :: Text-                -> Int32-                -> Bool-                -> Maybe Bool-                -> 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-    }-----------------------------------------------------------------------------------data DeleteStreamCompleted-    = DeleteStreamCompleted-      { _deleteCompletedResult          :: Required 1 (Enumeration OpResult)-      , _deleteCompletedMessage         :: Optional 2 (Value Text)-      , _deleteCompletedPreparePosition :: Optional 3 (Value Int64)-      , _deleteCompletedCommitPosition  :: Optional 4 (Value Int64)-      }-    deriving (Generic, Show)-----------------------------------------------------------------------------------instance Decode DeleteStreamCompleted-----------------------------------------------------------------------------------deleteStreamOperation :: Settings-                      -> MVar (OperationExceptional DeleteResult)-                      -> Text-                      -> ExpectedVersion-                      -> Maybe Bool-                      -> OperationParams-deleteStreamOperation settings mvar stream_id exp_ver hard_del =-    OperationParams-    { opSettings    = settings-    , opRequestCmd  = 0x8A-    , opResponseCmd = 0x8B--    , opRequest =-        let req_master  = s_requireMaster settings-            exp_ver_int = expVersionInt32 exp_ver-            request     = newDeleteStream stream_id-                                          exp_ver_int-                                          req_master-                                          hard_del in-         return request--    , opSuccess = inspect mvar stream_id exp_ver-    , opFailure = failed mvar-    }-----------------------------------------------------------------------------------inspect :: MVar (OperationExceptional DeleteResult)-        -> Text-        -> ExpectedVersion-        -> DeleteStreamCompleted-        -> IO Decision-inspect mvar stream exp_ver dsc = go (getField $ _deleteCompletedResult dsc)-  where-    go OP_SUCCESS                = succeed mvar dsc-    go OP_PREPARE_TIMEOUT        = return Retry-    go OP_FORWARD_TIMEOUT        = return Retry-    go OP_COMMIT_TIMEOUT         = return Retry-    go OP_WRONG_EXPECTED_VERSION = failed mvar wrong_version-    go OP_STREAM_DELETED         = failed mvar (StreamDeleted stream)-    go OP_INVALID_TRANSACTION    = failed mvar InvalidTransaction-    go OP_ACCESS_DENIED          = failed mvar (AccessDenied stream)--    wrong_version = WrongExpectedVersion stream exp_ver-----------------------------------------------------------------------------------succeed :: MVar (OperationExceptional DeleteResult)-        -> DeleteStreamCompleted-        -> IO Decision-succeed mvar wec = do-    putMVar mvar (Right wr)-    return EndOperation-  where-    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-    wr           = DeleteResult pos-----------------------------------------------------------------------------------failed :: MVar (OperationExceptional DeleteResult)-       -> OperationException-       -> IO Decision-failed mvar e = do-    putMVar mvar (Left e)-    return EndOperation

@@ -0,0 +1,145 @@+{-# LANGUAGE DataKinds      #-}+{-# LANGUAGE GADTs          #-}+{-# LANGUAGE KindSignatures #-}+{-# LANGUAGE TypeFamilies   #-}+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Internal.Operation.Read.Common+-- Copyright : (C) 2015 Yorick Laupa+-- License : (see the file LICENSE)+--+-- Maintainer : Yorick Laupa <yo.eight@gmail.com>+-- Stability : provisional+-- Portability : non-portable+--+--------------------------------------------------------------------------------+module Database.EventStore.Internal.Operation.Read.Common where++--------------------------------------------------------------------------------+import Control.Applicative+import Data.Foldable+import Data.Int+import Data.Monoid+import Data.Traversable++--------------------------------------------------------------------------------+import Data.Text++--------------------------------------------------------------------------------+import Database.EventStore.Internal.Stream+import Database.EventStore.Internal.Types++--------------------------------------------------------------------------------+import Prelude++--------------------------------------------------------------------------------+-- | Enumeration detailing the possible outcomes of reading a stream.+data ReadResult       :: StreamType -> * -> * where+    ReadSuccess       :: a -> ReadResult t a+    ReadNoStream      :: ReadResult 'RegularStream a+    ReadStreamDeleted :: Text -> ReadResult 'RegularStream a+    ReadNotModified   :: ReadResult t a+    ReadError         :: Maybe Text -> ReadResult t a+    ReadAccessDenied  :: StreamName -> ReadResult t a++--------------------------------------------------------------------------------+instance Eq a => Eq (ReadResult t a) where+    ReadSuccess a       == ReadSuccess b       = a == b+    ReadNoStream        == ReadNoStream        = True+    ReadStreamDeleted s == ReadStreamDeleted v = s == v+    ReadNotModified     == ReadNotModified     = True+    ReadError e         == ReadError u         = e == u+    ReadAccessDenied s  == ReadAccessDenied v  = s == v+    _                   == _                   = False++--------------------------------------------------------------------------------+instance Show a => Show (ReadResult t a) where+    show (ReadSuccess a)       = "ReadSuccess " ++ show a+    show ReadNoStream          = "ReadNoStream"+    show (ReadStreamDeleted s) = "ReadStreamDeleted" ++ show s+    show ReadNotModified       = "ReadNoModified"+    show (ReadError e)         = "ReadError" ++ show e+    show (ReadAccessDenied s)  = "ReadAccessDenied " ++ show s++--------------------------------------------------------------------------------+instance Functor (ReadResult t) where+    fmap f (ReadSuccess a)       = ReadSuccess (f a)+    fmap _ ReadNoStream          = ReadNoStream+    fmap _ (ReadStreamDeleted s) = ReadStreamDeleted s+    fmap _ ReadNotModified       = ReadNotModified+    fmap _ (ReadError e)         = ReadError e+    fmap _ (ReadAccessDenied s)  = ReadAccessDenied s++--------------------------------------------------------------------------------+instance Foldable (ReadResult t) where+    foldMap f (ReadSuccess a) = f a+    foldMap _ _               = mempty++--------------------------------------------------------------------------------+instance Traversable (ReadResult t) where+    traverse f (ReadSuccess a)       = fmap ReadSuccess $ f a+    traverse _ ReadNoStream          = pure ReadNoStream+    traverse _ (ReadStreamDeleted s) = pure $ ReadStreamDeleted s+    traverse _ ReadNotModified       = pure ReadNotModified+    traverse _ (ReadError e)         = pure $ ReadError e+    traverse _ (ReadAccessDenied s)  = pure $ ReadAccessDenied s++--------------------------------------------------------------------------------+-- | Gathers common slice operations.+class Slice a where+    type Loc a++    sliceEvents :: a -> [ResolvedEvent]+    -- ^ Gets slice's 'ResolvedEvent's.+    sliceDirection :: a -> ReadDirection+    -- ^ Gets slice's reading direction.+    sliceEOS :: a -> Bool+    -- ^ If the slice reaches the end of the stream.+    sliceFrom :: a -> Loc a+    -- ^ Gets the starting location of this slice.+    sliceNext :: a -> Loc a+    -- ^ Gets the next location of this slice.++--------------------------------------------------------------------------------+-- | Regular stream slice.+data StreamSlice =+    StreamSlice+    { sliceStream :: !Text+    , sliceLast   :: !Int32+    , _ssDir      :: !ReadDirection+    , _ssFrom     :: !Int32+    , _ssNext     :: !Int32+    , _ssEvents   :: ![ResolvedEvent]+    , _ssEOS      :: !Bool+    } deriving Show++--------------------------------------------------------------------------------+instance Slice StreamSlice where+    type Loc StreamSlice = Int32++    sliceEvents    = _ssEvents+    sliceDirection = _ssDir+    sliceEOS       = _ssEOS+    sliceFrom      = _ssFrom+    sliceNext      = _ssNext++--------------------------------------------------------------------------------+-- | Represents a slice of the $all stream.+data AllSlice =+    AllSlice+    { _saFrom   :: !Position+    , _saNext   :: !Position+    , _saDir    :: !ReadDirection+    , _saEvents :: ![ResolvedEvent]+    , _saEOS    :: !Bool+    } deriving Show++--------------------------------------------------------------------------------+instance Slice AllSlice where+    type Loc AllSlice = Position++    sliceEvents    = _saEvents+    sliceDirection = _saDir+    sliceEOS       = _saEOS+    sliceFrom      = _saFrom+    sliceNext      = _saNext

@@ -0,0 +1,62 @@+{-# LANGUAGE RecordWildCards #-}+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Internal.Operation.ReadAllEvents+-- Copyright : (C) 2015 Yorick Laupa+-- License : (see the file LICENSE)+--+-- Maintainer : Yorick Laupa <yo.eight@gmail.com>+-- Stability : provisional+-- Portability : non-portable+--+--------------------------------------------------------------------------------+module Database.EventStore.Internal.Operation.ReadAllEvents+    ( readAllEvents ) where++--------------------------------------------------------------------------------+import Data.Int+import Data.Maybe++--------------------------------------------------------------------------------+import Data.ProtocolBuffers++--------------------------------------------------------------------------------+import Database.EventStore.Internal.Operation+import Database.EventStore.Internal.Operation.Read.Common+import Database.EventStore.Internal.Operation.ReadAllEvents.Message+import Database.EventStore.Internal.Stream+import Database.EventStore.Internal.Types++--------------------------------------------------------------------------------+-- | Batch read on $all stream operation.+readAllEvents :: Settings+              -> Int64+              -> Int64+              -> Int32+              -> Bool+              -> ReadDirection+              -> Operation AllSlice+readAllEvents Settings{..} c_pos p_pos max_c tos dir = do+    let msg = newRequest c_pos p_pos max_c tos s_requireMaster+        cmd = case dir of+            Forward  -> 0xB6+            Backward -> 0xB8++        resp_cmd = case dir of+            Forward  -> 0xB7+            Backward -> 0xB9+    resp <- send cmd resp_cmd msg+    let r      = getField $ _Result resp+        err    = getField $ _Error resp+        nc_pos = getField $ _NextCommitPosition resp+        np_pos = getField $ _NextPreparePosition resp+        es     = getField $ _Events resp+        evts   = fmap newResolvedEventFromBuf es+        eos    = null evts+        f_pos  = Position c_pos p_pos+        n_pos  = Position nc_pos np_pos+        slice  = AllSlice f_pos n_pos dir evts eos+    case fromMaybe SUCCESS r of+        ERROR         -> serverError err+        ACCESS_DENIED -> accessDenied AllStream+        _             -> yield slice

@@ -0,0 +1,84 @@+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DataKinds     #-}+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Internal.Operation.ReadAllEvents.Message+-- Copyright : (C) 2015 Yorick Laupa+-- License : (see the file LICENSE)+--+-- Maintainer : Yorick Laupa <yo.eight@gmail.com>+-- Stability : provisional+-- Portability : non-portable+--+--------------------------------------------------------------------------------+module Database.EventStore.Internal.Operation.ReadAllEvents.Message where++--------------------------------------------------------------------------------+import Data.Int+import GHC.Generics++--------------------------------------------------------------------------------+import Data.ProtocolBuffers+import Data.Text++--------------------------------------------------------------------------------+import Database.EventStore.Internal.Types++--------------------------------------------------------------------------------+-- | Batch read on $all stream request.+data Request+    = Request+      { _commitPosition  :: Required 1 (Value Int64)+      , _preparePosition :: Required 2 (Value Int64)+      , _maxCount        :: Required 3 (Value Int32)+      , _resolveLinkTos  :: Required 4 (Value Bool)+      , _requireMaster   :: Required 5 (Value Bool)+      }+    deriving (Generic, Show)++--------------------------------------------------------------------------------+instance Encode Request++--------------------------------------------------------------------------------+-- | 'Request' smart constructor.+newRequest :: Int64+           -> Int64+           -> Int32+           -> Bool+           -> Bool+           -> Request+newRequest c_pos p_pos max_c res_link_tos req_master =+    Request+    { _commitPosition  = putField c_pos+    , _preparePosition = putField p_pos+    , _maxCount        = putField max_c+    , _resolveLinkTos  = putField res_link_tos+    , _requireMaster   = putField req_master+    }++--------------------------------------------------------------------------------+-- | Enumeration detailing the possible outcomes of reading a slice of $all+--   stream.+data Result+    = SUCCESS+    | NOT_MODIFIED+    | ERROR+    | ACCESS_DENIED+    deriving (Eq, Enum, Show)++--------------------------------------------------------------------------------+-- | Batch read on $all stream response.+data Response+    = Response+      { _CommitPosition      :: Required 1 (Value Int64)+      , _PreparePosition     :: Required 2 (Value Int64)+      , _Events              :: Repeated 3 (Message ResolvedEventBuf)+      , _NextCommitPosition  :: Required 4 (Value Int64)+      , _NextPreparePosition :: Required 5 (Value Int64)+      , _Result              :: Optional 6 (Enumeration Result)+      , _Error               :: Optional 7 (Value Text)+      }+    deriving (Generic, Show)++--------------------------------------------------------------------------------+instance Decode Response

@@ -1,200 +0,0 @@-{-# LANGUAGE DeriveGeneric     #-}-{-# LANGUAGE DataKinds         #-}-{-# LANGUAGE OverloadedStrings #-}------------------------------------------------------------------------------------ |--- Module : Database.EventStore.Internal.Operation.ReadAllEventsOperation--- 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.Internal.Operation.ReadAllEventsOperation-    ( AllEventsSlice(..)-    , ReadAllResult(..)-    , readAllEventsOperation-    ) where-----------------------------------------------------------------------------------import Control.Concurrent-import Data.Int-import Data.Maybe-import GHC.Generics (Generic)-----------------------------------------------------------------------------------import Data.Text hiding (null)-import Data.ProtocolBuffers-----------------------------------------------------------------------------------import Database.EventStore.Internal.Manager.Operation-import Database.EventStore.Internal.Types-----------------------------------------------------------------------------------data ReadAllEvents-    = ReadAllEvents-      { readAllEventsCommitPosition  :: Required 1 (Value Int64)-      , readAllEventsPreparePosition :: Required 2 (Value Int64)-      , readAllEventsMaxCount        :: Required 3 (Value Int32)-      , readAllEventsResolveLinkTos  :: Required 4 (Value Bool)-      , readAllEventsRequireMaster   :: Required 5 (Value Bool)-      }-    deriving (Generic, Show)-----------------------------------------------------------------------------------instance Encode ReadAllEvents-----------------------------------------------------------------------------------newReadAllEvents :: Int64-                 -> Int64-                 -> Int32-                 -> Bool-                 -> Bool-                 -> ReadAllEvents-newReadAllEvents c_pos p_pos max_c res_link_tos req_master =-    ReadAllEvents-    { readAllEventsCommitPosition  = putField c_pos-    , readAllEventsPreparePosition = putField p_pos-    , readAllEventsMaxCount        = putField max_c-    , readAllEventsResolveLinkTos  = putField res_link_tos-    , readAllEventsRequireMaster   = putField req_master-    }------------------------------------------------------------------------------------- | Enumeration detailing the possible outcomes of reading a slice of $all---   stream.-data ReadAllResult-    = RA_SUCCESS-    | RA_NOT_MODIFIED-    | RA_ERROR-    | RA_ACCESS_DENIED-    deriving (Eq, Enum, Show)-----------------------------------------------------------------------------------data ReadAllEventsCompleted-    = ReadAllEventsCompleted-      { readAECCommitPosition      :: Required 1 (Value Int64)-      , readAECPreparePosition     :: Required 2 (Value Int64)-      , readAECEvents              :: Repeated 3 (Message ResolvedEventBuf)-      , readAECNextCommitPosition  :: Required 4 (Value Int64)-      , readAECNextPreparePosition :: Required 5 (Value Int64)-      , readAECResult              :: Optional 6 (Enumeration ReadAllResult)-      , readAECError               :: Optional 7 (Value Text)-      }-    deriving (Generic, Show)-----------------------------------------------------------------------------------instance Decode ReadAllEventsCompleted------------------------------------------------------------------------------------- | The result of a read operation from the $all stream.-data AllEventsSlice-    = AllEventsSlice-      { allEventsSliceResult :: !ReadAllResult-        -- ^ Representing the status of the read attempt.-      , allEventsSliceFrom :: !Position-        -- ^ Representing the position where the next slice should be read-        --   from.-      , allEventsSliceNext :: !Position-        -- ^ Representing the position where the next slice should be read from.-      , allEventsSliceIsEOS :: !Bool-        -- ^ Representing whether or not this is the end of the $all stream.-      , allEventsSliceEvents :: ![ResolvedEvent]-        -- ^ The events read.-      , allEventsSliceDirection :: !ReadDirection-        -- ^ The direction of read request.-      }-    deriving Show-----------------------------------------------------------------------------------newAllEventsSlice :: ReadDirection -> ReadAllEventsCompleted -> AllEventsSlice-newAllEventsSlice dir raec = aes-  where-    res      = fromMaybe RA_SUCCESS (getField $ readAECResult raec)-    evts     = fmap newResolvedEventFromBuf (getField $ readAECEvents raec)-    r_com    = getField $ readAECCommitPosition raec-    r_pre    = getField $ readAECPreparePosition raec-    r_n_com  = getField $ readAECNextCommitPosition raec-    r_n_pre  = getField $ readAECNextPreparePosition raec-    from_pos = Position r_com r_pre-    next_pos = Position r_n_com r_n_pre--    aes = AllEventsSlice-          { allEventsSliceResult    = res-          , allEventsSliceFrom      = from_pos-          , allEventsSliceNext      = next_pos-          , allEventsSliceIsEOS     = null evts-          , allEventsSliceEvents    = evts-          , allEventsSliceDirection = dir-          }-----------------------------------------------------------------------------------readAllEventsOperation :: Settings-                       -> ReadDirection-                       -> MVar (OperationExceptional AllEventsSlice)-                       -> Int64-                       -> Int64-                       -> Int32-                       -> Bool-                       -> OperationParams-readAllEventsOperation settings dir mvar c_pos p_pos max_c res_link_tos =-    OperationParams-    { opSettings    = settings-    , opRequestCmd  = req-    , opResponseCmd = resp--    , opRequest =-        let req_master = s_requireMaster settings-            request    = newReadAllEvents c_pos-                                          p_pos-                                          max_c-                                          res_link_tos-                                          req_master in-         return request--    , opSuccess = inspect mvar dir-    , opFailure = failed mvar-    }-  where-    req = case dir of-              Forward  -> 0xB6-              Backward -> 0xB8--    resp = case dir of-               Forward  -> 0xB7-               Backward -> 0xB9-----------------------------------------------------------------------------------inspect :: MVar (OperationExceptional AllEventsSlice)-        -> ReadDirection-        -> ReadAllEventsCompleted-        -> IO Decision-inspect mvar dir raec = go res-  where-    res     = fromMaybe RA_SUCCESS (getField $ readAECResult raec)-    may_err = getField $ readAECError raec--    go RA_ERROR         = failed mvar (ServerError may_err)-    go RA_ACCESS_DENIED = failed mvar (AccessDenied "$all")-    go _                = succeed mvar dir raec-----------------------------------------------------------------------------------succeed :: MVar (OperationExceptional AllEventsSlice)-        -> ReadDirection-        -> ReadAllEventsCompleted-        -> IO Decision-succeed mvar dir raec = do-    putMVar mvar (Right ses)-    return EndOperation-  where-    ses = newAllEventsSlice dir raec-----------------------------------------------------------------------------------failed :: MVar (OperationExceptional AllEventsSlice)-       -> OperationException-       -> IO Decision-failed mvar e = do-    putMVar mvar (Left e)-    return EndOperation

@@ -0,0 +1,67 @@+{-# LANGUAGE DataKinds       #-}+{-# LANGUAGE RecordWildCards #-}+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Internal.Operation.DeleteStream+-- Copyright : (C) 2015 Yorick Laupa+-- License : (see the file LICENSE)+--+-- Maintainer : Yorick Laupa <yo.eight@gmail.com>+-- Stability : provisional+-- Portability : non-portable+--+--------------------------------------------------------------------------------+module Database.EventStore.Internal.Operation.ReadEvent+    ( ReadEvent(..)+    , readEvent+    ) where++--------------------------------------------------------------------------------+import Data.Int++--------------------------------------------------------------------------------+import Data.ProtocolBuffers+import Data.Text++--------------------------------------------------------------------------------+import Database.EventStore.Internal.Operation+import Database.EventStore.Internal.Operation.ReadEvent.Message+import Database.EventStore.Internal.Operation.Read.Common+import Database.EventStore.Internal.Stream+import Database.EventStore.Internal.Types++--------------------------------------------------------------------------------+-- | Represents the result of looking up a specific event number from a stream.+data ReadEvent+    = ReadEventNotFound+      { readEventStream :: !Text+      , readEventNumber :: !Int32+      }+    | ReadEvent+      { readEventStream   :: !Text+      , readEventNumber   :: !Int32+      , readEventResolved :: !ResolvedEvent+      } deriving Show++--------------------------------------------------------------------------------+-- | Read a specific event given event number operation.+readEvent :: Settings+          -> Text+          -> Int32+          -> Bool+          -> Operation (ReadResult 'RegularStream ReadEvent)+readEvent Settings{..} s evtn tos = do+    let msg = newRequest s evtn tos s_requireMaster+    resp <- send 0xB0 0xB1 msg+    let r         = getField $ _result resp+        evt       = newResolvedEvent $ getField $ _indexedEvent resp+        err       = getField $ _error resp+        not_found = ReadSuccess $ ReadEventNotFound s evtn+        found     = ReadSuccess $ ReadEvent s evtn evt+    case r of+        NOT_FOUND      -> yield not_found+        NO_STREAM      -> yield ReadNoStream+        STREAM_DELETED -> yield $ ReadStreamDeleted s+        ERROR          -> yield (ReadError err)+        ACCESS_DENIED  -> yield $ ReadAccessDenied $ StreamName s+        SUCCESS        -> yield found

@@ -0,0 +1,75 @@+{-# LANGUAGE DeriveGeneric         #-}+{-# LANGUAGE DataKinds             #-}+{-# OPTIONS_GHC -fcontext-stack=26 #-}+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Internal.Operation.ReadEvent.Message+-- Copyright : (C) 2015 Yorick Laupa+-- License : (see the file LICENSE)+--+-- Maintainer : Yorick Laupa <yo.eight@gmail.com>+-- Stability : provisional+-- Portability : non-portable+--+--------------------------------------------------------------------------------+module Database.EventStore.Internal.Operation.ReadEvent.Message where++--------------------------------------------------------------------------------+import Data.Int+import GHC.Generics++--------------------------------------------------------------------------------+import Data.ProtocolBuffers+import Data.Text++--------------------------------------------------------------------------------+import Database.EventStore.Internal.Types++--------------------------------------------------------------------------------+-- | Read event on a regular stream request.+data Request+    = Request+      { _streamId       :: Required 1 (Value Text)+      , _eventNumber    :: Required 2 (Value Int32)+      , _resolveLinkTos :: Required 3 (Value Bool)+      , _requireMaster  :: Required 4 (Value Bool)+      }+    deriving (Generic, Show)++--------------------------------------------------------------------------------+instance Encode Request++--------------------------------------------------------------------------------+-- | 'Request' smart constructor.+newRequest :: Text -> Int32 -> Bool -> Bool -> Request+newRequest stream_id evt_num res_link_tos req_master =+    Request+    { _streamId       = putField stream_id+    , _eventNumber    = putField evt_num+    , _resolveLinkTos = putField res_link_tos+    , _requireMaster  = putField req_master+    }++--------------------------------------------------------------------------------+-- | Enumeration representing the status of a single event read operation.+data Result+    = SUCCESS+    | NOT_FOUND+    | NO_STREAM+    | STREAM_DELETED+    | ERROR+    | ACCESS_DENIED+    deriving (Eq, Enum, Show)++--------------------------------------------------------------------------------+-- | Read event on a regular stream response.+data Response+    = Response+      { _result       :: Required 1 (Enumeration Result)+      , _indexedEvent :: Required 2 (Message ResolvedIndexedEvent)+      , _error        :: Optional 3 (Value Text)+      }+    deriving (Generic, Show)++--------------------------------------------------------------------------------+instance Decode Response

@@ -1,170 +0,0 @@-{-# LANGUAGE    DeriveGeneric      #-}-{-# LANGUAGE    DataKinds          #-}-{-# OPTIONS_GHC -fcontext-stack=26 #-}------------------------------------------------------------------------------------ |--- Module : Database.EventStore.Internal.Operation.ReadEventOperation--- 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.Internal.Operation.ReadEventOperation-    ( ReadResult(..)-    , ReadEventResult(..)-    , readEventOperation-    ) where-----------------------------------------------------------------------------------import Control.Concurrent-import Data.Int-import GHC.Generics (Generic)-----------------------------------------------------------------------------------import Data.ProtocolBuffers-import Data.Text-----------------------------------------------------------------------------------import Database.EventStore.Internal.Manager.Operation-import Database.EventStore.Internal.Types-----------------------------------------------------------------------------------data ReadEvent-    = ReadEvent-      { readEventStreamId       :: Required 1 (Value Text)-      , readEventNumber         :: Required 2 (Value Int32)-      , readEventResolveLinkTos :: Required 3 (Value Bool)-      , readEventRequireMaster  :: Required 4 (Value Bool)-      }-    deriving (Generic, Show)-----------------------------------------------------------------------------------instance Encode ReadEvent-----------------------------------------------------------------------------------newReadEvent :: Text -> Int32 -> Bool -> Bool -> ReadEvent-newReadEvent stream_id evt_num res_link_tos req_master =-    ReadEvent-    { readEventStreamId       = putField stream_id-    , readEventNumber         = putField evt_num-    , readEventResolveLinkTos = putField res_link_tos-    , readEventRequireMaster  = putField req_master-    }------------------------------------------------------------------------------------- | Enumeration representing the status of a single event read operation.-data ReadEventResult-    = RE_SUCCESS-    | RE_NOT_FOUND-    | RE_NO_STREAM-    | RE_STREAM_DELETED-    | RE_ERROR-    | RE_ACCESS_DENIED-    deriving (Eq, Enum, Show)-----------------------------------------------------------------------------------data ReadEventCompleted-    = ReadEventCompleted-      { readCompletedResult       :: Required 1 (Enumeration ReadEventResult)-      , readCompletedIndexedEvent :: Required 2 (Message ResolvedIndexedEvent)-      , readCompletedError        :: Optional 3 (Value Text)-      }-    deriving (Generic, Show)-----------------------------------------------------------------------------------instance Decode ReadEventCompleted------------------------------------------------------------------------------------- | Result of a single event read operation to the EventStore.-data ReadResult-    = ReadResult-      { readResultStatus        :: !ReadEventResult       -- ^ Attempt status-      , readResultStreamId      :: !Text                  -- ^ Stream name-      , readResultEventNumber   :: !Int32                 -- ^ Event number-      , readResultResolvedEvent :: !(Maybe ResolvedEvent)-      }-    deriving Show-----------------------------------------------------------------------------------newReadResult :: ReadEventResult-              -> Text-              -> Int32-              -> ResolvedIndexedEvent-              -> ReadResult-newReadResult status stream_id evt_num rie = rr-  where-    may_re =-        case status of-            RE_SUCCESS -> Just $ newResolvedEvent rie-            _          -> Nothing--    rr = ReadResult-         { readResultStatus        = status-         , readResultStreamId      = stream_id-         , readResultEventNumber   = evt_num-         , readResultResolvedEvent = may_re-         }-----------------------------------------------------------------------------------readEventOperation :: Settings-                   -> MVar (OperationExceptional ReadResult)-                   -> Text-                   -> Int32-                   -> Bool -- ^ Resolve link TOS-                   -> OperationParams-readEventOperation settings mvar stream_id evt_num res_link_tos =-    OperationParams-    { opSettings    = settings-    , opRequestCmd  = 0xB0-    , opResponseCmd = 0xB1--    , opRequest =-        let req_master = s_requireMaster settings-            request    = newReadEvent stream_id-                                      evt_num-                                      res_link_tos-                                      req_master in-         return request--    , opSuccess = inspect mvar stream_id evt_num-    , opFailure = failed mvar-    }-----------------------------------------------------------------------------------inspect :: MVar (OperationExceptional ReadResult)-        -> Text-        -> Int32-        -> ReadEventCompleted-        -> IO Decision-inspect mvar stream_id evt_num reco = go (getField $ readCompletedResult reco)-  where-    may_err = getField $ readCompletedError reco--    go RE_ERROR         = failed mvar (ServerError may_err)-    go RE_ACCESS_DENIED = failed mvar (AccessDenied stream_id)-    go _                = succeed mvar stream_id evt_num reco-----------------------------------------------------------------------------------succeed :: MVar (OperationExceptional ReadResult)-        -> Text-        -> Int32-        -> ReadEventCompleted-        -> IO Decision-succeed mvar stream_id evt_num reco = do-    putMVar mvar (Right rr)-    return EndOperation-  where-    status = getField $ readCompletedResult reco-    rie    = getField $ readCompletedIndexedEvent reco-    rr     = newReadResult status stream_id evt_num rie-----------------------------------------------------------------------------------failed :: MVar (OperationExceptional ReadResult)-       -> OperationException-       -> IO Decision-failed mvar e = do-    putMVar mvar (Left e)-    return EndOperation

@@ -0,0 +1,66 @@+{-# LANGUAGE DataKinds       #-}+{-# LANGUAGE RecordWildCards #-}+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Internal.Operation.ReadStreamEvents+-- Copyright : (C) 2015 Yorick Laupa+-- License : (see the file LICENSE)+--+-- Maintainer : Yorick Laupa <yo.eight@gmail.com>+-- Stability : provisional+-- Portability : non-portable+--+--------------------------------------------------------------------------------+module Database.EventStore.Internal.Operation.ReadStreamEvents+    ( readStreamEvents ) where++--------------------------------------------------------------------------------+import Data.Int++--------------------------------------------------------------------------------+import Data.ProtocolBuffers+import Data.Text++--------------------------------------------------------------------------------+import Database.EventStore.Internal.Operation+import Database.EventStore.Internal.Operation.Read.Common+import Database.EventStore.Internal.Operation.ReadStreamEvents.Message+import Database.EventStore.Internal.Stream+import Database.EventStore.Internal.Types++--------------------------------------------------------------------------------+-- | Batch read from a regular stream operation.+readStreamEvents :: Settings+                 -> ReadDirection+                 -> Text+                 -> Int32+                 -> Int32+                 -> Bool+                 -> Operation (ReadResult 'RegularStream StreamSlice)+readStreamEvents Settings{..} dir s st cnt tos = do+    let req_cmd =+            case dir of+                Forward  -> 0xB2+                Backward -> 0xB4+        resp_cmd =+            case dir of+                Forward  -> 0xB3+                Backward -> 0xB5++        msg = newRequest s st cnt tos s_requireMaster+    resp <- send req_cmd resp_cmd msg+    let r     = getField $ _result resp+        es    = getField $ _events resp+        evts  = fmap newResolvedEvent es+        err   = getField $ _error resp+        eos   = getField $ _endOfStream resp+        nxt   = getField $ _nextNumber resp+        lst   = getField $ _lastNumber resp+        found = StreamSlice s lst dir st nxt evts eos+    case r of+        NO_STREAM      -> yield ReadNoStream+        STREAM_DELETED -> yield $ ReadStreamDeleted s+        NOT_MODIFIED   -> yield ReadNotModified+        ERROR          -> yield (ReadError err)+        ACCESS_DENIED  -> yield $ ReadAccessDenied $ StreamName s+        SUCCESS        -> yield (ReadSuccess found)

@@ -0,0 +1,80 @@+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DataKinds     #-}+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Internal.Operation.ReadStreamEvents.Message+-- Copyright : (C) 2015 Yorick Laupa+-- License : (see the file LICENSE)+--+-- Maintainer : Yorick Laupa <yo.eight@gmail.com>+-- Stability : provisional+-- Portability : non-portable+--+--------------------------------------------------------------------------------+module Database.EventStore.Internal.Operation.ReadStreamEvents.Message where++--------------------------------------------------------------------------------+import Data.Int+import GHC.Generics++--------------------------------------------------------------------------------+import Data.ProtocolBuffers+import Data.Text++--------------------------------------------------------------------------------+import Database.EventStore.Internal.Types++--------------------------------------------------------------------------------+-- | Batch read on regular stream request.+data Request+    = Request+      { _streamId       :: Required 1 (Value Text)+      , _eventNumber    :: Required 2 (Value Int32)+      , _maxCount       :: Required 3 (Value Int32)+      , _resolveLinkTos :: Required 4 (Value Bool)+      , _requireMaster  :: Required 5 (Value Bool)+      }+    deriving (Generic, Show)++--------------------------------------------------------------------------------+-- | 'Request' smart constructor.+newRequest :: Text -> Int32 -> Int32 -> Bool -> Bool -> Request+newRequest stream_id evt_num max_c res_link_tos req_master =+    Request+    { _streamId       = putField stream_id+    , _eventNumber    = putField evt_num+    , _maxCount       = putField max_c+    , _resolveLinkTos = putField res_link_tos+    , _requireMaster  = putField req_master+    }++--------------------------------------------------------------------------------+instance Encode Request++--------------------------------------------------------------------------------+-- | Enumeration detailing the possible outcomes of reading a slice of a stream+data Result+    = SUCCESS+    | NO_STREAM+    | STREAM_DELETED+    | NOT_MODIFIED+    | ERROR+    | ACCESS_DENIED+    deriving (Eq, Enum, Show)++--------------------------------------------------------------------------------+-- | Batch read on regular stream response.+data Response+    = Response+      { _events             :: Repeated 1 (Message ResolvedIndexedEvent)+      , _result             :: Required 2 (Enumeration Result)+      , _nextNumber         :: Required 3 (Value Int32)+      , _lastNumber         :: Required 4 (Value Int32)+      , _endOfStream        :: Required 5 (Value Bool)+      , _lastCommitPosition :: Required 6 (Value Int64)+      , _error              :: Optional 7 (Value Text)+      }+    deriving (Generic, Show)++--------------------------------------------------------------------------------+instance Decode Response

@@ -1,205 +0,0 @@-{-# LANGUAGE DeriveGeneric #-}-{-# LANGUAGE DataKinds     #-}------------------------------------------------------------------------------------ |--- Module : Database.EventStore.Internal.Operation.ReadStreamEventsOperation--- 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.Internal.Operation.ReadStreamEventsOperation-    ( StreamEventsSlice(..)-    , ReadStreamResult(..)-    , readStreamEventsOperation-    ) where-----------------------------------------------------------------------------------import Control.Concurrent-import Data.Int-import GHC.Generics (Generic)-----------------------------------------------------------------------------------import Data.ProtocolBuffers-import Data.Text-----------------------------------------------------------------------------------import Database.EventStore.Internal.Manager.Operation-import Database.EventStore.Internal.Types-----------------------------------------------------------------------------------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)-      }-    deriving (Generic, Show)-----------------------------------------------------------------------------------newReadStreamEvents :: Text-                    -> Int32-                    -> Int32-                    -> Bool-                    -> Bool-                    -> 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-    }-----------------------------------------------------------------------------------instance Encode ReadStreamEvents------------------------------------------------------------------------------------- | Enumeration detailing the possible outcomes of reading a slice of a stream-data ReadStreamResult-    = RS_SUCCESS-    | RS_NO_STREAM-    | RS_STREAM_DELETED-    | RS_NOT_MODIFIED-    | RS_ERROR-    | RS_ACCESS_DENIED-    deriving (Eq, Enum, Show)-----------------------------------------------------------------------------------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)-      }-    deriving (Generic, Show)-----------------------------------------------------------------------------------instance Decode ReadStreamEventsCompleted------------------------------------------------------------------------------------- | Represents the result of a single read operation to the EventStore.-data StreamEventsSlice-    = StreamEventsSlice-      { streamEventsSliceResult :: !ReadStreamResult-        -- ^ Representing the status of the read attempt.-      , streamEventsSliceStreamId :: !Text-        -- ^ The name of the stream read.-      , streamEventsSliceStart :: !Int32-        -- ^ The starting point (represented as a sequence number) of the read-        --   operation.-      , streamEventsSliceNext :: !Int32-        -- ^ The next event number that can be read.-      , streamEventsSliceLast :: !Int32-        -- ^ The last event number in the stream.-      , streamEventsSliceIsEOS :: !Bool-        -- ^ Representing whether or not this is the end of the stream.-      , streamEventsSliceEvents :: ![ResolvedEvent]-        -- ^ The events read represented as 'ResolvedEvent'-      , streamEventsSliceDirection :: !ReadDirection-        -- ^ The direction of the read request.-      }-    deriving Show-----------------------------------------------------------------------------------newStreamEventsSlice :: Text-                     -> Int32-                     -> ReadDirection-                     -> ReadStreamEventsCompleted-                     -> StreamEventsSlice-newStreamEventsSlice stream_id start dir reco = ses-  where-    evts = getField $ _readSECEvents reco--    ses = StreamEventsSlice-          { streamEventsSliceResult    = getField $ _readSECResult reco-          , streamEventsSliceStreamId  = stream_id-          , streamEventsSliceStart     = start-          , streamEventsSliceNext      = getField $ _readSECNextNumber reco-          , streamEventsSliceLast      = getField $ _readSECLastNumber reco-          , streamEventsSliceIsEOS     = getField $ _readSECEndOfStream reco-          , streamEventsSliceEvents    = fmap newResolvedEvent evts-          , streamEventsSliceDirection = dir-          }-----------------------------------------------------------------------------------readStreamEventsOperation :: Settings-                          -> ReadDirection-                          -> MVar (OperationExceptional StreamEventsSlice)-                          -> Text-                          -> Int32-                          -> Int32-                          -> Bool-                          -> OperationParams-readStreamEventsOperation settings dir mvar stream_id start cnt res_link_tos =-    OperationParams-    { opSettings    = settings-    , opRequestCmd  = req-    , opResponseCmd = resp--    , opRequest =-        let req_master = s_requireMaster settings-            request    = newReadStreamEvents stream_id-                                             start-                                             cnt-                                             res_link_tos-                                             req_master in-         return request--    , opSuccess = inspect mvar dir stream_id start-    , opFailure = failed mvar-    }-  where-    req = case dir of-              Forward  -> 0xB2-              Backward -> 0xB4--    resp = case dir of-               Forward  -> 0xB3-               Backward -> 0xB5-----------------------------------------------------------------------------------inspect :: MVar (OperationExceptional StreamEventsSlice)-        -> ReadDirection-        -> Text-        -> Int32-        -> ReadStreamEventsCompleted-        -> IO Decision-inspect mvar dir stream_id start rsec = go (getField $ _readSECResult rsec)-  where-    may_err = getField $ _readSECError rsec--    go RS_ERROR         = failed mvar (ServerError may_err)-    go RS_ACCESS_DENIED = failed mvar (AccessDenied stream_id)-    go _                = succeed mvar dir stream_id start rsec-----------------------------------------------------------------------------------succeed :: MVar (OperationExceptional StreamEventsSlice)-        -> ReadDirection-        -> Text-        -> Int32-        -> ReadStreamEventsCompleted-        -> IO Decision-succeed mvar dir stream_id start rsec = do-    putMVar mvar (Right ses)-    return EndOperation-  where-    ses = newStreamEventsSlice stream_id start dir rsec-----------------------------------------------------------------------------------failed :: MVar (OperationExceptional StreamEventsSlice)-       -> OperationException-       -> IO Decision-failed mvar e = do-    putMVar mvar (Left e)-    return EndOperation

@@ -0,0 +1,89 @@+{-# LANGUAGE DataKinds         #-}+{-# LANGUAGE GADTs             #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE Rank2Types        #-}+{-# LANGUAGE RecordWildCards   #-}+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Internal.Operation.StreamMetadata+-- Copyright : (C) 2015 Yorick Laupa+-- License : (see the file LICENSE)+--+-- Maintainer : Yorick Laupa <yo.eight@gmail.com>+-- Stability : provisional+-- Portability : non-portable+--+--------------------------------------------------------------------------------+module Database.EventStore.Internal.Operation.StreamMetadata+    ( readMetaStream+    , setMetaStream+    ) where++--------------------------------------------------------------------------------+import Data.Int+import Data.Monoid ((<>))++--------------------------------------------------------------------------------+import Data.Aeson (decode)+import Data.ByteString.Lazy (fromStrict)+import Data.Text (Text)++--------------------------------------------------------------------------------+import Database.EventStore.Internal.Operation+import Database.EventStore.Internal.Operation.Read.Common+import Database.EventStore.Internal.Operation.ReadEvent+import Database.EventStore.Internal.Operation.Write.Common+import Database.EventStore.Internal.Operation.WriteEvents+import Database.EventStore.Internal.Stream+import Database.EventStore.Internal.Types++--------------------------------------------------------------------------------+metaStream :: Text -> Text+metaStream s = "$$" <> s++--------------------------------------------------------------------------------+-- | Read stream metadata operation.+readMetaStream :: Settings -> Text -> Operation StreamMetadataResult+readMetaStream setts s =+    foreach (readEvent setts (metaStream s) (-1) False) $ \tmp -> do+        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++--------------------------------------------------------------------------------+-- | Set stream metadata operation.+setMetaStream :: Settings+              -> Text+              -> ExpectedVersion+              -> StreamMetadata+              -> Operation WriteResult+setMetaStream setts s v meta =+    let stream = metaStream s+        json   = streamMetadataJSON meta+        evt    = createEvent "$metadata" Nothing (withJson json)+        inner  = writeEvents setts stream v [evt] in+    foreach inner yield++--------------------------------------------------------------------------------+invalidFormat :: OperationError+invalidFormat = InvalidOperation "Invalid metadata format"++--------------------------------------------------------------------------------+streamNotFound :: OperationError+streamNotFound = InvalidOperation "Read metadata on an inexistant stream"++--------------------------------------------------------------------------------+onReadResult :: ReadResult 'RegularStream ReadEvent+             -> (Text -> Int32 -> ResolvedEvent -> SM a b)+             -> SM a b+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

@@ -0,0 +1,109 @@+{-# LANGUAGE DataKinds       #-}+{-# LANGUAGE GADTs           #-}+{-# LANGUAGE KindSignatures  #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE TypeFamilies    #-}+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Internal.Operation.Transaction+-- Copyright : (C) 2015 Yorick Laupa+-- License : (see the file LICENSE)+--+-- Maintainer : Yorick Laupa <yo.eight@gmail.com>+-- Stability : provisional+-- Portability : non-portable+--+--------------------------------------------------------------------------------+module Database.EventStore.Internal.Operation.Transaction+    ( transactionStart+    , transactionWrite+    , transactionCommit+    ) where++--------------------------------------------------------------------------------+import Data.Int+import Data.Maybe+import Data.Traversable++--------------------------------------------------------------------------------+import Data.ProtocolBuffers+import Data.Text++--------------------------------------------------------------------------------+import Database.EventStore.Internal.Operation+import Database.EventStore.Internal.Operation.Transaction.Message+import Database.EventStore.Internal.Operation.Write.Common+import Database.EventStore.Internal.Stream+import Database.EventStore.Internal.Types++--------------------------------------------------------------------------------+import Prelude++--------------------------------------------------------------------------------+-- | Start transaction operation.+transactionStart :: Settings -> Text -> ExpectedVersion -> Operation Int64+transactionStart Settings{..} stream exp_v = do+    let msg = newStart stream (expVersionInt32 exp_v) s_requireMaster+    resp <- send 0x84 0x85 msg+    let tid = getField $ _transId resp+        r   = getField $ _result resp+    case r of+        OP_PREPARE_TIMEOUT        -> retry+        OP_FORWARD_TIMEOUT        -> retry+        OP_COMMIT_TIMEOUT         -> retry+        OP_WRONG_EXPECTED_VERSION -> wrongVersion stream exp_v+        OP_STREAM_DELETED         -> streamDeleted stream+        OP_INVALID_TRANSACTION    -> invalidTransaction+        OP_ACCESS_DENIED          -> accessDenied $ StreamName stream+        OP_SUCCESS                -> yield tid++--------------------------------------------------------------------------------+-- | Transactional write operation.+transactionWrite :: Settings+                 -> Text+                 -> ExpectedVersion+                 -> Int64+                 -> [Event]+                 -> Operation ()+transactionWrite Settings{..} stream exp_v trans_id evts = do+    nevts <- traverse eventToNewEvent evts+    let msg = newWrite trans_id nevts s_requireMaster+    resp <- send 0x86 0x87 msg+    let r = getField $ _wwResult resp+    case r of+        OP_PREPARE_TIMEOUT        -> retry+        OP_FORWARD_TIMEOUT        -> retry+        OP_COMMIT_TIMEOUT         -> retry+        OP_WRONG_EXPECTED_VERSION -> wrongVersion stream exp_v+        OP_STREAM_DELETED         -> streamDeleted stream+        OP_INVALID_TRANSACTION    -> invalidTransaction+        OP_ACCESS_DENIED          -> accessDenied $ StreamName stream+        OP_SUCCESS                -> yield ()++--------------------------------------------------------------------------------+-- | Transactional commit operation.+transactionCommit :: Settings+                  -> Text+                  -> ExpectedVersion+                  -> Int64+                  -> Operation WriteResult+transactionCommit Settings{..} stream exp_v trans_id = do+    let msg = newCommit trans_id s_requireMaster+    resp <- send 0x88 0x89 msg+    let r = getField $ _ccResult resp+        com_pos = getField $ _commitPosition resp+        pre_pos = getField $ _preparePosition resp+        lst_num = getField $ _lastNumber resp+        p_int   = fromMaybe (-1) pre_pos+        c_int   = fromMaybe (-1) com_pos+        pos     = Position c_int p_int+        res     = WriteResult lst_num pos+    case r of+        OP_PREPARE_TIMEOUT        -> retry+        OP_FORWARD_TIMEOUT        -> retry+        OP_COMMIT_TIMEOUT         -> retry+        OP_WRONG_EXPECTED_VERSION -> wrongVersion stream exp_v+        OP_STREAM_DELETED         -> streamDeleted stream+        OP_INVALID_TRANSACTION    -> invalidTransaction+        OP_ACCESS_DENIED          -> accessDenied $ StreamName stream+        OP_SUCCESS                -> yield res

@@ -0,0 +1,136 @@+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DataKinds     #-}+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Internal.Operation.Transaction.Message+-- Copyright : (C) 2015 Yorick Laupa+-- License : (see the file LICENSE)+--+-- Maintainer : Yorick Laupa <yo.eight@gmail.com>+-- Stability : provisional+-- Portability : non-portable+--+--------------------------------------------------------------------------------+module Database.EventStore.Internal.Operation.Transaction.Message where++--------------------------------------------------------------------------------+import Data.Int+import GHC.Generics++--------------------------------------------------------------------------------+import Data.ProtocolBuffers+import Data.Text++--------------------------------------------------------------------------------+import Database.EventStore.Internal.Operation+import Database.EventStore.Internal.Types++--------------------------------------------------------------------------------+-- | Start transaction request.+data Start =+    Start+    { _streamId        :: Required 1 (Value Text)+    , _expectedVersion :: Required 2 (Value Int32)+    , _requireMaster   :: Required 3 (Value Bool)+    }+    deriving (Generic, Show)++--------------------------------------------------------------------------------+instance Encode Start++--------------------------------------------------------------------------------+-- | 'Start' smart constructor.+newStart :: Text -> Int32 -> Bool -> Start+newStart stream_id exp_ver req_master =+    Start+    { _streamId        = putField stream_id+    , _expectedVersion = putField exp_ver+    , _requireMaster   = putField req_master+    }++--------------------------------------------------------------------------------+-- | Start transaction response.+data Started =+    Started+    { _transId :: Required 1 (Value Int64)+    , _result  :: Required 2 (Enumeration OpResult)+    , _message :: Optional 3 (Value Text)+    }+    deriving (Generic, Show)++--------------------------------------------------------------------------------+instance Decode Started++--------------------------------------------------------------------------------+-- | Write transactional events request.+data Write =+    Write+    { _wTransId       :: Required 1 (Value Int64)+    , _events         :: Repeated 2 (Message NewEvent)+    , _wRequireMaster :: Required 3 (Value Bool)+    }+    deriving (Generic, Show)++--------------------------------------------------------------------------------+instance Encode Write++--------------------------------------------------------------------------------+-- | 'Write' smart constructor.+newWrite :: Int64 -> [NewEvent] -> Bool -> Write+newWrite trans_id evts req_master =+    Write+    { _wTransId       = putField trans_id+    , _events         = putField evts+    , _wRequireMaster = putField req_master+    }++--------------------------------------------------------------------------------+-- | Write transactional events response.+data Written =+    Written+    { _wwTransId :: Required 1 (Value Int64)+    , _wwResult  :: Required 2 (Enumeration OpResult)+    , _wwMessage :: Optional 3 (Value Text)+    }+    deriving (Generic, Show)++--------------------------------------------------------------------------------+instance Decode Written++--------------------------------------------------------------------------------+-- | Commit transaction request.+data Commit =+    Commit+    { _cTransId       :: Required 1 (Value Int64)+    , _cRequireMaster :: Required 2 (Value Bool)+    }+    deriving (Generic, Show)++--------------------------------------------------------------------------------+instance Encode Commit++--------------------------------------------------------------------------------+-- | 'Commit' smart constructor.+newCommit :: Int64 -> Bool -> Commit+newCommit trans_id req_master =+    Commit+    { _cTransId       = putField trans_id+    , _cRequireMaster = putField req_master+    }++--------------------------------------------------------------------------------+-- | Commit transaction response.+data Committed =+    Committed+    { _ccTransId       :: Required 1 (Value Int64)+    , _ccResult        :: Required 2 (Enumeration OpResult)+    , _ccMessage       :: Optional 3 (Value Text)+    , _firstNumber     :: Required 4 (Value Int32)+    , _lastNumber      :: Required 5 (Value Int32)+    , _preparePosition :: Optional 6 (Value Int64)+    , _commitPosition  :: Optional 7 (Value Int64)+    }+    deriving (Generic, Show)++--------------------------------------------------------------------------------+instance Decode Committed

@@ -1,283 +0,0 @@-{-# LANGUAGE RecordWildCards #-}------------------------------------------------------------------------------------ |--- Module : Database.EventStore.Internal.Operation.TransactionStartOperation--- 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.Internal.Operation.TransactionStartOperation-    ( transactionStartOperation ) where-----------------------------------------------------------------------------------import Control.Concurrent-import Control.Exception-import Data.Int-import Data.Maybe-import Data.Traversable-import Prelude-----------------------------------------------------------------------------------import Control.Concurrent.Async-import Data.ProtocolBuffers-import Data.Text-----------------------------------------------------------------------------------import Database.EventStore.Internal.Manager.Operation-import Database.EventStore.Internal.Processor-import Database.EventStore.Internal.Types-----------------------------------------------------------------------------------data TransactionEnv-    = TransactionEnv-      { _transSettings        :: Settings-      , _transProcessor       :: Cmd -> IO ()-      , _transStreamId        :: Text-      , _transExpectedVersion :: ExpectedVersion-      }-----------------------------------------------------------------------------------transactionStartOperation :: Settings-                          -> (Cmd -> IO ())-                          -> MVar (OperationExceptional Transaction)-                          -> Text-                          -> ExpectedVersion-                          -> OperationParams-transactionStartOperation settings procss mvar stream_id exp_ver =-    OperationParams-    { opSettings    = settings-    , opRequestCmd  = 0x84-    , opResponseCmd = 0x85--    , opRequest =-        let req_master  = s_requireMaster settings-            exp_ver_int = expVersionInt32 exp_ver-            request     = newTransactionStart stream_id-                                              exp_ver_int-                                              req_master in-         return request--    , opSuccess = inspectTrans env mvar-    , opFailure = failed mvar-    }-  where-    env = TransactionEnv-          { _transSettings        = settings-          , _transProcessor       = procss-          , _transStreamId        = stream_id-          , _transExpectedVersion = exp_ver-          }-----------------------------------------------------------------------------------inspectTrans :: TransactionEnv-             -> MVar (OperationExceptional Transaction)-             -> TransactionStartCompleted-             -> IO Decision-inspectTrans env mvar tsc = go (getField $ transactionSCResult tsc)-  where-    go OP_SUCCESS                = succeedTrans env mvar tsc-    go OP_PREPARE_TIMEOUT        = return Retry-    go OP_FORWARD_TIMEOUT        = return Retry-    go OP_COMMIT_TIMEOUT         = return Retry-    go OP_WRONG_EXPECTED_VERSION = failed mvar wrong_version-    go OP_STREAM_DELETED         = failed mvar (StreamDeleted stream_id)-    go OP_INVALID_TRANSACTION    = failed mvar InvalidTransaction-    go OP_ACCESS_DENIED          = failed mvar (AccessDenied stream_id)--    exp_ver       = _transExpectedVersion env-    stream_id     = _transStreamId env-    wrong_version = WrongExpectedVersion stream_id exp_ver-----------------------------------------------------------------------------------succeedTrans :: TransactionEnv-             -> MVar (OperationExceptional Transaction)-             -> TransactionStartCompleted-             -> IO Decision-succeedTrans env mvar tsc = do-    putMVar mvar (Right trans)-    return EndOperation-  where-    trans_id = getField $ transactionSCId tsc-    trans    = createTransaction env trans_id-----------------------------------------------------------------------------------failed :: MVar (OperationExceptional a)-       -> OperationException-       -> IO Decision-failed mvar e = do-    putMVar mvar (Left e)-    return EndOperation-----------------------------------------------------------------------------------createTransaction :: TransactionEnv -> Int64 -> Transaction-createTransaction env@TransactionEnv{..} trans_id = trans-  where-    trans = Transaction-            { transactionId              = trans_id-            , transactionStreamId        = _transStreamId-            , transactionExpectedVersion = _transExpectedVersion--            , transactionCommit = do-                  (as, mvar) <- createAsync--                  let op = transactionCommitOperation env trans_id mvar--                  _transProcessor (NewOperation op)-                  return as--            , transactionSendEvents = \evts -> do-                  (as, mvar) <- createAsync--                  let op = transactionWriteOperation env trans_id mvar evts--                  _transProcessor (NewOperation op)-                  return as--            , transactionRollback = return ()-            }-----------------------------------------------------------------------------------transactionWriteOperation :: TransactionEnv-                          -> Int64-                          -> MVar (OperationExceptional ())-                          -> [Event]-                          -> OperationParams-transactionWriteOperation env trans_id mvar evts =-    OperationParams-    { opSettings    = settings-    , opRequestCmd  = 0x86-    , opResponseCmd = 0x87--    , opRequest = do-        new_evts <- traverse eventToNewEvent evts--        let request = newTransactionWrite trans_id-                      new_evts-                      req_master--        return request--    , opSuccess = inspectWrite env mvar-    , opFailure = failed mvar-    }-  where-    settings   = _transSettings env-    req_master = s_requireMaster settings-----------------------------------------------------------------------------------transactionCommitOperation :: TransactionEnv-                           -> Int64-                           -> MVar (OperationExceptional WriteResult)-                           -> OperationParams-transactionCommitOperation env trans_id mvar =-    OperationParams-    { opSettings    = settings-    , opRequestCmd  = 0x88-    , opResponseCmd = 0x89--    , opRequest =-        let request = newTransactionCommit trans_id req_master in--         return request--    , opSuccess = inspectCommit env mvar-    , opFailure = failed mvar-    }-  where-    settings   = _transSettings env-    req_master = s_requireMaster settings-----------------------------------------------------------------------------------inspectWrite :: TransactionEnv-             -> MVar (OperationExceptional ())-             -> TransactionWriteCompleted-             -> IO Decision-inspectWrite env mvar twc = go (getField $ transactionWCResult twc)-  where-    go OP_SUCCESS                = succeedWrite mvar twc-    go OP_PREPARE_TIMEOUT        = return Retry-    go OP_FORWARD_TIMEOUT        = return Retry-    go OP_COMMIT_TIMEOUT         = return Retry-    go OP_WRONG_EXPECTED_VERSION = failed mvar wrong_version-    go OP_STREAM_DELETED         = failed mvar (StreamDeleted stream_id)-    go OP_INVALID_TRANSACTION    = failed mvar InvalidTransaction-    go OP_ACCESS_DENIED          = failed mvar (AccessDenied stream_id)--    exp_ver       = _transExpectedVersion env-    stream_id     = _transStreamId env-    wrong_version = WrongExpectedVersion stream_id exp_ver-----------------------------------------------------------------------------------succeedWrite :: MVar (OperationExceptional ())-             -> TransactionWriteCompleted-             -> IO Decision-succeedWrite mvar _ = do-    putMVar mvar (Right ())-    return EndOperation-----------------------------------------------------------------------------------inspectCommit :: TransactionEnv-              -> MVar (OperationExceptional WriteResult)-              -> TransactionCommitCompleted-              -> IO Decision-inspectCommit env mvar tcc = go (getField $ transactionCCResult tcc)-  where-    go OP_SUCCESS                = succeedCommit mvar tcc-    go OP_PREPARE_TIMEOUT        = return Retry-    go OP_FORWARD_TIMEOUT        = return Retry-    go OP_COMMIT_TIMEOUT         = return Retry-    go OP_WRONG_EXPECTED_VERSION = failed mvar wrong_version-    go OP_STREAM_DELETED         = failed mvar (StreamDeleted stream_id)-    go OP_INVALID_TRANSACTION    = failed mvar InvalidTransaction-    go OP_ACCESS_DENIED          = failed mvar (AccessDenied stream_id)--    exp_ver       = _transExpectedVersion env-    stream_id     = _transStreamId env-    wrong_version = WrongExpectedVersion stream_id exp_ver-----------------------------------------------------------------------------------succeedCommit :: MVar (OperationExceptional WriteResult)-              -> TransactionCommitCompleted-              -> IO Decision-succeedCommit mvar tcc = do-    putMVar mvar (Right wr)-    return EndOperation-  where-    last_evt_num = getField $ transactionCCLastNumber tcc-    com_pos      = getField $ transactionCCCommitPosition tcc-    pre_pos      = getField $ transactionCCPreparePosition tcc-    com_pos_int  = fromMaybe (-1) com_pos-    pre_pos_int  = fromMaybe (-1) pre_pos-    pos          = Position com_pos_int pre_pos_int-    wr           = WriteResult last_evt_num pos-----------------------------------------------------------------------------------eventToNewEvent :: Event -> IO NewEvent-eventToNewEvent evt =-    newEvent evt_type-             evt_id-             evt_data_type-             evt_metadata_type-             evt_data_bytes-             evt_metadata_bytes-  where-    evt_type           = eventType evt-    evt_id             = eventId evt-    evt_data_bytes     = eventDataBytes $ eventData evt-    evt_data_type      = eventDataType $ eventData evt-    evt_metadata_bytes = eventMetadataBytes $ eventData evt-    evt_metadata_type  = eventMetadataType $ eventData evt-----------------------------------------------------------------------------------createAsync :: IO (Async a, MVar (OperationExceptional a))-createAsync = do-    mvar <- newEmptyMVar-    as   <- async $ do-        res <- readMVar mvar-        either throwIO return res--    return (as, mvar)