eventstore 0.13.1.3 → 0.13.1.4
raw patch · 37 files changed
+334/−380 lines, 37 filesdep +classy-preludedep −asyncdep −bytestringdep ~aesondep ~textPVP: major bump suggested
API removals or changes: PVP suggests a major version bump
Dependencies added: classy-prelude
Dependencies removed: async, bytestring
Dependency ranges changed: aeson, text
API changes (from Hackage documentation)
- Database.EventStore: infixr 6 <>
+ Database.EventStore: data Command
+ Database.EventStore: waitAsync :: MonadIO m => Async a -> m a
- Database.EventStore: (<>) :: Monoid m => m -> m -> m
+ Database.EventStore: (<>) :: Semigroup a => a -> a -> a
- Database.EventStore: InvalidServerResponse :: Word8 -> Word8 -> OperationError
+ Database.EventStore: InvalidServerResponse :: Command -> Command -> OperationError
- Database.EventStore.Logging: PackageReceived :: Word8 -> UUID -> InfoMessage
+ Database.EventStore.Logging: PackageReceived :: Command -> UUID -> InfoMessage
- Database.EventStore.Logging: PackageSent :: Word8 -> UUID -> InfoMessage
+ Database.EventStore.Logging: PackageSent :: Command -> UUID -> InfoMessage
Files
- CHANGELOG.markdown +5/−0
- Database/EventStore.hs +7/−13
- Database/EventStore/Internal/Command.hs +33/−0
- Database/EventStore/Internal/Connection.hs +113/−108
- Database/EventStore/Internal/Discovery.hs +1/−25
- Database/EventStore/Internal/Execution/Production.hs +5/−11
- Database/EventStore/Internal/Generator.hs +1/−0
- Database/EventStore/Internal/Manager/Operation/Model.hs +24/−26
- Database/EventStore/Internal/Manager/Subscription.hs +7/−7
- Database/EventStore/Internal/Manager/Subscription/Driver.hs +13/−17
- Database/EventStore/Internal/Manager/Subscription/Message.hs +1/−3
- Database/EventStore/Internal/Manager/Subscription/Model.hs +16/−17
- Database/EventStore/Internal/Manager/Subscription/Packages.hs +1/−5
- Database/EventStore/Internal/Operation.hs +6/−15
- Database/EventStore/Internal/Operation/Catchup.hs +1/−2
- Database/EventStore/Internal/Operation/DeleteStream.hs +1/−1
- Database/EventStore/Internal/Operation/DeleteStream/Message.hs +1/−2
- Database/EventStore/Internal/Operation/Read/Common.hs +2/−8
- Database/EventStore/Internal/Operation/ReadAllEvents.hs +1/−0
- Database/EventStore/Internal/Operation/ReadAllEvents/Message.hs +1/−2
- Database/EventStore/Internal/Operation/ReadEvent.hs +1/−1
- Database/EventStore/Internal/Operation/ReadEvent/Message.hs +1/−2
- Database/EventStore/Internal/Operation/ReadStreamEvents.hs +1/−1
- Database/EventStore/Internal/Operation/ReadStreamEvents/Message.hs +1/−2
- Database/EventStore/Internal/Operation/StreamMetadata.hs +1/−3
- Database/EventStore/Internal/Operation/Transaction.hs +1/−5
- Database/EventStore/Internal/Operation/Transaction/Message.hs +1/−2
- Database/EventStore/Internal/Operation/Write/Common.hs +3/−0
- Database/EventStore/Internal/Operation/WriteEvents.hs +1/−5
- Database/EventStore/Internal/Operation/WriteEvents/Message.hs +1/−2
- Database/EventStore/Internal/Processor.hs +6/−6
- Database/EventStore/Internal/Stream.hs +1/−1
- Database/EventStore/Internal/Types.hs +25/−36
- Database/EventStore/Logging.hs +8/−12
- README.md +3/−2
- eventstore.cabal +8/−7
- tests/Tests.hs +31/−31
CHANGELOG.markdown view
@@ -1,3 +1,8 @@+0.13.1.4+--------+* Bump `aeson` version.+* Internal connection refactoring.+ 0.13.1.3 -------- * Bump `http-client` version.
Database/EventStore.hs view
@@ -170,6 +170,7 @@ , positionStart , positionEnd -- * Misc+ , Command , DropReason(..) , ExpectedVersion , anyVersion@@ -178,7 +179,7 @@ , exactEventVersion , streamExists -- * Re-export- , module Control.Concurrent.Async+ , waitAsync , (<>) , NonEmpty(..) , nonEmpty@@ -186,24 +187,17 @@ ) where ---------------------------------------------------------------------------------import Control.Concurrent-import Control.Concurrent.STM-import Control.Exception-import Control.Monad (when)-import Data.ByteString (ByteString) import Data.Int import Data.Maybe-import Data.Monoid ((<>))-import Data.Typeable-import Network.Connection (TLSSettings) ---------------------------------------------------------------------------------import Control.Concurrent.Async+import ClassyPrelude hiding (Builder, group) import Data.List.NonEmpty(NonEmpty(..), nonEmpty)-import Data.Text hiding (group) import Data.UUID+import Network.Connection (TLSSettings) --------------------------------------------------------------------------------+import Database.EventStore.Internal.Command import Database.EventStore.Internal.Connection import Database.EventStore.Internal.Discovery import qualified Database.EventStore.Internal.Manager.Subscription as S@@ -303,7 +297,7 @@ waitTillCatchup :: Subscription S.Catchup -> IO () waitTillCatchup sub = atomically $ do caughtUp <- _hasCaughtUp sub- when (not caughtUp) retry+ when (not caughtUp) retrySTM -------------------------------------------------------------------------------- _hasCaughtUp :: Subscription S.Catchup -> STM Bool@@ -381,7 +375,7 @@ nextEvent sub = atomically $ do m <- _nextEventMaybe sub case m of- Nothing -> retry+ Nothing -> retrySTM Just e -> return e --------------------------------------------------------------------------------
+ Database/EventStore/Internal/Command.hs view
@@ -0,0 +1,33 @@+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Internal.Command+-- Copyright : (C) 2016 Yorick Laupa+-- License : (see the file LICENSE)+--+-- Maintainer : Yorick Laupa <yo.eight@gmail.com>+-- Stability : provisional+-- Portability : non-portable+--+--------------------------------------------------------------------------------+module Database.EventStore.Internal.Command (Command(..)) where++--------------------------------------------------------------------------------+import Data.Word+import Numeric++--------------------------------------------------------------------------------+import ClassyPrelude++--------------------------------------------------------------------------------+-- | Internal command representation.+newtype Command = Command { cmdWord8 :: Word8 } deriving (Eq, Ord, Num)++--------------------------------------------------------------------------------+padding :: String -> String+padding [x] = ['0',x]+padding xs = xs++--------------------------------------------------------------------------------+instance Show Command where+ show (Command w) = "0x" ++ padding (showHex w "")
Database/EventStore/Internal/Connection.hs view
@@ -25,22 +25,17 @@ ) where ---------------------------------------------------------------------------------import Control.Concurrent-import Control.Concurrent.STM-import Control.Exception-import qualified Data.ByteString as B-import Data.Foldable (for_)-import Data.IORef-import Data.Typeable-import Text.Printf+import Text.Printf --------------------------------------------------------------------------------+import ClassyPrelude import Data.Serialize import Data.UUID import Data.UUID.V4 import Network.Connection --------------------------------------------------------------------------------+import Database.EventStore.Internal.Command import Database.EventStore.Internal.Discovery import Database.EventStore.Internal.Types import Database.EventStore.Logging@@ -70,6 +65,14 @@ Recv :: In Package --------------------------------------------------------------------------------+-- | Represents connection logic action to carry out.+data Status a where+ Noop :: Status ()+ WithConnection :: UUID -> Connection -> In a -> Status a+ CreateConnection :: In a -> Status a+ Errored :: ConnectionException -> Status a++-------------------------------------------------------------------------------- -- | Internal representation of a connection with the server. data InternalConnection = InternalConnection@@ -128,113 +131,115 @@ _ -> return False --------------------------------------------------------------------------------+-- Connection Logic+--------------------------------------------------------------------------------+onlineLogic :: forall a. TMVar State+ -> UUID+ -> Connection+ -> In a+ -> STM (Status a)+onlineLogic var uuid conn input =+ let status = WithConnection uuid conn input+ state =+ case input of+ Close -> Closed+ _ -> Online uuid conn in+ status <$ putTMVar var state++--------------------------------------------------------------------------------+offlineLogic :: forall a. TMVar State -> In a -> STM (Status a)+offlineLogic var Close = Noop <$ putTMVar var Closed+offlineLogic _ other = return $ CreateConnection other++--------------------------------------------------------------------------------+closedLogic :: forall a. TMVar State -> In a -> STM (Status a)+closedLogic var input = do+ putTMVar var Closed+ case input of+ Close -> return Noop+ _ -> return $ Errored ClosedConnection++--------------------------------------------------------------------------------+connectionLogic :: forall a. TMVar State -> In a -> STM (Status a)+connectionLogic var input = do+ state <- takeTMVar var+ case state of+ Online uuid conn -> onlineLogic var uuid conn input+ Offline -> offlineLogic var input+ Closed -> closedLogic var input++--------------------------------------------------------------------------------+--------------------------------------------------------------------------------+handleInput :: forall a. UUID -> Connection -> In a -> IO a+handleInput _ conn (Send pkg) = send conn pkg+handleInput _ conn Recv = recv conn+handleInput uuid _ Id = return uuid+handleInput _ conn Close = liftIO $ connectionClose conn++-------------------------------------------------------------------------------- -- | Main connection logic. It will automatically reconnect to the server when -- a exception occured while the 'Handle' is accessed. execute :: forall a. InternalConnection -> In a -> IO a-execute InternalConnection{..} i = do- res <- atomically $ do- s <- takeTMVar _var- case s of- Offline -> return $ Right Nothing- Online u con -> return $ Right $ Just (u, con)- Closed -> return $ Left ClosedConnection- case i of- Close ->- case res of- Left _ -> atomically $ putTMVar _var Closed- Right Nothing -> atomically $ putTMVar _var Closed- Right (Just (_, con)) -> do- connectionClose con- atomically $ putTMVar _var Closed- other ->- case res of- Left e -> do- atomically $ putTMVar _var Closed- throwIO e- Right alt -> do- sres <- case alt of- Nothing -> newState _setts _ctx _last _disc- Just (u, h) -> return $ Right $ Online u h- case sres of- Left e -> do- atomically $ putTMVar _var Closed- throwIO e- Right s -> do- atomically $ putTMVar _var s- let Online u con = s- case other of- Id -> return u- Send pkg -> send con pkg- Recv -> recv con- Close -> error "impossible execute"+execute iconn input = do+ res <- atomically $ connectionLogic (_var iconn) input + case res of+ Noop -> return ()+ Errored e -> throwIO e+ WithConnection uuid conn op -> handleInput uuid conn op+ CreateConnection op -> do+ (uuid, conn) <- openConnection iconn+ atomically $ putTMVar (_var iconn) (Online uuid conn)+ handleInput uuid conn op+ ---------------------------------------------------------------------------------newState :: Settings- -> ConnectionContext- -> IORef (Maybe EndPoint)- -> Discovery- -> IO (Either ConnectionException State)-newState sett ctx ref disc =- case s_retry sett of- AtMost n ->- let loop i = do- _settingsLog sett (Info $ Connecting i)- let action = do- old <- readIORef ref- ept_opt <- runDiscovery disc old- case ept_opt of- Nothing -> do- threadDelay delay- if n <= i- then return $- Left MaxAttemptConnectionReached- else loop (i + 1)- Just ept -> do- let host = endPointIp ept- port = endPointPort ept- st <- connect sett ctx host port- writeIORef ref (Just ept)- return $ Right st- catch action $ \(_ :: SomeException) -> do- threadDelay delay- if n <= i- then return $- Left MaxAttemptConnectionReached- else loop (i + 1) in- loop 1- KeepRetrying ->- let endlessly i = do- _settingsLog sett (Info $ Connecting i)- let action = do- old <- readIORef ref- ept_opt <- runDiscovery disc old- case ept_opt of- Nothing -> threadDelay delay- >> endlessly (i + 1)- Just ept -> do- let host = endPointIp ept- port = endPointPort ept- st <- connect sett ctx host port- writeIORef ref (Just ept)- return $ Right st- catch action $ \(_ :: SomeException) ->- threadDelay delay >> endlessly (i + 1) in- endlessly (1 :: Int)+reachedMaxAttempt :: Retry -> Int -> Bool+reachedMaxAttempt KeepRetrying _ = False+reachedMaxAttempt (AtMost n) cur = n <= cur++--------------------------------------------------------------------------------+openConnection :: InternalConnection -> IO (UUID, Connection)+openConnection InternalConnection{..} = attempt 1 where- delay = s_reconnect_delay_secs sett * secs+ delay = s_reconnect_delay_secs _setts * secs + handleFailure trialCount = do+ threadDelay delay+ when (reachedMaxAttempt (s_retry _setts) trialCount) $ do+ atomically $ putTMVar _var Closed+ throwIO MaxAttemptConnectionReached+ attempt (trialCount + 1)++ attempt trialCount = do+ _settingsLog _setts (Info $ Connecting trialCount)+ old <- readIORef _last+ ept_opt <- runDiscovery _disc old+ case ept_opt of+ Nothing -> handleFailure trialCount+ Just ept -> do+ let host = endPointIp ept+ port = endPointPort ept+ res <- tryAny $ connect _setts _ctx host port+ case res of+ Left _ -> handleFailure trialCount+ Right st -> st <$ writeIORef _last (Just ept)+ -------------------------------------------------------------------------------- secs :: Int secs = 1000000 ---------------------------------------------------------------------------------connect :: Settings -> ConnectionContext -> String -> Int -> IO State+connect :: Settings+ -> ConnectionContext+ -> String+ -> Int+ -> IO (UUID, Connection) connect sett ctx host port = do let params = ConnectionParams host (fromIntegral port) (s_ssl sett) Nothing conn <- connectTo ctx params uuid <- nextRandom _settingsLog sett (Info $ Connected uuid)- return $ Online uuid conn+ return (uuid, conn) -------------------------------------------------------------------------------- -- Binary operations@@ -261,28 +266,28 @@ -------------------------------------------------------------------------------- -- | Serializes a 'Package' into raw bytes. putPackage :: Package -> Put-putPackage pack = do+putPackage pkg = do putWord32le length_prefix- putWord8 (packageCmd pack)+ putWord8 (cmdWord8 $ packageCmd pkg) putWord8 flag_word8 putLazyByteString corr_bytes for_ cred_m $ \(Credentials login passw) -> do- putWord8 $ fromIntegral $ B.length login+ putWord8 $ fromIntegral $ olength login putByteString login- putWord8 $ fromIntegral $ B.length passw+ putWord8 $ fromIntegral $ olength passw putByteString passw putByteString pack_data where- pack_data = packageData pack+ pack_data = packageData pkg cred_len = maybe 0 credSize cred_m- length_prefix = fromIntegral (B.length pack_data + mandatorySize + cred_len)- cred_m = packageCred pack+ length_prefix = fromIntegral (olength pack_data + mandatorySize + cred_len)+ cred_m = packageCred pkg flag_word8 = maybe 0x00 (const 0x01) cred_m- corr_bytes = toByteString $ packageCorrelation pack+ corr_bytes = toByteString $ packageCorrelation pkg -------------------------------------------------------------------------------- credSize :: Credentials -> Int-credSize (Credentials login passw) = B.length login + B.length passw + 2+credSize (Credentials login passw) = olength login + olength passw + 2 -------------------------------------------------------------------------------- -- | The minimun size a 'Package' should have. It's basically a command byte,@@ -307,7 +312,7 @@ dta <- getBytes rest let pkg = Package- { packageCmd = cmd+ { packageCmd = Command cmd , packageCorrelation = col , packageData = dta , packageCred = cred
Database/EventStore/Internal/Discovery.hs view
@@ -34,25 +34,14 @@ ) where ---------------------------------------------------------------------------------import Control.Applicative-import Control.Exception-import Control.Monad-import Data.Foldable (toList)-import Data.IORef-import Data.List (sortBy) import Data.Maybe-import Data.Ord-import Data.Typeable-import GHC.Generics hiding (from, to)-import Prelude --------------------------------------------------------------------------------+import ClassyPrelude import Data.Aeson import Data.Aeson.Types import Data.Array.IO-import Data.ByteString (ByteString) import Data.DotNet.TimeSpan-import Data.Int import Data.List.NonEmpty (NonEmpty) import Data.UUID import Network.HTTP.Client@@ -501,16 +490,3 @@ | otherwise = do res <- k cur if isJust res then return res else loop len (cur + 1)------------------------------------------------------------------------------------- | Taken from base >= 4.8 because prior base don't have it.--- Sort a list by comparing the results of a key function applied to each--- element. @sortOn f@ is equivalent to @sortBy . comparing f@, but has the--- performance advantage of only evaluating @f@ once for each element in the--- input list. This is called the decorate-sort-undecorate paradigm, or--- Schwartzian transform.------ @since 4.8.0.0-sortOn :: Ord b => (a -> b) -> [a] -> [a]-sortOn f =- map snd . sortBy (comparing fst) . map (\x -> let y = f x in y `seq` (y, x))
Database/EventStore/Internal/Execution/Production.hs view
@@ -39,18 +39,12 @@ ) where ---------------------------------------------------------------------------------import Prelude hiding (take)-import Control.Concurrent-import Control.Concurrent.STM-import Control.Exception-import Control.Monad+import Control.Exception (AsyncException(..), asyncExceptionFromException) import Control.Monad.Fix-import Data.IORef import Data.Int-import Data.Foldable ---------------------------------------------------------------------------------import Data.Text+import ClassyPrelude import Data.UUID --------------------------------------------------------------------------------@@ -64,7 +58,7 @@ , nakPersist , abort )-import Database.EventStore.Internal.Operation hiding (retry)+import Database.EventStore.Internal.Operation import Database.EventStore.Internal.Processor import Database.EventStore.Internal.Types import Database.EventStore.Logging@@ -514,7 +508,7 @@ -- Waits the runner thread to deal with its jobs list. atomically $ do end <- isEmptyCycleQueue _jobQueue- unless end retry+ unless end retrySTM traverse_ killThread rutid @@ -554,5 +548,5 @@ _ <- forkFinally (bootstrap env) handler return $ Prod nxt_sub $ do closed <- connIsClosed conn- unless closed retry+ unless closed retrySTM readTMVar disposed
Database/EventStore/Internal/Generator.hs view
@@ -18,6 +18,7 @@ ) where --------------------------------------------------------------------------------+import ClassyPrelude import Data.UUID import System.Random
Database/EventStore/Internal/Manager/Operation/Model.hs view
@@ -22,15 +22,13 @@ ) where ---------------------------------------------------------------------------------import Data.Word-----------------------------------------------------------------------------------import qualified Data.HashMap.Strict as H-import Data.ProtocolBuffers-import Data.Serialize-import Data.UUID+import ClassyPrelude+import Data.ProtocolBuffers+import Data.Serialize+import Data.UUID --------------------------------------------------------------------------------+import Database.EventStore.Internal.Command import Database.EventStore.Internal.Generator import Database.EventStore.Internal.Operation import Database.EventStore.Internal.Types@@ -41,7 +39,7 @@ forall a resp. Decode resp => Elem { _opOp :: Operation a- , _opCmd :: Word8+ , _opCmd :: Command , _opCont :: resp -> SM a () , _opCb :: Either OperationError a -> r }@@ -52,13 +50,13 @@ State { _gen :: Generator -- ^ 'UUID' generator.- , _pending :: H.HashMap UUID (Elem r)+ , _pending :: HashMap UUID (Elem r) -- ^ Contains all running 'Operation's. } -------------------------------------------------------------------------------- initState :: Generator -> State r-initState g = State g H.empty+initState g = State g mempty -------------------------------------------------------------------------------- -- | Type of requests handled by the model.@@ -114,7 +112,7 @@ -> Transition r runOperation setts cb op start init_st = go init_st start where- go st (Return _) = Await $ Model $ handle setts st+ go st (Return _) = Await $ Model $ execute setts st go st (Yield a n) = Produce (cb $ Right a) (go st n) go st (FreshId k) = let (new_id, nxt_gen) = nextUUID $ _gen st@@ -129,50 +127,50 @@ , packageCred = s_credentials setts } elm = Elem op co k cb- ps = H.insert new_uuid elm $ _pending st+ ps = insertMap new_uuid elm $ _pending st nxt_st = st { _pending = ps , _gen = nxt_gen } in- Transmit pkg (Await $ Model $ handle setts nxt_st)+ Transmit pkg (Await $ Model $ execute setts nxt_st) go st (Failure m) = case m of- Just e -> Produce (cb $ Left e) (Await $ Model $ handle setts st)+ Just e -> Produce (cb $ Left e) (Await $ Model $ execute setts st) _ -> runOperation setts cb op op st -------------------------------------------------------------------------------- runPackage :: Settings -> State r -> Package -> Maybe (Transition r) runPackage setts st Package{..} = do- Elem op resp_cmd cont cb <- H.lookup packageCorrelation $ _pending st- let nxt_ps = H.delete packageCorrelation $ _pending st+ Elem op resp_cmd cont cb <- lookup packageCorrelation $ _pending st+ let nxt_ps = deleteMap packageCorrelation $ _pending st nxt_st = st { _pending = nxt_ps } if resp_cmd /= packageCmd then let r = cb $ Left $ InvalidServerResponse resp_cmd packageCmd in- return $ Produce r (Await $ Model $ handle setts nxt_st)+ return $ Produce r (Await $ Model $ execute setts nxt_st) else case runGet decodeMessage packageData of Left e -> let r = cb $ Left $ ProtobufDecodingError e in- return $ Produce r (Await $ Model $ handle setts nxt_st)+ return $ Produce r (Await $ Model $ execute setts nxt_st) Right m -> return $ runOperation setts cb op (cont m) nxt_st -------------------------------------------------------------------------------- abortOperations :: Settings -> State r -> Transition r-abortOperations setts init_st = go init_st $ H.toList $ _pending init_st+abortOperations setts init_st = go init_st $ mapToList $ _pending init_st where go st ((key, Elem _ _ _ k):xs) =- let ps = H.delete key $ _pending st+ let ps = deleteMap key $ _pending st nxt_st = st { _pending = ps } in Produce (k $ Left Aborted) $ go nxt_st xs- go st [] = Await $ Model $ handle setts st+ go st [] = Await $ Model $ execute setts st -------------------------------------------------------------------------------- -- | Creates a new 'Operation' model state-machine. newModel :: Settings -> Generator -> Model r-newModel setts g = Model $ handle setts $ initState g+newModel setts g = Model $ execute setts $ initState g ---------------------------------------------------------------------------------handle :: Settings -> State r -> Request r -> Maybe (Transition r)-handle setts st (New op cb) = Just $ runOperation setts cb op op st-handle setts st (Pkg pkg) = runPackage setts st pkg-handle setts st Abort = Just $ abortOperations setts st+execute :: Settings -> State r -> Request r -> Maybe (Transition r)+execute setts st (New op cb) = Just $ runOperation setts cb op op st+execute setts st (Pkg pkg) = runPackage setts st pkg+execute setts st Abort = Just $ abortOperations setts st
Database/EventStore/Internal/Manager/Subscription.hs view
@@ -38,8 +38,8 @@ import Data.Int ---------------------------------------------------------------------------------import Data.Sequence-import Data.Text (Text)+import ClassyPrelude+import Data.Sequence (ViewL(..), viewl, dropWhileL) -------------------------------------------------------------------------------- import Database.EventStore.Internal.Manager.Subscription.Driver@@ -167,13 +167,13 @@ -> Input Catchup a -> a catchingUp b s (Arrived e) =- Subscription $ catchingUp b (s |> e)+ Subscription $ catchingUp b (s `snoc` e) catchingUp b s ReadNext = case viewl b of EmptyL -> (Nothing, Subscription $ catchingUp b s) e :< rest -> (Just e, Subscription $ catchingUp rest s) catchingUp b s (BatchRead es eos nxt_pt) =- let nxt_b = foldl (|>) b es+ let nxt_b = foldl' snoc b es nxt_s = dropWhileL (beforeChk nxt_pt) s nxt = if eos then Subscription $ caughtUp nxt_b nxt_s@@ -185,7 +185,7 @@ -> Seq ResolvedEvent -> Input Catchup a -> a- caughtUp b s (Arrived e) = Subscription $ caughtUp b (s |> e)+ caughtUp b s (Arrived e) = Subscription $ caughtUp b (s `snoc` e) caughtUp b s ReadNext = case viewl b of EmptyL -> live s ReadNext@@ -197,7 +197,7 @@ caughtUp _ _ CaughtUp = False live :: forall a. Seq ResolvedEvent -> Input Catchup a -> a- live s (Arrived e) = Subscription $ live (s |> e)+ live s (Arrived e) = Subscription $ live (s `snoc` e) live s ReadNext = case viewl s of EmptyL -> (Nothing, Subscription $ live s)@@ -211,7 +211,7 @@ baseSubscription = Subscription $ go empty where go :: forall a. Seq ResolvedEvent -> Input t a -> a- go s (Arrived e) = Subscription $ go (s |> e)+ go s (Arrived e) = Subscription $ go (s `snoc` e) go s ReadNext = case viewl s of EmptyL -> (Nothing, Subscription $ go s)
Database/EventStore/Internal/Manager/Subscription/Driver.hs view
@@ -37,18 +37,14 @@ ) where ---------------------------------------------------------------------------------import Control.Exception import Data.Int import Data.Maybe-import Data.Typeable ---------------------------------------------------------------------------------import Data.ByteString-import qualified Data.HashMap.Strict as H-import Data.Serialize-import Data.ProtocolBuffers-import Data.Text-import Data.UUID+import ClassyPrelude+import Data.Serialize+import Data.ProtocolBuffers+import Data.UUID -------------------------------------------------------------------------------- import Database.EventStore.Internal.Generator@@ -321,14 +317,14 @@ -- ^ Subscription model. , _gen :: !Generator -- ^ 'UUID' generator.- , _reg :: !(H.HashMap UUID (Cmd r))+ , _reg :: !(HashMap UUID (Cmd r)) -- ^ Holds ongoing commands. When stored, it means an action hasn't been -- confirmed yet. } -------------------------------------------------------------------------------- initState :: Generator -> State r-initState gen = State newModel gen H.empty+initState gen = State newModel gen mempty -------------------------------------------------------------------------------- -- | Subscription driver state machine.@@ -341,7 +337,7 @@ where go :: forall a. State r -> In r a -> a go st@State{..} (Pkg Package{..}) = do- elm <- H.lookup packageCorrelation _reg+ elm <- lookup packageCorrelation _reg case packageCmd of 0xC2 -> do _ <- querySubscription packageCorrelation _model@@ -398,7 +394,7 @@ nxt_m = unsubscribed run _model dreason = toSubDropReason reason evt = Dropped dreason- nxt_reg = H.delete packageCorrelation _reg+ nxt_reg = deleteMap packageCorrelation _reg nxt_st = st { _model = nxt_m , _reg = nxt_reg } case elm of@@ -417,7 +413,7 @@ msg <- maybeDecodeMessage packageData _ <- queryPersistentAction packageCorrelation _model let nxt_m = confirmedAction packageCorrelation _model- nxt_rg = H.delete packageCorrelation _reg+ nxt_rg = deleteMap packageCorrelation _reg nxt_st = st { _model = nxt_m , _reg = nxt_rg }@@ -435,7 +431,7 @@ nxt_m = connectReg s tos u _model nxt_st = st { _model = nxt_m , _gen = nxt_g- , _reg = H.insert u cmd _reg } in+ , _reg = insertMap u cmd _reg } in (pkg, Driver $ go nxt_st) ConnectPersist _ gn n b -> let (u, nxt_g) = nextUUID _gen@@ -443,7 +439,7 @@ nxt_m = connectPersist gn n b u _model nxt_st = st { _model = nxt_m , _gen = nxt_g- , _reg = H.insert u cmd _reg } in+ , _reg = insertMap u cmd _reg } in (pkg, Driver $ go nxt_st) Unsubscribe r -> let pkg = createUnsubscribePackage setts $ runningUUID r in@@ -454,7 +450,7 @@ nxt_m = persistAction gn n u a _model nxt_st = st { _model = nxt_m , _gen = nxt_g- , _reg = H.insert u cmd _reg } in+ , _reg = insertMap u cmd _reg } in (pkg, Driver $ go nxt_st) PersistAck _ run evts -> let RunningPersist _ _ _ _ sid _ _ = run@@ -466,7 +462,7 @@ u = runningUUID run pkg = createNakPackage setts u sid na r evts in (pkg, Driver $ go st)- go st Abort = (H.elems $ _reg st) >>= _F+ go st Abort = (fmap snd $ mapToList $ _reg st) >>= _F where _F (ConnectReg k _ _) = [k $ Dropped SubAborted] _F (ConnectPersist k _ _ _) = [k $ Dropped SubAborted]
Database/EventStore/Internal/Manager/Subscription/Message.hs view
@@ -15,14 +15,12 @@ module Database.EventStore.Internal.Manager.Subscription.Message where ---------------------------------------------------------------------------------import Data.ByteString (ByteString) import Data.Int-import GHC.Generics (Generic) --------------------------------------------------------------------------------+import ClassyPrelude hiding (group) import Data.DotNet.TimeSpan import Data.ProtocolBuffers-import Data.Text (Text) -------------------------------------------------------------------------------- import Database.EventStore.Internal.Types
Database/EventStore/Internal/Manager/Subscription/Model.hs view
@@ -37,9 +37,8 @@ import Data.Int ---------------------------------------------------------------------------------import qualified Data.HashMap.Strict as H-import Data.Text-import Data.UUID+import ClassyPrelude+import Data.UUID -------------------------------------------------------------------------------- import Database.EventStore.Internal.Types@@ -61,7 +60,7 @@ } ---------------------------------------------------------------------------------type Register a = H.HashMap UUID a+type Register a = HashMap UUID a -------------------------------------------------------------------------------- -- | Represents a 'Subscription' which is about to be confirmed.@@ -247,7 +246,7 @@ -------------------------------------------------------------------------------- emptyState :: State-emptyState = State H.empty H.empty H.empty+emptyState = State mempty mempty mempty -------------------------------------------------------------------------------- -- | Subscription operations state machine. Keeps every information related to@@ -268,12 +267,12 @@ case c of ConnectReg n tos -> let p = PendingReg n tos- nxt_ps = H.insert u p $ _stPending s+ nxt_ps = insertMap u p $ _stPending s nxt_s = s { _stPending = nxt_ps } in Model $ modelHandle nxt_s ConnectPersist g n b -> let p = PendingPersist g n b- nxt_ps = H.insert u p $ _stPending s+ nxt_ps = insertMap u p $ _stPending s nxt_s = s { _stPending = nxt_ps } in Model $ modelHandle nxt_s Confirmed c ->@@ -281,38 +280,38 @@ ConfirmedConnection u tpe -> case tpe of RegularMeta lc le ->- case H.lookup u $ _stPending s of+ case lookup u $ _stPending s of Just (PendingReg n tos) -> let r = RunningReg u n tos lc le- nxt_rs = H.insert u r $ _stRunning s+ nxt_rs = insertMap u r $ _stRunning s nxt_s = s { _stRunning = nxt_rs } in Model $ modelHandle nxt_s _ -> Model $ modelHandle s PersistMeta sb lc le ->- case H.lookup u $ _stPending s of+ case lookup u $ _stPending s of Just (PendingPersist g n b) -> let r = RunningPersist u g n b sb lc le- nxt_rs = H.insert u r $ _stRunning s+ nxt_rs = insertMap u r $ _stRunning s nxt_s = s { _stRunning = nxt_rs } in Model $ modelHandle nxt_s _ -> Model $ modelHandle s ConfirmedPersistAction u ->- case H.lookup u $ _stAction s of+ case lookup u $ _stAction s of Just (PendingAction{}) ->- let nxt_as = H.delete u $ _stAction s+ let nxt_as = deleteMap u $ _stAction s nxt_s = s { _stAction = nxt_as } in Model $ modelHandle nxt_s _ -> Model $ modelHandle s Unsubscribed u ->- let nxt_ps = H.delete u $ _stRunning s+ let nxt_ps = deleteMap u $ _stRunning s nxt_s = s { _stRunning = nxt_ps } in Model $ modelHandle nxt_s PersistAction g n u t -> let a = PendingAction g n t- nxt_as = H.insert u a $ _stAction s+ nxt_as = insertMap u a $ _stAction s nxt_s = s { _stAction = nxt_as } in Model $ modelHandle nxt_s modelHandle s (Query q) = case q of- QuerySub u -> H.lookup u $ _stRunning s- QueryAction u -> H.lookup u $ _stAction s+ QuerySub u -> lookup u $ _stRunning s+ QueryAction u -> lookup u $ _stAction s
Database/EventStore/Internal/Manager/Subscription/Packages.hs view
@@ -16,19 +16,15 @@ import Data.Int ---------------------------------------------------------------------------------import Data.ByteString.Lazy (toStrict)+import ClassyPrelude import Data.ProtocolBuffers import Data.Serialize-import Data.Text import Data.UUID -------------------------------------------------------------------------------- import Database.EventStore.Internal.Manager.Subscription.Message import Database.EventStore.Internal.Manager.Subscription.Model import Database.EventStore.Internal.Types-----------------------------------------------------------------------------------import Prelude -------------------------------------------------------------------------------- -- | Creates a regular subscription connection 'Package'.
Database/EventStore/Internal/Operation.hs view
@@ -17,25 +17,16 @@ module Database.EventStore.Internal.Operation where ---------------------------------------------------------------------------------import Control.Applicative-import Control.Exception-import Control.Monad-import Data.Typeable----------------------------------------------------------------------------------+import ClassyPrelude import Data.ProtocolBuffers-import Data.Text import Data.UUID-import Data.Word --------------------------------------------------------------------------------+import Database.EventStore.Internal.Command import Database.EventStore.Internal.Stream import Database.EventStore.Internal.Types ---------------------------------------------------------------------------------import Prelude---------------------------------------------------------------------------------- -- | Operation result sent by the server. data OpResult = OP_SUCCESS@@ -55,7 +46,7 @@ | StreamDeleted Text -- ^ Stream | InvalidTransaction | AccessDenied StreamName -- ^ Stream- | InvalidServerResponse Word8 Word8 -- ^ Expected, Found+ | InvalidServerResponse Command Command -- ^ Expected, Found | ProtobufDecodingError String | ServerError (Maybe Text) -- ^ Reason | InvalidOperation Text@@ -79,7 +70,7 @@ | FreshId (UUID -> SM o a) -- ^ Asks for an unused 'UUID'. | forall rq rp. (Encode rq, Decode rp) =>- SendPkg Word8 Word8 rq (rp -> SM o a)+ SendPkg Command Command rq (rp -> SM o a) -- ^ Send a request message given a command and an expected command. -- response. It also carries a callback to call when response comes in. | Failure (Maybe OperationError)@@ -128,7 +119,7 @@ -------------------------------------------------------------------------------- -- | Sends a request to the server given a command request and response. It -- returns the expected deserialized message.-send :: (Encode rq, Decode rp) => Word8 -> Word8 -> rq -> SM o rp+send :: (Encode rq, Decode rp) => Command -> Command -> rq -> SM o rp send ci co rq = SendPkg ci co rq Return --------------------------------------------------------------------------------@@ -189,5 +180,5 @@ -------------------------------------------------------------------------------- -- | Raises 'InvalidServerResponse' exception.-invalidServerResponse :: Word8 -> Word8 -> SM o a+invalidServerResponse :: Command -> Command -> SM o a invalidServerResponse expe got = failure $ InvalidServerResponse expe got
Database/EventStore/Internal/Operation/Catchup.hs view
@@ -18,12 +18,11 @@ ) where ---------------------------------------------------------------------------------import Control.Monad import Data.Int import Data.Maybe ---------------------------------------------------------------------------------import Data.Text (Text)+import ClassyPrelude -------------------------------------------------------------------------------- import Database.EventStore.Internal.Manager.Subscription (Checkpoint(..))
Database/EventStore/Internal/Operation/DeleteStream.hs view
@@ -22,8 +22,8 @@ import Data.Maybe --------------------------------------------------------------------------------+import ClassyPrelude import Data.ProtocolBuffers-import Data.Text -------------------------------------------------------------------------------- import Database.EventStore.Internal.Operation
Database/EventStore/Internal/Operation/DeleteStream/Message.hs view
@@ -15,11 +15,10 @@ -------------------------------------------------------------------------------- import Data.Int-import GHC.Generics --------------------------------------------------------------------------------+import ClassyPrelude import Data.ProtocolBuffers-import Data.Text -------------------------------------------------------------------------------- import Database.EventStore.Internal.Operation
Database/EventStore/Internal/Operation/Read/Common.hs view
@@ -16,21 +16,15 @@ module Database.EventStore.Internal.Operation.Read.Common where ---------------------------------------------------------------------------------import Control.Applicative-import Data.Foldable+import Data.Foldable (foldMap) import Data.Int-import Data.Monoid-import Data.Traversable ---------------------------------------------------------------------------------import Data.Text+import ClassyPrelude -------------------------------------------------------------------------------- import Database.EventStore.Internal.Stream import Database.EventStore.Internal.Types-----------------------------------------------------------------------------------import Prelude -------------------------------------------------------------------------------- -- | Enumeration detailing the possible outcomes of reading a stream.
Database/EventStore/Internal/Operation/ReadAllEvents.hs view
@@ -18,6 +18,7 @@ import Data.Maybe --------------------------------------------------------------------------------+import ClassyPrelude import Data.ProtocolBuffers --------------------------------------------------------------------------------
Database/EventStore/Internal/Operation/ReadAllEvents/Message.hs view
@@ -15,11 +15,10 @@ -------------------------------------------------------------------------------- import Data.Int-import GHC.Generics --------------------------------------------------------------------------------+import ClassyPrelude import Data.ProtocolBuffers-import Data.Text -------------------------------------------------------------------------------- import Database.EventStore.Internal.Types
Database/EventStore/Internal/Operation/ReadEvent.hs view
@@ -20,8 +20,8 @@ import Data.Int --------------------------------------------------------------------------------+import ClassyPrelude import Data.ProtocolBuffers-import Data.Text -------------------------------------------------------------------------------- import Database.EventStore.Internal.Operation
Database/EventStore/Internal/Operation/ReadEvent/Message.hs view
@@ -16,11 +16,10 @@ -------------------------------------------------------------------------------- import Data.Int-import GHC.Generics --------------------------------------------------------------------------------+import ClassyPrelude import Data.ProtocolBuffers-import Data.Text -------------------------------------------------------------------------------- import Database.EventStore.Internal.Types
Database/EventStore/Internal/Operation/ReadStreamEvents.hs view
@@ -18,8 +18,8 @@ import Data.Int --------------------------------------------------------------------------------+import ClassyPrelude import Data.ProtocolBuffers-import Data.Text -------------------------------------------------------------------------------- import Database.EventStore.Internal.Operation
Database/EventStore/Internal/Operation/ReadStreamEvents/Message.hs view
@@ -15,11 +15,10 @@ -------------------------------------------------------------------------------- import Data.Int-import GHC.Generics --------------------------------------------------------------------------------+import ClassyPrelude import Data.ProtocolBuffers-import Data.Text -------------------------------------------------------------------------------- import Database.EventStore.Internal.Types
Database/EventStore/Internal/Operation/StreamMetadata.hs view
@@ -21,12 +21,10 @@ -------------------------------------------------------------------------------- import Data.Int-import Data.Monoid ((<>)) --------------------------------------------------------------------------------+import ClassyPrelude import Data.Aeson (decode)-import Data.ByteString.Lazy (fromStrict)-import Data.Text (Text) -------------------------------------------------------------------------------- import Database.EventStore.Internal.Operation
Database/EventStore/Internal/Operation/Transaction.hs view
@@ -23,11 +23,10 @@ -------------------------------------------------------------------------------- import Data.Int import Data.Maybe-import Data.Traversable --------------------------------------------------------------------------------+import ClassyPrelude import Data.ProtocolBuffers-import Data.Text -------------------------------------------------------------------------------- import Database.EventStore.Internal.Operation@@ -35,9 +34,6 @@ import Database.EventStore.Internal.Operation.Write.Common import Database.EventStore.Internal.Stream import Database.EventStore.Internal.Types-----------------------------------------------------------------------------------import Prelude -------------------------------------------------------------------------------- -- | Start transaction operation.
Database/EventStore/Internal/Operation/Transaction/Message.hs view
@@ -15,11 +15,10 @@ -------------------------------------------------------------------------------- import Data.Int-import GHC.Generics --------------------------------------------------------------------------------+import ClassyPrelude import Data.ProtocolBuffers-import Data.Text -------------------------------------------------------------------------------- import Database.EventStore.Internal.Operation
Database/EventStore/Internal/Operation/Write/Common.hs view
@@ -15,6 +15,9 @@ import Data.Int --------------------------------------------------------------------------------+import ClassyPrelude++-------------------------------------------------------------------------------- import Database.EventStore.Internal.Operation import Database.EventStore.Internal.Types
Database/EventStore/Internal/Operation/WriteEvents.hs view
@@ -15,11 +15,10 @@ -------------------------------------------------------------------------------- import Data.Maybe-import Data.Traversable --------------------------------------------------------------------------------+import ClassyPrelude import Data.ProtocolBuffers-import Data.Text -------------------------------------------------------------------------------- import Database.EventStore.Internal.Operation@@ -27,9 +26,6 @@ import Database.EventStore.Internal.Operation.WriteEvents.Message import Database.EventStore.Internal.Stream import Database.EventStore.Internal.Types-----------------------------------------------------------------------------------import Prelude -------------------------------------------------------------------------------- -- | Write events operation.
Database/EventStore/Internal/Operation/WriteEvents/Message.hs view
@@ -15,11 +15,10 @@ -------------------------------------------------------------------------------- import Data.Int-import GHC.Generics --------------------------------------------------------------------------------+import ClassyPrelude import Data.ProtocolBuffers-import Data.Text -------------------------------------------------------------------------------- import Database.EventStore.Internal.Operation
Database/EventStore/Internal/Processor.hs view
@@ -34,7 +34,7 @@ import Data.Int ---------------------------------------------------------------------------------import Data.Text+import ClassyPrelude import Data.UUID --------------------------------------------------------------------------------@@ -229,7 +229,7 @@ loopOpTransition st (Op.Transmit pkg nxt) = Transmit pkg (loopOpTransition st nxt) loopOpTransition st (Op.Await m) =- let nxt_st = st { _opModel = m } in Await $ Processor $ handle nxt_st+ let nxt_st = st { _opModel = m } in Await $ Processor $ execute nxt_st -------------------------------------------------------------------------------- abortTransition :: State r -> Op.Transition r -> [r] -> Transition r@@ -239,12 +239,12 @@ abortOp (Op.Transmit _ nxt) = abortOp nxt abortOp _ = abortSub init_rs - abortSub [] = Await $ Processor $ handle st+ abortSub [] = Await $ Processor $ execute st abortSub (r:rs) = Produce r (abortSub rs) ---------------------------------------------------------------------------------handle :: State r -> In r -> Transition r-handle = go+execute :: State r -> In r -> Transition r+execute = go where go st (Cmd tpe) = case tpe of@@ -319,4 +319,4 @@ -------------------------------------------------------------------------------- -- | Creates a new 'Processor' state-machine. newProcessor :: Settings -> Generator -> Processor r-newProcessor setts gen = Processor $ handle $ initState setts gen+newProcessor setts gen = Processor $ execute $ initState setts gen
Database/EventStore/Internal/Stream.hs view
@@ -15,7 +15,7 @@ module Database.EventStore.Internal.Stream where ---------------------------------------------------------------------------------import Data.Text+import ClassyPrelude -------------------------------------------------------------------------------- -- | A stream can either point to $all or a regular one.
Database/EventStore/Internal/Types.hs view
@@ -18,35 +18,24 @@ module Database.EventStore.Internal.Types where --------------------------------------------------------------------------------+import Data.Maybe import Data.Monoid (Endo(..))-----------------------------------------------------------------------------------import Control.Applicative-import Control.Exception-import Control.Monad (mzero)-import Data.ByteString (ByteString)-import Data.ByteString.Lazy (fromStrict, toStrict)-import Data.Int-import Data.Maybe-import qualified Data.Set as S-import Data.Typeable-import Data.Word-import Foreign.C.Types (CTime(..))-import GHC.Generics (Generic)+import Foreign.C.Types (CTime(..)) ---------------------------------------------------------------------------------import qualified Data.Aeson as A+import ClassyPrelude hiding (Builder)+import qualified Data.Aeson as A import Data.Aeson.Types (Object, ToJSON(..), Pair, Parser, (.=)) import Data.DotNet.TimeSpan-import qualified Data.HashMap.Strict as H+import Data.HashMap.Strict (filterWithKey) import Data.ProtocolBuffers-import Data.Text (Text)-import Data.Time+import Data.Time (NominalDiffTime) import Data.Time.Clock.POSIX import Data.UUID (UUID, fromByteString, toByteString) import Network.Connection (TLSSettings) --------------------------------------------------------------------------------+import Database.EventStore.Internal.Command import Database.EventStore.Logging --------------------------------------------------------------------------------@@ -297,14 +286,14 @@ -------------------------------------------------------------------------------- instance Eq Position where- Position ac ap == Position bc bp = ac == bc && ap == bp+ Position ac aap == Position bc bp = ac == bc && aap == bp -------------------------------------------------------------------------------- instance Ord Position where- compare (Position ac ap) (Position bc bp) =- if ac < bc || (ac == bc && ap < bp)+ compare (Position ac aap) (Position bc bp) =+ if ac < bc || (ac == bc && aap < bp) then LT- else if ac > bc || (ac == bc && ap > bp)+ else if ac > bc || (ac == bc && aap > bp) then GT else EQ @@ -494,7 +483,7 @@ -- | Represents a package exchanged between the client and the server. data Package = Package- { packageCmd :: !Word8+ { packageCmd :: !Command , packageCorrelation :: !UUID , packageData :: !ByteString , packageCred :: !(Maybe Credentials)@@ -644,7 +633,7 @@ -------------------------------------------------------------------------------- -- | Gets a custom property value from metadata. getCustomPropertyValue :: StreamMetadata -> Text -> Maybe A.Value-getCustomPropertyValue s k = H.lookup k obj+getCustomPropertyValue s k = lookup k obj where obj = streamMetadataCustom s @@ -675,13 +664,13 @@ , streamMetadataTruncateBefore = Nothing , streamMetadataCacheControl = Nothing , streamMetadataACL = emptyStreamACL- , streamMetadataCustom = H.empty+ , streamMetadataCustom = mempty } -------------------------------------------------------------------------------- -- | Maps an 'Object' to a list of 'Pair' to ease the 'StreamMetadata'. customMetaToPairs :: Object -> [Pair]-customMetaToPairs = fmap go . H.toList+customMetaToPairs = fmap go . mapToList where go (k,v) = k .= v @@ -769,21 +758,21 @@ -------------------------------------------------------------------------------- -- | Gathers every internal metadata properties into a 'Set'. It used to safely -- 'StreamMetadata' in JSON.-internalMetaProperties :: S.Set Text+internalMetaProperties :: Set Text internalMetaProperties =- S.fromList [ p_maxAge- , p_maxCount- , p_truncateBefore- , p_cacheControl- , p_acl- ]+ setFromList [ p_maxAge+ , p_maxCount+ , p_truncateBefore+ , p_cacheControl+ , p_acl+ ] -------------------------------------------------------------------------------- -- | Only keeps the properties the users has set. keepUserProperties :: Object -> Object-keepUserProperties = H.filterWithKey go+keepUserProperties = filterWithKey go where- go k _ = not $ S.member k internalMetaProperties+ go k _ = notMember k internalMetaProperties -------------------------------------------------------------------------------- -- | Parses a 'NominalDiffTime' from an 'Object' given a JSON property.@@ -939,7 +928,7 @@ setCustomProperty :: ToJSON a => Text -> a -> StreamMetadataBuilder setCustomProperty k v = Endo $ \s -> let m = streamMetadataCustom s- m' = H.insert k (toJSON v) m in+ m' = insertMap k (toJSON v) m in s { streamMetadataCustom = m' } --------------------------------------------------------------------------------
Database/EventStore/Logging.hs view
@@ -17,13 +17,15 @@ -------------------------------------------------------------------------------- import Control.Exception-import Data.Word-import Numeric --------------------------------------------------------------------------------+import ClassyPrelude import Data.UUID --------------------------------------------------------------------------------+import Database.EventStore.Internal.Command++-------------------------------------------------------------------------------- -- | Logging main data structure. data Log = Error ErrorMessage@@ -45,9 +47,9 @@ -- ^ Indicates connection 'UUID'. | Disconnected UUID -- ^ Indicates connection 'UUID'- | PackageSent Word8 UUID+ | PackageSent Command UUID -- ^ Indicates a package has been sent.- | PackageReceived Word8 UUID+ | PackageReceived Command UUID -- ^ Indicates the client's received a package from the server. --------------------------------------------------------------------------------@@ -61,12 +63,6 @@ show (Disconnected u) = "Disconnected [" ++ show u ++ "]" show (PackageSent cmd u) =- "Package send 0x" ++ padding (showHex cmd "") ++ " [" ++ show u ++ "]"+ "Package send " ++ show cmd ++ " [" ++ show u ++ "]" show (PackageReceived cmd u) =- "Package received 0x" ++- padding (showHex cmd "") ++ " [" ++ show u ++ "]"-----------------------------------------------------------------------------------padding :: String -> String-padding [x] = ['0',x]-padding xs = xs+ "Package received " ++ show cmd ++ " [" ++ show u ++ "]"
README.md view
@@ -8,6 +8,7 @@ Requirements ============+ * 64bits system * GHC >= 7.8.3 * Cabal >= 1.18 * EventStore >= 3.0.0 (>= 3.1.0 if you want competing consumers)@@ -53,7 +54,7 @@ -- features like its Complex Event Processing (CEP) capabality. import Database.EventStore--- Note that import also re-exports 'Control.Concurrent.Async' module, allowing the use of 'wait'+-- Note that import also re-exports 'waitAsync' -- function for instance. There are also 'NonEmpty' data constructor and 'nonEmpty' function from -- 'Data.List.NonEmpty'. @@ -73,7 +74,7 @@ -- EventStore interactions are fundamentally asynchronous. Nothing requires you to wait -- for the completion of an operation, but it's good to know if something went wrong.- _ <- wait as+ _ <- waitAsync as -- Again, if you decide to `shutdown` an EventStore connection, it means your application is -- about to terminate.
eventstore.cabal view
@@ -10,7 +10,7 @@ -- PVP summary: +-+------- breaking API changes -- | | +----- non-breaking API additions -- | | | +--- code changes with no API change-version: 0.13.1.3+version: 0.13.1.4 tested-with: GHC >= 7.8.3 && < 7.11 @@ -62,7 +62,8 @@ exposed-modules: Database.EventStore Database.EventStore.Logging -- Modules included in this library but not exported.- other-modules: Database.EventStore.Internal.Connection+ other-modules: Database.EventStore.Internal.Command+ Database.EventStore.Internal.Connection Database.EventStore.Internal.Discovery Database.EventStore.Internal.Execution.Production Database.EventStore.Internal.Generator@@ -97,16 +98,15 @@ -- LANGUAGE extensions used by modules in this package. -- other-extensions: + default-extensions: NoImplicitPrelude+ -- Other library packages from which modules are imported. build-depends: base >=4.7 && <5- , aeson >=0.8 && <0.12- , async >=2.0 && <2.2- , bytestring >=0.10.4 && <0.11+ , aeson >=0.8 && <1.1 , cereal >=0.4 && <0.6- , containers >=0.5 && <0.6+ , containers ==0.5.* , protobuf >=0.2.1.1 && <0.3 , random ==1.*- , text >=1.1.1 && <1.3 , time >=1.4 && <1.7 , uuid ==1.3.* , unordered-containers@@ -117,6 +117,7 @@ , http-client== 0.5.* , dotnet-timespan , connection ==0.2.*+ , classy-prelude ==1.* -- Directories containing source files. -- hs-source-dirs:
tests/Tests.hs view
@@ -58,7 +58,7 @@ evt = createEvent "foo" Nothing $ withJson js as <- sendEvent conn "write-event-test" anyVersion evt- _ <- wait as+ _ <- waitAsync as return () --------------------------------------------------------------------------------@@ -67,9 +67,9 @@ let js = object [ "baz" .= True ] evt = createEvent "foo" Nothing $ withJson js as <- sendEvent conn "read-event-test" anyVersion evt- _ <- wait as+ _ <- waitAsync as bs <- readEvent conn "read-event-test" 0 False- rs <- wait bs+ rs <- waitAsync bs case rs of ReadSuccess re -> case re of@@ -86,7 +86,7 @@ deleteStreamTest conn = do let js = object [ "baz" .= True ] evt = createEvent "foo" Nothing $ withJson js- _ <- sendEvent conn "delete-stream-test" anyVersion evt >>= wait+ _ <- sendEvent conn "delete-stream-test" anyVersion evt >>= waitAsync _ <- deleteStream conn "delete-stream-test" anyVersion Nothing return () @@ -95,15 +95,15 @@ transactionTest conn = do let js = object [ "baz" .= True ] evt = createEvent "foo" Nothing $ withJson js- t <- startTransaction conn "transaction-test" anyVersion >>= wait- _ <- transactionWrite t [evt] >>= wait- rs <- readEvent conn "transaction-test" 0 False >>= wait+ t <- startTransaction conn "transaction-test" anyVersion >>= waitAsync+ _ <- transactionWrite t [evt] >>= waitAsync+ rs <- readEvent conn "transaction-test" 0 False >>= waitAsync case rs of ReadNoStream -> return () e -> fail $ "transaction-test stream is supposed to not exist " ++ show e- _ <- transactionCommit t >>= wait- rs2 <- readEvent conn "transaction-test" 0 False >>= wait+ _ <- transactionCommit t >>= waitAsync+ rs2 <- readEvent conn "transaction-test" 0 False >>= waitAsync case rs2 of ReadSuccess re -> case re of@@ -123,8 +123,8 @@ , object [ "bar" .= True] ] evts = fmap (createEvent "foo" Nothing . withJson) jss- _ <- sendEvents conn "read-forward-test" anyVersion evts >>= wait- rs <- readStreamEventsForward conn "read-forward-test" 0 10 False >>= wait+ _ <- sendEvents conn "read-forward-test" anyVersion evts >>= waitAsync+ rs <- readStreamEventsForward conn "read-forward-test" 0 10 False >>= waitAsync case rs of ReadSuccess sl -> do let jss_evts = catMaybes $ fmap resolvedEventDataAsJson@@ -140,8 +140,8 @@ , object [ "bar" .= True] ] evts = fmap (createEvent "foo" Nothing . withJson) jss- _ <- sendEvents conn "read-backward-test" anyVersion evts >>= wait- rs <- readStreamEventsBackward conn "read-backward-test" 2 10 False >>= wait+ _ <- sendEvents conn "read-backward-test" anyVersion evts >>= waitAsync+ rs <- readStreamEventsBackward conn "read-backward-test" 2 10 False >>= waitAsync case rs of ReadSuccess sl -> do let jss_evts = catMaybes $ fmap resolvedEventDataAsJson@@ -152,13 +152,13 @@ -------------------------------------------------------------------------------- readAllEventsForwardTest :: Connection -> IO () readAllEventsForwardTest conn = do- sl <- readAllEventsForward conn positionStart 3 False >>= wait+ sl <- readAllEventsForward conn positionStart 3 False >>= waitAsync assertEqual "Events is not empty" False (null $ sliceEvents sl) -------------------------------------------------------------------------------- readAllEventsBackwardTest :: Connection -> IO () readAllEventsBackwardTest conn = do- sl <- readAllEventsBackward conn positionEnd 3 False >>= wait+ sl <- readAllEventsBackward conn positionEnd 3 False >>= waitAsync assertEqual "Events is not empty" False (null $ sliceEvents sl) --------------------------------------------------------------------------------@@ -171,7 +171,7 @@ evts = fmap (createEvent "foo" Nothing . withJson) jss sub <- subscribe conn "subscribe-test" False _ <- waitConfirmation sub- _ <- sendEvents conn "subscribe-test" anyVersion evts >>= wait+ _ <- sendEvents conn "subscribe-test" anyVersion evts >>= waitAsync let loop 3 = return [] loop i = do e <- nextEvent sub@@ -200,10 +200,10 @@ alljss = jss ++ jss2 evts = fmap (createEvent "foo" Nothing . withJson) jss evts2 = fmap (createEvent "foo" Nothing . withJson) jss2- _ <- sendEvents conn "subscribe-from-test" anyVersion evts >>= wait+ _ <- sendEvents conn "subscribe-from-test" anyVersion evts >>= waitAsync sub <- subscribeFrom conn "subscribe-from-test" False Nothing (Just 1) _ <- waitConfirmation sub- _ <- sendEvents conn "subscribe-from-test" anyVersion evts2 >>= wait+ _ <- sendEvents conn "subscribe-from-test" anyVersion evts2 >>= waitAsync let loop [] = do m <- nextEventMaybe sub@@ -229,15 +229,15 @@ setStreamMetadataTest :: Connection -> IO () setStreamMetadataTest conn = do let metadata = buildStreamMetadata $ setCustomProperty "foo" (1 :: Int)- _ <- setStreamMetadata conn "set-metadata-test" anyVersion metadata >>= wait+ _ <- setStreamMetadata conn "set-metadata-test" anyVersion metadata >>= waitAsync return () -------------------------------------------------------------------------------- getStreamMetadataTest :: Connection -> IO () getStreamMetadataTest conn = do let metadata = buildStreamMetadata $ setCustomProperty "foo" (1 :: Int)- _ <- setStreamMetadata conn "get-metadata-test" anyVersion metadata >>= wait- r <- getStreamMetadata conn "get-metadata-test" >>= wait+ _ <- setStreamMetadata conn "get-metadata-test" anyVersion metadata >>= waitAsync+ r <- getStreamMetadata conn "get-metadata-test" >>= waitAsync case r of StreamMetadataResult _ _ m -> case getCustomProperty m "foo" of@@ -249,7 +249,7 @@ createPersistentTest :: Connection -> IO () createPersistentTest conn = do let def = defaultPersistentSubscriptionSettings- r <- createPersistentSubscription conn "group" "create-sub" def >>= wait+ r <- createPersistentSubscription conn "group" "create-sub" def >>= waitAsync case r of Nothing -> return () Just e -> fail $ "Exception arised: " ++ show e@@ -258,8 +258,8 @@ updatePersistentTest :: Connection -> IO () updatePersistentTest conn = do let def = defaultPersistentSubscriptionSettings- _ <- createPersistentSubscription conn "group" "update-sub" def >>= wait- r <- updatePersistentSubscription conn "group" "update-sub" def >>= wait+ _ <- createPersistentSubscription conn "group" "update-sub" def >>= waitAsync+ r <- updatePersistentSubscription conn "group" "update-sub" def >>= waitAsync case r of Nothing -> return () Just e -> fail $ "Exception arised: " ++ show e@@ -268,8 +268,8 @@ deletePersistentTest :: Connection -> IO () deletePersistentTest conn = do let def = defaultPersistentSubscriptionSettings- _ <- createPersistentSubscription conn "group" "delete-sub" def >>= wait- r <- deletePersistentSubscription conn "group" "delete-sub" >>= wait+ _ <- createPersistentSubscription conn "group" "delete-sub" def >>= waitAsync+ r <- deletePersistentSubscription conn "group" "delete-sub" >>= waitAsync case r of Nothing -> return () Just e -> fail $ "Exception arised: " ++ show e@@ -284,8 +284,8 @@ , js2 ] evts = fmap (createEvent "foo" Nothing . withJson) jss- _ <- createPersistentSubscription conn "group" "connect-sub" def >>= wait- _ <- sendEvents conn "connect-sub" anyVersion evts >>= wait+ _ <- createPersistentSubscription conn "group" "connect-sub" def >>= waitAsync+ _ <- sendEvents conn "connect-sub" anyVersion evts >>= waitAsync sub <- connectToPersistentSubscription conn "group" "connect-sub" 1 _ <- waitConfirmation sub r <- nextEvent sub@@ -316,9 +316,9 @@ metadata = buildStreamMetadata $ setMaxAge timespan evt = createEvent "foo" Nothing $ withJson (object ["type" .= (3 :: Int)])- _ <- sendEvent conn "test-max-age" anyVersion evt >>= wait- _ <- setStreamMetadata conn "test-max-age" anyVersion metadata >>= wait- r <- getStreamMetadata conn "test-max-age" >>= wait+ _ <- sendEvent conn "test-max-age" anyVersion evt >>= waitAsync+ _ <- setStreamMetadata conn "test-max-age" anyVersion metadata >>= waitAsync+ r <- getStreamMetadata conn "test-max-age" >>= waitAsync case r of StreamMetadataResult _ _ m -> assertEqual "Should have equal timespan" (Just timespan)