cqrs 0.4.0 → 0.5.0

raw patch · 11 files changed

+411/−323 lines, 11 filesdep +enumeratorPVP ok

version bump matches the API change (PVP)

Dependencies added: enumerator

API changes (from Hackage documentation)

Files

• cqrs.cabal +7/−3
• src/Data/CQRS.hs +2/−0
• src/Data/CQRS/EventStore.hs +4/−45
• src/Data/CQRS/EventStore/Backend.hs +47/−0
• src/Data/CQRS/EventStore/Backend/Sqlite3.hs +229/−0
• src/Data/CQRS/EventStore/Sqlite3.hs +0/−208
• src/Data/CQRS/GUID.hs +2/−12
• src/Data/CQRS/Internal/AggregateRef.hs +17/−14
• src/Data/CQRS/Internal/EventStore.hs +69/−0
• src/Data/CQRS/Serialize.hs +10/−0
• src/Data/CQRS/Transaction.hs +24/−41

@@ -1,5 +1,5 @@ Name:                cqrs-Version:             0.4.0+Version:             0.5.0 Synopsis:            Command-Query Responsibility Segregation Description:         Haskell implementation of the CQRS architectural pattern.   An SQLite3-based backend is included.@@ -16,6 +16,7 @@                , bytestring >= 0.9.0.1                , cereal >= 0.3.3 && < 0.4                , containers >= 0.4+               , enumerator >= 0.4.15 && < 0.5                , data-default >= 0.3 && < 0.4                , direct-sqlite >= 1.1 && < 1.2                , random >= 1.0 && < 1.1@@ -35,7 +36,10 @@                        Data.CQRS.AggregateRef                        Data.CQRS.Eventable                        Data.CQRS.EventStore-                       Data.CQRS.EventStore.Sqlite3+                       Data.CQRS.EventStore.Backend+                       Data.CQRS.EventStore.Backend.Sqlite3                        Data.CQRS.GUID                        Data.CQRS.Transaction-  Other-modules:       Data.CQRS.Internal.AggregateRef+  Other-modules:       Data.CQRS.Serialize+                       Data.CQRS.Internal.AggregateRef+                       Data.CQRS.Internal.EventStore

@@ -4,6 +4,7 @@         ( module Data.CQRS.Aggregate         , module Data.CQRS.AggregateRef         , module Data.CQRS.Eventable+        , module Data.CQRS.EventStore         , module Data.CQRS.GUID         , module Data.CQRS.Transaction         ) where@@ -11,5 +12,6 @@ import Data.CQRS.Aggregate import Data.CQRS.AggregateRef import Data.CQRS.Eventable+import Data.CQRS.EventStore import Data.CQRS.GUID(GUID, newGUID) import Data.CQRS.Transaction

@@ -1,49 +1,8 @@--- | Event store data types.+-- | Event store functions. module Data.CQRS.EventStore-       ( EventStore(..)+       ( EventStore+       , enumerateEventStore        , withEventStore        ) where -import Control.Exception (bracket)-import Data.ByteString (ByteString)-import Data.CQRS.GUID---- | Event stores are the backend used for reading and storing all the--- information about recorded events.-data EventStore =-  EventStore {-    -- | Store a sequence of events for aggregate identified by GUID-    -- into the event store, starting at the provided version number.-    -- If the version number does not match the expected value, a-    -- failure occurs.-    storeEvents :: forall a. GUID a -> Int -> [ByteString] -> IO (),-    -- | Retrieve the sequence of events associated with the aggregate-    -- identified by the given GUID. Only events at or after the given-    -- version number are retrieved. The events are returned in-    -- increasing order of version number. The version number of the-    -- last event is returned as well.-    retrieveEvents :: forall a. GUID a -> Int -> IO (Int,[ByteString]),-    -- | Read events and pass them to the provided monadic action.-    readAllEvents :: Int -> Int -> (ByteString -> IO ()) -> IO (),-    -- | Write snapshot for aggregate identified by GUID and-    -- the given version number. The version number is NOT checked-    -- for validity. If the event store does not support snapshots-    -- this function may do nothing.-    writeSnapshot :: forall a. GUID a -> (Int,ByteString) -> IO (),-    -- | Get latest snapshot of an aggregate identified by GUID.-    -- Returns the version number of the snapshot in addition to the-    -- data. An event store which does not support snapshots is-    -- permitted to return 'Nothing' in all cases.-    getLatestSnapshot :: forall a. GUID a -> IO (Maybe (Int,ByteString)),-    -- | Run transaction against the event store. The transaction is-    -- expected to commit if the supplied IO action runs to completion-    -- (i.e. doesn't throw an exception) and to rollback otherwise.-    withTransaction :: forall a . IO a -> IO a,-    -- | Close the event store.-    closeEventStore :: IO ()-    }---- | Perform an IO action with an open event store.-withEventStore :: (IO EventStore) -> (EventStore -> IO a) -> IO a-withEventStore open action = bracket open closeEventStore action-+import Data.CQRS.Internal.EventStore

@@ -0,0 +1,47 @@+-- | Event store backend. You only need to import this+-- module if you're planning on implementing a custom+-- event store backend.+module Data.CQRS.EventStore.Backend+       ( EventStoreBackend(..)+       ) where++import Data.ByteString (ByteString)+import Data.CQRS.GUID+import Data.Enumerator (Enumerator)++-- | Event stores are the backend used for reading and storing all the+-- information about recorded events.+data EventStoreBackend =+  EventStoreBackend {+    -- | Store a sequence of events for aggregate identified by GUID+    -- into the event store, starting at the provided version number.+    -- If the version number does not match the expected value, a+    -- failure occurs.+    esbStoreEvents :: forall a. GUID a -> Int -> [(ByteString,Int)] -> IO (),+    -- | Retrieve the sequence of events associated with the aggregate+    -- identified by the given GUID. Only events at or after the given+    -- version number are retrieved. The events are returned in+    -- increasing order of version number. The version number of the+    -- last event is returned as well.+    esbRetrieveEvents :: forall a. GUID a -> Int -> IO (Int,[ByteString]),+    -- | Enumerate all events later than given logical timestamp.+    esbEnumerateAllEvents :: forall a b. Int -> Enumerator (Int, (GUID a, Int, ByteString)) IO b,+    -- | Write snapshot for aggregate identified by GUID and+    -- the given version number. The version number is NOT checked+    -- for validity. If the event store does not support snapshots+    -- this function may do nothing.+    esbWriteSnapshot :: forall a. GUID a -> (Int,ByteString) -> IO (),+    -- | Get latest snapshot of an aggregate identified by GUID.+    -- Returns the version number of the snapshot in addition to the+    -- data. An event store which does not support snapshots is+    -- permitted to return 'Nothing' in all cases.+    esbGetLatestSnapshot :: forall a. GUID a -> IO (Maybe (Int,ByteString)),+    -- | Get the latest version global number.+    esbGetLatestVersion :: IO Int,+    -- | Run transaction against the event store. The transaction is+    -- expected to commit if the supplied IO action runs to completion+    -- (i.e. doesn't throw an exception) and to rollback otherwise.+    esbWithTransaction :: forall a . IO a -> IO a,+    -- | Close the event store.+    esbCloseEventStoreBackend :: IO ()+    }

@@ -0,0 +1,229 @@+{-| Implementation of an SQLite3-based event store. -}+module Data.CQRS.EventStore.Backend.Sqlite3+       ( openSqliteEventStore+       ) where++import           Control.Exception (catch, bracket, finally, onException, SomeException)+import           Control.Monad (when, forM_, liftM)+import           Data.ByteString (ByteString)+import           Data.CQRS.EventStore.Backend (EventStoreBackend(..))+import           Data.CQRS.GUID (GUID)+import           Data.CQRS.Serialize (decode')+import           Data.Enumerator (Enumerator, Iteratee(..), tryIO, continue, (>>==), ($=), run_, Stream(..))+import qualified Data.Enumerator.List as EL+import           Data.Enumerator.Internal (checkContinue0)+import           Data.Serialize (encode)+import qualified Database.SQLite3 as SQL+import           Database.SQLite3 (Database, Statement, SQLData(..), StepResult(..))+import           Prelude hiding (catch)++-- Convenience class for converting values to SQLData.+class ToSQLData a where+  toSQLData :: a -> SQLData++instance ToSQLData (GUID a) where+  toSQLData = SQLBlob . encode++instance ToSQLData Int where+  toSQLData = SQLInteger . fromIntegral++instance ToSQLData ByteString where+  toSQLData = SQLBlob++-- SQL+createEventsSql :: String+createEventsSql = "CREATE TABLE IF NOT EXISTS events ( guid BLOB , ev_data BLOB , version INTEGER , gversion INTEGER, PRIMARY KEY (guid, version) );"++selectEventsSql :: String+selectEventsSql = "SELECT version, ev_data FROM events WHERE guid = ? AND version > ? ORDER BY version ASC;"++enumerateAllEventsSql :: String+enumerateAllEventsSql = "SELECT gversion, guid, version, ev_data FROM events WHERE gversion >= ? ORDER BY gversion ASC;"++insertEventSql :: String+insertEventSql = "INSERT INTO events ( guid, version, ev_data, gversion ) VALUES (?, ?, ?, ?);"++createAggregateVersionsSql :: String+createAggregateVersionsSql = "CREATE TABLE IF NOT EXISTS versions ( guid BLOB PRIMARY KEY , version INTEGER );"++getCurrentVersionSql :: String+getCurrentVersionSql = "SELECT version FROM versions WHERE guid = ?;"++getLatestVersionSql :: String+getLatestVersionSql = "SELECT COALESCE(MAX(gversion), 0) FROM events;"++updateCurrentVersionSql :: String+updateCurrentVersionSql = "INSERT OR REPLACE INTO versions ( guid, version ) VALUES (?,?);"++createSnapshotSql :: String+createSnapshotSql = "CREATE TABLE IF NOT EXISTS snapshots ( guid BLOB PRIMARY KEY , data BLOB , version INTEGER );"++writeSnapshotSql :: String+writeSnapshotSql = "INSERT OR REPLACE INTO snapshots ( guid , data, version ) VALUES ( ?, ?, ? );"++selectSnapshotSql :: String+selectSnapshotSql = "SELECT data, version FROM snapshots WHERE guid = ?;"++beginTransaction :: Database -> IO ()+beginTransaction database = execSql database "BEGIN TRANSACTION;" []++commitTransaction :: Database -> IO ()+commitTransaction database = execSql database "COMMIT TRANSACTION;" []++rollbackTransaction :: Database -> IO ()+rollbackTransaction database = execSql database "ROLLBACK TRANSACTION;" []++withSqlStatement :: Database -> String -> [SQLData] -> (Statement -> IO a) -> IO a+withSqlStatement database sql parameters action =+  bracket (SQL.prepare database sql) SQL.finalize $ \statement -> do+    SQL.bind statement parameters+    action statement++execSql :: Database -> String -> [SQLData] -> IO ()+execSql database sql parameters =+  withSqlStatement database sql parameters $ \stmt -> do+    _ <- SQL.step stmt+    return ()++enumQueryResults :: Statement -> Enumerator [SQLData] IO b+enumQueryResults stmt = checkContinue0 $ \loop k ->+  do+    nextResult <- tryIO $ SQL.step stmt+    case nextResult of+      Done -> continue k+      Row -> do+        cols <- tryIO $ SQL.columns stmt+        k (Chunks [cols]) >>== loop++enumQueryResult :: Database -> String -> [SQLData] -> Enumerator [SQLData] IO b+enumQueryResult database sql parameters step = do+  stmt <- tryIO $ SQL.prepare database sql+  Iteratee $ finally+    (do+        SQL.bind stmt parameters+        runIteratee $ enumQueryResults stmt step)+    (SQL.finalize stmt)++badQueryResultMsg :: [String] -> [SQLData] -> String+badQueryResultMsg params columns = concat ["Invalid query result shape. Params: ", show params, ". Result columns: ", show columns]++versionConflict :: (Show a, Show b) => a -> b -> IO c+versionConflict ov cv =+  fail $ concat [ "Version conflict detected (expected ", show ov+                , ", saw ", show cv, ")"+                ]++storeEvents :: forall a. Database -> GUID a -> Int -> [(ByteString,Int)] -> IO ()+storeEvents database guid originatingVersion events = do+  -- Column unpacking.+  let unpackColumns [ SQLInteger v ] = v+      unpackColumns columns = error $ badQueryResultMsg [show guid] columns+  -- Get the current version number of the aggregate.+  curVer <- run_ $ EL.fold (\x -> max x . unpackColumns) 0 >>==+              (enumQueryResult database getCurrentVersionSql [toSQLData guid])++  -- Sanity check current version number.+  when (fromIntegral curVer /= originatingVersion) $+    versionConflict originatingVersion curVer++  -- Update de-normalized version number.+  execSql database updateCurrentVersionSql+    [ toSQLData guid+    , toSQLData $ originatingVersion + length events+    ]++  -- Store the supplied events.+  forM_ (zip [1 + originatingVersion..] events) $ \(v,(e,gv)) -> do+    execSql database insertEventSql+      [ toSQLData guid+      , toSQLData v+      , toSQLData e+      , toSQLData gv+      ]++retrieveEvents :: Database -> GUID a -> Int -> IO (Int,[ByteString])+retrieveEvents database guid v0 = do+  -- Unpack the columns into tuples.+  let unpackColumns [SQLInteger version, SQLBlob eventData] = (version, eventData)+      unpackColumns columns = error $ badQueryResultMsg [show guid, show v0] columns+  -- Find events with version numbers.+  results <-+    run_ $ EL.consume >>==+    (enumQueryResult database selectEventsSql [toSQLData guid, toSQLData v0] $=+     (EL.map unpackColumns))+  -- Find the max version number.+  let maxVersion = maximum $ (:) (fromIntegral v0) $ map fst results+  return (fromIntegral maxVersion, map snd results)++enumerateAllEvents :: forall a b. Database -> Int -> Enumerator (Int,(GUID a, Int, ByteString)) IO b+enumerateAllEvents database minVersion = do+  enumQueryResult database enumerateAllEventsSql [toSQLData minVersion] $= EL.map+    (\columns -> do+        case columns of+          [ SQLInteger gv, SQLBlob g, SQLInteger v, SQLBlob ed ] ->+            ( fromIntegral gv, (decode' g, fromIntegral v, ed) )+          _ ->+            error $ badQueryResultMsg [show minVersion] columns)++writeSnapshot :: Database -> GUID a -> (Int, ByteString) -> IO ()+writeSnapshot database guid (v,a) = do+  execSql database writeSnapshotSql+    [ toSQLData guid+    , toSQLData a+    , toSQLData v+    ]++getLatestSnapshot :: Database -> GUID a -> IO (Maybe (Int, ByteString))+getLatestSnapshot database guid = do+  -- Unpack columns from result.+  let unpackColumns :: [SQLData] -> Maybe (Int,ByteString)+      unpackColumns [SQLBlob a, SQLInteger v] = Just (fromIntegral v, a)+      unpackColumns columns                    = error $ badQueryResultMsg [show guid] columns+  -- Run the query.+  run_ $ EL.fold const Nothing >>==+    (enumQueryResult database selectSnapshotSql [toSQLData guid] $= (EL.map unpackColumns))++getLatestVersion :: Database -> IO Int+getLatestVersion database = do+  -- Unpack columns from result.+  let unpackColumns :: [SQLData] -> Int+      unpackColumns [SQLInteger v] = fromIntegral v+      unpackColumns columns        = error $ badQueryResultMsg [] columns+  -- Run the query.+  liftM head $ run_ $ EL.consume >>== (enumQueryResult database getLatestVersionSql [] $= (EL.map unpackColumns))++withTransaction :: forall a . Database -> IO a -> IO a+withTransaction database action = do+  beginTransaction database+  onException runAction tryRollback+  where+    runAction = do+      r <- action+      commitTransaction database+      return r++    tryRollback =+      -- Try rollback while discarding exception; we want to preserve+      -- original exception.+      catch (rollbackTransaction database) (\(_::SomeException) -> return ())++-- | Open an SQLite3-based event store using the named SQLite database file.+-- The database file is created if it does not exist.+openSqliteEventStore :: String -> IO EventStoreBackend+openSqliteEventStore databaseFileName = do+  -- Create the database.+  database <- SQL.open databaseFileName+  -- Set up tables.+  execSql database createEventsSql []+  execSql database createAggregateVersionsSql []+  execSql database createSnapshotSql []+  -- Return event store.+  return $ EventStoreBackend { esbStoreEvents = storeEvents database+                             , esbRetrieveEvents = retrieveEvents database+                             , esbEnumerateAllEvents = enumerateAllEvents database+                             , esbWriteSnapshot = writeSnapshot database+                             , esbGetLatestSnapshot = getLatestSnapshot database+                             , esbGetLatestVersion = getLatestVersion database+                             , esbWithTransaction = withTransaction database+                             , esbCloseEventStoreBackend = SQL.close database+                             }

@@ -1,208 +0,0 @@-{-| Implementation of an SQLite3-based event store. -}-module Data.CQRS.EventStore.Sqlite3-       ( openSqliteEventStore-       , closeEventStore -- Re-export for convenience-       ) where--import           Control.Exception (catch, bracket, onException, SomeException)-import           Control.Monad (when, forM_)-import           Data.ByteString (ByteString)-import           Data.CQRS.EventStore-import           Data.CQRS.GUID (GUID)-import qualified Data.CQRS.GUID as G-import qualified Database.SQLite3 as SQL-import           Database.SQLite3 (Database, Statement, SQLData(..), StepResult(..))-import           Prelude hiding (catch)---- Convenience class for converting values to SQLData.-class ToSQLData a where-  toSQLData :: a -> SQLData--instance ToSQLData (GUID a) where-  toSQLData = SQLBlob . G.toByteString--instance ToSQLData Int where-  toSQLData = SQLInteger . fromIntegral--instance ToSQLData ByteString where-  toSQLData = SQLBlob---- SQL-createEventsSql :: String-createEventsSql = "CREATE TABLE IF NOT EXISTS events ( guid BLOB , ev_data BLOB , version INTEGER , gversion INTEGER, PRIMARY KEY (guid, version) );"--selectEventsSql :: String-selectEventsSql = "SELECT version, ev_data FROM events WHERE guid = ? AND version > ? ORDER BY version ASC;"--selectAllEventsSql :: String-selectAllEventsSql = "SELECT version, ev_data FROM events WHERE gversion >= ? AND gversion < ? ORDER BY gversion ASC;"--insertEventSql :: String-insertEventSql = "INSERT INTO events ( guid, version, ev_data, gversion ) VALUES (?, ?, ?, COALESCE((SELECT MAX(gversion) FROM events ), 0) + 1);"--createAggregateVersionsSql :: String-createAggregateVersionsSql = "CREATE TABLE IF NOT EXISTS versions ( guid BLOB PRIMARY KEY , version INTEGER );"--getCurrentVersionSql :: String-getCurrentVersionSql = "SELECT version FROM versions WHERE guid = ?;"--updateCurrentVersionSql :: String-updateCurrentVersionSql = "INSERT OR REPLACE INTO versions ( guid, version ) VALUES (?,?);"--createSnapshotSql :: String-createSnapshotSql = "CREATE TABLE IF NOT EXISTS snapshots ( guid BLOB PRIMARY KEY , data BLOB , version INTEGER );"--writeSnapshotSql :: String-writeSnapshotSql = "INSERT OR REPLACE INTO snapshots ( guid , data, version ) VALUES ( ?, ?, ? );"--selectSnapshotSql :: String-selectSnapshotSql = "SELECT data, version FROM snapshots WHERE guid = ?;"--beginTransaction :: Database -> IO ()-beginTransaction database = execSql database "BEGIN TRANSACTION;" []--commitTransaction :: Database -> IO ()-commitTransaction database = execSql database "COMMIT TRANSACTION;" []--rollbackTransaction :: Database -> IO ()-rollbackTransaction database = execSql database "ROLLBACK TRANSACTION;" []--withSqlStatement :: Database -> String -> [SQLData] -> (Statement -> IO a) -> IO a-withSqlStatement database sql parameters action =-  bracket (SQL.prepare database sql) SQL.finalize $ \statement -> do-    SQL.bind statement parameters-    action statement--execSql :: Database -> String -> [SQLData] -> IO ()-execSql database sql parameters =-  withSqlStatement database sql parameters $ \stmt -> do-    _ <- SQL.step stmt-    return ()--querySql :: Database -> String -> [SQLData] -> a -> ([SQLData] -> a -> IO a) -> IO a-querySql database sql parameters a0 reader =-  withSqlStatement database sql parameters go-  where-    go statement = loop a0-      where-        loop acc = do-          res <- SQL.step statement-          case res of-            Done -> return acc-            Row -> do-              cols <- SQL.columns statement-              acc' <- reader cols acc-              loop acc'--badQueryResult :: [String] -> [SQLData] -> IO b-badQueryResult params columns =-  fail $ concat ["Invalid query result shape. Params: ", show params, ". Result columns: ", show columns]--versionConflict :: (Show a, Show b) => a -> b -> IO c-versionConflict ov cv =-  fail $ concat [ "Version conflict detected (expected ", show ov-                , ", saw ", show cv, ")"-                ]--storeEvents_ :: forall a. Database -> GUID a -> Int -> [ByteString] -> IO ()-storeEvents_ database guid originatingVersion events = do-  -- Get the current version number of the aggregate.-  versions <- querySql database getCurrentVersionSql [toSQLData guid] [] $ \columns acc ->-    case columns of-      [ SQLInteger v ] -> return (v:acc)-      _ -> badQueryResult [show guid] columns-  let curVer = maximum (0 : versions)--  -- Sanity check current version number.-  when (fromIntegral curVer /= originatingVersion) $-    versionConflict originatingVersion curVer--  -- Update de-normalized version number.-  execSql database updateCurrentVersionSql-    [ toSQLData guid-    , toSQLData $ originatingVersion + length events-    ]--  -- Store the supplied events.-  forM_ (zip [1 + originatingVersion..] events) $ \(v,e) -> do-    execSql database insertEventSql-      [ toSQLData guid-      , toSQLData v-      , toSQLData e-      ]--retrieveEvents_ :: Database -> GUID a -> Int -> IO (Int,[ByteString])-retrieveEvents_ database guid v0 = do-  -- Find events.-  results <- fmap reverse $ querySql database selectEventsSql [toSQLData guid, toSQLData v0] [] $ \columns acc -> do-    case columns of-      [SQLInteger version, SQLBlob eventData] ->-        return $ (version, eventData) : acc-      _ ->-        badQueryResult [show guid] columns--  -- Find the max version number.-  let maxVersion = maximum $ (:) (fromIntegral v0) $ map fst results-  return (fromIntegral maxVersion, map snd results)--readAllEvents_ :: Database -> Int -> Int -> (ByteString -> IO ()) -> IO ()-readAllEvents_ database minVersion maxVersion handler = do-  querySql database selectAllEventsSql [toSQLData minVersion, toSQLData maxVersion] undefined $ \columns _ -> do-    case columns of-      [SQLInteger version, SQLBlob eventData] -> do-        handler eventData-      _ ->-        badQueryResult [] columns-  return ()--writeSnapshot_ :: Database -> GUID a -> (Int, ByteString) -> IO ()-writeSnapshot_ database guid (v,a) = do-  execSql database writeSnapshotSql-    [ toSQLData guid-    , toSQLData a-    , toSQLData v-    ]--getLatestSnapshot_ :: Database -> GUID a -> IO (Maybe (Int, ByteString))-getLatestSnapshot_ database guid = do-  querySql database selectSnapshotSql [toSQLData guid] Nothing $ \columns _ -> do-    case columns of-      [SQLBlob aggData, SQLInteger version] ->-        return $ Just (fromIntegral version, aggData)-      _ ->-        badQueryResult [show guid] columns--withTransaction_ :: forall a . Database -> IO a -> IO a-withTransaction_ database action = do-  beginTransaction database-  onException runAction tryRollback-  where-    runAction = do-      r <- action-      commitTransaction database-      return r--    tryRollback =-      -- Try rollback while discarding exception; we want to preserve-      -- original exception.-      catch (rollbackTransaction database) (\(_::SomeException) -> return ())---- | Open an SQLite3-based event store using the named SQLite database file.--- The database file is created if it does not exist.-openSqliteEventStore :: String -> IO EventStore-openSqliteEventStore databaseFileName = do-  -- Create the database.-  database <- SQL.open databaseFileName-  -- Set up tables.-  execSql database createEventsSql []-  execSql database createAggregateVersionsSql []-  execSql database createSnapshotSql []-  -- Return event store.-  return $ EventStore { storeEvents = storeEvents_ database-                      , retrieveEvents = retrieveEvents_ database-                      , readAllEvents = readAllEvents_ database-                      , writeSnapshot = writeSnapshot_ database-                      , getLatestSnapshot = getLatestSnapshot_ database-                      , withTransaction = withTransaction_ database-                      , closeEventStore = SQL.close database-                      }

@@ -1,9 +1,7 @@ -- | Globally Unique IDentifiers. module Data.CQRS.GUID        ( GUID-       , fromByteString        , newGUID-       , toByteString        ) where  import           Control.Monad (liftM)@@ -34,15 +32,7 @@   uuid <- sequence $ replicate 32 randomWord8   return $ GUID $ B.pack uuid --- | Convert GUID to ByteString.-toByteString :: GUID a -> ByteString-toByteString (GUID s) = s---- | Convert ByteString to GUID.-fromByteString :: ByteString -> GUID a-fromByteString s = GUID s- -- | Serialize instance. instance Serialize (GUID a) where-    put = put . toByteString-    get = liftM fromByteString get+    put (GUID s) = put s+    get = liftM GUID get

@@ -10,17 +10,20 @@        , readValue        ) where -import Control.Monad (liftM)-import Control.Monad.IO.Class (MonadIO, liftIO)-import Data.CQRS.GUID (GUID)-import Data.CQRS.Eventable (Eventable(..))-import Data.IORef (IORef, modifyIORef, newIORef, readIORef)-import Data.Typeable (Typeable)+import           Control.Monad (liftM)+import           Control.Monad.IO.Class (MonadIO, liftIO)+import           Data.CQRS.GUID (GUID)+import           Data.CQRS.Eventable (Eventable(..))+import           Data.Foldable (toList)+import           Data.IORef (IORef, modifyIORef, newIORef, readIORef)+import           Data.Sequence (Seq, (|>))+import qualified Data.Sequence as S+import           Data.Typeable (Typeable)  -- | Aggregate root reference. data AggregateRef a e =   AggregateRef { arValue :: IORef a-               , arEvents :: IORef [e]+               , arEvents :: IORef (Seq (e, Int))                , arGUID :: GUID a                , arStartVersion :: Int                , arSnapshotVersion :: Int@@ -31,20 +34,20 @@ mkAggregateRef :: (MonadIO m) => a -> GUID a -> Int -> Int -> m (AggregateRef a e) mkAggregateRef a guid originatingVersion snapshotVersion = do   a' <- liftIO $ newIORef a-  e' <- liftIO $ newIORef []+  e' <- liftIO $ newIORef S.empty   return $ AggregateRef a' e' guid originatingVersion snapshotVersion  -- | Publish event to aggregate.-publishEvent :: (MonadIO m, Eventable a e) => AggregateRef a e -> e -> m ()-publishEvent aggregateRef event = liftIO $ do+publishEvent :: (MonadIO m, Eventable a e) => AggregateRef a e -> e -> Int -> m ()+publishEvent aggregateRef event gv = liftIO $ do   -- Apply event to aggregate state.   modifyIORef (arValue aggregateRef) $ applyEvent event   -- Add event to aggregate.-  modifyIORef (arEvents aggregateRef) $ \es -> event:es+  modifyIORef (arEvents aggregateRef) $ \events -> events |> (event,gv)  -- | Read aggregate events.-readEvents :: (MonadIO m) => AggregateRef a e -> m [e]-readEvents = liftIO . readIORef . arEvents+readEvents :: (MonadIO m) => AggregateRef a e -> m [(e,Int)]+readEvents = liftM toList . liftIO . readIORef . arEvents  -- | Read aggregate state. readValue :: (MonadIO m) => AggregateRef a e -> m a@@ -53,5 +56,5 @@ -- | Get the current version of the aggregate in aggregate ref. getCurrentVersion :: (MonadIO m) => AggregateRef a e -> m Int getCurrentVersion a = do-  nevs <- liftM length $ liftIO $ readIORef $ arEvents a+  nevs <- liftM S.length $ liftIO $ readIORef $ arEvents a   return $ nevs + (arStartVersion a)

@@ -0,0 +1,69 @@+-- | Event store functions.+module Data.CQRS.Internal.EventStore+       ( EventStore+       , enumerateAllEvents+       , enumerateEventStore+       , getLatestSnapshot+       , getLatestVersion+       , retrieveEvents+       , storeEvents+       , withEventStore+       , withTransaction+       , writeSnapshot+       ) where++import           Control.Exception (bracket)+import           Control.Monad (liftM)+import           Data.ByteString (ByteString)+import           Data.CQRS.EventStore.Backend (EventStoreBackend(..))+import           Data.CQRS.GUID+import           Data.CQRS.Serialize (decode')+import           Data.Enumerator (Enumerator, ($=))+import qualified Data.Enumerator.List as EL+import           Data.Serialize (Serialize, encode)++-- Utilities.+map1st :: (a -> c) -> (a, b) -> (c, b)+map1st f (a,b) = (f a, b)++map2nd :: (b -> c) -> (a,b) -> (a,c)+map2nd f (a,b) = (a, f b)++map3rd :: (c -> d) -> (a,b,c) -> (a,b,d)+map3rd f (a,b,c) = (a,b,f c)++-- Provide a type alias.+data EventStore e = EventStore { esBackend :: EventStoreBackend+                               }++-- | Perform an IO action with an open event store.+withEventStore :: (IO EventStoreBackend) -> (EventStore e -> IO a) -> IO a+withEventStore open action = bracket (liftM EventStore open) (esbCloseEventStoreBackend . esBackend) action++-- | Enumerate all the events from an event store that occur at or later+-- than a given logical timestamp.+enumerateEventStore :: forall a b e . (Serialize e) => EventStore e -> Int -> Enumerator (Int, (GUID a, Int, e)) IO b+enumerateEventStore es minVersion =+  esbEnumerateAllEvents (esBackend es) minVersion $= EL.map (\(gv,(g,ev,ed)) -> (gv,(g,ev,decode' ed :: e)))++withTransaction :: EventStore e -> IO a -> IO a+withTransaction (EventStore esb) = esbWithTransaction esb++storeEvents :: Serialize e => EventStore e -> GUID a -> Int -> [(e,Int)] -> IO ()+storeEvents (EventStore esb) guid v0 evs = esbStoreEvents esb guid v0 (map (map1st encode) evs)++retrieveEvents :: Serialize e => EventStore e -> GUID a -> Int -> IO (Int,[e])+retrieveEvents (EventStore esb) guid v0 = liftM (map2nd $ map decode') $ esbRetrieveEvents esb guid v0++enumerateAllEvents :: forall a b e. Serialize e => EventStore e -> Int -> Enumerator (Int, (GUID a, Int, e)) IO b+enumerateAllEvents (EventStore esb) v0 =+  esbEnumerateAllEvents esb v0 $= EL.map (map2nd $ map3rd decode')++writeSnapshot ::   EventStore e -> GUID a -> (Int, ByteString) -> IO ()+writeSnapshot (EventStore esb) = esbWriteSnapshot esb++getLatestSnapshot :: EventStore e -> GUID a -> IO (Maybe (Int, ByteString))+getLatestSnapshot (EventStore esb) = esbGetLatestSnapshot esb++getLatestVersion :: EventStore e -> IO Int+getLatestVersion (EventStore esb) = esbGetLatestVersion esb

@@ -0,0 +1,10 @@+module Data.CQRS.Serialize+       ( decode'+       ) where++import Data.ByteString (ByteString)+import Data.Serialize (Serialize, decode)++-- | Decode a serialized value "irrefutably".+decode' :: Serialize e => ByteString -> e+decode' = either error id . decode

@@ -3,25 +3,23 @@        ( TransactionT        , getAggregateRoot        , publishEvent-       , retrieveEvents        , runTransactionT        ) where -import           Control.Monad (forM_, when)+import           Control.Monad (forM_, when, liftM) import           Control.Monad.IO.Class (MonadIO, liftIO) import           Control.Monad.Trans.Class (MonadTrans(..), lift)-import           Control.Monad.Trans.State (StateT, get, modify, runStateT)-import           Data.ByteString (ByteString)+import           Control.Monad.Trans.State (StateT, get, gets, modify, runStateT) import           Data.CQRS.Aggregate (Aggregate(..)) import           Data.CQRS.Internal.AggregateRef (AggregateRef, mkAggregateRef) import qualified Data.CQRS.Internal.AggregateRef as AR import           Data.CQRS.Eventable (Eventable(..))-import           Data.CQRS.EventStore (EventStore, withTransaction)-import qualified Data.CQRS.EventStore as ES+import           Data.CQRS.EventStore (EventStore)+import qualified Data.CQRS.Internal.EventStore as ES import           Data.CQRS.GUID (GUID) import           Data.Default (Default(..)) import           Data.List (find)-import           Data.Serialize (Serialize, decode, encode)+import           Data.Serialize (Serialize) import           Data.Typeable (Typeable, cast)  -- | Transaction monad transformer.@@ -39,25 +37,24 @@ type Transaction e = StateT (TransactionState e)  data TransactionState e =-  TransactionState { eventStore :: EventStore+  TransactionState { eventStore :: EventStore e                    , aggregateRefsToCommit :: [BoxedAggregateRef e]+                   , tsCurrentVersion :: Int                    } --- | Decode a serialized value "irrefutably".-decode' :: Serialize e => ByteString -> e-decode' = either error id . decode- -- | Run transaction against an event store.-runTransactionT :: (Typeable e, Serialize e) => EventStore -> TransactionT e IO c -> IO c+runTransactionT :: (Typeable e, Serialize e) => EventStore e -> TransactionT e IO c -> IO c runTransactionT eventStore_ (TransactionT transaction) = do-  withTransaction eventStore_ $ do+  ES.withTransaction eventStore_ $ do+    -- Get the latest global version number to use.+    gv0 <- liftM ((+) 1) $ ES.getLatestVersion eventStore_     -- Run the computation.-    (r,s) <- runStateT transaction s0-    -- Write out all the accumulated aggregates+    (r,s) <- runStateT transaction $ s0 gv0+    -- Write out all the aggregates.     forM_ (aggregateRefsToCommit s) $ \(BoxedAggregateRef a) -> do-      es <- AR.readEvents a-      ES.storeEvents eventStore_ (AR.arGUID a) (AR.arStartVersion a)-        (map encode es)+      -- Write out accumulated events.+      evs <- AR.readEvents a+      ES.storeEvents eventStore_ (AR.arGUID a) (AR.arStartVersion a) evs       -- If we've advanced N events past the last snapshot, we       -- create a new snapshot.       v <- AR.getCurrentVersion a@@ -110,7 +107,7 @@   -- Get events.   (latestVersion, events) <- lift $ ES.retrieveEvents es guid v0   -- Build the aggregate state from all the events.-  let a = foldr applyEvent a0 $ map (\e -> decode' e :: e) events+  let a = foldr applyEvent a0 events   -- Make the aggregate itself   (a' :: AggregateRef a e) <- lift $ mkAggregateRef a guid latestVersion v0   -- Add to set of aggregates to commit later.@@ -119,27 +116,13 @@   return a'  -- | Publish event for an aggregate root.-publishEvent :: (MonadIO m, Serialize e, Eventable a e) => AggregateRef a e -> e -> TransactionT e m ()-publishEvent aggregateRef event = lift $ AR.publishEvent aggregateRef event---- | Retrieve a range of events, invoking a callback function for each event.--- For example,------    @--- retrieveEvents 10 20 $ \\e -> do---   putStrLn $ \"Event: \" ++ show e---    @------ would print all events between "logical time" 20 (inclusive)--- and "logical time" 30 (not inclusive). "Logical time" starts--- at 1. It is NOT an error to provide bounds for which there--- are no events.-retrieveEvents :: forall e . (Serialize e) => Int -> Int -> (e -> IO ()) -> TransactionT e IO ()-retrieveEvents minVersion maxVersion h = TransactionT $ do-  es <- fmap eventStore get-  lift $ ES.readAllEvents es minVersion maxVersion $ \ed -> do-    let e :: e = decode' ed-    liftIO $ h e+publishEvent :: (MonadIO m, Serialize e, Typeable a, Typeable e, Aggregate a, Eventable a e, Default a) => AggregateRef a e -> e -> TransactionT e m ()+publishEvent aggregateRef event = TransactionT $ do+  -- Publish event to aggregate itself.+  currentGlobalVersion <- gets tsCurrentVersion+  lift $ AR.publishEvent aggregateRef event currentGlobalVersion+  -- Increment global version number.+  modify $ \s -> s { tsCurrentVersion = currentGlobalVersion + 1 }  -- | Get aggregate root. getAggregateRoot :: (Default a, Serialize e, Typeable a, Typeable e, Aggregate a, Eventable a e) => GUID a -> TransactionT e IO (AggregateRef a e, a)