diff --git a/cqrs-example.cabal b/cqrs-example.cabal --- a/cqrs-example.cabal +++ b/cqrs-example.cabal @@ -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 diff --git a/src/Aggregates.hs b/src/Aggregates.hs --- a/src/Aggregates.hs +++ b/src/Aggregates.hs @@ -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 diff --git a/src/Events.hs b/src/Events.hs --- a/src/Events.hs +++ b/src/Events.hs @@ -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 diff --git a/src/Main.hs b/src/Main.hs --- a/src/Main.hs +++ b/src/Main.hs @@ -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 to quit" + _ <- getLine + return () diff --git a/src/Query.hs b/src/Query.hs new file mode 100644 --- /dev/null +++ b/src/Query.hs @@ -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