cqrs-example 0.5.0.1 → 0.6.0
raw patch · 5 files changed
+372/−81 lines, 5 filesdep +HUnitdep +QuickCheckdep +aesondep ~cqrs
Dependencies added: HUnit, QuickCheck, aeson, cqrs-sqlite3, direct-sqlite, happstack-lite, test-framework, test-framework-hunit, test-framework-quickcheck, time
Dependency ranges changed: cqrs
Files
- cqrs-example.cabal +44/−8
- src/Aggregates.hs +92/−8
- src/Events.hs +33/−3
- src/Main.hs +92/−26
- src/Query.hs +111/−36
cqrs-example.cabal view
@@ -1,31 +1,67 @@ Name: cqrs-example-Version: 0.5.0.1+Version: 0.6.0 Synopsis: Example for cqrs package+Description: Example for cqrs package License: MIT License-file: LICENSE Author: Bardur Arantsson Maintainer: spam@scientician.net Category: Web Build-type: Simple-Cabal-version: >=1.6.0.1+Cabal-version: >=1.10 Executable cqrs-example Main-is: Main.hs+ Ghc-options: -Wall Build-depends: base == 4.*+ , aeson >= 0.3.2.12 && <0.4 , bytestring >= 0.9 && < 0.10 , cereal >= 0.3 && < 0.4 , containers >= 0.4 && < 0.5- , cqrs >= 0.5 && < 0.6+ , cqrs >= 0.6 && < 0.7+ , cqrs-sqlite3 >= 0.6 && < 0.7 , data-default >= 0.3 && < 0.4+ , direct-sqlite >= 1.1 && < 1.2 , enumerator >= 0.4.15 && < 0.5+ , happstack-lite >= 6.0 && < 7.0 , text >= 0.11 && < 0.12+ , time >= 1.2 && < 1.3 , transformers >= 0.2.2 && < 0.3- Ghc-options: -Wall- Extensions: DeriveDataTypeable- MultiParamTypeClasses- OverloadedStrings- ScopedTypeVariables+ Default-extensions: DeriveDataTypeable+ MultiParamTypeClasses+ OverloadedStrings+ ScopedTypeVariables Hs-source-dirs: src Other-modules: Aggregates Events Query+ Default-language: Haskell2010++Test-Suite cqrs-example-tests+ type: exitcode-stdio-1.0+ hs-source-dirs: src+ main-is: Test.hs+ build-depends: base == 4.*+ , bytestring >= 0.9 && < 0.10+ , cereal >= 0.3 && < 0.4+ , containers >= 0.4 && < 0.5+ , cqrs >= 0.5 && < 0.6+ , data-default >= 0.3 && < 0.4+ , direct-sqlite >= 1.1 && < 1.2+ , enumerator >= 0.4.15 && < 0.5+ , text >= 0.11 && < 0.12+ , time >= 1.2 && < 1.3+ , transformers >= 0.2.2 && < 0.3+ -- Test framework:+ , HUnit >= 1.2 && < 2.0+ , test-framework >= 0.4.0+ , test-framework-hunit >= 0.2.0+ , test-framework-quickcheck >= 0.2.0+ , QuickCheck >= 1.1 && < 2+ default-language: Haskell2010+ ghc-options: -threaded -O0+ default-extensions: DeriveDataTypeable+ MultiParamTypeClasses+ OverloadedStrings+ ScopedTypeVariables+
src/Aggregates.hs view
@@ -3,39 +3,54 @@ , ProjectId , Task(..) , TaskId+ , User(..)+ , UserId+ , WorkUnit(..) ) where import Control.Monad (liftM) import Data.CQRS (Aggregate(..), GUID) import Data.Default (Default(..))+import Data.Map (Map) import Data.Serialize (Serialize(..), decode, encode) import Data.Text (Text) import Data.Text.Encoding (decodeUtf8, encodeUtf8)+import Data.Time.Calendar(Day(..), toGregorian, fromGregorian) import Data.Typeable (Typeable) import Data.Word (Word8) +-- Instances for Haskell built-ins.+instance Serialize Day where+ put = put . toGregorian+ get = do+ (a,b,c) <- get+ return $ fromGregorian a b c+ -- Projects. -type ProjectId = GUID Project+type ProjectId = GUID data Project = DefaultProject | ActiveProject { projectName :: Text+ , projectCustomer :: Text }- deriving (Typeable)+ deriving (Typeable, Eq, Show) instance Serialize Project where put DefaultProject = do put (0 :: Word8)- put (ActiveProject pn) = do+ put (ActiveProject pn pc) = do put (1 :: Word8) put $ encodeUtf8 pn+ put $ encodeUtf8 pc get = do i :: Word8 <- get case i of 0 -> return DefaultProject 1 -> do pn <- liftM decodeUtf8 get- return $ ActiveProject pn+ pc <- liftM decodeUtf8 get+ return $ ActiveProject pn pc _ -> fail $ "Cannot decode project state" ++ show i @@ -51,21 +66,25 @@ -- Tasks. -type TaskId = GUID Task+type TaskId = GUID data Task = NewTask | ActiveTask { taskProjectId :: ProjectId , taskShortDescription :: Text+ , taskLongDescription :: Text+ , taskWorkUnits :: Map Day WorkUnit }- deriving (Typeable)+ deriving (Typeable, Eq, Show) instance Serialize Task where put NewTask = do put (0 :: Word8)- put (ActiveTask tpid tsd) = do+ put (ActiveTask tpid tsd tld twu) = do put (1 :: Word8) put tpid put $ encodeUtf8 tsd+ put $ encodeUtf8 tld+ put twu get = do i :: Word8 <- get case i of@@ -73,7 +92,9 @@ 1 -> do tpid <- get tsd <- liftM decodeUtf8 get- return $ ActiveTask tpid tsd+ tld <- liftM decodeUtf8 get+ twu <- get+ return $ ActiveTask tpid tsd tld twu _ -> fail $ "Cannot decode task state: " ++ show i @@ -87,3 +108,66 @@ instance Default Task where def = NewTask++-- Users.++type UserId = GUID++data User = UncreatedUser+ | ActiveUser { auUserName :: Text+ , auLastName :: Text+ , auFirstName :: Text+ }+ deriving (Typeable, Eq, Show)++instance Serialize User where+ put UncreatedUser = do+ put $ (0 :: Word8)+ put (ActiveUser auun auln aufn) = do+ put $ (1 :: Word8)+ put $ encodeUtf8 auun+ put $ encodeUtf8 auln+ put $ encodeUtf8 aufn+ get = do+ i :: Word8 <- get+ case i of+ 0 -> return UncreatedUser+ 1 -> do+ auun <- liftM decodeUtf8 get+ auln <- liftM decodeUtf8 get+ aufn <- liftM decodeUtf8 get+ return $ ActiveUser auun auln aufn+ _ ->+ fail $ "Cannot decode user state: " ++ show i++instance Aggregate User where+ encodeAggregate = encode+ decodeAggregate s =+ case decode s of+ Left e -> error e+ Right a -> a++instance Default User where+ def = UncreatedUser+++-- Work Units (NOT aggregates!)+data WorkUnit = WorkUnit { workUnitId :: GUID+ , workUnitComment :: Text+ , workUnitDuration :: Integer+ , workUnitUser :: UserId+ }+ deriving (Eq, Show)++instance Serialize WorkUnit where+ put (WorkUnit wuid wuc wud wuu) = do+ put $ wuid+ put $ encodeUtf8 wuc+ put wud+ put wuu+ get = do+ wuid <- get+ wuc <- liftM decodeUtf8 get+ wud <- get+ wuu <- get+ return $ WorkUnit wuid wuc wud wuu
src/Events.hs view
@@ -4,21 +4,26 @@ import Aggregates import Control.Monad (liftM)+import Data.CQRS (GUID) import Data.Serialize (Serialize(..)) import Data.Text (Text) import Data.Text.Encoding (decodeUtf8, encodeUtf8)+import Data.Time.Calendar (Day) import Data.Typeable (Typeable) import Data.Word (Word8) -data Event = ProjectCreated Text+data Event = ProjectCreated Text Text | ProjectRenamed Text | TaskAdded ProjectId Text+ | RecordedWorkUnit GUID Day Integer Text UserId+ | UserCreated Text Text Text deriving (Typeable, Show) instance Serialize Event where- put (ProjectCreated name) = do+ put (ProjectCreated name shortDesc) = do put (0 :: Word8) put $ encodeUtf8 name+ put $ encodeUtf8 shortDesc put (ProjectRenamed name) = do put (1 :: Word8) put $ encodeUtf8 name@@ -26,12 +31,25 @@ put (2 :: Word8) put $ pid put $ encodeUtf8 tsd+ put (RecordedWorkUnit wuId wuDay wuDuration wuComment wuUserId) = do+ put (3 :: Word8)+ put $ wuId+ put $ wuDay+ put $ wuDuration+ put $ encodeUtf8 wuComment+ put wuUserId+ put (UserCreated ucUserName ucFirst ucLast) = do+ put (4 :: Word8)+ put $ encodeUtf8 ucUserName+ put $ encodeUtf8 ucFirst+ put $ encodeUtf8 ucLast get = do i :: Word8 <- get case i of 0 -> do n <- liftM decodeUtf8 get- return $ ProjectCreated n+ sd <- liftM decodeUtf8 get+ return $ ProjectCreated n sd 1 -> do n <- liftM decodeUtf8 get return $ ProjectRenamed n@@ -39,6 +57,18 @@ pid <- get tsd <- liftM decodeUtf8 get return $ TaskAdded pid tsd+ 3 -> do+ wuId <- get+ wuDay <- get+ wuDuration <- get+ wuComment <- liftM decodeUtf8 get+ wuUserId <- get+ return $ RecordedWorkUnit wuId wuDay wuDuration wuComment wuUserId+ 4 -> do+ ucUserName <- liftM decodeUtf8 get+ ucFirst <- liftM decodeUtf8 get+ ucLast <- liftM decodeUtf8 get+ return $ UserCreated ucUserName ucFirst ucLast _ -> do fail $ "Unrecognized event type " ++ show i
src/Main.hs view
@@ -1,58 +1,124 @@-import Control.Concurrent (forkIO, threadDelay)-import Control.Monad.Trans.Class (lift)-import Data.CQRS-import Data.CQRS.EventStore.Backend.Sqlite3-import Data.IORef (newIORef, readIORef)-import Aggregates-import Events-import Instances ()-import Query+import Control.Concurrent (forkIO, threadDelay)+import Control.Monad (liftM)+import Control.Monad.Trans.Class (lift)+import Data.Aeson (ToJSON, encode)+import Data.CQRS+import Data.CQRS.GUID (hexEncode)+import Data.CQRS.EventStore.Backend.Sqlite3+import qualified Data.Map as M+import qualified Data.Text as T+import qualified Data.Text.Encoding as TE+import Data.Time (fromGregorian)+import Database.SQLite3 (Database)+import Aggregates+import Events+import Instances ()+import Query+import Happstack.Lite sqliteFile :: String sqliteFile = "example.db" -eventSourcingThread :: EventStore Event -> IO ()-eventSourcingThread eventStore = do- qsRef <- newIORef emptyQueryState- loop qsRef+eventSourcingThread :: Database -> EventStore Event -> IO ()+eventSourcingThread queryDb eventStore = do+ loop where- loop qsRef = do+ loop = do putStrLn "Sourcing events..."- updateQueryStateRef qsRef eventStore- qs <- readIORef qsRef- putStrLn $ " => qs: " ++ show qs+ updateQueryStateRef queryDb eventStore -- Wait 1 second threadDelay 1000000 -- Go again.- loop qsRef+ loop +-- Serve JSON ok response.+okJson :: ToJSON a => a -> ServerPart Response+okJson a = ok $ toResponseBS "text/json" $ encode a +-- Serve JSON project list.+projectsJson :: Database -> ServerPart Response+projectsJson queryDb = do+ p <- liftM (map f) $ lift $ qProjectList queryDb+ okJson $ p+ where+ f (g,pn,psd) =+ M.fromList [ (T.pack "id" , TE.decodeUtf8 $ hexEncode g)+ , (T.pack "name" , pn)+ , (T.pack "short_desc", psd)+ ]++-- Serve JSON task list.+tasksJson :: Database -> ServerPart Response+tasksJson queryDb = do+ p <- liftM (map f) $ lift $ qTaskList queryDb+ okJson $ p+ where+ f (g,tsd) =+ M.fromList [ (T.pack "id" , TE.decodeUtf8 $ hexEncode g)+ , (T.pack "short_description", tsd)+ ]++-- Serve JSON data.+jsonPart :: Database -> ServerPart Response+jsonPart queryDb =+ msum+ [ dir "projects" $ projectsJson queryDb+ , dir "tasks" $ tasksJson queryDb+ , dir "time-sheet" $ okJson ([ ] :: [String])+ ]++-- Serve.+myApp :: Database -> ServerPart Response+myApp queryDb = msum+ [ dir "js" $ serveDirectory EnableBrowsing [ ] "static/js"+ , dir "css" $ serveDirectory EnableBrowsing [ ] "static/css"+ , dir "json" $ jsonPart queryDb+ , serveDirectory EnableBrowsing [ "index.html" ] "static"+ ]++-- test1 :: IO () test1 = do+ let queryDbFile = "query.db" -- Start sourcing events. _ <- forkIO $ do- withEventStore (openSqliteEventStore sqliteFile) eventSourcingThread+ queryDb <- initializeQueryDatabase queryDbFile+ withEventStore (openSqliteEventStore sqliteFile) $ eventSourcingThread queryDb + -- Web serving thread.+ _ <- forkIO $ do+ queryDb <- initializeQueryDatabase queryDbFile+ serve Nothing $ myApp queryDb+ -- 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 $ ProjectCreated "my project"+ projectId <- lift $ newGUID+ (projectRef, _ :: Project) <- getAggregateRoot projectId+ publishEvent projectRef $ ProjectCreated "my project" "short desc of my project" -- Add a couple of tasks- taskId1 :: GUID Task <- lift $ newGUID- (taskRef1, _) <- getAggregateRoot taskId1+ taskId1 <- lift $ newGUID+ (taskRef1, _ :: Task) <- getAggregateRoot taskId1 publishEvent taskRef1 $ TaskAdded projectId "tweak knob" - taskId2 :: GUID Task <- lift $ newGUID- (taskRef2, _) <- getAggregateRoot taskId2+ taskId2 <- lift $ newGUID+ (taskRef2, _ :: Task) <- getAggregateRoot taskId2 publishEvent taskRef2 $ TaskAdded projectId "pull lever" -- Rename project. publishEvent projectRef $ ProjectRenamed "old project"++ -- Create user.+ userId1 <- lift $ newGUID+ (userRef1, _ :: User) <- getAggregateRoot userId1+ publishEvent userRef1 $ UserCreated "bardur" "Bardur" "Arantsson"++ -- Record time for task #2.+ workUnitId1 <- lift $ newGUID+ publishEvent taskRef2 $ RecordedWorkUnit workUnitId1 (fromGregorian 2011 12 01) 3 "found the lever to pull" userId1 main :: IO ()
src/Query.hs view
@@ -1,48 +1,123 @@-module Query ( emptyQueryState+module Query ( initializeQueryDatabase , updateQueryStateRef- , QueryState+ , qProjectList+ , qTaskList ) where -import Control.Monad (forM_)+import Control.Monad (forM_, liftM) import Data.ByteString (ByteString) import Data.CQRS (enumerateEventStore, GUID, EventStore)-import Data.Enumerator (run_, (>>==))+import Data.CQRS.EventStore.Backend.Sqlite3Utils (withTransaction, execSql, enumQueryResult)+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.Serialize (Serialize, decode, encode) import Data.Text (Text)+import qualified Data.Text as T+import qualified Database.SQLite3 as SQL+import Database.SQLite3 (Database, SQLData(..)) 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 ()+-- TODO: Export Data.CQRS.Serialize decode' instead?+decode' :: Serialize a => ByteString -> a+decode' = either error id . decode --- 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 }+-- Initialize the query state.+initializeQueryDatabase :: String -> IO Database+initializeQueryDatabase databaseFileName = do+ -- Open database.+ database <- SQL.open databaseFileName+ -- Create prerequisite tables if necessary.+ execSql database "CREATE TABLE IF NOT EXISTS gversion ( gversion INTEGER )" []+ execSql database "CREATE TABLE IF NOT EXISTS projects ( guid BLOB , name TEXT , short_desc TEXT )" []+ execSql database "CREATE TABLE IF NOT EXISTS tasks ( guid BLOB , project_guid BLOB , short_desc TEXT , long_desc TEXT )" []+ execSql database "CREATE TABLE IF NOT EXISTS users ( user_id BLOB , user_name TEXT , first_name TEXT , last_name TEXT )" []+ execSql database "CREATE TABLE IF NOT EXISTS work_units ( task_guid BLOB , wu_guid BLOB , wu_date TEXT , wu_duration INTEGER , wu_comment TEXT , user_id BLOB )" []+ -- Return the open database.+ return database --- 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)+-- React to event.+reactToEvent :: Database -> (Int, (GUID, Int, Event)) -> IO ()+reactToEvent database (gv, (g, _, e)) = do+ putStrLn $ show $ "gv(" ++ show gv ++ "): event: " ++ show e+ react e+ where+ react (ProjectCreated pn psd) = do+ execSql database "INSERT INTO projects ( guid , name , short_desc ) VALUES ( ? , ? , ? );" [ SQLBlob $ encode g , SQLText $ T.unpack pn , SQLText $ T.unpack psd ] --- Create "empty" data structure for querying.-emptyQueryState :: QueryState-emptyQueryState = QueryState 0 M.empty+ react (ProjectRenamed pn) = do+ execSql database "UPDATE projects SET name = ? WHERE guid = ?;" [ SQLText $ T.unpack pn , SQLBlob $ encode g ]++ react (TaskAdded pid tsd) = do+ execSql database "INSERT INTO tasks ( guid , project_guid , short_desc ) VALUES (?, ?, ?);" [ SQLBlob $ encode g, SQLBlob $ encode pid , SQLText $ T.unpack tsd ]++ react (RecordedWorkUnit wuId wuDay wuDuration wuComment wuUserId) = do+ execSql database "INSERT INTO work_units ( task_guid , wu_guid , wu_date , wu_duration , wu_comment , user_id ) VALUES ( ? , ? , ? , ? , ? , ? );"+ [ SQLBlob $ encode g+ , SQLBlob $ encode wuId+ , SQLText $ show wuDay+ , SQLInteger $ fromIntegral wuDuration+ , SQLText $ T.unpack wuComment+ , SQLBlob $ encode wuUserId+ ]++ react (UserCreated ucUserName ucFirstName ucLastName) = do+ execSql database "INSERT INTO users ( user_id , user_name , first_name , last_name ) VALUES ( ? , ? , ? , ? );"+ [ SQLBlob $ encode g+ , SQLText $ T.unpack ucUserName+ , SQLText $ T.unpack ucFirstName+ , SQLText $ T.unpack ucLastName+ ]++-- Get the global version number.+getStartVersion :: Database -> IO Int+getStartVersion queryDatabase = do+ lastVersion <-+ liftM head $+ run_ $ EL.consume >>==+ (enumQueryResult queryDatabase "SELECT COALESCE(MAX(gversion), 1) FROM gversion" [] $=+ (EL.map (\columns -> case columns of+ [ SQLInteger i ] -> fromIntegral i+ _ -> error "Invalid columns in SQL query result")))+ return $ lastVersion++updateStartVersion :: Database -> Int -> IO ()+updateStartVersion queryDatabase newStartVersion = do+ execSql queryDatabase "DELETE FROM gversion;" []+ execSql queryDatabase "INSERT INTO gversion (gversion) VALUES (?);" [ SQLInteger $ fromIntegral $ newStartVersion ]++-- Update query state with latest events.+updateQueryStateRef :: Database -> EventStore Event -> IO ()+updateQueryStateRef queryDatabase eventStore =+ withTransaction queryDatabase $ do+ -- Find the version number to start at.+ startVersion <- getStartVersion queryDatabase+ -- Enumerate all the events form the event store since last run.+ evs <- run_ (EL.consume >>== enumerateEventStore eventStore startVersion)+ -- React to all the events.+ forM_ evs $ reactToEvent queryDatabase+ -- Update the global version number.+ updateStartVersion queryDatabase $ startVersion + length evs+ -- Done.+ return ()++-- Query project list+qProjectList :: Database -> IO [(GUID, Text, Text)]+qProjectList queryDatabase = do+ run_ $ EL.consume >>==+ (enumQueryResult queryDatabase "SELECT guid, name, short_desc FROM projects" [] $=+ (EL.map (\columns -> case columns of+ [ SQLBlob guid, SQLText name, SQLText short_desc ] ->+ (decode' guid , T.pack name, T.pack short_desc)+ _ ->+ error "Invalid columns in SQL query result")))++-- Query task list+qTaskList :: Database -> IO [(GUID, Text)]+qTaskList queryDatabase = do+ run_ $ EL.consume >>==+ (enumQueryResult queryDatabase "SELECT guid, short_desc FROM tasks;" [] $=+ (EL.map (\columns -> case columns of+ [ SQLBlob guid, SQLText short_desc ] ->+ (decode' guid, T.pack short_desc)+ _ ->+ error "Invalid columns in SQL query result")))