cqrs 0.7.1 → 0.8.0

raw patch · 6 files changed

+59/−43 lines, 6 filesdep +conduitdep +safecopydep −enumeratorPVP ok

version bump matches the API change (PVP)

Dependencies added: conduit, safecopy

Dependencies removed: enumerator

API changes (from Hackage documentation)

Files

• cqrs.cabal +3/−2
• src/Data/CQRS/EventStore/Backend.hs +12/−5
• src/Data/CQRS/GUID.hs +5/−5
• src/Data/CQRS/Internal/EventStore.hs +13/−14
• src/Data/CQRS/Serialize.hs +10/−4
• src/Data/CQRS/Transaction.hs +16/−13

@@ -1,5 +1,5 @@ Name:                cqrs-Version:             0.7.1+Version:             0.8.0 Synopsis:            Command-Query Responsibility Segregation Description:         Haskell implementation of the CQRS architectural pattern. License:             MIT@@ -15,10 +15,11 @@                , 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-               , enumerator >= 0.4.15 && < 0.5                , random >= 1.0 && < 1.1+               , safecopy >= 0.6 && < 0.7                , transformers >= 0.2.2 && < 0.3   Extensions:          DeriveDataTypeable                        ExistentialQuantification

@@ -4,17 +4,24 @@ 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)-import Data.Enumerator (Enumerator)  -- | 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 =@@ -28,19 +35,19 @@     -- 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 -> IO [RawEvent],+    esbRetrieveEvents :: GUID -> Int -> Source IO RawEvent,     -- | Enumerate all events later than given logical timestamp.-    esbEnumerateAllEvents :: forall a. Int -> Enumerator RawEvent IO a,+    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 -> (Int,ByteString) -> IO (),+    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 (Int,ByteString)),+    esbGetLatestSnapshot :: GUID -> IO (Maybe RawSnapshot),     -- | Get the latest version global number.     esbGetLatestVersion :: IO Int,     -- | Run transaction against the event store. The transaction is

@@ -8,13 +8,13 @@        , toByteString        ) where -import           Control.Monad (liftM)+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.Serialize (Serialize(..))+import           Data.SafeCopy (SafeCopy(..), contain, safePut, safeGet) import           Data.Typeable (Typeable) import           Data.Word (Word8) import           System.Random (randomRIO)@@ -38,9 +38,9 @@   return $ GUID $ B.pack uuid  -- | Serialize instance.-instance Serialize GUID where-    put = put . toByteString-    get = liftM fromByteString get+instance SafeCopy GUID where+    putCopy = contain . safePut . toByteString+    getCopy = contain $ fromByteString <$> safeGet  -- | Hex encode a GUID. hexEncode :: GUID -> ByteString

@@ -13,14 +13,13 @@  import           Control.Exception (bracket) import           Control.Monad (liftM)-import           Data.ByteString (ByteString)-import           Data.CQRS.EventStore.Backend (EventStoreBackend(..))+import           Data.CQRS.EventStore.Backend (EventStoreBackend(..), RawSnapshot) import           Data.CQRS.GUID import           Data.CQRS.Internal.PersistedEvent (PersistedEvent(..), mapPersistedEvent)-import           Data.CQRS.Serialize (decode')-import           Data.Enumerator (Enumerator, ($=))-import qualified Data.Enumerator.List as EL-import           Data.Serialize (Serialize, encode)+import           Data.CQRS.Serialize (decode', encode)+import           Data.Conduit (Source, ($=))+import qualified Data.Conduit.List as CL+import           Data.SafeCopy (SafeCopy)  -- Provide a type alias. data EventStore e = EventStore { esBackend :: EventStoreBackend@@ -32,23 +31,23 @@  -- | Enumerate all the events from an event store that occur at or later -- than a given logical timestamp.-enumerateEventStore :: forall a e . (Serialize e) => EventStore e -> Int -> Enumerator (PersistedEvent e) IO a+enumerateEventStore :: forall e . (SafeCopy e) => EventStore e -> Int -> Source IO (PersistedEvent e) enumerateEventStore es minVersion =-  esbEnumerateAllEvents (esBackend es) minVersion $= EL.map (mapPersistedEvent decode')+  esbEnumerateAllEvents (esBackend es) minVersion $= CL.map (mapPersistedEvent decode')  withTransaction :: EventStore e -> IO a -> IO a withTransaction (EventStore esb) = esbWithTransaction esb -storeEvents :: Serialize e => EventStore e -> GUID -> Int -> [PersistedEvent e] -> IO ()-storeEvents (EventStore esb) guid v0 evs = esbStoreEvents esb guid v0 $ (map $ mapPersistedEvent encode) evs+storeEvents :: SafeCopy e => EventStore e -> GUID -> Int -> [PersistedEvent e] -> IO ()+storeEvents (EventStore esb) guid v0 evs = esbStoreEvents esb guid v0 $ (map $ mapPersistedEvent $ encode) evs -retrieveEvents :: Serialize e => EventStore e -> GUID -> Int -> IO [PersistedEvent e]-retrieveEvents (EventStore esb) guid v0 = liftM (map $ mapPersistedEvent decode') $ esbRetrieveEvents esb guid v0+retrieveEvents :: SafeCopy e => EventStore e -> GUID -> Int -> Source IO (PersistedEvent e)+retrieveEvents (EventStore esb) guid v0 = fmap (mapPersistedEvent decode') $ esbRetrieveEvents esb guid v0 -writeSnapshot ::   EventStore e -> GUID -> (Int, ByteString) -> IO ()+writeSnapshot ::   EventStore e -> GUID -> RawSnapshot -> IO () writeSnapshot (EventStore esb) = esbWriteSnapshot esb -getLatestSnapshot :: EventStore e -> GUID -> IO (Maybe (Int, ByteString))+getLatestSnapshot :: EventStore e -> GUID -> IO (Maybe RawSnapshot) getLatestSnapshot (EventStore esb) = esbGetLatestSnapshot esb  getLatestVersion :: EventStore e -> IO Int

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

@@ -10,9 +10,12 @@ 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@@ -20,7 +23,7 @@ import           Data.CQRS.PersistedEvent (PersistedEvent(..)) import           Data.Default (Default(..)) import           Data.List (find)-import           Data.Serialize (Serialize)+import           Data.SafeCopy (SafeCopy) import           Data.Typeable (Typeable, cast)  -- | Transaction monad transformer.@@ -32,7 +35,7 @@  -- Existential wrapper for AggregateRef. data BoxedAggregateRef e =-  forall a . (Typeable a, Typeable e, Serialize e, Default a, Aggregate a, Eventable a e) => BoxedAggregateRef (AggregateRef a 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)@@ -44,7 +47,7 @@                    }  -- | Run transaction against an event store.-runTransactionT :: (Typeable e, Serialize e) => EventStore e -> TransactionT e IO c -> IO c+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.@@ -61,13 +64,13 @@       v <- AR.getCurrentVersion a       when (v - AR.arSnapshotVersion a > 10) $ do         av <- AR.readValue a-        ES.writeSnapshot eventStore_ (AR.arGUID a) (v, encodeAggregate av)+        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, Serialize e, Default a, Aggregate a, Eventable a e) => GUID -> TransactionT e IO (AggregateRef a e)+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@@ -84,26 +87,26 @@  -- 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, Serialize e, Default a, Aggregate a) => GUID -> Transaction e IO (Int,a)+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 (v',a') -> do-      case decodeAggregate a' :: Maybe a of-        Just a'' -> return (v', a'')+    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, Serialize e, Default a, Aggregate a, Eventable a e) => GUID -> Transaction e IO (AggregateRef a e)+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 $ ES.retrieveEvents es guid v0+  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@@ -115,7 +118,7 @@   return a'  -- | Publish event for an aggregate root.-publishEvent :: (MonadIO m, Serialize e, Typeable a, Typeable e, Aggregate a, Eventable a e, Default a) => AggregateRef a e -> e -> TransactionT e m ()+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@@ -124,7 +127,7 @@   modify $ \s -> s { tsCurrentVersion = currentGlobalVersion + 1 }  -- | Get aggregate root.-getAggregateRoot :: (Default a, Serialize e, Typeable a, Typeable e, Aggregate a, Eventable a e) => GUID -> TransactionT e IO (AggregateRef a e, a)+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