eventsource-api 1.3.1 → 1.5.0

raw patch · 9 files changed

+168/−964 lines, 9 filesdep +streamingPVP ok

version bump matches the API change (PVP)

Dependencies added: streaming

API changes (from Hackage documentation)

Files

• CHANGELOG.md +5/−0
• README.md +13/−13
• eventsource-api.cabal +14/−18
• library/EventSource/Aggregate.hs +0/−399
• library/EventSource/Aggregate/Simple.hs +0/−153
• library/EventSource/Store.hs +25/−367
• library/EventSource/Store/Internal/Iterator.hs +86/−0
• library/EventSource/Types.hs +22/−12
• package.yaml +3/−2

@@ -1,3 +1,8 @@+1.5.0+=====+  * Introduce a streaming interface.+  * Remove aggregate interface.+ 1.3.1 =====   * Fix GHC 8.4 build.

@@ -1,4 +1,4 @@-# [eventsource-api][]+# eventsource-api  This project provides what we think be a lean eventsourcing API. The goal is to set a common ground for eventsourcing-based application, yet doesn't force you@@ -11,30 +11,32 @@ --   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                -> [a]                -> m (Async EventNumber) -  -- | Appends a batch of events at the end of a stream.-  readBatch :: MonadIO m-            => store-            -> StreamName-            -> Batch-            -> m (Async (ReadStatus Slice))+  -- | Reads a stream in a stream-processing fashion.+  readStream :: MonadBase IO m+             => store+             -> StreamName+             -> Batch+             -> Stream (Of SavedEvent) (ExceptT ReadFailure m) ()    -- | Subscribes to given stream.-  subscribe :: MonadIO m => store -> StreamName -> m Subscription+  subscribe :: MonadBase IO m => store -> StreamName -> m Subscription++ ```  The idea is to use any backend of your liking as long as it's able to derive that typeclass with its related implicit constraints.    1. `appendEvents` MUST add events at the end of a stream without doing any other alteration on the stream. The order of the event's array MUST be respected within the store. Most important, the implementation MUST comply to the given `ExpectedVersion` passed by the user.-  -  2. `readBatch` SHOULD retrieve a set of event given a starting position and a batch size (represented by the `Batch` parameter). The returned array MUST have its events ordered by their `EventNumber` property. +  2. `readStream` SHOULD retrieve a set of event given a starting position and a batch size (represented by the `Batch` parameter). The returned array MUST have its events ordered by their `EventNumber` property.+   3. `subscribe` SHOULD register a subscription to the given stream. An implementation MUST allow to subscribe to a stream that doesn't exist yet. A subscription allows the user to be notified of any event added to the stream. There is no mandatory timing to meet regarding how fast the subscription is notified by a change. We only suggest to dispatch new events as soon as possible.  #### `ExpectedVersion`@@ -56,5 +58,3 @@ ```  On a write, if the `ExpectedVersion` condition given by the user is not met within the store, the implementation should raise an exception. At the moment, the API doesn't capture this situation in the type system.--[eventsource-api]: https://github.com/YoEight/eventsource-api

@@ -1,33 +1,36 @@--- This file has been generated from package.yaml by hpack version 0.28.2.+cabal-version: 1.12++-- This file has been generated from package.yaml by hpack version 0.31.1. -- -- see: https://github.com/sol/hpack ----- hash: 25d3e7c905840b2773e327d1dc91e90a9be3ecf9ff4c72ad67d8f999f0f05b9b+-- hash: 139d55fed31480f82f48ab1b236c875dac7ae19b648a917872b898b4f8c5a00b  name:           eventsource-api-version:        1.3.1+version:        1.5.0 synopsis:       Provides an eventsourcing high level API. 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+homepage:       https://gitlab.com/YoEight/eventsource-api-hs author:         Yorick Laupa maintainer:     yo.eight@gmail.com license:        BSD3 license-file:   LICENSE.md build-type:     Simple-cabal-version:  >= 1.10 extra-source-files:     CHANGELOG.md     LICENSE.md     package.yaml     README.md -source-repository head-  type: git-  location: https://github.com/YoEight/eventsource-api- library+  exposed-modules:+      EventSource+      EventSource.Store+      EventSource.Store.Internal.Iterator+      EventSource.Types+  other-modules:+      Paths_eventsource_api   hs-source-dirs:       library   ghc-options: -Wall@@ -44,17 +47,10 @@     , mtl     , stm     , stm-chans+    , streaming     , 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

@@ -1,399 +0,0 @@-{-# 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--

@@ -1,153 +0,0 @@-{-# 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'))-

@@ -14,73 +14,37 @@ -- -------------------------------------------------------------------------------- module EventSource.Store-  ( Batch(..)+  ( Batch'(..)+  , Batch   , Subscription(..)   , SubscriptionId   , ExpectedVersionException(..)   , Store(..)   , SomeStore(..)-  , StreamIterator-  , iteratorNext-  , iteratorNextEvent-  , iteratorReadAll-  , iteratorReadAllEvents-  , streamIterator   , freshSubscriptionId   , startFrom-  , nextEventAs-  , foldSub-  , foldSubAsync   , appendEvent-  , forEvents-  , forEventsWithNumber-  , foldEventsWithNumberM-  , foldEventsM-  , foldEvents-  , forSavedEvents-  , foldSavedEventsM-  , foldSavedEvents-  , foldSubSaved-  , foldSubSavedAsync-  , ForEventFailure(..)   , unhandled   ) where  ---------------------------------------------------------------------------------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.Exception (Exception, throwIO)  ---------------------------------------------------------------------------------import Control.Concurrent.Async.Lifted (Async, async, wait)+import Control.Concurrent.Async.Lifted (Async) 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 Control.Monad.Except (ExceptT, runExceptT) import Data.UUID (UUID) import Data.UUID.V4 (nextRandom)+import Streaming (hoist)+import Streaming.Prelude (Stream, Of)  -------------------------------------------------------------------------------- import EventSource.Types------------------------------------------------------------------------------------- | Represents batch information needed to read a stream.-data Batch =-  Batch { batchFrom :: EventNumber-        , batchSize :: Int32-        }+import EventSource.Store.Internal.Iterator (Batch'(..), startFrom)  ----------------------------------------------------------------------------------- | Starts a 'Batch' from a given point. The batch size is set to default,---   which is 500.-startFrom :: EventNumber -> Batch-startFrom from = Batch from 500+type Batch = Batch' EventNumber  -------------------------------------------------------------------------------- -- | Represents a subscription id.@@ -96,81 +60,11 @@ data Subscription =   Subscription { subscriptionId :: SubscriptionId -               , nextEvent :: forall m. MonadBase IO m-                           => m (Either SomeException SavedEvent)+               , subscriptionStream+                    :: forall m. MonadBase IO m => Stream (Of SavedEvent) m ()                }  ----------------------------------------------------------------------------------- | Waits for the next event and deserializes it on the go.-nextEventAs :: (DecodeEvent a, MonadBase IO m)-            => Subscription-            -> m (Either SomeException a)-nextEventAs sub = do-  res <- nextEvent sub-  let action = do-        event <- res-        first (toException . DecodeEventException)-          $ decodeEvent-          $ savedEvent event--  return action------------------------------------------------------------------------------------- | Folds over every event coming from the subscription until the end of the---   universe, unless an 'Exception' raises from the subscription.---   '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, MonadBase IO m)-        => Subscription-        -> (a -> m ())-        -> (SomeException -> m ())-        -> m ()-foldSub sub onEvent onError = loop-  where-    loop = do-      res <- nextEventAs sub-      case res of-        Left e -> onError e-        Right a -> onEvent a >> loop------------------------------------------------------------------------------------- | Asynchronous version of 'foldSub'.-foldSubAsync :: (MonadBaseControl IO m, DecodeEvent a)-             => Subscription-             -> (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 :: MonadBase IO m-             => Subscription-             -> (SavedEvent -> m ())-             -> (SomeException -> m ())-             -> m ()-foldSubSaved sub onEvent onError = loop-  where-    loop = do-      res <- nextEvent sub-      case res of-        Left e -> onError e-        Right a -> onEvent a >> loop------------------------------------------------------------------------------------- | Asynchronous version of 'foldSubSaved'.-foldSubSavedAsync :: MonadBaseControl IO m-                  => Subscription-                  -> (SavedEvent -> m ())-                  -> (SomeException -> m ())-                  -> m (Async (StM m ()))-foldSubSavedAsync sub onEvent onError =-  async $ foldSubSaved sub onEvent onError---------------------------------------------------------------------------------- data ExpectedVersionException   = ExpectedVersionException     { versionExceptionExpected :: ExpectedVersion@@ -192,12 +86,12 @@                -> [a]                -> m (Async EventNumber) -  -- | Appends a batch of events at the end of a stream.-  readBatch :: MonadBase IO m-            => store-            -> StreamName-            -> Batch-            -> m (Async (ReadStatus Slice))+  -- | Reads a stream in a stream-processing fashion.+  readStream :: MonadBase IO m+             => store+             -> StreamName+             -> Batch+             -> Stream (Of SavedEvent) (ExceptT ReadFailure m) ()    -- | Subscribes to given stream.   subscribe :: MonadBase IO m => store -> StreamName -> m Subscription@@ -213,7 +107,7 @@ -------------------------------------------------------------------------------- instance Store SomeStore where   appendEvents (SomeStore store) = appendEvents store-  readBatch (SomeStore store)    = readBatch store+  readStream (SomeStore store)   = readStream store   subscribe (SomeStore store)    = subscribe store  --------------------------------------------------------------------------------@@ -227,248 +121,12 @@ appendEvent store stream ver a = appendEvents store stream ver [a]  ----------------------------------------------------------------------------------- | Represents failures that can occurs when using 'forEvents'.-data ForEventFailure-  = ForEventReadFailure ReadFailure-  | ForEventDecodeFailure Text-  deriving Show-----------------------------------------------------------------------------------instance Exception ForEventFailure------------------------------------------------------------------------------------- | Iterates over all events of stream given a starting point and a batch size.-forEvents :: (MonadBase IO m, DecodeEvent a, Store store)-          => store-          -> StreamName-          -> (a -> m ())-          -> ExceptT ForEventFailure m ()-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-    ReadFailure e -> throwError $ ForEventReadFailure e-  where-    loop i = do-      opt <- iteratorNext i-      for_ opt $ \saved ->-        case decodeEvent $ savedEvent saved of-          Left e -> throwError $ ForEventDecodeFailure e-          Right a -> lift (k (eventNumber saved) a) >> loop i------------------------------------------------------------------------------------- | Like 'forEvents' but expose signature similar to 'foldM'.-foldEventsM :: (MonadBase IO m, DecodeEvent a, Store store)-            => store-            -> StreamName-            -> (s -> a -> m s)-            -> s-            -> ExceptT ForEventFailure m s-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-      forEventsWithNumber store stream $ \num a -> do-        s <- get-        s' <- lift $ k num s a-        put s'-      get-------------------------------------------------------------------------------------- | Like 'foldEventsM' but expose signature similar to 'foldl'.-foldEvents :: (MonadBase IO m, DecodeEvent a, Store store)-           => store-           -> StreamName-           -> (s -> a -> s)-           -> s-           -> ExceptT ForEventFailure m s-foldEvents store stream k seed =-  foldEventsM store stream (\s a -> return $ k s a) seed------------------------------------------------------------------------------------- | Like `forEvents` but provides access to 'SavedEvent' instead of---   decoded event.-forSavedEvents :: (MonadBase IO m, Store store)-               => store-               -> StreamName-               -> (SavedEvent -> m ())-               -> ExceptT ForEventFailure m ()-forSavedEvents store name k = do-  res <- streamIterator store name-  case res of-    ReadSuccess i -> loop i-    ReadFailure e -> throwError $ ForEventReadFailure e-  where-    loop i = do-      opt <- iteratorNext i-      for_ opt $ \saved -> lift (k saved) >> loop i------------------------------------------------------------------------------------- | Like 'forSavedEvents' but expose signature similar to 'foldM'.-foldSavedEventsM :: (MonadBase IO m, Store store)-                 => store-                 -> StreamName-                 -> (s -> SavedEvent -> m s)-                 -> s-                 -> ExceptT ForEventFailure m s-foldSavedEventsM store stream k seed = mapExceptT trans action-  where-    trans m = evalStateT m seed--    action = do-      forSavedEvents store stream $ \a -> do-        s <- get-        s' <- lift $ k s a-        put s'-      get------------------------------------------------------------------------------------- | Like 'foldSavedEventsM' but expose signature similar to 'foldl'.-foldSavedEvents :: (MonadBase IO m, Store store)-                => store-                -> StreamName-                -> (s -> SavedEvent -> s)-                -> s-                -> ExceptT ForEventFailure m s-foldSavedEvents store stream k seed =-  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. MonadBase IO m => m (Maybe SavedEvent) }-----------------------------------------------------------------------------------instance Show StreamIterator where-  show _ = "StreamIterator"------------------------------------------------------------------------------------- | Reads the next available event from the 'StreamIterator' and try to---   deserialize it at the same time.-iteratorNextEvent :: (DecodeEvent a, MonadBase IO m, MonadPlus m)-                  => StreamIterator-                  -> m (Maybe a)-iteratorNextEvent i = do-  res <- iteratorNext i-  case res of-    Nothing -> return Nothing-    Just s ->-      case decodeEvent $ savedEvent s of-        Left _ -> mzero-        Right a -> return $ Just a------------------------------------------------------------------------------------- | Reads all events from the 'StreamIterator' until reaching end of stream.-iteratorReadAll :: MonadBase IO m => StreamIterator -> m [SavedEvent]-iteratorReadAll i = do-  res <- iteratorNext i-  case res of-    Nothing -> return []-    Just s -> fmap (s:) $ iteratorReadAll i------------------------------------------------------------------------------------- | Like 'iteratorReadAll' but try to deserialize the events at the same time.-iteratorReadAllEvents :: (DecodeEvent a, MonadBase IO m, MonadPlus m)-                      => StreamIterator-                      -> m [a]-iteratorReadAllEvents i = do-  res <- iteratorNextEvent i-  case res of-    Nothing -> return []-    Just a -> fmap (a:) $ iteratorReadAllEvents i------------------------------------------------------------------------------------- | Returns a 'StreamIterator' for the given stream name. The returned---   'StreamIterator' IS NOT THREADSAFE.-streamIterator :: (Store store, MonadBase IO m)-               => store-               -> StreamName-               -> m (ReadStatus StreamIterator)-streamIterator store name = do-  w <- readBatch store name (startFrom 0)-  res <- liftBase $ wait w-  for res $ \slice -> do-    ref <- newIORef $ IteratorOverAvailable slice-    return $ StreamIterator $ iterateOver store ref name-----------------------------------------------------------------------------------data IteratorOverState-  = IteratorOverAvailable Slice-  | IteratorOverClosed-----------------------------------------------------------------------------------data IteratorOverAction-  = IteratorOverEvent SavedEvent-  | IteratorOverNextBatch EventNumber-  | IteratorOverEndOfStream-----------------------------------------------------------------------------------iterateOver :: (Store store, MonadBase IO m)-            => store-            -> IORef IteratorOverState-            -> StreamName-            -> m (Maybe SavedEvent)-iterateOver store ref name = go+unhandled :: (MonadBase IO m, Exception e)+          => Stream (Of a) (ExceptT e m) ()+          -> Stream (Of a) m ()+unhandled = hoist go   where-    go = do-      action <- atomicModifyIORef' ref $ \st ->-        case st of-          IteratorOverAvailable slice ->-            case sliceEvents slice of-              e:es ->-                let nextSlice = slice { sliceEvents = es }-                    nxtSt = IteratorOverAvailable nextSlice in-                (nxtSt, IteratorOverEvent e)-              [] | sliceEndOfStream slice-                   -> (IteratorOverClosed, IteratorOverEndOfStream)-                 | otherwise-                   -> let resp = IteratorOverNextBatch $-                                 sliceNextEventNumber slice in-                     (st, resp)-          IteratorOverClosed -> (st, IteratorOverEndOfStream)--      case action of-        IteratorOverEvent e -> return $ Just e-        IteratorOverEndOfStream -> return Nothing-        IteratorOverNextBatch num -> do-          w <- readBatch store name (startFrom num)-          res <- liftBase $ wait w-          case res of-            ReadFailure _ -> do-              atomicModifyIORef' ref $ \_ -> (IteratorOverClosed, ())-              return Nothing-            ReadSuccess slice -> do-              let nxtSt = IteratorOverAvailable slice-              atomicModifyIORef' ref $ \_ -> (nxtSt, ())-              go+    go m = runExceptT m >>= \case+      Left e  -> liftBase $ throwIO e+      Right a -> pure a

@@ -0,0 +1,86 @@+{-# LANGUAGE FlexibleContexts #-}+--------------------------------------------------------------------------------+-- |+-- Module    :  EventSource.Store.Iterator+-- Copyright :  (C) 2018 Yorick Laupa+-- License   :  (see the file LICENSE)+-- Maintainer:  Yorick Laupa <yo.eight@gmail.com>+-- Stability :  experimental+-- Portability: non-portable+--+--------------------------------------------------------------------------------+module EventSource.Store.Internal.Iterator where++--------------------------------------------------------------------------------+import Data.Int (Int32)++--------------------------------------------------------------------------------+import Control.Concurrent.Async.Lifted (Async, wait)+import Control.Monad.Base (MonadBase, liftBase)+import Data.IORef.Lifted (IORef, atomicModifyIORef')++--------------------------------------------------------------------------------+import EventSource.Types++--------------------------------------------------------------------------------+-- | Represents batch information needed to read a stream.+data Batch' a =+  Batch' { batchFrom :: !a+         , batchSize :: !Int32+         }++--------------------------------------------------------------------------------+-- | Starts a 'Batch' from a given point. The batch size is set to default,+--   which is 500.+startFrom :: a -> Batch' a+startFrom from = Batch' from 500++--------------------------------------------------------------------------------+data IteratorOverState a+  = IteratorOverAvailable (Slice' a)+  | IteratorOverClosed++--------------------------------------------------------------------------------+data IteratorOverAction a+  = IteratorOverEvent SavedEvent+  | IteratorOverNextBatch a+  | IteratorOverEndOfStream++--------------------------------------------------------------------------------+iterateOver :: MonadBase IO m+            => IORef (IteratorOverState a)+            -> (Batch' a -> m (Async (ReadStatus (Slice' a))))+            -> m (Maybe SavedEvent)+iterateOver ref puller = go+  where+    go = do+      action <- atomicModifyIORef' ref $ \st ->+        case st of+          IteratorOverAvailable slice ->+            case sliceEvents slice of+              e:es ->+                let nextSlice = slice { sliceEvents = es }+                    nxtSt = IteratorOverAvailable nextSlice in+                (nxtSt, IteratorOverEvent e)+              [] | sliceEndOfStream slice+                   -> (IteratorOverClosed, IteratorOverEndOfStream)+                 | otherwise+                   -> let resp = IteratorOverNextBatch $+                                 sliceNext slice in+                     (st, resp)+          IteratorOverClosed -> (st, IteratorOverEndOfStream)++      case action of+        IteratorOverEvent e -> return $ Just e+        IteratorOverEndOfStream -> return Nothing+        IteratorOverNextBatch num -> do+          w <- puller (startFrom num)+          res <- liftBase $ wait w+          case res of+            ReadFailure _ -> do+              atomicModifyIORef' ref $ \_ -> (IteratorOverClosed, ())+              return Nothing+            ReadSuccess slice -> do+              let nxtSt = IteratorOverAvailable slice+              atomicModifyIORef' ref $ \_ -> (nxtSt, ())+              go

@@ -224,10 +224,10 @@ -------------------------------------------------------------------------------- -- | Encapsulates an event which is about to be saved. data Event =-  Event { eventType :: EventType-        , eventId :: EventId-        , eventPayload :: Data-        , eventMetadata :: Maybe Properties+  Event { eventType :: !EventType+        , eventId :: !EventId+        , eventPayload :: !Data+        , eventMetadata :: !(Maybe Properties)         } deriving Show  --------------------------------------------------------------------------------@@ -238,9 +238,9 @@ -------------------------------------------------------------------------------- -- | Represents an event that's saved into the event store. data SavedEvent =-  SavedEvent { eventNumber :: EventNumber-             , savedEvent :: Event-             , linkEvent :: Maybe Event+  SavedEvent { eventNumber :: !EventNumber+             , savedEvent :: !Event+             , linkEvent :: !(Maybe Event)              } deriving Show  --------------------------------------------------------------------------------@@ -250,13 +250,20 @@  -------------------------------------------------------------------------------- -- | Represents batch of events read from a store.-data Slice =-  Slice { sliceEvents :: [SavedEvent]-        , sliceEndOfStream :: Bool-        , sliceNextEventNumber :: EventNumber-        } deriving Show+data Slice' a =+  Slice' { sliceEvents :: ![SavedEvent]+         , sliceEndOfStream :: !Bool+         , sliceNext :: !a+         } deriving Show  --------------------------------------------------------------------------------+type Slice = Slice' EventNumber++--------------------------------------------------------------------------------+sliceNextEventNumber :: Slice -> EventNumber+sliceNextEventNumber = sliceNext++-------------------------------------------------------------------------------- -- | Deserializes a 'Slice''s events. sliceEventsAs :: DecodeEvent a => Slice -> Either Text [a] sliceEventsAs = traverse savedEventAs . sliceEvents@@ -326,6 +333,9 @@   | ReadError (Maybe Text)   | AccessDenied StreamName   deriving Show++--------------------------------------------------------------------------------+instance Exception ReadFailure  -------------------------------------------------------------------------------- instance Functor ReadStatus where

@@ -9,7 +9,7 @@ - package.yaml - README.md ghc-options: -Wall-github: YoEight/eventsource-api+homepage: https://gitlab.com/YoEight/eventsource-api-hs library:   dependencies:     - base >=4.9 && <5@@ -29,6 +29,7 @@     - stm     - monad-loops ==0.4.*     - enclosed-exceptions+    - streaming   source-dirs: library license: BSD3 license-file: LICENSE.md@@ -36,4 +37,4 @@ maintainer: yo.eight@gmail.com name: eventsource-api synopsis: Provides an eventsourcing high level API.-version: '1.3.1'+version: '1.5.0'