cqrs-memory (empty) → 0.10.0
raw patch · 9 files changed
+385/−0 lines, 9 filesdep +basedep +bytestringdep +containerssetup-changed
Dependencies added: base, bytestring, containers, cqrs-core, cqrs-memory, cqrs-testkit, hspec, io-streams, random, stm, uuid-types
Files
- LICENSE +19/−0
- Setup.hs +2/−0
- cqrs-memory.cabal +44/−0
- src-test/Main.hs +36/−0
- src/Data/CQRS/Memory.hs +13/−0
- src/Data/CQRS/Memory/Internal/ArchiveStore.hs +123/−0
- src/Data/CQRS/Memory/Internal/EventStore.hs +82/−0
- src/Data/CQRS/Memory/Internal/SnapshotStore.hs +30/−0
- src/Data/CQRS/Memory/Internal/Storage.hs +36/−0
+ LICENSE view
@@ -0,0 +1,19 @@+Copyright (c) 2011-2015 Bardur Arantsson++Permission is hereby granted, free of charge, to any person obtaining a copy+of this software and associated documentation files (the "Software"), to deal+in the Software without restriction, including without limitation the rights+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell+copies of the Software, and to permit persons to whom the Software is+furnished to do so, subject to the following conditions:++The above copyright notice and this permission notice shall be included in+all copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN+THE SOFTWARE.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ cqrs-memory.cabal view
@@ -0,0 +1,44 @@+Name: cqrs-memory+Version: 0.10.0+Synopsis: Memory backend for the cqrs package.+Description: Memory backend for the cqrs package.+License: MIT+License-file: LICENSE+Category: Data+Cabal-version: >=1.10+Build-type: Simple+Author: Bardur Arantsson+Maintainer: Bardur Arantsson <bardur@scientician.net>++Library+ Build-Depends: base >= 4.8 && < 5+ , bytestring >= 0.9.0.1+ , containers >= 0.5+ , cqrs-core >= 0.10.0 && < 0.11.0+ , io-streams >= 1.2 && < 2+ , stm >= 2.4 && < 3+ , uuid-types >= 1.0 && < 1.1+ Default-language: Haskell2010+ ghc-options: -Wall+ hs-source-dirs: src+ Exposed-modules: Data.CQRS.Memory.Internal.ArchiveStore+ Data.CQRS.Memory.Internal.EventStore+ Data.CQRS.Memory.Internal.SnapshotStore+ Data.CQRS.Memory.Internal.Storage+ Data.CQRS.Memory++Test-Suite tests+ Type: exitcode-stdio-1.0+ Hs-source-dirs: src-test+ Main-is: Main.hs+ Build-depends: base == 4.*+ -- Self-dependency+ , cqrs-memory+ , cqrs-core >= 0.10.0 && < 0.11.0+ , cqrs-testkit >= 0.10.0 && < 0.11.0+ -- Test framework+ , hspec >= 2.2.0 && < 3.0+ -- Libraries+ , random >= 1.1 && < 2+ Ghc-options: -Wall+ Default-language: Haskell2010
+ src-test/Main.hs view
@@ -0,0 +1,36 @@+module Main (main) where++import Test.Hspec (hspec)+import Data.CQRS.Memory (newEventStore, newArchiveStore, newStorage, newSnapshotStore)+import Data.CQRS.Test.TestKit (mkArchiveStoreSpec, mkEventStoreSpec, mkRepositorySpec, mkSnapshotStoreSpec, TestKitSettings(..))+import System.Random (randomIO)++-- Run all the test suites.+main :: IO ()+main = do+ -- Setup+ let testKitSettings = TestKitSettings+ { tksMakeContext = \_ -> return ()+ , tksSetUp = newStorage+ , tksTearDown = \_ -> return ()+ }+ -- Run the full test kit.+ hspec $ do+ mkArchiveStoreSpec $ testKitSettings {+ tksMakeContext = \c -> do+ as <- newArchiveStore randomIO c+ es <- newEventStore c+ return $ (as, es)+ }+ mkEventStoreSpec $ testKitSettings {+ tksMakeContext = newEventStore+ }+ mkRepositorySpec $ testKitSettings {+ tksMakeContext = \c -> do+ es <- newEventStore c+ ss <- newSnapshotStore+ return (es, ss)+ }+ mkSnapshotStoreSpec $ testKitSettings {+ tksMakeContext = \_ -> newSnapshotStore+ }
+ src/Data/CQRS/Memory.hs view
@@ -0,0 +1,13 @@+-- | Memory-based event store. Used primarily for testing.+module Data.CQRS.Memory+ ( Storage+ , newEventStore+ , newArchiveStore+ , newStorage+ , newSnapshotStore+ ) where++import Data.CQRS.Memory.Internal.ArchiveStore+import Data.CQRS.Memory.Internal.EventStore+import Data.CQRS.Memory.Internal.Storage+import Data.CQRS.Memory.Internal.SnapshotStore
+ src/Data/CQRS/Memory/Internal/ArchiveStore.hs view
@@ -0,0 +1,123 @@+module Data.CQRS.Memory.Internal.ArchiveStore+ ( newArchiveStore+ ) where++import Control.Concurrent.STM (STM, atomically)+import Control.Concurrent.STM.TVar (readTVar, readTVarIO, writeTVar)+import Control.Monad (liftM)+import Data.CQRS.Types.ArchiveRef+import Data.CQRS.Types.ArchiveMetadata+import Data.CQRS.Types.ArchiveStore (ArchiveStore(..))+import Data.CQRS.Types.PersistedEvent (PersistedEvent(..))+import Data.CQRS.Memory.Internal.Storage+import qualified Data.Foldable as F+import Data.Sequence (Seq, ViewR(..), ViewL(..), viewr, viewl, (><), (|>))+import qualified Data.Sequence as S+import Data.UUID.Types (UUID)+import System.IO.Streams (InputStream)+import qualified System.IO.Streams.List as SL++lastOption :: Seq a -> Maybe a+lastOption xs = case viewr xs of+ EmptyR -> Nothing+ (_ :> x) -> Just $ x++uncons :: Seq a -> Maybe (a, Seq a)+uncons s = case viewl s of+ EmptyL -> Nothing+ (x :< xs) -> Just (x, xs)++inCurrentArchive :: Event e -> Bool+inCurrentArchive (Event _ _ CurrentArchive) = True+inCurrentArchive (Event _ _ (NamedArchive _)) = False++readLatestArchiveMetadata :: Storage e -> IO (Maybe ArchiveMetadata)+readLatestArchiveMetadata (Storage store) = do+ archives <- fmap msArchives $ readTVarIO store+ return $ lastOption archives++readArchiveMetadata :: Storage e -> UUID -> IO (Maybe ArchiveMetadata)+readArchiveMetadata (Storage store) archiveId = do+ archives <- fmap msArchives $ readTVarIO store+ return $ F.find (\a -> amArchiveId a == archiveId) archives++readArchive :: Storage e -> ArchiveRef -> (InputStream (UUID, PersistedEvent e) -> IO a) -> IO a+readArchive (Storage store) archiveRef f = do+ events <- liftM msEvents $ readTVarIO store+ eventStream <- SL.fromList $ F.toList+ $ fmap (\e -> (eAggregateId e, ePersistedEvent e))+ $ S.filter (\e -> archiveRef == eArchiveRef e)+ $ events+ f eventStream++getUnarchivedEventCount :: Storage e -> IO Int+getUnarchivedEventCount (Storage store) = atomically getCurrentEventCount+ where+ getCurrentEventCount :: STM Int+ getCurrentEventCount = do+ events <- fmap msEvents $ readTVar store+ return $ S.length $ S.filter inCurrentArchive events++archiveEvents :: Storage e -> IO UUID -> Int -> IO (Maybe UUID)+archiveEvents (Storage store) uuidSupply archiveSize =+ if (archiveSize > 0) then+ do+ newArchiveId <- uuidSupply+ doArchiveEvents newArchiveId+ return $ Just newArchiveId+ else+ return $ Nothing+ where+ doArchiveEvents :: UUID -> IO ()+ doArchiveEvents newArchiveId = do+ atomically $ do+ memoryStorage <- readTVar store+ let archives = msArchives memoryStorage+ let events = msEvents memoryStorage+ -- Find previous archive to point to the new archive.+ let maybePreviousArchive = lastOption archives+ -- Update the last archive's metadata (if necessary).+ let archives' = case maybePreviousArchive of+ Nothing ->+ archives -- No changes necessary+ Just previousArchive ->+ let previousArchive' = previousArchive { amNextArchiveId = NamedArchive newArchiveId } in+ S.update (S.length archives - 1) previousArchive' archives+ -- Create new archive+ let newArchiveMetadata = ArchiveMetadata+ { amArchiveId = newArchiveId+ , amPreviousArchiveId = fmap amArchiveId maybePreviousArchive+ , amNextArchiveId = CurrentArchive+ }+ let archives'' = archives' |> newArchiveMetadata+ -- Move <archiveSize'> events in Current to new archive. We+ -- exploit the fact that events are added in chronological+ -- order.+ let (archivedEvents, unarchivedEvents) = S.partition (not . inCurrentArchive) events+ let (archivedEvents', unarchivedEvents') = archive archiveSize newArchiveId archivedEvents unarchivedEvents+ let events' = archivedEvents' >< unarchivedEvents'+ -- Actually perform the update+ writeTVar store $ memoryStorage { msArchives = archives''+ , msEvents = events' }++ archive :: Int -> UUID -> Seq (Event e) -> Seq (Event e) -> (Seq (Event e), Seq (Event e))+ archive 0 _ archivedEvents unarchivedEvents = (archivedEvents, unarchivedEvents)+ archive _ _ archivedEvents unarchivedEvents | S.null unarchivedEvents = (archivedEvents, unarchivedEvents)+ archive n newArchiveId archivedEvents unarchivedEvents =+ -- Take off the first unarchived event+ let (unarchivedEvent, unarchivedEvents') = maybe (error "unarchivedEvents unexpectedly empty") id (uncons unarchivedEvents) in+ -- Archive it+ let archivedEvent = unarchivedEvent { eArchiveRef = NamedArchive newArchiveId } in+ -- Bundle up+ archive (n - 1) newArchiveId (archivedEvents |> archivedEvent) unarchivedEvents'++-- | Create a memory-backed archive store.+newArchiveStore :: Show e => IO UUID -> Storage e -> IO (ArchiveStore e)+newArchiveStore uuidSupply storage = do+ return $ ArchiveStore+ { asGetUnarchivedEventCount = getUnarchivedEventCount storage+ , asArchiveEvents = archiveEvents storage uuidSupply+ , asReadLatestArchiveMetadata = readLatestArchiveMetadata storage+ , asReadArchiveMetadata = readArchiveMetadata storage+ , asReadArchive = readArchive storage+ }
+ src/Data/CQRS/Memory/Internal/EventStore.hs view
@@ -0,0 +1,82 @@+module Data.CQRS.Memory.Internal.EventStore+ ( newEventStore+ ) where++import Control.Concurrent.STM (STM, atomically)+import Control.Concurrent.STM.TVar (TVar, readTVar, modifyTVar')+import Control.Monad (when, forM_)+import Control.Monad.STM (throwSTM)+import Data.CQRS.Types.ArchiveRef+import Data.CQRS.Types.EventStore (EventStore(..))+import Data.CQRS.Types.PersistedEvent (PersistedEvent(..))+import Data.CQRS.Types.StoreError (StoreError(..))+import Data.CQRS.Memory.Internal.Storage+import qualified Data.Foldable as F+import Data.List (nub, sortBy)+import Data.Ord (comparing)+import Data.Sequence (Seq, (><))+import qualified Data.Sequence as S+import Data.UUID.Types (UUID)+import System.IO.Streams (InputStream)+import qualified System.IO.Streams.List as SL+import qualified System.IO.Streams.Combinators as SC++storeEvents :: Storage e -> UUID -> [PersistedEvent e] -> IO ()+storeEvents (Storage store) aggregateId newEvents = atomically $ do+ -- Extract all existing events for this aggregate.+ events <- eventsByAggregateId store aggregateId+ let eventSequenceNumbers = F.toList $ fmap peSequenceNumber events+ -- Check for duplicates in the input list itself+ let newSequenceNumbers = F.toList $ fmap peSequenceNumber newEvents+ when ((nub newSequenceNumbers) /= newSequenceNumbers) $ throwSTM $ VersionConflict aggregateId+ -- Check for duplicate events+ forM_ newSequenceNumbers $ \newSequenceNumber -> do+ when (elem newSequenceNumber eventSequenceNumbers) $+ throwSTM $ VersionConflict aggregateId+ -- Check version numbers; this exists as a sanity check for tests+ let vE = lastStoredVersion $ F.toList events+ let v0 = if null newEvents then+ vE + 1+ else+ minimum $ map peSequenceNumber newEvents+ when (v0 /= vE + 1) $ error "Mismatched version numbers"+ -- Append to the list of all previous events+ modifyTVar' store addEvents+ where+ mkEvent persistedEvent = Event aggregateId persistedEvent CurrentArchive++ addEvents ms = ms { msEvents = (msEvents ms) >< (S.fromList $ map mkEvent newEvents) }++ lastStoredVersion [ ] = (-1)+ lastStoredVersion es = maximum $ map peSequenceNumber es++retrieveEvents :: Storage e -> UUID -> Int -> (InputStream (PersistedEvent e) -> IO a) -> IO a+retrieveEvents (Storage store) aggregateId v0 f = do+ events <- fmap F.toList $ atomically $ eventsByAggregateId store aggregateId+ SL.fromList events >>= SC.filter (\e -> peSequenceNumber e > v0) >>= f++retrieveAllEvents :: Storage e -> (InputStream (UUID, PersistedEvent e) -> IO a) -> IO a+retrieveAllEvents (Storage store) f = do+ -- We won't bother with efficiency since this is only+ -- really used for debugging/tests.+ events <- fmap msEvents $ atomically $ readTVar store+ let eventList = F.toList events+ inputStream <- SL.fromList $ sortBy (comparing cf) eventList+ SC.map (\(Event aggregateId event _) -> (aggregateId, event)) inputStream >>= f+ where+ cf e = (eAggregateId e, peSequenceNumber $ ePersistedEvent e)+++eventsByAggregateId :: TVar (Store e) -> UUID -> STM (Seq (PersistedEvent e))+eventsByAggregateId store aggregateId = do+ events <- readTVar store+ return $ fmap ePersistedEvent $ S.filter (\e -> aggregateId == eAggregateId e) $ msEvents events++-- | Create a memory-backend event store.+newEventStore :: Show e => Storage e -> IO (EventStore e)+newEventStore storage = do+ return $ EventStore+ { esStoreEvents = storeEvents storage+ , esRetrieveEvents = retrieveEvents storage+ , esRetrieveAllEvents = retrieveAllEvents storage+ }
+ src/Data/CQRS/Memory/Internal/SnapshotStore.hs view
@@ -0,0 +1,30 @@+module Data.CQRS.Memory.Internal.SnapshotStore+ ( newSnapshotStore+ ) where++import Control.Concurrent.STM (atomically)+import Control.Concurrent.STM.TVar (TVar, newTVar, readTVar, modifyTVar')+import Control.Monad (liftM)+import Data.CQRS.Types.Snapshot+import Data.CQRS.Types.SnapshotStore+import Data.Map.Strict (Map)+import qualified Data.Map.Strict as M+import Data.UUID.Types (UUID)++type SnapStore a = TVar (Map UUID (Snapshot a))++writeSnapshot :: SnapStore a -> UUID -> Snapshot a -> IO ()+writeSnapshot store aggregateId snapshot = do+ atomically $ modifyTVar' store (M.insert aggregateId snapshot)++readSnapshot :: SnapStore a -> UUID -> IO (Maybe (Snapshot a))+readSnapshot store aggregateId = do+ atomically $ liftM (M.lookup aggregateId) $ readTVar store++-- | Create a new memory-backed snapshot store.+newSnapshotStore :: IO (SnapshotStore a)+newSnapshotStore = do+ store <- atomically $ newTVar M.empty+ return $ SnapshotStore+ (writeSnapshot store)+ (readSnapshot store)
+ src/Data/CQRS/Memory/Internal/Storage.hs view
@@ -0,0 +1,36 @@+module Data.CQRS.Memory.Internal.Storage+ ( Event(..)+ , Store(..)+ , Storage(..)+ , newStorage+ ) where++import Control.Concurrent.STM (atomically)+import Control.Concurrent.STM.TVar (TVar, newTVar)+import Data.CQRS.Types.ArchiveRef+import Data.CQRS.Types.ArchiveMetadata+import Data.CQRS.Types.PersistedEvent (PersistedEvent(..))+import Data.Sequence (Seq)+import qualified Data.Sequence as S+import Data.UUID.Types (UUID)++data Event e =+ Event { eAggregateId :: UUID+ , ePersistedEvent :: PersistedEvent e+ , eArchiveRef :: ArchiveRef+ }+ deriving (Show)++data Store e = Store+ { msEvents :: Seq (Event e)+ -- Archives: Archives are kept in chronological order; first to last.+ , msArchives :: Seq ArchiveMetadata+ }++-- | Storage used for memory-backed EventStore and ArchiveStore.+newtype Storage e = Storage (TVar (Store e))++-- | Create backing memory for a memory-based event store+-- or archive store.+newStorage :: IO (Storage e)+newStorage = atomically $ fmap Storage $ newTVar $ Store S.empty S.empty