eventstore 0.7.1.0 → 0.7.2.0
raw patch · 4 files changed
+468/−15 lines, 4 filesdep +attoparsecdep +stmdep +unordered-containers
Dependencies added: attoparsec, stm, unordered-containers
Files
- CHANGELOG.markdown +5/−0
- Database/EventStore.hs +104/−0
- Database/EventStore/Internal/Types.hs +355/−14
- eventstore.cabal +4/−1
CHANGELOG.markdown view
@@ -1,3 +1,8 @@+0.7.2.0+-------+* Add `setStreamMetadata`+* Add `getStreamMetadata`+ 0.7.1.0 ------- * Internal connection changes
Database/EventStore.hs view
@@ -31,15 +31,61 @@ , connect , shutdown -- * Read Operations+ , StreamMetadataResult(..) , readEvent , readAllEventsBackward , readAllEventsForward , readStreamEventsBackward , readStreamEventsForward+ , getStreamMetadata -- * Write Operations+ , StreamACL(..)+ , StreamMetadata(..)+ , emptyStreamACL+ , emptyStreamMetadata , deleteStream , sendEvent , sendEvents+ , setStreamMetadata+ -- * Builder+ , Builder+ -- * Stream ACL Builder+ , StreamACLBuilder+ , buildStreamACL+ , modifyStreamACL+ , setReadRoles+ , setReadRole+ , setWriteRoles+ , setWriteRole+ , setDeleteRoles+ , setDeleteRole+ , setMetaReadRoles+ , setMetaReadRole+ , setMetaWriteRoles+ , setMetaWriteRole+ -- * Stream Metadata Builder+ , StreamMetadataBuilder+ , buildStreamMetadata+ , modifyStreamMetadata+ , setMaxCount+ , setMaxAge+ , setTruncateBefore+ , setCacheControl+ , setACL+ , modifyACL+ , setCustomProperty+ -- * TimeSpan+ , TimeSpan+ , timeSpanTicks+ , timeSpanHoursMinsSecs+ , timeSpanDaysHoursMinsSecs+ , timeSpanDaysHoursMinsSecsMillis+ , timeSpanGetTicks+ , timeSpanGetDays+ , timeSpanGetHours+ , timeSpanGetMinutes+ , timeSpanGetSeconds+ , timeSpanGetMillis -- * Transaction , Transaction , transactionStart@@ -95,20 +141,26 @@ , exactStream -- * Re-export , module Control.Concurrent.Async+ , (<>) ) where -------------------------------------------------------------------------------- import Control.Concurrent+import Control.Concurrent.STM (atomically) import Control.Exception+import Data.ByteString.Lazy (fromStrict) import Data.Int+import Data.Monoid ((<>)) -------------------------------------------------------------------------------- import Control.Concurrent.Async+import Data.Aeson (decode) import Data.Text -------------------------------------------------------------------------------- import Database.EventStore.Catchup import Database.EventStore.Internal.Processor+import Database.EventStore.Internal.TimeSpan import Database.EventStore.Internal.Types import Database.EventStore.Internal.Operation.DeleteStreamOperation import Database.EventStore.Internal.Operation.ReadAllEventsOperation@@ -377,6 +429,54 @@ get_sub = subscribeToAll conn res_lnk_tos --------------------------------------------------------------------------------+-- | Asynchronously sets the metadata for a stream.+setStreamMetadata :: Connection+ -> Text+ -> ExpectedVersion+ -> StreamMetadata+ -> IO (Async WriteResult)+setStreamMetadata conn evt_stream exp_ver metadata =+ let dat = withJson $ streamMetadataJSON metadata+ evt = createEvent "$metadata" Nothing dat in+ sendEvent conn (metaStreamOf evt_stream) exp_ver evt++--------------------------------------------------------------------------------+getStreamMetadata :: Connection -> Text -> IO (Async StreamMetadataResult)+getStreamMetadata conn evt_stream = do+ as <- readEvent conn (metaStreamOf evt_stream) (-1) False+ async $ atomically $ waitSTM as >>= extractStreamMetadataResult evt_stream++--------------------------------------------------------------------------------+extractStreamMetadataResult :: Monad m+ => Text+ -> ReadResult+ -> m StreamMetadataResult+extractStreamMetadataResult stream rres =+ case readResultStatus rres of+ RE_SUCCESS ->+ case action of+ Just orig ->+ case decode $ fromStrict $ recordedEventData orig of+ Just s ->+ let res = StreamMetadataResult+ { streamMetaResultStream = stream+ , streamMetaResultDeleted = False+ , streamMetaResultVersion = evt_number+ , streamMetaResultData = s+ } in+ return res+ Nothing -> fail "StreamMetadata: wrong format."+ Nothing -> fail "impossible: extractStreamMetadataResult"+ RE_STREAM_DELETED -> return $ DeletedStreamMetadataResult stream+ RE_NOT_FOUND -> return $ NotFoundStreamMetadataResult stream+ RE_NO_STREAM -> return $ NotFoundStreamMetadataResult stream+ _ -> fail "unexpected ReadEventResult"++ where+ action = readResultResolvedEvent rres >>= resolvedEventOriginal+ evt_number = readResultEventNumber rres++-------------------------------------------------------------------------------- createAsync :: IO (Async a, MVar (OperationExceptional a)) createAsync = do mvar <- newEmptyMVar@@ -385,3 +485,7 @@ either throwIO return res return (as, mvar)++--------------------------------------------------------------------------------+metaStreamOf :: Text -> Text+metaStreamOf s = "$$" <> s
Database/EventStore/Internal/Types.hs view
@@ -1,6 +1,8 @@ {-# LANGUAGE DeriveDataTypeable #-} {-# LANGUAGE DeriveGeneric #-} {-# LANGUAGE DataKinds #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-} -------------------------------------------------------------------------------- -- | -- Module : Database.EventStore.Internal.Types@@ -15,20 +17,27 @@ 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 Foreign.C.Types (CTime(..))-import GHC.Generics (Generic)+import Data.Monoid (Endo(..)) --------------------------------------------------------------------------------+import Control.Applicative+import Control.Exception+import Control.Monad+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 Control.Concurrent.Async hiding (link)-import qualified Data.Aeson as A+import qualified Data.Aeson as A+import Data.Aeson.Types (Object, ToJSON(..), Pair, Parser, (.=))+import qualified Data.HashMap.Strict as H import Data.ProtocolBuffers import Data.Text (Text) import Data.Time@@ -37,6 +46,9 @@ import System.Random --------------------------------------------------------------------------------+import Database.EventStore.Internal.TimeSpan++-------------------------------------------------------------------------------- -- Exceptions -------------------------------------------------------------------------------- data InternalException@@ -123,8 +135,9 @@ -- idempotency assurances given by the EventStore. -- -- The EventStore will assure idempotency for all operations using any value--- in 'ExpectedVersion' except for 'anyStream'. When using 'anyStream' the EventStore will--- do its best to assure idempotency but will not guarantee idempotency.+-- in 'ExpectedVersion' except for 'anyStream'. When using 'anyStream' the+-- EventStore will do its best to assure idempotency but will not guarantee+-- idempotency. data ExpectedVersion = Any | NoStream@@ -154,7 +167,7 @@ -- | The stream should exist and should be empty. If it does not exist or -- is not empty, treat that as a concurrency problem. emptyStream :: ExpectedVersion-emptyStream =EmptyStream+emptyStream = EmptyStream -------------------------------------------------------------------------------- -- | States that the last event written to the stream should have a@@ -622,3 +635,331 @@ -- | Millisecond timespan msDiffTime :: Float -> NominalDiffTime msDiffTime i = fromRational $ toRational (i / 1000)++--------------------------------------------------------------------------------+-- | Represents an access control list for a stream.+data StreamACL+ = StreamACL+ { streamACLReadRoles :: ![Text]+ -- ^ Roles and users permitted to read the stream.+ , streamACLWriteRoles :: ![Text]+ -- ^ Roles and users permitted to write to the stream.+ , streamACLDeleteRoles :: ![Text]+ -- ^ Roles and users permitted to delete to the stream.+ , streamACLMetaReadRoles :: ![Text]+ -- ^ Roles and users permitted to read stream metadata.+ , streamACLMetaWriteRoles :: ![Text]+ -- ^ Roles and users permitted to write stream metadata.+ } deriving Show++--------------------------------------------------------------------------------+-- | 'StreamACL' with no role or users whatsoever.+emptyStreamACL :: StreamACL+emptyStreamACL = StreamACL+ { streamACLReadRoles = []+ , streamACLWriteRoles = []+ , streamACLDeleteRoles = []+ , streamACLMetaReadRoles = []+ , streamACLMetaWriteRoles = []+ }++--------------------------------------------------------------------------------+-- | Represents stream metadata with strongly typed properties for system values+-- and a dictionary-like interface for custom values.+data StreamMetadata+ = StreamMetadata+ { streamMetadataMaxCount :: !(Maybe Int32)+ -- ^ The maximum number of events allowed in the stream.+ , streamMetadataMaxAge :: !(Maybe TimeSpan)+ -- ^ The maximum age of events allowed in the stream.+ , streamMetadataTruncateBefore :: !(Maybe Int32)+ -- ^ The event number from which previous events can be scavenged. This+ -- is used to implement soft-deletion of streams.+ , streamMetadataCacheControl :: !(Maybe TimeSpan)+ -- ^ The amount of time for which the stream head is cachable.+ , streamMetadataACL :: !StreamACL+ -- ^ The access control list for the stream.+ , streamMetadataCustom :: !Object+ -- ^ An enumerable of key-value pairs of keys to JSON text for+ -- user-provider metadata.+ } deriving Show++--------------------------------------------------------------------------------+instance A.FromJSON StreamMetadata where+ parseJSON = parseStreamMetadata++--------------------------------------------------------------------------------+-- | 'StreamMetadata' with everything set to 'Nothing', using 'emptyStreamACL'+-- and an empty 'Object'.+emptyStreamMetadata :: StreamMetadata+emptyStreamMetadata = StreamMetadata+ { streamMetadataMaxCount = Nothing+ , streamMetadataMaxAge = Nothing+ , streamMetadataTruncateBefore = Nothing+ , streamMetadataCacheControl = Nothing+ , streamMetadataACL = emptyStreamACL+ , streamMetadataCustom = H.empty+ }++--------------------------------------------------------------------------------+customMetaToPairs :: Object -> [Pair]+customMetaToPairs = fmap go . H.toList+ where+ go (k,v) = k .= v++--------------------------------------------------------------------------------+streamACLJSON :: StreamACL -> A.Value+streamACLJSON StreamACL{..} =+ A.object [ p_readRoles .= streamACLReadRoles+ , p_writeRoles .= streamACLWriteRoles+ , p_deleteRoles .= streamACLDeleteRoles+ , p_metaReadRoles .= streamACLMetaReadRoles+ , p_metaWriteRoles .= streamACLMetaWriteRoles+ ]++--------------------------------------------------------------------------------+streamMetadataJSON :: StreamMetadata -> A.Value+streamMetadataJSON StreamMetadata{..} =+ A.object $ [ p_maxAge .= streamMetadataMaxAge+ , p_maxCount .= streamMetadataMaxCount+ , p_truncateBefore .= streamMetadataTruncateBefore+ , p_cacheControl .= streamMetadataCacheControl+ , p_acl .= streamACLJSON streamMetadataACL+ ] ++ custPairs+ where+ custPairs = customMetaToPairs streamMetadataCustom++--------------------------------------------------------------------------------+-- Stream ACL Properties+--------------------------------------------------------------------------------+p_readRoles :: Text+p_readRoles = "$r"++--------------------------------------------------------------------------------+p_writeRoles :: Text+p_writeRoles = "$w"++--------------------------------------------------------------------------------+p_deleteRoles :: Text+p_deleteRoles = "$d"++--------------------------------------------------------------------------------+p_metaReadRoles :: Text+p_metaReadRoles = "$mr"++--------------------------------------------------------------------------------+p_metaWriteRoles :: Text+p_metaWriteRoles = "$mw"++--------------------------------------------------------------------------------+-- Internal MetaData Properties+--------------------------------------------------------------------------------+p_maxAge :: Text+p_maxAge = "$maxAge"++--------------------------------------------------------------------------------+p_maxCount :: Text+p_maxCount = "$maxCount"++--------------------------------------------------------------------------------+p_truncateBefore :: Text+p_truncateBefore = "$tb"++--------------------------------------------------------------------------------+p_cacheControl :: Text+p_cacheControl = "$cacheControl"++--------------------------------------------------------------------------------+p_acl :: Text+p_acl = "$acl"++--------------------------------------------------------------------------------+internalMetaProperties :: S.Set Text+internalMetaProperties =+ S.fromList [ p_maxAge+ , p_maxCount+ , p_truncateBefore+ , p_cacheControl+ , p_acl+ ]++--------------------------------------------------------------------------------+keepUserProperties :: Object -> Object+keepUserProperties = H.filterWithKey go+ where+ go k _ = not $ S.member k internalMetaProperties++--------------------------------------------------------------------------------+--------------------------------------------------------------------------------+parseNominalDiffTime :: Text -> Object -> Parser (Maybe NominalDiffTime)+parseNominalDiffTime k m = fmap (fmap go) (m A..: k)+ where+ go i = (realToFrac $ CTime i)++--------------------------------------------------------------------------------+parseStreamACL :: A.Value -> Parser StreamACL+parseStreamACL (A.Object m) =+ StreamACL <$>+ m A..: p_readRoles <*>+ m A..: p_writeRoles <*>+ m A..: p_deleteRoles <*>+ m A..: p_metaReadRoles <*>+ m A..: p_metaWriteRoles+parseStreamACL _ = mzero++--------------------------------------------------------------------------------+parseStreamMetadata :: A.Value -> Parser StreamMetadata+parseStreamMetadata (A.Object m) =+ StreamMetadata <$>+ m A..: p_maxCount <*>+ m A..: p_maxAge <*>+ m A..: p_truncateBefore <*>+ m A..: p_cacheControl <*>+ (m A..: p_acl >>= parseStreamACL) <*>+ pure (keepUserProperties m)+parseStreamMetadata _ = mzero++--------------------------------------------------------------------------------+-- Builder+--------------------------------------------------------------------------------+type Builder a = Endo a++--------------------------------------------------------------------------------+build :: a -> Builder a -> a+build a (Endo k) = k a++--------------------------------------------------------------------------------+type StreamACLBuilder = Builder StreamACL++--------------------------------------------------------------------------------+-- | Sets role names with read permission for the stream.+setReadRoles :: [Text] -> StreamACLBuilder+setReadRoles xs = Endo $ \s -> s { streamACLReadRoles = xs }++--------------------------------------------------------------------------------+-- | Sets a single role name with read permission for the stream.+setReadRole :: Text -> StreamACLBuilder+setReadRole x = setReadRoles [x]++--------------------------------------------------------------------------------+-- | Sets role names with write permission for the stream.+setWriteRoles :: [Text] -> StreamACLBuilder+setWriteRoles xs = Endo $ \s -> s { streamACLWriteRoles = xs }++--------------------------------------------------------------------------------+-- | Sets a single role name with write permission for the stream.+setWriteRole :: Text -> StreamACLBuilder+setWriteRole x = setWriteRoles [x]++--------------------------------------------------------------------------------+-- | Sets role names with delete permission for the stream.+setDeleteRoles :: [Text] -> StreamACLBuilder+setDeleteRoles xs = Endo $ \s -> s { streamACLDeleteRoles = xs }++--------------------------------------------------------------------------------+-- | Sets a single role name with delete permission for the stream.+setDeleteRole :: Text -> StreamACLBuilder+setDeleteRole x = setDeleteRoles [x]++--------------------------------------------------------------------------------+-- | Sets role names with metadata read permission for the stream.+setMetaReadRoles :: [Text] -> StreamACLBuilder+setMetaReadRoles xs = Endo $ \s -> s { streamACLMetaReadRoles = xs }++--------------------------------------------------------------------------------+-- | Sets a single role name with metadata read permission for the stream.+setMetaReadRole :: Text -> StreamACLBuilder+setMetaReadRole x = setMetaReadRoles [x]++--------------------------------------------------------------------------------+-- | Sets role names with metadata write permission for the stream.+setMetaWriteRoles :: [Text] -> StreamACLBuilder+setMetaWriteRoles xs = Endo $ \s -> s { streamACLMetaWriteRoles = xs }++--------------------------------------------------------------------------------+-- | Sets a single role name with metadata write permission for the stream.+setMetaWriteRole :: Text -> StreamACLBuilder+setMetaWriteRole x = setMetaWriteRoles [x]++--------------------------------------------------------------------------------+-- | Builds a 'StreamACL' from a 'StreamACLBuilder'.+buildStreamACL :: StreamACLBuilder -> StreamACL+buildStreamACL b = modifyStreamACL b emptyStreamACL++--------------------------------------------------------------------------------+-- | Modifies a 'StreamACL' using a 'StreamACLBuilder'.+modifyStreamACL :: StreamACLBuilder -> StreamACL -> StreamACL+modifyStreamACL b acl = build acl b++--------------------------------------------------------------------------------+type StreamMetadataBuilder = Builder StreamMetadata++--------------------------------------------------------------------------------+-- | Sets the maximum number of events allowed in the stream.+setMaxCount :: Int32 -> StreamMetadataBuilder+setMaxCount i = Endo $ \s -> s { streamMetadataMaxCount = Just i }++--------------------------------------------------------------------------------+-- | Sets the maximum age of events allowed in the stream.+setMaxAge :: TimeSpan -> StreamMetadataBuilder+setMaxAge d = Endo $ \s -> s { streamMetadataMaxAge = Just d }++--------------------------------------------------------------------------------+-- | Sets the event number from which previous events can be scavenged.+setTruncateBefore :: Int32 -> StreamMetadataBuilder+setTruncateBefore i = Endo $ \s -> s { streamMetadataTruncateBefore = Just i }++--------------------------------------------------------------------------------+-- | Sets the amount of time for which the stream head is cachable.+setCacheControl :: TimeSpan -> StreamMetadataBuilder+setCacheControl d = Endo $ \s -> s { streamMetadataCacheControl = Just d }++--------------------------------------------------------------------------------+setACL :: StreamACL -> StreamMetadataBuilder+setACL a = Endo $ \s -> s { streamMetadataACL = a }++--------------------------------------------------------------------------------+modifyACL :: StreamACLBuilder -> StreamMetadataBuilder+modifyACL b = Endo $ \s ->+ let old = streamMetadataACL s+ in s { streamMetadataACL = modifyStreamACL b old }++--------------------------------------------------------------------------------+-- | Sets a custom metadata property.+setCustomProperty :: ToJSON a => Text -> a -> StreamMetadataBuilder+setCustomProperty k v = Endo $ \s ->+ let m = streamMetadataCustom s+ m' = H.insert k (toJSON v) m in+ s { streamMetadataCustom = m' }++--------------------------------------------------------------------------------+-- | Builds a 'StreamMetadata' from a 'StreamMetadataBuilder'.+buildStreamMetadata :: StreamMetadataBuilder -> StreamMetadata+buildStreamMetadata b = modifyStreamMetadata b emptyStreamMetadata++--------------------------------------------------------------------------------+-- | Modifies a 'StreamMetadata' using a 'StreamMetadataBuilder'+modifyStreamMetadata :: StreamMetadataBuilder+ -> StreamMetadata+ -> StreamMetadata+modifyStreamMetadata b meta = build meta b++--------------------------------------------------------------------------------+-- | Represents stream metadata as a series of properties for system data and a+-- 'StreamMetadata' object for user metadata.+data StreamMetadataResult+ = StreamMetadataResult+ { streamMetaResultStream :: !Text+ -- ^ The name of the stream.+ , streamMetaResultDeleted :: !Bool+ -- ^ True if the stream is soft-deleted.+ , streamMetaResultVersion :: !Int32+ -- ^ The version of the metadata format.+ , streamMetaResultData :: !StreamMetadata+ -- ^ A 'StreamMetadata' containing user-specified metadata.+ }+ | NotFoundStreamMetadataResult { streamMetaResultStream :: !Text }+ -- ^ When the stream is either not found or 'no stream'.+ | DeletedStreamMetadataResult { streamMetaResultStream :: !Text }+ -- ^ When the stream is deleted.+ deriving Show
eventstore.cabal view
@@ -10,7 +10,7 @@ -- PVP summary: +-+------- breaking API changes -- | | +----- non-breaking API additions -- | | | +--- code changes with no API change-version: 0.7.1.0+version: 0.7.2.0 -- A short (one-line) description of the package. synopsis: EventStore TCP Client@@ -91,6 +91,9 @@ , text >=1.1.1 && <1.3 , time >=1.4 && <1.6 , uuid ==1.3.*+ , unordered-containers+ , stm+ , attoparsec -- Directories containing source files. -- hs-source-dirs: