eventsource-api 1.1.1 → 1.2.0

raw patch · 8 files changed

+724/−155 lines, 8 filesdep +bytestringdep +enclosed-exceptionsdep +lifted-asyncdep −protoludePVP ok

version bump matches the API change (PVP)

Dependencies added: bytestring, enclosed-exceptions, lifted-async, lifted-base, monad-control, monad-loops, stm, stm-chans, string-conversions, text, transformers-base

Dependencies removed: protolude

API changes (from Hackage documentation)

Files

• CHANGELOG.md +4/−0
• eventsource-api.cabal +24/−10
• library/EventSource/Aggregate.hs +399/−0
• library/EventSource/Aggregate/Simple.hs +153/−0
• library/EventSource/Store.hs +96/−47
• library/EventSource/Types.hs +35/−26
• package.yaml +13/−6
• stack.yaml +0/−66

@@ -1,3 +1,7 @@+1.2.0+=====+  * Introduce Aggregate API.+ 1.1.1 =====   * Fix GHC 8.2.1 and Stackage LTS-9 build.

@@ -1,11 +1,13 @@--- This file has been generated from package.yaml by hpack version 0.17.1.+-- This file has been generated from package.yaml by hpack version 0.20.0. -- -- see: https://github.com/sol/hpack+--+-- hash: 210f402f20d9901165999ae9e117c9e55c55fc725f1ecf4f895f42ec9c3e41d0  name:           eventsource-api-version:        1.1.1+version:        1.2.0 synopsis:       Provides an eventsourcing high level API.-description:    Please read README.md.+description:    A high-level eventsourcing library. category:       Eventsourcing homepage:       https://github.com/YoEight/eventsource-api#readme bug-reports:    https://github.com/YoEight/eventsource-api/issues@@ -21,7 +23,6 @@     LICENSE.md     package.yaml     README.md-    stack.yaml  source-repository head   type: git@@ -30,18 +31,31 @@ library   hs-source-dirs:       library-  default-extensions: NoImplicitPrelude   ghc-options: -Wall   build-depends:-      base >=4.9 && <5-    , protolude >= 0.1.10 && <0.3-    , uuid-    , aeson-    , mtl+      aeson+    , base >=4.9 && <5+    , bytestring     , containers+    , enclosed-exceptions+    , lifted-async+    , lifted-base+    , monad-control+    , monad-loops ==0.4.*+    , mtl+    , stm+    , stm-chans+    , string-conversions+    , text+    , transformers-base     , unordered-containers+    , uuid   exposed-modules:       EventSource+      EventSource.Aggregate+      EventSource.Aggregate.Simple       EventSource.Store       EventSource.Types+  other-modules:+      Paths_eventsource_api   default-language: Haskell2010

@@ -0,0 +1,399 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE GADTs            #-}+{-# LANGUAGE Rank2Types       #-}+{-# LANGUAGE TypeFamilies     #-}+--------------------------------------------------------------------------------+-- |+-- Module    :  EventSource.Aggregate+-- Copyright :  (C) 2017 Yorick Laupa+-- License   :  (see the file LICENSE)+-- Maintainer:  Yorick Laupa <yo.eight@gmail.com>+-- Stability :  experimental+-- Portability: non-portable+--+-- Implementation of an aggregate abstraction.+-- Link: https://en.wikipedia.org/wiki/Domain-driven_design.+--------------------------------------------------------------------------------+module EventSource.Aggregate+  ( StreamId(..)+  -- * Aggregate+  , Aggregate(..)+  , Validate(..)+  , Decision+  , Agg+  , aggId+  , runAgg+  , newAgg+  , loadAgg+  , loadOrCreateAgg+  -- * Interactions+  , submitCmd+  , submitEvt+  , snapshot+  , route+  , closeAgg+  , execute+  -- * Internal+  , Action'(..)+  , Action+  , askEnv+  , getState+  , putState+  , AggEnv(..)+  , AggState(..)+  , persist+  ) where++--------------------------------------------------------------------------------+import Control.Exception (SomeException, throwIO)+import Control.Monad (ap)+import Data.Foldable (for_, traverse_)++--------------------------------------------------------------------------------+import           Control.Concurrent.Async.Lifted (wait)+import qualified Control.Concurrent.Lifted as Concurrent+import           Control.Concurrent.STM (atomically)+import qualified Control.Concurrent.STM.TBMQueue as Queue+import           Control.Exception.Enclosed (tryAny)+import           Control.Monad.Base (MonadBase, liftBase)+import           Control.Monad.Except (runExceptT)+import           Control.Monad.Loops (whileJust_)+import           Control.Monad.Trans (MonadTrans(..))+import           Control.Monad.Trans.Control (MonadBaseControl)+import           Data.IORef.Lifted (IORef, newIORef, readIORef, atomicWriteIORef)++--------------------------------------------------------------------------------+import EventSource++--------------------------------------------------------------------------------+-- | Maps an id to a 'StreamName'.+class StreamId a where+  toStreamName :: a -> StreamName++--------------------------------------------------------------------------------+-- | Represents a stream aggregate. An aggregate can rebuild its internal state+--   by replaying all the stream's events that aggregate is responsible for.+class Aggregate a where+  -- | Type of the id associated to the aggregate.+  type Id a  :: *++  -- | Type of event handled by the aggregate.+  type Evt a :: *++  -- | Type of monad stack used by the aggregate.+  type M a :: * -> *++  -- | Given current aggregate state, updates it according to the event the+  --   aggregate receives.+  apply :: a -> Evt a -> M a a++--------------------------------------------------------------------------------+-- | When validating a command, tells if the command was valid. If the command+--   is valid, it returns an event. Otherwise, it returns an error.+type Decision a = Either (Err a) (Evt a)++--------------------------------------------------------------------------------+-- | Represents an aggregate that support validation. An aggregate that supports+--   validation can receive command and decide if it was valid or not. When the+--   validation is successful, The aggregate emits an event that will be+--   persisted and pass to 'apply' function.+class Aggregate a => Validate a where+  -- | Type of command supported by the aggregate.+  type Cmd a :: *++  -- | Type of error that aggregate can yield.+  type Err a :: *++  -- | Validates a command. If the command validation succeeds, it will emits+  --   an event. Otherwise, it will returns an error.+  validate :: a -> Cmd a -> M a (Decision a)++--------------------------------------------------------------------------------+-- | Internal aggregate action. An action is executed by an aggregate. An action+--   embodies fundamental operations like submitting event, validating command+--   or returning the current snapshot of an aggregate. Action are CPS-ed+--   encoded so the execution model can be flexible. An action can perform+--   synchronously or asynchronously.+newtype Action' e s m a =+  Action' { runAction :: e+                      -> s+                      -> (s -> a -> m ())+                      -> m () }++--------------------------------------------------------------------------------+instance Functor (Action' e s m) where+  fmap f (Action' k) = Action' $ \e s resp ->+    k e s (\s' a -> resp s' (f a))++--------------------------------------------------------------------------------+instance Applicative (Action' e s m) where+  pure  = return+  (<*>) = ap++--------------------------------------------------------------------------------+instance Monad (Action' e s m) where+  return a = Action' $ \_ s resp -> resp s a++  Action' k >>= f = Action' $ \e s resp ->+    k e s (\s' a -> runAction (f a) e s' resp)++--------------------------------------------------------------------------------+instance MonadTrans (Action' e s) where+  lift m = Action' $ \_ s resp -> m >>= resp s++--------------------------------------------------------------------------------+-- | Returns an action environment.+askEnv :: Action' e s m e+askEnv = Action' $ \e s resp -> resp s e++--------------------------------------------------------------------------------+-- | Returns an action current state.+getState :: Action' e s m s+getState = Action' $ \_ s resp -> resp s s++--------------------------------------------------------------------------------+-- | Set an action state.+putState :: s -> Action' e s m ()+putState s = Action' $ \_ _ resp -> resp s ()++--------------------------------------------------------------------------------+-- | An action configured to aggregate internal types.+type Action a r = Action' (AggEnv a) (AggState a) (M a) r++--------------------------------------------------------------------------------+-- | Aggregate internal environment.+data AggEnv a =+  AggEnv { aggEnvStore :: SomeStore+           -- ^ Handle to an eventstore.+         , aggEnvId :: Id a+           -- ^ Identification of the aggregate.+         }++--------------------------------------------------------------------------------+-- | Aggregate internal state.+data AggState a =+  AggState { aggStateVersion :: !ExpectedVersion+             -- ^ Expected version of the next write. This isn't expose to+             --   user and it's updated automatically.+           , aggState :: !a+             -- ^ Aggregate current state.+           }++--------------------------------------------------------------------------------+-- | A stream aggregate. An aggregate updates its internal based on the event+--  it receives. You can read its current state by using 'snapshot'. If it+--  supports validation, through 'Validated' typeclass, it can receive+--  command and emits an event if the command was successful. Otherwise, it will+--  yield an error. When receiving valid command, an aggregate will persist the+--  resulting event. An aggregate is only responsible of its own stream.+data Agg a where+  Agg :: AggEnv a+      -> M a ()+      -> IORef (Maybe SomeException) -- Last exception ever captured.+      -> (forall r. Action a r -> (r -> M a ()) -> M a ())+      -> Agg a++--------------------------------------------------------------------------------+-- | Returns an aggregate id.+aggId :: Agg a -> Id a+aggId (Agg env _ _ _) = aggEnvId env++--------------------------------------------------------------------------------+-- | Executes an action on an aggregate.+runAgg :: MonadBase IO (M a) => Agg a -> Action a r -> (r -> M a ()) -> M a ()+runAgg (Agg _ _ errRef k) action resp = do+  errLast <- readIORef errRef+  traverse_ (liftBase . throwIO) errLast+  k action resp++--------------------------------------------------------------------------------+-- | Holds an existantially quantified action so it can passed around easily+--   to aggregate's internal concurrent channel.+data Msg a where+  Msg :: Action a r -> (r -> M a ()) -> Msg a++--------------------------------------------------------------------------------+-- | Creates a new aggregate given an eventstore handle, an id and an initial+--   state.+newAgg :: (Aggregate a, MonadBaseControl IO (M a))+       => SomeStore+       -> Id a+       -> a+       -> M a (Agg a)+newAgg store aId seed = do+  queue   <- liftBase $ Queue.newTBMQueueIO 500 -- Max parked messages.+  ref     <- newIORef (AggState AnyVersion seed)+  errLast <- newIORef Nothing++  let takeFromQueue  = liftBase $ atomically $ Queue.readTBMQueue queue+      closeAggThread = liftBase $ atomically $ Queue.closeTBMQueue queue+      env            = AggEnv store aId++  _ <- Concurrent.fork $ whileJust_ takeFromQueue $ \(Msg action k) ->+         do s   <- readIORef ref+            res <- tryAny $ runAction action env s $ \s' r -> do+              atomicWriteIORef ref s'+              k r++            case res of+              Left e ->+                do atomicWriteIORef errLast (Just e)+                   liftBase $ atomically $ Queue.closeTBMQueue queue++              _ -> pure ()++  pure $ Agg env closeAggThread errLast $ \action k ->+    liftBase $ atomically $ Queue.writeTBMQueue queue (Msg action k)++--------------------------------------------------------------------------------+-- | Creates an aggregate and replays its entire stream to rebuild its+--   internal state.+loadAgg :: (Aggregate a, StreamId (Id a), DecodeEvent (Evt a), MonadBaseControl IO (M a))+        => SomeStore+        -> Id a+        -> a+        -> M a (Either ForEventFailure (Agg a))+loadAgg store aId seed = do+  agg <- newAgg store aId seed+  res <- execute agg (loadEventsAction aId)++  -- We don't need to let a running thread for nothing if we couldn't load the+  -- aggregate properly.+  case res of+    Left _ -> closeAgg agg+    _      -> pure ()++  pure (agg <$ res)++--------------------------------------------------------------------------------+-- | Like 'loadAgg' but call 'loadAgg' in case of 'ForEventFailure' error.+loadOrCreateAgg :: (Aggregate a, StreamId (Id a), DecodeEvent (Evt a), MonadBaseControl IO (M a))+                => SomeStore+                -> Id a+                -> a+                -> M a (Agg a)+loadOrCreateAgg store aId seed = do+  agg <- newAgg store aId seed+  _   <- execute agg (loadEventsAction aId)+  pure agg++--------------------------------------------------------------------------------+-- | Submits a command to the aggregate. If the command was valid, it returns+-- an event otherwise an error. In case of a valid command, the aggregate+-- persist the resulting event to the eventstore. The aggregate will also+-- update its internal state accordingly.+submitCmd :: (Validate a, MonadBase IO (M a), StreamId (Id a), EncodeEvent (Evt a))+          => Agg a+          -> Cmd a+          -> M a (Decision a)+submitCmd agg cmd = execute agg (submitCmdAction cmd)++--------------------------------------------------------------------------------+-- | Submits an event. The aggregate will update its internal state accondingly.+submitEvt :: (Aggregate a, MonadBase IO (M a)) => Agg a -> Evt a -> M a ()+submitEvt agg evt = execute agg (submitEvtAction evt)++--------------------------------------------------------------------------------+-- | Returns current aggregate state.+snapshot :: MonadBase IO (M a) => Agg a -> M a a+snapshot agg = execute agg snapshotAction++--------------------------------------------------------------------------------+-- | Uses usually by Root aggregates which usually have unusual workflow and+--  make great use of a CPS-ed computation.+--   http://blog.sapiensworks.com/post/2016/07/14/DDD-Aggregate-Decoded-1+route :: MonadBase IO (M a)+      => Agg a+      -> (SomeStore -> a -> (a -> r -> M a ()) -> M a ())+      -> M a r+route agg k = execute agg (routeAction k)++--------------------------------------------------------------------------------+-- | Executes an action.+execute :: MonadBase IO (M a) => Agg a -> Action a r -> M a r+execute agg action = do+  var <- Concurrent.newEmptyMVar+  runAgg agg action (Concurrent.putMVar var)+  Concurrent.takeMVar var++--------------------------------------------------------------------------------+-- | Persists an event to the eventstore.+persist :: (StreamId id, EncodeEvent event, MonadBase IO m)+        => SomeStore+        -> id+        -> ExpectedVersion+        -> event+        -> m EventNumber+persist store aid ver event =+  liftBase (appendEvent store (toStreamName aid) ver event >>= wait)++--------------------------------------------------------------------------------+-- | Closes internal aggregate state.+closeAgg :: Agg a -> M a ()+closeAgg (Agg _ action _ _) = action++--------------------------------------------------------------------------------+-- // Internal commands.+--------------------------------------------------------------------------------+submitCmdAction :: (Validate a, MonadBase IO (M a), StreamId (Id a), EncodeEvent (Evt a))+                => Cmd a+                -> Action a (Decision a)+submitCmdAction cmd = do+  env    <- askEnv+  s      <- getState+  result <- lift $ validate (aggState s) cmd++  for_ result $ \event ->+    do next <- lift $ persist (aggEnvStore env)+                              (aggEnvId env)+                              (aggStateVersion s)+                              event++       let s' = s { aggStateVersion = ExactVersion next }++       putState s'+       submitEvtAction event++  pure result++--------------------------------------------------------------------------------+submitEvtAction :: (Aggregate a, Monad (M a)) => Evt a -> Action a ()+submitEvtAction event = do+  s  <- getState+  a' <- lift $ apply (aggState s) event+  let s' = s { aggState = a' }+  putState s'++--------------------------------------------------------------------------------+snapshotAction :: Monad (M a) => Action a a+snapshotAction = fmap aggState getState++--------------------------------------------------------------------------------+loadEventsAction :: (Aggregate a, StreamId (Id a), DecodeEvent (Evt a), MonadBase IO (M a))+                 => Id a+                 -> Action a (Either ForEventFailure a)+loadEventsAction aId = do+  seed <- getState+  env  <- askEnv++  let go num s event =+        do a <- apply (aggState s) event+           pure s { aggState        = a+                  , aggStateVersion = ExactVersion num+                  }++  res <- lift $ runExceptT+       $ foldEventsWithNumberM (aggEnvStore env) (toStreamName aId) go seed++  traverse_ putState res+  pure (fmap aggState res)++--------------------------------------------------------------------------------+routeAction :: (SomeStore -> a -> (a -> r -> M a ()) -> M a ())+            -> Action a r+routeAction k = Action' $ \env s resp -> do+  k (aggEnvStore env) (aggState s) $ \a r ->+    let s' = s { aggState = a } in+    resp s' r++

@@ -0,0 +1,153 @@+{-# LANGUAGE FunctionalDependencies #-}+{-# LANGUAGE MultiParamTypeClasses  #-}+{-# LANGUAGE TypeFamilies           #-}+--------------------------------------------------------------------------------+-- |+-- Module    :  EventSource.Aggregate.Simple+-- Copyright :  (C) 2017 Yorick Laupa+-- License   :  (see the file LICENSE)+-- Maintainer:  Yorick Laupa <yo.eight@gmail.com>+-- Stability :  experimental+-- Portability: non-portable+--+--------------------------------------------------------------------------------+module EventSource.Aggregate.Simple+  ( AggIO+  , AggregateIO(..)+  , ValidateIO(..)+  , Simple+  , newAgg+  , loadAgg+  , loadOrCreateAgg+  , submitCmd+  , submitEvt+  , closeAgg+  , snapshot+  , route+  ) where++--------------------------------------------------------------------------------+import Control.Exception (SomeException)++--------------------------------------------------------------------------------+import qualified EventSource.Aggregate as Self+import           EventSource++--------------------------------------------------------------------------------+-- | Simple aggregate abstraction.+newtype Simple id command event state = Simple state++--------------------------------------------------------------------------------+-- | A stream aggregate. An aggregate updates its internal based on the event+--   it receives. You can read its current state by using 'snapshot'. If it+--   supports validation, through 'Validated' typeclass, it can receive+--   command and emits an event if the command was successful. Otherwise, it will+--   yield an error. When receiving valid command, an aggregate will persist the+--   resulting event. An aggregate is only responsible of its own stream.+type AggIO id command event state = Self.Agg (Simple id command event state)++--------------------------------------------------------------------------------+-- | Represents a stream aggregate. An aggregate can rebuild its internal state+--   by replaying all the stream's events that aggregate is responsible for.+class AggregateIO event state | event -> state where++  -- | Given current aggregate state, updates it according to the event the+  --   aggregate receives.+  applyIO :: state -> event -> IO state++--------------------------------------------------------------------------------+-- | Represents an aggregate that support validation. An aggregate that supports+--   validation can receive command and decide if it was valid or not. When the+--   validation is successful, The aggregate emits an event that will be+--   persisted and pass to 'apply' function.+class AggregateIO event state => ValidateIO command event state | command -> state, command -> event where++  -- | Validates a command. If the command validation succeeds, it will emits+  --   an event. Otherwise, it will returns an error.+  validateIO :: state -> command -> IO (Either SomeException event)++--------------------------------------------------------------------------------+instance AggregateIO event state => Self.Aggregate (Simple id command event state) where+  type Id  (Simple id command event state) = id+  type Evt (Simple id command event state) = event+  type M   (Simple id command event state) = IO++  apply (Simple a) e = fmap Simple (applyIO a e)++--------------------------------------------------------------------------------+instance ValidateIO command event state => Self.Validate (Simple id command event state) where+  type Cmd (Simple id command event state) = command+  type Err (Simple id command event state) = SomeException++  validate (Simple a) cmd = validateIO a cmd++--------------------------------------------------------------------------------+-- | Creates a new aggregate given an eventstore handle, an id and an initial+--   state.+newAgg :: AggregateIO event state+       => SomeStore+       -> id+       -> state+       -> IO (AggIO id command event state)+newAgg store aId seed = Self.newAgg store aId (Simple seed)++--------------------------------------------------------------------------------+-- | Creates an aggregate and replays its entire stream to rebuild its+--   internal state.+loadAgg :: (AggregateIO event state, Self.StreamId id, DecodeEvent event)+        => SomeStore+        -> id+        -> state+        -> IO (Either ForEventFailure (AggIO id command event state))+loadAgg store aId seed = Self.loadAgg store aId (Simple seed)++--------------------------------------------------------------------------------+-- | Like 'loadAgg' but call 'loadAgg' in case of 'ForEventFailure' error.+loadOrCreateAgg :: (AggregateIO event state, Self.StreamId id, DecodeEvent event)+                => SomeStore+                -> id+                -> state+                -> IO (AggIO id command event state)+loadOrCreateAgg store aId seed = Self.loadOrCreateAgg store aId (Simple seed)++--------------------------------------------------------------------------------+-- | Submits a command to the aggregate. If the command was valid, it returns+--   an event otherwise an error. In case of a valid command, the aggregate+--   persist the resulting event to the eventstore. The aggregate will also+--   update its internal state accordingly.+submitCmd :: (ValidateIO command event state, Self.StreamId id, EncodeEvent event)+          => AggIO id command event state+          -> command+          -> IO (Either SomeException event)+submitCmd agg cmd = Self.submitCmd agg cmd++--------------------------------------------------------------------------------+-- | Submits an event. The aggregate will update its internal state accondingly.+submitEvt :: AggregateIO event state+          => AggIO id command event state+          -> event+          -> IO ()+submitEvt agg event = Self.submitEvt agg event++--------------------------------------------------------------------------------+-- | Returns current aggregate state.+snapshot :: AggIO id command event state -> IO state+snapshot agg = do+  Simple a <- Self.snapshot agg+  pure a++--------------------------------------------------------------------------------+-- | Closes internal aggregate state.+closeAgg :: AggIO id command event state -> IO ()+closeAgg agg = Self.closeAgg agg++--------------------------------------------------------------------------------+-- | Uses usually by Root aggregates which usually have unusual workflow and+--  make great use of a CPS-ed computation.+--   http://blog.sapiensworks.com/post/2016/07/14/DDD-Aggregate-Decoded-1+route :: AggIO id command event state+      -> (SomeStore -> state -> (state -> r -> IO ()) -> IO ())+      -> IO r+route agg k = Self.route agg $ \store (Simple state) resp ->+  k store state (\state' -> resp (Simple state'))+

@@ -1,5 +1,6 @@ {-# LANGUAGE ExistentialQuantification #-} {-# LANGUAGE FlexibleContexts          #-}+{-# LANGUAGE LambdaCase                #-} {-# LANGUAGE Rank2Types                #-} -------------------------------------------------------------------------------- -- |@@ -32,6 +33,8 @@   , foldSubAsync   , appendEvent   , forEvents+  , forEventsWithNumber+  , foldEventsWithNumberM   , foldEventsM   , foldEvents   , forSavedEvents@@ -40,17 +43,28 @@   , foldSubSaved   , foldSubSavedAsync   , ForEventFailure(..)+  , unhandled   ) where  ---------------------------------------------------------------------------------import Prelude (Show(..))+import Control.Monad (MonadPlus, mzero)+import Control.Exception (Exception, SomeException, toException, throwIO)+import Data.Bifunctor (first)+import Data.Foldable (for_)+import Data.Int (Int32)+import Data.Traversable (for)  ---------------------------------------------------------------------------------import Control.Monad.Except-import Data.IORef-import Data.UUID-import Data.UUID.V4-import Protolude hiding (from, show, trans)+import Control.Concurrent.Async.Lifted (Async, async, wait)+import Control.Monad.Base (MonadBase, liftBase)+import Control.Monad.Except (ExceptT, runExceptT, mapExceptT, throwError)+import Control.Monad.State (get, put, evalStateT)+import Control.Monad.Trans (lift)+import Control.Monad.Trans.Control (MonadBaseControl, StM)+import Data.IORef.Lifted (IORef, newIORef, atomicModifyIORef')+import Data.Text (Text)+import Data.UUID (UUID)+import Data.UUID.V4 (nextRandom)  -------------------------------------------------------------------------------- import EventSource.Types@@ -74,21 +88,21 @@  -------------------------------------------------------------------------------- -- | Returns a fresh subscription id.-freshSubscriptionId :: MonadIO m => m SubscriptionId-freshSubscriptionId = liftIO $ fmap SubscriptionId nextRandom+freshSubscriptionId :: MonadBase IO m => m SubscriptionId+freshSubscriptionId = liftBase $ fmap SubscriptionId nextRandom  -------------------------------------------------------------------------------- -- | A subscription allows to be notified on every change occuring on a stream. data Subscription =   Subscription { subscriptionId :: SubscriptionId -               , nextEvent :: forall m. MonadIO m+               , nextEvent :: forall m. MonadBase IO m                            => m (Either SomeException SavedEvent)                }  -------------------------------------------------------------------------------- -- | Waits for the next event and deserializes it on the go.-nextEventAs :: (DecodeEvent a, MonadIO m)+nextEventAs :: (DecodeEvent a, MonadBase IO m)             => Subscription             -> m (Either SomeException a) nextEventAs sub = do@@ -107,7 +121,7 @@ --   'SomeException' is used because we let the underlying subscription model --   exposed its own 'Exception'. If the callback that handles incoming events --   throws an exception, it will not be catch by the error callback.-foldSub :: (DecodeEvent a, MonadIO m)+foldSub :: (DecodeEvent a, MonadBase IO m)         => Subscription         -> (a -> m ())         -> (SomeException -> m ())@@ -122,18 +136,18 @@  -------------------------------------------------------------------------------- -- | Asynchronous version of 'foldSub'.-foldSubAsync :: DecodeEvent a+foldSubAsync :: (MonadBaseControl IO m, DecodeEvent a)              => Subscription-             -> (a -> IO ())-             -> (SomeException -> IO ())-             -> IO (Async ())+             -> (a -> m ())+             -> (SomeException -> m ())+             -> m (Async (StM m ())) foldSubAsync sub onEvent onError =   async $ foldSub sub onEvent onError  -------------------------------------------------------------------------------- -- | Similar to 'foldSub' but provides access to the 'SavedEvent' instead of --   decoded event.-foldSubSaved :: (MonadIO m)+foldSubSaved :: MonadBase IO m              => Subscription              -> (SavedEvent -> m ())              -> (SomeException -> m ())@@ -148,10 +162,11 @@  -------------------------------------------------------------------------------- -- | Asynchronous version of 'foldSubSaved'.-foldSubSavedAsync :: Subscription-                  -> (SavedEvent -> IO ())-                  -> (SomeException -> IO ())-                  -> IO (Async ())+foldSubSavedAsync :: MonadBaseControl IO m+                  => Subscription+                  -> (SavedEvent -> m ())+                  -> (SomeException -> m ())+                  -> m (Async (StM m ())) foldSubSavedAsync sub onEvent onError =   async $ foldSubSaved sub onEvent onError @@ -170,7 +185,7 @@ --   an event store. class Store store where   -- | Appends a batch of events at the end of a stream.-  appendEvents :: (EncodeEvent a, MonadIO m)+  appendEvents :: (EncodeEvent a, MonadBase IO m)                => store                -> StreamName                -> ExpectedVersion@@ -178,15 +193,19 @@                -> m (Async EventNumber)    -- | Appends a batch of events at the end of a stream.-  readBatch :: MonadIO m+  readBatch :: MonadBase IO m             => store             -> StreamName             -> Batch             -> m (Async (ReadStatus Slice))    -- | Subscribes to given stream.-  subscribe :: MonadIO m => store -> StreamName -> m Subscription+  subscribe :: MonadBase IO m => store -> StreamName -> m Subscription +  -- | Encapsulates to an abstract store.+  toStore :: store -> SomeStore+  toStore = SomeStore+ -------------------------------------------------------------------------------- -- | Utility type to pass any store that implements 'Store' typeclass. data SomeStore = forall store. Store store => SomeStore store@@ -199,7 +218,7 @@  -------------------------------------------------------------------------------- -- | Appends a single event at the end of a stream.-appendEvent :: (EncodeEvent a, MonadIO m, Store store)+appendEvent :: (EncodeEvent a, MonadBase IO m, Store store)             => store             -> StreamName             -> ExpectedVersion@@ -219,12 +238,22 @@  -------------------------------------------------------------------------------- -- | Iterates over all events of stream given a starting point and a batch size.-forEvents :: (MonadIO m, DecodeEvent a, Store store)+forEvents :: (MonadBase IO m, DecodeEvent a, Store store)           => store           -> StreamName           -> (a -> m ())           -> ExceptT ForEventFailure m ()-forEvents store name k = do+forEvents store name k = forEventsWithNumber store name (const k)++--------------------------------------------------------------------------------+-- | Iterates over all events of stream given a starting point and a batch size.+--   It also passes the 'EventNumber' at each recursion step.+forEventsWithNumber :: (MonadBase IO m, DecodeEvent a, Store store)+                    => store+                    -> StreamName+                    -> (EventNumber -> a -> m ())+                    -> ExceptT ForEventFailure m ()+forEventsWithNumber store name k = do   res <- streamIterator store name   case res of     ReadSuccess i -> loop i@@ -235,30 +264,43 @@       for_ opt $ \saved ->         case decodeEvent $ savedEvent saved of           Left e -> throwError $ ForEventDecodeFailure e-          Right a -> lift (k a) >> loop i+          Right a -> lift (k (eventNumber saved) a) >> loop i  -------------------------------------------------------------------------------- -- | Like 'forEvents' but expose signature similar to 'foldM'.-foldEventsM :: (MonadIO m, DecodeEvent a, Store store)+foldEventsM :: (MonadBase IO m, DecodeEvent a, Store store)             => store             -> StreamName             -> (s -> a -> m s)             -> s             -> ExceptT ForEventFailure m s-foldEventsM store stream k seed = mapExceptT trans action+foldEventsM store stream k seed =+  foldEventsWithNumberM store stream (const k) seed++--------------------------------------------------------------------------------+-- | Like 'forEvents' but expose signature similar to 'foldM' and also passes+--   the 'EventNumber' at each recursion step.+foldEventsWithNumberM :: (MonadBase IO m, DecodeEvent a, Store store)+            => store+            -> StreamName+            -> (EventNumber -> s -> a -> m s)+            -> s+            -> ExceptT ForEventFailure m s+foldEventsWithNumberM store stream k seed = mapExceptT trans action   where     trans m = evalStateT m seed      action = do-      forEvents store stream $ \a -> do+      forEventsWithNumber store stream $ \num a -> do         s <- get-        s' <- lift $ k s a+        s' <- lift $ k num s a         put s'       get + -------------------------------------------------------------------------------- -- | Like 'foldEventsM' but expose signature similar to 'foldl'.-foldEvents :: (MonadIO m, DecodeEvent a, Store store)+foldEvents :: (MonadBase IO m, DecodeEvent a, Store store)            => store            -> StreamName            -> (s -> a -> s)@@ -270,7 +312,7 @@ -------------------------------------------------------------------------------- -- | Like `forEvents` but provides access to 'SavedEvent' instead of --   decoded event.-forSavedEvents :: (MonadIO m, Store store)+forSavedEvents :: (MonadBase IO m, Store store)                => store                -> StreamName                -> (SavedEvent -> m ())@@ -287,7 +329,7 @@  -------------------------------------------------------------------------------- -- | Like 'forSavedEvents' but expose signature similar to 'foldM'.-foldSavedEventsM :: (MonadIO m, Store store)+foldSavedEventsM :: (MonadBase IO m, Store store)                  => store                  -> StreamName                  -> (s -> SavedEvent -> m s)@@ -306,7 +348,7 @@  -------------------------------------------------------------------------------- -- | Like 'foldSavedEventsM' but expose signature similar to 'foldl'.-foldSavedEvents :: (MonadIO m, Store store)+foldSavedEvents :: (MonadBase IO m, Store store)                 => store                 -> StreamName                 -> (s -> SavedEvent -> s)@@ -316,9 +358,16 @@   foldSavedEventsM store stream (\s a -> return $ k s a) seed  --------------------------------------------------------------------------------+-- | Throws an exception in case 'ExceptT' was a 'Left'.+unhandled :: (MonadBase IO m, Exception e) => ExceptT e m a -> m a+unhandled m = runExceptT m >>= \case+  Left e  -> liftBase $ throwIO e+  Right a -> pure a++-------------------------------------------------------------------------------- -- | Allows to easily iterate over a stream's events. newtype StreamIterator =-  StreamIterator { iteratorNext :: forall m. MonadIO m => m (Maybe SavedEvent) }+  StreamIterator { iteratorNext :: forall m. MonadBase IO m => m (Maybe SavedEvent) }  -------------------------------------------------------------------------------- instance Show StreamIterator where@@ -327,7 +376,7 @@ -------------------------------------------------------------------------------- -- | Reads the next available event from the 'StreamIterator' and try to --   deserialize it at the same time.-iteratorNextEvent :: (DecodeEvent a, MonadIO m, MonadPlus m)+iteratorNextEvent :: (DecodeEvent a, MonadBase IO m, MonadPlus m)                   => StreamIterator                   -> m (Maybe a) iteratorNextEvent i = do@@ -341,7 +390,7 @@  -------------------------------------------------------------------------------- -- | Reads all events from the 'StreamIterator' until reaching end of stream.-iteratorReadAll :: MonadIO m => StreamIterator -> m [SavedEvent]+iteratorReadAll :: MonadBase IO m => StreamIterator -> m [SavedEvent] iteratorReadAll i = do   res <- iteratorNext i   case res of@@ -350,7 +399,7 @@  -------------------------------------------------------------------------------- -- | Like 'iteratorReadAll' but try to deserialize the events at the same time.-iteratorReadAllEvents :: (DecodeEvent a, MonadIO m, MonadPlus m)+iteratorReadAllEvents :: (DecodeEvent a, MonadBase IO m, MonadPlus m)                       => StreamIterator                       -> m [a] iteratorReadAllEvents i = do@@ -362,15 +411,15 @@ -------------------------------------------------------------------------------- -- | Returns a 'StreamIterator' for the given stream name. The returned --   'StreamIterator' IS NOT THREADSAFE.-streamIterator :: (Store store, MonadIO m)+streamIterator :: (Store store, MonadBase IO m)                => store                -> StreamName                -> m (ReadStatus StreamIterator) streamIterator store name = do   w <- readBatch store name (startFrom 0)-  res <- liftIO $ wait w+  res <- liftBase $ wait w   for res $ \slice -> do-    ref <- liftIO $ newIORef $ IteratorOverAvailable slice+    ref <- newIORef $ IteratorOverAvailable slice     return $ StreamIterator $ iterateOver store ref name  --------------------------------------------------------------------------------@@ -385,7 +434,7 @@   | IteratorOverEndOfStream  ---------------------------------------------------------------------------------iterateOver :: (Store store, MonadIO m)+iterateOver :: (Store store, MonadBase IO m)             => store             -> IORef IteratorOverState             -> StreamName@@ -393,7 +442,7 @@ iterateOver store ref name = go   where     go = do-      action <- liftIO $ atomicModifyIORef' ref $ \st ->+      action <- atomicModifyIORef' ref $ \st ->         case st of           IteratorOverAvailable slice ->             case sliceEvents slice of@@ -414,12 +463,12 @@         IteratorOverEndOfStream -> return Nothing         IteratorOverNextBatch num -> do           w <- readBatch store name (startFrom num)-          res <- liftIO $ wait w+          res <- liftBase $ wait w           case res of             ReadFailure _ -> do-              liftIO $ atomicModifyIORef' ref $ \_ -> (IteratorOverClosed, ())+              atomicModifyIORef' ref $ \_ -> (IteratorOverClosed, ())               return Nothing             ReadSuccess slice -> do               let nxtSt = IteratorOverAvailable slice-              liftIO $ atomicModifyIORef' ref $ \_ -> (nxtSt, ())+              atomicModifyIORef' ref $ \_ -> (nxtSt, ())               go

@@ -1,3 +1,4 @@+{-# LANGUAGE FlexibleContexts           #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE OverloadedStrings          #-} --------------------------------------------------------------------------------@@ -14,18 +15,26 @@ module EventSource.Types where  ---------------------------------------------------------------------------------import Prelude (Show(..))-import Data.Foldable-import Data.String+import Control.Exception (Exception)+import Control.Monad (MonadPlus, mzero)+import Data.Bifunctor (first)+import Data.Foldable (foldlM)+import Data.Int (Int32)+import Data.String (IsString(..))+import Data.String.Conversions (convertString)  ---------------------------------------------------------------------------------import Protolude hiding (show)-import Data.Aeson-import Data.Aeson.Types-import Data.UUID hiding (fromString)-import Data.UUID.V4-import qualified Data.HashMap.Strict as H-import qualified Data.Map.Strict as M+import           Control.Monad.Base (MonadBase, liftBase)+import           Control.Monad.State.Strict (State, modify, put)+import           Data.Aeson (ToJSON(..), FromJSON(..), Value, (.=), (.:))+import qualified Data.Aeson as Aeson+import           Data.Aeson.Types (Parser, parseEither)+import           Data.ByteString (ByteString)+import qualified Data.HashMap.Strict as HashMap+import qualified Data.Map.Strict as Map+import           Data.Text (Text)+import           Data.UUID (UUID, toText, fromText)+import           Data.UUID.V4 (nextRandom)  -------------------------------------------------------------------------------- -- | Opaque data type used to store raw data.@@ -68,7 +77,7 @@ -- | Returns 'Data' content as a 'ByteString'. dataAsBytes :: Data -> ByteString dataAsBytes (Data bs) = bs-dataAsBytes (DataAsJson v) = toS $ encode v+dataAsBytes (DataAsJson v) = convertString $ Aeson.encode v  -------------------------------------------------------------------------------- -- | Creates a 'Data' object from a raw 'ByteString'.@@ -83,15 +92,15 @@ -------------------------------------------------------------------------------- -- | Returns 'Data' content as any value that implements 'FromJSON' type-class. dataAsJson :: FromJSON a => Data -> Either Text a-dataAsJson (Data bs) = first toS $ eitherDecodeStrict bs-dataAsJson (DataAsJson v) = first toS $ parseEither parseJSON v+dataAsJson (Data bs) = first convertString $ Aeson.eitherDecodeStrict bs+dataAsJson (DataAsJson v) = first convertString $ parseEither parseJSON v  -------------------------------------------------------------------------------- -- | Uses a 'JsonParsing' comuputation to extract a value. dataAsParsing :: Data -> JsonParsing a -> Either Text a dataAsParsing dat (JsonParsing k) = do   value <- dataAsJson dat-  first toS $ parseEither k value+  first convertString $ parseEither k value  -------------------------------------------------------------------------------- -- | Like 'dataAsParsing' but doesn't require you to use 'JsonParsing'.@@ -101,7 +110,7 @@ -------------------------------------------------------------------------------- -- | Used to store a set a properties. One example is to be used as 'Event' --   metadata.-newtype Properties = Properties (Map Text Text)+newtype Properties = Properties (Map.Map Text Text)  -------------------------------------------------------------------------------- instance Monoid Properties where@@ -114,21 +123,21 @@  -------------------------------------------------------------------------------- instance ToJSON Properties where-  toJSON = object . fmap go . properties+  toJSON = Aeson.object . fmap go . properties     where       go (k, v) = k .= v  -------------------------------------------------------------------------------- instance FromJSON Properties where-  parseJSON = withObject "Properties" $ \o ->+  parseJSON = Aeson.withObject "Properties" $ \o ->     let go p k = fmap (\v -> setProperty k v p) (o .: k) in-    foldlM go mempty (H.keys o)+    foldlM go mempty (HashMap.keys o)  -------------------------------------------------------------------------------- -- | Retrieves a value associated with the given key. property :: MonadPlus m => Text -> Properties -> m Text property k (Properties m) =-  case M.lookup k m of+  case Map.lookup k m of     Nothing -> mzero     Just v -> return v @@ -140,12 +149,12 @@ -------------------------------------------------------------------------------- -- | Adds a pair of key-value into given 'Properties'. setProperty :: Text -> Text -> Properties -> Properties-setProperty key value (Properties m) = Properties $ M.insert key value m+setProperty key value (Properties m) = Properties $ Map.insert key value m  -------------------------------------------------------------------------------- -- | Returns all associated key-value pairs as a list. properties :: Properties -> [(Text, Text)]-properties (Properties m) = M.toList m+properties (Properties m) = Map.toList m  -------------------------------------------------------------------------------- -- | Used to identify an event.@@ -157,7 +166,7 @@  -------------------------------------------------------------------------------- instance FromJSON EventId where-  parseJSON = withText "EventId" $ \t ->+  parseJSON = Aeson.withText "EventId" $ \t ->     case fromText t of       Just u  -> return $ EventId u       Nothing -> mzero@@ -168,8 +177,8 @@  -------------------------------------------------------------------------------- -- | Generates a fresh 'EventId'.-freshEventId :: MonadIO m => m EventId-freshEventId = fmap EventId $ liftIO nextRandom+freshEventId :: MonadBase IO m => m EventId+freshEventId = fmap EventId $ liftBase nextRandom  -------------------------------------------------------------------------------- -- | Represents a stream name.@@ -316,9 +325,9 @@ -------------------------------------------------------------------------------- -- | Represents the different kind of failure you can get when reading. data ReadFailure-  = StreamNotFound+  = StreamNotFound StreamName   | ReadError (Maybe Text)-  | AccessDenied+  | AccessDenied StreamName   deriving Show  --------------------------------------------------------------------------------

@@ -2,26 +2,33 @@ # Cabal file when you run `stack build`. See the hpack website for help with # this file: <https://github.com/sol/hpack>. category: Eventsourcing-description: Please read README.md.+description: A high-level eventsourcing library. extra-source-files: - CHANGELOG.md - LICENSE.md - package.yaml - README.md-- stack.yaml ghc-options: -Wall github: YoEight/eventsource-api library:-  default-extensions:-    - NoImplicitPrelude   dependencies:     - base >=4.9 && <5-    - protolude >= 0.1.10 && <0.3     - uuid     - aeson     - mtl     - containers     - unordered-containers+    - lifted-async+    - lifted-base+    - monad-control+    - transformers-base+    - text+    - bytestring+    - string-conversions+    - stm-chans+    - stm+    - monad-loops ==0.4.*+    - enclosed-exceptions   source-dirs: library license: BSD3 license-file: LICENSE.md@@ -29,4 +36,4 @@ maintainer: yo.eight@gmail.com name: eventsource-api synopsis: Provides an eventsourcing high level API.-version: '1.1.1'+version: '1.2.0'

@@ -1,66 +0,0 @@-# This file was automatically generated by 'stack init'-#-# Some commonly used options have been documented as comments in this file.-# For advanced use and comprehensive documentation of the format, please see:-# http://docs.haskellstack.org/en/stable/yaml_configuration/--# Resolver to choose a 'specific' stackage snapshot or a compiler version.-# A snapshot resolver dictates the compiler version and the set of packages-# to be used for project dependencies. For example:-#-# resolver: lts-3.5-# resolver: nightly-2015-09-21-# resolver: ghc-7.10.2-# resolver: ghcjs-0.1.0_ghc-7.10.2-# resolver:-#  name: custom-snapshot-#  location: "./custom-snapshot.yaml"-resolver: nightly-2017-08-14--# User packages to be built.-# Various formats can be used as shown in the example below.-#-# packages:-# - some-directory-# - https://example.com/foo/bar/baz-0.0.2.tar.gz-# - location:-#    git: https://github.com/commercialhaskell/stack.git-#    commit: e7b331f14bcffb8367cd58fbfc8b40ec7642100a-# - location: https://github.com/commercialhaskell/stack/commit/e7b331f14bcffb8367cd58fbfc8b40ec7642100a-#   extra-dep: true-#  subdirs:-#  - auto-update-#  - wai-#-# A package marked 'extra-dep: true' will only be built if demanded by a-# non-dependency (i.e. a user package), and its test suites and benchmarks-# will not be run. This is useful for tweaking upstream packages.-packages:-- '.'-# Dependency packages to be pulled from upstream that are not in the resolver-# (e.g., acme-missiles-0.3)-extra-deps: [ protolude-0.2 ]--# Override default flag values for local packages and extra-deps-flags: {}--# Extra package databases containing global packages-extra-package-dbs: []--# Control whether we use the GHC we find on the path-# system-ghc: true-#-# Require a specific version of stack, using version ranges-# require-stack-version: -any # Default-# require-stack-version: ">=1.3"-#-# Override the architecture used by stack, especially useful on Windows-# arch: i386-# arch: x86_64-#-# Extra directories used by stack for building-# extra-include-dirs: [/path/to/dir]-# extra-lib-dirs: [/path/to/dir]-#-# Allow a newer minor version of GHC than the snapshot specifies-# compiler-check: newer-minor