cqrs-sqlite3 0.6.0 → 0.7.0
raw patch · 2 files changed
+15/−18 lines, 2 filesdep ~cqrs
Dependency ranges changed: cqrs
Files
cqrs-sqlite3.cabal view
@@ -1,5 +1,5 @@ Name: cqrs-sqlite3-Version: 0.6.0+Version: 0.7.0 Synopsis: SQLite3 backend for the cqrs package. Description: SQLite3 backend for the cqrs package. License: MIT@@ -14,7 +14,7 @@ build-depends: base == 4.* , bytestring >= 0.9.0.1 , cereal >= 0.3.3 && < 0.4- , cqrs >= 0.6.0 && < 0.7+ , cqrs >= 0.6.1 && < 0.7 , enumerator >= 0.4.15 && < 0.5 , direct-sqlite >= 1.1 && < 1.2 extensions: ScopedTypeVariables
src/Data/CQRS/EventStore/Backend/Sqlite3.hs view
@@ -5,9 +5,10 @@ import Control.Monad (when, forM_, liftM) import Data.ByteString (ByteString)-import Data.CQRS.EventStore.Backend (EventStoreBackend(..))+import Data.CQRS.EventStore.Backend (EventStoreBackend(..), RawEvent) import Data.CQRS.EventStore.Backend.Sqlite3Utils (withTransaction, execSql, enumQueryResult) import Data.CQRS.GUID (GUID)+import Data.CQRS.PersistedEvent (PersistedEvent(..)) import Data.CQRS.Serialize (decode') import Data.Enumerator (Enumerator, (>>==), ($=), run_) import qualified Data.Enumerator.List as EL@@ -34,7 +35,7 @@ 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;"+selectEventsSql = "SELECT version, gversion, 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;"@@ -72,7 +73,7 @@ , ", saw ", show cv, ")" ] -storeEvents :: Database -> GUID -> Int -> [(ByteString,Int)] -> IO ()+storeEvents :: Database -> GUID -> Int -> [RawEvent] -> IO () storeEvents database guid originatingVersion events = do -- Column unpacking. let unpackColumns [ SQLInteger v ] = v@@ -92,35 +93,31 @@ ] -- Store the supplied events.- forM_ (zip [1 + originatingVersion..] events) $ \(v,(e,gv)) -> do+ forM_ events $ \e -> do execSql database insertEventSql [ toSQLData guid- , toSQLData v- , toSQLData e- , toSQLData gv+ , toSQLData $ peSequenceNumber e+ , toSQLData $ peEvent e+ , toSQLData $ peGlobalVer e ] -retrieveEvents :: Database -> GUID -> Int -> IO (Int,[ByteString])+retrieveEvents :: Database -> GUID -> Int -> IO [RawEvent] retrieveEvents database guid v0 = do -- Unpack the columns into tuples.- let unpackColumns [SQLInteger version, SQLBlob eventData] = (version, eventData)+ let unpackColumns [SQLInteger version, SQLInteger gversion, SQLBlob eventData] = PersistedEvent guid eventData (fromIntegral version) (fromIntegral gversion) unpackColumns columns = error $ badQueryResultMsg [show guid, show v0] columns -- Find events with version numbers.- results <-- run_ $ EL.consume >>==+ 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 :: Database -> Int -> Enumerator (Int,(GUID, Int, ByteString)) IO a+enumerateAllEvents :: Database -> Int -> Enumerator RawEvent IO a 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) )+ PersistedEvent (decode' g) ed (fromIntegral v) (fromIntegral gv) _ -> error $ badQueryResultMsg [show minVersion] columns)