diff --git a/Database/EventStore.hs b/Database/EventStore.hs new file mode 100644 --- /dev/null +++ b/Database/EventStore.hs @@ -0,0 +1,290 @@ +-------------------------------------------------------------------------------- +-- | +-- Module : Database.EventStore +-- Copyright : (C) 2014 Yorick Laupa +-- License : (see the file LICENSE) +-- +-- Maintainer : Yorick Laupa +-- Stability : provisional +-- Portability : non-portable +-- +-------------------------------------------------------------------------------- +module Database.EventStore + ( Event + , EventData + , EventStoreConnection + , ExpectedVersion(..) + , HostName + , Port + -- * Result + , AllEventsSlice(..) + , DeleteResult(..) + , WriteResult(..) + , ReadResult(..) + , RecordedEvent(..) + , StreamEventsSlice(..) + , eventResolved + , resolvedEventOriginal + , resolvedEventOriginalStreamId + -- * Event + , createEvent + , withJson + , withJsonAndMetadata + -- * Connection manager + , defaultSettings + , connect + , deleteStream + , readEvent + , readAllEventsBackward + , readAllEventsForward + , readStreamEventsBackward + , readStreamEventsForward + , sendEvent + , sendEvents + , shutdown + , transactionStart + -- * Transaction + , Transaction + , transactionCommit + , transactionRollback + , transactionSendEvents + -- * Re-export + , module Control.Concurrent.Async + ) where + +-------------------------------------------------------------------------------- +import Control.Concurrent +import Control.Concurrent.STM +import Data.Int + +-------------------------------------------------------------------------------- +import Control.Concurrent.Async +import Data.Text + +-------------------------------------------------------------------------------- +import Database.EventStore.Internal.Processor +import Database.EventStore.Internal.Types +import Database.EventStore.Internal.Operation.DeleteStreamOperation +import Database.EventStore.Internal.Operation.ReadAllEventsOperation +import Database.EventStore.Internal.Operation.ReadEventOperation +import Database.EventStore.Internal.Operation.ReadStreamEventsOperation +import Database.EventStore.Internal.Operation.TransactionStartOperation +import Database.EventStore.Internal.Operation.WriteEventsOperation + +-------------------------------------------------------------------------------- +type HostName = String +type Port = Int + +-------------------------------------------------------------------------------- +-- EventStoreConnection +-------------------------------------------------------------------------------- +data EventStoreConnection + = EventStoreConnection + { mgrChan :: TChan Msg + , mgrSettings :: Settings + , mgrThreadId :: ThreadId + } + +-------------------------------------------------------------------------------- +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 +-- 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 +-- 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 host port = do + chan <- newTChanIO + app <- newProcessor settings chan host port + tid <- forkFinally (appProcess app) (\_ -> appFinalizer app) + + return $ EventStoreConnection chan settings tid + +-------------------------------------------------------------------------------- +shutdown :: EventStoreConnection -> IO () +shutdown mgr = killThread tid + where + tid = mgrThreadId mgr + +-------------------------------------------------------------------------------- +sendEvent :: EventStoreConnection + -> Text -- ^ Stream + -> ExpectedVersion + -> Event + -> IO (Async WriteResult) +sendEvent mgr evt_stream exp_ver evt = + sendEvents mgr evt_stream exp_ver [evt] + +-------------------------------------------------------------------------------- +sendEvents :: EventStoreConnection + -> Text -- ^ Stream + -> ExpectedVersion + -> [Event] + -> IO (Async WriteResult) +sendEvents mgr evt_stream exp_ver evts = do + (as, mvar) <- createAsync + + let op = writeEventsOperation settings mvar evt_stream exp_ver evts + + msgQueue mgr (RegisterOperation op) + return as + where + settings = mgrSettings mgr + +-------------------------------------------------------------------------------- +deleteStream :: EventStoreConnection + -> Text + -> ExpectedVersion + -> Maybe Bool -- ^ Hard delete + -> IO (Async DeleteResult) +deleteStream mgr evt_stream exp_ver hard_del = do + (as, mvar) <- createAsync + + let op = deleteStreamOperation settings mvar evt_stream exp_ver hard_del + + msgQueue mgr (RegisterOperation op) + return as + where + settings = mgrSettings mgr + +-------------------------------------------------------------------------------- +transactionStart :: EventStoreConnection + -> Text + -> ExpectedVersion + -> IO (Async Transaction) +transactionStart mgr evt_stream exp_ver = do + (as, mvar) <- createAsync + + let op = transactionStartOperation settings chan mvar evt_stream exp_ver + + msgQueue mgr (RegisterOperation op) + return as + where + chan = mgrChan mgr + settings = mgrSettings mgr + +-------------------------------------------------------------------------------- +readEvent :: EventStoreConnection + -> Text + -> Int32 + -> Bool + -> IO (Async ReadResult) +readEvent mgr stream_id evt_num res_link_tos = do + (as, mvar) <- createAsync + + let op = readEventOperation settings mvar stream_id evt_num res_link_tos + + msgQueue mgr (RegisterOperation op) + return as + where + settings = mgrSettings mgr + +-------------------------------------------------------------------------------- +readStreamEventsForward :: EventStoreConnection + -> Text + -> Int32 + -> Int32 + -> Bool + -> IO (Async StreamEventsSlice) +readStreamEventsForward mgr = + readStreamEventsCommon mgr Forward + +-------------------------------------------------------------------------------- +readStreamEventsBackward :: EventStoreConnection + -> Text + -> Int32 + -> Int32 + -> Bool + -> IO (Async StreamEventsSlice) +readStreamEventsBackward mgr = + readStreamEventsCommon mgr Backward + +-------------------------------------------------------------------------------- +readStreamEventsCommon :: EventStoreConnection + -> ReadDirection + -> Text + -> Int32 + -> Int32 + -> Bool + -> IO (Async StreamEventsSlice) +readStreamEventsCommon mgr dir stream_id start cnt res_link_tos = do + (as, mvar) <- createAsync + + let op = readStreamEventsOperation settings + dir + mvar + stream_id + start + cnt + res_link_tos + + msgQueue mgr (RegisterOperation op) + return as + where + settings = mgrSettings mgr + +-------------------------------------------------------------------------------- +readAllEventsForward :: EventStoreConnection + -> Int64 + -> Int64 + -> Int32 + -> Bool + -> IO (Async AllEventsSlice) +readAllEventsForward mgr = + readAllEventsCommon mgr Forward + +-------------------------------------------------------------------------------- +readAllEventsBackward :: EventStoreConnection + -> Int64 + -> Int64 + -> Int32 + -> Bool + -> IO (Async AllEventsSlice) +readAllEventsBackward mgr = + readAllEventsCommon mgr Backward + +-------------------------------------------------------------------------------- +readAllEventsCommon :: EventStoreConnection + -> ReadDirection + -> Int64 + -> Int64 + -> Int32 + -> Bool + -> IO (Async AllEventsSlice) +readAllEventsCommon mgr dir c_pos p_pos max_c res_link_tos = do + (as, mvar) <- createAsync + + let op = readAllEventsOperation settings + dir + mvar + c_pos + p_pos + max_c + res_link_tos + + msgQueue mgr (RegisterOperation op) + return as + where + settings = mgrSettings mgr + +-------------------------------------------------------------------------------- +createAsync :: IO (Async a, TMVar (OperationExceptional a)) +createAsync = do + mvar <- atomically newEmptyTMVar + as <- async $ atomically $ do + res <- readTMVar mvar + either throwSTM return res + + return (as, mvar) diff --git a/Database/EventStore/Internal/Operation/Common.hs b/Database/EventStore/Internal/Operation/Common.hs new file mode 100644 --- /dev/null +++ b/Database/EventStore/Internal/Operation/Common.hs @@ -0,0 +1,80 @@ +-------------------------------------------------------------------------------- +-- | +-- 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 new file mode 100644 --- /dev/null +++ b/Database/EventStore/Internal/Operation/DeleteStreamOperation.hs @@ -0,0 +1,94 @@ +-------------------------------------------------------------------------------- +-- | +-- Module : Database.EventStore.Internal.Operation.DeleteStreamOperation +-- Copyright : (C) 2014 Yorick Laupa +-- License : (see the file LICENSE) +-- +-- Maintainer : Yorick Laupa +-- Stability : provisional +-- Portability : non-portable +-- +-------------------------------------------------------------------------------- +module Database.EventStore.Internal.Operation.DeleteStreamOperation + ( deleteStreamOperation ) where + +-------------------------------------------------------------------------------- +import Control.Concurrent.STM +import Data.Maybe + +-------------------------------------------------------------------------------- +import Data.Text + +-------------------------------------------------------------------------------- +import Database.EventStore.Internal.Operation.Common +import Database.EventStore.Internal.Types + +-------------------------------------------------------------------------------- +deleteStreamOperation :: Settings + -> TMVar (OperationExceptional DeleteResult) + -> Text + -> ExpectedVersion + -> Maybe Bool + -> Operation +deleteStreamOperation settings mvar stream_id exp_ver hard_del = + createOperation params + where + params = OperationParams + { opSettings = settings + , opRequestCmd = DeleteStreamCmd + , opResponseCmd = DeleteStreamCompletedCmd + + , 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 + } + +-------------------------------------------------------------------------------- +inspect :: TMVar (OperationExceptional DeleteResult) + -> Text + -> ExpectedVersion + -> DeleteStreamCompleted + -> IO Decision +inspect mvar stream exp_ver dsc = go (getField $ deleteCompletedResult dsc) + where + go OP_SUCCESS = succeed mvar dsc + go OP_PREPARE_TIMEOUT = return Retry + go OP_FORWARD_TIMEOUT = return Retry + go OP_COMMIT_TIMEOUT = return Retry + go OP_WRONG_EXPECTED_VERSION = failed mvar wrong_version + go OP_STREAM_DELETED = failed mvar (StreamDeleted stream) + go OP_INVALID_TRANSACTION = failed mvar InvalidTransaction + go OP_ACCESS_DENIED = failed mvar (AccessDenied stream) + + wrong_version = WrongExpectedVersion stream exp_ver + +-------------------------------------------------------------------------------- +succeed :: TMVar (OperationExceptional DeleteResult) + -> DeleteStreamCompleted + -> IO Decision +succeed mvar wec = do + atomically $ putTMVar mvar (Right wr) + return EndOperation + where + com_pos = getField $ deleteCompletedCommitPosition wec + pre_pos = getField $ deleteCompletedPreparePosition wec + com_pos_int = fromMaybe (-1) com_pos + pre_pos_int = fromMaybe (-1) pre_pos + pos = Position com_pos_int pre_pos_int + wr = DeleteResult pos + +-------------------------------------------------------------------------------- +failed :: TMVar (OperationExceptional DeleteResult) + -> OperationException + -> IO Decision +failed mvar e = do + atomically $ putTMVar mvar (Left e) + return EndOperation diff --git a/Database/EventStore/Internal/Operation/ReadAllEventsOperation.hs b/Database/EventStore/Internal/Operation/ReadAllEventsOperation.hs new file mode 100644 --- /dev/null +++ b/Database/EventStore/Internal/Operation/ReadAllEventsOperation.hs @@ -0,0 +1,94 @@ +{-# LANGUAGE OverloadedStrings #-} +-------------------------------------------------------------------------------- +-- | +-- Module : Database.EventStore.Internal.Operation.ReadAllEventsOperation +-- Copyright : (C) 2014 Yorick Laupa +-- License : (see the file LICENSE) +-- +-- Maintainer : Yorick Laupa +-- Stability : provisional +-- Portability : non-portable +-- +-------------------------------------------------------------------------------- +module Database.EventStore.Internal.Operation.ReadAllEventsOperation + ( readAllEventsOperation ) where + +-------------------------------------------------------------------------------- +import Control.Concurrent.STM +import Data.Int +import Data.Maybe + +-------------------------------------------------------------------------------- +import Database.EventStore.Internal.Operation.Common +import Database.EventStore.Internal.Types + +-------------------------------------------------------------------------------- +readAllEventsOperation :: Settings + -> ReadDirection + -> TMVar (OperationExceptional AllEventsSlice) + -> Int64 + -> Int64 + -> Int32 + -> Bool + -> Operation +readAllEventsOperation settings dir mvar c_pos p_pos max_c res_link_tos = + createOperation params + where + req = case dir of + Forward -> ReadAllEventsForwardCmd + Backward -> ReadAllEventsBackwardCmd + + 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 + } + +-------------------------------------------------------------------------------- +inspect :: TMVar (OperationExceptional AllEventsSlice) + -> ReadDirection + -> ReadAllEventsCompleted + -> IO Decision +inspect mvar dir raec = go res + where + res = fromMaybe RA_SUCCESS (getField $ readAECResult raec) + may_err = getField $ readAECError raec + + go RA_ERROR = failed mvar (ServerError may_err) + go RA_ACCESS_DENIED = failed mvar (AccessDenied "$all") + go _ = succeed mvar dir raec + +-------------------------------------------------------------------------------- +succeed :: TMVar (OperationExceptional AllEventsSlice) + -> ReadDirection + -> ReadAllEventsCompleted + -> IO Decision +succeed mvar dir raec = do + atomically $ putTMVar mvar (Right ses) + return EndOperation + where + ses = newAllEventsSlice dir raec + +-------------------------------------------------------------------------------- +failed :: TMVar (OperationExceptional AllEventsSlice) + -> OperationException + -> IO Decision +failed mvar e = do + atomically $ putTMVar mvar (Left e) + return EndOperation diff --git a/Database/EventStore/Internal/Operation/ReadEventOperation.hs b/Database/EventStore/Internal/Operation/ReadEventOperation.hs new file mode 100644 --- /dev/null +++ b/Database/EventStore/Internal/Operation/ReadEventOperation.hs @@ -0,0 +1,87 @@ +-------------------------------------------------------------------------------- +-- | +-- Module : Database.EventStore.Internal.Operation.ReadEventOperation +-- Copyright : (C) 2014 Yorick Laupa +-- License : (see the file LICENSE) +-- +-- Maintainer : Yorick Laupa +-- Stability : provisional +-- Portability : non-portable +-- +-------------------------------------------------------------------------------- +module Database.EventStore.Internal.Operation.ReadEventOperation + ( readEventOperation ) where + +-------------------------------------------------------------------------------- +import Control.Concurrent.STM +import Data.Int + +-------------------------------------------------------------------------------- +import Data.Text + +-------------------------------------------------------------------------------- +import Database.EventStore.Internal.Operation.Common +import Database.EventStore.Internal.Types + +-------------------------------------------------------------------------------- +readEventOperation :: Settings + -> TMVar (OperationExceptional ReadResult) + -> Text + -> Int32 + -> Bool -- ^ Resolve link TOS + -> Operation +readEventOperation settings mvar stream_id evt_num res_link_tos = + createOperation params + where + params = OperationParams + { opSettings = settings + , opRequestCmd = ReadEventCmd + , opResponseCmd = ReadEventCompletedCmd + + , 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 + } + +-------------------------------------------------------------------------------- +inspect :: TMVar (OperationExceptional ReadResult) + -> Text + -> Int32 + -> ReadEventCompleted + -> IO Decision +inspect mvar stream_id evt_num reco = go (getField $ readCompletedResult reco) + where + may_err = getField $ readCompletedError reco + + go RE_ERROR = failed mvar (ServerError may_err) + go RE_ACCESS_DENIED = failed mvar (AccessDenied stream_id) + go _ = succeed mvar stream_id evt_num reco + +-------------------------------------------------------------------------------- +succeed :: TMVar (OperationExceptional ReadResult) + -> Text + -> Int32 + -> ReadEventCompleted + -> IO Decision +succeed mvar stream_id evt_num reco = do + atomically $ putTMVar mvar (Right rr) + return EndOperation + where + status = getField $ readCompletedResult reco + rie = getField $ readCompletedIndexedEvent reco + rr = newReadResult status stream_id evt_num rie + +-------------------------------------------------------------------------------- +failed :: TMVar (OperationExceptional ReadResult) + -> OperationException + -> IO Decision +failed mvar e = do + atomically $ putTMVar mvar (Left e) + return EndOperation diff --git a/Database/EventStore/Internal/Operation/ReadStreamEventsOperation.hs b/Database/EventStore/Internal/Operation/ReadStreamEventsOperation.hs new file mode 100644 --- /dev/null +++ b/Database/EventStore/Internal/Operation/ReadStreamEventsOperation.hs @@ -0,0 +1,98 @@ +-------------------------------------------------------------------------------- +-- | +-- Module : Database.EventStore.Internal.Operation.ReadStreamEventsOperation +-- Copyright : (C) 2014 Yorick Laupa +-- License : (see the file LICENSE) +-- +-- Maintainer : Yorick Laupa +-- Stability : provisional +-- Portability : non-portable +-- +-------------------------------------------------------------------------------- +module Database.EventStore.Internal.Operation.ReadStreamEventsOperation + ( readStreamEventsOperation ) where + +-------------------------------------------------------------------------------- +import Control.Concurrent.STM +import Data.Int + +-------------------------------------------------------------------------------- +import Data.Text + +-------------------------------------------------------------------------------- +import Database.EventStore.Internal.Operation.Common +import Database.EventStore.Internal.Types + +-------------------------------------------------------------------------------- +readStreamEventsOperation :: Settings + -> ReadDirection + -> TMVar (OperationExceptional StreamEventsSlice) + -> Text + -> Int32 + -> Int32 + -> Bool + -> Operation +readStreamEventsOperation settings dir mvar stream_id start cnt res_link_tos = + createOperation params + where + req = case dir of + Forward -> ReadStreamEventsForwardCmd + Backward -> ReadStreamEventsBackwardCmd + + 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 + } + +-------------------------------------------------------------------------------- +inspect :: TMVar (OperationExceptional StreamEventsSlice) + -> ReadDirection + -> Text + -> Int32 + -> ReadStreamEventsCompleted + -> IO Decision +inspect mvar dir stream_id start rsec = go (getField $ readSECResult rsec) + where + may_err = getField $ readSECError rsec + + go RS_ERROR = failed mvar (ServerError may_err) + go RS_ACCESS_DENIED = failed mvar (AccessDenied stream_id) + go _ = succeed mvar dir stream_id start rsec + +-------------------------------------------------------------------------------- +succeed :: TMVar (OperationExceptional StreamEventsSlice) + -> ReadDirection + -> Text + -> Int32 + -> ReadStreamEventsCompleted + -> IO Decision +succeed mvar dir stream_id start rsec = do + atomically $ putTMVar mvar (Right ses) + return EndOperation + where + ses = newStreamEventsSlice stream_id start dir rsec + +-------------------------------------------------------------------------------- +failed :: TMVar (OperationExceptional StreamEventsSlice) + -> OperationException + -> IO Decision +failed mvar e = do + atomically $ putTMVar mvar (Left e) + return EndOperation diff --git a/Database/EventStore/Internal/Operation/TransactionStartOperation.hs b/Database/EventStore/Internal/Operation/TransactionStartOperation.hs new file mode 100644 --- /dev/null +++ b/Database/EventStore/Internal/Operation/TransactionStartOperation.hs @@ -0,0 +1,289 @@ +-------------------------------------------------------------------------------- +-- | +-- Module : Database.EventStore.Internal.Operation.TransactionStartOperation +-- Copyright : (C) 2014 Yorick Laupa +-- License : (see the file LICENSE) +-- +-- Maintainer : Yorick Laupa +-- Stability : provisional +-- Portability : non-portable +-- +-------------------------------------------------------------------------------- +module Database.EventStore.Internal.Operation.TransactionStartOperation + ( transactionStartOperation ) where + +-------------------------------------------------------------------------------- +import Control.Concurrent.STM +import Data.Maybe +import Data.Traversable + +-------------------------------------------------------------------------------- +import Data.Int +import Data.Text + +-------------------------------------------------------------------------------- +import Control.Concurrent.Async +import Database.EventStore.Internal.Operation.Common +import Database.EventStore.Internal.Types + +-------------------------------------------------------------------------------- +data TransactionEnv + = TransactionEnv + { _transSettings :: Settings + , _transChan :: TChan Msg + , _transStreamId :: Text + , _transExpectedVersion :: ExpectedVersion + } + +-------------------------------------------------------------------------------- +transactionStartOperation :: Settings + -> TChan Msg + -> TMVar (OperationExceptional Transaction) + -> Text + -> ExpectedVersion + -> Operation +transactionStartOperation settings chan mvar stream_id exp_ver = + createOperation params + where + env = TransactionEnv + { _transSettings = settings + , _transChan = chan + , _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) + -> TransactionStartCompleted + -> IO Decision +inspectTrans env mvar tsc = go (getField $ transactionSCResult tsc) + where + go OP_SUCCESS = succeedTrans env mvar tsc + go OP_PREPARE_TIMEOUT = return Retry + go OP_FORWARD_TIMEOUT = return Retry + go OP_COMMIT_TIMEOUT = return Retry + go OP_WRONG_EXPECTED_VERSION = failed mvar wrong_version + go OP_STREAM_DELETED = failed mvar (StreamDeleted stream_id) + go OP_INVALID_TRANSACTION = failed mvar InvalidTransaction + go OP_ACCESS_DENIED = failed mvar (AccessDenied stream_id) + + exp_ver = _transExpectedVersion env + stream_id = _transStreamId env + wrong_version = WrongExpectedVersion stream_id exp_ver + +-------------------------------------------------------------------------------- +succeedTrans :: TransactionEnv + -> TMVar (OperationExceptional Transaction) + -> TransactionStartCompleted + -> IO Decision +succeedTrans env mvar tsc = do + atomically $ putTMVar mvar (Right trans) + return EndOperation + where + trans_id = getField $ transactionSCId tsc + trans = createTransaction env trans_id + +-------------------------------------------------------------------------------- +failed :: TMVar (OperationExceptional a) + -> OperationException + -> IO Decision +failed mvar e = do + atomically $ putTMVar mvar (Left e) + return EndOperation + +-------------------------------------------------------------------------------- +createTransaction :: TransactionEnv -> Int64 -> Transaction +createTransaction env 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 + + , transactionCommit = do + (as, mvar) <- createAsync + + let op = transactionCommitOperation env trans_id mvar + + sendMsg chan (RegisterOperation op) + return as + + , transactionSendEvents = \evts -> do + (as, mvar) <- createAsync + + let op = transactionWriteOperation env trans_id mvar evts + + sendMsg chan (RegisterOperation op) + return as + + , transactionRollback = return () + } + +-------------------------------------------------------------------------------- +transactionWriteOperation :: TransactionEnv + -> Int64 + -> TMVar (OperationExceptional ()) + -> [Event] + -> Operation +transactionWriteOperation env trans_id mvar evts = + createOperation params + where + settings = _transSettings env + req_master = _requireMaster settings + + params = OperationParams + { opSettings = settings + , opRequestCmd = TransactionWriteCmd + , opResponseCmd = TransactionWriteCompletedCmd + + , opRequest = do + new_evts <- traverse eventToNewEvent evts + + let request = newTransactionWrite trans_id + new_evts + req_master + + return request + + , opSuccess = inspectWrite env mvar + , opFailure = failed mvar + } + +-------------------------------------------------------------------------------- +transactionCommitOperation :: TransactionEnv + -> Int64 + -> TMVar (OperationExceptional WriteResult) + -> Operation +transactionCommitOperation env trans_id mvar = + createOperation params + where + settings = _transSettings env + req_master = _requireMaster settings + + params = OperationParams + { opSettings = settings + , opRequestCmd = TransactionCommitCmd + , opResponseCmd = TransactionCommitCompletedCmd + + , opRequest = + let request = newTransactionCommit trans_id req_master in + + return request + + , opSuccess = inspectCommit env mvar + , opFailure = failed mvar + } + +-------------------------------------------------------------------------------- +inspectWrite :: TransactionEnv + -> TMVar (OperationExceptional ()) + -> TransactionWriteCompleted + -> IO Decision +inspectWrite env mvar twc = go (getField $ transactionWCResult twc) + where + go OP_SUCCESS = succeedWrite mvar twc + go OP_PREPARE_TIMEOUT = return Retry + go OP_FORWARD_TIMEOUT = return Retry + go OP_COMMIT_TIMEOUT = return Retry + go OP_WRONG_EXPECTED_VERSION = failed mvar wrong_version + go OP_STREAM_DELETED = failed mvar (StreamDeleted stream_id) + go OP_INVALID_TRANSACTION = failed mvar InvalidTransaction + go OP_ACCESS_DENIED = failed mvar (AccessDenied stream_id) + + exp_ver = _transExpectedVersion env + stream_id = _transStreamId env + wrong_version = WrongExpectedVersion stream_id exp_ver + +-------------------------------------------------------------------------------- +succeedWrite :: TMVar (OperationExceptional ()) + -> TransactionWriteCompleted + -> IO Decision +succeedWrite mvar _ = do + atomically $ putTMVar mvar (Right ()) + return EndOperation + +-------------------------------------------------------------------------------- +inspectCommit :: TransactionEnv + -> TMVar (OperationExceptional WriteResult) + -> TransactionCommitCompleted + -> IO Decision +inspectCommit env mvar tcc = go (getField $ transactionCCResult tcc) + where + go OP_SUCCESS = succeedCommit mvar tcc + go OP_PREPARE_TIMEOUT = return Retry + go OP_FORWARD_TIMEOUT = return Retry + go OP_COMMIT_TIMEOUT = return Retry + go OP_WRONG_EXPECTED_VERSION = failed mvar wrong_version + go OP_STREAM_DELETED = failed mvar (StreamDeleted stream_id) + go OP_INVALID_TRANSACTION = failed mvar InvalidTransaction + go OP_ACCESS_DENIED = failed mvar (AccessDenied stream_id) + + exp_ver = _transExpectedVersion env + stream_id = _transStreamId env + wrong_version = WrongExpectedVersion stream_id exp_ver + +-------------------------------------------------------------------------------- +succeedCommit :: TMVar (OperationExceptional WriteResult) + -> TransactionCommitCompleted + -> IO Decision +succeedCommit mvar tcc = do + atomically $ putTMVar mvar (Right wr) + return EndOperation + where + last_evt_num = getField $ transactionCCLastNumber tcc + com_pos = getField $ transactionCCCommitPosition tcc + pre_pos = getField $ transactionCCPreparePosition tcc + com_pos_int = fromMaybe (-1) com_pos + pre_pos_int = fromMaybe (-1) pre_pos + pos = Position com_pos_int pre_pos_int + wr = WriteResult last_evt_num pos + +-------------------------------------------------------------------------------- +eventToNewEvent :: Event -> IO NewEvent +eventToNewEvent evt = + newEvent evt_type + evt_data_type + evt_metadata_type + evt_data_bytes + evt_metadata_bytes + where + evt_type = eventType evt + evt_data_bytes = eventDataBytes $ eventData evt + evt_data_type = eventDataType $ eventData evt + evt_metadata_bytes = eventMetadataBytes $ eventData evt + evt_metadata_type = eventMetadataType $ eventData evt + +-------------------------------------------------------------------------------- +sendMsg :: TChan Msg -> Msg -> IO () +sendMsg chan msg = atomically $ writeTChan chan msg + +-------------------------------------------------------------------------------- +createAsync :: IO (Async a, TMVar (OperationExceptional a)) +createAsync = do + mvar <- atomically newEmptyTMVar + as <- async $ atomically $ do + res <- readTMVar mvar + either throwSTM return res + + return (as, mvar) diff --git a/Database/EventStore/Internal/Operation/WriteEventsOperation.hs b/Database/EventStore/Internal/Operation/WriteEventsOperation.hs new file mode 100644 --- /dev/null +++ b/Database/EventStore/Internal/Operation/WriteEventsOperation.hs @@ -0,0 +1,114 @@ +-------------------------------------------------------------------------------- +-- | +-- Module : Database.EventStore.Internal.Operation.WriteEventsOperation +-- Copyright : (C) 2014 Yorick Laupa +-- License : (see the file LICENSE) +-- +-- Maintainer : Yorick Laupa +-- Stability : provisional +-- Portability : non-portable +-- +-------------------------------------------------------------------------------- +module Database.EventStore.Internal.Operation.WriteEventsOperation + ( writeEventsOperation ) where + +-------------------------------------------------------------------------------- +import Control.Concurrent.STM +import Data.Maybe +import Data.Traversable + +-------------------------------------------------------------------------------- +import Data.Text + +-------------------------------------------------------------------------------- +import Database.EventStore.Internal.Operation.Common +import Database.EventStore.Internal.Types + +-------------------------------------------------------------------------------- +writeEventsOperation :: Settings + -> TMVar (OperationExceptional WriteResult) + -> Text + -> ExpectedVersion + -> [Event] + -> Operation +writeEventsOperation settings mvar evt_stream exp_ver evts = + createOperation params + where + params = + OperationParams + { opSettings = settings + , opRequestCmd = WriteEventsCmd + , opResponseCmd = WriteEventsCompletedCmd + + , opRequest = do + new_evts <- traverse eventToNewEvent evts + + let require_master = _requireMaster settings + exp_ver_int32 = expVersionInt32 exp_ver + request = newWriteEvents evt_stream + exp_ver_int32 + new_evts + require_master + return request + + , opSuccess = inspect mvar evt_stream exp_ver + , opFailure = failed mvar + } + +-------------------------------------------------------------------------------- +inspect :: TMVar (OperationExceptional WriteResult) + -> Text + -> ExpectedVersion + -> WriteEventsCompleted + -> IO Decision +inspect mvar stream exp_ver wec = go (getField $ writeCompletedResult wec) + where + go OP_SUCCESS = succeed mvar wec + go OP_PREPARE_TIMEOUT = return Retry + go OP_FORWARD_TIMEOUT = return Retry + go OP_COMMIT_TIMEOUT = return Retry + go OP_WRONG_EXPECTED_VERSION = failed mvar wrong_version + go OP_STREAM_DELETED = failed mvar (StreamDeleted stream) + go OP_INVALID_TRANSACTION = failed mvar InvalidTransaction + go OP_ACCESS_DENIED = failed mvar (AccessDenied stream) + + wrong_version = WrongExpectedVersion stream exp_ver + +-------------------------------------------------------------------------------- +succeed :: TMVar (OperationExceptional WriteResult) + -> WriteEventsCompleted + -> IO Decision +succeed mvar wec = do + atomically $ putTMVar mvar (Right wr) + return EndOperation + where + last_evt_num = getField $ writeCompletedLastNumber wec + com_pos = getField $ writeCompletedCommitPosition wec + pre_pos = getField $ writeCompletedPreparePosition wec + com_pos_int = fromMaybe (-1) com_pos + pre_pos_int = fromMaybe (-1) pre_pos + pos = Position com_pos_int pre_pos_int + wr = WriteResult last_evt_num pos + +-------------------------------------------------------------------------------- +failed :: TMVar (OperationExceptional WriteResult) + -> OperationException + -> IO Decision +failed mvar e = do + atomically $ putTMVar mvar (Left e) + return EndOperation + +-------------------------------------------------------------------------------- +eventToNewEvent :: Event -> IO NewEvent +eventToNewEvent evt = + newEvent evt_type + evt_data_type + evt_metadata_type + evt_data_bytes + evt_metadata_bytes + where + evt_type = eventType evt + evt_data_bytes = eventDataBytes $ eventData evt + evt_data_type = eventDataType $ eventData evt + evt_metadata_bytes = eventMetadataBytes $ eventData evt + evt_metadata_type = eventMetadataType $ eventData evt diff --git a/Database/EventStore/Internal/Packages.hs b/Database/EventStore/Internal/Packages.hs new file mode 100644 --- /dev/null +++ b/Database/EventStore/Internal/Packages.hs @@ -0,0 +1,105 @@ +-------------------------------------------------------------------------------- +-- | +-- Module : Database.EventStore.Internal.Packages +-- Copyright : (C) 2014 Yorick Laupa +-- License : (see the file LICENSE) +-- +-- Maintainer : Yorick Laupa +-- Stability : provisional +-- Portability : non-portable +-- +-------------------------------------------------------------------------------- +module Database.EventStore.Internal.Packages + ( -- * Package Smart Contructors + deleteStreamPackage + , 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 + +-------------------------------------------------------------------------------- +import Database.EventStore.Internal.Types + +-------------------------------------------------------------------------------- +-- 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 + , packageFlag = None + , packageCorrelation = uuid + , packageData = B.empty + } + + return pack + +-------------------------------------------------------------------------------- +heartbeatResponsePackage :: UUID -> Package +heartbeatResponsePackage uuid = + Package + { packageCmd = HeartbeatResponse + , packageFlag = None + , packageCorrelation = uuid + , packageData = B.empty + } + +-------------------------------------------------------------------------------- +putPackage :: Package -> Put +putPackage pack = + putWord32le length_prefix >> + putWord8 cmd_word8 >> + 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 +mandatorySize = 18 diff --git a/Database/EventStore/Internal/Processor.hs b/Database/EventStore/Internal/Processor.hs new file mode 100644 --- /dev/null +++ b/Database/EventStore/Internal/Processor.hs @@ -0,0 +1,351 @@ +-------------------------------------------------------------------------------- +-- | +-- Module : Database.EventStore.Internal.Processor +-- Copyright : (C) 2014 Yorick Laupa +-- License : (see the file LICENSE) +-- +-- Maintainer : Yorick Laupa +-- Stability : provisional +-- Portability : non-portable +-- +-------------------------------------------------------------------------------- +module Database.EventStore.Internal.Processor + ( Application(..) + , 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 Data.Time +import Data.UUID +import Network +import System.Random + +-------------------------------------------------------------------------------- +import Database.EventStore.Internal.Packages +import Database.EventStore.Internal.Reader +import Database.EventStore.Internal.Types + +-------------------------------------------------------------------------------- +-- Env +-------------------------------------------------------------------------------- +data Env + = Env + { _hostname :: HostName + , _port :: Int + , _settings :: Settings + , _chan :: TChan Msg + , _finalizer :: TVar (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 + +-------------------------------------------------------------------------------- +newEnv :: Settings -> TChan Msg -> HostName -> Int -> IO Env +newEnv settings chan host port = do + ref <- newTVarIO (return ()) + + return $ Env host port settings chan ref + +-------------------------------------------------------------------------------- +registerFinalizer :: Env -> IO () -> IO () +registerFinalizer env action = atomically $ writeTVar var action + where + var = _finalizer env + +-------------------------------------------------------------------------------- +runFinalizer :: Env -> IO () +runFinalizer env = do + action <- atomically $ do + act <- readTVar var + writeTVar var (return ()) + return act + action + where + var = _finalizer env + +-------------------------------------------------------------------------------- +-- Connection +-------------------------------------------------------------------------------- +data Connection + = Connection + { _connId :: UUID + , _connHandle :: Handle + , _connReaderThreadId :: ThreadId + } + +-------------------------------------------------------------------------------- +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 + +-------------------------------------------------------------------------------- +-------------------------------------------------------------------------------- +type Processor = Env -> State -> 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) + } + +-------------------------------------------------------------------------------- +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 } + +-------------------------------------------------------------------------------- +incrPackageNumber :: State -> State +incrPackageNumber cur_state = new_state + where + new_package_number = _packageNumber cur_state + 1 + new_state = cur_state { _packageNumber = new_package_number } + +-------------------------------------------------------------------------------- +-- | 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) + } + + state = State + { _lastTime = cur_time + , _heartbeatInfo = info + , _packageNumber = package_num + , _operations = M.empty + } + + return state + +-------------------------------------------------------------------------------- +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 + } + + return app + +-------------------------------------------------------------------------------- +connecting :: Processor +connecting env state = do + hdl <- connectTo host (PortNumber $ fromIntegral port) + hSetBuffering hdl NoBuffering + + 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 + } + + printf "Connected %s\n" (toString conn_id) + registerFinalizer env $ + connectionClose conn + + connected conn env new_state + + where + port = _port env + host = _hostname env + chan = _chan env + + recovering (Left some_ex)= + case fromException some_ex of + Just e -> + case e of + ConnectionClosedByServer + -> sendMsg env Reconnect + Stopped + -> return () + _ -> sendMsg env Reconnect + + recovering _ = return () + +-------------------------------------------------------------------------------- +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 + +-------------------------------------------------------------------------------- +registerOperation :: Connection -> Operation -> Processor +registerOperation conn op env state = do + uuid <- randomIO + pack <- operationCreatePackage op uuid + + let new_op_map = M.insert uuid op op_map + new_state = state { _operations = new_op_map } + + connectionSend conn (putPackage pack) + connected conn env new_state + where + op_map = _operations state + +-------------------------------------------------------------------------------- +handlePackage :: Connection -> Env -> State -> Package -> IO State +handlePackage conn env state pack = go (packageCmd pack) + where + go HeartbeatRequest = do + handleHeartbeatRequest conn pack + return new_state + + go HeartbeatResponse = + return new_state + + go _ = + case M.lookup corr_id op_map of + Just op -> handleOperation env new_state pack op + _ -> fmap (const new_state) $ unhandledPackage pack + + corr_id = packageCorrelation pack + op_map = _operations state + new_state = incrPackageNumber state + +-------------------------------------------------------------------------------- +handleHeartbeatRequest :: Connection -> Package -> IO () +handleHeartbeatRequest conn pack = + connectionSend conn (putPackage pack_resp) + where + corr_id = packageCorrelation pack + pack_resp = heartbeatResponsePackage corr_id + +-------------------------------------------------------------------------------- +unhandledPackage :: Package -> IO () +unhandledPackage pack = printf "Unhandled command: %s\n" cmd_str + where + cmd_str = show $ packageCmd pack + +-------------------------------------------------------------------------------- +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 } + +-- -------------------------------------------------------------------------------- +-- 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 + +-- 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 diff --git a/Database/EventStore/Internal/Reader.hs b/Database/EventStore/Internal/Reader.hs new file mode 100644 --- /dev/null +++ b/Database/EventStore/Internal/Reader.hs @@ -0,0 +1,97 @@ +-------------------------------------------------------------------------------- +-- | +-- Module : Database.EventStore.Internal.Reader +-- Copyright : (C) 2014 Yorick Laupa +-- License : (see the file LICENSE) +-- +-- Maintainer : Yorick Laupa +-- Stability : provisional +-- Portability : non-portable +-- +-------------------------------------------------------------------------------- +module Database.EventStore.Internal.Reader (readerThread) where + +-------------------------------------------------------------------------------- +import Prelude hiding (take) +import Control.Concurrent.STM +import qualified Data.ByteString as B +import System.IO +import Text.Printf + +-------------------------------------------------------------------------------- +import Data.Serialize.Get +import Data.UUID + +-------------------------------------------------------------------------------- +import Database.EventStore.Internal.Types + +-------------------------------------------------------------------------------- +readerThread :: TChan Msg -> Handle -> IO () +readerThread chan h = loop + 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 + +-------------------------------------------------------------------------------- +-- Parsers +-------------------------------------------------------------------------------- +getLengthPrefix :: Get Int +getLengthPrefix = fmap fromIntegral getWord32le + +-------------------------------------------------------------------------------- +getPackage :: Get Package +getPackage = do + cmd <- getCmd + flg <- getFlag + col <- getUUID + rest <- remaining + dta <- getBytes rest + + let pack = Package + { packageCmd = cmd + , packageFlag = flg + , packageCorrelation = col + , packageData = dta + } + + 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 +getFlag = do + wd <- getWord8 + case wd of + 0x00 -> return None + 0x01 -> return Authenticated + _ -> fail $ printf "TCP: Unhandled flag value 0x%x" wd + +-------------------------------------------------------------------------------- +getUUID :: Get UUID +getUUID = do + bs <- getLazyByteString 16 + case fromByteString bs of + Just uuid -> return uuid + _ -> fail "TCP: Wrong UUID format" diff --git a/Database/EventStore/Internal/Types.hs b/Database/EventStore/Internal/Types.hs new file mode 100644 --- /dev/null +++ b/Database/EventStore/Internal/Types.hs @@ -0,0 +1,941 @@ +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE DataKinds #-} +{-# OPTIONS_GHC -fcontext-stack=26 #-} +-------------------------------------------------------------------------------- +-- | +-- Module : Database.EventStore.Internal.Types +-- Copyright : (C) 2014 Yorick Laupa +-- License : (see the file LICENSE) +-- +-- Maintainer : Yorick Laupa +-- Stability : provisional +-- Portability : non-portable +-- +-------------------------------------------------------------------------------- +module Database.EventStore.Internal.Types where + +-------------------------------------------------------------------------------- +import Control.Applicative ((<|>)) +import Control.Exception +import Data.ByteString (ByteString) +import Data.ByteString.Lazy (fromStrict, toStrict) +import Data.Int +import Data.Maybe +import Data.Typeable +import Data.Word +import GHC.Generics (Generic) + +-------------------------------------------------------------------------------- +import Control.Concurrent.Async hiding (link) +import qualified Data.Aeson as A +import Data.ProtocolBuffers +import Data.Text (Text) +import Data.Time +import Data.Time.Clock.POSIX +import Data.UUID (UUID, fromByteString, toByteString) +import System.Random + +-------------------------------------------------------------------------------- +-- Exceptions +-------------------------------------------------------------------------------- +data InternalException + = ConnectionClosedByServer + | Stopped + deriving (Show, Typeable) + +-------------------------------------------------------------------------------- +instance Exception InternalException + +-------------------------------------------------------------------------------- +data OperationException + = WrongExpectedVersion Text ExpectedVersion -- ^ Stream and Expected Version + | StreamDeleted Text -- ^ Stream + | InvalidTransaction + | AccessDenied Text -- ^ Stream + | InvalidServerResponse Command Command -- ^ Expected, Found + | ProtobufDecodingError String + | ServerError (Maybe Text) + deriving (Show, Typeable) + +-------------------------------------------------------------------------------- +instance Exception OperationException + +-------------------------------------------------------------------------------- +type OperationExceptional a = Either OperationException a + +-------------------------------------------------------------------------------- +-- Event +-------------------------------------------------------------------------------- +data Event + = Event + { eventType :: !Text + , eventData :: !EventData + } + +-------------------------------------------------------------------------------- +createEvent :: Text -> EventData -> Event +createEvent = Event + +-------------------------------------------------------------------------------- +data EventData + = Json A.Value (Maybe A.Value) + +-------------------------------------------------------------------------------- +eventDataType :: EventData -> Int32 +eventDataType (Json _ _) = 1 + +-------------------------------------------------------------------------------- +eventMetadataType :: EventData -> Int32 +eventMetadataType _ = 0 + +-------------------------------------------------------------------------------- +withJson :: A.Value -> EventData +withJson value = Json value Nothing + +-------------------------------------------------------------------------------- +withJsonAndMetadata :: A.Value -> A.Value -> EventData +withJsonAndMetadata value metadata = Json value (Just metadata) + +-------------------------------------------------------------------------------- +eventDataBytes :: EventData -> ByteString +eventDataBytes (Json value _) = toStrict $ A.encode value + +-------------------------------------------------------------------------------- +eventMetadataBytes :: EventData -> Maybe ByteString +eventMetadataBytes (Json _ meta_m) = fmap (toStrict . A.encode) meta_m + +-------------------------------------------------------------------------------- +-- Expected Version +-------------------------------------------------------------------------------- +data ExpectedVersion + = Any -- ^ Says that you should not conflict with anything + | NoStream -- ^ Stream should not exist when doing your write + | EmptyStream -- ^ Stream should exist but be empty when doing the write + deriving Show + +-------------------------------------------------------------------------------- +expVersionInt32 :: ExpectedVersion -> Int32 +expVersionInt32 Any = -2 +expVersionInt32 NoStream = -1 +expVersionInt32 EmptyStream = 0 + +-------------------------------------------------------------------------------- +-- EventStore Messages +-------------------------------------------------------------------------------- +data Operation + = Operation + { operationCreatePackage :: UUID -> IO Package + , operationInspect :: Package -> IO Decision + } + +-------------------------------------------------------------------------------- +data OpResult + = OP_SUCCESS + | OP_PREPARE_TIMEOUT + | OP_COMMIT_TIMEOUT + | OP_FORWARD_TIMEOUT + | OP_WRONG_EXPECTED_VERSION + | OP_STREAM_DELETED + | OP_INVALID_TRANSACTION + | OP_ACCESS_DENIED + deriving (Eq, Enum, Show) + +-------------------------------------------------------------------------------- +data NewEvent + = NewEvent + { newEventId :: Required 1 (Value ByteString) + , newEventType :: Required 2 (Value Text) + , newEventDataType :: Required 3 (Value Int32) + , newEventMetadataType :: Required 4 (Value Int32) + , newEventData :: Required 5 (Value ByteString) + , newEventMetadata :: Optional 6 (Value ByteString) + } + deriving (Generic, Show) + +-------------------------------------------------------------------------------- +instance Encode NewEvent + +-------------------------------------------------------------------------------- +newEvent :: Text -- ^ Event type + -> Int32 -- ^ Data content type + -> Int32 -- ^ Metadata content type + -> ByteString -- ^ Event data + -> Maybe ByteString -- ^ Metadata + -> IO NewEvent +newEvent evt_type data_type meta_type evt_data evt_meta = do + new_uuid <- randomIO + let uuid_bytes = toStrict $ toByteString new_uuid + new_evt = NewEvent + { newEventId = putField uuid_bytes + , newEventType = putField evt_type + , newEventDataType = putField data_type + , newEventMetadataType = putField meta_type + , newEventData = putField evt_data + , newEventMetadata = putField evt_meta + } + + 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) + , transactionStartExpectedVersion :: Required 2 (Value Int32) + , transactionStartRequireMaster :: Required 3 (Value Bool) + } + deriving (Generic, Show) + +-------------------------------------------------------------------------------- +newTransactionStart :: Text + -> Int32 + -> Bool + -> TransactionStart +newTransactionStart stream_id exp_ver req_master = + TransactionStart + { transactionStartStreamId = putField stream_id + , transactionStartExpectedVersion = putField exp_ver + , transactionStartRequireMaster = putField req_master + } + +-------------------------------------------------------------------------------- +instance Encode TransactionStart + +-------------------------------------------------------------------------------- +data TransactionStartCompleted + = TransactionStartCompleted + { transactionSCId :: Required 1 (Value Int64) + , transactionSCResult :: Required 2 (Enumeration OpResult) + , transactionSCMessage :: Optional 3 (Value Text) + } + deriving (Generic, Show) + +-------------------------------------------------------------------------------- +instance Decode TransactionStartCompleted + +-------------------------------------------------------------------------------- +data TransactionWrite + = TransactionWrite + { transactionWriteId :: Required 1 (Value Int64) + , transactionWriteEvents :: Repeated 2 (Message NewEvent) + , transactionWriteRequireMaster :: Required 3 (Value Bool) + } + deriving (Generic, Show) + +-------------------------------------------------------------------------------- +instance Encode TransactionWrite + +-------------------------------------------------------------------------------- +newTransactionWrite :: Int64 -> [NewEvent] -> Bool -> TransactionWrite +newTransactionWrite trans_id evts req_master = + TransactionWrite + { transactionWriteId = putField trans_id + , transactionWriteEvents = putField evts + , transactionWriteRequireMaster = putField req_master + } + +-------------------------------------------------------------------------------- +data TransactionWriteCompleted + = TransactionWriteCompleted + { transactionWCId :: Required 1 (Value Int64) + , transactionWCResult :: Required 2 (Enumeration OpResult) + , transactionWCMessage :: Optional 3 (Value Text) + } + deriving (Generic, Show) + +-------------------------------------------------------------------------------- +instance Decode TransactionWriteCompleted + +-------------------------------------------------------------------------------- +data TransactionCommit + = TransactionCommit + { transactionCommitId :: Required 1 (Value Int64) + , transactionCommitRequireMaster :: Required 2 (Value Bool) + } + deriving (Generic, Show) + +-------------------------------------------------------------------------------- +instance Encode TransactionCommit + +-------------------------------------------------------------------------------- +newTransactionCommit :: Int64 -> Bool -> TransactionCommit +newTransactionCommit trans_id req_master = + TransactionCommit + { transactionCommitId = putField trans_id + , transactionCommitRequireMaster = putField req_master + } + +-------------------------------------------------------------------------------- +data TransactionCommitCompleted + = TransactionCommitCompleted + { transactionCCId :: Required 1 (Value Int64) + , transactionCCResult :: Required 2 (Enumeration OpResult) + , transactionCCMessage :: Optional 3 (Value Text) + , transactionCCFirstNumber :: Required 4 (Value Int32) + , transactionCCLastNumber :: Required 5 (Value Int32) + , transactionCCPreparePosition :: Optional 6 (Value Int64) + , transactionCCCommitPosition :: Optional 7 (Value Int64) + } + deriving (Generic, Show) + +-------------------------------------------------------------------------------- +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) + , eventRecordNumber :: Required 2 (Value Int32) + , eventRecordId :: Required 3 (Value ByteString) + , eventRecordType :: Required 4 (Value Text) + , eventRecordDataType :: Required 5 (Value Int32) + , eventRecordMetadataType :: Required 6 (Value Int32) + , eventRecordData :: Required 7 (Value ByteString) + , eventRecordMetadata :: Optional 8 (Value ByteString) + , eventRecordCreated :: Optional 9 (Value Int64) + , eventRecordCreatedEpoch :: Optional 10 (Value Int64) + } + deriving (Generic, Show) + +-------------------------------------------------------------------------------- +instance Decode EventRecord + +-------------------------------------------------------------------------------- +data ResolvedIndexedEvent + = ResolvedIndexedEvent + { resolvedIndexedRecord :: Optional 1 (Message EventRecord) + , resolvedIndexedLink :: Optional 2 (Message EventRecord) + } + deriving (Generic, Show) + +-------------------------------------------------------------------------------- +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) + , resolvedEventBufLink :: Optional 2 (Message EventRecord) + , resolvedEventBufCommitPosition :: Required 3 (Value Int64) + , resolvedEventBufPreparePosition :: Required 4 (Value Int64) + } + deriving (Generic, Show) + +-------------------------------------------------------------------------------- +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 + = DoNothing + | EndOperation + | Retry + | Reconnection + | Subscribed + +-------------------------------------------------------------------------------- +data Position + = Position + { positionCommit :: !Int64 + , positionPrepare :: !Int64 + } + deriving Show + +-------------------------------------------------------------------------------- +data WriteResult + = WriteResult + { writeNextExpectedVersion :: !Int32 + , writePosition :: !Position + } + deriving Show + +-------------------------------------------------------------------------------- +newtype DeleteResult + = DeleteResult { deleteStreamPosition :: Position } + deriving Show + +-------------------------------------------------------------------------------- +data RecordedEvent + = RecordedEvent + { recordedEventStreamId :: !Text + , recordedEventId :: !UUID + , recordedEventNumber :: !Int32 + , recordedEventType :: !Text + , recordedEventData :: !ByteString + , recordedEventMetadata :: !(Maybe ByteString) + , recordedEventIsJson :: !Bool + , recordedEventCreated :: !(Maybe UTCTime) + , recordedEventCreatedEpoch :: !(Maybe Integer) + } + deriving Show + +-------------------------------------------------------------------------------- +newRecordedEvent :: EventRecord -> RecordedEvent +newRecordedEvent er = re + where + evt_id = getField $ eventRecordId er + evt_uuid = fromJust $ fromByteString $ fromStrict evt_id + data_type = getField $ eventRecordDataType er + created = getField $ eventRecordCreated er + epoch = getField $ eventRecordCreatedEpoch er + utc_created = fmap (posixSecondsToUTCTime . fromInteger . toInteger) created + + re = RecordedEvent + { recordedEventStreamId = getField $ eventRecordStreamId er + , recordedEventNumber = getField $ eventRecordNumber er + , recordedEventId = evt_uuid + , recordedEventType = getField $ eventRecordType er + , recordedEventData = getField $ eventRecordData er + , recordedEventMetadata = getField $ eventRecordMetadata er + , recordedEventIsJson = data_type == 1 + , recordedEventCreated = utc_created + , recordedEventCreatedEpoch = fmap toInteger epoch + } + +-------------------------------------------------------------------------------- +data ResolvedEvent + = ResolvedEvent + { resolvedEventRecord :: !(Maybe RecordedEvent) + , resolvedEventLink :: !(Maybe RecordedEvent) + } + deriving Show + +-------------------------------------------------------------------------------- +newResolvedEvent :: ResolvedIndexedEvent -> ResolvedEvent +newResolvedEvent rie = re + where + record = getField $ resolvedIndexedRecord rie + link = getField $ resolvedIndexedLink rie + re = ResolvedEvent + { resolvedEventRecord = fmap newRecordedEvent record + , resolvedEventLink = fmap newRecordedEvent link + } + +-------------------------------------------------------------------------------- +newResolvedEventFromBuf :: ResolvedEventBuf -> ResolvedEvent +newResolvedEventFromBuf reb = re + where + record = Just $ newRecordedEvent $ getField $ resolvedEventBufEvent reb + link = getField $ resolvedEventBufLink reb + re = ResolvedEvent + { resolvedEventRecord = record + , resolvedEventLink = fmap newRecordedEvent link + } + +-------------------------------------------------------------------------------- +resolvedEventOriginal :: ResolvedEvent -> Maybe RecordedEvent +resolvedEventOriginal (ResolvedEvent record link) = + link <|> record + +-------------------------------------------------------------------------------- +eventResolved :: ResolvedEvent -> Bool +eventResolved = isJust . resolvedEventOriginal + +-------------------------------------------------------------------------------- +resolvedEventOriginalStreamId :: ResolvedEvent -> Maybe Text +resolvedEventOriginalStreamId = + 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 + = Transaction + { transactionId :: Int64 + , transactionStreamId :: Text + , transactionExpectedVersion :: ExpectedVersion + , transactionCommit :: IO (Async WriteResult) + , transactionSendEvents :: [Event] -> IO (Async ()) + , transactionRollback :: IO () + } + +-------------------------------------------------------------------------------- +-- Flag +-------------------------------------------------------------------------------- +data Flag + = None + | Authenticated + deriving Show + +-------------------------------------------------------------------------------- +flagWord8 :: Flag -> Word8 +flagWord8 None = 0x00 +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 + +-------------------------------------------------------------------------------- +-------------------------------------------------------------------------------- +data Package + = Package + { packageCmd :: !Command + , packageFlag :: !Flag + , packageCorrelation :: !UUID + , packageData :: !ByteString + } + deriving Show + +-------------------------------------------------------------------------------- +data Msg + = Reconnect + | RecvPackage Package + | RegisterOperation Operation + | Notice String + | Tick + +-------------------------------------------------------------------------------- +-- Settings +-------------------------------------------------------------------------------- +-- | Global @ConnectionManager@ settings +data Settings + = Settings + { _heartbeatInterval :: NominalDiffTime + , _heartbeatTimeout :: NominalDiffTime + , _requireMaster :: Bool + } + +-------------------------------------------------------------------------------- +defaultSettings :: Settings +defaultSettings = Settings + { _heartbeatInterval = msDiffTime 750 -- 750ms + , _heartbeatTimeout = msDiffTime 1500 -- 1500ms + , _requireMaster = True + } + +-------------------------------------------------------------------------------- +-- | Millisecond timespan +msDiffTime :: Float -> NominalDiffTime +msDiffTime i = fromRational $ toRational (i / 1000) diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,30 @@ +Copyright (c) 2014, Yorick Laupa + +All rights reserved. + +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions are met: + + * Redistributions of source code must retain the above copyright + notice, this list of conditions and the following disclaimer. + + * Redistributions in binary form must reproduce the above + copyright notice, this list of conditions and the following + disclaimer in the documentation and/or other materials provided + with the distribution. + + * Neither the name of Yorick Laupa nor the names of other + contributors may be used to endorse or promote products derived + from this software without specific prior written permission. + +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -0,0 +1,26 @@ +EventStore Haskell TCP client +============================= + +Basically, all we have for now are: + + 1. NewEvent + 2. DeleteStream + 3. Transaction + 4. ReadEvent + 5. ReadStreamEvents (Forward and Backward) + 6. ReadAllEvents (Forward and Backward) + +TODO +==== + + 1. All kind of Subscriptions + 2. Authentication + 3. SSL + +Requirements +============ + 1. GHC >= 7.8.3 + 2. Cabal >= 1.20 + 3. EventStore >= 3.0.0 + +(Don't know if it works on Windows) diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/eventstore.cabal b/eventstore.cabal new file mode 100644 --- /dev/null +++ b/eventstore.cabal @@ -0,0 +1,95 @@ +-- Initial eventstore.cabal generated by cabal init. For further +-- documentation, see http://haskell.org/cabal/users-guide/ + +-- The name of the package. +name: eventstore + +-- The package version. See the Haskell package versioning policy (PVP) +-- for standards guiding when and how versions should be incremented. +-- http://www.haskell.org/haskellwiki/Package_versioning_policy +-- PVP summary: +-+------- breaking API changes +-- | | +----- non-breaking API additions +-- | | | +--- code changes with no API change +version: 0.1.0.0 + +-- A short (one-line) description of the package. +synopsis: EventStore Haskell TCP Client + +-- A longer description of the package. +description: EventStore Haskell TCP Client + +-- The license under which the package is released. +license: BSD3 + +-- The file containing the license text. +license-file: LICENSE + +-- The package author(s). +author: Yorick Laupa + +-- An email address to which users can send suggestions, bug reports, and +-- patches. +maintainer: yo.eight@gmail.com + +-- A copyright notice. +-- copyright: + +category: Database + +build-type: Simple + +-- Extra files to be distributed with the package, such as examples or a +-- README. +extra-source-files: README.md + +-- Constraint on the version of Cabal needed to build this package. +cabal-version: >=1.10 + +source-repository head + type: git + location: git://github.com/YoEight/eventstore.git + +library + -- Modules exported by the library. + exposed-modules: Database.EventStore + + -- Modules included in this library but not exported. + other-modules: Database.EventStore.Internal.Packages + Database.EventStore.Internal.Processor + Database.EventStore.Internal.Reader + Database.EventStore.Internal.Types + Database.EventStore.Internal.Operation.Common + 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 + + -- LANGUAGE extensions used by modules in this package. + -- other-extensions: + + -- Other library packages from which modules are imported. + build-depends: base >=4.7 && <4.8 + , 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.* + , 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. + -- hs-source-dirs: + + -- Base language which the package is written in. + default-language: Haskell2010