cqrs 0.8.0 → 0.9.0

raw patch · 23 files changed

+548/−428 lines, 23 filesdep +HUnitdep +SafeSemaphoredep +asyncdep −base16-bytestringdep −cerealdep −data-defaultdep ~conduitdep ~containersdep ~transformersPVP ok

version bump matches the API change (PVP)

Dependencies added: HUnit, SafeSemaphore, async, cqrs, cqrs-test, cqrs-types, deepseq, hspec, pool-conduit, stm

Dependencies removed: base16-bytestring, cereal, data-default, safecopy

Dependency ranges changed: conduit, containers, transformers

API changes (from Hackage documentation)

Files

• cqrs.cabal +49/−26
• src-test/Main.hs +11/−0
• src/Data/CQRS.hs +0/−17
• src/Data/CQRS/Aggregate.hs +0/−18
• src/Data/CQRS/AggregateRef.hs +0/−6
• src/Data/CQRS/Command.hs +32/−0
• src/Data/CQRS/EventStore.hs +0/−8
• src/Data/CQRS/EventStore/Backend.hs +0/−59
• src/Data/CQRS/EventStore/Backend/Memory.hs +73/−0
• src/Data/CQRS/Eventable.hs +0/−9
• src/Data/CQRS/GUID.hs +3/−60
• src/Data/CQRS/Internal/AggregateRef.hs +15/−12
• src/Data/CQRS/Internal/EventBus.hs +40/−0
• src/Data/CQRS/Internal/EventStore.hs +23/−40
• src/Data/CQRS/Internal/PersistedEvent.hs +0/−18
• src/Data/CQRS/Internal/Repository.hs +84/−0
• src/Data/CQRS/Internal/UnitOfWork.hs +183/−0
• src/Data/CQRS/PersistedEvent.hs +0/−5
• src/Data/CQRS/Query.hs +19/−0
• src/Data/CQRS/Repository.hs +11/−0
• src/Data/CQRS/Serializable.hs +5/−0
• src/Data/CQRS/Serialize.hs +0/−16
• src/Data/CQRS/Transaction.hs +0/−134

@@ -1,46 +1,69 @@ Name:                cqrs-Version:             0.8.0+Version:             0.9.0 Synopsis:            Command-Query Responsibility Segregation Description:         Haskell implementation of the CQRS architectural pattern. License:             MIT License-file:        LICENSE Category:            Data-Cabal-version:       >=1.6.0.1+Cabal-version:       >=1.10 Build-type:          Simple Author:              Bardur Arantsson Maintainer:          Bardur Arantsson <bardur@scientician.net>  Library-  Build-Depends: base == 4.*-               , base16-bytestring >= 0.1.1.3 && < 0.2-               , bytestring >= 0.9.0.1-               , cereal >= 0.3.3 && < 0.4-               , conduit >= 0.1 && <0.2-               , containers >= 0.4-               , data-default >= 0.3 && < 0.4-               , random >= 1.0 && < 1.1-               , safecopy >= 0.6 && < 0.7-               , transformers >= 0.2.2 && < 0.3-  Extensions:          DeriveDataTypeable+  Build-Depends:       base == 4.*+                     , bytestring >= 0.9.0.1+                     , conduit >= 1.0 && < 2+                     , containers >= 0.5+                     , cqrs-types >= 0.9.0 && < 0.10+                     , deepseq >= 1.3 && < 2+                     , random >= 1.0 && < 1.1+                     , pool-conduit >= 0.1 && < 0.2+                     , SafeSemaphore >= 0.9 && < 1.0+                     , stm >= 2.4 && < 3+                     , transformers >= 0.2.2 && < 0.4+  Default-language:    Haskell2010+  Default-Extensions:  DeriveDataTypeable                        ExistentialQuantification-                       FunctionalDependencies                        GeneralizedNewtypeDeriving                        MultiParamTypeClasses-                       OverloadedStrings-                       Rank2Types+                       PackageImports                        ScopedTypeVariables   ghc-options:         -Wall   hs-source-dirs:      src-  Exposed-modules:     Data.CQRS-                       Data.CQRS.Aggregate-                       Data.CQRS.AggregateRef-                       Data.CQRS.Eventable-                       Data.CQRS.EventStore-                       Data.CQRS.EventStore.Backend+  Exposed-modules:     Data.CQRS.Command+                       Data.CQRS.EventStore.Backend.Memory                        Data.CQRS.GUID-                       Data.CQRS.PersistedEvent-                       Data.CQRS.Serialize-                       Data.CQRS.Transaction+                       Data.CQRS.Repository+                       Data.CQRS.Query+                       Data.CQRS.Serializable   Other-modules:       Data.CQRS.Internal.AggregateRef+                       Data.CQRS.Internal.EventBus                        Data.CQRS.Internal.EventStore-                       Data.CQRS.Internal.PersistedEvent+                       Data.CQRS.Internal.Repository+                       Data.CQRS.Internal.UnitOfWork++Test-Suite cqrs-tests+  Type:                exitcode-stdio-1.0+  Hs-source-dirs:      src-test+  Main-is:             Main.hs+  Build-depends:       base == 4.*+                     , async >= 2.0.1 && < 3+                     , bytestring >= 0.9.0.1+                     , conduit >= 0.5 && < 0.6+                     , pool-conduit >= 0.1 && < 0.2+                     , stm >= 2.4 && < 3+                     , transformers >= 0.2.2 && < 0.4+                     -- Self-dependency+                     , cqrs+                     , cqrs-test >= 0.9 && < 0.10+                     -- Test framework+                     , hspec >= 1.3 && < 2.0+                     , HUnit >= 1.2 && < 2.0+  Ghc-options:         -Wall+  Default-language:    Haskell2010+  Default-extensions:  DeriveDataTypeable+                       MultiParamTypeClasses+                       OverloadedStrings+                       ScopedTypeVariables+                       TemplateHaskell

@@ -0,0 +1,11 @@+import Test.Hspec+import Data.CQRS.TransactionTest+import Data.CQRS.RepositoryTest+import Data.CQRS.EventStore.Backend.MemoryTest++main :: IO ()+main = do+  hspec $ do+    memoryTest+    transactionTest+    repositoryTest

@@ -1,17 +0,0 @@-{-| CQRS module. This module exports all the data types-and functions needed for typical CQRS applications. -}-module Data.CQRS-        ( 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--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,18 +0,0 @@--- | Aggregate type class.-module Data.CQRS.Aggregate-       ( Aggregate(..)-       ) where--import Data.ByteString (ByteString)---- | Type class for aggregates.-class Aggregate a where-  -- | Encode aggregate into a ByteString representation. The-  -- representation should contain some metadata (a UUID for example)-  -- which can be used to check reliably whether the encoded-  -- representation is valid upon decoding. This can be used if the-  -- actual aggregate structure changes.-  encodeAggregate :: a -> ByteString-  -- | Decode ByteString to aggregate state. If decoding is not-  -- possible should return 'Nothing'.-  decodeAggregate :: ByteString -> Maybe a

@@ -1,6 +0,0 @@-{-| Aggregate root references. -}-module Data.CQRS.AggregateRef-       ( AggregateRef-       ) where--import Data.CQRS.Internal.AggregateRef

@@ -0,0 +1,32 @@+{-| Module to import for the "Command" side of the application. -}+module Data.CQRS.Command+    ( -- Aggregate:+      Aggregate+      -- AggregateRef:+    , AggregateRef+      -- Eventable:+    , Eventable(..)+      -- EventStoreBackend:+    , EventStoreBackend+      -- GUID:+    , GUID+    , newGUID+      -- Repository:+    , Repository+      -- UnitOfWork:+    , UnitOfWorkT+    , createAggregate+    , createOrLoadAggregate+    , findAggregate+    , loadAggregate+    , publishEvent+    , runUnitOfWorkT+    ) where++import Data.CQRS.Aggregate (Aggregate)+import Data.CQRS.Internal.AggregateRef (AggregateRef)+import Data.CQRS.Internal.UnitOfWork+import Data.CQRS.Internal.Repository (Repository)+import Data.CQRS.Eventable (Eventable(..))+import Data.CQRS.EventStore.Backend (EventStoreBackend)+import Data.CQRS.GUID (GUID, newGUID)

@@ -1,8 +0,0 @@--- | Event store functions.-module Data.CQRS.EventStore-       ( EventStore-       , enumerateEventStore-       , withEventStore-       ) where--import Data.CQRS.Internal.EventStore

@@ -1,59 +0,0 @@--- | 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(..)-       , RawEvent-       , RawSnapshot(..)-       ) where--import Data.ByteString (ByteString)-import Data.Conduit (Source)-import Data.CQRS.GUID-import Data.CQRS.PersistedEvent (PersistedEvent)---- | Raw event type. The data associated with an event is not--- translated in any way.-type RawEvent = PersistedEvent ByteString---- | Raw snapshot.-data RawSnapshot =-  RawSnapshot { rsVersion :: Int-              , rsSnapshotData :: ByteString-              }---- | 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 :: GUID -> Int -> [RawEvent] -> 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.-    esbRetrieveEvents :: GUID -> Int -> Source IO RawEvent,-    -- | Enumerate all events later than given logical timestamp.-    esbEnumerateAllEvents :: Int -> Source IO RawEvent,-    -- | 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 :: GUID -> RawSnapshot -> 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 :: GUID -> IO (Maybe RawSnapshot),-    -- | 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,73 @@+-- | Memory-based event store backend. Used primarily+-- for testing.+module Data.CQRS.EventStore.Backend.Memory+    ( MemoryEventStoreBackend+    , createBackendPool+    ) where++-- External imports+import           Control.Concurrent.STM (atomically)+import           Control.Concurrent.STM.TVar (TVar, newTVar, readTVar, writeTVar)+import           Control.Concurrent.MSem as MSem+import           Control.Monad (unless, liftM)+import           Control.Monad.IO.Class (liftIO)+import qualified Data.Conduit.List as CL+import           Data.Conduit.Pool (Pool, createPool)+import qualified Data.Foldable as F+import           Data.Map.Strict (Map)+import qualified Data.Map.Strict as M+import           Data.Sequence (Seq, (><))+import qualified Data.Sequence as S++-- Internal imports+import Data.CQRS.GUID+import Data.CQRS.EventStore.Backend++-- | Memory-based event store backend.+data MemoryEventStoreBackend = MESB+    (TVar (Map GUID (Seq RawEvent)))+    (MSem Int)+    (TVar (Map GUID RawSnapshot))++-- | Pool of memory event store backends.+createBackendPool :: Int -> IO (Pool MemoryEventStoreBackend)+createBackendPool n = do+  -- New "backing store" for every pool.+  events <- atomically $ newTVar M.empty+  snapshots <- atomically $ newTVar M.empty+  lock <- MSem.new 1+  -- Create the pool+  createPool (return $ MESB events lock snapshots) (\_ -> return ()) 1 1 n++-- Instance+instance EventStoreBackend MemoryEventStoreBackend where++    esbStoreEvents (MESB eventsTVar _ _) g v0 newEvents =+      atomically $ do+        events <- readTVar eventsTVar+        let eventsForAggregate = M.findWithDefault S.empty g events+        let expectedV0 = S.length eventsForAggregate+        unless (expectedV0 == v0) $ fail "Mismatched version numbers"+        writeTVar eventsTVar $ M.insert g (eventsForAggregate >< S.fromList newEvents) events++    esbRetrieveEvents (MESB eventsTVar _ _) g v0 = do+        e <- liftIO $ atomically $ do+          events <- readTVar eventsTVar+          return $ F.toList $ M.findWithDefault S.empty g events+        CL.sourceList $ drop v0 e++    esbEnumerateAllEvents (MESB eventsTVar _ _) = do+      events <- liftIO $ atomically $ readTVar eventsTVar+      CL.sourceList $ concatMap F.toList $ M.elems $ events++    esbWriteSnapshot (MESB _ _ v) g s =+        atomically $ do+          snapshots <- readTVar v+          writeTVar v $ M.insert g s $ snapshots++    esbGetLatestSnapshot (MESB _ _ v) g =+        atomically $ do+          liftM (M.lookup g) $ readTVar v++    esbWithTransaction (MESB _ lock _) io =+        MSem.with lock io

@@ -1,9 +0,0 @@--- | Eventable type class.-module Data.CQRS.Eventable-       ( Eventable(..)-       ) where---- | Type class for applying events to aggregates.-class Eventable a e | a -> e where-  -- | Apply an event to the aggregate and return the updated aggregate.-  applyEvent :: e -> a -> a

@@ -1,62 +1,5 @@--- | Globally Unique IDentifiers. module Data.CQRS.GUID-       ( GUID-       , fromByteString-       , hexDecode-       , hexEncode-       , newGUID-       , toByteString-       ) where--import           Control.Applicative ((<$>))-import           Data.ByteString (ByteString)-import qualified Data.ByteString as B-import qualified Data.ByteString.Base16 as B16-import           Data.ByteString.Char8 ()-import           Data.Default (Default(..))-import           Data.SafeCopy (SafeCopy(..), contain, safePut, safeGet)-import           Data.Typeable (Typeable)-import           Data.Word (Word8)-import           System.Random (randomRIO)---- | A Globally Unique IDentifier.-newtype GUID = GUID ByteString-              deriving (Show, Typeable, Eq, Ord)---- | Default GUID value.-instance Default GUID where-    def = GUID $ B.pack $ replicate 32 0---- | Get a random byte.-randomWord8 :: IO Word8-randomWord8 = fmap fromInteger $ randomRIO (0,255)---- | Create a new random GUID.-newGUID :: IO GUID-newGUID = do-  uuid <- sequence $ replicate 32 randomWord8-  return $ GUID $ B.pack uuid---- | Serialize instance.-instance SafeCopy GUID where-    putCopy = contain . safePut . toByteString-    getCopy = contain $ fromByteString <$> safeGet---- | Hex encode a GUID.-hexEncode :: GUID -> ByteString-hexEncode (GUID s) = B16.encode s---- | Decode a GUID from hex representation.-hexDecode :: ByteString -> Maybe GUID-hexDecode s =-  case B16.decode s of-    (a,b) | B.length b == 0 -> Just $ GUID a-    _                       -> Nothing---- | Convert from ByteString.-fromByteString :: ByteString -> GUID-fromByteString = GUID+    ( module G+    ) where --- | Convert to ByteString.-toByteString :: GUID -> ByteString-toByteString (GUID g) = g+import "cqrs-types" Data.CQRS.GUID as G

@@ -10,49 +10,52 @@        , readValue        ) where +import           Control.DeepSeq (NFData, ($!!)) import           Control.Monad (liftM) import           Control.Monad.IO.Class (MonadIO, liftIO) import           Data.CQRS.Eventable (Eventable(..)) import           Data.CQRS.GUID (GUID) import           Data.CQRS.PersistedEvent (PersistedEvent(..)) import           Data.Foldable (toList)-import           Data.IORef (IORef, modifyIORef, newIORef, readIORef)+import           Data.IORef (IORef, modifyIORef', newIORef, readIORef, writeIORef) 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+  AggregateRef { arValue :: IORef (Maybe a)                , arEvents :: IORef (Seq (PersistedEvent e))-               , arGUID :: GUID-               , arStartVersion :: Int-               , arSnapshotVersion :: Int+               , arGUID :: !GUID+               , arStartVersion :: !Int+               , arSnapshotVersion :: !Int                }   deriving (Typeable)  -- | Make aggregate-mkAggregateRef :: (MonadIO m) => a -> GUID -> Int -> Int -> m (AggregateRef a e)+mkAggregateRef :: (MonadIO m) => Maybe a -> GUID -> Int -> Int -> m (AggregateRef a e) mkAggregateRef a guid originatingVersion snapshotVersion = do   a' <- liftIO $ newIORef a   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 -> Int -> m ()-publishEvent aggregateRef event gv = liftIO $ do+publishEvent :: (MonadIO m, Eventable a e, NFData e, NFData a) => AggregateRef a e -> e -> m ()+publishEvent aggregateRef event = liftIO $ do   -- Apply event to aggregate state.-  modifyIORef (arValue aggregateRef) $ applyEvent event+  a <- readIORef $ arValue aggregateRef+  let a' = applyEvent a $!! event+  writeIORef (arValue aggregateRef) $!! a'   -- Add event to aggregate.-  modifyIORef (arEvents aggregateRef) $ \events ->-    events |> (PersistedEvent (arGUID aggregateRef) event (arStartVersion aggregateRef + 1 + S.length events) gv)+  modifyIORef' (arEvents aggregateRef) $ \events -> do+    (events |>) $!! (PersistedEvent (arGUID aggregateRef) event (arStartVersion aggregateRef + 1 + S.length events))  -- | Read aggregate events. readEvents :: (MonadIO m) => AggregateRef a e -> m [PersistedEvent e] readEvents = liftM toList . liftIO . readIORef . arEvents  -- | Read aggregate state.-readValue :: (MonadIO m) => AggregateRef a e -> m a+readValue :: (MonadIO m) => AggregateRef a e -> m (Maybe a) readValue = liftIO . readIORef . arValue  -- | Get the current version of the aggregate in aggregate ref.

@@ -0,0 +1,40 @@+module Data.CQRS.Internal.EventBus+    ( EventBus(..)+    , Subscription(..)+    , newEventBus+    ) where++-- External imports+import           Control.Concurrent.STM (atomically)+import qualified Control.Concurrent.STM.TChan as C+import           Control.DeepSeq (NFData, ($!!))+import           Control.Monad.IO.Class (MonadIO, liftIO)++-- Internal imports+import           Data.CQRS.PersistedEvent++-- | Event bus that broadcasts events of a certain type.+data EventBus e = EventBus+    { publishEventsToBus :: [PersistedEvent e] -> IO ()+    , subscribeToEventBus :: IO (Subscription e)+    , unsubscribeFromEventBus :: Subscription e -> IO ()+    }++-- | A subscription to an event bus.+newtype Subscription e = Subscription+    { readEventFromSubscription :: IO [PersistedEvent e]+    }++-- | Create a new event bus.+newEventBus :: (MonadIO m, Show e, NFData e) => m (EventBus e)+newEventBus = do+  c <- liftIO $ atomically $ C.newBroadcastTChan+  let publish events = do+        atomically $ C.writeTChan c $!! events+      subscribe = do+        c' <- atomically $ C.dupTChan c+        return $ Subscription (atomically $ C.readTChan c')+      unsubscribe _ = do+        -- TODO: Should we set a flag to detect post-unsubscribe use?+        return () -- Needs no cleanup.+  return $ EventBus publish subscribe unsubscribe

@@ -1,54 +1,37 @@ -- | Event store functions. module Data.CQRS.Internal.EventStore-       ( EventStore+       ( EventStore(..)        , enumerateEventStore-       , getLatestSnapshot-       , getLatestVersion        , retrieveEvents        , storeEvents-       , withEventStore-       , withTransaction-       , writeSnapshot        ) where -import           Control.Exception (bracket)-import           Control.Monad (liftM)-import           Data.CQRS.EventStore.Backend (EventStoreBackend(..), RawSnapshot)-import           Data.CQRS.GUID-import           Data.CQRS.Internal.PersistedEvent (PersistedEvent(..), mapPersistedEvent)-import           Data.CQRS.Serialize (decode', encode)-import           Data.Conduit (Source, ($=))+-- External imports+import           Data.ByteString (ByteString)+import           Data.Conduit (ResourceT, Source, ($=)) import qualified Data.Conduit.List as CL-import           Data.SafeCopy (SafeCopy)+-- Internal imports+import           Data.CQRS.EventStore.Backend+import           Data.CQRS.GUID+import           Data.CQRS.Serializable+import           Data.CQRS.PersistedEvent (PersistedEvent(..))  -- 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 e . (SafeCopy e) => EventStore e -> Int -> Source IO (PersistedEvent e)-enumerateEventStore es minVersion =-  esbEnumerateAllEvents (esBackend es) minVersion $= CL.map (mapPersistedEvent decode')--withTransaction :: EventStore e -> IO a -> IO a-withTransaction (EventStore esb) = esbWithTransaction esb--storeEvents :: SafeCopy e => EventStore e -> GUID -> Int -> [PersistedEvent e] -> IO ()-storeEvents (EventStore esb) guid v0 evs = esbStoreEvents esb guid v0 $ (map $ mapPersistedEvent $ encode) evs+data EventStore e b = EventStore { esBackend :: b } -retrieveEvents :: SafeCopy e => EventStore e -> GUID -> Int -> Source IO (PersistedEvent e)-retrieveEvents (EventStore esb) guid v0 = fmap (mapPersistedEvent decode') $ esbRetrieveEvents esb guid v0+enumerateEventStore :: forall e b . (Serializable e, EventStoreBackend b) => EventStore e b -> Source (ResourceT IO) [PersistedEvent e]+enumerateEventStore es =+  esbEnumerateAllEvents (esBackend es) $= CL.map (\re -> [fmap deserialize' re]) -writeSnapshot ::   EventStore e -> GUID -> RawSnapshot -> IO ()-writeSnapshot (EventStore esb) = esbWriteSnapshot esb+storeEvents :: (Serializable e, EventStoreBackend b) => EventStore e b -> GUID -> Int -> [PersistedEvent e] -> IO ()+storeEvents (EventStore esb) guid v0 evs = do+  esbStoreEvents esb guid v0 $ (map $ fmap $ serialize) evs -getLatestSnapshot :: EventStore e -> GUID -> IO (Maybe RawSnapshot)-getLatestSnapshot (EventStore esb) = esbGetLatestSnapshot esb+retrieveEvents :: (Serializable e, EventStoreBackend b) => EventStore e b -> GUID -> Int -> Source (ResourceT IO) (PersistedEvent e)+retrieveEvents (EventStore esb) guid v0 = (esbRetrieveEvents esb guid v0) $= (CL.map (fmap deserialize')) -getLatestVersion :: EventStore e -> IO Int-getLatestVersion (EventStore esb) = esbGetLatestVersion esb+deserialize' :: Serializable e => ByteString -> e+deserialize' s =+    case deserialize s of+      Nothing -> error "Could not deserialize event -- event deserialization MUST NOT fail"+      Just e -> e

@@ -1,18 +0,0 @@-module Data.CQRS.Internal.PersistedEvent-       ( PersistedEvent(..)-       , mapPersistedEvent-       ) where--import Data.CQRS.GUID (GUID)---- | Persisted Event.-data PersistedEvent e =-  PersistedEvent { peAggregateGUID :: !GUID -- ^ GUID of the aggregate.-                 , peEvent :: e             -- ^ Event.-                 , peSequenceNumber :: !Int -- ^ Sequence number within the aggregate.-                 , peGlobalVer :: !Int      -- ^ Global sequence number.-                 }---- Transform the event data.-mapPersistedEvent :: (e -> e') -> PersistedEvent e -> PersistedEvent e'-mapPersistedEvent f (PersistedEvent g e v gv) = PersistedEvent g (f e) v gv

@@ -0,0 +1,84 @@+module Data.CQRS.Internal.Repository+    ( Repository(..)+    , Settings(..)+    , defaultSettings+    , enumerateEventStore+    , enumerateAndStreamEvents+    , newRepository+    , withEventStoreBackend+    ) where++-- External imports+import           Control.DeepSeq (NFData)+import           Control.Monad (forever)+import           Control.Monad.Trans.Class (lift)+import           Data.Conduit (ResourceT, Source, addCleanup, bracketP, yield)+import           Data.Conduit.Pool (Pool, withResource, takeResource, mrReuse, mrValue)++-- Internal imports+import           Data.CQRS.Internal.EventBus+import qualified Data.CQRS.Internal.EventStore as ES+import           Data.CQRS.Internal.EventStore (EventStore(..))+import           Data.CQRS.EventStore.Backend+import           Data.CQRS.PersistedEvent+import           Data.CQRS.Serializable++-- | Repository settings+data Settings = Settings+    { settingsSnapshotFrequency :: Maybe Int+    }++-- | Default repository settings.+defaultSettings :: Settings+defaultSettings = Settings+    { settingsSnapshotFrequency = Nothing+    }++-- | Repository consisting of an event store and an event bus.+data Repository e b = Repository+    { repositoryEventStoreBackendPool :: Pool b  -- ^ Event store backend pool+    , repositoryEventBus :: EventBus e           -- ^ Event bus to use+    , repositorySettings :: Settings             -- ^ Repository settings+    }++-- | Create a repository from a pool of event store backends.+newRepository :: (EventStoreBackend b, Show e, NFData e) => Settings -> Pool b -> IO (Repository e b)+newRepository settings eventStoreBackendPool = do+  eventBus <- newEventBus+  return $ Repository { repositoryEventStoreBackendPool = eventStoreBackendPool+                      , repositoryEventBus = eventBus+                      , repositorySettings = settings+                      }++-- | Use event store backed from repository to perform an action.+withEventStoreBackend :: (EventStoreBackend b) => Repository e b -> (b -> IO a) -> IO a+withEventStoreBackend (Repository p _ _) action =+    withResource p (\eventStoreBackend -> action eventStoreBackend)++-- | Enumerate all events which satisfy certain criteria from event+-- store associated with repository.+enumerateEventStore :: (Serializable e, EventStoreBackend b) => Repository e b -> Source (ResourceT IO) [PersistedEvent e]+enumerateEventStore (Repository p _ _) = do+  eventStoreBackend <- lift $ takeResource p+  addCleanup (mrReuse eventStoreBackend) $ ES.enumerateEventStore (EventStore $ mrValue eventStoreBackend)++-- State for the streaming thread.+data StreamingState e = Buffering [PersistedEvent e]+                      | EnumerationComplete [PersistedEvent e]+                      | Streaming+                        deriving (Show)++-- | Enumerate all events which satisfy criteria and stream+-- all new events from repository. All events which arrive+-- while enumerating are buffered until enumeration completes.+enumerateAndStreamEvents :: (Show e, Serializable e, EventStoreBackend b) => Repository e b -> Source (ResourceT IO) [PersistedEvent e]+enumerateAndStreamEvents repository@(Repository _ eventBus _) = do+  bracketP (subscribeToEventBus eventBus)+           (unsubscribeFromEventBus eventBus)+           (\subscription -> do+              enumerateEventStore repository+              subscriptionSource subscription)+  where+    subscriptionSource subscription = forever $ do+      e <- lift $ lift $ readEventFromSubscription subscription+      yield e

@@ -0,0 +1,183 @@+-- | Run commands against repositories.+module Data.CQRS.Internal.UnitOfWork+       ( UnitOfWorkT+       , createAggregate+       , createOrLoadAggregate+       , findAggregate+       , loadAggregate+       , publishEvent+       , runUnitOfWorkT+       ) where++-- External imports+import           Control.DeepSeq (NFData)+import           Control.Monad (forM, when)+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.Conduit (($$), runResourceT)+import qualified Data.Conduit.List as CL+import           Data.Foldable (forM_)+import           Data.List (find)+import           Data.Typeable (Typeable, cast)++-- Internal imports+import           Data.CQRS.Aggregate+import           Data.CQRS.Eventable (Eventable(..))+import           Data.CQRS.Internal.EventBus+import           Data.CQRS.EventStore.Backend (EventStoreBackend(..), RawSnapshot(..))+import           Data.CQRS.GUID (GUID)+import           Data.CQRS.Internal.AggregateRef (AggregateRef, mkAggregateRef)+import qualified Data.CQRS.Internal.AggregateRef as AR+import           Data.CQRS.Internal.EventStore (EventStore(..))+import qualified Data.CQRS.Internal.EventStore as ES+import           Data.CQRS.Internal.Repository+import           Data.CQRS.PersistedEvent (PersistedEvent(..))+import           Data.CQRS.Serializable++-- | UnitOfWork monad transformer.+newtype UnitOfWorkT e m a = UnitOfWorkT (UnitOfWorkM e m a)+    deriving (Functor, Monad)++instance MonadTrans (UnitOfWorkT e) where+  lift m = UnitOfWorkT $ lift m++-- Existential wrapper for AggregateRef.+data BoxedAggregateRef e =+  forall a . (Typeable a, Typeable e, Serializable e, Aggregate a, Eventable a e) => BoxedAggregateRef (AggregateRef a e)++-- Existential wrapper for event store.+data BoxedEventStore e =+    forall b . (EventStoreBackend b) => BoxedEventStore (EventStore e b)++-- Existential wrapper for event store backend+data BoxedEventStoreBackend =+    forall b . (EventStoreBackend b) => BoxedEventStoreBackend b++-- | UnitOfWork monad.+type UnitOfWorkM e = StateT (UnitOfWork e)++data UnitOfWork e =+    UnitOfWork { txnEventStore :: BoxedEventStore e+               , txnEventStoreBackend :: BoxedEventStoreBackend+               , aggregateRefsToCommit :: [BoxedAggregateRef e]+               }++-- | Run transaction against an event store.+runUnitOfWorkT :: forall b c e . (Typeable e, Serializable e, EventStoreBackend b) => Repository e b -> UnitOfWorkT e IO c -> IO c+runUnitOfWorkT repository (UnitOfWorkT transaction) = do+  (r, writtenEvents) <- withEventStoreBackend repository $ \eventStoreBackend -> do+    esbWithTransaction eventStoreBackend $ do+      -- Run the computation.+      let eventStore = EventStore eventStoreBackend+      (r,s) <- runStateT transaction $ UnitOfWork (BoxedEventStore eventStore) (BoxedEventStoreBackend eventStoreBackend) []+      -- Write out all the aggregates.+      writtenEvents <- forM (aggregateRefsToCommit s) $ \(BoxedAggregateRef a) -> do+        -- 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.+        forM_ (settingsSnapshotFrequency $ repositorySettings repository) $ \f -> do+          v <- AR.getCurrentVersion a+          when (v - AR.arSnapshotVersion a > f) $ do+            mav <- AR.readValue a+            case mav of+              Just av ->+                  esbWriteSnapshot eventStoreBackend (AR.arGUID a) $ RawSnapshot v $ serialize av+              Nothing ->+                  return ()+        -- Return the written events for accumulator+        return $ evs+      -- Return the value.+      return (r, concat writtenEvents)+  -- Publish+  publishEventsToBus (repositoryEventBus repository) writtenEvents+  -- Return command return value+  return r++-- Get an aggregate ref by GUID.+getById :: forall a e . (Typeable a, Typeable e, Serializable e, Aggregate a, Eventable a e) => GUID -> UnitOfWorkT e IO (AggregateRef a e)+getById guid = UnitOfWorkT $ do+  -- Check through list to see if we've given out a reference to the aggregate before.+  aggregateRefs <- fmap aggregateRefsToCommit get+  case find (\(BoxedAggregateRef a) -> AR.arGUID a == guid) aggregateRefs of+    Just (BoxedAggregateRef a) ->+      case cast a of+        Just (a' :: AggregateRef a e) -> return a'+        Nothing ->+          -- This cast could only really fail if there are duplicate GUIDs for+          -- different types of aggregates/events.+          fail $ concat ["Duplicate GUID ", show guid, "!" ]+    Nothing -> do+      getByIdFromEventStore guid++-- Get the latest snapshot from database, filling in a default+-- if a) no snapshot exists, or b) snapshot state was not decodable.+getLatestSnapshot :: forall a e . (Typeable a, Typeable e, Serializable e, Aggregate a) => GUID -> UnitOfWorkM e IO (Int, Maybe a)+getLatestSnapshot guid = do+  (BoxedEventStoreBackend eventStoreBackend) <- fmap txnEventStoreBackend get+  r <- liftIO $ esbGetLatestSnapshot eventStoreBackend guid+  case r of+    Just (RawSnapshot v a) -> do+      case deserialize a :: Maybe a of+        Just a' -> return (v, Just a')+        Nothing -> return (0, Nothing)+    Nothing -> do+      return (0, Nothing)++-- Retrieve aggregate from event store.+getByIdFromEventStore :: forall a e . (Typeable a, Typeable e, Serializable e, Aggregate a, Eventable a e) => GUID -> UnitOfWorkM e IO (AggregateRef a e)+getByIdFromEventStore guid = do+  (BoxedEventStore es) <- fmap txnEventStore get+  -- Get latest snapshot (if any).+  (v0,ma0) <- getLatestSnapshot guid+  -- Get events.+  events <- lift $ runResourceT $ (ES.retrieveEvents es guid v0 $$ CL.consume)+  let latestVersion = maximum $ (:) v0 (map peSequenceNumber events)+  -- Build the aggregate state from all the events.+  let a = foldl (\a0 e -> applyEvent a0 $ peEvent e) ma0 events+  -- Make the aggregate itself+  (a' :: AggregateRef a e) <- lift $ mkAggregateRef a guid latestVersion v0+  -- Add to set of aggregates to commit later.+  modify $ \s -> s { aggregateRefsToCommit = (BoxedAggregateRef a' : aggregateRefsToCommit s) }+  -- Return the aggregate.+  return $ a'++-- | Publish event for an aggregate root.+publishEvent :: (MonadIO m, Serializable e, Typeable a, Typeable e, Aggregate a, Eventable a e, NFData a, NFData e) => AggregateRef a e -> e -> UnitOfWorkT e m ()+publishEvent aggregateRef event = UnitOfWorkT $ do+  lift $ AR.publishEvent aggregateRef event++-- | Find aggregate root.+findAggregate :: (Serializable e, Typeable a, Typeable e, Aggregate a, Eventable a e) => GUID -> UnitOfWorkT e IO (Maybe (AggregateRef a e, a))+findAggregate guid = do+  aggregateRef <- getById guid+  aggregate <- lift $ AR.readValue aggregateRef+  case aggregate of+    Nothing -> return Nothing+    Just a -> return $ Just (aggregateRef, a)++-- | Load aggregate root. The aggregate root must exist.+loadAggregate :: (Serializable e, Typeable a, Typeable e, Aggregate a, Eventable a e) => GUID -> UnitOfWorkT e IO (AggregateRef a e, a)+loadAggregate guid = do+  mAggregate <- findAggregate guid+  case mAggregate of+    Nothing -> fail $ show $ "Aggregate with GUID " ++ show guid ++ " does not exist"+    Just a -> return a++-- | Add aggregate root. The aggregate root will be created upon+-- transaction commit.+createAggregate :: (Serializable e, Typeable a, Typeable e, Aggregate a, Eventable a e) => GUID -> UnitOfWorkT e IO (AggregateRef a e)+createAggregate guid = do+  aggregateRef <- getById guid+  aggregate <- lift $ AR.readValue aggregateRef+  case aggregate of+    Nothing -> do+      return aggregateRef+    Just _ -> fail $ show $ "Aggregate with GUID " ++ show guid ++ " already exists"++-- | Create or load aggregate. The aggregate root will be created (if necessary)+-- upon transaction commit.+createOrLoadAggregate :: (Serializable e, Typeable a, Typeable e, Aggregate a, Eventable a e) => GUID -> UnitOfWorkT e IO (AggregateRef a e)+createOrLoadAggregate guid = getById guid

@@ -1,5 +0,0 @@-module Data.CQRS.PersistedEvent-       ( PersistedEvent(..)-       ) where--import Data.CQRS.Internal.PersistedEvent

@@ -0,0 +1,19 @@+module Data.CQRS.Query+    ( -- cqrs-types:+      PersistedEvent(..)+      -- EventStoreBackend:+    , EventStoreBackend+      -- GUID:+    , GUID+      -- Repository:+    , Repository+    , enumerateAndStreamEvents+    , enumerateEventStore+    ) where++import "cqrs-types" Data.CQRS.PersistedEvent+import Data.CQRS.GUID (GUID)+import Data.CQRS.Internal.Repository (Repository,+                                      enumerateEventStore,+                                      enumerateAndStreamEvents)+import Data.CQRS.EventStore.Backend (EventStoreBackend)

@@ -0,0 +1,11 @@+module Data.CQRS.Repository+    ( -- * Repository+      Repository+    , newRepository+      -- * Settings+    , Settings+    , settingsSnapshotFrequency+    , defaultSettings+    ) where++import Data.CQRS.Internal.Repository

@@ -0,0 +1,5 @@+module Data.CQRS.Serializable+    ( module G+    ) where++import "cqrs-types" Data.CQRS.Serializable as G

@@ -1,16 +0,0 @@-module Data.CQRS.Serialize-       ( decode'-       , encode-       ) where--import Data.ByteString (ByteString)-import Data.SafeCopy (SafeCopy, safeGet, safePut)-import Data.Serialize (runGet, runPut)---- | Decode a serialized value irrefutably.-decode' :: SafeCopy e => ByteString -> e-decode' = either error id . runGet safeGet---- | Encode a value.-encode :: SafeCopy e => e -> ByteString-encode = runPut . safePut

@@ -1,134 +0,0 @@--- | Run transactions against event stores.-module Data.CQRS.Transaction-       ( TransactionT-       , getAggregateRoot-       , publishEvent-       , runTransactionT-       ) where--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, gets, modify, runStateT)-import           Data.Conduit (($$), runResourceT)-import qualified Data.Conduit.List as CL-import           Data.CQRS.Aggregate (Aggregate(..))-import           Data.CQRS.Eventable (Eventable(..))-import           Data.CQRS.EventStore (EventStore)-import           Data.CQRS.EventStore.Backend (RawSnapshot(..))-import           Data.CQRS.GUID (GUID)-import           Data.CQRS.Internal.AggregateRef (AggregateRef, mkAggregateRef)-import qualified Data.CQRS.Internal.AggregateRef as AR-import qualified Data.CQRS.Internal.EventStore as ES-import           Data.CQRS.PersistedEvent (PersistedEvent(..))-import           Data.Default (Default(..))-import           Data.List (find)-import           Data.SafeCopy (SafeCopy)-import           Data.Typeable (Typeable, cast)---- | Transaction monad transformer.-newtype TransactionT e m a = TransactionT (Transaction e m a)-                           deriving (Functor, Monad)--instance MonadTrans (TransactionT e) where-  lift m = TransactionT $ lift m---- Existential wrapper for AggregateRef.-data BoxedAggregateRef e =-  forall a . (Typeable a, Typeable e, SafeCopy e, Default a, Aggregate a, Eventable a e) => BoxedAggregateRef (AggregateRef a e)---- | Transaction monad itself.-type Transaction e = StateT (TransactionState e)--data TransactionState e =-  TransactionState { eventStore :: EventStore e-                   , aggregateRefsToCommit :: [BoxedAggregateRef e]-                   , tsCurrentVersion :: Int-                   }---- | Run transaction against an event store.-runTransactionT :: (Typeable e, SafeCopy e) => EventStore e -> TransactionT e IO c -> IO c-runTransactionT eventStore_ (TransactionT transaction) = 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 gv0-    -- Write out all the aggregates.-    forM_ (aggregateRefsToCommit s) $ \(BoxedAggregateRef a) -> do-      -- 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-      when (v - AR.arSnapshotVersion a > 10) $ do-        av <- AR.readValue a-        ES.writeSnapshot eventStore_ (AR.arGUID a) $ RawSnapshot v (encodeAggregate av)-    -- Return the value.-    return r-  where s0 = TransactionState eventStore_ [ ]---- | Get an aggregate ref by GUID.-getById :: forall a e . (Typeable a, Typeable e, SafeCopy e, Default a, Aggregate a, Eventable a e) => GUID -> TransactionT e IO (AggregateRef a e)-getById guid = TransactionT $ do-  -- Check through list to see if we've given out a reference to the aggregate before.-  aggregateRefs <- fmap aggregateRefsToCommit get-  case find (\(BoxedAggregateRef a) -> AR.arGUID a == guid) aggregateRefs of-    Just (BoxedAggregateRef a) ->-      case cast a of-        Just (a' :: AggregateRef a e) -> return a'-        Nothing ->-          -- This cast could only really fail if there are duplicate GUIDs for-          -- different types of aggregates/events.-          fail $ concat ["Duplicate GUID ", show guid, "!" ]-    Nothing -> do-      getByIdFromEventStore guid---- Get the latest snapshot from database, filling in a default--- if a) no snapshot exists, or b) snapshot state was not decodable.-getLatestSnapshot :: forall a e . (Typeable a, Typeable e, SafeCopy e, Default a, Aggregate a) => GUID -> Transaction e IO (Int,a)-getLatestSnapshot guid = do-  es <- fmap eventStore get-  r <- liftIO $ ES.getLatestSnapshot es guid-  case r of-    Just (RawSnapshot v a) -> do-      case decodeAggregate a :: Maybe a of-        Just a' -> return (v, a')-        Nothing -> return (0, def)-    Nothing -> do-      return (0, def)---- Retrieve aggregate from event store.-getByIdFromEventStore :: forall a e . (Typeable a, Typeable e, SafeCopy e, Default a, Aggregate a, Eventable a e) => GUID -> Transaction e IO (AggregateRef a e)-getByIdFromEventStore guid = do-  es <- fmap eventStore get-  -- Get latest snapshot (if any).-  (v0,a0) <- getLatestSnapshot guid-  -- Get events.-  events <- lift $ runResourceT $ ES.retrieveEvents es guid v0 $$ CL.consume-  let latestVersion = maximum $ (:) v0 (map peSequenceNumber events)-  -- Build the aggregate state from all the events.-  let a = foldr (applyEvent . peEvent) a0 events-  -- Make the aggregate itself-  (a' :: AggregateRef a e) <- lift $ mkAggregateRef a guid latestVersion v0-  -- Add to set of aggregates to commit later.-  modify $ \s -> s { aggregateRefsToCommit = (BoxedAggregateRef a' : aggregateRefsToCommit s) }-  -- Return the aggregate.-  return a'---- | Publish event for an aggregate root.-publishEvent :: (MonadIO m, SafeCopy 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, SafeCopy e, Typeable a, Typeable e, Aggregate a, Eventable a e) => GUID -> TransactionT e IO (AggregateRef a e, a)-getAggregateRoot guid = do-  aggregateRef <- getById guid-  aggregate <- lift $ AR.readValue aggregateRef-  return (aggregateRef, aggregate)