eventstore 0.14.0.2 → 0.15.0.0

raw patch · 79 files changed

+6296/−4972 lines, 79 filesdep +asyncdep +bifunctorsdep +bytestringdep −classy-preludedep ~aesondep ~basedep ~cerealPVP ok

version bump matches the API change (PVP)

Dependencies added: async, bifunctors, bytestring, clock, ekg-core, exceptions, fast-logger, hashable, interpolate, lifted-async, lifted-base, machines, monad-control, monad-logger, mono-traversable, safe-exceptions, stm-chans, tasty-hspec, text-format, transformers-base

Dependencies removed: classy-prelude

Dependency ranges changed: aeson, base, cereal, connection, containers, protobuf, semigroups, time

API changes (from Hackage documentation)

Files

• .gitignore +0/−0
• .travis.yml +159/−74
• CHANGELOG.markdown +9/−2
• Database/EventStore.hs +141/−156
• Database/EventStore/Internal/Callback.hs +111/−0
• Database/EventStore/Internal/Command.hs +395/−6
• Database/EventStore/Internal/Communication.hs +67/−0
• Database/EventStore/Internal/Connection.hs +181/−220
• Database/EventStore/Internal/ConnectionManager.hs +499/−0
• Database/EventStore/Internal/Control.hs +419/−0
• Database/EventStore/Internal/Discovery.hs +4/−3
• Database/EventStore/Internal/EndPoint.hs +9/−2
• Database/EventStore/Internal/Exec.hs +176/−0
• Database/EventStore/Internal/Execution/Production.hs +0/−587
• Database/EventStore/Internal/Generator.hs +0/−43
• Database/EventStore/Internal/Logger.hs +73/−0
• Database/EventStore/Internal/Manager/Operation/Model.hs +0/−204
• Database/EventStore/Internal/Manager/Operation/Registry.hs +427/−0
• Database/EventStore/Internal/Manager/Subscription/Command.hs +0/−103
• Database/EventStore/Internal/Manager/Subscription/Driver.hs +0/−550
• Database/EventStore/Internal/Manager/Subscription/Message.hs +0/−408
• Database/EventStore/Internal/Manager/Subscription/Model.hs +0/−317
• Database/EventStore/Internal/Manager/Subscription/Packages.hs +0/−133
• Database/EventStore/Internal/Manager/Subscription/Types.hs +0/−40
• Database/EventStore/Internal/Operation.hs +138/−75
• Database/EventStore/Internal/Operation/Catchup.hs +51/−41
• Database/EventStore/Internal/Operation/DeleteStream.hs +5/−3
• Database/EventStore/Internal/Operation/DeleteStream/Message.hs +1/−1
• Database/EventStore/Internal/Operation/Persist.hs +79/−0
• Database/EventStore/Internal/Operation/PersistOperations.hs +75/−0
• Database/EventStore/Internal/Operation/Read/Common.hs +55/−4
• Database/EventStore/Internal/Operation/ReadAllEvents.hs +8/−6
• Database/EventStore/Internal/Operation/ReadAllEvents/Message.hs +1/−1
• Database/EventStore/Internal/Operation/ReadEvent.hs +5/−3
• Database/EventStore/Internal/Operation/ReadEvent/Message.hs +9/−3
• Database/EventStore/Internal/Operation/ReadStreamEvents.hs +8/−6
• Database/EventStore/Internal/Operation/ReadStreamEvents/Message.hs +1/−1
• Database/EventStore/Internal/Operation/StreamMetadata.hs +14/−13
• Database/EventStore/Internal/Operation/Transaction.hs +9/−7
• Database/EventStore/Internal/Operation/Transaction/Message.hs +1/−1
• Database/EventStore/Internal/Operation/Volatile.hs +78/−0
• Database/EventStore/Internal/Operation/Write/Common.hs +2/−4
• Database/EventStore/Internal/Operation/WriteEvents.hs +5/−3
• Database/EventStore/Internal/Operation/WriteEvents/Message.hs +1/−1
• Database/EventStore/Internal/OperationManager.hs +60/−0
• Database/EventStore/Internal/Operations.hs +2/−0
• Database/EventStore/Internal/Prelude.hs +185/−0
• Database/EventStore/Internal/Processor.hs +0/−328
• Database/EventStore/Internal/Settings.hs +149/−0
• Database/EventStore/Internal/Stopwatch.hs +54/−0
• Database/EventStore/Internal/Stream.hs +15/−4
• Database/EventStore/Internal/Subscription.hs +0/−810
• Database/EventStore/Internal/Subscription/Api.hs +122/−0
• Database/EventStore/Internal/Subscription/Catchup.hs +197/−0
• Database/EventStore/Internal/Subscription/Command.hs +113/−0
• Database/EventStore/Internal/Subscription/Message.hs +413/−0
• Database/EventStore/Internal/Subscription/Packages.hs +135/−0
• Database/EventStore/Internal/Subscription/Persistent.hs +184/−0
• Database/EventStore/Internal/Subscription/Regular.hs +102/−0
• Database/EventStore/Internal/Subscription/Types.hs +138/−0
• Database/EventStore/Internal/Test.hs +41/−0
• Database/EventStore/Internal/TimerService.hs +74/−0
• Database/EventStore/Internal/Types.hs +66/−131
• Database/EventStore/Internal/Utils.hs +3/−2
• Database/EventStore/Logging.hs +0/−68
• README.md +5/−5
• eventstore-test-setup-osx.sh +5/−0
• eventstore-test-setup.sh +3/−3
• eventstore.cabal +174/−150
• tests/Main.hs +40/−0
• tests/Test/Bogus/Connection.hs +79/−0
• tests/Test/Bus.hs +55/−0
• tests/Test/Common.hs +65/−0
• tests/Test/Connection.hs +52/−0
• tests/Test/Integration.hs +26/−0
• tests/Test/Integration/Tests.hs +454/−0
• tests/Test/Operation.hs +69/−0
• tests/Tests.hs +0/−397
• tests/integration.hs +0/−53

@@ -1,91 +1,176 @@-# This file has been generated -- see https://github.com/hvr/multi-ghc-travis-language: c+# This is the complex Travis configuration, which is intended for use+# on open source libraries which need compatibility across multiple GHC+# versions, must work with cabal-install, and should be+# cross-platform. For more information and other options, see:+#+# https://docs.haskellstack.org/en/stable/travis_ci/+#+# Copy these contents into the root directory of your Github project in a file+# named .travis.yml++# Use new container infrastructure to enable caching sudo: false +# Do not choose a language; we provide our own build tools.+language: generic++# Caching so the next build will be fast too. cache:   directories:-    - $HOME/.cabsnap-    - $HOME/.cabal/packages--before_cache:-  - rm -fv $HOME/.cabal/packages/hackage.haskell.org/build-reports.log-  - rm -fv $HOME/.cabal/packages/hackage.haskell.org/00-index.tar+  - $HOME/.ghc+  - $HOME/.cabal+  - $HOME/.stack +# The different configurations we want to test. We have BUILD=cabal which uses+# cabal-install, and BUILD=stack which uses Stack. More documentation on each+# of those below.+#+# We set the compiler values here to tell Travis to use a different+# cache file per set of arguments.+#+# If you need to have different apt packages for each combination in the+# matrix, you can use a line such as:+#     addons: {apt: {packages: [libfcgi-dev,libgmp-dev]}} matrix:   include:-    - env: CABALVER=1.18 GHCVER=7.8.3-      compiler: ": #GHC 7.8.3"-      addons: {apt: {packages: [cabal-install-1.18,ghc-7.8.3], sources: [hvr-ghc]}}-    - env: CABALVER=1.18 GHCVER=7.8.4-      compiler: ": #GHC 7.8.4"-      addons: {apt: {packages: [cabal-install-1.18,ghc-7.8.4], sources: [hvr-ghc]}}-    - env: CABALVER=1.22 GHCVER=7.10.1-      compiler: ": #GHC 7.10.1"-      addons: {apt: {packages: [cabal-install-1.22,ghc-7.10.1], sources: [hvr-ghc]}}-    - env: CABALVER=1.22 GHCVER=7.10.2-      compiler: ": #GHC 7.10.2"-      addons: {apt: {packages: [cabal-install-1.22,ghc-7.10.2], sources: [hvr-ghc]}}-    - env: CABALVER=1.22 GHCVER=7.10.3-      compiler: ": #GHC 7.10.3"-      addons: {apt: {packages: [cabal-install-1.22,ghc-7.10.3], sources: [hvr-ghc]}}-    - env: CABALVER=1.24 GHCVER=8.0.1-      compiler: ": #GHC 8.0.1"-      addons: {apt: {packages: [cabal-install-1.24,ghc-8.0.1], sources: [hvr-ghc]}}+  # We grab the appropriate GHC and cabal-install versions from hvr's PPA. See:+  # https://github.com/hvr/multi-ghc-travis+  - env: BUILD=cabal GHCVER=7.8.4 CABALVER=head HAPPYVER=1.19.5 ALEXVER=3.1.7+    compiler: ": #GHC 7.8.4"+    addons: {apt: {packages: [cabal-install-head,ghc-7.8.4,happy-1.19.5,alex-3.1.7], sources: [hvr-ghc]}}+  - env: BUILD=cabal GHCVER=7.10.3 CABALVER=head HAPPYVER=1.19.5 ALEXVER=3.1.7+    compiler: ": #GHC 7.10.3"+    addons: {apt: {packages: [cabal-install-head,ghc-7.10.3,happy-1.19.5,alex-3.1.7], sources: [hvr-ghc]}}+  - env: BUILD=cabal GHCVER=8.0.2 CABALVER=head HAPPYVER=1.19.5 ALEXVER=3.1.7+    compiler: ": #GHC 8.0.2"+    addons: {apt: {packages: [cabal-install-head,ghc-8.0.2,happy-1.19.5,alex-3.1.7], sources: [hvr-ghc]}}++  # Build with the newest GHC and cabal-install. This is an accepted failure,+  # see below.+  - env: BUILD=cabal GHCVER=head  CABALVER=head HAPPYVER=1.19.5 ALEXVER=3.1.7+    compiler: ": #GHC HEAD"+    addons: {apt: {packages: [cabal-install-head,ghc-head,happy-1.19.5,alex-3.1.7], sources: [hvr-ghc]}}++  # The Stack builds. We can pass in arbitrary Stack arguments via the ARGS+  # variable, such as using --stack-yaml to point to a different file.+  - env: BUILD=stack ARGS=""+    compiler: ": #stack default"+    addons: {apt: {packages: [libgmp-dev]}}++  - env: BUILD=stack ARGS="--resolver lts-8"+    compiler: ": #stack 8.0.2"+    addons: {apt: {packages: [libgmp-dev]}}++  # Nightly builds are allowed to fail+  - env: BUILD=stack ARGS="--resolver nightly"+    compiler: ": #stack nightly"+    addons: {apt: {packages: [libgmp-dev]}}++  # Build on OS X in addition to Linux+  - env: BUILD=stack ARGS=""+    compiler: ": #stack default osx"+    os: osx++  # Travis includes an OS X which is incompatible with GHC 7.8.4++  - env: BUILD=stack ARGS="--resolver lts-8"+    compiler: ": #stack 8.0.2 osx"+    os: osx++  - env: BUILD=stack ARGS="--resolver nightly"+    compiler: ": #stack nightly osx"+    os: osx++  allow_failures:+  - env: BUILD=cabal GHCVER=head  CABALVER=head HAPPYVER=1.19.5 ALEXVER=3.1.7+  - env: BUILD=stack ARGS="--resolver nightly"+ before_install:- - unset CC- - export PATH=/opt/ghc/$GHCVER/bin:/opt/cabal/$CABALVER/bin:$PATH- - bash eventstore-test-setup.sh+# Using compiler above sets CC to an invalid value, so unset it+- unset CC +# Setting up GetEventStore database.+- |+  if [ `uname` = "Darwin" ]+  then+    bash eventstore-test-setup-osx.sh+  else+    bash eventstore-test-setup.sh+  fi++# We want to always allow newer versions of packages when building on GHC HEAD+- CABALARGS=""+- if [ "x$GHCVER" = "xhead" ]; then CABALARGS=--allow-newer; fi++# Download and unpack the stack executable+- export PATH=/opt/ghc/$GHCVER/bin:/opt/cabal/$CABALVER/bin:$HOME/.local/bin:/opt/alex/$ALEXVER/bin:/opt/happy/$HAPPYVER/bin:$HOME/.cabal/bin:$PATH+- mkdir -p ~/.local/bin+- |+  if [ `uname` = "Darwin" ]+  then+    travis_retry curl --insecure -L https://www.stackage.org/stack/osx-x86_64 | tar xz --strip-components=1 --include '*/stack' -C ~/.local/bin+  else+    travis_retry curl -L https://www.stackage.org/stack/linux-x86_64 | tar xz --wildcards --strip-components=1 -C ~/.local/bin '*/stack'+  fi++  # Use the more reliable S3 mirror of Hackage+  mkdir -p $HOME/.cabal+  echo 'remote-repo: hackage.haskell.org:http://hackage.fpcomplete.com/' > $HOME/.cabal/config+  echo 'remote-repo-cache: $HOME/.cabal/packages' >> $HOME/.cabal/config++  if [ "$CABALVER" != "1.16" ]+  then+    echo 'jobs: $ncpus' >> $HOME/.cabal/config+  fi+ install:- - cabal --version- - echo "$(ghc --version) [$(ghc --print-project-git-commit-id 2> /dev/null || echo '?')]"- - if [ -f $HOME/.cabal/packages/hackage.haskell.org/00-index.tar.gz ];-   then-     zcat $HOME/.cabal/packages/hackage.haskell.org/00-index.tar.gz >-          $HOME/.cabal/packages/hackage.haskell.org/00-index.tar;-   fi- - travis_retry cabal update -v- - sed -i 's/^jobs:/-- jobs:/' ${HOME}/.cabal/config- - cabal install --only-dependencies --enable-tests --enable-benchmarks --dry -v > installplan.txt- - sed -i -e '1,/^Resolving /d' installplan.txt; cat installplan.txt+- echo "$(ghc --version) [$(ghc --print-project-git-commit-id 2> /dev/null || echo '?')]"+- if [ -f configure.ac ]; then autoreconf -i; fi+- |+  set -ex+  case "$BUILD" in+    stack)+      stack --no-terminal --install-ghc $ARGS test --bench --only-dependencies+      ;;+    cabal)+      cabal --version+      travis_retry cabal update -# check whether current requested install-plan matches cached package-db snapshot- - if diff -u installplan.txt $HOME/.cabsnap/installplan.txt;-   then-     echo "cabal build-cache HIT";-     rm -rfv .ghc;-     cp -a $HOME/.cabsnap/ghc $HOME/.ghc;-     cp -a $HOME/.cabsnap/lib $HOME/.cabsnap/share $HOME/.cabsnap/bin $HOME/.cabal/;-   else-     echo "cabal build-cache MISS";-     rm -rf $HOME/.cabsnap;-     mkdir -p $HOME/.ghc $HOME/.cabal/lib $HOME/.cabal/share $HOME/.cabal/bin;-     cabal install --only-dependencies --enable-tests --enable-benchmarks;-   fi+      # Get the list of packages from the stack.yaml file+      PACKAGES=$(stack --install-ghc query locals | grep '^ *path' | sed 's@^ *path:@@') -# snapshot package-db on cache miss- - if [ ! -d $HOME/.cabsnap ];-   then-      echo "snapshotting package-db to build-cache";-      mkdir $HOME/.cabsnap;-      cp -a $HOME/.ghc $HOME/.cabsnap/ghc;-      cp -a $HOME/.cabal/lib $HOME/.cabal/share $HOME/.cabal/bin installplan.txt $HOME/.cabsnap/;-   fi+      cabal install --only-dependencies --enable-tests --enable-benchmarks --force-reinstalls --ghc-options=-O0 --reorder-goals --max-backjumps=-1 $CABALARGS $PACKAGES+      ;;+  esac+  set +ex -# Here starts the actual work to be performed for the package under test;-# any command which exits with a non-zero exit code causes the build to fail. script:- - if [ -f configure.ac ]; then autoreconf -i; fi- - cabal configure --enable-tests --enable-benchmarks -v2  # -v2 provides useful information for debugging- - cabal build   # this builds all libraries and executables (including tests/benchmarks)- - cabal test- - cabal check- - cabal sdist   # tests that a source-distribution can be generated+- |+  set -ex+  case "$BUILD" in+    stack)+      stack --no-terminal $ARGS test --bench --no-run-benchmarks --haddock --no-haddock-deps+      ;;+    cabal)+      cabal install --enable-tests --enable-benchmarks --force-reinstalls --ghc-options=-O0 --reorder-goals --max-backjumps=-1 $CABALARGS $PACKAGES -# Check that the resulting source distribution can be built & installed.-# If there are no other `.tar.gz` files in `dist`, this can be even simpler:-# `cabal install --force-reinstalls dist/*-*.tar.gz`- - SRC_TGZ=$(cabal info . | awk '{print $2;exit}').tar.gz &&-   (cd dist && cabal install --force-reinstalls "$SRC_TGZ")+      ORIGDIR=$(pwd)+      for dir in $PACKAGES+      do+        cd $dir+        cabal check || [ "$CABALVER" == "1.16" ]+        cabal sdist+        PKGVER=$(cabal info . | awk '{print $2;exit}')+        SRC_TGZ=$PKGVER.tar.gz+        cd dist+        tar zxfv "$SRC_TGZ"+        cd "$PKGVER"+        cabal configure --enable-tests+        cabal build+        cd $ORIGDIR+      done+      ;;+  esac+  set +ex  # EOF

@@ -1,6 +1,13 @@-0.14.0.2+0.15.0.0 ---------* Bump aeson version.+* Overall internal components refactoring.+* Decrease memory pressure by 10 folds.+* Improve general performance.+* Provide proper logging support using `fast-logger` library.+* Support operation timeout detection.+* Detects if the server is overwhelmed and act accordingly.+* Improve connection management code.+* Expose EKG monitoring metrics.  0.14.0.1 ----------

@@ -3,6 +3,7 @@ {-# LANGUAGE ExistentialQuantification #-} {-# LANGUAGE OverloadedStrings         #-} {-# LANGUAGE RecordWildCards           #-}+{-# LANGUAGE ScopedTypeVariables       #-} -------------------------------------------------------------------------------- -- | -- Module : Database.EventStore@@ -18,7 +19,6 @@     ( -- * Connection       Connection     , ConnectionType(..)-    , ConnectionException(..)     , Credentials     , Settings(..)     , Retry@@ -51,6 +51,8 @@     , withJsonAndMetadata     , withBinary     , withBinaryAndMetadata+     -- * Common Operation types+    , OperationMaxAttemptReached(..)      -- * Read Operations     , StreamMetadataResult(..)     , readEvent@@ -110,32 +112,32 @@     , SubscriptionId     , Subscription     , SubDropReason(..)+    , SubDetails     , waitConfirmation     , unsubscribeConfirmed     , unsubscribeConfirmedSTM     , waitUnsubscribeConfirmed+    , nextEventMaybeSTM+    , getSubscriptionDetailsSTM+    , unsubscribe+    , subscriptionStream       -- * Volatile Subscription-    , Regular+    , RegularSubscription     , subscribe     , subscribeToAll-    , getSubId-    , getSubStream+    , getSubscriptionId     , isSubscribedToAll-    , unsubscribe     , nextEvent     , nextEventMaybe-    , getSubResolveLinkTos-    , getSubLastCommitPos-    , getSubLastEventNumber       -- * Catch-up Subscription-    , Catchup+    , CatchupSubscription     , subscribeFrom     , subscribeToAllFrom     , waitTillCatchup     , hasCaughtUp     , hasCaughtUpSTM      -- * Persistent Subscription-    , Persistent+    , PersistentSubscription     , PersistentSubscriptionSettings(..)     , SystemConsumerStrategy(..)     , NakAction(..)@@ -171,9 +173,14 @@     , resolvedEventDataAsJson     , resolvedEventOriginalStreamId     , resolvedEventOriginalId+    , resolvedEventOriginalEventNumber     , recordedEventDataAsJson     , positionStart     , positionEnd+      -- * Logging+    , LogLevel(..)+    , LogType(..)+    , LoggerFilter(..)       -- * Misc     , Command     , DropReason(..)@@ -183,37 +190,48 @@     , emptyStreamVersion     , exactEventVersion     , streamExists+    , msDiffTime       -- * Re-export-    , waitAsync     , (<>)     , NonEmpty(..)     , nonEmpty     , TLSSettings+    , NominalDiffTime     ) where  --------------------------------------------------------------------------------+import Prelude (String) import Data.Int import Data.Maybe+import Data.Time (NominalDiffTime)  ---------------------------------------------------------------------------------import ClassyPrelude hiding (Builder, group) import Data.List.NonEmpty(NonEmpty(..), nonEmpty) import Network.Connection (TLSSettings)  --------------------------------------------------------------------------------+import           Database.EventStore.Internal.Callback import           Database.EventStore.Internal.Command-import           Database.EventStore.Internal.Connection+import           Database.EventStore.Internal.Communication+import           Database.EventStore.Internal.Connection (connectionBuilder)+import           Database.EventStore.Internal.Control hiding (subscribe) import           Database.EventStore.Internal.Discovery-import           Database.EventStore.Internal.Subscription-import           Database.EventStore.Internal.Manager.Subscription.Driver hiding (unsubscribe)-import           Database.EventStore.Internal.Manager.Subscription.Message+import           Database.EventStore.Internal.Exec+import           Database.EventStore.Internal.Subscription.Api+import           Database.EventStore.Internal.Subscription.Catchup+import           Database.EventStore.Internal.Subscription.Message+import           Database.EventStore.Internal.Subscription.Persistent+import           Database.EventStore.Internal.Subscription.Types+import           Database.EventStore.Internal.Subscription.Regular+import           Database.EventStore.Internal.Logger+import           Database.EventStore.Internal.Manager.Operation.Registry import           Database.EventStore.Internal.Operation (OperationError(..)) import qualified Database.EventStore.Internal.Operations as Op import           Database.EventStore.Internal.Operation.Read.Common import           Database.EventStore.Internal.Operation.Write.Common+import           Database.EventStore.Internal.Prelude import           Database.EventStore.Internal.Stream import           Database.EventStore.Internal.Types-import           Database.EventStore.Internal.Execution.Production  -------------------------------------------------------------------------------- -- Connection@@ -230,7 +248,7 @@ -- | Represents a connection to a single EventStore node. data Connection     = Connection-      { _prod     :: Production+      { _exec     :: Exec       , _settings :: Settings       , _type     :: ConnectionType       }@@ -249,33 +267,37 @@ --   performance out of the connection it is generally recommended to use it in --   this way. connect :: Settings -> ConnectionType -> IO Connection-connect settings tpe = do+connect settings@Settings{..} tpe = do     disc <- case tpe of         Static host port -> return $ staticEndPointDiscovery host port         Cluster setts    -> clusterDnsEndPointDiscovery setts         Dns dom srv port -> return $ simpleDnsEndPointDiscovery dom srv port-    prod <- newExecutionModel settings disc-    return $ Connection prod settings tpe +    logRef  <- newLoggerRef s_loggerType s_loggerFilter s_loggerDetailed+    mainBus <- newBus logRef settings+    builder <- connectionBuilder settings+    exec    <- newExec settings mainBus builder disc+    return $ Connection exec settings tpe+ -------------------------------------------------------------------------------- -- | Waits the 'Connection' to be closed. waitTillClosed :: Connection -> IO ()-waitTillClosed Connection{..} = prodWaitTillClosed _prod+waitTillClosed Connection{..} = execWaitTillClosed _exec  ----------------------------------------------------------------------------------- | Returns a 'Connection''s 'Settings'.+-- | Returns a 'Connection' 's 'Settings'. connectionSettings :: Connection -> Settings connectionSettings = _settings  -------------------------------------------------------------------------------- -- | Asynchronously closes the 'Connection'. shutdown :: Connection -> IO ()-shutdown Connection{..} = shutdownExecutionModel _prod+shutdown Connection{..} = publishWith _exec SystemShutdown  -------------------------------------------------------------------------------- -- | Sends a single 'Event' to given stream. sendEvent :: Connection-          -> Text             -- ^ Stream name+          -> StreamName              -- ^ Stream name           -> ExpectedVersion           -> Event           -> IO (Async WriteResult)@@ -285,28 +307,28 @@ -------------------------------------------------------------------------------- -- | Sends a list of 'Event' to given stream. sendEvents :: Connection-           -> Text             -- ^ Stream name+           -> StreamName             -- ^ Stream name            -> ExpectedVersion            -> [Event]            -> IO (Async WriteResult) sendEvents Connection{..} evt_stream exp_ver evts = do-    (k, as)  <- createOpAsync-    let op = Op.writeEvents _settings evt_stream exp_ver evts-    pushOperation _prod k op-    return as+    p <- newPromise+    let op = Op.writeEvents _settings (streamNameRaw evt_stream) exp_ver evts+    publishWith _exec (SubmitOperation p op)+    async (retrieve p)  -------------------------------------------------------------------------------- -- | Deletes given stream. deleteStream :: Connection-             -> Text             -- ^ Stream name+             -> StreamName             -- ^ Stream name              -> ExpectedVersion              -> Maybe Bool       -- ^ Hard delete              -> IO (Async Op.DeleteResult) deleteStream Connection{..} evt_stream exp_ver hard_del = do-    (k, as) <- createOpAsync-    let op = Op.deleteStream _settings evt_stream exp_ver hard_del-    pushOperation _prod k op-    return as+    p <- newPromise+    let op = Op.deleteStream _settings (streamNameRaw evt_stream) exp_ver hard_del+    publishWith _exec (SubmitOperation p op)+    async (retrieve p)  -------------------------------------------------------------------------------- -- | Represents a multi-request transaction with the EventStore.@@ -332,43 +354,43 @@ -------------------------------------------------------------------------------- -- | Starts a transaction on given stream. startTransaction :: Connection-                 -> Text            -- ^ Stream name+                 -> StreamName            -- ^ Stream name                  -> ExpectedVersion                  -> IO (Async Transaction) startTransaction conn@Connection{..} evt_stream exp_ver = do-    (k, as) <- createOpAsync-    let op = Op.transactionStart _settings evt_stream exp_ver-    pushOperation _prod k op-    let _F trans_id =-            Transaction-            { _tStream  = evt_stream-            , _tTransId = TransactionId trans_id-            , _tExpVer  = exp_ver-            , _tConn    = conn-            }-    return $ fmap _F as+    p <- newPromise+    let op = Op.transactionStart _settings (streamNameRaw evt_stream) exp_ver+    publishWith _exec (SubmitOperation p op)+    async $ do+        tid <- retrieve p+        return Transaction+               { _tStream  = streamNameRaw evt_stream+               , _tTransId = TransactionId tid+               , _tExpVer  = exp_ver+               , _tConn    = conn+               }  -------------------------------------------------------------------------------- -- | Asynchronously writes to a transaction in the EventStore. transactionWrite :: Transaction -> [Event] -> IO (Async ()) transactionWrite Transaction{..} evts = do-    (k, as) <- createOpAsync+    p <- newPromise     let Connection{..} = _tConn         raw_id = _unTransId _tTransId         op     = Op.transactionWrite _settings _tStream _tExpVer raw_id evts-    pushOperation _prod k op-    return as+    publishWith _exec (SubmitOperation p op)+    async (retrieve p)  -------------------------------------------------------------------------------- -- | Asynchronously commits this transaction. transactionCommit :: Transaction -> IO (Async WriteResult) transactionCommit Transaction{..} = do-    (k, as) <- createOpAsync+    p <- newPromise     let Connection{..} = _tConn         raw_id = _unTransId _tTransId         op     = Op.transactionCommit _settings _tStream _tExpVer raw_id-    pushOperation _prod k op-    return as+    publishWith _exec (SubmitOperation p op)+    async (retrieve p)  -------------------------------------------------------------------------------- -- | There isn't such of thing in EventStore parlance. Basically, if you want to@@ -379,20 +401,20 @@ -------------------------------------------------------------------------------- -- | Reads a single event from given stream. readEvent :: Connection-          -> Text       -- ^ Stream name+          -> StreamName       -- ^ Stream name           -> Int32      -- ^ Event number           -> Bool       -- ^ Resolve Link Tos           -> IO (Async (ReadResult 'RegularStream Op.ReadEvent)) readEvent Connection{..} stream_id evt_num res_link_tos = do-    (k, as) <- createOpAsync-    let op = Op.readEvent _settings stream_id evt_num res_link_tos-    pushOperation _prod k op-    return as+    p <- newPromise+    let op = Op.readEvent _settings (streamNameRaw stream_id) evt_num res_link_tos+    publishWith _exec (SubmitOperation p op)+    async (retrieve p)  -------------------------------------------------------------------------------- -- | Reads events from a given stream forward. readStreamEventsForward :: Connection-                        -> Text       -- ^ Stream name+                        -> StreamName       -- ^ Stream name                         -> Int32      -- ^ From event number                         -> Int32      -- ^ Batch size                         -> Bool       -- ^ Resolve Link Tos@@ -403,7 +425,7 @@ -------------------------------------------------------------------------------- -- | Reads events from a given stream backward. readStreamEventsBackward :: Connection-                         -> Text       -- ^ Stream name+                         -> StreamName       -- ^ Stream name                          -> Int32      -- ^ From event number                          -> Int32      -- ^ Batch size                          -> Bool       -- ^ Resolve Link Tos@@ -414,16 +436,17 @@ -------------------------------------------------------------------------------- readStreamEventsCommon :: Connection                        -> ReadDirection-                       -> Text+                       -> StreamName                        -> Int32                        -> Int32                        -> Bool                        -> IO (Async (ReadResult 'RegularStream StreamSlice)) readStreamEventsCommon Connection{..} dir stream_id start cnt res_link_tos = do-    (k, as) <- createOpAsync-    let op = Op.readStreamEvents _settings dir stream_id start cnt res_link_tos-    pushOperation _prod k op-    return as+    p <- newPromise+    let name = streamNameRaw stream_id+        op   = Op.readStreamEvents _settings dir name start cnt res_link_tos+    publishWith _exec (SubmitOperation p op)+    async (retrieve p)  -------------------------------------------------------------------------------- -- | Reads events from the $all stream forward.@@ -453,49 +476,28 @@                     -> Bool                     -> IO (Async AllSlice) readAllEventsCommon Connection{..} dir pos max_c res_link_tos = do-    (k, as) <- createOpAsync+    p <- newPromise     let op = Op.readAllEvents _settings c_pos p_pos max_c res_link_tos dir-    pushOperation _prod k op-    return as+    publishWith _exec (SubmitOperation p op)+    async (retrieve p)   where     Position c_pos p_pos = pos  ---------------------------------------------------------------------------------mkSubEnv :: Connection -> SubEnv-mkSubEnv Connection{..} =-    SubEnv-    { subSettings = _settings-    , subPushOp = pushOperation _prod-    , subPushConnect = \k cmd ->-          case cmd of-              PushRegular stream tos ->-                  pushConnectStream _prod k stream tos-              PushPersistent group stream size ->-                  pushConnectPersist _prod k group stream size-    , subPushUnsub = pushUnsubscribe _prod-    , subAckCmd = \cmd run uuids ->-          case cmd of-              AckCmd -> pushAckPersist _prod run uuids-              NakCmd act res -> pushNakPersist _prod run act res uuids-    , subForceReconnect = \node ->-          pushForceReconnect _prod node-    }---------------------------------------------------------------------------------- -- | Subcribes to given stream. subscribe :: Connection-          -> Text       -- ^ Stream name+          -> StreamName       -- ^ Stream name           -> Bool       -- ^ Resolve Link Tos-          -> IO (Subscription Regular)-subscribe conn streamId resLnkTos =-    regularSub (mkSubEnv conn) streamId resLnkTos+          -> IO RegularSubscription+subscribe Connection{..} stream resLnkTos =+    newRegularSubscription _exec stream resLnkTos  -------------------------------------------------------------------------------- -- | Subcribes to $all stream. subscribeToAll :: Connection                -> Bool       -- ^ Resolve Link Tos-               -> IO (Subscription Regular)-subscribeToAll conn = subscribe conn ""+               -> IO RegularSubscription+subscribeToAll conn = subscribe conn AllStream  -------------------------------------------------------------------------------- -- | Subscribes to given stream. If last checkpoint is defined, this will@@ -503,15 +505,15 @@ --   beginning. Once last stream event reached up, a subscription request will --   be sent using 'subscribe'. subscribeFrom :: Connection-              -> Text        -- ^ Stream name+              -> StreamName        -- ^ Stream name               -> Bool        -- ^ Resolve Link Tos               -> Maybe Int32 -- ^ Last checkpoint               -> Maybe Int32 -- ^ Batch size-              -> IO (Subscription Catchup)+              -> IO CatchupSubscription subscribeFrom conn streamId resLnkTos lastChkPt batch =     subscribeFromCommon conn resLnkTos batch tpe   where-    tpe = Op.RegularCatchup streamId (fromMaybe 0 lastChkPt)+    tpe = Op.RegularCatchup (streamNameRaw streamId) (fromMaybe 0 lastChkPt)  -------------------------------------------------------------------------------- -- | Same as 'subscribeFrom' but applied to $all stream.@@ -519,112 +521,95 @@                    -> Bool           -- ^ Resolve Link Tos                    -> Maybe Position -- ^ Last checkpoint                    -> Maybe Int32    -- ^ Batch size-                   -> IO (Subscription Catchup)+                   -> IO CatchupSubscription subscribeToAllFrom conn resLnkTos lastChkPt batch =     subscribeFromCommon conn resLnkTos batch tpe   where     Position cPos pPos = fromMaybe positionStart lastChkPt-    tpe = Op.AllCatchup cPos pPos+    tpe = Op.AllCatchup (Position cPos pPos)  -------------------------------------------------------------------------------- subscribeFromCommon :: Connection                     -> Bool                     -> Maybe Int32                     -> Op.CatchupState-                    -> IO (Subscription Catchup)-subscribeFromCommon conn resLnkTos batch tpe =-    catchupSub (mkSubEnv conn) params-  where-    params = CatchupParams { catchupResLnkTos = resLnkTos-                           , catchupState = tpe-                           , catchupBatchSize = batch-                           }+                    -> IO CatchupSubscription+subscribeFromCommon Connection{..} resLnkTos batch tpe =+    newCatchupSubscription _exec resLnkTos batch tpe  -------------------------------------------------------------------------------- -- | Asynchronously sets the metadata for a stream. setStreamMetadata :: Connection-                  -> Text+                  -> StreamName                   -> ExpectedVersion                   -> StreamMetadata                   -> IO (Async WriteResult) setStreamMetadata Connection{..} evt_stream exp_ver metadata = do-    (k, as) <- createOpAsync-    let op = Op.setMetaStream _settings evt_stream exp_ver metadata-    pushOperation _prod k op-    return as+    p <- newPromise+    let name = streamNameRaw evt_stream+        op = Op.setMetaStream _settings name exp_ver metadata+    publishWith _exec (SubmitOperation p op)+    async (retrieve p)  -------------------------------------------------------------------------------- -- | Asynchronously gets the metadata of a stream.-getStreamMetadata :: Connection -> Text -> IO (Async StreamMetadataResult)+getStreamMetadata :: Connection -> StreamName -> IO (Async StreamMetadataResult) getStreamMetadata Connection{..} evt_stream = do-    (k, as) <- createOpAsync-    let op = Op.readMetaStream _settings evt_stream-    pushOperation _prod k op-    return as+    p <- newPromise+    let op = Op.readMetaStream _settings (streamNameRaw evt_stream)+    publishWith _exec (SubmitOperation p op)+    async (retrieve p)  -------------------------------------------------------------------------------- -- | Asynchronously create a persistent subscription group on a stream. createPersistentSubscription :: Connection                              -> Text-                             -> Text+                             -> StreamName                              -> PersistentSubscriptionSettings                              -> IO (Async (Maybe PersistActionException))-createPersistentSubscription Connection{..} group stream sett = do-    mvar <- newEmptyTMVarIO-    let _F res = atomically $-            case res of-                Left e -> putTMVar mvar (Just e)-                _      -> putTMVar mvar Nothing-    pushCreatePersist _prod _F group stream sett-    async $ atomically $ readTMVar mvar+createPersistentSubscription Connection{..} grp stream sett = do+    p <- newPromise+    let op = Op.createPersist grp (streamNameRaw stream) sett+    publishWith _exec (SubmitOperation p op)+    async (persistAsync p)  -------------------------------------------------------------------------------- -- | Asynchronously update a persistent subscription group on a stream. updatePersistentSubscription :: Connection                              -> Text-                             -> Text+                             -> StreamName                              -> PersistentSubscriptionSettings                              -> IO (Async (Maybe PersistActionException))-updatePersistentSubscription Connection{..} group stream sett = do-    mvar <- newEmptyTMVarIO-    let _F res = atomically $-            case res of-                Left e -> putTMVar mvar (Just e)-                _      -> putTMVar mvar Nothing-    pushUpdatePersist _prod _F group stream sett-    async $ atomically $ readTMVar mvar+updatePersistentSubscription Connection{..} grp stream sett = do+    p <- newPromise+    let op = Op.updatePersist grp (streamNameRaw stream) sett+    publishWith _exec (SubmitOperation p op)+    async (persistAsync p)  -------------------------------------------------------------------------------- -- | Asynchronously delete a persistent subscription group on a stream. deletePersistentSubscription :: Connection                              -> Text-                             -> Text+                             -> StreamName                              -> IO (Async (Maybe PersistActionException))-deletePersistentSubscription Connection{..} group stream = do-    mvar <- newEmptyTMVarIO-    let _F res = atomically $-            case res of-                Left e -> putTMVar mvar (Just e)-                _      -> putTMVar mvar Nothing-    pushDeletePersist _prod _F group stream-    async $ atomically $ readTMVar mvar+deletePersistentSubscription Connection{..} grp stream = do+    p <- newPromise+    let op = Op.deletePersist grp (streamNameRaw stream)+    publishWith _exec (SubmitOperation p op)+    async (persistAsync p)  -------------------------------------------------------------------------------- -- | Asynchronously connect to a persistent subscription given a group on a --   stream. connectToPersistentSubscription :: Connection                                 -> Text-                                -> Text+                                -> StreamName                                 -> Int32-                                -> IO (Subscription Persistent)-connectToPersistentSubscription conn group stream bufSize =-    persistentSub (mkSubEnv conn) group stream bufSize+                                -> IO PersistentSubscription+connectToPersistentSubscription Connection{..} group stream bufSize =+    newPersistentSubscription _exec group stream bufSize  ---------------------------------------------------------------------------------createOpAsync :: IO (Either OperationError a -> IO (), Async a)-createOpAsync = do-    mvar <- newEmptyMVar-    as   <- async $ do-        res <- readMVar mvar-        either throwIO return res-    return (putMVar mvar, as)+persistAsync :: Callback (Maybe PersistActionException)+             -> IO (Maybe PersistActionException)+persistAsync = either throw return <=< tryRetrieve

@@ -0,0 +1,111 @@+{-# LANGUAGE GADTs               #-}+{-# LANGUAGE RankNTypes          #-}+{-# LANGUAGE ScopedTypeVariables #-}+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Internal.Callback+-- Copyright : (C) 2017 Yorick Laupa+-- License : (see the file LICENSE)+--+-- Maintainer : Yorick Laupa <yo.eight@gmail.com>+-- Stability : provisional+-- Portability : non-portable+--+--------------------------------------------------------------------------------+module Database.EventStore.Internal.Callback+  ( Callback+  , newPromise+  , newCallback+  , fulfill+  , reject+  , retrieve+  , tryRetrieve+  , fromEither+  ) where++--------------------------------------------------------------------------------+import Database.EventStore.Internal.Prelude++--------------------------------------------------------------------------------+newtype Callback a =+  Callback { runCallback :: forall b. Stage a b -> IO b }++--------------------------------------------------------------------------------+data Stage a b where+  Fulfill  :: a -> Stage a ()+  Reject   :: Exception e => e -> Stage a ()+  Retrieve :: Stage a (Either SomeException a)++--------------------------------------------------------------------------------+fulfill :: MonadIO m => Callback a -> a -> m ()+fulfill cb a = liftIO $ runCallback cb (Fulfill a)++--------------------------------------------------------------------------------+reject :: (Exception e, MonadIO m) => Callback a -> e -> m ()+reject cb e = liftIO $ runCallback cb (Reject e)++--------------------------------------------------------------------------------+tryRetrieve :: Callback a -> IO (Either SomeException a)+tryRetrieve cb = runCallback cb Retrieve++--------------------------------------------------------------------------------+retrieve :: Callback a -> IO a+retrieve p = do+  tryRetrieve p >>= \case+    Left e  -> throw e+    Right a -> return a++--------------------------------------------------------------------------------+fromEither :: Exception e => Callback a -> Either e a -> IO ()+fromEither p (Left e)  = reject p e+fromEither p (Right a) = fulfill p a++--------------------------------------------------------------------------------+newPromise :: IO (Callback a)+newPromise = do+  mvar <- newEmptyTMVarIO+  return $ promise mvar++--------------------------------------------------------------------------------+newCallback :: (Either SomeException a -> IO ()) -> IO (Callback a)+newCallback k = do+  mvar <- newEmptyTMVarIO+  return $ callback mvar k++--------------------------------------------------------------------------------+promise :: forall a. TMVar (Either SomeException a) -> Callback a+promise mvar = Callback go+  where+    go :: forall b. Stage a b -> IO b+    go (Fulfill a) = atomically $+      whenM (isEmptyTMVar mvar) $+        putTMVar mvar (Right a)++    go (Reject e) = atomically $+      whenM (isEmptyTMVar mvar) $+        putTMVar mvar (Left $ toException e)++    go Retrieve = atomically $ readTMVar mvar++--------------------------------------------------------------------------------+callback :: forall a. TMVar (Either SomeException a)+         -> (Either SomeException a -> IO ())+         -> Callback a+callback mvar k = Callback go+  where+    go :: forall b. Stage a b -> IO b+    go (Fulfill a) = do+      atomically $+        unlessM (tryPutTMVar mvar (Right a)) $ do+          _ <- swapTMVar mvar (Right a)+          return ()+      k (Right a)+    go (Reject e) = do+      let err = Left $ toException e+      atomically $ do+        unlessM (tryPutTMVar mvar err) $ do+          _ <- swapTMVar mvar err++          return ()+      k err+    go Retrieve = atomically $ readTMVar mvar

@@ -1,4 +1,5 @@ {-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE OverloadedStrings          #-} -------------------------------------------------------------------------------- -- | -- Module : Database.EventStore.Internal.Command@@ -10,18 +11,406 @@ -- Portability : non-portable -- ---------------------------------------------------------------------------------module Database.EventStore.Internal.Command (Command(..)) where-----------------------------------------------------------------------------------import ClassyPrelude+module Database.EventStore.Internal.Command where  --------------------------------------------------------------------------------+import Database.EventStore.Internal.Prelude import Database.EventStore.Internal.Utils  -------------------------------------------------------------------------------- -- | Internal command representation.-newtype Command = Command { cmdWord8 :: Word8 } deriving (Eq, Ord, Num)+data Command =+    Command { cmdWord8 :: !Word8+            , cmdDesc  :: !Text+            }  -------------------------------------------------------------------------------- instance Show Command where-    show (Command w) = prettyWord8 w+    show c = "(" <> prettyWord8 (cmdWord8 c) <> ")" <> unpack (cmdDesc c)++--------------------------------------------------------------------------------+instance Eq Command where+    c1 == c2 = cmdWord8 c1 == cmdWord8 c2++--------------------------------------------------------------------------------+instance Ord Command where+    compare c1 c2 = compare (cmdWord8 c1) (cmdWord8 c2)++--------------------------------------------------------------------------------+heartbeatRequestCmd :: Command+heartbeatRequestCmd =+    Command { cmdWord8 = 0x01+            , cmdDesc  = "[heartbeat-request]"+            }++--------------------------------------------------------------------------------+heartbeatResponseCmd :: Command+heartbeatResponseCmd =+    Command { cmdWord8 = 0x02+            , cmdDesc  = "[heartbeat-response]"+            }++--------------------------------------------------------------------------------+writeEventsCmd :: Command+writeEventsCmd =+    Command { cmdWord8 = 0x82+            , cmdDesc  = "[write-events]"+            }++--------------------------------------------------------------------------------+writeEventsCompletedCmd :: Command+writeEventsCompletedCmd =+    Command { cmdWord8 = 0x83+            , cmdDesc  = "[write-events-completed]"+            }++--------------------------------------------------------------------------------+transactionStartCmd :: Command+transactionStartCmd =+    Command { cmdWord8 = 0x84+            , cmdDesc  = "[transaction-start]"+            }++--------------------------------------------------------------------------------+transactionStartCompletedCmd :: Command+transactionStartCompletedCmd =+    Command { cmdWord8 = 0x85+            , cmdDesc  = "[transaction-start-completed]"+            }++--------------------------------------------------------------------------------+transactionWriteCmd :: Command+transactionWriteCmd =+    Command { cmdWord8 = 0x86+            , cmdDesc  = "[transaction-write]"+            }++--------------------------------------------------------------------------------+transactionWriteCompletedCmd :: Command+transactionWriteCompletedCmd =+    Command { cmdWord8 = 0x87+            , cmdDesc  = "[transaction-write-completed]"+            }++--------------------------------------------------------------------------------+transactionCommitCmd :: Command+transactionCommitCmd =+    Command { cmdWord8 = 0x88+            , cmdDesc  = "[transaction-commit]"+            }++--------------------------------------------------------------------------------+transactionCommitCompletedCmd :: Command+transactionCommitCompletedCmd =+    Command { cmdWord8 = 0x89+            , cmdDesc  = "[transaction-commit-completed]"+            }++--------------------------------------------------------------------------------+deleteStreamCmd :: Command+deleteStreamCmd =+    Command { cmdWord8 = 0x8A+            , cmdDesc  = "[delete-stream]"+            }++--------------------------------------------------------------------------------+deleteStreamCompletedCmd :: Command+deleteStreamCompletedCmd =+    Command { cmdWord8 = 0x8B+            , cmdDesc  = "[delete-stream-completed]"+            }++--------------------------------------------------------------------------------+readEventCmd :: Command+readEventCmd =+    Command { cmdWord8 = 0xB0+            , cmdDesc  = "[read-event]"+            }++--------------------------------------------------------------------------------+readEventCompletedCmd :: Command+readEventCompletedCmd =+    Command { cmdWord8 = 0xB1+            , cmdDesc  = "[read-event-completed]"+            }++--------------------------------------------------------------------------------+readStreamEventsForwardCmd :: Command+readStreamEventsForwardCmd =+    Command { cmdWord8 = 0xB2+            , cmdDesc  = "[read-stream-events-forward]"+            }++--------------------------------------------------------------------------------+readStreamEventsForwardCompletedCmd :: Command+readStreamEventsForwardCompletedCmd =+    Command { cmdWord8 = 0xB3+            , cmdDesc  = "[read-stream-events-forward-completed]"+            }++--------------------------------------------------------------------------------+readStreamEventsBackwardCmd :: Command+readStreamEventsBackwardCmd =+    Command { cmdWord8 = 0xB4+            , cmdDesc  = "[read-stream-events-backward]"+            }++--------------------------------------------------------------------------------+readStreamEventsBackwardCompletedCmd :: Command+readStreamEventsBackwardCompletedCmd =+    Command { cmdWord8 = 0xB5+            , cmdDesc  = "[read-stream-events-backward]"+            }++--------------------------------------------------------------------------------+readAllEventsForwardCmd :: Command+readAllEventsForwardCmd =+    Command { cmdWord8 = 0xB6+            , cmdDesc  = "[read-all-events-forward]"+            }++--------------------------------------------------------------------------------+readAllEventsForwardCompletedCmd :: Command+readAllEventsForwardCompletedCmd =+    Command { cmdWord8 = 0xB7+            , cmdDesc  = "[read-all-events-forward-completed]"+            }++--------------------------------------------------------------------------------+readAllEventsBackwardCmd :: Command+readAllEventsBackwardCmd =+    Command { cmdWord8 = 0xB8+            , cmdDesc  = "[read-all-events-backward]"+            }++--------------------------------------------------------------------------------+readAllEventsBackwardCompletedCmd :: Command+readAllEventsBackwardCompletedCmd =+    Command { cmdWord8 = 0xB9+            , cmdDesc  = "[read-all-events-backward-completed]"+            }++--------------------------------------------------------------------------------+subscribeToStreamCmd :: Command+subscribeToStreamCmd =+    Command { cmdWord8 = 0xC0+            , cmdDesc  = "[subscribe-to-stream]"+            }++--------------------------------------------------------------------------------+subscriptionConfirmationCmd :: Command+subscriptionConfirmationCmd =+    Command { cmdWord8 = 0xC1+            , cmdDesc  = "[subscription-confirmation]"+            }++--------------------------------------------------------------------------------+streamEventAppearedCmd :: Command+streamEventAppearedCmd =+    Command { cmdWord8 = 0xC2+            , cmdDesc  = "[stream-event-appeared]"+            }++--------------------------------------------------------------------------------+unsubscribeFromStreamCmd :: Command+unsubscribeFromStreamCmd =+    Command { cmdWord8 = 0xC3+            , cmdDesc  = "[unsubscribe-from-stream]"+            }++--------------------------------------------------------------------------------+subscriptionDroppedCmd :: Command+subscriptionDroppedCmd =+    Command { cmdWord8 = 0xC4+            , cmdDesc  = "[subscription-dropped]"+            }++--------------------------------------------------------------------------------+connectToPersistentSubscriptionCmd :: Command+connectToPersistentSubscriptionCmd =+    Command { cmdWord8 = 0xC5+            , cmdDesc  = "[connect-to-persistent-subscription]"+            }++--------------------------------------------------------------------------------+persistentSubscriptionConfirmationCmd :: Command+persistentSubscriptionConfirmationCmd =+    Command { cmdWord8 = 0xC6+            , cmdDesc  = "[persistent-subscription-confirmation]"+            }++--------------------------------------------------------------------------------+persistentSubscriptionStreamEventAppearedCmd :: Command+persistentSubscriptionStreamEventAppearedCmd =+    Command { cmdWord8 = 0xC7+            , cmdDesc  = "[persistent-subscription-stream-event-appeared]"+            }++--------------------------------------------------------------------------------+createPersistentSubscriptionCmd :: Command+createPersistentSubscriptionCmd =+    Command { cmdWord8 = 0xC8+            , cmdDesc  = "[create-persistent-subscription]"+            }++--------------------------------------------------------------------------------+createPersistentSubscriptionCompletedCmd :: Command+createPersistentSubscriptionCompletedCmd =+    Command { cmdWord8 = 0xC9+            , cmdDesc  = "[create-persistent-subscription-completed]"+            }++--------------------------------------------------------------------------------+deletePersistentSubscriptionCmd :: Command+deletePersistentSubscriptionCmd =+    Command { cmdWord8 = 0xCA+            , cmdDesc  = "[delete-persistent-subscription]"+            }++--------------------------------------------------------------------------------+deletePersistentSubscriptionCompletedCmd :: Command+deletePersistentSubscriptionCompletedCmd =+    Command { cmdWord8 = 0xCB+            , cmdDesc  = "[delete-persistent-subscription-completed]"+            }++--------------------------------------------------------------------------------+persistentSubscriptionAckEventsCmd :: Command+persistentSubscriptionAckEventsCmd =+    Command { cmdWord8 = 0xCC+            , cmdDesc  = "[persistent-subscription-ack-events]"+            }++--------------------------------------------------------------------------------+persistentSubscriptionNakEventsCmd :: Command+persistentSubscriptionNakEventsCmd =+    Command { cmdWord8 = 0xCD+            , cmdDesc  = "[persistent-subscription-nak-events]"+            }++--------------------------------------------------------------------------------+updatePersistentSubscriptionCmd :: Command+updatePersistentSubscriptionCmd =+    Command { cmdWord8 = 0xCE+            , cmdDesc  = "[update-persistent-subscription]"+            }++--------------------------------------------------------------------------------+updatePersistentSubscriptionCompletedCmd :: Command+updatePersistentSubscriptionCompletedCmd =+    Command { cmdWord8 = 0xCF+            , cmdDesc  = "[update-persistent-subscription-completed]"+            }++--------------------------------------------------------------------------------+badRequestCmd :: Command+badRequestCmd =+    Command { cmdWord8 = 0xF0+            , cmdDesc  = "[bad-request]"+            }++--------------------------------------------------------------------------------+notHandledCmd :: Command+notHandledCmd =+    Command { cmdWord8 = 0xF1+            , cmdDesc  = "[not-handled]"+            }++--------------------------------------------------------------------------------+authenticateCmd :: Command+authenticateCmd =+    Command { cmdWord8 = 0xF2+            , cmdDesc  = "[authenticate]"+            }++--------------------------------------------------------------------------------+authenticatedCmd :: Command+authenticatedCmd =+    Command { cmdWord8 = 0xF3+            , cmdDesc  = "[authenticated]"+            }++--------------------------------------------------------------------------------+notAuthenticatedCmd :: Command+notAuthenticatedCmd =+    Command { cmdWord8 = 0xF4+            , cmdDesc  = "[not-authenticated]"+            }++--------------------------------------------------------------------------------+identifyClientCmd :: Command+identifyClientCmd =+    Command { cmdWord8 = 0xF5+            , cmdDesc  = "[identify-client]"+            }++--------------------------------------------------------------------------------+clientIdentifiedCmd :: Command+clientIdentifiedCmd =+    Command { cmdWord8 = 0xF6+            , cmdDesc  = "[client-identified]"+            }++--------------------------------------------------------------------------------+unknownCmd :: Word8 -> Command+unknownCmd w =+    Command { cmdWord8 = w+            , cmdDesc  = "[unknown: "<> pack (prettyWord8 w) <> "]"+            }++--------------------------------------------------------------------------------+getCommand :: Word8 -> Command+getCommand w =+    case lookup w _cmdDict of+        Just cmd -> cmd+        Nothing  -> unknownCmd w++--------------------------------------------------------------------------------+_cmdDict :: HashMap Word8 Command+_cmdDict = mapFromList+    [ (0x01, heartbeatRequestCmd)+    , (0x02, heartbeatResponseCmd)+    , (0x82, writeEventsCmd)+    , (0x83, writeEventsCompletedCmd)+    , (0x84, transactionStartCmd)+    , (0x85, transactionStartCompletedCmd)+    , (0x86, transactionWriteCmd)+    , (0x87, transactionWriteCompletedCmd)+    , (0x88, transactionCommitCmd)+    , (0x89, transactionCommitCompletedCmd)+    , (0x8A, deleteStreamCmd)+    , (0x8B, deleteStreamCompletedCmd)+    , (0xB0, readEventCmd)+    , (0xB1, readEventCompletedCmd)+    , (0xB2, readStreamEventsForwardCmd)+    , (0xB3, readStreamEventsForwardCompletedCmd)+    , (0xB4, readStreamEventsBackwardCmd)+    , (0xB5, readStreamEventsBackwardCompletedCmd)+    , (0xB6, readAllEventsForwardCmd)+    , (0xB7, readAllEventsForwardCompletedCmd)+    , (0xB8, readAllEventsBackwardCmd)+    , (0xB9, readAllEventsBackwardCompletedCmd)+    , (0xC0, subscribeToStreamCmd)+    , (0xC1, subscriptionConfirmationCmd)+    , (0xC2, streamEventAppearedCmd)+    , (0xC3, unsubscribeFromStreamCmd)+    , (0xC4, subscriptionDroppedCmd)+    , (0xC5, connectToPersistentSubscriptionCmd)+    , (0xC6, persistentSubscriptionConfirmationCmd)+    , (0xC7, persistentSubscriptionStreamEventAppearedCmd)+    , (0xC8, createPersistentSubscriptionCmd)+    , (0xC9, createPersistentSubscriptionCompletedCmd)+    , (0xCA, deletePersistentSubscriptionCmd)+    , (0xCB, deletePersistentSubscriptionCompletedCmd)+    , (0xCC, persistentSubscriptionAckEventsCmd)+    , (0xCD, persistentSubscriptionNakEventsCmd)+    , (0xCE, updatePersistentSubscriptionCmd)+    , (0xCF, updatePersistentSubscriptionCompletedCmd)+    , (0xF0, badRequestCmd)+    , (0xF1, notHandledCmd)+    , (0xF2, authenticateCmd)+    , (0xF3, authenticatedCmd)+    , (0xF4, notAuthenticatedCmd)+    , (0xF5, identifyClientCmd)+    , (0xF6, clientIdentifiedCmd)+    ]

@@ -0,0 +1,67 @@+{-# LANGUAGE DeriveGeneric             #-}+{-# LANGUAGE DeriveDataTypeable        #-}+{-# LANGUAGE ExistentialQuantification #-}+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Internal.Communication+-- Copyright : (C) 2017 Yorick Laupa+-- License : (see the file LICENSE)+--+-- Maintainer : Yorick Laupa <yo.eight@gmail.com>+-- Stability : provisional+-- Portability : non-portable+--+--------------------------------------------------------------------------------+module Database.EventStore.Internal.Communication where++--------------------------------------------------------------------------------+import Data.Typeable++--------------------------------------------------------------------------------+import Database.EventStore.Internal.Callback+import Database.EventStore.Internal.Operation+import Database.EventStore.Internal.Prelude+import Database.EventStore.Internal.Types++--------------------------------------------------------------------------------+data SystemInit = SystemInit deriving Typeable++--------------------------------------------------------------------------------+data SystemShutdown = SystemShutdown deriving Typeable++--------------------------------------------------------------------------------+data Service+  = ConnectionManager+  | TimerService+  deriving (Show, Eq, Enum, Bounded, Typeable, Generic)++--------------------------------------------------------------------------------+instance Hashable Service++--------------------------------------------------------------------------------+data Initialized = Initialized Service deriving Typeable++--------------------------------------------------------------------------------+data InitFailed = InitFailed Service deriving Typeable++--------------------------------------------------------------------------------+data FatalException+  = forall e. Exception e => FatalException e+  | FatalCondition Text+  deriving Typeable++--------------------------------------------------------------------------------+data SubmitOperation =+  forall a. SubmitOperation (Callback a) (Operation a)+  deriving Typeable++--------------------------------------------------------------------------------+data ServiceTerminated = ServiceTerminated Service deriving Typeable++--------------------------------------------------------------------------------+data NewTimer =+  forall e. Typeable e => NewTimer e Duration Bool+  deriving Typeable++--------------------------------------------------------------------------------+newtype SendPackage = SendPackage Package deriving Typeable

@@ -1,11 +1,9 @@-{-# LANGUAGE DeriveDataTypeable  #-}-{-# LANGUAGE GADTs               #-}-{-# LANGUAGE RecordWildCards     #-}-{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE LambdaCase         #-} -------------------------------------------------------------------------------- -- | -- Module : Database.EventStore.Internal.Connection--- Copyright : (C) 2015 Yorick Laupa+-- Copyright : (C) 2017 Yorick Laupa -- License : (see the file LICENSE) -- -- Maintainer : Yorick Laupa <yo.eight@gmail.com>@@ -14,284 +12,247 @@ -- -------------------------------------------------------------------------------- module Database.EventStore.Internal.Connection-    ( InternalConnection-    , ConnectionException(..)-    , connUUID-    , connClose-    , connSend-    , connRecv-    , connIsClosed-    , newConnection-    , connForceReconnect-    ) where+  ( ConnectionBuilder(..)+  , Connection(..)+  , RecvOutcome(..)+  , PackageArrived(..)+  , ConnectionError(..)+  , ConnectionEstablished(..)+  , ConnectionClosed(..)+  , ConnectionRef(..)+  , getConnection+  , connectionBuilder+  , connectionError+  ) where  --------------------------------------------------------------------------------+import Prelude (String) import Text.Printf  ---------------------------------------------------------------------------------import ClassyPrelude-import Data.Serialize-import Data.UUID-import Data.UUID.V4-import Network.Connection+import           Control.Monad.Reader+import           Data.Serialize+import           Data.UUID+import qualified Network.Connection as Network  -------------------------------------------------------------------------------- import Database.EventStore.Internal.Command+import Database.EventStore.Internal.Control import Database.EventStore.Internal.EndPoint-import Database.EventStore.Internal.Discovery+import Database.EventStore.Internal.Logger+import Database.EventStore.Internal.Prelude import Database.EventStore.Internal.Types-import Database.EventStore.Logging  ----------------------------------------------------------------------------------- | Type of connection issue that can arise during the communication with the---   server.-data ConnectionException-    = MaxAttemptConnectionReached-      -- ^ The max reconnection attempt threshold has been reached.-    | ClosedConnection-      -- ^ Use of a close 'Connection'.-    | WrongPackageFraming-      -- ^ TCP package sent by the server had a wrong framing.-    | PackageParsingError String-      -- ^ Server sent a malformed TCP package.-    deriving (Show, Typeable)+newtype ConnectionBuilder =+  ConnectionBuilder { connect :: EndPoint -> EventStore Connection }  ---------------------------------------------------------------------------------instance Exception ConnectionException+data RecvOutcome+  = ResetByPeer+  | Recv Package+  | WrongFraming+  | ParsingError  ---------------------------------------------------------------------------------data In a where-    Id    :: In UUID-    Close :: In ()-    Send  :: Package -> In ()-    Recv  :: In Package-    ForceReconnect :: NodeEndPoints -> In ()+type ConnectionId = UUID  ----------------------------------------------------------------------------------- | Represents connection logic action to carry out.-data Status a where-    Noop :: Status ()-    WithConnection :: UUID -> Connection -> In a -> Status a-    CreateConnection :: In a -> Status a-    Errored :: ConnectionException -> Status a+newtype ConnectionRef =+  ConnectionRef { maybeConnection :: EventStore (Maybe Connection) }  ----------------------------------------------------------------------------------- | Internal representation of a connection with the server.-data InternalConnection =-    InternalConnection-    { _var    :: TMVar State-    , _last   :: IORef (Maybe EndPoint)-    , _disc   :: Discovery-    , _setts  :: Settings-    , _ctx    :: ConnectionContext-    }+getConnection :: ConnectionRef -> EventStore Connection+getConnection ref =+  maybeConnection ref >>= \case+    Just conn -> return conn+    Nothing   -> do+      $(logError) "Expected a connection but got none."+      throwString "No current connection (impossible situation)"  ---------------------------------------------------------------------------------data State-    = Offline-    | Online !UUID !Connection-    | Closed+data Connection =+  Connection { connectionId       :: ConnectionId+             , connectionEndPoint :: EndPoint+             , enqueuePackage     :: Package -> EventStore ()+             , dispose            :: EventStore ()+             }  ----------------------------------------------------------------------------------- | Creates a new 'InternalConnection'.-newConnection :: Settings -> Discovery -> IO InternalConnection-newConnection setts disc = do-    ctx <- initConnectionContext-    var <- newTMVarIO Offline-    ref <- newIORef Nothing-    return $ InternalConnection var ref disc setts ctx+instance Show Connection where+  show Connection{..} = "Connection [" <> show connectionId <> "] on "+                        <> show connectionEndPoint  ----------------------------------------------------------------------------------- | Gets current 'InternalConnection' 'UUID'.-connUUID :: InternalConnection -> IO UUID-connUUID conn = execute conn Id+instance Eq Connection where+  a == b = connectionId a == connectionId b  ----------------------------------------------------------------------------------- | Closes the 'InternalConnection'. It will not retry to reconnect after that---   call. it means a new 'InternalConnection' has to be created.---   'ClosedConnection' exception will be raised if the same---   'InternalConnection' object is used after a 'connClose' call.-connClose :: InternalConnection -> IO ()-connClose conn = execute conn Close+newtype ConnectionState =+  ConnectionState { _sendQueue :: TBMQueue Package }  ----------------------------------------------------------------------------------- | Sends 'Package' to the server.-connSend :: InternalConnection -> Package -> IO ()-connSend conn pkg = execute conn (Send pkg)+data PackageArrived = PackageArrived Connection Package deriving Typeable  ----------------------------------------------------------------------------------- | Asks the requested amount of bytes from the 'handle'.-connRecv :: InternalConnection -> IO Package-connRecv conn = execute conn Recv+data ConnectionError =+  ConnectionError Connection SomeException deriving Typeable  ----------------------------------------------------------------------------------- | Returns True if the connection is in closed state.-connIsClosed :: InternalConnection -> STM Bool-connIsClosed InternalConnection{..} = do-    r <- readTMVar _var-    case r of-        Closed -> return True-        _      -> return False+connectionError :: Exception e => Connection -> e -> ConnectionError+connectionError c = ConnectionError c . toException  ----------------------------------------------------------------------------------- | Forces reconnection on given node.-connForceReconnect :: InternalConnection -> NodeEndPoints -> IO ()-connForceReconnect conn = execute conn . ForceReconnect+data ConnectionClosed = ConnectionClosed Connection SomeException+  deriving Typeable  ----------------------------------------------------------------------------------- Connection Logic+data ConnectionEstablished =+  ConnectionEstablished Connection deriving Typeable+ ---------------------------------------------------------------------------------onlineLogic :: forall a. TMVar State-            -> UUID-            -> Connection-            -> In a-            -> STM (Status a)-onlineLogic var uuid conn input =-    let status = WithConnection uuid conn input-        state =-            case input of-                Close -> Closed-                _ -> Online uuid conn in-    status <$ putTMVar var state+newtype ConnectionResetByPeer = ConnectionResetByPeer SomeException+  deriving Typeable  ---------------------------------------------------------------------------------offlineLogic :: forall a. TMVar State -> In a -> STM (Status a)-offlineLogic var Close = Noop <$ putTMVar var Closed-offlineLogic _ other = return $ CreateConnection other+instance Show ConnectionResetByPeer where+  show (ConnectionResetByPeer reason) =+    "Connection reset by peer: " <> show reason  ---------------------------------------------------------------------------------closedLogic :: forall a. TMVar State -> In a -> STM (Status a)-closedLogic var input = do-    putTMVar var Closed-    case input of-        Close -> return Noop-        _ -> return $ Errored ClosedConnection+instance Exception ConnectionResetByPeer  ---------------------------------------------------------------------------------connectionLogic :: forall a. TMVar State -> In a -> STM (Status a)-connectionLogic var input = do-    state <- takeTMVar var-    case state of-        Online uuid conn -> onlineLogic var uuid conn input-        Offline -> offlineLogic var input-        Closed -> closedLogic var input+data ProtocolError+  = WrongFramingError !String+  | PackageParsingError !String+  deriving Typeable  --------------------------------------------------------------------------------+instance Show ProtocolError where+  show (WrongFramingError reason)   = "Package framing error: " <> reason+  show (PackageParsingError reason) = "Package parsing error: " <> reason+ ---------------------------------------------------------------------------------handleInput :: forall a. UUID -> Connection -> In a -> IO a-handleInput _ conn (Send pkg) = send conn pkg-handleInput _ conn Recv = recv conn-handleInput uuid _ Id = return uuid-handleInput _ conn Close = liftIO $ connectionClose conn-handleInput _ _ ForceReconnect{} = return ()+instance Exception ProtocolError  ----------------------------------------------------------------------------------- | Main connection logic. It will automatically reconnect to the server when---   a exception occured while the 'Handle' is accessed.-execute :: forall a. InternalConnection -> In a -> IO a-execute iconn input = do-    res <- atomically $ connectionLogic (_var iconn) input+connectionBuilder :: Settings -> IO ConnectionBuilder+connectionBuilder setts = do+  ctx <- Network.initConnectionContext+  return $ ConnectionBuilder $ \ept -> do+    cid <- freshUUID+    state <- createState -    case res of-        Noop -> return ()-        Errored e -> throwIO e-        WithConnection uuid conn op -> handleInput uuid conn op-        CreateConnection op -> do-            (uuid, conn) <--                case op of-                    ForceReconnect node -> openConnection iconn (Just node)-                    _ -> openConnection iconn Nothing-            atomically $ putTMVar (_var iconn) (Online uuid conn)-            handleInput uuid conn op+    mfix $ \self -> do+      tcpConnAsync <- async $+        tryAny (createConnection setts ctx ept) >>= \case+          Left e -> do+            publish (ConnectionClosed self e)+            throw e+          Right conn -> do+            publish (ConnectionEstablished self)+            return conn +      sendAsync <- async (sending state self tcpConnAsync)+      recvAsync <- async (receiving state self tcpConnAsync)+      return Connection { connectionId       = cid+                        , connectionEndPoint = ept+                        , enqueuePackage     = enqueue state+                        , dispose            = do+                            closeState state+                            disposeConnection tcpConnAsync+                            cancel sendAsync+                            cancel recvAsync+                        }+ ---------------------------------------------------------------------------------reachedMaxAttempt :: Retry -> Int -> Bool-reachedMaxAttempt KeepRetrying _ = False-reachedMaxAttempt (AtMost n) cur = n <= cur+createState :: EventStore ConnectionState+createState = ConnectionState <$> liftIO (newTBMQueueIO 500)  ---------------------------------------------------------------------------------isSsl :: Settings -> Bool-isSsl = isJust . s_ssl+closeState :: ConnectionState -> EventStore ()+closeState ConnectionState{..} = atomically $ closeTBMQueue _sendQueue  ---------------------------------------------------------------------------------openConnection :: InternalConnection-               -> Maybe NodeEndPoints-               -> IO (UUID, Connection)-openConnection InternalConnection{..} nodeM = attempt 1+createConnection :: Settings+                 -> Network.ConnectionContext+                 -> EndPoint+                 -> EventStore Network.Connection+createConnection setts ctx ept = liftIO $ Network.connectTo ctx params   where-    delay = s_reconnect_delay_secs _setts * secs--    handleFailure trialCount = do-        threadDelay delay-        when (reachedMaxAttempt (s_retry _setts) trialCount) $ do-            atomically $ putTMVar _var Closed-            throwIO MaxAttemptConnectionReached-        attempt (trialCount + 1)--    attempt trialCount = do-        _settingsLog _setts (Info $ Connecting trialCount)-        case nodeM of-            Just node -> do-                let ept = if isSsl _setts-                          then let Just pt = secureEndPoint node-                               in pt-                          else tcpEndPoint node--                    host = endPointIp ept-                    port = endPointPort ept-                res <- tryAny $ connect _setts _ctx host port-                case res of-                    Left _ -> handleFailure trialCount-                    Right st -> st <$ writeIORef _last (Just ept)-            Nothing -> do-                old <- readIORef _last-                ept_opt <- runDiscovery _disc old-                case ept_opt of-                    Nothing -> handleFailure trialCount-                    Just ept -> do-                        let host = endPointIp ept-                            port = endPointPort ept-                        res <- tryAny $ connect _setts _ctx host port-                        case res of-                            Left _ -> handleFailure trialCount-                            Right st -> st <$ writeIORef _last (Just ept)+    host   = endPointIp ept+    port   = fromIntegral $ endPointPort ept+    params = Network.ConnectionParams host port (s_ssl setts) Nothing  ---------------------------------------------------------------------------------secs :: Int-secs = 1000000+disposeConnection :: Async Network.Connection -> EventStore ()+disposeConnection as = traverse_ tryDisposing =<< poll as+  where+    tryDisposing = traverse_ disposing+    disposing    = liftIO . Network.connectionClose  ---------------------------------------------------------------------------------connect :: Settings-        -> ConnectionContext-        -> String-        -> Int-        -> IO (UUID, Connection)-connect sett ctx host port = do-    let params = ConnectionParams host (fromIntegral port) (s_ssl sett) Nothing-    conn <- connectTo ctx params-    uuid <- nextRandom-    _settingsLog sett (Info $ Connected uuid)-    return  (uuid, conn)+receivePackage :: Connection -> Network.Connection -> EventStore Package+receivePackage self conn =+  tryAny (liftIO $ Network.connectionGetExact conn 4) >>= \case+    Left e -> do+      publish (ConnectionClosed self e)+      throw e+    Right frame ->+      case runGet getLengthPrefix frame of+        Left reason -> do+          let cause = WrongFramingError reason+          publish (connectionError self cause)+          throw cause+        Right prefix -> do+          tryAny (liftIO $ Network.connectionGetExact conn prefix) >>= \case+            Left e -> do+              publish (ConnectionClosed self e)+              throw e+            Right payload ->+              case runGet getPackage payload of+                Left reason -> do+                  let cause = PackageParsingError reason+                  publish (connectionError self cause)+                  throw cause+                Right pkg -> return pkg  ----------------------------------------------------------------------------------- Binary operations+receiving :: ConnectionState+          -> Connection+          -> Async Network.Connection+          -> EventStore ()+receiving ConnectionState{..} self tcpConnAsync =+  forever . go =<< wait tcpConnAsync+  where+    go conn =+      publish . PackageArrived self =<< receivePackage self conn+ ---------------------------------------------------------------------------------recv :: Connection -> IO Package-recv con = do-    header_bs <- connectionGetExact con 4-    case runGet getLengthPrefix header_bs of-        Left _              -> throwIO WrongPackageFraming-        Right length_prefix -> do-            bs <- connectionGetExact con length_prefix-            case runGet getPackage bs of-                Left e    -> throwIO $ PackageParsingError e-                Right pkg -> return pkg+enqueue :: ConnectionState -> Package -> EventStore ()+enqueue ConnectionState{..} pkg@Package{..} = do+  $(logDebug) [i|Package enqueued: #{pkg}|]+  atomically $ writeTBMQueue _sendQueue pkg  ---------------------------------------------------------------------------------send :: Connection -> Package -> IO ()-send  con pkg = connectionPut con bs+sending :: ConnectionState+        -> Connection+        -> Async Network.Connection+        -> EventStore ()+sending ConnectionState{..} self tcpConnAsync = go =<< wait tcpConnAsync   where-    bs = runPut $ putPackage pkg+    go conn =+      let loop     = traverse_ send =<< atomically (readTBMQueue _sendQueue)+          send pkg =+            tryAny (liftIO $ Network.connectionPut conn bytes) >>= \case+              Left e  -> publish (ConnectionClosed self e)+              Right _ -> do+                monitorAddDataTransmitted (length bytes)+                loop+            where+              bytes = runPut $ putPackage pkg in+      loop  -------------------------------------------------------------------------------- -- Serialization@@ -304,22 +265,22 @@     putWord8 flag_word8     putLazyByteString corr_bytes     for_ cred_m $ \(Credentials login passw) -> do-        putWord8 $ fromIntegral $ olength login+        putWord8 $ fromIntegral $ length login         putByteString login-        putWord8 $ fromIntegral $ olength passw+        putWord8 $ fromIntegral $ length passw         putByteString passw     putByteString pack_data   where     pack_data     = packageData pkg     cred_len      = maybe 0 credSize cred_m-    length_prefix = fromIntegral (olength pack_data + mandatorySize + cred_len)+    length_prefix = fromIntegral (length pack_data + mandatorySize + cred_len)     cred_m        = packageCred pkg     flag_word8    = maybe 0x00 (const 0x01) cred_m     corr_bytes    = toByteString $ packageCorrelation pkg  -------------------------------------------------------------------------------- credSize :: Credentials -> Int-credSize (Credentials login passw) = olength login + olength passw + 2+credSize (Credentials login passw) = length login + length passw + 2  -------------------------------------------------------------------------------- -- | The minimun size a 'Package' should have. It's basically a command byte,@@ -344,7 +305,7 @@     dta  <- getBytes rest      let pkg = Package-              { packageCmd         = Command cmd+              { packageCmd         = getCommand cmd               , packageCorrelation = col               , packageData        = dta               , packageCred        = cred

@@ -0,0 +1,499 @@+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE OverloadedStrings  #-}+{-# LANGUAGE LambdaCase         #-}+{-# LANGUAGE RecordWildCards    #-}+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Internal.ConnectionManager+-- Copyright : (C) 2017 Yorick Laupa+-- License : (see the file LICENSE)+--+-- Maintainer : Yorick Laupa <yo.eight@gmail.com>+-- Stability : provisional+-- Portability : non-portable+--+--------------------------------------------------------------------------------+module Database.EventStore.Internal.ConnectionManager+  ( connectionManager ) where++--------------------------------------------------------------------------------+import Data.Typeable++--------------------------------------------------------------------------------+import Control.Monad.Reader+import Data.Time++--------------------------------------------------------------------------------+import           Database.EventStore.Internal.Callback+import           Database.EventStore.Internal.Command+import           Database.EventStore.Internal.Communication+import           Database.EventStore.Internal.Connection+import           Database.EventStore.Internal.Control+import           Database.EventStore.Internal.Discovery+import           Database.EventStore.Internal.EndPoint+import           Database.EventStore.Internal.Logger+import           Database.EventStore.Internal.Operation+import qualified Database.EventStore.Internal.OperationManager as Operation+import           Database.EventStore.Internal.Prelude+import           Database.EventStore.Internal.Stopwatch+import           Database.EventStore.Internal.Types++--------------------------------------------------------------------------------+data Stage+  = Init+  | Connecting Attempts ConnectingState+  | Connected Connection+  | Closed++--------------------------------------------------------------------------------+instance Show Stage where+  show Init             = "Init"+  show (Connecting a s) = "Connecting: " <> show (a, s)+  show (Connected c)    = "Connected on" <> show c+  show Closed           = "Closed"++--------------------------------------------------------------------------------+data ConnectingState+  = Reconnecting+  | EndpointDiscovery+  | ConnectionEstablishing Connection+  deriving Show++--------------------------------------------------------------------------------+data Attempts =+  Attempts { attemptCount     :: !Int+           , attemptLastStart :: !NominalDiffTime+           } deriving Show++--------------------------------------------------------------------------------+freshAttempt :: Stopwatch -> EventStore Attempts+freshAttempt = fmap (Attempts 1) . stopwatchElapsed++--------------------------------------------------------------------------------+data ConnectionMaxAttemptReached = ConnectionMaxAttemptReached+  deriving Typeable++--------------------------------------------------------------------------------+instance Show ConnectionMaxAttemptReached where+  show _ = "Reconnection limit reached."++--------------------------------------------------------------------------------+instance Exception ConnectionMaxAttemptReached++--------------------------------------------------------------------------------+data EstablishConnection = EstablishConnection EndPoint deriving Typeable++--------------------------------------------------------------------------------+newtype CloseConnection = CloseConnection SomeException+  deriving (Show, Typeable)++--------------------------------------------------------------------------------+instance Exception CloseConnection++--------------------------------------------------------------------------------+data Tick = Tick deriving Typeable++--------------------------------------------------------------------------------+timerPeriod :: Duration+timerPeriod = msDuration 200++--------------------------------------------------------------------------------+data HeartbeatStage = Interval | Timeout++--------------------------------------------------------------------------------+data HeartbeatTracker =+  HeartbeatTracker { _pkgNum         :: !Integer+                   , _heartbeatStage :: !HeartbeatStage+                   , _startedSince   :: !NominalDiffTime+                   }++--------------------------------------------------------------------------------+newHeartbeatTracker :: MonadBaseControl IO m+                    => Stopwatch+                    -> m (IORef HeartbeatTracker)+newHeartbeatTracker =+  newIORef . HeartbeatTracker 0 Interval <=< stopwatchElapsed++--------------------------------------------------------------------------------+initHeartbeatTracker :: Internal -> EventStore ()+initHeartbeatTracker Internal{..} = do+  elapsed <- stopwatchElapsed _stopwatch+  pkgNum  <- readIORef _lastPkgNum+  let tracker = HeartbeatTracker pkgNum Interval elapsed+  atomicWriteIORef _tracker tracker++--------------------------------------------------------------------------------+data Internal =+  Internal { _disc          :: Discovery+           , _builder       :: ConnectionBuilder+           , _stage         :: IORef Stage+           , _last          :: IORef (Maybe EndPoint)+           , _sending       :: TVar Bool+           , _opMgr         :: Operation.Manager+           , _stopwatch     :: Stopwatch+           , _lastCheck     :: IORef NominalDiffTime+           , _lastConnected :: IORef Bool+           , _tracker       :: IORef HeartbeatTracker+           , _lastPkgNum    :: IORef Integer+           }++--------------------------------------------------------------------------------+incrPackageNumber :: Internal -> EventStore ()+incrPackageNumber Internal{..} = do+  atomicModifyIORef' _lastPkgNum $ \n -> (n + 1, ())+  monitorIncrPkgCount++--------------------------------------------------------------------------------+connectionManager :: ConnectionBuilder+                  -> Discovery+                  -> Hub+                  -> IO ()+connectionManager builder disc mainBus = do+  stageRef <- newIORef Init+  let mkInternal = Internal disc builder stageRef+      connRef    = ConnectionRef $ lookingUpConnection stageRef++  stopwatch    <- newStopwatch+  timeoutCheck <- stopwatchElapsed stopwatch+  internal <- mkInternal <$> newIORef Nothing+                         <*> newTVarIO False+                         <*> Operation.new connRef+                         <*> return stopwatch+                         <*> newIORef timeoutCheck+                         <*> newIORef False+                         <*> newHeartbeatTracker stopwatch+                         <*> newIORef 0++  subscribe mainBus (onInit internal)+  subscribe mainBus (onEstablish internal)+  subscribe mainBus (onEstablished internal)+  subscribe mainBus (onArrived internal)+  subscribe mainBus (onSubmitOperation internal)+  subscribe mainBus (onConnectionError internal)+  subscribe mainBus (onConnectionClosed internal)+  subscribe mainBus (onCloseConnection internal)+  subscribe mainBus (onShutdown internal)+  subscribe mainBus (onTick internal)+  subscribe mainBus (onSendPackage internal)++--------------------------------------------------------------------------------+onInit :: Internal -> SystemInit -> EventStore ()+onInit self@Internal{..} _ = do+  publish (NewTimer Tick timerPeriod False)+  startConnect self++--------------------------------------------------------------------------------+startConnect :: Internal -> EventStore ()+startConnect self@Internal{..} =+  readIORef _stage >>= \case+    Init -> do+      atts <- freshAttempt _stopwatch+      atomicWriteIORef _stage (Connecting atts Reconnecting)+      discover self+    _ -> return ()++--------------------------------------------------------------------------------+discover :: Internal -> EventStore ()+discover Internal{..} =+  readIORef _stage >>= \case+    Connecting att p ->+      case p of+        Reconnecting{} -> do+          atomicWriteIORef _stage (Connecting att EndpointDiscovery)+          old <- readIORef _last+          _   <- fork $+              tryAny (liftIO $ runDiscovery _disc old) >>= \case+                Left e -> do+                  $logError+                    [i| Failed to resolve TCP endpoint to which to connect #{e}.|]+                  publish (CloseConnection e)+                Right opt ->+                  case opt of+                    Nothing -> do+                      $logWarn+                        "Failed to resolve TCP endpoint to which to connect."+                    Just ept -> publish (EstablishConnection ept)+          return ()+        _ -> return ()+    _ -> return ()++--------------------------------------------------------------------------------+establish :: Internal -> EndPoint -> EventStore ()+establish Internal{..} ept = do+  $(logDebug) [i|Establish tcp connection on [#{ept}]|]+  readIORef _stage >>= \case+    Connecting att s ->+      case s of+        EndpointDiscovery -> do+          conn <- connect _builder ept+          connected <- atomicModifyIORef' _lastConnected $ \c -> (True, c)+          unless connected $+            publish (Initialized ConnectionManager)+          atomicWriteIORef _stage (Connecting att (ConnectionEstablishing conn))+        _ -> return ()+    _ -> return ()++--------------------------------------------------------------------------------+established :: Internal -> Connection -> EventStore ()+established self@Internal{..} conn =+  readIORef _stage >>= \case+    Connecting _ (ConnectionEstablishing known) -> do+      when (conn == known) $ do+        $logDebug [i|TCP connection established: #{conn}.|]+        atomicWriteIORef _stage (Connected conn)+        initHeartbeatTracker self+    _ -> return ()++--------------------------------------------------------------------------------+onEstablished :: Internal -> ConnectionEstablished -> EventStore ()+onEstablished self (ConnectionEstablished conn) = established self conn++--------------------------------------------------------------------------------+closeConnection :: Exception e => Internal -> e -> EventStore ()+closeConnection self@Internal{..} cause = do+  $logDebug [i|CloseConnection: #{cause}.|]+  mConn <- lookupConnectionAndSwitchToClosed self+  Operation.cleanup _opMgr+  traverse_ (closeTcpConnection self cause) mConn+  $logInfo [i|CloseConnection: connection cleanup done for [#{cause}].|]+  publish (FatalException cause)++--------------------------------------------------------------------------------+lookupConnectionAndSwitchToClosed :: Internal -> EventStore (Maybe Connection)+lookupConnectionAndSwitchToClosed self@Internal{..} = do+  outcome <- lookupConnection self+  atomicWriteIORef _stage Closed+  return outcome++--------------------------------------------------------------------------------+closeTcpConnection :: Exception e => Internal -> e -> Connection -> EventStore ()+closeTcpConnection Internal{..} cause conn = do+  let cid = connectionId conn+  $logDebug [i|CloseTcpConnection: connection [#{cid}]. Cause: #{cause}.|]+  dispose conn+  $logDebug [i|CloseTcpConnection: connection [#{cid}] disposed.|]++  readIORef _stage >>= \case+    Closed -> return ()+    stage  -> do+      att <-+        case stage of+          Connecting old _ -> return old+          _                -> freshAttempt _stopwatch+      atomicWriteIORef _stage (Connecting att Reconnecting)++--------------------------------------------------------------------------------+data ForceReconnect = ForceReconnect EndPoint deriving (Typeable, Show)++--------------------------------------------------------------------------------+instance Exception ForceReconnect++--------------------------------------------------------------------------------+forceReconnect :: Internal -> NodeEndPoints -> EventStore ()+forceReconnect self@Internal{..} node = do+  setts <- getSettings+  let ept = if isJust $ s_ssl setts+            then let Just pt = secureEndPoint node in pt+            else tcpEndPoint node++  Connected conn <- readIORef _stage+  when (connectionEndPoint conn /= ept) $ do+    monitorIncrForceReconnect+    closeTcpConnection self (ForceReconnect ept) conn+    att <- freshAttempt _stopwatch+    atomicWriteIORef _stage (Connecting att EndpointDiscovery)+    $logInfo [i|#{conn}: going to reconnect to #{ept}.|]+    establish self ept++--------------------------------------------------------------------------------+onEstablish :: Internal -> EstablishConnection -> EventStore ()+onEstablish self (EstablishConnection ept) = establish self ept++--------------------------------------------------------------------------------+onTick :: Internal -> Tick -> EventStore ()+onTick self@Internal{..} _ = do+  setts <- getSettings+  readIORef _stage >>= \case+    Connecting Attempts{..} s+      | onGoingConnection s -> do+        elapsed <- stopwatchElapsed _stopwatch+        if elapsed - attemptLastStart >= s_reconnect_delay setts+          then do+            let retries = attemptCount + 1+                att     = Attempts retries elapsed+            atomicWriteIORef _stage (Connecting att Reconnecting)+            case s_retry setts of+              AtMost n+                | attemptCount <= n -> retryConnection attemptCount+                | otherwise -> maxAttemptReached+              KeepRetrying -> retryConnection attemptCount+          else return ()+      | otherwise -> manageHeartbeats self+    Connected _ -> do+      elapsed           <- stopwatchElapsed _stopwatch+      timeoutCheckStart <- readIORef _lastCheck++      when (elapsed - timeoutCheckStart >= s_operationTimeout setts) $ do+        Operation.check _opMgr+        atomicWriteIORef _lastCheck elapsed++      manageHeartbeats self+    _ -> return ()+  where+    onGoingConnection Reconnecting             = True+    onGoingConnection ConnectionEstablishing{} = True+    onGoingConnection _                        = False++    maxAttemptReached = do+      closeConnection self ConnectionMaxAttemptReached+      publish (FatalException ConnectionMaxAttemptReached)++    retryConnection cnt = do+      $logDebug [i|Checking reconnection... (attempt #{cnt}).|]+      discover self++--------------------------------------------------------------------------------+data ServerHeartbeatTimeout = ServerHeartbeatTimeout deriving Typeable++--------------------------------------------------------------------------------+instance Show ServerHeartbeatTimeout where+  show _ = "Server connection has heartbeat timeout"++--------------------------------------------------------------------------------+instance Exception ServerHeartbeatTimeout++--------------------------------------------------------------------------------+manageHeartbeats :: Internal -> EventStore ()+manageHeartbeats self@Internal{..} = traverse_ go =<< lookupConnection self+  where+    go conn = do+      elapsed <- stopwatchElapsed _stopwatch+      pkgNum  <- readIORef _lastPkgNum+      tracker <- readIORef _tracker+      setts   <- getSettings++      let interval    = s_heartbeatInterval setts+          timeout     = s_heartbeatInterval setts+          initTracker = tracker+                        { _heartbeatStage = Interval+                        , _startedSince   = elapsed+                        , _pkgNum         = pkgNum+                        }++      if _pkgNum tracker /= pkgNum+        then atomicWriteIORef _tracker initTracker+        else+          case _heartbeatStage tracker of+            Interval+              | elapsed - _startedSince tracker >= interval -> do+                uuid <- freshUUID+                let pkg        = heartbeatRequestPackage uuid+                    newTracker = tracker+                                 { _heartbeatStage = Timeout+                                 , _startedSince   = elapsed+                                 , _pkgNum         = pkgNum+                                 }+                enqueuePackage conn pkg+                atomicWriteIORef _tracker newTracker+              | otherwise -> return ()+            Timeout+              | elapsed - _startedSince tracker >= timeout -> do+                monitorIncrHeartbeatTimeouts+                $logInfo [i|Closing #{conn} due to HEARTBEAT TIMEOUT at pkgNum #{pkgNum}|]+                closeTcpConnection self ServerHeartbeatTimeout conn+              | otherwise -> return ()++--------------------------------------------------------------------------------+onArrived :: Internal -> PackageArrived -> EventStore ()+onArrived self@Internal{..} (PackageArrived conn pkg@Package{..}) = do+  withConnection $ \knownConn -> do+    if knownConn == conn+      then do+        $logDebug [i|Package received:  #{pkg}.|]+        incrPackageNumber self+        handlePackage+      else do+        $logDebug [i|Package IGNORED: #{pkg}.|]+  where+    withConnection :: (Connection -> EventStore ()) -> EventStore ()+    withConnection k = do+      readIORef _stage >>= \case+        Connecting _ (ConnectionEstablishing c) -> k c+        Connected c -> k c+        _ -> $logDebug [i|Package IGNORED: #{pkg}|]++    heartbeatResponse = heartbeatResponsePackage packageCorrelation++    handlePackage+      | packageCmd == heartbeatResponseCmd = return ()+      | packageCmd == heartbeatRequestCmd =+        enqueuePackage conn heartbeatResponse+      | otherwise =+        Operation.handle _opMgr pkg >>= \case+          Nothing       -> $logWarn [i|Package not handled: #{pkg}|]+          Just decision ->+            case decision of+              Operation.Handled        -> return ()+              Operation.Reconnect node -> forceReconnect self node++--------------------------------------------------------------------------------+isSameConnection :: Internal -> Connection -> EventStore Bool+isSameConnection Internal{..} conn = go <$> readIORef _stage+  where+    go (Connected known)                             = known == conn+    go (Connecting _ (ConnectionEstablishing known)) = known == conn+    go _                                             = False++--------------------------------------------------------------------------------+onConnectionError :: Internal -> ConnectionError -> EventStore ()+onConnectionError self@Internal{..} (ConnectionError conn e) =+  whenM (isSameConnection self conn) $ do+    $logError [i|TCP #{conn} error. Cause: #{e}.|]+    closeConnection self e++--------------------------------------------------------------------------------+onConnectionClosed :: Internal -> ConnectionClosed -> EventStore ()+onConnectionClosed self@Internal{..} (ConnectionClosed conn cause) =+  whenM (isSameConnection self conn) $ do+    closeTcpConnection self cause conn+    monitorIncrConnectionDrop++--------------------------------------------------------------------------------+onShutdown :: Internal -> SystemShutdown -> EventStore ()+onShutdown self@Internal{..} _ = do+  $logDebug "Shutting down..."+  mConn <- lookupConnectionAndSwitchToClosed self+  Operation.cleanup _opMgr+  traverse_ dispose mConn+  $logDebug "Shutdown properly."+  publish (ServiceTerminated ConnectionManager)++--------------------------------------------------------------------------------+onSubmitOperation :: Internal -> SubmitOperation -> EventStore ()+onSubmitOperation Internal{..} (SubmitOperation callback op) =+  readIORef _stage >>= \case+    Closed -> reject callback Aborted+    _      -> Operation.submit _opMgr op callback++--------------------------------------------------------------------------------+onCloseConnection :: Internal -> CloseConnection -> EventStore ()+onCloseConnection self e = closeConnection self e++--------------------------------------------------------------------------------+lookupConnection :: Internal -> EventStore (Maybe Connection)+lookupConnection Internal{..} = lookingUpConnection _stage++--------------------------------------------------------------------------------+lookingUpConnection :: IORef Stage -> EventStore (Maybe Connection)+lookingUpConnection ref = go <$> readIORef ref+  where+    go (Connected conn)                             = Just conn+    go (Connecting _ (ConnectionEstablishing conn)) = Just conn+    go _                                            = Nothing++--------------------------------------------------------------------------------+onSendPackage :: Internal -> SendPackage -> EventStore ()+onSendPackage self (SendPackage pkg) =+  traverse_ sending =<< lookupConnection self+  where+    sending conn = enqueuePackage conn pkg

@@ -0,0 +1,419 @@+{-# LANGUAGE CPP                        #-}+{-# LANGUAGE DeriveDataTypeable         #-}+{-# LANGUAGE GADTs                      #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Internal.Control+-- Copyright : (C) 2014 Yorick Laupa+-- License : (see the file LICENSE)+--+-- Maintainer : Yorick Laupa <yo.eight@gmail.com>+-- Stability : provisional+-- Portability : non-portable+--+--------------------------------------------------------------------------------+module Database.EventStore.Internal.Control+  (+    -- * Control+    EventStore+  , runEventStore+  , getSettings+  , freshUUID+    -- * Messaging+  , Pub(..)+  , Sub(..)+  , Hub+  , Subscribe+  , Publish+  , asPub+  , asSub+  , asHub+  , Bus+  , newBus+  , busStop+  , Message+  , toMsg+  , fromMsg+  , busProcessedEverything+  , publish+  , publishWith+  , subscribe+  , stopBus+  , publisher+    -- * Monitoring+  , monitorIncrPkgCount+  , monitorIncrConnectionDrop+  , monitorAddDataTransmitted+  , monitorIncrForceReconnect+  , monitorIncrHeartbeatTimeouts+    -- * Re-export+  , module Database.EventStore.Internal.Settings+  ) where++--------------------------------------------------------------------------------+import Data.Typeable+import Data.Typeable.Internal++--------------------------------------------------------------------------------+import Control.Monad.Reader+import Data.UUID+import Data.UUID.V4+import System.Metrics+import System.Metrics.Counter hiding (add)+import System.Metrics.Distribution++--------------------------------------------------------------------------------+import Database.EventStore.Internal.Logger+import Database.EventStore.Internal.Prelude+import Database.EventStore.Internal.Settings++--------------------------------------------------------------------------------+data Env =+  Env { __logRef   :: LoggerRef+      , __settings :: Settings+      , __bus      :: Bus+      , __monitor  :: Maybe Monitoring+      }++--------------------------------------------------------------------------------+newtype EventStore a =+  EventStore { unEventStore :: ReaderT Env IO a }+  deriving ( Functor+           , Applicative+           , Monad+           , MonadThrow+           , MonadCatch+           , MonadIO+           , MonadFix+           )++--------------------------------------------------------------------------------+getEnv :: EventStore Env+getEnv = EventStore ask++--------------------------------------------------------------------------------+getSettings :: EventStore Settings+getSettings = __settings <$> getEnv++--------------------------------------------------------------------------------+freshUUID :: EventStore UUID+freshUUID = liftIO nextRandom++--------------------------------------------------------------------------------+publisher :: EventStore Publish+publisher = fmap (asPub . __bus) getEnv++--------------------------------------------------------------------------------+stopBus :: EventStore ()+stopBus = busStop . __bus =<< getEnv++--------------------------------------------------------------------------------+instance MonadBase IO EventStore where+  liftBase m = EventStore $ liftBase m++--------------------------------------------------------------------------------+instance MonadBaseControl IO EventStore where+    type StM EventStore a = a+    liftBaseWith run = EventStore $ do+      env <- ask+      s   <- liftIO $ run (\m -> runReaderT (unEventStore m) env)+      restoreM s+    restoreM = return++--------------------------------------------------------------------------------+instance MonadLogger EventStore where+  monadLoggerLog loc src lvl msg  = do+    loggerRef <- __logRef <$> getEnv+    liftIO $ loggerCallback loggerRef loc src lvl (toLogStr msg)++--------------------------------------------------------------------------------+instance MonadLoggerIO EventStore where+  askLoggerIO = do+    loggerRef <- __logRef <$> getEnv+    return (loggerCallback loggerRef)++--------------------------------------------------------------------------------+runEventStore :: LoggerRef -> Settings -> Bus -> EventStore a -> IO a+runEventStore ref setts bus (EventStore action) =+  runReaderT action (Env ref setts bus (_monitoring bus))++--------------------------------------------------------------------------------+-- Messaging+--------------------------------------------------------------------------------+data Message where+  Message :: Typeable a => a -> Message+  deriving Typeable++--------------------------------------------------------------------------------+instance Show Message where+  show (Message a) = "Message: " <> show (typeOf a)++--------------------------------------------------------------------------------+toMsg :: Typeable a => a -> Message+toMsg = Message++--------------------------------------------------------------------------------+fromMsg :: Typeable a => Message -> Maybe a+fromMsg (Message a) = cast a++--------------------------------------------------------------------------------+class Pub p where+  publishSTM :: Typeable a => p -> a -> STM Bool++--------------------------------------------------------------------------------+publish :: Typeable a => a -> EventStore ()+publish a = do+  bus <- __bus <$> getEnv+  publishWith bus a++--------------------------------------------------------------------------------+publishWith :: (Pub p, Typeable a, MonadIO m) => p -> a -> m ()+publishWith p a = atomically $ do+  _ <- publishSTM p a+  return ()++--------------------------------------------------------------------------------+class Sub s where+  subscribeEventHandler :: s -> EventHandler -> IO ()++--------------------------------------------------------------------------------+subscribe :: (Sub s, Typeable a)+          => s+          -> (a -> EventStore ())+          -> IO ()+subscribe s k = subscribeEventHandler s (EventHandler Proxy k)++--------------------------------------------------------------------------------+data Publish = forall p. Pub p => Publish p++--------------------------------------------------------------------------------+instance Pub Publish where+  publishSTM (Publish p) a = publishSTM p a++--------------------------------------------------------------------------------+data Subscribe = forall p. Sub p => Subscribe p++--------------------------------------------------------------------------------+instance Sub Subscribe where+  subscribeEventHandler (Subscribe p) a = subscribeEventHandler p a++--------------------------------------------------------------------------------+data Hub = forall h. (Sub h, Pub h) => Hub h++--------------------------------------------------------------------------------+instance Sub Hub where+  subscribeEventHandler (Hub h) = subscribeEventHandler h++--------------------------------------------------------------------------------+instance Pub Hub where+  publishSTM (Hub h) = publishSTM h++--------------------------------------------------------------------------------+asSub :: Sub s => s -> Subscribe+asSub = Subscribe++--------------------------------------------------------------------------------+asPub :: Pub p => p -> Publish+asPub = Publish++--------------------------------------------------------------------------------+asHub :: (Sub h, Pub h) => h -> Hub+asHub = Hub++--------------------------------------------------------------------------------+data Type = Type TypeRep Fingerprint++--------------------------------------------------------------------------------+instance Show Type where+  show (Type rep _) = "type " <> show rep++--------------------------------------------------------------------------------+instance Eq Type where+  Type _ a == Type _ b = a == b++--------------------------------------------------------------------------------+instance Ord Type where+  compare (Type _ a) (Type _ b) = compare a b++--------------------------------------------------------------------------------+instance Hashable Type where+  hashWithSalt s (Type _ (Fingerprint b l)) = hashWithSalt s (b, l)++--------------------------------------------------------------------------------+data GetType+  = forall a. Typeable a => FromTypeable a+  | forall prx a. Typeable a => FromProxy (prx a)++--------------------------------------------------------------------------------+getFingerprint :: TypeRep -> Fingerprint+#if __GLASGOW_HASKELL__ == 708+getFingerprint (TypeRep fp _ _) = fp+#else+getFingerprint = typeRepFingerprint+#endif++--------------------------------------------------------------------------------+getType :: GetType -> Type+getType op = Type t (getFingerprint t)+  where+    t = case op of+          FromTypeable a -> typeOf a+          FromProxy prx  -> typeRep prx++--------------------------------------------------------------------------------+type EventHandlers = HashMap Type (Seq EventHandler)++--------------------------------------------------------------------------------+propagate :: Typeable a => a -> Seq EventHandler -> EventStore ()+propagate a = traverse_ $ \(EventHandler _ k) -> do+  let Just b = cast a+      tpe    = typeOf b+  outcome <- tryAny $ k b+  case outcome of+    Right _ -> return ()+    Left e  -> $(logError) [i|Exception when propagating #{tpe}: #{e}.|]++--------------------------------------------------------------------------------+data EventHandler where+  EventHandler :: Typeable a+               => Proxy a+               -> (a -> EventStore ())+               -> EventHandler++--------------------------------------------------------------------------------+instance Show EventHandler where+  show (EventHandler prx _) = "Handle " <> show (typeRep prx)++--------------------------------------------------------------------------------+data Bus =+  Bus { _busLoggerRef      :: LoggerRef+      , _busSettings       :: Settings+      , _busEventHandlers  :: IORef EventHandlers+      , _busQueue          :: TBMQueue Message+      , _workerAsync       :: Async ()+      , _monitoring        :: Maybe Monitoring+      }++--------------------------------------------------------------------------------+busStop :: MonadIO m => Bus -> m ()+busStop Bus{..} = atomically $ closeTBMQueue _busQueue++--------------------------------------------------------------------------------+busProcessedEverything :: Bus -> IO ()+busProcessedEverything Bus{..} = wait _workerAsync++--------------------------------------------------------------------------------+messageType :: Type+messageType = getType (FromProxy (Proxy :: Proxy Message))++--------------------------------------------------------------------------------+newBus :: LoggerRef -> Settings -> IO Bus+newBus ref setts = do+  bus <- mfix $ \b -> do+    Bus ref setts <$> newIORef mempty+                  <*> newTBMQueueIO 500+                  <*> async (worker b)+                  <*> traverse configureMonitoring (s_monitoring setts)++  return bus++--------------------------------------------------------------------------------+worker :: Bus -> IO ()+worker self@Bus{..} = loop+  where+    handleMsg (Message a) = do+      callbacks <- readIORef _busEventHandlers+      publishing self callbacks a+      loop++    loop = traverse_ handleMsg =<< atomically (readTBMQueue _busQueue)++--------------------------------------------------------------------------------+instance Sub Bus where+  subscribeEventHandler Bus{..} hdl@(EventHandler prx _) =+    atomicModifyIORef' _busEventHandlers update+    where+      update :: EventHandlers -> (EventHandlers, ())+      update callbacks =+        let tpe  = getType (FromProxy prx)+            next = alterMap $ \input ->+              case input of+                Nothing -> Just (singleton hdl)+                Just hs -> Just (snoc hs hdl) in+        (next tpe callbacks, ())++--------------------------------------------------------------------------------+instance Pub Bus where+  publishSTM Bus{..} a = do+    closed <- isClosedTBMQueue _busQueue+    writeTBMQueue _busQueue (toMsg a)+    return $ not closed++--------------------------------------------------------------------------------+publishing :: Typeable a => Bus -> EventHandlers -> a -> IO ()+publishing self@Bus{..} callbacks a = do+  let tpe = getType (FromTypeable a)+  runEventStore _busLoggerRef _busSettings self $ do+    $(logDebug) [i|Publishing message #{tpe}.|]+    traverse_ (propagate a) (lookup tpe callbacks)+    $(logDebug) [i|Message #{tpe} propagated.|]++    unless (tpe == messageType) $+      traverse_ (propagate (toMsg a)) (lookup messageType callbacks)++--------------------------------------------------------------------------------+-- Monitoring+--------------------------------------------------------------------------------+data Monitoring =+  Monitoring+  { _pkgCount     :: Counter+  , _connDrops    :: Counter+  , _dataTx       :: Distribution+  , _forceReco    :: Counter+  , _heartTimeout :: Counter+  }++--------------------------------------------------------------------------------+configureMonitoring :: Store -> IO Monitoring+configureMonitoring store =+  Monitoring <$> createCounter "eventstore.packages.received" store+             <*> createCounter "eventstore.connection.drops" store+             <*> createDistribution "eventstore.data.transmitted" store+             <*> createCounter "eventstore.force_reconnect" store+             <*> createCounter "eventstore.heartbeat.timeouts" store++--------------------------------------------------------------------------------+monitorIncrPkgCount :: EventStore ()+monitorIncrPkgCount = do+  Env{..} <- getEnv+  for_ __monitor  $ \Monitoring{..}->+    liftIO $ inc _pkgCount++--------------------------------------------------------------------------------+monitorIncrConnectionDrop :: EventStore ()+monitorIncrConnectionDrop = do+  Env{..} <- getEnv+  for_ __monitor  $ \Monitoring{..}->+    liftIO $ inc _connDrops++--------------------------------------------------------------------------------+monitorAddDataTransmitted :: Int -> EventStore ()+monitorAddDataTransmitted siz = do+  Env{..} <- getEnv+  for_ __monitor  $ \Monitoring{..}->+    liftIO $ add _dataTx (fromIntegral siz)++--------------------------------------------------------------------------------+monitorIncrForceReconnect :: EventStore ()+monitorIncrForceReconnect = do+  Env{..} <- getEnv+  for_ __monitor  $ \Monitoring{..}->+    liftIO $ inc _forceReco++--------------------------------------------------------------------------------+monitorIncrHeartbeatTimeouts :: EventStore ()+monitorIncrHeartbeatTimeouts = do+  Env{..} <- getEnv+  for_ __monitor  $ \Monitoring{..}->+    liftIO $ inc _heartTimeout

@@ -34,10 +34,10 @@     ) where  --------------------------------------------------------------------------------+import Prelude (String) import Data.Maybe  ---------------------------------------------------------------------------------import ClassyPrelude import Data.Aeson import Data.Aeson.Types import Data.Array.IO@@ -50,6 +50,7 @@  -------------------------------------------------------------------------------- import Database.EventStore.Internal.EndPoint+import Database.EventStore.Internal.Prelude  -------------------------------------------------------------------------------- data DnsDiscoveryException@@ -64,7 +65,7 @@ httpRequest :: EndPoint -> String -> IO Request httpRequest (EndPoint ip p) path = parseUrlThrow url   where-    url = "http://" ++ ip ++ ":" ++ show p ++ path+    url = "http://" <> ip <> ":" <> show p <> path  -------------------------------------------------------------------------------- -- | Represents a source of cluster gossip.@@ -228,7 +229,7 @@     parseJSON (String txt) =         case fromText txt of             Just uuid -> return $ GUUID uuid-            _         -> fail $ "Wrong UUID format " ++ show txt+            _         -> fail $ "Wrong UUID format " <> show txt     parseJSON invalid = typeMismatch "UUID" invalid  --------------------------------------------------------------------------------

@@ -12,15 +12,22 @@ module Database.EventStore.Internal.EndPoint where  ---------------------------------------------------------------------------------import ClassyPrelude+import Prelude (String)  --------------------------------------------------------------------------------+import Database.EventStore.Internal.Prelude++-------------------------------------------------------------------------------- -- | Gathers both an IPv4 and a port. data EndPoint =     EndPoint     { endPointIp   :: !String     , endPointPort :: !Int-    } deriving Show+    } deriving Eq++--------------------------------------------------------------------------------+instance Show EndPoint where+    show (EndPoint h p) = h <> ":" <> show p  -------------------------------------------------------------------------------- emptyEndPoint :: EndPoint

@@ -0,0 +1,176 @@+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE OverloadedStrings  #-}+{-# LANGUAGE RecordWildCards    #-}+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Internal.Exec+-- Copyright : (C) 2017 Yorick Laupa+-- License : (see the file LICENSE)+--+-- Maintainer : Yorick Laupa <yo.eight@gmail.com>+-- Stability : provisional+-- Portability : non-portable+--+--------------------------------------------------------------------------------+module Database.EventStore.Internal.Exec+  ( Exec+  , Terminated(..)+  , newExec+  , execSettings+  , execWaitTillClosed+  ) where++--------------------------------------------------------------------------------+import Prelude (String)+import Data.Typeable++--------------------------------------------------------------------------------+import Database.EventStore.Internal.Communication+import Database.EventStore.Internal.Connection+import Database.EventStore.Internal.ConnectionManager+import Database.EventStore.Internal.Control+import Database.EventStore.Internal.Discovery+import Database.EventStore.Internal.Logger+import Database.EventStore.Internal.Prelude+import Database.EventStore.Internal.TimerService++--------------------------------------------------------------------------------+type ServicePendingInit = HashMap Service ()++--------------------------------------------------------------------------------+data Stage+  = Init+  | Available Publish+  | Errored String++--------------------------------------------------------------------------------+newtype Terminated = Terminated String deriving (Show, Typeable)++--------------------------------------------------------------------------------+instance Exception Terminated++--------------------------------------------------------------------------------+data Exec =+  Exec { execSettings       :: Settings+       , _execPub           :: STM Publish+       , _internal          :: Internal+       , execWaitTillClosed :: IO ()+       }++--------------------------------------------------------------------------------+data Internal =+  Internal { _initRef   :: IORef ServicePendingInit+           , _finishRef :: IORef ServicePendingInit+           , _stageVar  :: TVar Stage+           }++--------------------------------------------------------------------------------+instance Pub Exec where+  publishSTM e a = do+    pub     <- _execPub e+    handled <- publishSTM pub a+    unless handled $+      throwSTM $ Terminated "Connection Closed."++    return handled++--------------------------------------------------------------------------------+stageSTM :: TVar Stage -> STM Publish+stageSTM var = do+  stage <- readTVar var+  case stage of+    Init          -> retrySTM+    Available pub -> return pub+    Errored msg   -> throwSTM $ Terminated msg++--------------------------------------------------------------------------------+errored :: TVar Stage -> String -> STM ()+errored var err = do+  stage <- readTVar var+  case stage of+    Errored _ -> return ()+    _         -> writeTVar var (Errored err)++--------------------------------------------------------------------------------+initServicePending :: ServicePendingInit+initServicePending = foldMap (\svc -> singletonMap svc ()) [minBound..]++--------------------------------------------------------------------------------+newExec :: Settings -> Bus -> ConnectionBuilder -> Discovery -> IO Exec+newExec setts mainBus builder disc = do++  internal <- Internal <$> newIORef initServicePending+                       <*> newIORef initServicePending+                       <*> newTVarIO Init++  let stagePub = stageSTM $ _stageVar internal+      exe      = Exec setts stagePub internal (busProcessedEverything mainBus)+      hub      = asHub mainBus++  timerService hub+  connectionManager builder disc hub++  subscribe mainBus (onInit internal)+  subscribe mainBus (onInitFailed internal)+  subscribe mainBus (onFatal internal)+  subscribe mainBus (onTerminated internal)+  subscribe mainBus (onShutdown internal)++  publishWith mainBus SystemInit++  return exe++--------------------------------------------------------------------------------+onInit :: Internal -> Initialized -> EventStore ()+onInit Internal{..} (Initialized svc) = do+  $logInfo [i|Service #{svc} initialized|]+  initialized <- atomicModifyIORef' _initRef $ \m ->+    let m' = deleteMap svc m in+    (m', null m')++  when initialized $ do+    $logInfo "Entire system initialized properly"+    pub <- publisher+    atomically $ writeTVar _stageVar (Available pub)++--------------------------------------------------------------------------------+onInitFailed :: Internal -> InitFailed -> EventStore ()+onInitFailed Internal{..} (InitFailed svc) = do+  atomically $ errored _stageVar "Driver failed to initialized"+  $logError [i|Service #{svc} failed to initialize.|]+  stopBus+  $logError "System can't start."++--------------------------------------------------------------------------------+onFatal :: Internal -> FatalException -> EventStore ()+onFatal self@Internal{..} situation = do+  case situation of+    FatalException e ->+      $(logOther "Fatal") [i|Fatal exception: #{e}|]+    FatalCondition msg ->+      $(logOther "Fatal") [i|Driver is in unrecoverable state: #{msg}.|]++  shutdown self++--------------------------------------------------------------------------------+onTerminated :: Internal -> ServiceTerminated -> EventStore ()+onTerminated Internal{..} (ServiceTerminated svc) = do+  $logInfo [i|Service #{svc} terminated.|]+  terminated <- atomicModifyIORef' _finishRef $ \m ->+    let m' = deleteMap svc m in+    (m', null m')++  when terminated $ do+    $logInfo "Entire system shutdown properly"+    stopBus++--------------------------------------------------------------------------------+onShutdown :: Internal -> SystemShutdown -> EventStore ()+onShutdown Internal{..} _ =+  atomically $ writeTVar _stageVar (Errored "Connection closed")++--------------------------------------------------------------------------------+shutdown :: Internal -> EventStore ()+shutdown Internal{..} = do+  atomically $ writeTVar _stageVar (Errored "Connection closed")+  publish SystemShutdown

@@ -1,587 +0,0 @@-{-# LANGUAGE DataKinds                 #-}-{-# LANGUAGE DeriveDataTypeable        #-}-{-# LANGUAGE ExistentialQuantification #-}-{-# LANGUAGE RecordWildCards           #-}-{-# LANGUAGE ScopedTypeVariables       #-}------------------------------------------------------------------------------------ |--- Module : Database.EventStore.Internal.Execution.Production--- Copyright : (C) 2015 Yorick Laupa--- License : (see the file LICENSE)------ Maintainer : Yorick Laupa <yo.eight@gmail.com>--- Stability : provisional--- Portability : non-portable------ Production execution model. It's striving for robustness. The model consists--- on 4 threads. The Reader thread that reads 'Package' from the connection, the--- Runner thread which executes finalizers submitted by the user (typically what--- to do on operation completion or when a event has arrived for a subscription)--- , the writer thread that sends 'Package' to the server, and the Manager--- thread that handles requests coming both from the user and the Reader--- thread. If the Reader or Runner threads die, it will be restarted by the--- Manager thread if the connection hasn't been closed by user in the meantime.----------------------------------------------------------------------------------module Database.EventStore.Internal.Execution.Production-    ( Production-    , newExecutionModel-    , pushOperation-    , shutdownExecutionModel-    , pushConnectStream-    , pushConnectPersist-    , pushCreatePersist-    , pushUpdatePersist-    , pushDeletePersist-    , pushAckPersist-    , pushNakPersist-    , pushUnsubscribe-    , prodWaitTillClosed-    , pushForceReconnect-    ) where-----------------------------------------------------------------------------------import Control.Exception (AsyncException(..), asyncExceptionFromException)-import Control.Monad.Fix-import Data.Int-----------------------------------------------------------------------------------import ClassyPrelude-import Data.UUID-----------------------------------------------------------------------------------import Database.EventStore.Internal.Connection-import Database.EventStore.Internal.Discovery-import Database.EventStore.Internal.EndPoint-import Database.EventStore.Internal.Generator-import Database.EventStore.Internal.Manager.Subscription.Driver hiding-    ( submitPackage-    , unsubscribe-    , ackPersist-    , nakPersist-    , abort-    )-import Database.EventStore.Internal.Manager.Subscription.Model-import Database.EventStore.Internal.Operation-import Database.EventStore.Internal.Processor-import Database.EventStore.Internal.Types-import Database.EventStore.Logging-----------------------------------------------------------------------------------data Worker-    = Reader ThreadId-    | Runner ThreadId-    | Writer ThreadId-    deriving Show------------------------------------------------------------------------------------- | Used to determine if we hit the end of the queue.-data Slot a = Slot !a | End------------------------------------------------------------------------------------- | A 'TQueue' that can be cycled.-newtype CycleQueue a = CycleQueue (TQueue (Slot a))------------------------------------------------------------------------------------- | Creates an empty 'CycleQueue'.-newCycleQueue :: IO (CycleQueue a)-newCycleQueue = fmap CycleQueue newTQueueIO------------------------------------------------------------------------------------- | Gets an element from the 'CycleQueue'.-readCycleQueue :: CycleQueue a -> STM a-readCycleQueue (CycleQueue q) = do-    Slot a <- readTQueue q-    return a------------------------------------------------------------------------------------- | Writes an element to the 'CycleQueue'.-writeCycleQueue :: CycleQueue a -> a -> STM ()-writeCycleQueue (CycleQueue q) a = writeTQueue q (Slot a)------------------------------------------------------------------------------------- | Empties a 'CycleQueue'.-emptyCycleQueue :: CycleQueue a -> STM ()-emptyCycleQueue (CycleQueue q) = writeTQueue q End >> go-  where-    go = do-        s <- readTQueue q-        case s of-            End -> return ()-            _   -> go------------------------------------------------------------------------------------- | Updates a 'CycleQueue'.-updateCycleQueue :: CycleQueue a -> (a -> STM (Maybe a)) -> STM ()-updateCycleQueue (CycleQueue q) k = writeTQueue q End >> go-  where-    go = do-        s <- readTQueue q-        case s of-            End    -> return ()-            Slot a -> do-                r <- k a-                case r of-                    Nothing -> go-                    Just a' -> writeTQueue q (Slot a') >> go------------------------------------------------------------------------------------- | Indicates if a 'CycleQueue' is empty.-isEmptyCycleQueue :: CycleQueue a -> STM Bool-isEmptyCycleQueue (CycleQueue q) = isEmptyTQueue q-----------------------------------------------------------------------------------wkUpdState :: Worker -> State -> State-wkUpdState (Reader tid) s = s { _reader = Just tid }-wkUpdState (Runner tid) s = s { _runner = Just tid }-wkUpdState (Writer tid) s = s { _writer = Just tid }------------------------------------------------------------------------------------- | Holds the execution model state.-data Production =-    Prod-    { _submit :: TVar (Msg -> IO ())-      -- ^ The action to call when pushing new command.-    , _waitClosed :: STM ()-      -- ^ Action that attests the execution model has been closed successfully.-      --   It doesn't mean the execution model hasn't been shutdown because of-      --   some random exception.-    }------------------------------------------------------------------------------------- | Main execution environment used among different transitions.-data Env =-    Env-    { _setts :: Settings-      -- ^ Global settings reference.-    , _queue :: CycleQueue Msg-      -- ^ That queue ties the user, the reader thread and the manager thread.-      --   The user and the reader push new messages onto the queue while the-      --   manager dequeue and handles one message at the time.-    , _pkgQueue :: CycleQueue Package-      -- ^ That queue ties the writer thread with the manager thread. The writer-      --   dequeue packages from that queue and sends those to the server. While-      --   the manager pushes new packages on every new submitted operation.-    , _jobQueue :: CycleQueue Job-      -- ^ That queue ties the runner thread with the manager thread. The runner-      --   dequeues IO action from it while the manager pushes new command-      --   finalizers as those arrived.-    , _state :: TVar State-      -- ^ Holds manager thread state.-    , _nextSubmit :: TVar (Msg -> IO ())-      -- ^ Indicates the action to call in order to push new commands.-    , _connVar :: TVar InternalConnection-      -- ^ Connection to the server.-    , _disposed :: TMVar ()-      -- ^ Indicates when the production execution model has been shutdown and-      --   disposed any ongoing operations.-    }-----------------------------------------------------------------------------------data Msg-    = Stopped Worker SomeException-    | Arrived Package-    | Shutdown-    | forall a.-      NewOperation (Either OperationError a -> IO ()) (Operation a)-    | ConnectStream (SubConnectEvent -> IO ()) Text Bool-    | ConnectPersist (SubConnectEvent -> IO ()) Text Text Int32-    | Unsubscribe Running-    | CreatePersist (Either PersistActionException ConfirmedAction -> IO ())-          Text Text PersistentSubscriptionSettings-    | UpdatePersist (Either PersistActionException ConfirmedAction -> IO ())-          Text Text PersistentSubscriptionSettings-    | DeletePersist (Either PersistActionException ConfirmedAction -> IO ())-          Text Text-    | AckPersist Running [UUID]-    | NakPersist Running NakAction (Maybe Text) [UUID]-    | ForceReconnect NodeEndPoints-----------------------------------------------------------------------------------pushCmd :: Production -> Msg -> IO ()-pushCmd (Prod _sender _) msg = do-    push <- readTVarIO _sender-    push msg------------------------------------------------------------------------------------- | Asks to shutdown the connection to the server asynchronously.-shutdownExecutionModel :: Production -> IO ()-shutdownExecutionModel prod = pushCmd prod Shutdown------------------------------------------------------------------------------------- | Pushes a new 'Operation' asynchronously.-pushOperation :: Production-             -> (Either OperationError a -> IO ())-             -> Operation a-             -> IO ()-pushOperation prod k op = pushCmd prod (NewOperation k op)------------------------------------------------------------------------------------- | Subscribes to a regular stream.-pushConnectStream :: Production-                  -> (SubConnectEvent -> IO ())-                  -> Text-                  -> Bool-                  -> IO ()-pushConnectStream prod k n tos = pushCmd prod (ConnectStream k n tos)------------------------------------------------------------------------------------- | Subscribes to a persistent subscription.-pushConnectPersist :: Production-                   -> (SubConnectEvent -> IO ())-                   -> Text-                   -> Text-                   -> Int32-                   -> IO ()-pushConnectPersist prod k g n buf = pushCmd prod (ConnectPersist k g n buf)------------------------------------------------------------------------------------- | Creates a persistent subscription.-pushCreatePersist :: Production-                  -> (Either PersistActionException ConfirmedAction -> IO ())-                  -> Text-                  -> Text-                  -> PersistentSubscriptionSettings-                  -> IO ()-pushCreatePersist prod k g n setts = pushCmd prod (CreatePersist k g n setts)------------------------------------------------------------------------------------- | Updates a persistent subscription.-pushUpdatePersist :: Production-                  -> (Either PersistActionException ConfirmedAction -> IO ())-                  -> Text-                  -> Text-                  -> PersistentSubscriptionSettings-                  -> IO ()-pushUpdatePersist prod k g n setts = pushCmd prod (UpdatePersist k g n setts)------------------------------------------------------------------------------------- | Deletes a persistent subscription.-pushDeletePersist :: Production-                  -> (Either PersistActionException ConfirmedAction -> IO ())-                  -> Text-                  -> Text-                  -> IO ()-pushDeletePersist prod k g n = pushCmd prod (DeletePersist k g n)------------------------------------------------------------------------------------- | Acknowledges a set of events has been successfully handled.-pushAckPersist :: Production -> Running -> [UUID] -> IO ()-pushAckPersist prod run evts = pushCmd prod (AckPersist run evts)------------------------------------------------------------------------------------- | Acknowledges a set of events hasn't been handled successfully.-pushNakPersist :: Production-               -> Running-               -> NakAction-               -> Maybe Text-               -> [UUID]-               -> IO ()-pushNakPersist prod run act res evts =-    pushCmd prod (NakPersist run act res evts)------------------------------------------------------------------------------------- | Unsubscribe from a subscription.-pushUnsubscribe :: Production -> Running -> IO ()-pushUnsubscribe prod r = pushCmd prod (Unsubscribe r)------------------------------------------------------------------------------------- | Waits the execution model to close properly.-prodWaitTillClosed :: Production -> IO ()-prodWaitTillClosed (Prod _ disposed) = atomically disposed-----------------------------------------------------------------------------------pushForceReconnect :: Production -> NodeEndPoints -> IO ()-pushForceReconnect prod n = pushCmd prod (ForceReconnect n)-----------------------------------------------------------------------------------newtype Job = Job (IO ())------------------------------------------------------------------------------------- Internal Production state.----------------------------------------------------------------------------------data State =-    State-    { _proc   :: !(Processor (IO ()))-    , _reader :: !(Maybe ThreadId)-    , _runner :: !(Maybe ThreadId)-    , _writer :: !(Maybe ThreadId)-    }-----------------------------------------------------------------------------------emptyState :: Settings -> Generator -> State-emptyState setts gen = State (newProcessor setts gen) Nothing Nothing Nothing-----------------------------------------------------------------------------------updateProc :: Processor (IO ()) -> State -> State-updateProc p s = s { _proc = p }------------------------------------------------------------------------------------- | Reader thread. Keeps reading 'Package' from the server.-reader :: Settings -> CycleQueue Msg -> InternalConnection -> IO ()-reader sett queue c = forever $ do-    pkg <- connRecv c-    let cmd  = packageCmd pkg-        uuid = packageCorrelation pkg--    atomically $ writeCycleQueue queue (Arrived pkg)-    _settingsLog sett $ Info $ PackageReceived cmd uuid------------------------------------------------------------------------------------- | Writer thread, writes incoming 'Package's----------------------------------------------------------------------------------writer :: Settings -> CycleQueue Package -> InternalConnection -> IO ()-writer setts pkg_queue conn = forever $ do-    pkg <- atomically $ readCycleQueue pkg_queue-    connSend conn pkg-    let cmd  = packageCmd pkg-        uuid = packageCorrelation pkg--    _settingsLog setts $ Info $ PackageSent cmd uuid------------------------------------------------------------------------------------- Runner thread. Keeps running job comming from the Manager thread.----------------------------------------------------------------------------------runner :: CycleQueue Job -> IO ()-runner job_queue = forever $ do-    Job j <- atomically $ readCycleQueue job_queue-    j------------------------------------------------------------------------------------- | Spawns a new thread worker.-spawn :: Env -> (ThreadId -> Worker) -> IO Worker-spawn Env{..} mk = do-    conn <- readTVarIO _connVar-    tid  <- mfix $ \tid ->-        let worker = mk tid-            action =-                case worker of-                    Reader _ -> reader _setts _queue conn-                    Runner _ -> runner _jobQueue-                    Writer _ -> writer _setts _pkgQueue conn in-        forkFinally action $ \r ->-            case r of-                Left e ->-                    case asyncExceptionFromException e of-                        Just ThreadKilled -> return ()-                        _ ->  atomically $ writeCycleQueue _queue-                                         $ Stopped worker e-                _      -> return ()-    return $ mk tid------------------------------------------------------------------------------------- | Loops over a 'Processor''s 'Transition' state machine, returning an updated---   'Processor' model at the end.-runTransition :: Env -> Transition (IO ()) -> STM (Processor (IO ()))-runTransition Env{..} = go-  where-    go (Produce j nxt) = do-        let job = Job j-        writeCycleQueue _jobQueue job-        go nxt-    go (Transmit pkg nxt) = do-        writeCycleQueue _pkgQueue pkg-        go nxt-    go (Await new_proc) = return new_proc-    go (ForceReconnectCmd node nxt) = do-        writeCycleQueue _queue (ForceReconnect node)-        go nxt------------------------------------------------------------------------------------- | First execution mode. It spawns initial reader, runner and writer threads.---   Then it switches to 'cruising' mode.-bootstrap :: Env -> IO ()-bootstrap env@Env{..} = do-    rew <- spawn env Reader-    ruw <- spawn env Runner-    wrw <- spawn env Writer-    let _F = wkUpdState rew .-             wkUpdState ruw .-             wkUpdState wrw-    atomically $ modifyTVar' _state _F-    cruising env-----------------------------------------------------------------------------------data ForceReconnectException = ForceReconnectionException NodeEndPoints-    deriving (Show, Typeable)-----------------------------------------------------------------------------------instance Exception ForceReconnectException------------------------------------------------------------------------------------- | Crusing execution mode. Reads and handle message coming from the channel as---   those are arrived. That mode is used when the connection to the server is---   still live. We might have deconnection once in a while but at the end, if---   we managed to reconnect to it, we consider everything is fine.-cruising :: Env -> IO ()-cruising env@Env{..} = do-    msg <- atomically $ readCycleQueue _queue-    s   <- readTVarIO _state-    case msg of-        Stopped _ e -> throwIO e-        ForceReconnect node -> throwIO $ ForceReconnectionException node-        Arrived pkg -> do-            let sm = submitPackage pkg $ _proc s-            atomically $ do-                new_proc <- runTransition env sm-                modifyTVar' _state $ updateProc new_proc-            cruising env-        Shutdown -> throwIO ClosedConnection-        NewOperation k op -> do-            let sm = newOperation k op $ _proc s-            atomically $ do-                new_proc <- runTransition env sm-                modifyTVar' _state $ updateProc new_proc-            cruising env-        ConnectStream k n tos -> do-            let sm = connectRegularStream k n tos $ _proc s-            atomically $ do-                new_proc <- runTransition env sm-                modifyTVar' _state $ updateProc new_proc-            cruising env-        ConnectPersist k g n b -> do-            let sm = connectPersistent k g n b $ _proc s-            atomically $ do-                new_proc <- runTransition env sm-                modifyTVar' _state $ updateProc new_proc-            cruising env-        Unsubscribe r -> do-            let sm = unsubscribe r $ _proc s-            atomically $ do-                new_proc <- runTransition env sm-                modifyTVar' _state $ updateProc new_proc-            cruising env-        CreatePersist k g n psetts -> do-            let sm = createPersistent k g n psetts $ _proc s-            atomically $ do-                new_proc <- runTransition env sm-                modifyTVar' _state $ updateProc new_proc-            cruising env-        UpdatePersist k g n psetts -> do-            let sm = updatePersistent k g n psetts $ _proc s-            atomically $ do-                new_proc <- runTransition env sm-                modifyTVar' _state $ updateProc new_proc-            cruising env-        DeletePersist k g n -> do-            let sm = deletePersistent k g n $ _proc s-            atomically $ do-                new_proc <- runTransition env sm-                modifyTVar' _state $ updateProc new_proc-            cruising env-        AckPersist run evts -> do-            let sm = ackPersist (return ()) run evts $ _proc s-            atomically $ do-                new_proc <- runTransition env sm-                modifyTVar' _state $ updateProc new_proc-            cruising env-        NakPersist run act res evts -> do-            let sm = nakPersist (return ()) run act res evts $ _proc s-            atomically $ do-                new_proc <- runTransition env sm-                modifyTVar' _state $ updateProc new_proc-            cruising env------------------------------------------------------------------------------------- | That mode is triggered either because the user asks to shutdown the---   connection or because the connection to server has been dropped and we---   can't reconnect.-closing :: Env -> IO ()-closing env@Env{..} = do-    State _ retid rutid  wutid <- readTVarIO _state-    -- We kill reader and writer threads to avoid in fly package in the cleaning-    -- phase.-    traverse_ killThread retid-    traverse_ killThread wutid--    -- Discards every 'Package' that was about to be sent.-    atomically $ emptyCycleQueue _pkgQueue--    -- Takes care of 'Package's that have already arrived. Just in case those-    -- are completing or moving forward ongoing operations. Every ongoing-    -- request is kept for later reconnection. Some transient operations like-    -- Ack, Nak or Unsubscribe are just discard.-    atomically $ updateCycleQueue _queue $ \nxt ->-        case nxt of-            Arrived pkg -> do-                s <- readTVar _state-                let sm = submitPackage pkg $ _proc s-                nxt_proc <- runTransition env sm-                modifyTVar' _state $ updateProc nxt_proc-                return Nothing-            Shutdown -> return Nothing-            AckPersist _ _ -> return Nothing-            NakPersist _ _ _ _ -> return Nothing-            Unsubscribe _ -> return Nothing-            Stopped _ _ -> return Nothing-            _ -> return $ Just nxt--    -- If the connection is already closed, it will throw an exception. We just-    -- make sure it doesn't interfere with the cleaning process.-    conn <- readTVarIO _connVar-    _    <- try $ connClose conn :: (IO (Either ConnectionException ()))-    atomically $ do-        s <- readTVar _state-        _ <- runTransition env $ abort $ _proc s-        return ()--    -- Waits the runner thread to deal with its jobs list.-    atomically $ do-        end <- isEmptyCycleQueue _jobQueue-        unless end retrySTM--    traverse_ killThread rutid-----------------------------------------------------------------------------------raiseException :: Exception e => e -> Msg -> IO ()-raiseException e _ = throwIO e-----------------------------------------------------------------------------------handles :: SomeException -> [Handler IO a] -> IO a-handles e [] = throwIO e-handles e (Handler k:hs) =-    case fromException e of-        Just t -> k t-        _ -> handles e hs------------------------------------------------------------------------------------- | Main Production execution model entry point.-newExecutionModel :: Settings -> Discovery -> IO Production-newExecutionModel setts disc = do-    gen       <- newGenerator-    queue     <- newCycleQueue-    pkg_queue <- newCycleQueue-    job_queue <- newCycleQueue-    conn      <- newConnection setts disc-    conn_var  <- newTVarIO conn-    var       <- newTVarIO $ emptyState setts gen-    nxt_sub   <- newTVarIO (atomically . writeCycleQueue queue)-    disposed  <- newEmptyTMVarIO-    let env = Env setts queue pkg_queue job_queue var nxt_sub conn_var disposed-        handler res = do-            closing env-            case res of-                Left e -> do-                    _settingsLog setts (Error $ UnexpectedException e)-                    handles e-                        [ Handler $ \(_ :: ConnectionException) ->-                              atomically $ do-                                  writeTVar nxt_sub (raiseException e)-                                  putTMVar disposed ()-                        , Handler $ \(ForceReconnectionException node) -> do-                              new_conn <- newConnection setts disc-                              connForceReconnect new_conn node-                              atomically $ writeTVar conn_var new_conn-                              _ <- forkFinally (bootstrap env) handler-                              return ()-                        , Handler $ \(_ :: SomeException) -> do-                              new_conn <- newConnection setts disc-                              atomically $ writeTVar conn_var new_conn-                              _ <- forkFinally (bootstrap env) handler-                              return ()-                        ]-                _ -> atomically $ putTMVar disposed ()-    _ <- forkFinally (bootstrap env) handler-    return $ Prod nxt_sub $ do-        curConn <- readTVar conn_var-        unlessM (connIsClosed curConn) retrySTM-        readTMVar disposed

@@ -1,43 +0,0 @@------------------------------------------------------------------------------------ |--- Module : Database.EventStore.Internal.Generator--- Copyright : (C) 2015 Yorick Laupa--- License : (see the file LICENSE)------ Maintainer : Yorick Laupa <yo.eight@gmail.com>--- Stability : provisional--- Portability : non-portable------ Pure UUID generator.----------------------------------------------------------------------------------module Database.EventStore.Internal.Generator-    ( Generator-    , nextUUID-    , newGenerator-    , splitGenerator-    ) where-----------------------------------------------------------------------------------import ClassyPrelude-import Data.UUID-import System.Random------------------------------------------------------------------------------------- | Pure 'UUID' generator.-newtype Generator = Generator StdGen------------------------------------------------------------------------------------- | Gets the next fresh 'UUID'.-nextUUID :: Generator -> (UUID, Generator)-nextUUID (Generator g) =  let (u, nxt) = random g in (u, Generator nxt)------------------------------------------------------------------------------------- | Builds 2 new 'Generator's out of one.-splitGenerator :: Generator -> (Generator, Generator)-splitGenerator (Generator g) =-    let (g1, g2) = split g in (Generator g1, Generator g2)------------------------------------------------------------------------------------- | Creates a new 'Generator'.-newGenerator :: IO Generator-newGenerator = fmap Generator getStdGen

@@ -0,0 +1,73 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards   #-}+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Internal.Logger+-- Copyright : (C) 2017 Yorick Laupa+-- License : (see the file LICENSE)+--+-- Maintainer : Yorick Laupa <yo.eight@gmail.com>+-- Stability : provisional+-- Portability : non-portable+--+--------------------------------------------------------------------------------+module Database.EventStore.Internal.Logger+  ( LoggerRef+  , LoggerFilter(..)+  , newLoggerRef+  , loggerCallback+  , module Control.Monad.Logger+  , module Data.String.Interpolate.IsString+  , module System.Log.FastLogger+  ) where++--------------------------------------------------------------------------------+import Control.Monad.Logger+import Data.String.Interpolate.IsString+import System.Log.FastLogger hiding (check)++--------------------------------------------------------------------------------+import Database.EventStore.Internal.Prelude++--------------------------------------------------------------------------------+data LoggerFilter+  = LoggerFilter (LogSource -> LogLevel -> Bool)+  | LoggerLevel LogLevel++--------------------------------------------------------------------------------+toLogPredicate :: LoggerFilter -> (LogSource -> LogLevel -> Bool)+toLogPredicate (LoggerFilter k)  = k+toLogPredicate (LoggerLevel lvl) = \_ t -> t >= lvl++--------------------------------------------------------------------------------+data LoggerRef+  = LoggerRef !TimedFastLogger !LoggerFilter !Bool !(IO ())+  | NoLogger++--------------------------------------------------------------------------------+loggerCallback :: LoggerRef -> (Loc -> LogSource -> LogLevel -> LogStr -> IO ())+loggerCallback NoLogger = \_ _ _ _ -> return ()+loggerCallback (LoggerRef logger filt detailed _) = \loc src lvl msg ->+  when (predicate src lvl) $+    loggerFormat logger (if detailed then loc else defaultLoc) src lvl msg+  where+    predicate = toLogPredicate filt++--------------------------------------------------------------------------------+loggerFormat :: TimedFastLogger+             -> (Loc -> LogSource -> LogLevel -> LogStr -> IO ())+loggerFormat logger = \loc src lvl msg ->+  logger $ \t ->+    toLogStr ("["`mappend` t `mappend`"]") `mappend` " eventstore "+                                           `mappend` defaultLogStr loc src lvl msg++--------------------------------------------------------------------------------+newLoggerRef :: LogType -> LoggerFilter -> Bool -> IO LoggerRef+newLoggerRef LogNone _ _ = return NoLogger+newLoggerRef typ filt detailed =+  case typ of+    LogNone -> return NoLogger+    other   -> do+      cache             <- newTimeCache simpleTimeFormat+      (logger, cleanup) <- newTimedFastLogger cache other+      return $ LoggerRef logger filt detailed cleanup

@@ -1,204 +0,0 @@-{-# LANGUAGE ExistentialQuantification #-}-{-# LANGUAGE RecordWildCards           #-}------------------------------------------------------------------------------------ |--- Module : Database.EventStore.Internal.Manager.Operation.Model--- Copyright : (C) 2015 Yorick Laupa--- License : (see the file LICENSE)------ Maintainer : Yorick Laupa <yo.eight@gmail.com>--- Stability : provisional--- Portability : non-portable------ Main operation bookkeeping structure.----------------------------------------------------------------------------------module Database.EventStore.Internal.Manager.Operation.Model-    ( Model-    , Transition(..)-    , newModel-    , pushOperation-    , submitPackage-    , abort-    ) where-----------------------------------------------------------------------------------import ClassyPrelude-import Data.ProtocolBuffers-import Data.Serialize-import Data.UUID-----------------------------------------------------------------------------------import Database.EventStore.Internal.Command-import Database.EventStore.Internal.Generator-import Database.EventStore.Internal.Operation-import Database.EventStore.Internal.Types------------------------------------------------------------------------------------- | Entry of a running 'Operation'.-data Elem r =-    forall a resp. Decode resp =>-    Elem-    { _opOp   :: Operation a-    , _opCmd  :: Command-    , _opCont :: resp -> SM a ()-    , _opCb   :: Either OperationError a -> r-    }------------------------------------------------------------------------------------- | Operation internal state.-data State r =-    State-    { _gen :: Generator-      -- ^ 'UUID' generator.-    , _pending :: HashMap UUID (Elem r)-      -- ^ Contains all running 'Operation's.-    }-----------------------------------------------------------------------------------initState :: Generator -> State r-initState g = State g mempty------------------------------------------------------------------------------------- | Type of requests handled by the model.-data Request r-    = forall a. New (Operation a) (Either OperationError a -> r)-      -- ^ Register a new 'Operation'.-    | Pkg Package-      -- ^ Submit a package.-    | Abort-      -- ^ Aborts every pending operation.------------------------------------------------------------------------------------- | Output produces by the interpretation of an 'Operation'.-data Transition r-    = Produce r (Transition r)-      -- ^ Produces an intermediary value.-    | Transmit Package (Transition r)-      -- ^ Asks for sending the given 'Package'.-    | Await (Model r)-      -- ^ waits for more input.-    | NotHandled MasterInfo (Transition r)------------------------------------------------------------------------------------- | Main 'Operation' bookkeeping state machine.-newtype Model r = Model (Request r -> Maybe (Transition r))------------------------------------------------------------------------------------- | Pushes a new 'Operation' to model. The given 'Operation' state-machine is---   initialized and produces a 'Package'.-pushOperation :: (Either OperationError a -> r)-              -> Operation a-              -> Model r-              -> Transition r-pushOperation cb op (Model k) = let Just t = k (New op cb) in t------------------------------------------------------------------------------------- | Submits a 'Package' to the model. If the model isn't concerned by the---   'Package', it will returns 'Nothing'. Because 'Operation' can implement---   complex logic (retry for instance), it returns a 'Step'.-submitPackage :: Package -> Model r -> Maybe (Transition r)-submitPackage pkg (Model k) = k (Pkg pkg)------------------------------------------------------------------------------------- | Aborts every pending operation.-abort :: Model r -> Transition r-abort (Model k) = let Just t = k Abort in t-----------------------------------------------------------------------------------runOperation :: Settings-             -> (Either OperationError a -> r)-             -> Operation a-             -> SM a ()-             -> State r-             -> Transition r-runOperation setts cb op start init_st = go init_st start-  where-    go st (Return _) = Await $ Model $ execute setts st-    go st (Yield a n) = Produce (cb $ Right a) (go st n)-    go st (FreshId k) =-        let (new_id, nxt_gen) = nextUUID $ _gen st-            nxt_st            = st { _gen = nxt_gen } in-        go nxt_st $ k new_id-    go st (SendPkg ci co rq k) =-        let (new_uuid, nxt_gen) = nextUUID $ _gen st-            pkg = Package-                  { packageCmd         = ci-                  , packageCorrelation = new_uuid-                  , packageData        = runPut $ encodeMessage rq-                  , packageCred        = s_credentials setts-                  }-            elm    = Elem op co k cb-            ps     = insertMap new_uuid elm $ _pending st-            nxt_st = st { _pending = ps-                        , _gen     = nxt_gen-                        } in-        Transmit pkg (Await $ Model $ execute setts nxt_st)-    go st (Failure m) =-        case m of-            Just e -> Produce (cb $ Left e) (Await $ Model $ execute setts st)-            _      -> runOperation setts cb op op st-----------------------------------------------------------------------------------runPackage :: Settings -> State r -> Package -> Maybe (Transition r)-runPackage setts st pkg@Package{..} = do-    Elem op resp_cmd cont cb <- lookup packageCorrelation $ _pending st-    let nxt_ps = deleteMap packageCorrelation $ _pending st-        nxt_st = st { _pending = nxt_ps }-    case packageCmd of-        -- Bad request-        0xF0 -> do-            let reason = packageDataAsText pkg-                resp  = ServerError reason-                value = cb $ Left $ resp-            return $ Produce value (Await $ Model $ execute setts nxt_st)-        0xF4 -> do-            let value = cb $ Left NotAuthenticatedOp-            return $ Produce value (Await $ Model $ execute setts nxt_st)-        -- Not handled-        0xF1 -> do-            msg <- maybeDecodeMessage packageData-            let reason = getField $ notHandledReason msg-            case reason of-                N_NotMaster -> do-                    info <- getField $ notHandledAdditionalInfo msg-                    let next = runOperation setts cb op op nxt_st-                    return $ NotHandled (masterInfo info) next-                -- In this case with just retry the operation.-                _ -> return $ runOperation setts cb op op nxt_st-        _ | packageCmd /= resp_cmd -> do-              let r = cb $ Left $ InvalidServerResponse resp_cmd packageCmd-              return $ Produce r (Await $ Model $ execute setts nxt_st)-          | otherwise ->-              case runGet decodeMessage packageData of-                  Left e  ->-                    let r = cb $ Left $ ProtobufDecodingError e in-                    return $ Produce r (Await $ Model $ execute setts nxt_st)-                  Right m -> return $ runOperation setts cb op (cont m) nxt_st-----------------------------------------------------------------------------------abortOperations :: Settings -> State r -> Transition r-abortOperations setts init_st = go init_st $ mapToList $ _pending init_st-  where-    go st ((key, Elem _ _ _ k):xs) =-        let ps     = deleteMap key $ _pending st-            nxt_st = st { _pending = ps } in-        Produce (k $ Left Aborted) $ go nxt_st xs-    go st [] = Await $ Model $ execute setts st------------------------------------------------------------------------------------- | Creates a new 'Operation' model state-machine.-newModel :: Settings -> Generator -> Model r-newModel setts g = Model $ execute setts $ initState g-----------------------------------------------------------------------------------execute :: Settings -> State r -> Request r -> Maybe (Transition r)-execute setts st (New op cb) = Just $ runOperation setts cb op op st-execute setts st (Pkg pkg)   = runPackage setts st pkg-execute setts st Abort       = Just $ abortOperations setts st-----------------------------------------------------------------------------------maybeDecodeMessage :: Decode a => ByteString -> Maybe a-maybeDecodeMessage bytes =-    case runGet decodeMessage bytes of-        Right a -> Just a-        _       -> Nothing

@@ -0,0 +1,427 @@+{-# LANGUAGE DeriveDataTypeable        #-}+{-# LANGUAGE ExistentialQuantification #-}+{-# LANGUAGE RecordWildCards           #-}+{-# LANGUAGE TypeFamilies              #-}+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Internal.Manager.Operation.Registry+-- Copyright : (C) 2017 Yorick Laupa+-- License : (see the file LICENSE)+--+-- Maintainer : Yorick Laupa <yo.eight@gmail.com>+-- Stability : provisional+-- Portability : non-portable+--+-- Main operation bookkeeping structure.+--------------------------------------------------------------------------------+module Database.EventStore.Internal.Manager.Operation.Registry+    ( Registry+    , OperationMaxAttemptReached(..)+    , Decision(..)+    , newRegistry+    , register+    , handlePackage+    , abortPendingRequests+    , checkAndRetry+    , startAwaitings+    ) where++--------------------------------------------------------------------------------+import Data.ProtocolBuffers+import Data.Serialize+import Data.Time+import Data.UUID+import Data.UUID.V4++--------------------------------------------------------------------------------+import Database.EventStore.Internal.Callback+import Database.EventStore.Internal.Command+import Database.EventStore.Internal.Control+import Database.EventStore.Internal.Connection+import Database.EventStore.Internal.EndPoint+import Database.EventStore.Internal.Logger+import Database.EventStore.Internal.Operation hiding (retry)+import Database.EventStore.Internal.Prelude+import Database.EventStore.Internal.Stopwatch+import Database.EventStore.Internal.Types++--------------------------------------------------------------------------------+data Request =+  Request { _requestCmd     :: !Command+          , _requestPayload :: !ByteString+          }++--------------------------------------------------------------------------------+data Suspend a+  = Required !(Package -> Operation a)+  | Optional !(Maybe Package -> Operation a)+  | Resolved !(Operation a)++--------------------------------------------------------------------------------+type SessionId  = Integer+type SessionMap = HashMap SessionId Session++--------------------------------------------------------------------------------+data Session =+  forall result.+  Session { sessionId       :: !SessionId+          , sessionOp       :: !(Operation result)+          , sessionStack    :: !(IORef (Suspend result))+          , sessionCallback :: !(Callback result)+          }++--------------------------------------------------------------------------------+rejectSession :: Exception e => Session -> e -> EventStore ()+rejectSession Session{..} = reject sessionCallback++--------------------------------------------------------------------------------+resumeSession :: Registry -> Session -> Package -> EventStore ()+resumeSession reg session@Session{..} pkg = do+  atomicModifyIORef' sessionStack $ \case+      Required k -> (Resolved $ k pkg, ())+      Optional k -> (Resolved $ k (Just pkg), ())+      same       -> (same, ())++  execute reg session++--------------------------------------------------------------------------------+resumeNoPkgSession :: Registry -> Session -> EventStore ()+resumeNoPkgSession reg session@Session{..} = do+  atomicModifyIORef' sessionStack $ \case+    Optional k -> (Resolved $ k Nothing, ())+    same       -> (same, ())++  execute reg session++--------------------------------------------------------------------------------+reinitSession :: Session -> EventStore ()+reinitSession Session{..} = atomicWriteIORef sessionStack (Resolved sessionOp)++--------------------------------------------------------------------------------+restartSession :: Registry -> Session -> EventStore ()+restartSession reg session = do+  reinitSession session+  execute reg session++--------------------------------------------------------------------------------+data Sessions =+  Sessions { sessionsNextId :: IORef SessionId+           , sessionsMap    :: IORef SessionMap+           }++--------------------------------------------------------------------------------+createSession :: Sessions -> Operation a -> Callback a -> EventStore Session+createSession Sessions{..} op cb = do+  sid <- atomicModifyIORef' sessionsNextId $ \n -> (succ n, n)+  ref <- newIORef (Resolved op)+  atomicModifyIORef' sessionsMap $ \m ->+    let session =+          Session { sessionId       = sid+                  , sessionOp       = op+                  , sessionStack    = ref+                  , sessionCallback = cb+                  } in+    (insertMap sid session m, session)++--------------------------------------------------------------------------------+destroySession :: Sessions -> Session -> EventStore ()+destroySession Sessions{..} s =+  atomicModifyIORef' sessionsMap $ \m -> (deleteMap (sessionId s) m, ())++--------------------------------------------------------------------------------+sessionDisposed :: Sessions -> Session -> EventStore Bool+sessionDisposed Sessions{..} s =+  notMember (sessionId s) <$> readIORef sessionsMap++--------------------------------------------------------------------------------+newSessions :: IO Sessions+newSessions =+  Sessions <$> newIORef 0 +           <*> newIORef mempty++--------------------------------------------------------------------------------+packageOf :: Settings -> Request -> UUID -> Package+packageOf setts Request{..} uuid =+  Package { packageCmd         = _requestCmd+          , packageCorrelation = uuid+          , packageData        = _requestPayload+          , packageCred        = s_credentials setts+          }++--------------------------------------------------------------------------------+data Pending =+    Pending { _pendingRequest :: !(Maybe Request)+            , _pendingSession :: !Session+            , _pendingRetries :: !Int+            , _pendingLastTry :: !NominalDiffTime+            , _pendingConnId  :: !UUID+            }++--------------------------------------------------------------------------------+destroyPendingSession :: Sessions -> Pending -> EventStore ()+destroyPendingSession sessions Pending{..} =+  destroySession sessions _pendingSession++--------------------------------------------------------------------------------+type PendingRequests = HashMap UUID Pending++--------------------------------------------------------------------------------+data Awaiting+  = Awaiting !Session+  | AwaitingRequest !Session !Request++--------------------------------------------------------------------------------+type Awaitings = [Awaiting]++--------------------------------------------------------------------------------+rejectPending :: Exception e => Pending -> e -> EventStore ()+rejectPending Pending{..} = rejectSession _pendingSession++--------------------------------------------------------------------------------+applyResponse :: Registry -> Pending -> Package -> EventStore ()+applyResponse reg Pending{..} = resumeSession reg _pendingSession++--------------------------------------------------------------------------------+restartPending :: Registry -> Pending -> EventStore ()+restartPending reg Pending{..} = restartSession reg _pendingSession++--------------------------------------------------------------------------------+data Registry =+    Registry  { _regConnRef   :: ConnectionRef+              , _regPendings  :: IORef PendingRequests+              , _regAwaitings :: IORef Awaitings+              , _stopwatch    :: Stopwatch+              , _sessions     :: Sessions+              }++--------------------------------------------------------------------------------+newRegistry :: ConnectionRef -> IO Registry+newRegistry ref =+   Registry ref <$> newIORef mempty+                <*> newIORef []+                <*> newStopwatch+                <*> newSessions++--------------------------------------------------------------------------------+scheduleAwait :: Registry -> Awaiting -> EventStore ()+scheduleAwait Registry{..} aw =+  atomicModifyIORef' _regAwaitings $ \stack ->+    (aw : stack, ())++--------------------------------------------------------------------------------+execute :: Registry -> Session -> EventStore ()+execute self@Registry{..} session@Session{..} =+  readIORef sessionStack >>= \case+    Resolved action -> loop action+    _               -> return ()+  where+    loop (MachineT m) =+      case m of+        Failed e -> do+          destroySession _sessions session+          rejectSession session e+        Retry -> do+          reinitSession session+          execute self session+        Proceed s ->+          case s of+            Stop  -> destroySession _sessions session+            Yield a next -> do+              fulfill sessionCallback a+              loop next+            Await k tpe _ ->+              case tpe of+                NeedUUID  -> loop . k =<< freshUUID+                NeedRemote cmd payload -> do+                  let req = Request { _requestCmd     = cmd+                                    , _requestPayload = payload+                                    }+                  atomicWriteIORef sessionStack (Required k)+                  maybeConnection _regConnRef >>= \case+                    Nothing   -> scheduleAwait self (AwaitingRequest session req)+                    Just conn -> issueRequest self session conn req++                WaitRemote uuid -> do+                  atomicWriteIORef sessionStack (Optional k)+                  let mkNewPending conn = do+                        let connId = connectionId conn+                        pending <- createPending session _stopwatch connId Nothing+                        insertPending self uuid pending++                  traverse_ mkNewPending =<< maybeConnection _regConnRef++--------------------------------------------------------------------------------+issueRequest :: Registry+             -> Session+             -> Connection+             -> Request+             -> EventStore ()+issueRequest reg@Registry{..} session conn req = do+  uuid    <- liftBase nextRandom+  pending <- createPending session _stopwatch (connectionId conn) (Just req)++  insertPending reg uuid pending+  setts <- getSettings+  enqueuePackage conn (packageOf setts req uuid)++--------------------------------------------------------------------------------+createPending :: MonadBaseControl IO m+              => Session+              -> Stopwatch+              -> UUID+              -> Maybe Request+              -> m Pending+createPending session stopwatch connId mReq = do+  elapsed <- stopwatchElapsed stopwatch+  let pending =+        Pending { _pendingRequest = mReq+                , _pendingSession = session+                , _pendingRetries = 1+                , _pendingLastTry = elapsed+                , _pendingConnId  = connId+                }++  return pending++--------------------------------------------------------------------------------+insertPending :: Registry -> UUID -> Pending -> EventStore ()+insertPending Registry{..} key pending =+  atomicModifyIORef' _regPendings $ \pendings ->+    (insertMap key pending pendings, ())++--------------------------------------------------------------------------------+register :: Registry -> Operation a -> Callback a -> EventStore ()+register reg op cb = do+  session <- createSession (_sessions reg) op cb+  execute reg session++--------------------------------------------------------------------------------+abortPendingRequests :: Registry -> EventStore ()+abortPendingRequests Registry{..} = do+  m <- atomicModifyIORef' _regPendings $ \pendings -> (mempty, pendings)++  for_ m $ \p -> rejectPending p Aborted++--------------------------------------------------------------------------------+data Decision+  = Handled+  | Reconnect NodeEndPoints++--------------------------------------------------------------------------------+handlePackage :: Registry -> Package -> EventStore (Maybe Decision)+handlePackage reg@Registry{..} pkg@Package{..} = do+  outcome <- atomicModifyIORef' _regPendings $ \pendings ->+    let uuid = packageCorrelation in+    (deleteMap uuid pendings, lookup uuid pendings)++  case outcome of+    Nothing      -> return Nothing+    Just pending -> Just <$> executePending reg pkg pending++--------------------------------------------------------------------------------+executePending :: Registry -> Package -> Pending -> EventStore Decision+executePending reg@Registry{..} pkg@Package{..} p@Pending{..} =+  case packageCmd of+    cmd | cmd == badRequestCmd -> do+            let reason = packageDataAsText pkg++            rejectPending p (ServerError reason)+            return Handled+        | cmd == notAuthenticatedCmd -> do+            rejectPending p NotAuthenticatedOp+            return Handled+        | cmd == notHandledCmd -> do+            $(logDebug) [i|Not handled response received: #{pkg}.|]+            let Just msg = maybeDecodeMessage packageData+                reason   = getField $ notHandledReason msg+            case reason of+              N_NotMaster -> do+                let Just details = getField $ notHandledAdditionalInfo msg+                    info         = masterInfo details+                    node         = masterInfoNodeEndPoints info++                restartPending reg p+                return $ Reconnect node+              -- In this case with just retry the operation.+              _ -> Handled <$ restartPending reg p+        | otherwise -> do+            applyResponse reg p pkg+            return Handled++--------------------------------------------------------------------------------+-- | Occurs when an operation has been retried more than 's_operationRetry'.+data OperationMaxAttemptReached =+  OperationMaxAttemptReached UUID Command+  deriving (Show, Typeable)++--------------------------------------------------------------------------------+instance Exception OperationMaxAttemptReached++--------------------------------------------------------------------------------+checkAndRetry :: Registry -> EventStore ()+checkAndRetry self@Registry{..} = do+  pendings    <- readIORef _regPendings+  elapsed     <- stopwatchElapsed _stopwatch+  newPendings <- foldM (checking elapsed) pendings (mapToList pendings)+  atomicWriteIORef _regPendings newPendings+  where+    checking elapsed reg (key, p) = do+      conn  <- getConnection _regConnRef+      setts <- getSettings+      case _pendingRequest p of+        Nothing+          | connectionId conn == _pendingConnId p -> return reg+          | otherwise -> do+            resumeNoPkgSession self (_pendingSession p)+            disposed <- sessionDisposed _sessions (_pendingSession p)+            let newReg = if disposed then deleteMap key reg else reg+            return newReg+        Just req -> do+          let lastTry    = _pendingLastTry p+              hasTimeout = elapsed - lastTry >= s_operationTimeout setts+          if hasTimeout || _pendingConnId p /= connectionId conn+            then do+              let retry = do+                    uuid <- liftBase nextRandom+                    let pkg     = packageOf setts req uuid+                        pending =+                          p { _pendingLastTry = elapsed+                            , _pendingRetries = _pendingRetries p + 1+                            , _pendingConnId  = connectionId conn+                            }+                        nextReg = deleteMap key $ insertMap uuid pending reg++                    enqueuePackage conn pkg+                    return nextReg++              case s_operationRetry setts of+                AtMost maxAttempts+                  | _pendingRetries p <= maxAttempts+                    -> retry+                  | otherwise -> do+                    let cmd = _requestCmd req+                    destroyPendingSession _sessions p+                    rejectPending p (OperationMaxAttemptReached key cmd)+                    return $ deleteMap key reg+                KeepRetrying -> retry+            else return reg++--------------------------------------------------------------------------------+startAwaitings :: Registry -> EventStore ()+startAwaitings reg@Registry{..} = do+  awaitings <- atomicModifyIORef' _regAwaitings $ \stack ->+    ([], reverse stack)++  traverse_ starting awaitings+  where+    starting (Awaiting session)            = execute reg session+    starting (AwaitingRequest session req) = do+      conn <- getConnection _regConnRef+      issueRequest reg session conn req++--------------------------------------------------------------------------------+maybeDecodeMessage :: Decode a => ByteString -> Maybe a+maybeDecodeMessage bytes =+    case runGet decodeMessage bytes of+        Right a -> Just a+        _       -> Nothing

@@ -1,103 +0,0 @@------------------------------------------------------------------------------------ |--- Module : Database.EventStore.Internal.Manager.Subscription.Command--- Copyright : (C) 2016 Yorick Laupa--- License : (see the file LICENSE)------ Maintainer : Yorick Laupa <yo.eight@gmail.com>--- Stability : provisional--- Portability : non-portable-------------------------------------------------------------------------------------module Database.EventStore.Internal.Manager.Subscription.Command where-----------------------------------------------------------------------------------import ClassyPrelude-import Data.ProtocolBuffers-import Data.Serialize-----------------------------------------------------------------------------------import Database.EventStore.Internal.Command-import Database.EventStore.Internal.Manager.Subscription.Message-import Database.EventStore.Internal.Manager.Subscription.Types-import Database.EventStore.Internal.Types-----------------------------------------------------------------------------------data ServerMessage-    = EventAppearedMsg !ResolvedEvent-    | PersistentEventAppearedMsg !ResolvedEvent-    | ConfirmationMsg !Int64 !(Maybe Int32)-    | PersistentConfirmationMsg !Text !Int64 !(Maybe Int32)-    | PersistentCreatedMsg CreatePersistentSubscriptionResult-    | PersistentUpdatedMsg UpdatePersistentSubscriptionResult-    | PersistentDeletedMsg DeletePersistentSubscriptionResult-    | DroppedMsg SubDropReason-    | BadRequestMsg !(Maybe Text)-    | NotHandledMsg !NotHandledReason !(Maybe MasterInfo)-    | NotAuthenticatedMsg !(Maybe Text)-    | UnknownMsg !(Maybe Command)-----------------------------------------------------------------------------------toSubDropReason :: DropReason -> SubDropReason-toSubDropReason D_Unsubscribed                  = SubUnsubscribed-toSubDropReason D_NotFound                      = SubNotFound-toSubDropReason D_AccessDenied                  = SubAccessDenied-toSubDropReason D_PersistentSubscriptionDeleted = SubPersistDeleted-toSubDropReason D_SubscriberMaxCountReached     = SubSubscriberMaxCountReached-----------------------------------------------------------------------------------decodeServerMessage :: Package -> ServerMessage-decodeServerMessage pkg = fromMaybe (UnknownMsg $ Just cmd) $ go cmd-  where-    cmd = packageCmd pkg-    go 0xC2 = do-        msg <- maybeDecodeMessage $ packageData pkg-        let evt = newResolvedEventFromBuf $ getField $ streamResolvedEvent msg-        return $ EventAppearedMsg evt-    go 0xC7 = do-        msg <- maybeDecodeMessage $ packageData pkg-        let evt = newResolvedEvent $ getField $ psseaEvt msg-        return $ PersistentEventAppearedMsg evt-    go 0xC1 = do-        msg <- maybeDecodeMessage $ packageData pkg-        let lcp = getField $ subscribeLastCommitPos msg-            len = getField $ subscribeLastEventNumber msg-        return $ ConfirmationMsg lcp len-    go 0xC6 = do-        msg <- maybeDecodeMessage $ packageData pkg-        let lcp = getField $ pscLastCommitPos msg-            sid = getField $ pscId msg-            len = getField $ pscLastEvtNumber msg-        return $ PersistentConfirmationMsg sid lcp len-    go 0xC9 =-        fmap (PersistentCreatedMsg . getField . cpscResult)-            $ maybeDecodeMessage-            $ packageData pkg-    go 0xCF =-        fmap (PersistentUpdatedMsg . getField . upscResult)-            $ maybeDecodeMessage-            $ packageData pkg-    go 0xCB =-        fmap (PersistentDeletedMsg . getField . dpscResult)-            $ maybeDecodeMessage-            $ packageData pkg-    go 0xC4 = do-        msg <- maybeDecodeMessage $ packageData pkg-        let reason  = fromMaybe D_Unsubscribed $ getField $ dropReason msg-        return $ DroppedMsg $ toSubDropReason reason-    go 0xF0 = return $ BadRequestMsg $ packageDataAsText pkg-    go 0xF4 = return $ NotAuthenticatedMsg $ packageDataAsText pkg-    go 0xF1 = do-        msg <- maybeDecodeMessage $ packageData pkg-        let info = fmap masterInfo $ getField $ notHandledAdditionalInfo msg-            reason = getField $ notHandledReason msg-        return $ NotHandledMsg reason info--    go _ = Nothing-----------------------------------------------------------------------------------maybeDecodeMessage :: Decode a => ByteString -> Maybe a-maybeDecodeMessage bytes =-    case runGet decodeMessage bytes of-        Right a -> Just a-        _       -> Nothing

@@ -1,550 +0,0 @@-{-# LANGUAGE DeriveDataTypeable         #-}-{-# LANGUAGE GADTs                      #-}-{-# LANGUAGE GeneralizedNewtypeDeriving #-}-{-# LANGUAGE OverloadedStrings          #-}-{-# LANGUAGE Rank2Types                 #-}-{-# LANGUAGE RecordWildCards            #-}-{-# LANGUAGE ScopedTypeVariables        #-}------------------------------------------------------------------------------------ |--- Module : Database.EventStore.Internal.Manager.Subscription.Driver--- Copyright : (C) 2015 Yorick Laupa--- License : (see the file LICENSE)------ Maintainer : Yorick Laupa <yo.eight@gmail.com>--- Stability : provisional--- Portability : non-portable------ Subscription model driver. It drivers the model accordingly depending on the--- 'Package' or commands submitted to it.----------------------------------------------------------------------------------module Database.EventStore.Internal.Manager.Subscription.Driver-    ( SubDropReason(..)-    , SubConnectEvent(..)-    , PersistActionException(..)-    , ConfirmedAction(..)-    , NakAction(..)-    , Driver-    , newDriver-    , submitPackage-    , connectToStream-    , connectToPersist-    , createPersist-    , updatePersist-    , deletePersist-    , ackPersist-    , nakPersist-    , unsubscribe-    , abort-    ) where-----------------------------------------------------------------------------------import Data.Int-import Data.Maybe-----------------------------------------------------------------------------------import ClassyPrelude hiding (group)-import Control.Monad.State-import Data.UUID-----------------------------------------------------------------------------------import Database.EventStore.Internal.Generator-import Database.EventStore.Internal.Manager.Subscription.Command-import Database.EventStore.Internal.Manager.Subscription.Message-import Database.EventStore.Internal.Manager.Subscription.Model-import Database.EventStore.Internal.Manager.Subscription.Packages-import Database.EventStore.Internal.Manager.Subscription.Types-import Database.EventStore.Internal.Types------------------------------------------------------------------------------------- | Set of events that can occurs during a subscription lifetime.-data SubConnectEvent-    = EventAppeared ResolvedEvent-      -- ^ A wild event appeared !-    | Dropped SubDropReason-      -- ^ The subscription connection dropped.-    | SubConfirmed Running-      -- ^ Subscription connection is confirmed. It means that subscription can-      --   receive events from the server.-    | Unsubscribed------------------------------------------------------------------------------------- | Enumerates all persistent action exceptions.-data PersistActionException-    = PersistActionFail-      -- ^ The action failed.-    | PersistActionAlreadyExist-      -- ^ Happens when creating a persistent subscription on a stream with a-      --   group name already taken.-    | PersistActionDoesNotExist-      -- ^ An operation tried to do something on a persistent subscription or a-      --   stream that don't exist.-    | PersistActionAccessDenied-      -- ^ The current user is not allowed to operate on the supplied stream or-      --   persistent subscription.-    | PersistActionAborted-      -- ^ That action has been aborted because the user shutdown the connection-      --   to the server or the connection to the server is no longer possible.-    deriving (Show, Typeable)-----------------------------------------------------------------------------------instance Exception PersistActionException------------------------------------------------------------------------------------- | Emitted when a persistent action has been carried out successfully.-data ConfirmedAction =-    ConfirmedAction-    { caId     :: !UUID-      -- ^ Action id.-    , caGroup  :: !Text-      -- ^ Subscription group name.-    , caStream :: !Text-      -- ^ Stream name.-    , caAction :: !PersistAction-      -- ^ Persistent action type.-    }------------------------------------------------------------------------------------- | Submits a 'Package' to a subscription driver. If the 'Package' was---   processed by the driver, it will return a final value and a new driver with---   its internal state updated accordingly.-submitPackage :: Package -> Driver r -> Maybe (r, Driver r)-submitPackage pkg (Driver k) = k (Pkg pkg)------------------------------------------------------------------------------------- | Starts a regular subscription connection. It returns the associated---   'Package' and updates driver internal state.-connectToStream :: (SubConnectEvent -> r)-                -> Text -- ^ Stream name.-                -> Bool -- ^ Resolve Link TOS-                -> Driver r-                -> (Package, Driver r)-connectToStream c s t (Driver k) = k (Cmd $ ConnectReg c s t)------------------------------------------------------------------------------------- | Starts a persistent subscription connection. It returns the associated---   'Package' and updates driver internal state.-connectToPersist :: (SubConnectEvent -> r)-                 -> Text  -- ^ Group name.-                 -> Text  -- ^ Stream name.-                 -> Int32 -- ^ Buffer size.-                 -> Driver r-                 -> (Package, Driver r)-connectToPersist c g s b (Driver k) = k (Cmd $ ConnectPersist c g s b)------------------------------------------------------------------------------------- | Creates a persistent subscription. It returns the associated 'Package' and---   updates driver internal state.-createPersist :: (Either PersistActionException ConfirmedAction -> r)-              -> Text -- ^ Group name.-              -> Text -- ^ Stream name.-              -> PersistentSubscriptionSettings-              -> Driver r-              -> (Package, Driver r)-createPersist c g s ss (Driver k) =-    k (Cmd $ ApplyPersistAction c g s (PersistCreate ss))------------------------------------------------------------------------------------- | Updates a persistent subscription. It returns the associated 'Package' and---   updates driver internal state.-updatePersist :: (Either PersistActionException ConfirmedAction -> r)-              -> Text -- ^ Group name.-              -> Text -- ^ Stream name.-              -> PersistentSubscriptionSettings-              -> Driver r-              -> (Package, Driver r)-updatePersist c g s ss (Driver k) =-    k (Cmd $ ApplyPersistAction c g s (PersistUpdate ss))------------------------------------------------------------------------------------- | Deletes a persistent subscription. It returns the associated 'Package' and--- updates driver internal state.-deletePersist :: (Either PersistActionException ConfirmedAction -> r)-              -> Text -- ^ Group name.-              -> Text -- ^ Stream name.-              -> Driver r-              -> (Package, Driver r)-deletePersist c g s (Driver k) =-    k (Cmd $ ApplyPersistAction c g s PersistDelete)------------------------------------------------------------------------------------- | Given a persistent subscription, acknowledges a set of events have been---   successfully processed. It returns the associated 'Package' and updates---   driver internal state.-ackPersist :: r-           -> Running-           -> [UUID] -- ^ Event ids.-           -> Driver r-           -> (Package, Driver r)-ackPersist r i evts (Driver k) = k (Cmd $ PersistAck r i evts)------------------------------------------------------------------------------------- | Given a persistent subscription, indicates a set of event haven't been---   processed correctly. It returns the associated 'Package' and updates driver---   internal state.-nakPersist :: r-           -> Running-           -> NakAction-           -> Maybe Text -- ^ Reason.-           -> [UUID]     -- ^ Event ids.-           -> Driver r-           -> (Package, Driver r)-nakPersist r i na mt evts (Driver k) =-    k (Cmd $ PersistNak r i na mt evts)------------------------------------------------------------------------------------- | Unsubscribe from a subscription.-unsubscribe :: Running -> Driver r -> (Package, Driver r)-unsubscribe r (Driver k) = k (Cmd $ Unsubscribe r)------------------------------------------------------------------------------------- | Aborts every pending action.-abort :: Driver r -> [r]-abort (Driver k) = k Abort------------------------------------------------------------------------------------- EventStore result mappers:--- =========================--- EventStore protocol has several values that means the exact same thing. Those--- functions convert a specific EventStore to uniform result type common to all--- persistent actions.----------------------------------------------------------------------------------createRException :: CreatePersistentSubscriptionResult-                 -> Maybe PersistActionException-createRException CPS_Success       = Nothing-createRException CPS_AlreadyExists = Just PersistActionAlreadyExist-createRException CPS_Fail          = Just PersistActionFail-createRException CPS_AccessDenied  = Just PersistActionAccessDenied-----------------------------------------------------------------------------------deleteRException :: DeletePersistentSubscriptionResult-                 -> Maybe PersistActionException-deleteRException DPS_Success      = Nothing-deleteRException DPS_DoesNotExist = Just PersistActionDoesNotExist-deleteRException DPS_Fail         = Just PersistActionFail-deleteRException DPS_AccessDenied = Just PersistActionAccessDenied-----------------------------------------------------------------------------------updateRException :: UpdatePersistentSubscriptionResult-                 -> Maybe PersistActionException-updateRException UPS_Success      = Nothing-updateRException UPS_DoesNotExist = Just PersistActionDoesNotExist-updateRException UPS_Fail         = Just PersistActionFail-updateRException UPS_AccessDenied = Just PersistActionAccessDenied---------------------------------------------------------------------------------------------------------------------------------------------------------------------- | Type of inputs handled by the 'Subscription' driver.-data In r a where-    -- A command consists of receiving some parameters, updating the-    -- 'Subscription' model accordingly and thus modifying driver internal-    -- state.-    Cmd :: Cmd r -> In r (Package, Driver r)-    -- A 'Package' has been submitted to the 'Subscription' driver. If the-    -- driver recognize that 'Package', it returns a final value and update-    -- the driver internal state.-    Pkg :: Package -> In r (Maybe (r, Driver r))-    -- Aborts every pending action.-    Abort :: In r [r]------------------------------------------------------------------------------------- | Set of commands handled by the driver.-data Cmd r-    = ConnectReg (SubConnectEvent -> r) Text Bool-      -- ^ Creates a regular 'Subscription' connection. When a 'SubConnectEvent'-      --   has arrived, the driver will use the provided callback and emit a-      --   final value. It holds a stream name and `Resolve Link TOS` setting.-    | ConnectPersist (SubConnectEvent -> r)-                     Text-                     Text-                     Int32-      -- ^ Creates a persistent 'Subscription' connection. When a-      --   'SubConnectEvent' has arrived, the driver will use the provided-      --   callback  and emit a final value. It holds a group name, stream-      --   name and a buffer size.--    | Unsubscribe Running-      -- ^ Unsubscribe from a subscription.--    | ApplyPersistAction (Either PersistActionException ConfirmedAction -> r)-                         Text-                         Text-                         PersistAction-      -- ^ Creates a persistent action. Depending of the failure or the success-      --   of that action, the driver will use the provided callback to emit a-      --   final value. It hols a group name, a stream name and a persistent-      --   action.--    | PersistAck r Running [UUID]-      -- ^ Acks a set of Event 'UUID' to notify those events have been correctly-      --   handled. It holds a 'Running' subscription and a set of `UUID`-      --   representing event id. When the ack would be confirmed the driver-      --   will return the supplied final value.-    | PersistNak r-                 Running-                 NakAction-                 (Maybe Text)-                 [UUID]-      -- ^ Naks a set of Event 'UUID' to notify those events haven't been-      --   handled correctly. it holds a 'Running' subscription, a 'NakAction',-      --   an optional reason and a set of event ids. When the nak would be-      --   confirmed, the driver will return the provided final value.-----------------------------------------------------------------------------------cmdSubCallback :: Cmd r -> Maybe (SubConnectEvent -> r)-cmdSubCallback (ConnectReg k _ _) = Just k-cmdSubCallback (ConnectPersist k _ _ _) = Just k-cmdSubCallback _ = Nothing------------------------------------------------------------------------------------- | Driver internal state.-data Internal r =-    Internal-    { _model :: !Model-      -- ^ Subscription model.-    , _gen :: !Generator-      -- ^ 'UUID' generator.-    , _reg :: !(HashMap UUID (Cmd r))-      -- ^ Holds ongoing commands. When stored, it means an action hasn't been-      --   confirmed yet.-    }-----------------------------------------------------------------------------------initInternal :: Generator -> Internal r-initInternal gen = Internal newModel gen mempty------------------------------------------------------------------------------------- | Subscription driver state machine.-newtype Driver r = Driver (forall a. In r a -> a)-----------------------------------------------------------------------------------newtype DriverM r m a = DriverM (ReaderT Settings (StateT (Internal r) m) a)-    deriving ( Functor-             , Applicative-             , Monad-             , MonadReader Settings-             , MonadState (Internal r)-             )-----------------------------------------------------------------------------------instance MonadTrans (DriverM r) where-    lift m = DriverM $ lift $ lift m-----------------------------------------------------------------------------------noop :: DriverM r Maybe a-noop = lift Nothing-----------------------------------------------------------------------------------modelSubRunning :: UUID -> DriverM r Maybe Running-modelSubRunning uuid = do-    model <- gets _model-    lift $ querySubscription uuid $ model-----------------------------------------------------------------------------------modelSubConfirmed :: Monad m => UUID -> Meta -> DriverM r m ()-modelSubConfirmed uuid meta = do-    model <- gets _model-    let nxt = confirmedSubscription uuid meta model-    modify $ \s -> s { _model = nxt }-----------------------------------------------------------------------------------modelActionConfirmed :: Monad m => UUID -> DriverM r m ()-modelActionConfirmed uuid =-    modify $ \s -> s { _model = confirmedAction uuid $ _model s }-----------------------------------------------------------------------------------modelUnsubscribed :: UUID -> DriverM r Maybe ()-modelUnsubscribed uuid = do-    run <- modelSubRunning uuid-    model <- gets _model-    modify $ \s -> s { _model = unsubscribed run model }-----------------------------------------------------------------------------------registerDelete :: Monad m => UUID -> DriverM r m ()-registerDelete uuid = do-    reg <- gets _reg-    let nxtR = deleteMap uuid reg-    modify $ \s -> s { _reg = nxtR }-----------------------------------------------------------------------------------registerAdd :: Monad m => UUID -> Cmd r -> DriverM r m ()-registerAdd uuid cmd = do-    reg <- gets _reg-    modify $ \s -> s { _reg = insertMap uuid cmd reg }-----------------------------------------------------------------------------------modelPersistentAction :: UUID -> DriverM r Maybe PendingAction-modelPersistentAction uuid = do-    model <- gets _model-    lift $ queryPersistentAction uuid model-----------------------------------------------------------------------------------freshUUID :: Monad m => DriverM r m UUID-freshUUID = do-    (uuid, nxtG) <- gets (nextUUID . _gen)-    modify $ \s -> s { _gen = nxtG }-    return uuid-----------------------------------------------------------------------------------modelConnectReg :: Monad m => Text -> Bool -> DriverM r m UUID-modelConnectReg stream tos = do-    uuid <- freshUUID-    model <- gets _model-    modify $ \s -> s { _model = connectReg stream tos uuid model }-    return uuid-----------------------------------------------------------------------------------modelConnectPersist :: Monad m => Text -> Text -> Int32 -> DriverM r m UUID-modelConnectPersist group name batch = do-  uuid <- freshUUID-  model <- gets _model-  modify $ \s -> s { _model = connectPersist group name batch uuid model }-  return uuid-----------------------------------------------------------------------------------modelPersistAction :: Monad m-                   => Text-                   -> Text-                   -> PersistAction-                   -> DriverM r m UUID-modelPersistAction group name action = do-    uuid <- freshUUID-    model <- gets _model-    modify $ \s -> s { _model = persistAction group name uuid action model }-    return uuid-----------------------------------------------------------------------------------runDriverM :: Monad m-           => Settings-           -> Internal r-           -> DriverM r m a-           -> m (a, Driver r)-runDriverM setts st (DriverM m) = do-    (a, nxtSt) <- runStateT (runReaderT m setts) st-    return (a, Driver $ handleDriver setts nxtSt)-----------------------------------------------------------------------------------runDriver :: Settings-          -> Internal r-          -> DriverM r Identity a-          -> (a, Driver r)-runDriver setts st action = runIdentity $ runDriverM setts st action------------------------------------------------------------------------------------- Driver main state machine.-handleDriver :: Settings -> Internal r -> In r a -> a-handleDriver setts st (Pkg pkg) = do-    let corrId = packageCorrelation pkg-    cmd <- lookup corrId $ _reg st-    let action = handleMsg corrId cmd $ decodeServerMessage pkg-    runDriverM setts st action-handleDriver setts st (Cmd cmd) =-    let action = handleCmd cmd in-    runDriver setts st action-handleDriver _ st Abort = (fmap snd $ mapToList $ _reg st) >>= _F-  where-    _F (ConnectReg k _ _)           = [k $ Dropped SubAborted]-    _F (ConnectPersist k _ _ _)     = [k $ Dropped SubAborted]-    _F (ApplyPersistAction k _ _ _) = [k $ Left PersistActionAborted]-    _F _                            = []------------------------------------------------------------------------------------- | Handles 'Package's coming from the server.-handleMsg :: UUID -> Cmd r -> ServerMessage -> DriverM r Maybe r-handleMsg corrId = go-  where-    go (ConnectReg k _ _) (EventAppearedMsg evt) = do-        _ <- modelSubRunning corrId-        return $ k $ EventAppeared evt-    go (ConnectPersist k _ _ _) (PersistentEventAppearedMsg evt) = do-        _ <- modelSubRunning corrId-        return $ k $ EventAppeared evt-    go (ConnectReg k _ _) (ConfirmationMsg lcp len) = do-        let meta = RegularMeta lcp len-        modelSubConfirmed corrId meta-        run <- modelSubRunning corrId-        return $ k $ SubConfirmed run-    go (ConnectPersist k _ _ _) (PersistentConfirmationMsg sid lcp len) = do-        let meta = PersistMeta sid lcp len-        modelSubConfirmed corrId meta-        run <- modelSubRunning corrId-        return $ k $ SubConfirmed run-    go cmd (PersistentCreatedMsg res) =-        confirmPAction cmd $ createRException res-    go cmd (PersistentUpdatedMsg res) =-        confirmPAction cmd $ updateRException res-    go cmd (PersistentDeletedMsg res) =-        confirmPAction cmd $ deleteRException res-    go cmd (DroppedMsg reason) = do-        modelUnsubscribed corrId-        registerDelete corrId-        let evt =-              case reason of-                  SubUnsubscribed -> Unsubscribed-                  _ -> Dropped reason-        k <- lift $ cmdSubCallback cmd-        return $ k evt-    go cmd (BadRequestMsg msg) =-        go cmd (DroppedMsg $ SubServerError msg)-    go cmd (NotAuthenticatedMsg msg) =-        go cmd (DroppedMsg $ SubNotAuthenticated msg)-    go cmd (NotHandledMsg reason info) =-        go cmd (DroppedMsg $ SubNotHandled reason info)-    go cmd (UnknownMsg pkgCmdM) = do-        k <- lift $ cmdSubCallback cmd-        let msgM = fmap (\c -> "unknown command: " <> tshow c) pkgCmdM-        return $ k $ Dropped $ SubServerError msgM-    go cmd _ = do-        k <- lift $ cmdSubCallback cmd-        let msg = "Logic error in Subscription Driver (the impossible happened)"-        return $ k $ Dropped $ SubClientError msg--    confirmPAction :: Cmd r-                   -> Maybe PersistActionException-                   -> DriverM r Maybe r-    confirmPAction (ApplyPersistAction k g n c) em = do-        _ <- modelPersistentAction corrId-        modelActionConfirmed corrId-        registerDelete corrId-        let evt = ConfirmedAction corrId g n c-        case em of-            Just e  -> return $ k $ Left e-            Nothing -> return $ k $ Right evt-    confirmPAction _ _ = noop------------------------------------------------------------------------------------- | Handles commands coming from the user.-handleCmd :: Monad m => Cmd r -> DriverM r m Package-handleCmd cmd@(ConnectReg _ s tos) = do-    setts <- ask-    uuid <- modelConnectReg s tos-    registerAdd uuid cmd-    return $ createConnectRegularPackage setts uuid s tos-handleCmd cmd@(ConnectPersist _ gn n b) = do-    setts <- ask-    uuid <- modelConnectPersist gn n b-    registerAdd uuid cmd-    return $ createConnectPersistPackage setts uuid gn n b-handleCmd (Unsubscribe r) = do-    setts <- ask-    return $ createUnsubscribePackage setts $ runningUUID r-handleCmd cmd@(ApplyPersistAction _ gn n a) = do-    setts <- ask-    uuid <- modelPersistAction gn n a-    registerAdd uuid cmd-    return $ createPersistActionPackage setts uuid gn n a-handleCmd (PersistAck _ run evts) = do-    setts <- ask-    let RunningPersist _ _ _ _ sid _ _ = run-        uuid = runningUUID run-    return $ createAckPackage setts uuid sid evts-handleCmd (PersistNak _ run na r evts) = do-    setts <- ask-    let RunningPersist _ _ _ _ sid _ _ = run-        uuid = runningUUID run-    return $ createNakPackage setts uuid sid na r evts------------------------------------------------------------------------------------- | Creates a new subscription 'Driver' state machine.-newDriver :: forall r. Settings -> Generator -> Driver r-newDriver setts gen = Driver $ handleDriver setts (initInternal gen)

@@ -1,408 +0,0 @@-{-# LANGUAGE DataKinds                 #-}-{-# LANGUAGE DeriveGeneric             #-}-{-# OPTIONS_GHC -fcontext-stack=26     #-}------------------------------------------------------------------------------------ |--- Module : Database.EventStore.Internal.Manager.Subscription.Message--- Copyright : (C) 2015 Yorick Laupa--- License : (see the file LICENSE)------ Maintainer : Yorick Laupa <yo.eight@gmail.com>--- Stability : provisional--- Portability : non-portable-------------------------------------------------------------------------------------module Database.EventStore.Internal.Manager.Subscription.Message where-----------------------------------------------------------------------------------import Data.Int-----------------------------------------------------------------------------------import ClassyPrelude hiding (group)-import Data.DotNet.TimeSpan-import Data.ProtocolBuffers-----------------------------------------------------------------------------------import Database.EventStore.Internal.Types------------------------------------------------------------------------------------- | Stream subscription connection request.-data SubscribeToStream-    = SubscribeToStream-      { subscribeStreamId       :: Required 1 (Value Text)-      , subscribeResolveLinkTos :: Required 2 (Value Bool)-      }-    deriving (Generic, Show)-----------------------------------------------------------------------------------instance Encode SubscribeToStream------------------------------------------------------------------------------------- | 'SubscribeToStream' smart constructor.-subscribeToStream :: Text -> Bool -> SubscribeToStream-subscribeToStream stream_id res_link_tos =-    SubscribeToStream-    { subscribeStreamId       = putField stream_id-    , subscribeResolveLinkTos = putField res_link_tos-    }------------------------------------------------------------------------------------- | Stream subscription connection response.-data SubscriptionConfirmation-    = SubscriptionConfirmation-      { subscribeLastCommitPos   :: Required 1 (Value Int64)-      , subscribeLastEventNumber :: Optional 2 (Value Int32)-      }-    deriving (Generic, Show)-----------------------------------------------------------------------------------instance Decode SubscriptionConfirmation------------------------------------------------------------------------------------- | Serialized event sent by the server when a new event has been appended to a---   stream.-data StreamEventAppeared-    = StreamEventAppeared-      { streamResolvedEvent :: Required 1 (Message ResolvedEventBuf) }-    deriving (Generic, Show)-----------------------------------------------------------------------------------instance Decode StreamEventAppeared------------------------------------------------------------------------------------- | Represents the reason subscription drop happened.-data DropReason-    = D_Unsubscribed-    | D_AccessDenied-    | D_NotFound-    | D_PersistentSubscriptionDeleted-    | D_SubscriberMaxCountReached-    deriving (Enum, Eq, Show)------------------------------------------------------------------------------------- | A message sent by the server when a subscription has been dropped.-data SubscriptionDropped-    = SubscriptionDropped-      { dropReason :: Optional 1 (Enumeration DropReason) }-    deriving (Generic, Show)-----------------------------------------------------------------------------------instance Decode SubscriptionDropped------------------------------------------------------------------------------------- | A message sent to the server to indicate the user asked to end a---   subscription.-data UnsubscribeFromStream = UnsubscribeFromStream deriving (Generic, Show)-----------------------------------------------------------------------------------instance Encode UnsubscribeFromStream------------------------------------------------------------------------------------- | Create persistent subscription request.-data CreatePersistentSubscription =-    CreatePersistentSubscription-    { cpsGroupName         :: Required 1  (Value Text)-    , cpsStreamId          :: Required 2  (Value Text)-    , cpsResolveLinkTos    :: Required 3  (Value Bool)-    , cpsStartFrom         :: Required 4  (Value Int32)-    , cpsMsgTimeout        :: Required 5  (Value Int32)-    , cpsRecordStats       :: Required 6  (Value Bool)-    , cpsLiveBufSize       :: Required 7  (Value Int32)-    , cpsReadBatchSize     :: Required 8  (Value Int32)-    , cpsBufSize           :: Required 9  (Value Int32)-    , cpsMaxRetryCount     :: Required 10 (Value Int32)-    , cpsPreferRoundRobin  :: Required 11 (Value Bool)-    , cpsChkPtAfterTime    :: Required 12 (Value Int32)-    , cpsChkPtMaxCount     :: Required 13 (Value Int32)-    , cpsChkPtMinCount     :: Required 14 (Value Int32)-    , cpsSubMaxCount       :: Required 15 (Value Int32)-    , cpsNamedConsStrategy :: Optional 16 (Value Text)-    } deriving (Generic, Show)------------------------------------------------------------------------------------- | 'CreatePersistentSubscription' smart constructor.-_createPersistentSubscription :: Text-                              -> Text-                              -> PersistentSubscriptionSettings-                              -> CreatePersistentSubscription-_createPersistentSubscription group stream sett =-    CreatePersistentSubscription-    { cpsGroupName         = putField group-    , cpsStreamId          = putField stream-    , cpsResolveLinkTos    = putField $ psSettingsResolveLinkTos sett-    , cpsStartFrom         = putField $ psSettingsStartFrom sett-    , cpsMsgTimeout        = putField-                             . fromIntegral-                             . (truncate :: Double -> Int64)-                             . totalMillis-                             $ psSettingsMsgTimeout sett-    , cpsRecordStats       = putField $ psSettingsExtraStats sett-    , cpsLiveBufSize       = putField $ psSettingsLiveBufSize sett-    , cpsReadBatchSize     = putField $ psSettingsReadBatchSize sett-    , cpsBufSize           = putField $ psSettingsHistoryBufSize sett-    , cpsMaxRetryCount     = putField $ psSettingsMaxRetryCount sett-    , cpsPreferRoundRobin  = putField False-    , cpsChkPtAfterTime    = putField-                             . fromIntegral-                             . (truncate :: Double -> Int64)-                             . totalMillis-                             $ psSettingsCheckPointAfter sett-    , cpsChkPtMaxCount     = putField $ psSettingsMaxCheckPointCount sett-    , cpsChkPtMinCount     = putField $ psSettingsMinCheckPointCount sett-    , cpsSubMaxCount       = putField $ psSettingsMaxSubsCount sett-    , cpsNamedConsStrategy = putField $ Just strText-    }-  where-    strText = strategyText $ psSettingsNamedConsumerStrategy sett-----------------------------------------------------------------------------------instance Encode CreatePersistentSubscription------------------------------------------------------------------------------------- | Create persistent subscription outcome.-data CreatePersistentSubscriptionResult-    = CPS_Success-    | CPS_AlreadyExists-    | CPS_Fail-    | CPS_AccessDenied-    deriving (Enum, Eq, Show)------------------------------------------------------------------------------------- | Create persistent subscription response.-data CreatePersistentSubscriptionCompleted =-    CreatePersistentSubscriptionCompleted-    { cpscResult :: Required 1 (Enumeration CreatePersistentSubscriptionResult)-    , cpscReason :: Optional 2 (Value Text)-    } deriving (Generic, Show)-----------------------------------------------------------------------------------instance Decode CreatePersistentSubscriptionCompleted------------------------------------------------------------------------------------- | Delete persistent subscription request.-data DeletePersistentSubscription =-    DeletePersistentSubscription-    { dpsGroupName :: Required 1 (Value Text)-    , dpsStreamId  :: Required 2 (Value Text)-    } deriving (Generic, Show)-----------------------------------------------------------------------------------instance Encode DeletePersistentSubscription------------------------------------------------------------------------------------- | 'DeletePersistentSubscription' smart construction.-_deletePersistentSubscription :: Text -> Text -> DeletePersistentSubscription-_deletePersistentSubscription group_name stream_id =-    DeletePersistentSubscription-    { dpsGroupName = putField group_name-    , dpsStreamId  = putField stream_id-    }------------------------------------------------------------------------------------- | Delete persistent subscription outcome.-data DeletePersistentSubscriptionResult-    = DPS_Success-    | DPS_DoesNotExist-    | DPS_Fail-    | DPS_AccessDenied-    deriving (Enum, Eq, Show)------------------------------------------------------------------------------------- | Delete persistent subscription response.-data DeletePersistentSubscriptionCompleted =-    DeletePersistentSubscriptionCompleted-    { dpscResult :: Required 1 (Enumeration DeletePersistentSubscriptionResult)-    , dpscReason :: Optional 2 (Value Text)-    } deriving (Generic, Show)-----------------------------------------------------------------------------------instance Decode DeletePersistentSubscriptionCompleted------------------------------------------------------------------------------------- | Update persistent subscription request.-data UpdatePersistentSubscription =-    UpdatePersistentSubscription-    { upsGroupName         :: Required 1  (Value Text)-    , upsStreamId          :: Required 2  (Value Text)-    , upsResolveLinkTos    :: Required 3  (Value Bool)-    , upsStartFrom         :: Required 4  (Value Int32)-    , upsMsgTimeout        :: Required 5  (Value Int32)-    , upsRecordStats       :: Required 6  (Value Bool)-    , upsLiveBufSize       :: Required 7  (Value Int32)-    , upsReadBatchSize     :: Required 8  (Value Int32)-    , upsBufSize           :: Required 9  (Value Int32)-    , upsMaxRetryCount     :: Required 10 (Value Int32)-    , upsPreferRoundRobin  :: Required 11 (Value Bool)-    , upsChkPtAfterTime    :: Required 12 (Value Int32)-    , upsChkPtMaxCount     :: Required 13 (Value Int32)-    , upsChkPtMinCount     :: Required 14 (Value Int32)-    , upsSubMaxCount       :: Required 15 (Value Int32)-    , upsNamedConsStrategy :: Optional 16 (Value Text)-    } deriving (Generic, Show)------------------------------------------------------------------------------------- | 'UpdatePersistentSubscription' smart constructor.-_updatePersistentSubscription :: Text-                              -> Text-                              -> PersistentSubscriptionSettings-                              -> UpdatePersistentSubscription-_updatePersistentSubscription group stream sett =-    UpdatePersistentSubscription-    { upsGroupName         = putField group-    , upsStreamId          = putField stream-    , upsResolveLinkTos    = putField $ psSettingsResolveLinkTos sett-    , upsStartFrom         = putField $ psSettingsStartFrom sett-    , upsMsgTimeout        = putField-                             . fromIntegral-                             . (truncate :: Double -> Int64)-                             . totalMillis-                             $ psSettingsMsgTimeout sett-    , upsRecordStats       = putField $ psSettingsExtraStats sett-    , upsLiveBufSize       = putField $ psSettingsLiveBufSize sett-    , upsReadBatchSize     = putField $ psSettingsReadBatchSize sett-    , upsBufSize           = putField $ psSettingsHistoryBufSize sett-    , upsMaxRetryCount     = putField $ psSettingsMaxRetryCount sett-    , upsPreferRoundRobin  = putField False-    , upsChkPtAfterTime    = putField-                             . fromIntegral-                             . (truncate :: Double -> Int64)-                             . totalMillis-                             $ psSettingsCheckPointAfter sett-    , upsChkPtMaxCount     = putField $ psSettingsMaxCheckPointCount sett-    , upsChkPtMinCount     = putField $ psSettingsMinCheckPointCount sett-    , upsSubMaxCount       = putField $ psSettingsMaxSubsCount sett-    , upsNamedConsStrategy = putField $ Just strText-    }-  where-    strText = strategyText $ psSettingsNamedConsumerStrategy sett-----------------------------------------------------------------------------------instance Encode UpdatePersistentSubscription------------------------------------------------------------------------------------- | Update persistent subscription outcome.-data UpdatePersistentSubscriptionResult-    = UPS_Success-    | UPS_DoesNotExist-    | UPS_Fail-    | UPS_AccessDenied-    deriving (Enum, Eq, Show)------------------------------------------------------------------------------------- | Update persistent subscription response.-data UpdatePersistentSubscriptionCompleted =-    UpdatePersistentSubscriptionCompleted-    { upscResult :: Required 1 (Enumeration UpdatePersistentSubscriptionResult)-    , upscReason :: Optional 2 (Value Text)-    } deriving (Generic, Show)-----------------------------------------------------------------------------------instance Decode UpdatePersistentSubscriptionCompleted------------------------------------------------------------------------------------- | Connect to a persistent subscription request.-data ConnectToPersistentSubscription =-    ConnectToPersistentSubscription-    { ctsId                  :: Required 1 (Value Text)-    , ctsStreamId            :: Required 2 (Value Text)-    , ctsAllowedInFlightMsgs :: Required 3 (Value Int32)-    } deriving (Generic, Show)-----------------------------------------------------------------------------------instance Encode ConnectToPersistentSubscription------------------------------------------------------------------------------------- | 'ConnectToPersistentSubscription' smart constructor.-_connectToPersistentSubscription :: Text-                                 -> Text-                                 -> Int32-                                 -> ConnectToPersistentSubscription-_connectToPersistentSubscription sub_id stream_id all_fly_msgs =-    ConnectToPersistentSubscription-    { ctsId                  = putField sub_id-    , ctsStreamId            = putField stream_id-    , ctsAllowedInFlightMsgs = putField all_fly_msgs-    }------------------------------------------------------------------------------------- | Ack processed events request.-data PersistentSubscriptionAckEvents =-    PersistentSubscriptionAckEvents-    { psaeId              :: Required 1 (Value Text)-    , psaeProcessedEvtIds :: Repeated 2 (Value ByteString)-    } deriving (Generic, Show)-----------------------------------------------------------------------------------instance Encode PersistentSubscriptionAckEvents------------------------------------------------------------------------------------- | 'PersistentSubscriptionAckEvents' smart constructor.-persistentSubscriptionAckEvents :: Text-                                -> [ByteString]-                                -> PersistentSubscriptionAckEvents-persistentSubscriptionAckEvents sub_id evt_ids =-    PersistentSubscriptionAckEvents-    { psaeId              = putField sub_id-    , psaeProcessedEvtIds = putField evt_ids-    }------------------------------------------------------------------------------------- | Gathers every possible Nak actions.-data NakAction-    = NA_Unknown-    | NA_Park-    | NA_Retry-    | NA_Skip-    | NA_Stop-    deriving (Enum, Eq, Show)------------------------------------------------------------------------------------- | Nak processed events request.-data PersistentSubscriptionNakEvents =-    PersistentSubscriptionNakEvents-    { psneId              :: Required 1 (Value Text)-    , psneProcessedEvtIds :: Repeated 2 (Value ByteString)-    , psneMsg             :: Optional 3 (Value Text)-    , psneAction          :: Required 4 (Enumeration NakAction)-    } deriving (Generic, Show)-----------------------------------------------------------------------------------instance Encode PersistentSubscriptionNakEvents------------------------------------------------------------------------------------- | 'PersistentSubscriptionNakEvents' smart constructor.-persistentSubscriptionNakEvents :: Text-                                -> [ByteString]-                                -> Maybe Text-                                -> NakAction-                                -> PersistentSubscriptionNakEvents-persistentSubscriptionNakEvents sub_id evt_ids msg action =-    PersistentSubscriptionNakEvents-    { psneId              = putField sub_id-    , psneProcessedEvtIds = putField evt_ids-    , psneMsg             = putField msg-    , psneAction          = putField action-    }------------------------------------------------------------------------------------- | Connection to persistent subscription response.-data PersistentSubscriptionConfirmation =-    PersistentSubscriptionConfirmation-    { pscLastCommitPos :: Required 1 (Value Int64)-    , pscId            :: Required 2 (Value Text)-    , pscLastEvtNumber :: Optional 3 (Value Int32)-    } deriving (Generic, Show)-----------------------------------------------------------------------------------instance Decode PersistentSubscriptionConfirmation------------------------------------------------------------------------------------- | Avalaible event sent by the server in the context of a persistent---   subscription..-data PersistentSubscriptionStreamEventAppeared =-    PersistentSubscriptionStreamEventAppeared-    { psseaEvt :: Required 1 (Message ResolvedIndexedEvent) }-    deriving (Generic, Show)-----------------------------------------------------------------------------------instance Decode PersistentSubscriptionStreamEventAppeared

@@ -1,317 +0,0 @@-{-# LANGUAGE GADTs           #-}-{-# LANGUAGE Rank2Types      #-}-{-# LANGUAGE RecordWildCards #-}------------------------------------------------------------------------------------ |--- Module : Database.EventStore.Internal.Manager.Subscription.Model--- Copyright : (C) 2015 Yorick Laupa--- License : (see the file LICENSE)------ Maintainer : Yorick Laupa <yo.eight@gmail.com>--- Stability : provisional--- Portability : non-portable------ Main Subscription bookkeeping structure.----------------------------------------------------------------------------------module Database.EventStore.Internal.Manager.Subscription.Model-    ( PersistAction(..)-    , PendingAction(..)-    , Running(..)-    , Meta(..)-    , Model-    , runningUUID-    , runningLastEventNumber-    , runningLastCommitPosition-    , querySubscription-    , queryPersistentAction-    , confirmedSubscription-    , confirmedAction-    , newModel-    , unsubscribed-    , connectReg-    , connectPersist-    , persistAction-    ) where-----------------------------------------------------------------------------------import Data.Int-----------------------------------------------------------------------------------import ClassyPrelude-import Data.UUID-----------------------------------------------------------------------------------import Database.EventStore.Internal.Types------------------------------------------------------------------------------------- | Type of persistent action.-data PersistAction-    = PersistCreate PersistentSubscriptionSettings-    | PersistUpdate PersistentSubscriptionSettings-    | PersistDelete------------------------------------------------------------------------------------- | Represents an persistent action that hasn't been completed yet.-data PendingAction =-    PendingAction-    { _paGroup  :: !Text-    , _paStream :: !Text-    , _paTpe    :: !PersistAction-    }-----------------------------------------------------------------------------------type Register a = HashMap UUID a------------------------------------------------------------------------------------- | Represents a 'Subscription' which is about to be confirmed.-data Pending-    = PendingReg Text Bool-      -- ^ Related to regular subscription. In order of appearance:-      ---      --   * Stream name.-      ---      --   * Resolve Link TOS.-    | PendingPersist Text Text Int32-      -- ^ Related to persistent subscription. In order of appearance:-      ---      --   * Group name.-      ---      --   * Stream name.-      ---      --   * Buffer size.-      deriving Show------------------------------------------------------------------------------------- | Represents a running subscription. Gathers useful information.-data Running-    = RunningReg UUID Text Bool Int64 (Maybe Int32)-      -- ^ Related regular subscription. In order of appearance:-      ---      --   * Subscription id.-      ---      --   * Stream name.-      ---      --   * Resolve Link TOS.-      ---      --   * Last commit position.-      ---      --   * Last event number.-    | RunningPersist UUID Text Text Int32 Text Int64 (Maybe Int32)-      -- ^ Related to persistent subscription. In order of appearance:-      ---      --   * Subscription id.-      ---      --   * Group name.-      ---      --   * Stream name.-      ---      --   * Buffer size.-      ---      --   * Persistence subscription id.-      ---      --   * Last commit position.-      ---      --   * Last event number.-      deriving Show------------------------------------------------------------------------------------- | Gets the event number of a running subscription.-runningLastEventNumber :: Running -> Maybe Int32-runningLastEventNumber (RunningReg _ _ _ _ i) = i-runningLastEventNumber (RunningPersist _ _ _ _ _ _ i) = i------------------------------------------------------------------------------------- | Gets the commit position of a running subscription.-runningLastCommitPosition :: Running -> Int64-runningLastCommitPosition (RunningReg _ _ _ i _) = i-runningLastCommitPosition (RunningPersist _ _ _ _ _ i _) = i------------------------------------------------------------------------------------- | Gets the 'UUID' of a running subscription.-runningUUID :: Running -> UUID-runningUUID (RunningReg i _ _ _ _)         = i-runningUUID (RunningPersist i _ _ _ _ _ _) = i------------------------------------------------------------------------------------- | Type of requests handled by the model.-data Request a where-    -- Read request.-    Query :: Query a -> Request a-    -- Write request.-    Execute :: Action -> Request Model------------------------------------------------------------------------------------- | Set of a piece of information we can query from the 'Subscription' model.-data Query a where-    -- Query a running 'Subscription'.-    QuerySub :: UUID -> Query (Maybe Running)-    -- Query a pending persistent action.-    QueryAction :: UUID -> Query (Maybe PendingAction)------------------------------------------------------------------------------------- | Set of actions handled by the 'Subscription' model.-data Action-    = Connect UUID Connect-      -- ^ Subscription connection.-    | Confirmed Confirmed-      -- ^ Subscription action confirmation.-    | Unsubscribed UUID-      -- ^ Subscription no longer exist.-    | PersistAction Text Text UUID PersistAction-      -- ^ Add a new persist action.------------------------------------------------------------------------------------- | Subscription connection information.-data Connect-    = ConnectReg Text Bool-      --         |    |---- Resolve TOS link.-      --         |--------- Stream name.-    | ConnectPersist Text Text Int32-      --             |    |    |---- Buffer size.-      --             |    |--------- Stream name.-      --             |-------------- Group name.------------------------------------------------------------------------------------- | Information related to a confirmed 'Subscription'.-data Meta-    = RegularMeta Int64 (Maybe Int32)-      --          |     |------------- Last commit position.-      --          |------------------- Last event number.-    | PersistMeta Text Int64 (Maybe Int32)-      --          |    |     |------------- Subscription Id.-      --          |    |------------------- Last commit position.-      --          |------------------------ Last event number.------------------------------------------------------------------------------------- | Subscription action confirmation.-data Confirmed-    = ConfirmedConnection UUID Meta-      -- ^ Confirms a 'Subscription' connection has handled successfully.-    | ConfirmedPersistAction UUID-      -- ^ Confirms a persist action has been handled successfully.------------------------------------------------------------------------------------- | Retrieves a running 'Subscription'.-querySubscription :: UUID -> Model -> Maybe Running-querySubscription u (Model k) = k $ Query $ QuerySub u------------------------------------------------------------------------------------- | Retrieves an ongoing persistent action.-queryPersistentAction :: UUID -> Model -> Maybe PendingAction-queryPersistentAction u (Model k) = k $ Query $ QueryAction u------------------------------------------------------------------------------------- | Registers a regular 'Subscription' request.-connectReg :: Text -> Bool -> UUID -> Model -> Model-connectReg n t u (Model k) = k $ Execute $ Connect u (ConnectReg n t)------------------------------------------------------------------------------------- | Registers a persistent 'Subscription' request.-connectPersist :: Text -> Text -> Int32 -> UUID -> Model -> Model-connectPersist g n b u (Model k) =-    k $ Execute $ Connect u (ConnectPersist g n b)------------------------------------------------------------------------------------- | Registers a persistent action.-persistAction :: Text -> Text -> UUID -> PersistAction -> Model -> Model-persistAction g n u a (Model k) = k $ Execute $ PersistAction g n u a------------------------------------------------------------------------------------- | Confirms a subscription.-confirmedSubscription :: UUID -> Meta -> Model -> Model-confirmedSubscription u m (Model k) =-    k $ Execute $ Confirmed $ ConfirmedConnection u m------------------------------------------------------------------------------------- | Confirms a persistent action. It doesn't assume if the action went well.-confirmedAction :: UUID -> Model -> Model-confirmedAction u (Model k) = k $ Execute $ Confirmed $ ConfirmedPersistAction u------------------------------------------------------------------------------------- | Remove a 'Subscription'.-unsubscribed :: Running -> Model -> Model-unsubscribed r (Model k) = k $ Execute $ Unsubscribed $ runningUUID r------------------------------------------------------------------------------------- | 'Subscription' model internal state.-data State =-    State-    { _stPending :: !(Register Pending)-      -- ^ Holds all pending 'Subscription's-    , _stRunning :: !(Register Running)-      -- ^ Holds all 'Subscription's that are currently running.-    , _stAction  :: !(Register PendingAction)-      -- ^ Holds all pending persistent actions.-    }-----------------------------------------------------------------------------------emptyState :: State-emptyState = State mempty mempty mempty------------------------------------------------------------------------------------- | Subscription operations state machine. Keeps every information related to---   subscription updated.-newtype Model = Model (forall a. Request a -> a)------------------------------------------------------------------------------------- | Creates a new 'Subscription' model.-newModel :: Model-newModel = Model $ modelHandle emptyState------------------------------------------------------------------------------------- | Main model handler.-modelHandle :: State -> Request a -> a-modelHandle s (Execute e) =-    case e of-        Connect u c ->-            case c of-                ConnectReg n tos ->-                    let p      = PendingReg n tos-                        nxt_ps = insertMap u p $ _stPending s-                        nxt_s  = s { _stPending = nxt_ps } in-                    Model $ modelHandle nxt_s-                ConnectPersist g n b ->-                    let p      = PendingPersist g n b-                        nxt_ps = insertMap u p $ _stPending s-                        nxt_s  = s { _stPending = nxt_ps } in-                    Model $ modelHandle nxt_s-        Confirmed c ->-            case c of-                ConfirmedConnection u tpe ->-                    case tpe of-                        RegularMeta lc le ->-                            case lookup u $ _stPending s of-                              Just (PendingReg n tos) ->-                                  let r      = RunningReg u n tos lc le-                                      nxt_rs = insertMap u r $ _stRunning s-                                      nxt_s  = s { _stRunning = nxt_rs } in-                                  Model $ modelHandle nxt_s-                              _ -> Model $ modelHandle s-                        PersistMeta sb lc le ->-                            case lookup u $ _stPending s of-                                Just (PendingPersist g n b) ->-                                    let r      = RunningPersist u g n b sb lc le-                                        nxt_rs = insertMap u r $ _stRunning s-                                        nxt_s  = s { _stRunning = nxt_rs } in-                                    Model $ modelHandle nxt_s-                                _ -> Model $ modelHandle s-                ConfirmedPersistAction u ->-                    case lookup u $ _stAction s of-                        Just (PendingAction{}) ->-                            let nxt_as = deleteMap u $ _stAction s-                                nxt_s  = s { _stAction = nxt_as } in-                            Model $ modelHandle nxt_s-                        _ -> Model $ modelHandle s-        Unsubscribed u ->-            let nxt_ps = deleteMap u $ _stRunning s-                nxt_s  = s { _stRunning = nxt_ps } in-            Model $ modelHandle nxt_s-        PersistAction g n u t ->-            let a      = PendingAction g n t-                nxt_as = insertMap u a $ _stAction s-                nxt_s  = s { _stAction = nxt_as } in-            Model $ modelHandle nxt_s-modelHandle s (Query q) =-    case q of-        QuerySub u    -> lookup u $ _stRunning s-        QueryAction u -> lookup u $ _stAction s

@@ -1,133 +0,0 @@-{-# LANGUAGE RecordWildCards #-}------------------------------------------------------------------------------------ |--- Module : Database.EventStore.Internal.Manager.Subscription.Packages--- Copyright : (C) 2015 Yorick Laupa--- License : (see the file LICENSE)------ Maintainer : Yorick Laupa <yo.eight@gmail.com>--- Stability : provisional--- Portability : non-portable-------------------------------------------------------------------------------------module Database.EventStore.Internal.Manager.Subscription.Packages where-----------------------------------------------------------------------------------import Data.Int-----------------------------------------------------------------------------------import ClassyPrelude-import Data.ProtocolBuffers-import Data.Serialize-import Data.UUID-----------------------------------------------------------------------------------import Database.EventStore.Internal.Manager.Subscription.Message-import Database.EventStore.Internal.Manager.Subscription.Model-import Database.EventStore.Internal.Types------------------------------------------------------------------------------------- | Creates a regular subscription connection 'Package'.-createConnectRegularPackage :: Settings -> UUID -> Text -> Bool -> Package-createConnectRegularPackage Settings{..} uuid stream tos =-    Package-    { packageCmd         = 0xC0-    , packageCorrelation = uuid-    , packageData        = runPut $ encodeMessage msg-    , packageCred        = s_credentials-    }-  where-    msg = subscribeToStream stream tos------------------------------------------------------------------------------------- | Creates a persistent subscription connection 'Package'.-createConnectPersistPackage :: Settings-                            -> UUID-                            -> Text-                            -> Text-                            -> Int32-                            -> Package-createConnectPersistPackage Settings{..} uuid grp stream bufSize =-    Package-    { packageCmd         = 0xC5-    , packageCorrelation = uuid-    , packageData        = runPut $ encodeMessage msg-    , packageCred        = s_credentials-    }-  where-    msg = _connectToPersistentSubscription grp stream bufSize------------------------------------------------------------------------------------- | Creates a persistent subscription 'Package'.-createPersistActionPackage :: Settings-                           -> UUID-                           -> Text-                           -> Text-                           -> PersistAction-                           -> Package-createPersistActionPackage Settings{..} u grp strm tpe =-    Package-    { packageCmd         = cmd-    , packageCorrelation = u-    , packageData        = runPut msg-    , packageCred        = s_credentials-    }-  where-    msg =-        case tpe of-            PersistCreate sett ->-                encodeMessage $ _createPersistentSubscription grp strm sett-            PersistUpdate sett ->-                encodeMessage $ _updatePersistentSubscription grp strm sett-            PersistDelete ->-                encodeMessage $ _deletePersistentSubscription grp strm-    cmd =-        case tpe of-            PersistCreate _  -> 0xC8-            PersistUpdate _  -> 0xCE-            PersistDelete    -> 0xCA------------------------------------------------------------------------------------- | Creates Ack 'Package'.-createAckPackage :: Settings -> UUID -> Text -> [UUID] -> Package-createAckPackage Settings{..} corr sid eids =-    Package-    { packageCmd         = 0xCC-    , packageCorrelation = corr-    , packageData        = runPut $ encodeMessage msg-    , packageCred        = s_credentials-    }-  where-    bytes = fmap (toStrict . toByteString) eids-    msg   = persistentSubscriptionAckEvents sid bytes------------------------------------------------------------------------------------- | Create Nak 'Package'.-createNakPackage :: Settings-                 -> UUID-                 -> Text-                 -> NakAction-                 -> Maybe Text-                 -> [UUID]-                 -> Package-createNakPackage Settings{..} corr sid act txt eids =-    Package-    { packageCmd         = 0xCD-    , packageCorrelation = corr-    , packageData        = runPut $ encodeMessage msg-    , packageCred        = s_credentials-    }-  where-    bytes = fmap (toStrict . toByteString) eids-    msg   = persistentSubscriptionNakEvents sid bytes txt act------------------------------------------------------------------------------------- | Create an unsubscribe 'Package'.-createUnsubscribePackage :: Settings -> UUID -> Package-createUnsubscribePackage Settings{..} uuid =-    Package-    { packageCmd         = 0xC3-    , packageCorrelation = uuid-    , packageData        = runPut $ encodeMessage UnsubscribeFromStream-    , packageCred        = s_credentials-    }

@@ -1,40 +0,0 @@------------------------------------------------------------------------------------ |--- Module : Database.EventStore.Internal.Manager.Subscription.Types--- Copyright : (C) 2016 Yorick Laupa--- License : (see the file LICENSE)------ Maintainer : Yorick Laupa <yo.eight@gmail.com>--- Stability : provisional--- Portability : non-portable-------------------------------------------------------------------------------------module Database.EventStore.Internal.Manager.Subscription.Types where-----------------------------------------------------------------------------------import ClassyPrelude-----------------------------------------------------------------------------------import Database.EventStore.Internal.Types------------------------------------------------------------------------------------- | Indicates why a subscription has been dropped.-data SubDropReason-    = SubUnsubscribed-      -- ^ Subscription connection has been closed by the user.-    | SubAccessDenied-      -- ^ The current user is not allowed to operate on the supplied stream.-    | SubNotFound-      -- ^ Given stream name doesn't exist.-    | SubPersistDeleted-      -- ^ Given stream is deleted.-    | SubAborted-      -- ^ Occurs when the user shutdown the connection from the server or if-      -- the connection to the server is no longer possible.-    | SubNotAuthenticated (Maybe Text)-    | SubServerError (Maybe Text)-      -- ^ Unexpected error from the server.-    | SubNotHandled !NotHandledReason !(Maybe MasterInfo)-    | SubClientError !Text-    | SubSubscriberMaxCountReached-    deriving (Show, Eq)

@@ -1,8 +1,10 @@-{-# LANGUAGE DataKinds          #-}-{-# LANGUAGE DeriveDataTypeable #-}-{-# LANGUAGE GADTs              #-}-{-# LANGUAGE KindSignatures     #-}-{-# LANGUAGE Rank2Types         #-}+{-# LANGUAGE DataKinds                  #-}+{-# LANGUAGE DeriveDataTypeable         #-}+{-# LANGUAGE GADTs                      #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE KindSignatures             #-}+{-# LANGUAGE MultiParamTypeClasses      #-}+{-# LANGUAGE Rank2Types                 #-} -------------------------------------------------------------------------------- -- | -- Module : Database.EventStore.Internal.Operation@@ -14,15 +16,43 @@ -- Portability : non-portable -- ---------------------------------------------------------------------------------module Database.EventStore.Internal.Operation where+module Database.EventStore.Internal.Operation+  ( OpResult(..)+  , OperationError(..)+  , Operation+  , Need(..)+  , Code+  , Execution(..)+  , Expect(..)+  , freshId+  , failure+  , retry+  , send+  , request+  , waitFor+  , waitForOr+  , wrongVersion+  , streamDeleted+  , invalidTransaction+  , accessDenied+  , protobufDecodingError+  , serverError+  , invalidServerResponse+  , module Data.Machine+  ) where  ---------------------------------------------------------------------------------import ClassyPrelude+import Prelude (String)++--------------------------------------------------------------------------------+import Data.Machine import Data.ProtocolBuffers+import Data.Serialize import Data.UUID  -------------------------------------------------------------------------------- import Database.EventStore.Internal.Command+import Database.EventStore.Internal.Prelude import Database.EventStore.Internal.Stream import Database.EventStore.Internal.Types @@ -50,6 +80,7 @@     | ProtobufDecodingError String     | ServerError (Maybe Text)                  -- ^ Reason     | InvalidOperation Text+    | StreamNotFound Text     | NotAuthenticatedOp       -- ^ Invalid operation state. If happens, it's a driver bug.     | Aborted@@ -61,125 +92,157 @@ instance Exception OperationError  ----------------------------------------------------------------------------------- | Main operation state machine instruction.-data SM o a-    = Return a-      -- ^ Lifts a pure value into the intruction tree. Also marks the end of-      --   an instruction tree.-    | Yield o (SM o a)-      -- ^ Emits an operation return value.-    | FreshId (UUID -> SM o a)-      -- ^ Asks for an unused 'UUID'.-    | forall rq rp. (Encode rq, Decode rp) =>-      SendPkg Command Command rq (rp -> SM o a)-      -- ^ Send a request message given a command and an expected command.-      --   response. It also carries a callback to call when response comes in.-    | Failure (Maybe OperationError)-      -- ^ Ends the instruction interpretation. If holds Nothing, the-      --   interpretation should resume from the beginning. Otherwise it ends-      --   by indicating what went wrong.+data Execution a+  = Proceed a+  | Retry+  | Failed !OperationError  ---------------------------------------------------------------------------------instance Functor (SM o) where-    fmap f (Return a)          = Return (f a)-    fmap f (Yield o n)         = Yield o (fmap f n)-    fmap f (FreshId k)         = FreshId (fmap f . k)-    fmap f (SendPkg ci co p k) = SendPkg ci co p (fmap f . k)-    fmap _ (Failure e)         = Failure e+instance Functor Execution where+  fmap f (Proceed a) = Proceed (f a)+  fmap _ Retry       = Retry+  fmap _ (Failed e)  = Failed e  ---------------------------------------------------------------------------------instance Applicative (SM o) where-    pure = return-    (<*>) = ap+instance Applicative Execution where+  pure = return+  (<*>) = ap  ---------------------------------------------------------------------------------instance Monad (SM o) where-    return = Return+instance Monad Execution where+  return = Proceed -    Return a          >>= f = f a-    Yield o n         >>= f = Yield o (n >>= f)-    FreshId k         >>= f = FreshId ((f =<<) . k)-    SendPkg ci co p k >>= f = SendPkg ci co p ((f =<<) . k)-    Failure e         >>= _ = Failure e+  Proceed a >>= f = f a+  Retry     >>= _ = Retry+  Failed e  >>= _ = Failed e  --------------------------------------------------------------------------------+type Operation output = MachineT Execution Need output++--------------------------------------------------------------------------------+data Need a where+  NeedUUID   :: Need UUID+  NeedRemote :: Command -> ByteString -> Need Package+  WaitRemote :: UUID -> Need (Maybe Package)++--------------------------------------------------------------------------------+-- | Instruction that composed an 'Operation'.+type Code o a = PlanT Need o Execution a++-------------------------------------------------------------------------------- -- | Asks for a unused 'UUID'.-freshId :: SM o UUID-freshId = FreshId Return+freshId :: Code o UUID+freshId = awaits NeedUUID  -------------------------------------------------------------------------------- -- | Raises an 'OperationError'.-failure :: OperationError -> SM o a-failure e = Failure $ Just e+failure :: OperationError -> Code o a+failure = lift . Failed  -------------------------------------------------------------------------------- -- | Asks to resume the interpretation from the beginning.-retry :: SM o a-retry = Failure Nothing+retry :: Code o a+retry = lift Retry  ----------------------------------------------------------------------------------- | Sends a request to the server given a command request and response. It---   returns the expected deserialized message.-send :: (Encode rq, Decode rp) => Command -> Command -> rq -> SM o rp-send ci co rq = SendPkg ci co rq Return+-- | Like 'request' except it discards the correlation id of the network+--   exchange.+send :: (Encode req, Decode resp)+     => Command+     -> Command+     -> req+     -> Code o resp+send reqCmd expCmd req = do+  let payload = runPut $ encodeMessage req+  pkg <- awaits $ NeedRemote reqCmd payload+  let gotCmd = packageCmd pkg +  when (gotCmd /= expCmd)+    (invalidServerResponse expCmd gotCmd)++  case runGet decodeMessage (packageData pkg) of+    Left e     -> protobufDecodingError e+    Right resp -> return resp+ ----------------------------------------------------------------------------------- | Emits operation return value.-yield :: o -> SM o ()-yield o = Yield o (Return ())+data Expect o where+  Expect :: Decode resp => Command -> (UUID -> resp -> Code o ()) -> Expect o  ----------------------------------------------------------------------------------- | Replaces every emitted value, via 'yield' function by calling the given---   callback.-foreach :: SM a x -> (a -> SM b x) -> SM b x-foreach start k = go start-  where-    go (Return x)           = Return x-    go (Yield a n)          = k a >> go n-    go (FreshId ki)         = FreshId (go . ki)-    go (SendPkg ci co p kp) = SendPkg ci co p (go . kp)-    go (Failure e)          = Failure e+-- | Runs the first expection that matches.+runFirstMatch :: Package -> [Expect o] -> Code o ()+runFirstMatch _ [] = invalidOperation "No expection was fulfilled"+runFirstMatch pkg (Expect cmd k:rest)+  | packageCmd pkg /= cmd = runFirstMatch pkg rest+  | otherwise =+    case runGet decodeMessage (packageData pkg) of+      Left e     -> protobufDecodingError e+      Right resp -> k (packageCorrelation pkg) resp  ----------------------------------------------------------------------------------- | Maps every emitted value, via 'yield', using given function.-mapOp :: (a -> b) -> SM a () -> SM b ()-mapOp k sm = foreach sm (yield . k)+-- | Sends a message to remote server. It returns the expected deserialized+--   message along with the correlation id of the network exchange.+request :: Encode req+        => Command+        -> req+        -> [Expect o]+        -> Code o ()+request reqCmd rq exps = do+  let payload = runPut $ encodeMessage rq+  pkg <- awaits $ NeedRemote reqCmd payload+  runFirstMatch pkg exps  ----------------------------------------------------------------------------------- | An operation is just a 'SM' tree.-type Operation a = SM a ()+-- | Like 'waitForOr' but will 'stop' if the connection reset.+waitFor :: UUID -> [Expect o] -> Code o ()+waitFor pid exps = waitForOr pid stop exps  --------------------------------------------------------------------------------+-- | @waitForElse uuid alternative expects@ Waits for a message from the server+--   at the given /uuid/. If the connection has been reset in the meantime, it+--   will use /alternative/.+waitForOr :: UUID -> (Code o ()) -> [Expect o] -> Code o ()+waitForOr pid alt exps =+  awaits (WaitRemote pid) >>= \case+    Nothing  -> alt+    Just pkg ->+      runFirstMatch pkg exps++-------------------------------------------------------------------------------- -- | Raises 'WrongExpectedVersion' exception.-wrongVersion :: Text -> ExpectedVersion -> SM o a+wrongVersion :: Text -> ExpectedVersion -> Code o a wrongVersion stream ver = failure (WrongExpectedVersion stream ver)  -------------------------------------------------------------------------------- -- | Raises 'StreamDeleted' exception.-streamDeleted :: Text -> SM o a+streamDeleted :: Text -> Code o a streamDeleted stream = failure (StreamDeleted stream)  -------------------------------------------------------------------------------- -- | Raises 'InvalidTransaction' exception.-invalidTransaction :: SM o a+invalidTransaction :: Code o a invalidTransaction = failure InvalidTransaction  -------------------------------------------------------------------------------- -- | Raises 'AccessDenied' exception.-accessDenied :: StreamName -> SM o a+accessDenied :: StreamName -> Code oconcat a accessDenied = failure . AccessDenied  -------------------------------------------------------------------------------- -- | Raises 'ProtobufDecodingError' exception.-protobufDecodingError :: String -> SM o a+protobufDecodingError :: String -> Code o a protobufDecodingError = failure . ProtobufDecodingError  -------------------------------------------------------------------------------- -- | Raises 'ServerError' exception.-serverError :: Maybe Text -> SM o a+serverError :: Maybe Text -> Code o a serverError = failure . ServerError  -------------------------------------------------------------------------------- -- | Raises 'InvalidServerResponse' exception.-invalidServerResponse :: Command -> Command -> SM o a+invalidServerResponse :: Command -> Command -> Code o a invalidServerResponse expe got = failure $ InvalidServerResponse expe got++--------------------------------------------------------------------------------+invalidOperation :: Text -> Code o a+invalidOperation = failure . InvalidOperation

@@ -17,7 +17,6 @@     , CatchupOpResult(..)     , Checkpoint(..)     , catchup-    , catchupStreamName     ) where  --------------------------------------------------------------------------------@@ -25,14 +24,15 @@ import Data.Maybe  ---------------------------------------------------------------------------------import ClassyPrelude---------------------------------------------------------------------------------- import Database.EventStore.Internal.Operation import Database.EventStore.Internal.Operation.Read.Common import Database.EventStore.Internal.Operation.ReadAllEvents import Database.EventStore.Internal.Operation.ReadStreamEvents+import Database.EventStore.Internal.Operation.Volatile+import Database.EventStore.Internal.Prelude+import Database.EventStore.Internal.Settings import Database.EventStore.Internal.Stream+import Database.EventStore.Internal.Subscription.Types import Database.EventStore.Internal.Types  --------------------------------------------------------------------------------@@ -47,22 +47,58 @@  -------------------------------------------------------------------------------- streamNotFound :: Text -> OperationError-streamNotFound stream =-  InvalidOperation $ "Catchup. inexistant stream [" <> stream <> "]"+streamNotFound stream = StreamNotFound stream  -------------------------------------------------------------------------------- -- | Catchup operation state. data CatchupState     = RegularCatchup Text Int32       -- ^ Indicates the stream name and the next event number to start from.-    | AllCatchup Int64 Int64+    | AllCatchup Position       -- ^ Indicates the commit and prepare position. Used when catching up from       --   the $all stream.+    deriving Show  --------------------------------------------------------------------------------+fetch :: Settings -> Int32 -> Bool -> CatchupState -> Code o SomeSlice+fetch setts batch tos state =+    case state of+        RegularCatchup stream n -> do+            outcome <- deconstruct $ fmap Left $+                           readStreamEvents setts Forward stream n batch tos+            fromReadResult stream outcome (return . toSlice)+        AllCatchup (Position com pre) ->+            deconstruct $ fmap (Left . toSlice) $+                readAllEvents setts com pre batch tos Forward++--------------------------------------------------------------------------------+updateState :: CatchupState -> Location -> CatchupState+updateState (RegularCatchup stream _) (StreamEventNumber n) =+    RegularCatchup stream n+updateState (AllCatchup _) (StreamPosition p) = AllCatchup p+updateState x y = error $ "Unexpected input updateState: " <> show (x,y)++--------------------------------------------------------------------------------+sourceStream :: Settings+             -> Int32+             -> Bool+             -> CatchupState+             -> Operation SubAction+sourceStream setts batch tos start = unfoldPlan start go+  where+    go state = do+        s <- fetch setts batch tos state+        traverse_ (yield . Submit) (sliceEvents s)++        when (sliceEOS s)+            stop++        return $ updateState state (sliceNext s)++-------------------------------------------------------------------------------- catchupStreamName :: CatchupState -> Text catchupStreamName (RegularCatchup stream _) = stream-catchupStreamName _ = "$all"+catchupStreamName _ = ""  -------------------------------------------------------------------------------- data CatchupOpResult =@@ -77,44 +113,18 @@         -> CatchupState         -> Bool         -> Maybe Int32-        -> Operation CatchupOpResult -- ([ResolvedEvent], Bool, Checkpoint)-catchup setts init_tpe tos bat_siz = go init_tpe+        -> Operation SubAction+catchup setts state tos batchSiz =+    sourceStream setts batch tos state <> volatile stream tos   where-    batch = fromMaybe defaultBatchSize bat_siz-    go tpe = do-        let action =-                case tpe of-                    RegularCatchup stream cur_evt ->-                        let op = readStreamEvents setts Forward stream cur_evt-                                 batch tos in-                        mapOp Left op-                    AllCatchup c_pos p_pos ->-                        let op = readAllEvents setts c_pos p_pos batch-                                 tos Forward in-                        mapOp Right op--        foreach action $ \res -> do-            (eos, evts, nchk, nxt_tpe) <- case res of-                Right as -> do-                    let Position nxt_c nxt_p = sliceNext as-                        tmp_tpe = AllCatchup nxt_c nxt_p-                        chk     = CheckpointPosition $ sliceNext as-                    return (sliceEOS as, sliceEvents as, chk, tmp_tpe)-                Left rr -> fromReadResult (catchupStreamName tpe) rr $ \as ->-                    let RegularCatchup s _ = tpe-                        nxt = sliceNext as-                        tmp_tpe = RegularCatchup s nxt-                        chk = CheckpointNumber nxt in-                    return (sliceEOS as, sliceEvents as, chk, tmp_tpe)--            yield (CatchupOpResult evts eos nchk)-            unless eos $ go nxt_tpe+    batch  = fromMaybe defaultBatchSize batchSiz+    stream = catchupStreamName state  -------------------------------------------------------------------------------- fromReadResult :: Text                -> ReadResult 'RegularStream a-               -> (a -> SM b x)-               -> SM b x+               -> (a -> Code o x)+               -> Code o x fromReadResult stream res k =     case res of         ReadNoStream        -> failure $ streamNotFound stream

@@ -22,12 +22,14 @@ import Data.Maybe  ---------------------------------------------------------------------------------import ClassyPrelude import Data.ProtocolBuffers  --------------------------------------------------------------------------------+import Database.EventStore.Internal.Command import Database.EventStore.Internal.Operation import Database.EventStore.Internal.Operation.DeleteStream.Message+import Database.EventStore.Internal.Prelude+import Database.EventStore.Internal.Settings import Database.EventStore.Internal.Stream import Database.EventStore.Internal.Types @@ -42,9 +44,9 @@              -> ExpectedVersion              -> Maybe Bool              -> Operation DeleteResult-deleteStream Settings{..} s v hard = do+deleteStream Settings{..} s v hard = construct $ do     let msg = newRequest s (expVersionInt32 v) s_requireMaster hard-    resp <- send 0x8A 0x8B msg+    resp <- send deleteStreamCmd deleteStreamCompletedCmd msg     let r            = getField $ _result resp         com_pos      = getField $ _commitPosition resp         prep_pos     = getField $ _preparePosition resp

@@ -17,11 +17,11 @@ import Data.Int  ---------------------------------------------------------------------------------import ClassyPrelude import Data.ProtocolBuffers  -------------------------------------------------------------------------------- import Database.EventStore.Internal.Operation+import Database.EventStore.Internal.Prelude  -------------------------------------------------------------------------------- -- | Delete stream request.

@@ -0,0 +1,79 @@+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Internal.Operation.Persist+-- Copyright : (C) 2017 Yorick Laupa+-- License : (see the file LICENSE)+--+-- Maintainer : Yorick Laupa <yo.eight@gmail.com>+-- Stability : provisional+-- Portability : non-portable+--+--------------------------------------------------------------------------------+module Database.EventStore.Internal.Operation.Persist (persist) where++--------------------------------------------------------------------------------+import Data.ProtocolBuffers+import Data.UUID++--------------------------------------------------------------------------------+import Database.EventStore.Internal.Command+import Database.EventStore.Internal.Operation+import Database.EventStore.Internal.Prelude+import Database.EventStore.Internal.Subscription.Message+import Database.EventStore.Internal.Subscription.Types+import Database.EventStore.Internal.Types++--------------------------------------------------------------------------------+persist :: Text -> Text -> Int32 -> Operation SubAction+persist grp stream bufSize = construct (issueRequest grp stream bufSize)++--------------------------------------------------------------------------------+issueRequest :: Text -> Text -> Int32 -> Code SubAction ()+issueRequest grp stream bufSize = do+  let req = _connectToPersistentSubscription grp stream bufSize+  request connectToPersistentSubscriptionCmd req+    [ Expect subscriptionDroppedCmd $ \_ d ->+        handleDropped d+    , Expect persistentSubscriptionConfirmationCmd $ \sid c -> do+        let lcp      = getField $ pscLastCommitPos c+            subSubId = getField $ pscId c+            len      = getField $ pscLastEvtNumber c+            details  =+              SubDetails+              { subId           = sid+              , subCommitPos    = lcp+              , subLastEventNum = len+              , subSubId        = Just subSubId+              }+        yield (Confirmed details)+        live sid+    ]++--------------------------------------------------------------------------------+eventAppeared :: PersistentSubscriptionStreamEventAppeared -> Code SubAction ()+eventAppeared e = do+  let evt = newResolvedEvent $ getField $ psseaEvt e+  yield (Submit evt)++--------------------------------------------------------------------------------+live :: UUID -> Code SubAction ()+live subscriptionId = loop+  where+    loop =+      waitForOr subscriptionId connectionReset+        [ Expect subscriptionDroppedCmd $ \_ d ->+            handleDropped d+        , Expect persistentSubscriptionStreamEventAppearedCmd $ \_ e -> do+            eventAppeared e+            loop+        ]++--------------------------------------------------------------------------------+connectionReset :: Code SubAction ()+connectionReset = yield ConnectionReset++--------------------------------------------------------------------------------+handleDropped :: SubscriptionDropped -> Code SubAction ()+handleDropped d = do+  let reason = fromMaybe D_Unsubscribed (getField $ dropReason d)+  yield (Dropped $ toSubDropReason reason)

@@ -0,0 +1,75 @@+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Internal.Operation.PersistOperations+-- Copyright : (C) 2017 Yorick Laupa+-- License : (see the file LICENSE)+--+-- Maintainer : Yorick Laupa <yo.eight@gmail.com>+-- Stability : provisional+-- Portability : non-portable+--+--------------------------------------------------------------------------------+module Database.EventStore.Internal.Operation.PersistOperations+  ( createPersist+  , updatePersist+  , deletePersist+  ) where++--------------------------------------------------------------------------------+import Data.ProtocolBuffers++--------------------------------------------------------------------------------+import Database.EventStore.Internal.Command+import Database.EventStore.Internal.Operation+import Database.EventStore.Internal.Prelude+import Database.EventStore.Internal.Subscription.Message+import Database.EventStore.Internal.Subscription.Types+import Database.EventStore.Internal.Types++--------------------------------------------------------------------------------+persistOperation :: Text+                 -> Text+                 -> PersistAction+                 -> Operation (Maybe PersistActionException)+persistOperation grp stream tpe = construct go+  where+    go =+      case tpe of+        PersistCreate ss -> do+          let req = _createPersistentSubscription grp stream ss+          resp <- send createPersistentSubscriptionCmd+                       createPersistentSubscriptionCompletedCmd req+          let result = createRException $ getField $ cpscResult resp+          yield result+        PersistUpdate ss -> do+          let req = _updatePersistentSubscription grp stream ss+          resp <- send updatePersistentSubscriptionCmd+                       updatePersistentSubscriptionCompletedCmd req+          let result = updateRException $ getField $ upscResult resp+          yield result+        PersistDelete -> do+          let req = _deletePersistentSubscription grp stream+          resp <- send deletePersistentSubscriptionCmd+                       deletePersistentSubscriptionCompletedCmd req+          let result = deleteRException $ getField $ dpscResult resp+          yield result++--------------------------------------------------------------------------------+createPersist :: Text+              -> Text+              -> PersistentSubscriptionSettings+              -> Operation (Maybe PersistActionException)+createPersist grp stream ss = persistOperation grp stream (PersistCreate ss)++--------------------------------------------------------------------------------+updatePersist :: Text+              -> Text+              -> PersistentSubscriptionSettings+              -> Operation (Maybe PersistActionException)+updatePersist grp stream ss = persistOperation grp stream (PersistUpdate ss)++--------------------------------------------------------------------------------+deletePersist :: Text+              -> Text+              -> Operation (Maybe PersistActionException)+deletePersist grp stream = persistOperation grp stream PersistDelete

@@ -16,17 +16,21 @@ module Database.EventStore.Internal.Operation.Read.Common where  ---------------------------------------------------------------------------------import Data.Foldable (foldMap)+import Control.Applicative+import Data.Foldable+import Data.Monoid+import Data.Traversable import Data.Int  ---------------------------------------------------------------------------------import ClassyPrelude----------------------------------------------------------------------------------+import Database.EventStore.Internal.Prelude import Database.EventStore.Internal.Stream import Database.EventStore.Internal.Types  --------------------------------------------------------------------------------+import Prelude++-------------------------------------------------------------------------------- -- | Enumeration detailing the possible outcomes of reading a stream. data ReadResult       :: StreamType -> * -> * where     ReadSuccess       :: a -> ReadResult t a@@ -93,6 +97,8 @@     -- ^ Gets the starting location of this slice.     sliceNext :: a -> Loc a     -- ^ Gets the next location of this slice.+    toSlice :: a -> SomeSlice+    -- ^ Returns a common view of a slice.  -------------------------------------------------------------------------------- -- | Regular stream slice.@@ -117,6 +123,15 @@     sliceFrom      = _ssFrom     sliceNext      = _ssNext +    toSlice s =+        SomeSlice+        { __events = sliceEvents s+        , __eos    = sliceEOS s+        , __dir    = sliceDirection s+        , __from   = StreamEventNumber $ sliceFrom s+        , __next   = StreamEventNumber $ sliceNext s+        }+ -------------------------------------------------------------------------------- -- | Represents a slice of the $all stream. data AllSlice =@@ -137,3 +152,39 @@     sliceEOS       = _saEOS     sliceFrom      = _saFrom     sliceNext      = _saNext++    toSlice s =+        SomeSlice+        { __events = sliceEvents s+        , __eos    = sliceEOS s+        , __dir    = sliceDirection s+        , __from   = StreamPosition $ sliceFrom s+        , __next   = StreamPosition $ sliceNext s+        }++--------------------------------------------------------------------------------+data Location+    = StreamEventNumber !Int32+    | StreamPosition !Position+    deriving Show++--------------------------------------------------------------------------------+data SomeSlice =+    SomeSlice+    { __events :: ![ResolvedEvent]+    , __eos    :: !Bool+    , __dir    :: !ReadDirection+    , __from   :: !Location+    , __next   :: !Location+    } deriving Show++--------------------------------------------------------------------------------+instance Slice SomeSlice where+    type Loc SomeSlice = Location++    sliceEvents    = __events+    sliceDirection = __dir+    sliceEOS       = __eos+    sliceFrom      = __from+    sliceNext      = __next+    toSlice        = id

@@ -18,13 +18,15 @@ import Data.Maybe  ---------------------------------------------------------------------------------import ClassyPrelude import Data.ProtocolBuffers  --------------------------------------------------------------------------------+import Database.EventStore.Internal.Command import Database.EventStore.Internal.Operation import Database.EventStore.Internal.Operation.Read.Common import Database.EventStore.Internal.Operation.ReadAllEvents.Message+import Database.EventStore.Internal.Prelude+import Database.EventStore.Internal.Settings import Database.EventStore.Internal.Stream import Database.EventStore.Internal.Types @@ -37,15 +39,15 @@               -> Bool               -> ReadDirection               -> Operation AllSlice-readAllEvents Settings{..} c_pos p_pos max_c tos dir = do+readAllEvents Settings{..} c_pos p_pos max_c tos dir = construct $ do     let msg = newRequest c_pos p_pos max_c tos s_requireMaster         cmd = case dir of-            Forward  -> 0xB6-            Backward -> 0xB8+            Forward  -> readAllEventsForwardCmd+            Backward -> readAllEventsBackwardCmd          resp_cmd = case dir of-            Forward  -> 0xB7-            Backward -> 0xB9+            Forward  -> readAllEventsForwardCompletedCmd+            Backward -> readAllEventsBackwardCompletedCmd     resp <- send cmd resp_cmd msg     let r      = getField $ _Result resp         err    = getField $ _Error resp

@@ -17,10 +17,10 @@ import Data.Int  ---------------------------------------------------------------------------------import ClassyPrelude import Data.ProtocolBuffers  --------------------------------------------------------------------------------+import Database.EventStore.Internal.Prelude import Database.EventStore.Internal.Types  --------------------------------------------------------------------------------

@@ -20,13 +20,15 @@ import Data.Int  ---------------------------------------------------------------------------------import ClassyPrelude import Data.ProtocolBuffers  --------------------------------------------------------------------------------+import Database.EventStore.Internal.Command import Database.EventStore.Internal.Operation import Database.EventStore.Internal.Operation.ReadEvent.Message import Database.EventStore.Internal.Operation.Read.Common+import Database.EventStore.Internal.Prelude+import Database.EventStore.Internal.Settings import Database.EventStore.Internal.Stream import Database.EventStore.Internal.Types @@ -50,9 +52,9 @@           -> Int32           -> Bool           -> Operation (ReadResult 'RegularStream ReadEvent)-readEvent Settings{..} s evtn tos = do+readEvent Settings{..} s evtn tos = construct $ do     let msg = newRequest s evtn tos s_requireMaster-    resp <- send 0xB0 0xB1 msg+    resp <- send readEventCmd readEventCompletedCmd msg     let r         = getField $ _result resp         evt       = newResolvedEvent $ getField $ _indexedEvent resp         err       = getField $ _error resp

@@ -1,6 +1,11 @@-{-# LANGUAGE DeriveGeneric         #-}-{-# LANGUAGE DataKinds             #-}+{-# LANGUAGE CPP           #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DataKinds     #-}+#if __GLASGOW_HASKELL__ < 800 {-# OPTIONS_GHC -fcontext-stack=26 #-}+#else+{-# OPTIONS_GHC -freduction-depth=26 #-}+#endif -------------------------------------------------------------------------------- -- | -- Module : Database.EventStore.Internal.Operation.ReadEvent.Message@@ -18,10 +23,10 @@ import Data.Int  ---------------------------------------------------------------------------------import ClassyPrelude import Data.ProtocolBuffers  --------------------------------------------------------------------------------+import Database.EventStore.Internal.Prelude import Database.EventStore.Internal.Types  --------------------------------------------------------------------------------@@ -72,3 +77,4 @@  -------------------------------------------------------------------------------- instance Decode Response+instance Encode Response

@@ -18,13 +18,15 @@ import Data.Int  ---------------------------------------------------------------------------------import ClassyPrelude import Data.ProtocolBuffers  --------------------------------------------------------------------------------+import Database.EventStore.Internal.Command import Database.EventStore.Internal.Operation import Database.EventStore.Internal.Operation.Read.Common import Database.EventStore.Internal.Operation.ReadStreamEvents.Message+import Database.EventStore.Internal.Prelude+import Database.EventStore.Internal.Settings import Database.EventStore.Internal.Stream import Database.EventStore.Internal.Types @@ -37,15 +39,15 @@                  -> Int32                  -> Bool                  -> Operation (ReadResult 'RegularStream StreamSlice)-readStreamEvents Settings{..} dir s st cnt tos = do+readStreamEvents Settings{..} dir s st cnt tos = construct $ do     let req_cmd =             case dir of-                Forward  -> 0xB2-                Backward -> 0xB4+                Forward  -> readStreamEventsForwardCmd+                Backward -> readStreamEventsBackwardCmd         resp_cmd =             case dir of-                Forward  -> 0xB3-                Backward -> 0xB5+                Forward  -> readStreamEventsForwardCompletedCmd+                Backward -> readStreamEventsBackwardCompletedCmd          msg = newRequest s st cnt tos s_requireMaster     resp <- send req_cmd resp_cmd msg

@@ -17,10 +17,10 @@ import Data.Int  ---------------------------------------------------------------------------------import ClassyPrelude import Data.ProtocolBuffers  --------------------------------------------------------------------------------+import Database.EventStore.Internal.Prelude import Database.EventStore.Internal.Types  --------------------------------------------------------------------------------

@@ -23,7 +23,6 @@ import Data.Int  ---------------------------------------------------------------------------------import ClassyPrelude import Data.Aeson (decode)  --------------------------------------------------------------------------------@@ -32,6 +31,8 @@ import Database.EventStore.Internal.Operation.ReadEvent import Database.EventStore.Internal.Operation.Write.Common import Database.EventStore.Internal.Operation.WriteEvents+import Database.EventStore.Internal.Prelude+import Database.EventStore.Internal.Settings import Database.EventStore.Internal.Stream import Database.EventStore.Internal.Types @@ -42,13 +43,14 @@ -------------------------------------------------------------------------------- -- | Read stream metadata operation. readMetaStream :: Settings -> Text -> Operation StreamMetadataResult-readMetaStream setts s =-    foreach (readEvent setts (metaStream s) (-1) False) $ \tmp -> do-        onReadResult tmp $ \n e_num evt -> do-            let bytes = recordedEventData $ resolvedEventOriginal evt-            case decode $ fromStrict bytes of-                Just pv -> yield $ StreamMetadataResult n e_num pv-                Nothing -> failure invalidFormat+readMetaStream setts s = construct $ do+    let op = readEvent setts (metaStream s) (-1) False+    tmp <- deconstruct (fmap Left op)+    onReadResult tmp $ \n e_num evt -> do+        let bytes = recordedEventData $ resolvedEventOriginal evt+        case decode $ fromStrict bytes of+            Just pv -> yield $ StreamMetadataResult n e_num pv+            Nothing -> failure invalidFormat  -------------------------------------------------------------------------------- -- | Set stream metadata operation.@@ -60,9 +62,8 @@ setMetaStream setts s v meta =     let stream = metaStream s         json   = streamMetadataJSON meta-        evt    = createEvent StreamMetadataType Nothing (withJson json)-        inner  = writeEvents setts stream v [evt] in-    foreach inner yield+        evt    = createEvent StreamMetadataType Nothing (withJson json) in+     writeEvents setts stream v [evt]  -------------------------------------------------------------------------------- invalidFormat :: OperationError@@ -74,8 +75,8 @@  -------------------------------------------------------------------------------- onReadResult :: ReadResult 'RegularStream ReadEvent-             -> (Text -> Int32 -> ResolvedEvent -> SM a b)-             -> SM a b+             -> (Text -> Int32 -> ResolvedEvent -> Code o a)+             -> Code o a onReadResult (ReadSuccess r) k =     case r of       ReadEvent s n e -> k s n e