eventstore 0.1.0.0 → 0.1.0.1
raw patch · 16 files changed
+1086/−1123 lines, 16 filesdep +sodiumdep −binarydep −suspenddep −timers
Dependencies added: sodium
Dependencies removed: binary, suspend, timers
Files
- Database/EventStore.hs +48/−67
- Database/EventStore/Internal/Manager/Operation.hs +187/−0
- Database/EventStore/Internal/Operation/Common.hs +0/−80
- Database/EventStore/Internal/Operation/DeleteStreamOperation.hs +62/−19
- Database/EventStore/Internal/Operation/ReadAllEventsOperation.hs +121/−26
- Database/EventStore/Internal/Operation/ReadEventOperation.hs +99/−19
- Database/EventStore/Internal/Operation/ReadStreamEventsOperation.hs +121/−26
- Database/EventStore/Internal/Operation/TransactionStartOperation.hs +75/−85
- Database/EventStore/Internal/Operation/WriteEventsOperation.hs +62/−18
- Database/EventStore/Internal/Packages.hs +4/−37
- Database/EventStore/Internal/Processor.hs +220/−269
- Database/EventStore/Internal/Reader.hs +12/−27
- Database/EventStore/Internal/Types.hs +3/−445
- Database/EventStore/Internal/Util/Sodium.hs +35/−0
- Database/EventStore/Internal/Writer.hs +32/−0
- eventstore.cabal +5/−5
Database/EventStore.hs view
@@ -1,3 +1,4 @@+{-# LANGUAGE RecordWildCards #-} -------------------------------------------------------------------------------- -- | -- Module : Database.EventStore@@ -12,7 +13,7 @@ module Database.EventStore ( Event , EventData- , EventStoreConnection+ , Connection , ExpectedVersion(..) , HostName , Port@@ -53,7 +54,6 @@ ) where ---------------------------------------------------------------------------------import Control.Concurrent import Control.Concurrent.STM import Data.Int @@ -76,50 +76,40 @@ type Port = Int ----------------------------------------------------------------------------------- EventStoreConnection+-- Connection ---------------------------------------------------------------------------------data EventStoreConnection- = EventStoreConnection- { mgrChan :: TChan Msg- , mgrSettings :: Settings- , mgrThreadId :: ThreadId+data Connection+ = Connection+ { conProcessor :: Processor+ , conSettings :: Settings } ---------------------------------------------------------------------------------msgQueue :: EventStoreConnection -> Msg -> IO ()-msgQueue mgr msg = atomically $ writeTChan chan msg- where- chan = mgrChan mgr---------------------------------------------------------------------------------- -- | Creates a new connection to a single node. It maintains a full duplex--- connection to the EventStore. An @EventStoreConnection@ operates quite+-- connection to the EventStore. An EventStore @Connection@ operates quite -- differently than say a SQL connection. Normally when you use a SQL -- 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 @EventStoreConnection@ all operation+-- Another difference is that with the EventStore @Connection@ all operation -- 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 -- this way-connect :: Settings -> HostName -> Port -> IO EventStoreConnection+connect :: Settings -> HostName -> Port -> IO Connection connect settings host port = do- chan <- newTChanIO- app <- newProcessor settings chan host port- tid <- forkFinally (appProcess app) (\_ -> appFinalizer app)+ processor <- newProcessor 3+ processorConnect processor host port - return $ EventStoreConnection chan settings tid+ return $ Connection processor settings ---------------------------------------------------------------------------------shutdown :: EventStoreConnection -> IO ()-shutdown mgr = killThread tid- where- tid = mgrThreadId mgr+shutdown :: Connection -> IO ()+shutdown Connection{..} = processorShutdown conProcessor ---------------------------------------------------------------------------------sendEvent :: EventStoreConnection+sendEvent :: Connection -> Text -- ^ Stream -> ExpectedVersion -> Event@@ -128,71 +118,66 @@ sendEvents mgr evt_stream exp_ver [evt] ---------------------------------------------------------------------------------sendEvents :: EventStoreConnection+sendEvents :: Connection -> Text -- ^ Stream -> ExpectedVersion -> [Event] -> IO (Async WriteResult)-sendEvents mgr evt_stream exp_ver evts = do+sendEvents Connection{..} evt_stream exp_ver evts = do (as, mvar) <- createAsync - let op = writeEventsOperation settings mvar evt_stream exp_ver evts+ let op = writeEventsOperation conSettings mvar evt_stream exp_ver evts - msgQueue mgr (RegisterOperation op)+ processorNewOperation conProcessor op return as- where- settings = mgrSettings mgr ---------------------------------------------------------------------------------deleteStream :: EventStoreConnection+deleteStream :: Connection -> Text -> ExpectedVersion -> Maybe Bool -- ^ Hard delete -> IO (Async DeleteResult)-deleteStream mgr evt_stream exp_ver hard_del = do+deleteStream Connection{..} evt_stream exp_ver hard_del = do (as, mvar) <- createAsync - let op = deleteStreamOperation settings mvar evt_stream exp_ver hard_del+ let op = deleteStreamOperation conSettings mvar evt_stream exp_ver hard_del - msgQueue mgr (RegisterOperation op)+ processorNewOperation conProcessor op return as- where- settings = mgrSettings mgr ---------------------------------------------------------------------------------transactionStart :: EventStoreConnection+transactionStart :: Connection -> Text -> ExpectedVersion -> IO (Async Transaction)-transactionStart mgr evt_stream exp_ver = do+transactionStart Connection{..} evt_stream exp_ver = do (as, mvar) <- createAsync - let op = transactionStartOperation settings chan mvar evt_stream exp_ver+ let op = transactionStartOperation conSettings+ conProcessor+ mvar+ evt_stream+ exp_ver - msgQueue mgr (RegisterOperation op)+ processorNewOperation conProcessor op return as- where- chan = mgrChan mgr- settings = mgrSettings mgr ---------------------------------------------------------------------------------readEvent :: EventStoreConnection+readEvent :: Connection -> Text -> Int32 -> Bool -> IO (Async ReadResult)-readEvent mgr stream_id evt_num res_link_tos = do+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+ let op = readEventOperation conSettings mvar stream_id evt_num res_link_tos - msgQueue mgr (RegisterOperation op)+ processorNewOperation conProcessor op return as- where- settings = mgrSettings mgr ---------------------------------------------------------------------------------readStreamEventsForward :: EventStoreConnection+readStreamEventsForward :: Connection -> Text -> Int32 -> Int32@@ -202,7 +187,7 @@ readStreamEventsCommon mgr Forward ---------------------------------------------------------------------------------readStreamEventsBackward :: EventStoreConnection+readStreamEventsBackward :: Connection -> Text -> Int32 -> Int32@@ -212,17 +197,17 @@ readStreamEventsCommon mgr Backward ---------------------------------------------------------------------------------readStreamEventsCommon :: EventStoreConnection+readStreamEventsCommon :: Connection -> ReadDirection -> Text -> Int32 -> Int32 -> Bool -> IO (Async StreamEventsSlice)-readStreamEventsCommon mgr dir stream_id start cnt res_link_tos = do+readStreamEventsCommon Connection{..} dir stream_id start cnt res_link_tos = do (as, mvar) <- createAsync - let op = readStreamEventsOperation settings+ let op = readStreamEventsOperation conSettings dir mvar stream_id@@ -230,13 +215,11 @@ cnt res_link_tos - msgQueue mgr (RegisterOperation op)+ processorNewOperation conProcessor op return as- where- settings = mgrSettings mgr ---------------------------------------------------------------------------------readAllEventsForward :: EventStoreConnection+readAllEventsForward :: Connection -> Int64 -> Int64 -> Int32@@ -246,7 +229,7 @@ readAllEventsCommon mgr Forward ---------------------------------------------------------------------------------readAllEventsBackward :: EventStoreConnection+readAllEventsBackward :: Connection -> Int64 -> Int64 -> Int32@@ -256,17 +239,17 @@ readAllEventsCommon mgr Backward ---------------------------------------------------------------------------------readAllEventsCommon :: EventStoreConnection+readAllEventsCommon :: Connection -> ReadDirection -> Int64 -> Int64 -> Int32 -> Bool -> IO (Async AllEventsSlice)-readAllEventsCommon mgr dir c_pos p_pos max_c res_link_tos = do+readAllEventsCommon Connection{..} dir c_pos p_pos max_c res_link_tos = do (as, mvar) <- createAsync - let op = readAllEventsOperation settings+ let op = readAllEventsOperation conSettings dir mvar c_pos@@ -274,10 +257,8 @@ max_c res_link_tos - msgQueue mgr (RegisterOperation op)+ processorNewOperation conProcessor op return as- where- settings = mgrSettings mgr -------------------------------------------------------------------------------- createAsync :: IO (Async a, TMVar (OperationExceptional a))
+ Database/EventStore/Internal/Manager/Operation.hs view
@@ -0,0 +1,187 @@+{-# 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+ ( 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 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 :: OperationParams -> Operation+createOperation params =+ Operation+ { operationCreatePackage = createPackage params+ , operationInspect = inspection params+ }++--------------------------------------------------------------------------------+createPackage :: OperationParams -> UUID -> IO Package+createPackage OperationParams{..} uuid = do+ req <- opRequest++ let pack = Package+ { packageCmd = opRequestCmd+ , packageCorrelation = uuid+ , packageFlag = None+ , packageData = runPut $ encodeMessage req+ }++ 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 :: (Package -> Reactive ())+ -> Reactive ()+ -> Event Package+ -> Reactive (OperationParams -> Reactive ())+operationNetwork 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 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
− Database/EventStore/Internal/Operation/Common.hs
@@ -1,80 +0,0 @@------------------------------------------------------------------------------------ |--- Module : Database.EventStore.Internal.Operation.Common--- 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.Common- ( OperationParams(..)- , createOperation- -- * Re-exports- , module Data.ProtocolBuffers- ) where-----------------------------------------------------------------------------------import Data.ProtocolBuffers-import Data.Serialize-import Data.UUID-----------------------------------------------------------------------------------import Database.EventStore.Internal.Types-----------------------------------------------------------------------------------data OperationParams request response- = OperationParams- { opSettings :: !Settings- , opRequestCmd :: !Command- , opResponseCmd :: !Command-- , opRequest :: IO request- , opSuccess :: response -> IO Decision- , opFailure :: OperationException -> IO Decision- }-----------------------------------------------------------------------------------createOperation :: (Encode a, Decode b) => OperationParams a b -> Operation-createOperation params =- Operation- { operationCreatePackage = createPackage params- , operationInspect = inspection params- }-----------------------------------------------------------------------------------createPackage :: Encode a => OperationParams a b -> UUID -> IO Package-createPackage params uuid = do- req <- opRequest params-- let pack = Package- { packageCmd = opRequestCmd params- , packageCorrelation = uuid- , packageFlag = None- , packageData = runPut $ encodeMessage req- }-- return pack-----------------------------------------------------------------------------------inspection :: Decode b => OperationParams a b -> Package -> IO Decision-inspection params pack- | found == exp_v = deeperInspection params pack- | otherwise = failed (InvalidServerResponse exp_v found)- where- exp_v = opResponseCmd params- failed = opFailure params- found = packageCmd pack-----------------------------------------------------------------------------------deeperInspection :: Decode b => OperationParams a b -> Package -> IO Decision-deeperInspection params pack =- case runGet decodeMessage bytes of- Left e -> failed (ProtobufDecodingError e)- Right msg -> succeed msg- where- failed = opFailure params- succeed = opSuccess params- bytes = packageData pack
Database/EventStore/Internal/Operation/DeleteStreamOperation.hs view
@@ -1,3 +1,5 @@+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DataKinds #-} -------------------------------------------------------------------------------- -- | -- Module : Database.EventStore.Internal.Operation.DeleteStreamOperation@@ -14,42 +16,83 @@ -------------------------------------------------------------------------------- import Control.Concurrent.STM+import Data.Int import Data.Maybe+import GHC.Generics (Generic) --------------------------------------------------------------------------------+import Data.ProtocolBuffers import Data.Text ---------------------------------------------------------------------------------import Database.EventStore.Internal.Operation.Common+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 -> TMVar (OperationExceptional DeleteResult) -> Text -> ExpectedVersion -> Maybe Bool- -> Operation+ -> OperationParams deleteStreamOperation settings mvar stream_id exp_ver hard_del =- createOperation params- where- params = OperationParams- { opSettings = settings- , opRequestCmd = DeleteStreamCmd- , opResponseCmd = DeleteStreamCompletedCmd+ OperationParams+ { opSettings = settings+ , opRequestCmd = 0x8A+ , opResponseCmd = 0x8B - , opRequest =- let req_master = _requireMaster settings- exp_ver_int = expVersionInt32 exp_ver- request = newDeleteStream stream_id- exp_ver_int- req_master- hard_del in- return request+ , opRequest =+ let req_master = _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- }+ , opSuccess = inspect mvar stream_id exp_ver+ , opFailure = failed mvar+ } -------------------------------------------------------------------------------- inspect :: TMVar (OperationExceptional DeleteResult)
Database/EventStore/Internal/Operation/ReadAllEventsOperation.hs view
@@ -1,3 +1,5 @@+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DataKinds #-} {-# LANGUAGE OverloadedStrings #-} -------------------------------------------------------------------------------- -- |@@ -11,18 +13,113 @@ -- -------------------------------------------------------------------------------- module Database.EventStore.Internal.Operation.ReadAllEventsOperation- ( readAllEventsOperation ) where+ ( AllEventsSlice(..)+ , readAllEventsOperation+ ) where -------------------------------------------------------------------------------- import Control.Concurrent.STM import Data.Int import Data.Maybe+import GHC.Generics (Generic) ---------------------------------------------------------------------------------import Database.EventStore.Internal.Operation.Common+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+ }++--------------------------------------------------------------------------------+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++--------------------------------------------------------------------------------+data AllEventsSlice+ = AllEventsSlice+ { allEventsSliceResult :: !ReadAllResult+ , allEventsSliceFrom :: !Position+ , allEventsSliceNext :: !Position+ , allEventsSliceIsEOS :: !Bool+ , allEventsSliceEvents :: ![ResolvedEvent]+ , allEventsSliceDirection :: !ReadDirection+ }+ 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 -> TMVar (OperationExceptional AllEventsSlice)@@ -30,35 +127,33 @@ -> Int64 -> Int32 -> Bool- -> Operation+ -> OperationParams readAllEventsOperation settings dir mvar c_pos p_pos max_c res_link_tos =- createOperation params+ OperationParams+ { opSettings = settings+ , opRequestCmd = req+ , opResponseCmd = resp++ , opRequest =+ let req_master = _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 -> ReadAllEventsForwardCmd- Backward -> ReadAllEventsBackwardCmd+ Forward -> 0xB6+ Backward -> 0xB8 resp = case dir of- Forward -> ReadAllEventsForwardCompletedCmd- Backward -> ReadAllEventsBackwardCompletedCmd-- params = OperationParams- { opSettings = settings- , opRequestCmd = req- , opResponseCmd = resp-- , opRequest =- let req_master = _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- }+ Forward -> 0xB7+ Backward -> 0xB9 -------------------------------------------------------------------------------- inspect :: TMVar (OperationExceptional AllEventsSlice)
Database/EventStore/Internal/Operation/ReadEventOperation.hs view
@@ -1,3 +1,6 @@+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DataKinds #-}+{-# OPTIONS_GHC -fcontext-stack=26 #-} -------------------------------------------------------------------------------- -- | -- Module : Database.EventStore.Internal.Operation.ReadEventOperation@@ -10,45 +13,122 @@ -- -------------------------------------------------------------------------------- module Database.EventStore.Internal.Operation.ReadEventOperation- ( readEventOperation ) where+ ( ReadResult(..)+ , readEventOperation+ ) where -------------------------------------------------------------------------------- import Control.Concurrent.STM import Data.Int+import GHC.Generics (Generic) --------------------------------------------------------------------------------+import Data.ProtocolBuffers import Data.Text ---------------------------------------------------------------------------------import Database.EventStore.Internal.Operation.Common+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+ }++--------------------------------------------------------------------------------+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++--------------------------------------------------------------------------------+data ReadResult+ = ReadResult+ { readResultStatus :: !ReadEventResult+ , readResultStreamId :: !Text+ , readResultEventNumber :: !Int32+ , 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 -> TMVar (OperationExceptional ReadResult) -> Text -> Int32 -> Bool -- ^ Resolve link TOS- -> Operation+ -> OperationParams readEventOperation settings mvar stream_id evt_num res_link_tos =- createOperation params- where- params = OperationParams- { opSettings = settings- , opRequestCmd = ReadEventCmd- , opResponseCmd = ReadEventCompletedCmd+ OperationParams+ { opSettings = settings+ , opRequestCmd = 0xB0+ , opResponseCmd = 0xB1 - , opRequest =- let req_master = _requireMaster settings- request = newReadEvent stream_id- evt_num- res_link_tos- req_master in- return request+ , opRequest =+ let req_master = _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- }+ , opSuccess = inspect mvar stream_id evt_num+ , opFailure = failed mvar+ } -------------------------------------------------------------------------------- inspect :: TMVar (OperationExceptional ReadResult)
Database/EventStore/Internal/Operation/ReadStreamEventsOperation.hs view
@@ -1,3 +1,5 @@+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DataKinds #-} -------------------------------------------------------------------------------- -- | -- Module : Database.EventStore.Internal.Operation.ReadStreamEventsOperation@@ -10,20 +12,115 @@ -- -------------------------------------------------------------------------------- module Database.EventStore.Internal.Operation.ReadStreamEventsOperation- ( readStreamEventsOperation ) where+ ( StreamEventsSlice(..)+ , readStreamEventsOperation+ ) where -------------------------------------------------------------------------------- import Control.Concurrent.STM import Data.Int+import GHC.Generics (Generic) --------------------------------------------------------------------------------+import Data.ProtocolBuffers import Data.Text ---------------------------------------------------------------------------------import Database.EventStore.Internal.Operation.Common+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++--------------------------------------------------------------------------------+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++--------------------------------------------------------------------------------+data StreamEventsSlice+ = StreamEventsSlice+ { streamEventsSliceResult :: !ReadStreamResult+ , streamEventsSliceStreamId :: !Text+ , streamEventsSliceStart :: !Int32+ , streamEventsSliceNext :: !Int32+ , streamEventsSliceLast :: !Int32+ , streamEventsSliceIsEOS :: !Bool+ , streamEventsSliceEvents :: ![ResolvedEvent]+ , streamEventsSliceDirection :: !ReadDirection+ }+ 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 -> TMVar (OperationExceptional StreamEventsSlice)@@ -31,35 +128,33 @@ -> Int32 -> Int32 -> Bool- -> Operation+ -> OperationParams readStreamEventsOperation settings dir mvar stream_id start cnt res_link_tos =- createOperation params+ OperationParams+ { opSettings = settings+ , opRequestCmd = req+ , opResponseCmd = resp++ , opRequest =+ let req_master = _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 -> ReadStreamEventsForwardCmd- Backward -> ReadStreamEventsBackwardCmd+ Forward -> 0xB2+ Backward -> 0xB4 resp = case dir of- Forward -> ReadStreamEventsForwardCompletedCmd- Backward -> ReadStreamEventsBackwardCompletedCmd-- params = OperationParams- { opSettings = settings- , opRequestCmd = req- , opResponseCmd = resp-- , opRequest =- let req_master = _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- }+ Forward -> 0xB3+ Backward -> 0xB5 -------------------------------------------------------------------------------- inspect :: TMVar (OperationExceptional StreamEventsSlice)
Database/EventStore/Internal/Operation/TransactionStartOperation.hs view
@@ -1,3 +1,4 @@+{-# LANGUAGE RecordWildCards #-} -------------------------------------------------------------------------------- -- | -- Module : Database.EventStore.Internal.Operation.TransactionStartOperation@@ -14,61 +15,61 @@ -------------------------------------------------------------------------------- import Control.Concurrent.STM+import Data.Int import Data.Maybe import Data.Traversable ---------------------------------------------------------------------------------import Data.Int+import Control.Concurrent.Async+import Data.ProtocolBuffers import Data.Text ---------------------------------------------------------------------------------import Control.Concurrent.Async-import Database.EventStore.Internal.Operation.Common+import Database.EventStore.Internal.Manager.Operation+import Database.EventStore.Internal.Processor import Database.EventStore.Internal.Types -------------------------------------------------------------------------------- data TransactionEnv = TransactionEnv { _transSettings :: Settings- , _transChan :: TChan Msg+ , _transProcessor :: Processor , _transStreamId :: Text , _transExpectedVersion :: ExpectedVersion } -------------------------------------------------------------------------------- transactionStartOperation :: Settings- -> TChan Msg+ -> Processor -> TMVar (OperationExceptional Transaction) -> Text -> ExpectedVersion- -> Operation-transactionStartOperation settings chan mvar stream_id exp_ver =- createOperation params+ -> OperationParams+transactionStartOperation settings procss mvar stream_id exp_ver =+ OperationParams+ { opSettings = settings+ , opRequestCmd = 0x84+ , opResponseCmd = 0x85++ , opRequest =+ let req_master = _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- , _transChan = chan+ , _transProcessor = procss , _transStreamId = stream_id , _transExpectedVersion = exp_ver } - params = OperationParams- { opSettings = settings- , opRequestCmd = TransactionStartCmd- , opResponseCmd = TransactionStartCompletedCmd-- , opRequest =- let req_master = _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- }- -------------------------------------------------------------------------------- inspectTrans :: TransactionEnv -> TMVar (OperationExceptional Transaction)@@ -111,89 +112,82 @@ -------------------------------------------------------------------------------- createTransaction :: TransactionEnv -> Int64 -> Transaction-createTransaction env trans_id = trans+createTransaction env@TransactionEnv{..} trans_id = trans where- stream_id = _transStreamId env- chan = _transChan env- exp_ver = _transExpectedVersion env- trans = Transaction- { transactionId = trans_id- , transactionStreamId = stream_id- , transactionExpectedVersion = exp_ver+ trans = Transaction+ { transactionId = trans_id+ , transactionStreamId = _transStreamId+ , transactionExpectedVersion = _transExpectedVersion - , transactionCommit = do- (as, mvar) <- createAsync+ , transactionCommit = do+ (as, mvar) <- createAsync - let op = transactionCommitOperation env trans_id mvar+ let op = transactionCommitOperation env trans_id mvar - sendMsg chan (RegisterOperation op)- return as+ processorNewOperation _transProcessor op+ return as - , transactionSendEvents = \evts -> do- (as, mvar) <- createAsync+ , transactionSendEvents = \evts -> do+ (as, mvar) <- createAsync - let op = transactionWriteOperation env trans_id mvar evts+ let op = transactionWriteOperation env trans_id mvar evts - sendMsg chan (RegisterOperation op)- return as+ processorNewOperation _transProcessor op+ return as - , transactionRollback = return ()- }+ , transactionRollback = return ()+ } -------------------------------------------------------------------------------- transactionWriteOperation :: TransactionEnv -> Int64 -> TMVar (OperationExceptional ()) -> [Event]- -> Operation+ -> OperationParams transactionWriteOperation env trans_id mvar evts =- createOperation params- where- settings = _transSettings env- req_master = _requireMaster settings-- params = OperationParams- { opSettings = settings- , opRequestCmd = TransactionWriteCmd- , opResponseCmd = TransactionWriteCompletedCmd+ OperationParams+ { opSettings = settings+ , opRequestCmd = 0x86+ , opResponseCmd = 0x87 - , opRequest = do- new_evts <- traverse eventToNewEvent evts+ , opRequest = do+ new_evts <- traverse eventToNewEvent evts - let request = newTransactionWrite trans_id- new_evts- req_master+ let request = newTransactionWrite trans_id+ new_evts+ req_master - return request+ return request - , opSuccess = inspectWrite env mvar- , opFailure = failed mvar- }+ , opSuccess = inspectWrite env mvar+ , opFailure = failed mvar+ }+ where+ settings = _transSettings env+ req_master = _requireMaster settings -------------------------------------------------------------------------------- transactionCommitOperation :: TransactionEnv -> Int64 -> TMVar (OperationExceptional WriteResult)- -> Operation+ -> OperationParams transactionCommitOperation env trans_id mvar =- createOperation params- where- settings = _transSettings env- req_master = _requireMaster settings-- params = OperationParams- { opSettings = settings- , opRequestCmd = TransactionCommitCmd- , opResponseCmd = TransactionCommitCompletedCmd+ OperationParams+ { opSettings = settings+ , opRequestCmd = 0x88+ , opResponseCmd = 0x89 - , opRequest =- let request = newTransactionCommit trans_id req_master in+ , opRequest =+ let request = newTransactionCommit trans_id req_master in - return request+ return request - , opSuccess = inspectCommit env mvar- , opFailure = failed mvar- }+ , opSuccess = inspectCommit env mvar+ , opFailure = failed mvar+ }+ where+ settings = _transSettings env+ req_master = _requireMaster settings -------------------------------------------------------------------------------- inspectWrite :: TransactionEnv@@ -273,10 +267,6 @@ evt_data_type = eventDataType $ eventData evt evt_metadata_bytes = eventMetadataBytes $ eventData evt evt_metadata_type = eventMetadataType $ eventData evt-----------------------------------------------------------------------------------sendMsg :: TChan Msg -> Msg -> IO ()-sendMsg chan msg = atomically $ writeTChan chan msg -------------------------------------------------------------------------------- createAsync :: IO (Async a, TMVar (OperationExceptional a))
Database/EventStore/Internal/Operation/WriteEventsOperation.hs view
@@ -1,3 +1,5 @@+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DataKinds #-} -------------------------------------------------------------------------------- -- | -- Module : Database.EventStore.Internal.Operation.WriteEventsOperation@@ -14,46 +16,88 @@ -------------------------------------------------------------------------------- import Control.Concurrent.STM+import Data.Int import Data.Maybe import Data.Traversable+import GHC.Generics (Generic) --------------------------------------------------------------------------------+import Data.ProtocolBuffers import Data.Text ---------------------------------------------------------------------------------import Database.EventStore.Internal.Operation.Common+import Database.EventStore.Internal.Manager.Operation import Database.EventStore.Internal.Types --------------------------------------------------------------------------------+data WriteEvents+ = WriteEvents+ { writeStreamId :: Required 1 (Value Text)+ , writeExpectedVersion :: Required 2 (Value Int32)+ , writeEvents :: Repeated 3 (Message NewEvent)+ , writeRequireMaster :: Required 4 (Value Bool)+ }+ deriving (Generic, Show)++--------------------------------------------------------------------------------+instance Encode WriteEvents++--------------------------------------------------------------------------------+newWriteEvents :: Text -- ^ Stream+ -> Int32 -- ^ Expected version+ -> [NewEvent] -- ^ Events+ -> Bool -- ^ Require master+ -> WriteEvents+newWriteEvents stream_id exp_ver evts req_master =+ WriteEvents+ { writeStreamId = putField stream_id+ , writeExpectedVersion = putField exp_ver+ , writeEvents = putField evts+ , writeRequireMaster = putField req_master+ }++--------------------------------------------------------------------------------+data WriteEventsCompleted+ = WriteEventsCompleted+ { writeCompletedResult :: Required 1 (Enumeration OpResult)+ , writeCompletedMessage :: Optional 2 (Value Text)+ , writeCompletedFirstNumber :: Required 3 (Value Int32)+ , writeCompletedLastNumber :: Required 4 (Value Int32)+ , writeCompletedPreparePosition :: Optional 5 (Value Int64)+ , writeCompletedCommitPosition :: Optional 6 (Value Int64)+ }+ deriving (Generic, Show)++--------------------------------------------------------------------------------+instance Decode WriteEventsCompleted++-------------------------------------------------------------------------------- writeEventsOperation :: Settings -> TMVar (OperationExceptional WriteResult) -> Text -> ExpectedVersion -> [Event]- -> Operation+ -> OperationParams writeEventsOperation settings mvar evt_stream exp_ver evts =- createOperation params- where- params =- OperationParams- { opSettings = settings- , opRequestCmd = WriteEventsCmd- , opResponseCmd = WriteEventsCompletedCmd+ OperationParams+ { opSettings = settings+ , opRequestCmd = 0x82+ , opResponseCmd = 0x83 - , opRequest = do- new_evts <- traverse eventToNewEvent evts+ , opRequest = do+ new_evts <- traverse eventToNewEvent evts - let require_master = _requireMaster settings- exp_ver_int32 = expVersionInt32 exp_ver- request = newWriteEvents evt_stream+ let require_master = _requireMaster settings+ exp_ver_int32 = expVersionInt32 exp_ver+ request = newWriteEvents evt_stream exp_ver_int32 new_evts require_master- return request+ return request - , opSuccess = inspect mvar evt_stream exp_ver- , opFailure = failed mvar- }+ , opSuccess = inspect mvar evt_stream exp_ver+ , opFailure = failed mvar+ } -------------------------------------------------------------------------------- inspect :: TMVar (OperationExceptional WriteResult)
Database/EventStore/Internal/Packages.hs view
@@ -11,22 +11,16 @@ -------------------------------------------------------------------------------- module Database.EventStore.Internal.Packages ( -- * Package Smart Contructors- deleteStreamPackage- , heartbeatPackage+ heartbeatPackage , heartbeatResponsePackage- , writeEventsPackage -- * Cereal Put , putPackage- -- * Decode- , getWriteEventsCompleted ) where -------------------------------------------------------------------------------- import qualified Data.ByteString as B ---------------------------------------------------------------------------------import Data.ProtocolBuffers-import Data.Serialize.Get import Data.Serialize.Put import Data.UUID import System.Random@@ -37,31 +31,11 @@ -------------------------------------------------------------------------------- -- Encode ---------------------------------------------------------------------------------writeEventsPackage :: UUID -> Flag -> WriteEvents -> Package-writeEventsPackage uuid flag msg =- Package- { packageCmd = WriteEventsCmd- , packageFlag = flag- , packageCorrelation = uuid- , packageData = runPut $ encodeMessage msg- }-----------------------------------------------------------------------------------deleteStreamPackage :: UUID -> Flag -> DeleteStream -> Package-deleteStreamPackage uuid flag msg =- Package- { packageCmd = DeleteStreamCmd- , packageFlag = flag- , packageCorrelation = uuid- , packageData = runPut $ encodeMessage msg- }---------------------------------------------------------------------------------- heartbeatPackage :: IO Package heartbeatPackage = do uuid <- randomIO let pack = Package- { packageCmd = HeartbeatRequest+ { packageCmd = 0x01 , packageFlag = None , packageCorrelation = uuid , packageData = B.empty@@ -73,7 +47,7 @@ heartbeatResponsePackage :: UUID -> Package heartbeatResponsePackage uuid = Package- { packageCmd = HeartbeatResponse+ { packageCmd = 0x02 , packageFlag = None , packageCorrelation = uuid , packageData = B.empty@@ -83,22 +57,15 @@ putPackage :: Package -> Put putPackage pack = putWord32le length_prefix >>- putWord8 cmd_word8 >>+ putWord8 (packageCmd pack) >> putWord8 flag_word8 >> putLazyByteString corr_bytes >> putByteString pack_data where pack_data = packageData pack length_prefix = fromIntegral (B.length pack_data + mandatorySize)- cmd_word8 = cmdWord8 $ packageCmd pack flag_word8 = flagWord8 $ packageFlag pack corr_bytes = toByteString $ packageCorrelation pack------------------------------------------------------------------------------------- Decode----------------------------------------------------------------------------------getWriteEventsCompleted :: Get WriteEventsCompleted-getWriteEventsCompleted = decodeMessage -------------------------------------------------------------------------------- mandatorySize :: Int
Database/EventStore/Internal/Processor.hs view
@@ -1,3 +1,6 @@+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE ScopedTypeVariables #-} -------------------------------------------------------------------------------- -- | -- Module : Database.EventStore.Internal.Processor@@ -10,342 +13,290 @@ -- -------------------------------------------------------------------------------- module Database.EventStore.Internal.Processor- ( Application(..)+ ( InternalException(..)+ , Processor(..) , newProcessor ) where ---------------------------------------------------------------------------------import Control.Concurrent-import Control.Concurrent.STM-import Control.Exception-import qualified Data.ByteString as B-import qualified Data.Map.Strict as M-import Data.Serialize.Put-import System.IO-import Text.Printf+import Control.Concurrent+import Control.Concurrent.STM+import Control.Exception+import Data.Monoid ((<>))+import Data.Typeable+import Data.Word+import System.IO+import Text.Printf ---------------------------------------------------------------------------------import Data.Time+import Control.Concurrent.Async import Data.UUID+import FRP.Sodium import Network import System.Random --------------------------------------------------------------------------------+import Database.EventStore.Internal.Manager.Operation import Database.EventStore.Internal.Packages import Database.EventStore.Internal.Reader-import Database.EventStore.Internal.Types+import Database.EventStore.Internal.Types hiding (Event, newEvent)+import Database.EventStore.Internal.Util.Sodium+import Database.EventStore.Internal.Writer ----------------------------------------------------------------------------------- Env+-- Processor ---------------------------------------------------------------------------------data Env- = Env- { _hostname :: HostName- , _port :: Int- , _settings :: Settings- , _chan :: TChan Msg- , _finalizer :: TVar (IO ())+data Processor+ = Processor+ { processorConnect :: HostName -> Int -> IO ()+ , processorShutdown :: IO ()+ , processorNewOperation :: OperationParams -> IO () } ---------------------------------------------------------------------------------getMsg :: Env -> IO Msg-getMsg env = atomically $ readTChan (_chan env)-----------------------------------------------------------------------------------sendMsg :: Env -> Msg -> IO ()-sendMsg env msg = atomically $ writeTChan (_chan env) msg---- ----------------------------------------------------------------------------------- heartbeatInterval :: Env -> NominalDiffTime--- heartbeatInterval env = _heartbeatInterval $ _settings env---- ----------------------------------------------------------------------------------- heartbeatTimeout :: Env -> NominalDiffTime--- heartbeatTimeout env = _heartbeatTimeout $ _settings env+newProcessor :: Int -> IO Processor+newProcessor max_at = sync $ network max_at ---------------------------------------------------------------------------------newEnv :: Settings -> TChan Msg -> HostName -> Int -> IO Env-newEnv settings chan host port = do- ref <- newTVarIO (return ())-- return $ Env host port settings chan ref-+-- Exception ---------------------------------------------------------------------------------registerFinalizer :: Env -> IO () -> IO ()-registerFinalizer env action = atomically $ writeTVar var action- where- var = _finalizer env+data ProcessorException+ = MaxAttempt HostName Int Int -- ^ Hostname Port MaxAttempt value+ deriving (Show, Typeable) ---------------------------------------------------------------------------------runFinalizer :: Env -> IO ()-runFinalizer env = do- action <- atomically $ do- act <- readTVar var- writeTVar var (return ())- return act- action- where- var = _finalizer env+instance Exception ProcessorException ----------------------------------------------------------------------------------- Connection+-- State ---------------------------------------------------------------------------------data Connection- = Connection- { _connId :: UUID- , _connHandle :: Handle- , _connReaderThreadId :: ThreadId+data State+ = Offline+ { _maxAttempt :: !Int }+ | Online+ { _uuidCon :: !UUID+ , _maxAttempt :: !Int+ , _packageCount :: !Int+ , _host :: !HostName+ , _port :: !Int+ , _cleanup :: !(IO ()) } ---------------------------------------------------------------------------------connectionSend :: Connection -> Put -> IO ()-connectionSend conn put = B.hPut hdl (runPut put) >> hFlush hdl- where- hdl = _connHandle conn-----------------------------------------------------------------------------------connectionClose :: Connection -> IO ()-connectionClose conn = do- killThread thread_id- hClose hdl- printf "Disconnected %s\n" conn_id_str- where- hdl = _connHandle conn- thread_id = _connReaderThreadId conn- conn_id_str = toString $ _connId conn+initState :: Int -> State+initState max_at = Offline max_at --------------------------------------------------------------------------------+-- Event ---------------------------------------------------------------------------------type Processor = Env -> State -> IO ()+data Connect = Connect HostName Int+data Connected = Connected HostName Int UUID (IO ())+data Cleanup = Cleanup+data Reconnect = Reconnect+data Reconnected = Reconnected UUID (IO ()) ----------------------------------------------------------------------------------data Application- = Application- { appProcess :: IO ()- , appFinalizer :: IO ()- } ----------------------------------------------------------------------------------- Manager state----------------------------------------------------------------------------------data HeartbeatInfo- = HeartbeatInfo- { _lastPackage :: !Int -- ^ Last package since last update- , _intervalStage :: !Bool- , _elapsedTime :: !NominalDiffTime -- ^ Elapsed time since last update- }------------------------------------------------------------------------------------- | Holds every needed piece of information in order to properly communicate--- with an EventStore backend-data State- = State- { _lastTime :: !UTCTime- , _heartbeatInfo :: !HeartbeatInfo- , _packageNumber :: !Int -- ^ Number of received packages- , _operations :: !(M.Map UUID Operation)- }+heartbeatRequestCmd :: Word8+heartbeatRequestCmd = 0x01 ---------------------------------------------------------------------------------updateHeartbeatInfo :: State- -> UTCTime -- ^ Current time- -> Bool -- ^ Is interval stage- -> Int -- ^ Package number- -> State-updateHeartbeatInfo cur_state cur_time is_interval_state package_num = new_state- where- last_time = _lastTime cur_state- elapsed_time = diffUTCTime cur_time last_time- new_heart_info = HeartbeatInfo package_num is_interval_state elapsed_time- new_state = cur_state { _heartbeatInfo = new_heart_info }+network :: Int -> Reactive Processor+network max_at = do+ (onConnect, pushConnect) <- newEvent+ (onConnected, pushConnected) <- newEvent+ (onCleanup, pushCleanup) <- newEvent+ (onReconnect, pushReconnect) <- newEvent+ (onReconnected, pushReconnected) <- newEvent+ (onReceived, pushReceived) <- newEvent+ (onSend, pushSend) <- newEvent ----------------------------------------------------------------------------------incrPackageNumber :: State -> State-incrPackageNumber cur_state = new_state- where- new_package_number = _packageNumber cur_state + 1- new_state = cur_state { _packageNumber = new_package_number }+ push_new_op <- operationNetwork pushSend+ (pushReconnect Reconnect)+ onReceived ------------------------------------------------------------------------------------ | Create an initial @State@-newState :: IO State-newState = do- cur_time <- getCurrentTime- let package_num = 0- info = HeartbeatInfo- { _lastPackage = package_num- , _intervalStage = True- , _elapsedTime = fromIntegral (0 :: Integer)- }+ let stateE = fmap connected onConnected <>+ fmap reconnected onReconnected <>+ fmap received onReceived - state = State- { _lastTime = cur_time- , _heartbeatInfo = info- , _packageNumber = package_num- , _operations = M.empty- }+ stateB <- accum (initState max_at) stateE - return state+ let heartbeatP pkg = packageCmd pkg == heartbeatRequestCmd+ onlyHeartbeats = filterE heartbeatP onReceived ----------------------------------------------------------------------------------newProcessor :: Settings -> TChan Msg -> HostName -> Int -> IO Application-newProcessor settings chan host port = do- env <- newEnv settings chan host port- state <- newState- let app = Application- { appProcess = connecting env state- , appFinalizer = runFinalizer env- }+ con_snap = snapshot connectSnapshot onConnect stateB+ reco_snap = snapshot reconnectSnapshot onReconnect stateB+ clean_snap = snapshot cleanupSnapshot onCleanup stateB - return app+ full_reco c = do+ pushCleanup Cleanup+ pushReconnect c ----------------------------------------------------------------------------------connecting :: Processor-connecting env state = do- hdl <- connectTo host (PortNumber $ fromIntegral port)- hSetBuffering hdl NoBuffering+ push_reco_io = pushAsync full_reco Reconnect+ push_recv_io = \pkg -> sync $ pushReceived pkg+ push_recod_io = pushAsync2 $ \u c -> pushReconnected $ Reconnected u c+ push_send_io = pushAsync pushSend+ push_con_io = pushAsync4 $ \h p u c ->+ pushConnected $ Connected h p u c - rid <- forkFinally (readerThread chan hdl) recovering- conn_id <- randomIO- cur_time <- getCurrentTime- let pack_num = _packageNumber state- new_state = updateHeartbeatInfo state cur_time True pack_num- conn = Connection- { _connId = conn_id- , _connHandle = hdl- , _connReaderThreadId = rid- }+ _ <- listen con_snap $ \(ConnectionSnapshot max_a host port) ->+ connection push_recv_io+ (push_con_io host port)+ push_reco_io+ onSend+ max_a+ host+ port - printf "Connected %s\n" (toString conn_id)- registerFinalizer env $- connectionClose conn+ _ <- listen reco_snap $ \(ConnectionSnapshot max_a host port) ->+ connection push_recv_io+ push_recod_io+ push_reco_io+ onSend+ max_a+ host+ port - connected conn env new_state+ _ <- listen clean_snap $ \(CleanupSnapshot finalizer) -> finalizer - where- port = _port env- host = _hostname env- chan = _chan env+ _ <- listen onlyHeartbeats $ \pkg ->+ push_send_io $ heartbeatResponsePackage (packageCorrelation pkg) - recovering (Left some_ex)=- case fromException some_ex of- Just e ->- case e of- ConnectionClosedByServer- -> sendMsg env Reconnect- Stopped- -> return ()- _ -> sendMsg env Reconnect+ let processor =+ Processor+ { processorConnect = \h p -> sync $ pushConnect $ Connect h p+ , processorShutdown = sync $ pushCleanup Cleanup+ , processorNewOperation = \o -> sync $ push_new_op o+ } - recovering _ = return ()+ return processor ---------------------------------------------------------------------------------connected :: Connection -> Processor-connected conn env state = getMsg env >>= go- where- go Reconnect = do- runFinalizer env- putStrLn "Reconnecting..."- connecting env state-- go (RecvPackage pack) = do- new_state <- handlePackage conn env state pack- connected conn env new_state-- go (RegisterOperation op) =- registerOperation conn op env state-- go (Notice msg) = do- print msg- connected conn env state-- go Tick =- connected conn env state-+-- Observer ---------------------------------------------------------------------------------registerOperation :: Connection -> Operation -> Processor-registerOperation conn op env state = do- uuid <- randomIO- pack <- operationCreatePackage op uuid+data ConnectionSnapshot+ = ConnectionSnapshot+ { _conMax :: !Int+ , _conHost :: !HostName+ , _conPort :: !Int+ } - let new_op_map = M.insert uuid op op_map- new_state = state { _operations = new_op_map }+--------------------------------------------------------------------------------+connectSnapshot :: Connect -> State -> ConnectionSnapshot+connectSnapshot (Connect host port) s =+ ConnectionSnapshot+ { _conMax = _maxAttempt s+ , _conHost = host+ , _conPort = port+ } - connectionSend conn (putPackage pack)- connected conn env new_state- where- op_map = _operations state+--------------------------------------------------------------------------------+reconnectDelay :: Int+reconnectDelay = 500000 ---------------------------------------------------------------------------------handlePackage :: Connection -> Env -> State -> Package -> IO State-handlePackage conn env state pack = go (packageCmd pack)+connection :: (Package -> IO ())+ -> (UUID -> IO () -> IO ())+ -> IO ()+ -> Event Package+ -> Int+ -> HostName+ -> Int+ -> IO ()+connection push_pkg push_con push_reco evt_pkg max_a host port = loop 1 where- go HeartbeatRequest = do- handleHeartbeatRequest conn pack- return new_state+ loop att+ | max_a == att =+ throwIO $ MaxAttempt host port max_a+ | otherwise =+ catch (doConnect att) $ \(_ :: SomeException) -> do+ threadDelay reconnectDelay+ loop (att + 1) - go HeartbeatResponse =- return new_state+ doConnect att = do+ printf "Connecting...Attempt %d\n" att+ hdl <- connectTo host (PortNumber $ fromIntegral port)+ hSetBuffering hdl NoBuffering - go _ =- case M.lookup corr_id op_map of- Just op -> handleOperation env new_state pack op- _ -> fmap (const new_state) $ unhandledPackage pack+ uuid <- randomIO+ chan <- newTChanIO+ as_rl <- async $ sync $ listen evt_pkg (atomically . writeTChan chan)+ rid <- forkFinally (readerThread push_pkg hdl) (recovering push_reco)+ wid <- forkFinally (writerThread chan hdl) (recovering push_reco) - corr_id = packageCorrelation pack- op_map = _operations state- new_state = incrPackageNumber state+ push_con uuid $ do+ throwTo rid Stopped+ throwTo wid Stopped+ hClose hdl+ rel_w <- wait as_rl+ rel_w+ printf "Disconnected %s\n" (toString uuid) ---------------------------------------------------------------------------------handleHeartbeatRequest :: Connection -> Package -> IO ()-handleHeartbeatRequest conn pack =- connectionSend conn (putPackage pack_resp)- where- corr_id = packageCorrelation pack- pack_resp = heartbeatResponsePackage corr_id+recovering :: IO () -> Either SomeException () -> IO ()+recovering recover (Left some_ex) = do+ case fromException some_ex of+ Just e ->+ case e of+ ConnectionClosedByServer+ -> recover+ Stopped+ -> return ()+ _ -> recover+recovering _ _ = return () ---------------------------------------------------------------------------------unhandledPackage :: Package -> IO ()-unhandledPackage pack = printf "Unhandled command: %s\n" cmd_str- where- cmd_str = show $ packageCmd pack+reconnectSnapshot :: Reconnect -> State -> ConnectionSnapshot+reconnectSnapshot _ s =+ ConnectionSnapshot+ { _conMax = _maxAttempt s+ , _conHost = _host s+ , _conPort = _port s+ } ---------------------------------------------------------------------------------handleOperation :: Env -> State -> Package -> Operation -> IO State-handleOperation env state pack op = do- decision <- operationInspect op pack- case decision of- DoNothing ->- return state- EndOperation ->- return new_state- Retry- -> do sendMsg env (RegisterOperation op)- return new_state- Reconnection- -> do sendMsg env Reconnect- sendMsg env (RegisterOperation op)- return new_state- _ -> fail "Unexpected decision Processor.hs"- where- corr_id = packageCorrelation pack- op_map = _operations state- new_op_map = M.delete corr_id op_map- new_state = state { _operations = new_op_map }+newtype CleanupSnapshot = CleanupSnapshot (IO ()) --- ----------------------------------------------------------------------------------- manageHeartbeats :: ConnectionManager -> IO ()--- manageHeartbeats mgr@ConnectionManager{..} = do--- elapsed <- managerElapsedTime mgr--- info <- atomically $ readTVar mgrHeartbeatInfo--- pack_num <- atomically $ readTVar mgrPackageNumber--- let timeout = if heartbeatIntervalStage info--- then heartbeatInterval--- else heartbeatTimeout+--------------------------------------------------------------------------------+cleanupSnapshot :: Cleanup -> State -> CleanupSnapshot+cleanupSnapshot _ s =+ case s of+ Offline {}+ -> CleanupSnapshot (return ())+ Online {}+ -> CleanupSnapshot $ _cleanup s --- if pack_num /= heartbeatLastPackage info--- then managerUpdateHeartbeatInfo mgr pack_num True--- else when (heartbeatIntervalStage info) $ do--- pack <- heartbeatPackage--- doSendPackage mgr pack--- managerUpdateHeartbeatInfo mgr (heartbeatLastPackage info) False+--------------------------------------------------------------------------------+-- Model+--------------------------------------------------------------------------------+connected :: Connected -> State -> State+connected (Connected host port uuid cl) s =+ case s of+ Offline {}+ -> Online+ { _uuidCon = uuid+ , _maxAttempt = _maxAttempt s+ , _packageCount = 0+ , _host = host+ , _port = port+ , _cleanup = cl+ }+ _ -> s++--------------------------------------------------------------------------------+reconnected :: Reconnected -> State -> State+reconnected (Reconnected uuid cl) s =+ case s of+ Online {}+ -> s { _uuidCon = uuid+ , _cleanup = cl+ }+ _ -> s++--------------------------------------------------------------------------------+received :: Package -> State -> State+received _ s =+ case s of+ Online {}+ -> let cnt = _packageCount s in s { _packageCount = cnt + 1 }+ _ -> s
Database/EventStore/Internal/Reader.hs view
@@ -13,7 +13,7 @@ -------------------------------------------------------------------------------- import Prelude hiding (take)-import Control.Concurrent.STM+import Control.Monad import qualified Data.ByteString as B import System.IO import Text.Printf@@ -26,26 +26,19 @@ import Database.EventStore.Internal.Types ---------------------------------------------------------------------------------readerThread :: TChan Msg -> Handle -> IO ()-readerThread chan h = loop+readerThread :: (Package -> IO ()) -> Handle -> IO ()+readerThread push_p h = forever $ do+ header_bs <- B.hGet h 4+ case runGet getLengthPrefix header_bs of+ Left _+ -> error "Wrong package framing"+ Right length_prefix+ -> B.hGet h length_prefix >>= parsePackage where parsePackage bs = case runGet getPackage bs of- Left e -> send chan (Notice $ printf "Parsing error [%s]\n" e)- Right pack -> send chan (RecvPackage pack)-- loop = do- header_bs <- B.hGet h 4- case runGet getLengthPrefix header_bs of- Left _- -> send chan (Notice "Wrong package framing\n")- Right length_prefix- -> B.hGet h length_prefix >>= parsePackage- loop-----------------------------------------------------------------------------------send :: TChan Msg -> Msg -> IO ()-send chan msg = atomically $ writeTChan chan msg+ Left e -> error $ printf "Parsing error [%s]" e+ Right pack -> push_p pack -------------------------------------------------------------------------------- -- Parsers@@ -56,7 +49,7 @@ -------------------------------------------------------------------------------- getPackage :: Get Package getPackage = do- cmd <- getCmd+ cmd <- getWord8 flg <- getFlag col <- getUUID rest <- remaining@@ -70,14 +63,6 @@ } return pack-----------------------------------------------------------------------------------getCmd :: Get Command-getCmd = do- wd <- getWord8- case word8Cmd wd of- Just cmd -> return cmd- _ -> fail $ printf "TCP: Unhandled command value 0x%x" wd -------------------------------------------------------------------------------- getFlag :: Get Flag
Database/EventStore/Internal/Types.hs view
@@ -53,7 +53,7 @@ | StreamDeleted Text -- ^ Stream | InvalidTransaction | AccessDenied Text -- ^ Stream- | InvalidServerResponse Command Command -- ^ Expected, Found+ | InvalidServerResponse Word8 Word8 -- ^ Expected, Found | ProtobufDecodingError String | ServerError (Maybe Text) deriving (Show, Typeable)@@ -123,13 +123,6 @@ -------------------------------------------------------------------------------- -- EventStore Messages ---------------------------------------------------------------------------------data Operation- = Operation- { operationCreatePackage :: UUID -> IO Package- , operationInspect :: Package -> IO Decision- }---------------------------------------------------------------------------------- data OpResult = OP_SUCCESS | OP_PREPARE_TIMEOUT@@ -178,88 +171,6 @@ return new_evt ---------------------------------------------------------------------------------data WriteEvents- = WriteEvents- { writeStreamId :: Required 1 (Value Text)- , writeExpectedVersion :: Required 2 (Value Int32)- , writeEvents :: Repeated 3 (Message NewEvent)- , writeRequireMaster :: Required 4 (Value Bool)- }- deriving (Generic, Show)-----------------------------------------------------------------------------------instance Encode WriteEvents-----------------------------------------------------------------------------------newWriteEvents :: Text -- ^ Stream- -> Int32 -- ^ Expected version- -> [NewEvent] -- ^ Events- -> Bool -- ^ Require master- -> WriteEvents-newWriteEvents stream_id exp_ver evts req_master =- WriteEvents- { writeStreamId = putField stream_id- , writeExpectedVersion = putField exp_ver- , writeEvents = putField evts- , writeRequireMaster = putField req_master- }-----------------------------------------------------------------------------------data WriteEventsCompleted- = WriteEventsCompleted- { writeCompletedResult :: Required 1 (Enumeration OpResult)- , writeCompletedMessage :: Optional 2 (Value Text)- , writeCompletedFirstNumber :: Required 3 (Value Int32)- , writeCompletedLastNumber :: Required 4 (Value Int32)- , writeCompletedPreparePosition :: Optional 5 (Value Int64)- , writeCompletedCommitPosition :: Optional 6 (Value Int64)- }- deriving (Generic, Show)-----------------------------------------------------------------------------------instance Decode WriteEventsCompleted-----------------------------------------------------------------------------------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---------------------------------------------------------------------------------- data TransactionStart = TransactionStart { transactionStartStreamId :: Required 1 (Value Text)@@ -364,39 +275,6 @@ instance Decode TransactionCommitCompleted ---------------------------------------------------------------------------------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- }-----------------------------------------------------------------------------------data ReadEventResult- = RE_SUCCESS- | RE_NOT_FOUND- | RE_NO_STREAM- | RE_STREAM_DELETED- | RE_ERROR- | RE_ACCESS_DENIED- deriving (Eq, Enum, Show)---------------------------------------------------------------------------------- data EventRecord = EventRecord { eventRecordStreamId :: Required 1 (Value Text)@@ -427,112 +305,6 @@ instance Decode ResolvedIndexedEvent ---------------------------------------------------------------------------------data ReadEventCompleted- = ReadEventCompleted- { readCompletedResult :: Required 1 (Enumeration ReadEventResult)- , readCompletedIndexedEvent :: Required 2 (Message ResolvedIndexedEvent)- , readCompletedError :: Optional 3 (Value Text)- }- deriving (Generic, Show)-----------------------------------------------------------------------------------instance Decode ReadEventCompleted-----------------------------------------------------------------------------------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-----------------------------------------------------------------------------------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-----------------------------------------------------------------------------------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- }-----------------------------------------------------------------------------------data ReadAllResult- = RA_SUCCESS- | RA_NOT_MODIFIED- | RA_ERROR- | RA_ACCESS_DENIED- deriving (Eq, Enum, Show)---------------------------------------------------------------------------------- data ResolvedEventBuf = ResolvedEventBuf { resolvedEventBufEvent :: Required 1 (Message EventRecord)@@ -546,22 +318,6 @@ instance Decode ResolvedEventBuf ---------------------------------------------------------------------------------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---------------------------------------------------------------------------------- -- Result -------------------------------------------------------------------------------- data Decision@@ -675,111 +431,12 @@ fmap recordedEventStreamId . resolvedEventOriginal ---------------------------------------------------------------------------------data ReadResult- = ReadResult- { readResultStatus :: !ReadEventResult- , readResultStreamId :: !Text- , readResultEventNumber :: !Int32- , 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- }---------------------------------------------------------------------------------- data ReadDirection = Forward | Backward deriving Show ---------------------------------------------------------------------------------data StreamEventsSlice- = StreamEventsSlice- { streamEventsSliceResult :: !ReadStreamResult- , streamEventsSliceStreamId :: !Text- , streamEventsSliceStart :: !Int32- , streamEventsSliceNext :: !Int32- , streamEventsSliceLast :: !Int32- , streamEventsSliceIsEOS :: !Bool- , streamEventsSliceEvents :: ![ResolvedEvent]- , streamEventsSliceDirection :: !ReadDirection- }- 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- }-----------------------------------------------------------------------------------data AllEventsSlice- = AllEventsSlice- { allEventsSliceResult :: !ReadAllResult- , allEventsSliceFrom :: !Position- , allEventsSliceNext :: !Position- , allEventsSliceIsEOS :: !Bool- , allEventsSliceEvents :: ![ResolvedEvent]- , allEventsSliceDirection :: !ReadDirection- }- 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- }---------------------------------------------------------------------------------- -- Transaction -------------------------------------------------------------------------------- data Transaction@@ -806,115 +463,16 @@ flagWord8 Authenticated = 0x01 ----------------------------------------------------------------------------------- Command----------------------------------------------------------------------------------data Command- = HeartbeatRequest- | HeartbeatResponse- | WriteEventsCmd- | WriteEventsCompletedCmd- | DeleteStreamCmd- | DeleteStreamCompletedCmd- | TransactionStartCmd- | TransactionStartCompletedCmd- | TransactionWriteCmd- | TransactionWriteCompletedCmd- | TransactionCommitCmd- | TransactionCommitCompletedCmd- | ReadEventCmd- | ReadEventCompletedCmd- | ReadStreamEventsForwardCmd- | ReadStreamEventsForwardCompletedCmd- | ReadStreamEventsBackwardCmd- | ReadStreamEventsBackwardCompletedCmd- | ReadAllEventsForwardCmd- | ReadAllEventsForwardCompletedCmd- | ReadAllEventsBackwardCmd- | ReadAllEventsBackwardCompletedCmd- -- | CreateChunk- -- | BadRequest- -- | NotHandled- deriving (Eq, Show)-----------------------------------------------------------------------------------cmdWord8 :: Command -> Word8-cmdWord8 cmd =- case cmd of- HeartbeatRequest -> 0x01- HeartbeatResponse -> 0x02- WriteEventsCmd -> 0x82- WriteEventsCompletedCmd -> 0x83- DeleteStreamCmd -> 0x8A- DeleteStreamCompletedCmd -> 0x8B- TransactionStartCmd -> 0x84- TransactionStartCompletedCmd -> 0x85- TransactionWriteCmd -> 0x86- TransactionWriteCompletedCmd -> 0x87- TransactionCommitCmd -> 0x88- TransactionCommitCompletedCmd -> 0x89- ReadEventCmd -> 0xB0- ReadEventCompletedCmd -> 0xB1- ReadStreamEventsForwardCmd -> 0xB2- ReadStreamEventsForwardCompletedCmd -> 0xB3- ReadStreamEventsBackwardCmd -> 0xB4- ReadStreamEventsBackwardCompletedCmd -> 0xB5- ReadAllEventsForwardCmd -> 0xB6- ReadAllEventsForwardCompletedCmd -> 0xB7- ReadAllEventsBackwardCmd -> 0xB8- ReadAllEventsBackwardCompletedCmd -> 0xB9- -- CreateChunk -> 0x12- -- BadRequest -> 0xF0- -- NotHandled -> 0xF1-----------------------------------------------------------------------------------word8Cmd :: Word8 -> Maybe Command-word8Cmd wd =- case wd of- 0x01 -> Just HeartbeatRequest- 0x02 -> Just HeartbeatResponse- 0x82 -> Just WriteEventsCmd- 0x83 -> Just WriteEventsCompletedCmd- 0x8A -> Just DeleteStreamCmd- 0x8B -> Just DeleteStreamCompletedCmd- 0x84 -> Just TransactionStartCmd- 0x85 -> Just TransactionStartCompletedCmd- 0x86 -> Just TransactionWriteCmd- 0x87 -> Just TransactionWriteCompletedCmd- 0x88 -> Just TransactionCommitCmd- 0x89 -> Just TransactionCommitCompletedCmd- 0xB0 -> Just ReadEventCmd- 0xB1 -> Just ReadEventCompletedCmd- 0xB2 -> Just ReadStreamEventsForwardCmd- 0xB3 -> Just ReadStreamEventsForwardCompletedCmd- 0xB4 -> Just ReadStreamEventsBackwardCmd- 0xB5 -> Just ReadStreamEventsBackwardCompletedCmd- 0xB6 -> Just ReadAllEventsForwardCmd- 0xB7 -> Just ReadAllEventsForwardCompletedCmd- 0xB8 -> Just ReadAllEventsBackwardCmd- 0xB9 -> Just ReadAllEventsBackwardCompletedCmd- -- 0x12 -> Just CreateChunk- -- 0xF0 -> Just BadRequest- -- 0xF1 -> Just NotHandled- _ -> Nothing----------------------------------------------------------------------------------+-- Package -------------------------------------------------------------------------------- data Package = Package- { packageCmd :: !Command+ { packageCmd :: !Word8 , packageFlag :: !Flag , packageCorrelation :: !UUID , packageData :: !ByteString } deriving Show-----------------------------------------------------------------------------------data Msg- = Reconnect- | RecvPackage Package- | RegisterOperation Operation- | Notice String- | Tick -------------------------------------------------------------------------------- -- Settings
+ Database/EventStore/Internal/Util/Sodium.hs view
@@ -0,0 +1,35 @@+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Internal.Util.Sodium+-- 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.Util.Sodium where++--------------------------------------------------------------------------------+import Control.Concurrent (forkIO)+import Data.Functor (void)++--------------------------------------------------------------------------------+import FRP.Sodium++--------------------------------------------------------------------------------+pushAsync :: (a -> Reactive ()) -> a -> IO ()+pushAsync push a = void $ forkIO $ sync $ push a++--------------------------------------------------------------------------------+pushAsync2 :: (a -> b -> Reactive ()) -> a -> b -> IO ()+pushAsync2 push a b = void $ forkIO $ sync $ push a b++--------------------------------------------------------------------------------+pushAsync3 :: (a -> b -> c -> Reactive ()) -> a -> b -> c -> IO ()+pushAsync3 push a b c = void $ forkIO $ sync $ push a b c++--------------------------------------------------------------------------------+pushAsync4 :: (a -> b -> c -> d -> Reactive ()) -> a -> b -> c -> d -> IO ()+pushAsync4 push a b c d = void $ forkIO $ sync $ push a b c d
+ Database/EventStore/Internal/Writer.hs view
@@ -0,0 +1,32 @@+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Internal.Writer+-- 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.Writer (writerThread) where++--------------------------------------------------------------------------------+import Control.Monad+import Control.Concurrent.STM+import qualified Data.ByteString as B+import System.IO++--------------------------------------------------------------------------------+import Data.Serialize.Put++--------------------------------------------------------------------------------+import Database.EventStore.Internal.Packages+import Database.EventStore.Internal.Types++--------------------------------------------------------------------------------+writerThread :: TChan Package -> Handle -> IO ()+writerThread chan hdl = forever $ do+ pkg <- atomically $ readTChan chan+ B.hPut hdl (runPut $ putPackage pkg)+ hFlush hdl
eventstore.cabal view
@@ -10,7 +10,7 @@ -- PVP summary: +-+------- breaking API changes -- | | +----- non-breaking API additions -- | | | +--- code changes with no API change-version: 0.1.0.0+version: 0.1.0.1 -- A short (one-line) description of the package. synopsis: EventStore Haskell TCP Client@@ -58,13 +58,15 @@ Database.EventStore.Internal.Processor Database.EventStore.Internal.Reader Database.EventStore.Internal.Types- Database.EventStore.Internal.Operation.Common+ Database.EventStore.Internal.Manager.Operation Database.EventStore.Internal.Operation.DeleteStreamOperation Database.EventStore.Internal.Operation.ReadAllEventsOperation Database.EventStore.Internal.Operation.ReadEventOperation Database.EventStore.Internal.Operation.ReadStreamEventsOperation Database.EventStore.Internal.Operation.TransactionStartOperation Database.EventStore.Internal.Operation.WriteEventsOperation+ Database.EventStore.Internal.Util.Sodium+ Database.EventStore.Internal.Writer -- LANGUAGE extensions used by modules in this package. -- other-extensions:@@ -74,18 +76,16 @@ , attoparsec >=0.12 && <0.13 , aeson >=0.8 && <0.9 , async >=2.0 && <2.1- , binary ==0.7.* , bytestring >=0.10.4 && <0.10.5 , cereal >=0.4 && <0.5 , containers >=0.5 && <0.6 , network ==2.6.* , protobuf >=0.2 && <0.3 , random ==1.*+ , sodium ==0.11.* , stm >=2.2 && <2.5- , suspend ==0.2.* , text >=1.1.1 && <1.2 , time >=1.4 && <1.6- , timers ==0.2.* , uuid ==1.3.* -- Directories containing source files.