eventium-memory 0.3.2 → 0.4.0
raw patch · 3 files changed
+119/−9 lines, 3 filesdep ~eventium-core
Dependency ranges changed: eventium-core
Files
- eventium-memory.cabal +3/−3
- src/Eventium/Store/Memory.hs +12/−6
- tests/Eventium/ReadModelSpec.hs +104/−0
eventium-memory.cabal view
@@ -5,7 +5,7 @@ -- see: https://github.com/sol/hpack name: eventium-memory-version: 0.3.2+version: 0.4.0 synopsis: In-memory implementations for eventium description: Eventium-memory provides in-memory implementations of event stores and projection caches for the Eventium event sourcing framework. This package is ideal for development, testing, and prototyping@@ -49,7 +49,7 @@ build-depends: base >=4.9 && <5 , containers >=0.6 && <0.8- , eventium-core >=0.3.0 && <0.4.0+ , eventium-core >=0.4.0 && <0.5.0 , mtl >=2.2 && <2.4 , safe ==0.3.* , stm ==2.5.*@@ -89,7 +89,7 @@ HUnit , base >=4.9 && <5 , containers >=0.6 && <0.8- , eventium-core >=0.3.0 && <0.4.0+ , eventium-core >=0.4.0 && <0.5.0 , eventium-testkit , hspec , mtl >=2.2 && <2.4
src/Eventium/Store/Memory.hs view
@@ -191,23 +191,29 @@ (StartQueryAt startSeq) -> startSeq storeEventMap ::- EventMap event -> UUID -> [event] -> (EventMap event, EventVersion)+ EventMap event -> UUID -> [event] -> (EventMap event, EventWriteResult) storeEventMap store@(EventMap uuidMap globalEvents) uuid events = let versStart = latestEventVersion store uuid- streamEvents = zipWith (\v e -> StreamEvent uuid v (emptyMetadata "") e) [versStart + 1 ..] events+ vers = take (length events) [versStart + 1 ..]+ streamEvents = zipWith (\v e -> StreamEvent uuid v (emptyMetadata "") e) vers events newMap = Map.insertWith (flip (><)) uuid (Seq.fromList streamEvents) uuidMap globalEvents' = globalEvents >< Seq.fromList streamEvents- in (EventMap newMap globalEvents', versStart + EventVersion (length events))+ -- The global reader numbers events 1-based by their position in+ -- 'globalEvents'; mirror that from the length before appending.+ globals = take (length events) [SequenceNumber (Seq.length globalEvents + 1) ..]+ in (EventMap newMap globalEvents', zip vers globals) storeEventMapTagged ::- EventMap event -> UUID -> [TaggedEvent event] -> (EventMap event, EventVersion)+ EventMap event -> UUID -> [TaggedEvent event] -> (EventMap event, EventWriteResult) storeEventMapTagged store@(EventMap uuidMap globalEvents) uuid taggedEvents = let versStart = latestEventVersion store uuid+ vers = take (length taggedEvents) [versStart + 1 ..] streamEvents = zipWith (\v (TaggedEvent meta e) -> StreamEvent uuid v meta e)- [versStart + 1 ..]+ vers taggedEvents newMap = Map.insertWith (flip (><)) uuid (Seq.fromList streamEvents) uuidMap globalEvents' = globalEvents >< Seq.fromList streamEvents- in (EventMap newMap globalEvents', versStart + EventVersion (length taggedEvents))+ globals = take (length taggedEvents) [SequenceNumber (Seq.length globalEvents + 1) ..]+ in (EventMap newMap globalEvents', zip vers globals)
tests/Eventium/ReadModelSpec.hs view
@@ -180,3 +180,107 @@ rB <- readTVarIO resetB rA `shouldBe` True rB `shouldBe` True++ describe "readModelPublisher (synchronous driver)" $ do+ it "applies the handler and advances the checkpoint during the write" $ do+ eventTVar <- eventMapTVar+ sumRef <- newTVarIO (0 :: Int)+ checkpointRef <- newTVarIO (0 :: SequenceNumber)+ let baseWriter = runEventStoreWriterUsing atomically (tvarEventStoreWriter eventTVar)+ globalReader = runEventStoreReaderUsing atomically (tvarGlobalEventStoreReader eventTVar)+ rm =+ ReadModel+ { initialize = pure (),+ eventHandler = EventHandler $ \globalEvent ->+ case globalEvent.payload.payload of+ Added n -> atomically $ modifyTVar' sumRef (+ n)+ _ -> return (),+ checkpointStore =+ CheckpointStore+ { getCheckpoint = readTVarIO checkpointRef,+ saveCheckpoint = atomically . writeTVar checkpointRef+ },+ reset = pure ()+ }+ -- Same ReadModel, driven synchronously in the write path.+ writer = publishingGlobalEventStoreWriter baseWriter (readModelPublisher rm)++ _ <- writer.storeEvents (uuidFromInteger 1) NoStream [Added 1, Added 2]+ _ <- writer.storeEvents (uuidFromInteger 2) NoStream [Added 10]++ -- Handler ran synchronously during the writes.+ readTVarIO sumRef `shouldReturn` 13+ -- Checkpoint advanced in-line to the last assigned global position.+ readTVarIO checkpointRef `shouldReturn` 3+ -- Nothing is left for an async catch-up: the checkpoint is already current.+ newEvents <- globalReader.getEvents (eventsStartingAt () 4)+ length newEvents `shouldBe` 0++ describe "catchUpReadModel" $ do+ it "brings a model current without resetting, and is a no-op when current" $ do+ eventTVar <- eventMapTVar+ let writer = runEventStoreWriterUsing atomically (tvarEventStoreWriter eventTVar)+ globalReader = runEventStoreReaderUsing atomically (tvarGlobalEventStoreReader eventTVar)+ _ <- writer.storeEvents (uuidFromInteger 1) NoStream [Added 1, Added 2]+ _ <- writer.storeEvents (uuidFromInteger 2) NoStream [Added 10]++ sumRef <- newTVarIO (0 :: Int)+ checkpointRef <- newTVarIO (0 :: SequenceNumber)+ resetCount <- newTVarIO (0 :: Int)+ let rm =+ ReadModel+ { initialize = pure (),+ eventHandler = EventHandler $ \globalEvent ->+ case globalEvent.payload.payload of+ Added n -> atomically $ modifyTVar' sumRef (+ n)+ _ -> return (),+ checkpointStore =+ CheckpointStore+ { getCheckpoint = readTVarIO checkpointRef,+ saveCheckpoint = atomically . writeTVar checkpointRef+ },+ reset = atomically $ modifyTVar' resetCount (+ 1)+ }++ catchUpReadModel globalReader rm+ readTVarIO sumRef `shouldReturn` 13+ readTVarIO checkpointRef `shouldReturn` 3+ -- Did not reset, and a second catch-up applies nothing new.+ catchUpReadModel globalReader rm+ readTVarIO sumRef `shouldReturn` 13+ readTVarIO resetCount `shouldReturn` 0++ describe "rebuildReadModel resets the checkpoint itself" $ do+ it "replays from scratch even when reset clears only view data" $ do+ eventTVar <- eventMapTVar+ let writer = runEventStoreWriterUsing atomically (tvarEventStoreWriter eventTVar)+ globalReader = runEventStoreReaderUsing atomically (tvarGlobalEventStoreReader eventTVar)+ _ <- writer.storeEvents (uuidFromInteger 1) NoStream [Added 1, Added 2]++ sumRef <- newTVarIO (0 :: Int)+ checkpointRef <- newTVarIO (0 :: SequenceNumber)+ let rm =+ ReadModel+ { initialize = pure (),+ eventHandler = EventHandler $ \globalEvent ->+ case globalEvent.payload.payload of+ Added n -> atomically $ modifyTVar' sumRef (+ n)+ _ -> return (),+ checkpointStore =+ CheckpointStore+ { getCheckpoint = readTVarIO checkpointRef,+ saveCheckpoint = atomically . writeTVar checkpointRef+ },+ -- Clears view DATA only — deliberately does NOT touch the checkpoint.+ reset = atomically $ writeTVar sumRef 0+ }++ -- Advance the checkpoint as if previously caught up.+ catchUpReadModel globalReader rm+ readTVarIO sumRef `shouldReturn` 3+ readTVarIO checkpointRef `shouldReturn` 2++ -- Rebuild must still replay from the start: eventium resets the checkpoint+ -- even though reset left it at 2.+ rebuildReadModel globalReader rm+ readTVarIO sumRef `shouldReturn` 3