diff --git a/Database/EventStore.hs b/Database/EventStore.hs --- a/Database/EventStore.hs +++ b/Database/EventStore.hs @@ -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)) diff --git a/Database/EventStore/Internal/Manager/Operation.hs b/Database/EventStore/Internal/Manager/Operation.hs new file mode 100644 --- /dev/null +++ b/Database/EventStore/Internal/Manager/Operation.hs @@ -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 +-- 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 diff --git a/Database/EventStore/Internal/Operation/Common.hs b/Database/EventStore/Internal/Operation/Common.hs deleted file mode 100644 --- a/Database/EventStore/Internal/Operation/Common.hs +++ /dev/null @@ -1,80 +0,0 @@ --------------------------------------------------------------------------------- --- | --- Module : Database.EventStore.Internal.Operation.Common --- Copyright : (C) 2014 Yorick Laupa --- License : (see the file LICENSE) --- --- Maintainer : Yorick Laupa --- 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 diff --git a/Database/EventStore/Internal/Operation/DeleteStreamOperation.hs b/Database/EventStore/Internal/Operation/DeleteStreamOperation.hs --- a/Database/EventStore/Internal/Operation/DeleteStreamOperation.hs +++ b/Database/EventStore/Internal/Operation/DeleteStreamOperation.hs @@ -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) diff --git a/Database/EventStore/Internal/Operation/ReadAllEventsOperation.hs b/Database/EventStore/Internal/Operation/ReadAllEventsOperation.hs --- a/Database/EventStore/Internal/Operation/ReadAllEventsOperation.hs +++ b/Database/EventStore/Internal/Operation/ReadAllEventsOperation.hs @@ -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) diff --git a/Database/EventStore/Internal/Operation/ReadEventOperation.hs b/Database/EventStore/Internal/Operation/ReadEventOperation.hs --- a/Database/EventStore/Internal/Operation/ReadEventOperation.hs +++ b/Database/EventStore/Internal/Operation/ReadEventOperation.hs @@ -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) diff --git a/Database/EventStore/Internal/Operation/ReadStreamEventsOperation.hs b/Database/EventStore/Internal/Operation/ReadStreamEventsOperation.hs --- a/Database/EventStore/Internal/Operation/ReadStreamEventsOperation.hs +++ b/Database/EventStore/Internal/Operation/ReadStreamEventsOperation.hs @@ -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) diff --git a/Database/EventStore/Internal/Operation/TransactionStartOperation.hs b/Database/EventStore/Internal/Operation/TransactionStartOperation.hs --- a/Database/EventStore/Internal/Operation/TransactionStartOperation.hs +++ b/Database/EventStore/Internal/Operation/TransactionStartOperation.hs @@ -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)) diff --git a/Database/EventStore/Internal/Operation/WriteEventsOperation.hs b/Database/EventStore/Internal/Operation/WriteEventsOperation.hs --- a/Database/EventStore/Internal/Operation/WriteEventsOperation.hs +++ b/Database/EventStore/Internal/Operation/WriteEventsOperation.hs @@ -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) diff --git a/Database/EventStore/Internal/Packages.hs b/Database/EventStore/Internal/Packages.hs --- a/Database/EventStore/Internal/Packages.hs +++ b/Database/EventStore/Internal/Packages.hs @@ -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 diff --git a/Database/EventStore/Internal/Processor.hs b/Database/EventStore/Internal/Processor.hs --- a/Database/EventStore/Internal/Processor.hs +++ b/Database/EventStore/Internal/Processor.hs @@ -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 diff --git a/Database/EventStore/Internal/Reader.hs b/Database/EventStore/Internal/Reader.hs --- a/Database/EventStore/Internal/Reader.hs +++ b/Database/EventStore/Internal/Reader.hs @@ -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 diff --git a/Database/EventStore/Internal/Types.hs b/Database/EventStore/Internal/Types.hs --- a/Database/EventStore/Internal/Types.hs +++ b/Database/EventStore/Internal/Types.hs @@ -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 diff --git a/Database/EventStore/Internal/Util/Sodium.hs b/Database/EventStore/Internal/Util/Sodium.hs new file mode 100644 --- /dev/null +++ b/Database/EventStore/Internal/Util/Sodium.hs @@ -0,0 +1,35 @@ +-------------------------------------------------------------------------------- +-- | +-- Module : Database.EventStore.Internal.Util.Sodium +-- Copyright : (C) 2014 Yorick Laupa +-- License : (see the file LICENSE) +-- +-- Maintainer : Yorick Laupa +-- 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 diff --git a/Database/EventStore/Internal/Writer.hs b/Database/EventStore/Internal/Writer.hs new file mode 100644 --- /dev/null +++ b/Database/EventStore/Internal/Writer.hs @@ -0,0 +1,32 @@ +-------------------------------------------------------------------------------- +-- | +-- Module : Database.EventStore.Internal.Writer +-- Copyright : (C) 2014 Yorick Laupa +-- License : (see the file LICENSE) +-- +-- Maintainer : Yorick Laupa +-- 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 diff --git a/eventstore.cabal b/eventstore.cabal --- a/eventstore.cabal +++ b/eventstore.cabal @@ -10,7 +10,7 @@ -- PVP summary: +-+------- breaking API changes -- | | +----- non-breaking API additions -- | | | +--- code changes with no API change -version: 0.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.