cqrs-example 0.4.0 → 0.5.0
raw patch · 5 files changed
+138/−74 lines, 5 filesdep +bytestringdep +containersdep +enumerator
Dependencies added: bytestring, containers, enumerator
Files
- cqrs-example.cabal +5/−2
- src/Aggregates.hs +37/−53
- src/Events.hs +12/−12
- src/Main.hs +36/−7
- src/Query.hs +48/−0
cqrs-example.cabal view
@@ -1,5 +1,5 @@ Name: cqrs-example-Version: 0.4.0+Version: 0.5.0 Synopsis: Example for cqrs package License: MIT License-file: LICENSE@@ -12,9 +12,12 @@ Executable cqrs-example Main-is: Main.hs Build-depends: base == 4.*+ , bytestring >= 0.9 && < 0.10 , cereal >= 0.3 && < 0.4+ , containers >= 0.4 && < 0.5 , cqrs >= 0.4 && < 0.5 , data-default >= 0.3 && < 0.4+ , enumerator >= 0.4.15 && < 0.5 , text >= 0.11 && < 0.12 , transformers >= 0.2.2 && < 0.3 Ghc-options: -Wall@@ -25,4 +28,4 @@ Hs-source-dirs: src Other-modules: Aggregates Events- + Query
src/Aggregates.hs view
@@ -1,10 +1,8 @@ module Aggregates ( Project(..) , ProjectId- , ProjectState(..) , Task(..) , TaskId- , TaskState(..) ) where import Control.Monad (liftM)@@ -12,7 +10,6 @@ import Data.Default (Default(..)) import Data.Serialize (Serialize(..), decode, encode) import Data.Text (Text)-import qualified Data.Text as T import Data.Text.Encoding (decodeUtf8, encodeUtf8) import Data.Typeable (Typeable) import Data.Word (Word8)@@ -21,33 +18,26 @@ type ProjectId = GUID Project -data ProjectState = New- | Active- deriving (Eq, Typeable)--instance Serialize ProjectState where- put New = put (0 :: Word8)- put Active = put (1 :: Word8)- get = do- i :: Word8 <- get- case i of- 0 -> return New- 1 -> return Active- _ -> fail $ "Cannot decode project state: " ++ show i--data Project = Project { projectName :: Text- , projectState :: ProjectState- }- deriving (Typeable)+data Project = DefaultProject+ | ActiveProject { projectName :: Text+ }+ deriving (Typeable) instance Serialize Project where- put (Project pn ps) = do+ put DefaultProject = do+ put (0 :: Word8)+ put (ActiveProject pn) = do+ put (1 :: Word8) put $ encodeUtf8 pn- put ps get = do- pn <- liftM decodeUtf8 get- ps <- get- return $ Project pn ps+ i :: Word8 <- get+ case i of+ 0 -> return DefaultProject+ 1 -> do+ pn <- liftM decodeUtf8 get+ return $ ActiveProject pn+ _ ->+ fail $ "Cannot decode project state" ++ show i instance Aggregate Project where encodeAggregate = encode@@ -57,43 +47,37 @@ Right a -> a instance Default Project where- def = Project T.empty New+ def = DefaultProject -- Tasks. type TaskId = GUID Task -data TaskState = TaskNew- | TaskActive- deriving (Eq, Typeable)--instance Serialize TaskState where- put TaskNew = put (0 :: Word8)- put TaskActive = put (1 :: Word8)- get = do- i :: Word8 <- get- case i of- 0 -> return TaskNew- 1 -> return TaskActive- _ -> fail $ "Cannot decode task state: " ++ show i--data Task = Task { taskProjectId :: ProjectId- , taskState :: TaskState- , taskShortDescription :: Text- }- deriving (Typeable)+data Task = NewTask+ | ActiveTask { taskProjectId :: ProjectId+ , taskShortDescription :: Text+ }+ deriving (Typeable) instance Serialize Task where- put (Task tpid ts tsd) = do+ put NewTask = do+ put (0 :: Word8)+ put (ActiveTask tpid tsd) = do+ put (1 :: Word8) put tpid- put ts put $ encodeUtf8 tsd get = do- tpid <- get- ts <- get- tsd <- liftM decodeUtf8 get- return $ Task tpid ts tsd+ i :: Word8 <- get+ case i of+ 0 -> return NewTask+ 1 -> do+ tpid <- get+ tsd <- liftM decodeUtf8 get+ return $ ActiveTask tpid tsd+ _ ->+ fail $ "Cannot decode task state: " ++ show i + instance Aggregate Task where encodeAggregate = encode decodeAggregate s =@@ -102,4 +86,4 @@ Right a -> a instance Default Task where- def = Task def TaskNew T.empty+ def = NewTask
src/Events.hs view
@@ -1,5 +1,5 @@ module Events- ( ExampleEvent(..)+ ( Event(..) ) where import Aggregates@@ -10,19 +10,19 @@ import Data.Typeable (Typeable) import Data.Word (Word8) -data ExampleEvent = CreateProject Text- | RenameProject Text- | AddTask ProjectId Text- deriving (Typeable)+data Event = ProjectCreated Text+ | ProjectRenamed Text+ | TaskAdded ProjectId Text+ deriving (Typeable, Show) -instance Serialize ExampleEvent where- put (CreateProject name) = do+instance Serialize Event where+ put (ProjectCreated name) = do put (0 :: Word8) put $ encodeUtf8 name- put (RenameProject name) = do+ put (ProjectRenamed name) = do put (1 :: Word8) put $ encodeUtf8 name- put (AddTask pid tsd) = do+ put (TaskAdded pid tsd) = do put (2 :: Word8) put $ pid put $ encodeUtf8 tsd@@ -31,14 +31,14 @@ case i of 0 -> do n <- liftM decodeUtf8 get- return $ CreateProject n+ return $ ProjectCreated n 1 -> do n <- liftM decodeUtf8 get- return $ RenameProject n+ return $ ProjectRenamed n 2 -> do pid <- get tsd <- liftM decodeUtf8 get- return $ AddTask pid tsd+ return $ TaskAdded pid tsd _ -> do fail $ "Unrecognized event type " ++ show i
src/Main.hs view
@@ -1,35 +1,64 @@+import Control.Concurrent (forkIO, threadDelay) import Control.Monad.Trans.Class (lift) import Data.CQRS-import Data.CQRS.EventStore (withEventStore)-import Data.CQRS.EventStore.Sqlite3+import Data.CQRS.EventStore.Backend.Sqlite3+import Data.IORef (newIORef, readIORef) import Aggregates import Events import Instances ()+import Query sqliteFile :: String sqliteFile = "example.db" +eventSourcingThread :: EventStore Event -> IO ()+eventSourcingThread eventStore = do+ qsRef <- newIORef emptyQueryState+ loop qsRef+ where+ loop qsRef = do+ putStrLn "Sourcing events..."+ updateQueryStateRef qsRef eventStore+ qs <- readIORef qsRef+ putStrLn $ " => qs: " ++ show qs+ -- Wait 1 second+ threadDelay 1000000+ -- Go again.+ loop qsRef++ test1 :: IO () test1 =+ do+ -- Start sourcing events.+ _ <- forkIO $ do+ withEventStore (openSqliteEventStore sqliteFile) eventSourcingThread++ -- Do a few things withEventStore (openSqliteEventStore sqliteFile) $ \eventStore -> do runTransactionT eventStore $ do -- Create new project. projectId :: GUID Project <- lift $ newGUID (projectRef, _) <- getAggregateRoot projectId- publishEvent projectRef $ CreateProject "my project"+ publishEvent projectRef $ ProjectCreated "my project" -- Add a couple of tasks taskId1 :: GUID Task <- lift $ newGUID (taskRef1, _) <- getAggregateRoot taskId1- publishEvent taskRef1 $ AddTask projectId "tweak knob"+ publishEvent taskRef1 $ TaskAdded projectId "tweak knob" taskId2 :: GUID Task <- lift $ newGUID (taskRef2, _) <- getAggregateRoot taskId2- publishEvent taskRef2 $ AddTask projectId "pull lever"+ publishEvent taskRef2 $ TaskAdded projectId "pull lever" + -- Rename project.+ publishEvent projectRef $ ProjectRenamed "old project"++ main :: IO () main = do putStrLn "Running test1..." test1- putStrLn "Done."-+ putStrLn "Press <Enter> to quit"+ _ <- getLine+ return ()
+ src/Query.hs view
@@ -0,0 +1,48 @@+module Query ( emptyQueryState+ , updateQueryStateRef+ , QueryState+ ) where++import Control.Monad (forM_)+import Data.ByteString (ByteString)+import Data.CQRS (enumerateEventStore, GUID, EventStore)+import Data.Enumerator (run_, (>>==))+import qualified Data.Enumerator.List as EL+import Data.IORef (IORef, readIORef, atomicModifyIORef)+import Data.Map (Map)+import qualified Data.Map as M+import Data.Serialize (encode)+import Data.Text (Text)+import Events (Event(..))++-- Update query state with latest events.+updateQueryStateRef :: IORef QueryState -> EventStore Event -> IO ()+updateQueryStateRef queryStateRef eventStore = do+ lastVersion <- fmap ((+) 1 . qsLastVersion) $ readIORef queryStateRef+ evs <- run_ (EL.consume >>== enumerateEventStore eventStore lastVersion)+ forM_ evs $ \(gv,(g,v,e)) -> do+ putStrLn $ show $ "ev(" ++ show gv ++ "): " ++ show e+ atomicModifyIORef queryStateRef $ \s -> (updateState s gv g v e, ())+ return ()++-- Update state per event.+updateState :: QueryState -> Int -> GUID a -> Int -> Event -> QueryState+updateState queryState gv guid _ (ProjectCreated pn) =+ queryState { qsLastVersion = gv+ , qsProjectNamesById = M.insert (encode guid) pn $ qsProjectNamesById queryState+ }+updateState queryState gv guid v (ProjectRenamed pn) =+ updateState queryState gv guid v (ProjectCreated pn)+updateState queryState gv _ _ (TaskAdded _ _) =+ queryState { qsLastVersion = gv }++-- In-memory data structure for querying. We should really use+-- a persistent store for this, but this is just an example.+data QueryState = QueryState { qsLastVersion :: Int+ , qsProjectNamesById :: Map ByteString Text+ }+ deriving (Show)++-- Create "empty" data structure for querying.+emptyQueryState :: QueryState+emptyQueryState = QueryState 0 M.empty