diff --git a/Database/EventStore.hs b/Database/EventStore.hs --- a/Database/EventStore.hs +++ b/Database/EventStore.hs @@ -66,7 +66,8 @@ ) where -------------------------------------------------------------------------------- -import Control.Concurrent.STM +import Control.Concurrent +import Control.Exception import Data.Int -------------------------------------------------------------------------------- @@ -279,20 +280,20 @@ -> (Subscription -> Either DropReason ResolvedEvent -> IO ()) -> IO (Async Subscription) subscribe Connection{..} stream_id res_lnk_tos cb = do - tmp <- newEmptyTMVarIO + tmp <- newEmptyMVar processorNewSubcription conProcessor - (atomically . putTMVar tmp) + (putMVar tmp) cb stream_id res_lnk_tos - async $ atomically $ readTMVar tmp + async $ readMVar tmp -------------------------------------------------------------------------------- -createAsync :: IO (Async a, TMVar (OperationExceptional a)) +createAsync :: IO (Async a, MVar (OperationExceptional a)) createAsync = do - mvar <- newEmptyTMVarIO - as <- async $ atomically $ do - res <- readTMVar mvar - either throwSTM return res + mvar <- newEmptyMVar + as <- async $ do + res <- readMVar mvar + either throwIO return res return (as, mvar) diff --git a/Database/EventStore/Internal/Operation/DeleteStreamOperation.hs b/Database/EventStore/Internal/Operation/DeleteStreamOperation.hs --- a/Database/EventStore/Internal/Operation/DeleteStreamOperation.hs +++ b/Database/EventStore/Internal/Operation/DeleteStreamOperation.hs @@ -15,7 +15,7 @@ ( deleteStreamOperation ) where -------------------------------------------------------------------------------- -import Control.Concurrent.STM +import Control.Concurrent import Data.Int import Data.Maybe import GHC.Generics (Generic) @@ -70,7 +70,7 @@ -------------------------------------------------------------------------------- deleteStreamOperation :: Settings - -> TMVar (OperationExceptional DeleteResult) + -> MVar (OperationExceptional DeleteResult) -> Text -> ExpectedVersion -> Maybe Bool @@ -95,7 +95,7 @@ } -------------------------------------------------------------------------------- -inspect :: TMVar (OperationExceptional DeleteResult) +inspect :: MVar (OperationExceptional DeleteResult) -> Text -> ExpectedVersion -> DeleteStreamCompleted @@ -114,11 +114,11 @@ wrong_version = WrongExpectedVersion stream exp_ver -------------------------------------------------------------------------------- -succeed :: TMVar (OperationExceptional DeleteResult) +succeed :: MVar (OperationExceptional DeleteResult) -> DeleteStreamCompleted -> IO Decision succeed mvar wec = do - atomically $ putTMVar mvar (Right wr) + putMVar mvar (Right wr) return EndOperation where com_pos = getField $ deleteCompletedCommitPosition wec @@ -129,9 +129,9 @@ wr = DeleteResult pos -------------------------------------------------------------------------------- -failed :: TMVar (OperationExceptional DeleteResult) +failed :: MVar (OperationExceptional DeleteResult) -> OperationException -> IO Decision failed mvar e = do - atomically $ putTMVar mvar (Left e) + putMVar mvar (Left e) return EndOperation diff --git a/Database/EventStore/Internal/Operation/ReadAllEventsOperation.hs b/Database/EventStore/Internal/Operation/ReadAllEventsOperation.hs --- a/Database/EventStore/Internal/Operation/ReadAllEventsOperation.hs +++ b/Database/EventStore/Internal/Operation/ReadAllEventsOperation.hs @@ -19,7 +19,7 @@ ) where -------------------------------------------------------------------------------- -import Control.Concurrent.STM +import Control.Concurrent import Data.Int import Data.Maybe import GHC.Generics (Generic) @@ -123,7 +123,7 @@ -------------------------------------------------------------------------------- readAllEventsOperation :: Settings -> ReadDirection - -> TMVar (OperationExceptional AllEventsSlice) + -> MVar (OperationExceptional AllEventsSlice) -> Int64 -> Int64 -> Int32 @@ -157,7 +157,7 @@ Backward -> 0xB9 -------------------------------------------------------------------------------- -inspect :: TMVar (OperationExceptional AllEventsSlice) +inspect :: MVar (OperationExceptional AllEventsSlice) -> ReadDirection -> ReadAllEventsCompleted -> IO Decision @@ -171,20 +171,20 @@ go _ = succeed mvar dir raec -------------------------------------------------------------------------------- -succeed :: TMVar (OperationExceptional AllEventsSlice) +succeed :: MVar (OperationExceptional AllEventsSlice) -> ReadDirection -> ReadAllEventsCompleted -> IO Decision succeed mvar dir raec = do - atomically $ putTMVar mvar (Right ses) + putMVar mvar (Right ses) return EndOperation where ses = newAllEventsSlice dir raec -------------------------------------------------------------------------------- -failed :: TMVar (OperationExceptional AllEventsSlice) +failed :: MVar (OperationExceptional AllEventsSlice) -> OperationException -> IO Decision failed mvar e = do - atomically $ putTMVar mvar (Left e) + putMVar mvar (Left e) return EndOperation diff --git a/Database/EventStore/Internal/Operation/ReadEventOperation.hs b/Database/EventStore/Internal/Operation/ReadEventOperation.hs --- a/Database/EventStore/Internal/Operation/ReadEventOperation.hs +++ b/Database/EventStore/Internal/Operation/ReadEventOperation.hs @@ -19,7 +19,7 @@ ) where -------------------------------------------------------------------------------- -import Control.Concurrent.STM +import Control.Concurrent import Data.Int import GHC.Generics (Generic) @@ -108,7 +108,7 @@ -------------------------------------------------------------------------------- readEventOperation :: Settings - -> TMVar (OperationExceptional ReadResult) + -> MVar (OperationExceptional ReadResult) -> Text -> Int32 -> Bool -- ^ Resolve link TOS @@ -132,7 +132,7 @@ } -------------------------------------------------------------------------------- -inspect :: TMVar (OperationExceptional ReadResult) +inspect :: MVar (OperationExceptional ReadResult) -> Text -> Int32 -> ReadEventCompleted @@ -146,13 +146,13 @@ go _ = succeed mvar stream_id evt_num reco -------------------------------------------------------------------------------- -succeed :: TMVar (OperationExceptional ReadResult) +succeed :: MVar (OperationExceptional ReadResult) -> Text -> Int32 -> ReadEventCompleted -> IO Decision succeed mvar stream_id evt_num reco = do - atomically $ putTMVar mvar (Right rr) + putMVar mvar (Right rr) return EndOperation where status = getField $ readCompletedResult reco @@ -160,9 +160,9 @@ rr = newReadResult status stream_id evt_num rie -------------------------------------------------------------------------------- -failed :: TMVar (OperationExceptional ReadResult) +failed :: MVar (OperationExceptional ReadResult) -> OperationException -> IO Decision failed mvar e = do - atomically $ putTMVar mvar (Left e) + putMVar mvar (Left e) return EndOperation diff --git a/Database/EventStore/Internal/Operation/ReadStreamEventsOperation.hs b/Database/EventStore/Internal/Operation/ReadStreamEventsOperation.hs --- a/Database/EventStore/Internal/Operation/ReadStreamEventsOperation.hs +++ b/Database/EventStore/Internal/Operation/ReadStreamEventsOperation.hs @@ -18,7 +18,7 @@ ) where -------------------------------------------------------------------------------- -import Control.Concurrent.STM +import Control.Concurrent import Data.Int import GHC.Generics (Generic) @@ -124,7 +124,7 @@ -------------------------------------------------------------------------------- readStreamEventsOperation :: Settings -> ReadDirection - -> TMVar (OperationExceptional StreamEventsSlice) + -> MVar (OperationExceptional StreamEventsSlice) -> Text -> Int32 -> Int32 @@ -158,7 +158,7 @@ Backward -> 0xB5 -------------------------------------------------------------------------------- -inspect :: TMVar (OperationExceptional StreamEventsSlice) +inspect :: MVar (OperationExceptional StreamEventsSlice) -> ReadDirection -> Text -> Int32 @@ -173,22 +173,22 @@ go _ = succeed mvar dir stream_id start rsec -------------------------------------------------------------------------------- -succeed :: TMVar (OperationExceptional StreamEventsSlice) +succeed :: MVar (OperationExceptional StreamEventsSlice) -> ReadDirection -> Text -> Int32 -> ReadStreamEventsCompleted -> IO Decision succeed mvar dir stream_id start rsec = do - atomically $ putTMVar mvar (Right ses) + putMVar mvar (Right ses) return EndOperation where ses = newStreamEventsSlice stream_id start dir rsec -------------------------------------------------------------------------------- -failed :: TMVar (OperationExceptional StreamEventsSlice) +failed :: MVar (OperationExceptional StreamEventsSlice) -> OperationException -> IO Decision failed mvar e = do - atomically $ putTMVar mvar (Left e) + putMVar mvar (Left e) return EndOperation diff --git a/Database/EventStore/Internal/Operation/TransactionStartOperation.hs b/Database/EventStore/Internal/Operation/TransactionStartOperation.hs --- a/Database/EventStore/Internal/Operation/TransactionStartOperation.hs +++ b/Database/EventStore/Internal/Operation/TransactionStartOperation.hs @@ -14,7 +14,8 @@ ( transactionStartOperation ) where -------------------------------------------------------------------------------- -import Control.Concurrent.STM +import Control.Concurrent +import Control.Exception import Data.Int import Data.Maybe import Data.Traversable @@ -41,7 +42,7 @@ -------------------------------------------------------------------------------- transactionStartOperation :: Settings -> Processor - -> TMVar (OperationExceptional Transaction) + -> MVar (OperationExceptional Transaction) -> Text -> ExpectedVersion -> OperationParams @@ -72,7 +73,7 @@ -------------------------------------------------------------------------------- inspectTrans :: TransactionEnv - -> TMVar (OperationExceptional Transaction) + -> MVar (OperationExceptional Transaction) -> TransactionStartCompleted -> IO Decision inspectTrans env mvar tsc = go (getField $ transactionSCResult tsc) @@ -92,22 +93,22 @@ -------------------------------------------------------------------------------- succeedTrans :: TransactionEnv - -> TMVar (OperationExceptional Transaction) + -> MVar (OperationExceptional Transaction) -> TransactionStartCompleted -> IO Decision succeedTrans env mvar tsc = do - atomically $ putTMVar mvar (Right trans) + putMVar mvar (Right trans) return EndOperation where trans_id = getField $ transactionSCId tsc trans = createTransaction env trans_id -------------------------------------------------------------------------------- -failed :: TMVar (OperationExceptional a) +failed :: MVar (OperationExceptional a) -> OperationException -> IO Decision failed mvar e = do - atomically $ putTMVar mvar (Left e) + putMVar mvar (Left e) return EndOperation -------------------------------------------------------------------------------- @@ -141,7 +142,7 @@ -------------------------------------------------------------------------------- transactionWriteOperation :: TransactionEnv -> Int64 - -> TMVar (OperationExceptional ()) + -> MVar (OperationExceptional ()) -> [Event] -> OperationParams transactionWriteOperation env trans_id mvar evts = @@ -169,7 +170,7 @@ -------------------------------------------------------------------------------- transactionCommitOperation :: TransactionEnv -> Int64 - -> TMVar (OperationExceptional WriteResult) + -> MVar (OperationExceptional WriteResult) -> OperationParams transactionCommitOperation env trans_id mvar = OperationParams @@ -191,7 +192,7 @@ -------------------------------------------------------------------------------- inspectWrite :: TransactionEnv - -> TMVar (OperationExceptional ()) + -> MVar (OperationExceptional ()) -> TransactionWriteCompleted -> IO Decision inspectWrite env mvar twc = go (getField $ transactionWCResult twc) @@ -210,16 +211,16 @@ wrong_version = WrongExpectedVersion stream_id exp_ver -------------------------------------------------------------------------------- -succeedWrite :: TMVar (OperationExceptional ()) +succeedWrite :: MVar (OperationExceptional ()) -> TransactionWriteCompleted -> IO Decision succeedWrite mvar _ = do - atomically $ putTMVar mvar (Right ()) + putMVar mvar (Right ()) return EndOperation -------------------------------------------------------------------------------- inspectCommit :: TransactionEnv - -> TMVar (OperationExceptional WriteResult) + -> MVar (OperationExceptional WriteResult) -> TransactionCommitCompleted -> IO Decision inspectCommit env mvar tcc = go (getField $ transactionCCResult tcc) @@ -238,11 +239,11 @@ wrong_version = WrongExpectedVersion stream_id exp_ver -------------------------------------------------------------------------------- -succeedCommit :: TMVar (OperationExceptional WriteResult) +succeedCommit :: MVar (OperationExceptional WriteResult) -> TransactionCommitCompleted -> IO Decision succeedCommit mvar tcc = do - atomically $ putTMVar mvar (Right wr) + putMVar mvar (Right wr) return EndOperation where last_evt_num = getField $ transactionCCLastNumber tcc @@ -269,11 +270,11 @@ evt_metadata_type = eventMetadataType $ eventData evt -------------------------------------------------------------------------------- -createAsync :: IO (Async a, TMVar (OperationExceptional a)) +createAsync :: IO (Async a, MVar (OperationExceptional a)) createAsync = do - mvar <- atomically newEmptyTMVar - as <- async $ atomically $ do - res <- readTMVar mvar - either throwSTM return res + mvar <- newEmptyMVar + as <- async $ do + res <- readMVar mvar + either throwIO return res return (as, mvar) diff --git a/Database/EventStore/Internal/Operation/WriteEventsOperation.hs b/Database/EventStore/Internal/Operation/WriteEventsOperation.hs --- a/Database/EventStore/Internal/Operation/WriteEventsOperation.hs +++ b/Database/EventStore/Internal/Operation/WriteEventsOperation.hs @@ -15,7 +15,7 @@ ( writeEventsOperation ) where -------------------------------------------------------------------------------- -import Control.Concurrent.STM +import Control.Concurrent import Data.Int import Data.Maybe import Data.Traversable @@ -73,7 +73,7 @@ -------------------------------------------------------------------------------- writeEventsOperation :: Settings - -> TMVar (OperationExceptional WriteResult) + -> MVar (OperationExceptional WriteResult) -> Text -> ExpectedVersion -> [Event] @@ -100,7 +100,7 @@ } -------------------------------------------------------------------------------- -inspect :: TMVar (OperationExceptional WriteResult) +inspect :: MVar (OperationExceptional WriteResult) -> Text -> ExpectedVersion -> WriteEventsCompleted @@ -119,11 +119,11 @@ wrong_version = WrongExpectedVersion stream exp_ver -------------------------------------------------------------------------------- -succeed :: TMVar (OperationExceptional WriteResult) +succeed :: MVar (OperationExceptional WriteResult) -> WriteEventsCompleted -> IO Decision succeed mvar wec = do - atomically $ putTMVar mvar (Right wr) + putMVar mvar (Right wr) return EndOperation where last_evt_num = getField $ writeCompletedLastNumber wec @@ -135,11 +135,11 @@ wr = WriteResult last_evt_num pos -------------------------------------------------------------------------------- -failed :: TMVar (OperationExceptional WriteResult) +failed :: MVar (OperationExceptional WriteResult) -> OperationException -> IO Decision failed mvar e = do - atomically $ putTMVar mvar (Left e) + putMVar mvar (Left e) return EndOperation -------------------------------------------------------------------------------- diff --git a/Database/EventStore/Internal/Processor.hs b/Database/EventStore/Internal/Processor.hs --- a/Database/EventStore/Internal/Processor.hs +++ b/Database/EventStore/Internal/Processor.hs @@ -22,7 +22,6 @@ -------------------------------------------------------------------------------- import Control.Concurrent -import Control.Concurrent.STM import Control.Exception import Data.Monoid ((<>)) import Data.Typeable @@ -233,8 +232,8 @@ hSetBuffering hdl NoBuffering uuid <- randomIO - chan <- newTChanIO - as_rl <- async $ sync $ listen evt_pkg (atomically . writeTChan chan) + chan <- newChan + as_rl <- async $ sync $ listen evt_pkg (writeChan chan) rid <- forkFinally (readerThread push_pkg hdl) (recovering push_reco) wid <- forkFinally (writerThread chan hdl) (recovering push_reco) diff --git a/Database/EventStore/Internal/Writer.hs b/Database/EventStore/Internal/Writer.hs --- a/Database/EventStore/Internal/Writer.hs +++ b/Database/EventStore/Internal/Writer.hs @@ -12,8 +12,8 @@ module Database.EventStore.Internal.Writer (writerThread) where -------------------------------------------------------------------------------- +import Control.Concurrent import Control.Monad -import Control.Concurrent.STM import qualified Data.ByteString as B import System.IO @@ -25,8 +25,8 @@ import Database.EventStore.Internal.Types -------------------------------------------------------------------------------- -writerThread :: TChan Package -> Handle -> IO () +writerThread :: Chan Package -> Handle -> IO () writerThread chan hdl = forever $ do - pkg <- atomically $ readTChan chan + pkg <- readChan chan B.hPut hdl (runPut $ putPackage pkg) hFlush hdl diff --git a/eventstore.cabal b/eventstore.cabal --- a/eventstore.cabal +++ b/eventstore.cabal @@ -10,7 +10,7 @@ -- PVP summary: +-+------- breaking API changes -- | | +----- non-breaking API additions -- | | | +--- code changes with no API change -version: 0.2.0.0 +version: 0.2.0.1 -- A short (one-line) description of the package. synopsis: EventStore Haskell TCP Client @@ -74,7 +74,6 @@ -- 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 , bytestring >=0.10.4 && <0.10.5 @@ -84,7 +83,6 @@ , protobuf >=0.2 && <0.3 , random ==1.* , sodium ==0.11.* - , stm >=2.2 && <2.5 , text >=1.1.1 && <1.2 , time >=1.4 && <1.6 , uuid ==1.3.*