eventstore (empty) → 0.1.0.0
raw patch · 16 files changed
+2793/−0 lines, 16 filesdep +aesondep +asyncdep +attoparsecsetup-changed
Dependencies added: aeson, async, attoparsec, base, binary, bytestring, cereal, containers, network, protobuf, random, stm, suspend, text, time, timers, uuid
Files
- Database/EventStore.hs +290/−0
- Database/EventStore/Internal/Operation/Common.hs +80/−0
- Database/EventStore/Internal/Operation/DeleteStreamOperation.hs +94/−0
- Database/EventStore/Internal/Operation/ReadAllEventsOperation.hs +94/−0
- Database/EventStore/Internal/Operation/ReadEventOperation.hs +87/−0
- Database/EventStore/Internal/Operation/ReadStreamEventsOperation.hs +98/−0
- Database/EventStore/Internal/Operation/TransactionStartOperation.hs +289/−0
- Database/EventStore/Internal/Operation/WriteEventsOperation.hs +114/−0
- Database/EventStore/Internal/Packages.hs +105/−0
- Database/EventStore/Internal/Processor.hs +351/−0
- Database/EventStore/Internal/Reader.hs +97/−0
- Database/EventStore/Internal/Types.hs +941/−0
- LICENSE +30/−0
- README.md +26/−0
- Setup.hs +2/−0
- eventstore.cabal +95/−0
+ Database/EventStore.hs view
@@ -0,0 +1,290 @@+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore+-- Copyright : (C) 2014 Yorick Laupa+-- License : (see the file LICENSE)+--+-- Maintainer : Yorick Laupa <yo.eight@gmail.com>+-- Stability : provisional+-- Portability : non-portable+--+--------------------------------------------------------------------------------+module Database.EventStore+ ( 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)
+ Database/EventStore/Internal/Operation/Common.hs view
@@ -0,0 +1,80 @@+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Internal.Operation.Common+-- Copyright : (C) 2014 Yorick Laupa+-- License : (see the file LICENSE)+--+-- Maintainer : Yorick Laupa <yo.eight@gmail.com>+-- Stability : provisional+-- Portability : non-portable+--+--------------------------------------------------------------------------------+module Database.EventStore.Internal.Operation.Common+ ( OperationParams(..)+ , createOperation+ -- * Re-exports+ , module Data.ProtocolBuffers+ ) where++--------------------------------------------------------------------------------+import Data.ProtocolBuffers+import Data.Serialize+import Data.UUID++--------------------------------------------------------------------------------+import Database.EventStore.Internal.Types++--------------------------------------------------------------------------------+data OperationParams request response+ = OperationParams+ { opSettings :: !Settings+ , opRequestCmd :: !Command+ , opResponseCmd :: !Command++ , opRequest :: IO request+ , opSuccess :: response -> IO Decision+ , opFailure :: OperationException -> IO Decision+ }++--------------------------------------------------------------------------------+createOperation :: (Encode a, Decode b) => OperationParams a b -> Operation+createOperation params =+ Operation+ { operationCreatePackage = createPackage params+ , operationInspect = inspection params+ }++--------------------------------------------------------------------------------+createPackage :: Encode a => OperationParams a b -> UUID -> IO Package+createPackage params uuid = do+ req <- opRequest params++ let pack = Package+ { packageCmd = opRequestCmd params+ , packageCorrelation = uuid+ , packageFlag = None+ , packageData = runPut $ encodeMessage req+ }++ return pack++--------------------------------------------------------------------------------+inspection :: Decode b => OperationParams a b -> Package -> IO Decision+inspection params pack+ | found == exp_v = deeperInspection params pack+ | otherwise = failed (InvalidServerResponse exp_v found)+ where+ exp_v = opResponseCmd params+ failed = opFailure params+ found = packageCmd pack++--------------------------------------------------------------------------------+deeperInspection :: Decode b => OperationParams a b -> Package -> IO Decision+deeperInspection params pack =+ case runGet decodeMessage bytes of+ Left e -> failed (ProtobufDecodingError e)+ Right msg -> succeed msg+ where+ failed = opFailure params+ succeed = opSuccess params+ bytes = packageData pack
+ Database/EventStore/Internal/Operation/DeleteStreamOperation.hs view
@@ -0,0 +1,94 @@+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Internal.Operation.DeleteStreamOperation+-- Copyright : (C) 2014 Yorick Laupa+-- License : (see the file LICENSE)+--+-- Maintainer : Yorick Laupa <yo.eight@gmail.com>+-- Stability : provisional+-- Portability : non-portable+--+--------------------------------------------------------------------------------+module Database.EventStore.Internal.Operation.DeleteStreamOperation+ ( deleteStreamOperation ) where++--------------------------------------------------------------------------------+import Control.Concurrent.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
+ Database/EventStore/Internal/Operation/ReadAllEventsOperation.hs view
@@ -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 <yo.eight@gmail.com>+-- 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
+ Database/EventStore/Internal/Operation/ReadEventOperation.hs view
@@ -0,0 +1,87 @@+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Internal.Operation.ReadEventOperation+-- Copyright : (C) 2014 Yorick Laupa+-- License : (see the file LICENSE)+--+-- Maintainer : Yorick Laupa <yo.eight@gmail.com>+-- Stability : provisional+-- Portability : non-portable+--+--------------------------------------------------------------------------------+module Database.EventStore.Internal.Operation.ReadEventOperation+ ( 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
+ Database/EventStore/Internal/Operation/ReadStreamEventsOperation.hs view
@@ -0,0 +1,98 @@+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Internal.Operation.ReadStreamEventsOperation+-- Copyright : (C) 2014 Yorick Laupa+-- License : (see the file LICENSE)+--+-- Maintainer : Yorick Laupa <yo.eight@gmail.com>+-- Stability : provisional+-- Portability : non-portable+--+--------------------------------------------------------------------------------+module Database.EventStore.Internal.Operation.ReadStreamEventsOperation+ ( 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
+ Database/EventStore/Internal/Operation/TransactionStartOperation.hs view
@@ -0,0 +1,289 @@+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Internal.Operation.TransactionStartOperation+-- Copyright : (C) 2014 Yorick Laupa+-- License : (see the file LICENSE)+--+-- Maintainer : Yorick Laupa <yo.eight@gmail.com>+-- Stability : provisional+-- Portability : non-portable+--+--------------------------------------------------------------------------------+module Database.EventStore.Internal.Operation.TransactionStartOperation+ ( transactionStartOperation ) where++--------------------------------------------------------------------------------+import Control.Concurrent.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)
+ Database/EventStore/Internal/Operation/WriteEventsOperation.hs view
@@ -0,0 +1,114 @@+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Internal.Operation.WriteEventsOperation+-- Copyright : (C) 2014 Yorick Laupa+-- License : (see the file LICENSE)+--+-- Maintainer : Yorick Laupa <yo.eight@gmail.com>+-- Stability : provisional+-- Portability : non-portable+--+--------------------------------------------------------------------------------+module Database.EventStore.Internal.Operation.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
+ Database/EventStore/Internal/Packages.hs view
@@ -0,0 +1,105 @@+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Internal.Packages+-- Copyright : (C) 2014 Yorick Laupa+-- License : (see the file LICENSE)+--+-- Maintainer : Yorick Laupa <yo.eight@gmail.com>+-- Stability : provisional+-- Portability : non-portable+--+--------------------------------------------------------------------------------+module Database.EventStore.Internal.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
+ Database/EventStore/Internal/Processor.hs view
@@ -0,0 +1,351 @@+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Internal.Processor+-- Copyright : (C) 2014 Yorick Laupa+-- License : (see the file LICENSE)+--+-- Maintainer : Yorick Laupa <yo.eight@gmail.com>+-- Stability : provisional+-- Portability : non-portable+--+--------------------------------------------------------------------------------+module Database.EventStore.Internal.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
+ Database/EventStore/Internal/Reader.hs view
@@ -0,0 +1,97 @@+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Internal.Reader+-- Copyright : (C) 2014 Yorick Laupa+-- License : (see the file LICENSE)+--+-- Maintainer : Yorick Laupa <yo.eight@gmail.com>+-- Stability : provisional+-- Portability : non-portable+--+--------------------------------------------------------------------------------+module Database.EventStore.Internal.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"
+ Database/EventStore/Internal/Types.hs view
@@ -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 <yo.eight@gmail.com>+-- 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)
+ LICENSE view
@@ -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.
+ README.md view
@@ -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)
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ eventstore.cabal view
@@ -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