eventstore 0.13.1.7 → 0.14.0.0

raw patch · 25 files changed

+1609/−783 lines, 25 filesdep +mtlPVP ok

version bump matches the API change (PVP)

Dependencies added: mtl

API changes (from Hackage documentation)

Files

• .gitignore +1/−0
• CHANGELOG.markdown +7/−0
• Database/EventStore.hs +61/−327
• Database/EventStore/Internal/Command.hs +3/−9
• Database/EventStore/Internal/Connection.hs +41/−9
• Database/EventStore/Internal/Discovery.hs +3/−12
• Database/EventStore/Internal/EndPoint.hs +34/−0
• Database/EventStore/Internal/Execution/Production.hs +45/−10
• Database/EventStore/Internal/Manager/Operation/Model.hs +37/−9
• Database/EventStore/Internal/Manager/Subscription.hs +0/−220
• Database/EventStore/Internal/Manager/Subscription/Command.hs +103/−0
• Database/EventStore/Internal/Manager/Subscription/Driver.hs +243/−170
• Database/EventStore/Internal/Manager/Subscription/Message.hs +1/−0
• Database/EventStore/Internal/Manager/Subscription/Types.hs +40/−0
• Database/EventStore/Internal/Operation.hs +1/−0
• Database/EventStore/Internal/Operation/Catchup.hs +23/−8
• Database/EventStore/Internal/Operation/StreamMetadata.hs +1/−1
• Database/EventStore/Internal/Operation/Write/Common.hs +1/−1
• Database/EventStore/Internal/Processor.hs +7/−1
• Database/EventStore/Internal/Subscription.hs +794/−0
• Database/EventStore/Internal/Types.hs +125/−2
• Database/EventStore/Internal/Utils.hs +27/−0
• README.md +1/−0
• eventstore.cabal +9/−4
• tests/Tests.hs +1/−0

@@ -11,3 +11,4 @@ cabal.config *~ *.swp+.stack-work/

@@ -1,3 +1,10 @@+0.14.0.0+--------+* Fix deadlock issues in subscription code.+* Improve cluster connection fiability.+* Internal refactoring.+* Typeful `EventType`+ 0.13.1.7 -------- * Report stream name when facing a 'Stream not found' error from the server on catchup subscription.

@@ -45,6 +45,7 @@       -- * Event     , Event     , EventData+    , EventType(..)     , createEvent     , withJson     , withJsonAndMetadata@@ -108,11 +109,13 @@     , SubscriptionClosed(..)     , SubscriptionId     , Subscription-    , S.Running(..)-    , S.SubDropReason(..)+    , SubDropReason(..)     , waitConfirmation+    , unsubscribeConfirmed+    , unsubscribeConfirmedSTM+    , waitUnsubscribeConfirmed       -- * Volatile Subscription-    , S.Regular+    , Regular     , subscribe     , subscribeToAll     , getSubId@@ -125,17 +128,18 @@     , getSubLastCommitPos     , getSubLastEventNumber       -- * Catch-up Subscription-    , S.Catchup+    , Catchup     , subscribeFrom     , subscribeToAllFrom     , waitTillCatchup     , hasCaughtUp+    , hasCaughtUpSTM      -- * Persistent Subscription-    , S.Persistent+    , Persistent     , PersistentSubscriptionSettings(..)     , SystemConsumerStrategy(..)     , NakAction(..)-    , S.PersistActionException(..)+    , PersistActionException(..)     , acknowledge     , acknowledgeEvents     , failed@@ -194,14 +198,14 @@ -------------------------------------------------------------------------------- import ClassyPrelude hiding (Builder, group) import Data.List.NonEmpty(NonEmpty(..), nonEmpty)-import Data.UUID import Network.Connection (TLSSettings)  -------------------------------------------------------------------------------- import           Database.EventStore.Internal.Command import           Database.EventStore.Internal.Connection import           Database.EventStore.Internal.Discovery-import qualified Database.EventStore.Internal.Manager.Subscription as S+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.Operation (OperationError(..)) import qualified Database.EventStore.Internal.Operations as Op@@ -279,213 +283,6 @@     sendEvents mgr evt_stream exp_ver [evt]  ----------------------------------------------------------------------------------- | Represents a subscription id.-newtype SubscriptionId = SubId UUID deriving (Eq, Ord, Show)------------------------------------------------------------------------------------- | Determines whether or not any link events encontered in the stream will be---   resolved.-getSubResolveLinkTos :: Subscription S.Regular -> Bool-getSubResolveLinkTos = S._subTos . _subInner------------------------------------------------------------------------------------- | Non blocking version of `waitTillCatchup`.-hasCaughtUp :: Subscription S.Catchup -> IO Bool-hasCaughtUp sub = atomically $ _hasCaughtUp sub------------------------------------------------------------------------------------- | Waits until 'CatchupSubscription' subscription catch-up its stream.-waitTillCatchup :: Subscription S.Catchup -> IO ()-waitTillCatchup sub = atomically $ do-    caughtUp <- _hasCaughtUp sub-    when (not caughtUp) retrySTM-----------------------------------------------------------------------------------_hasCaughtUp :: Subscription S.Catchup -> STM Bool-_hasCaughtUp Subscription{..} = do-    res <- _subState-    case res of-        SubState sm _ -> return $ S.hasCaughtUp sm-        SubException e -> throwSTM e------------------------------------------------------------------------------------- | Tracks a 'Subcription' lifecycle. It holds a 'Subscription' state machine---   and `SubDropReason` if any.-data SubState a-    = SubState (S.Subscription a) (Maybe S.SubDropReason)-    | forall e. Exception e => SubException e-      -- ^ Hack used to cover a special exception that can arise from-      --   a catchup subscription for instance. One example is when asking a-      --   catchup subscription on stream that doesn't exist yet.------------------------------------------------------------------------------------- | It's possible to subscribe to a stream and be notified when new events are---   written to that stream. There are three types of subscription which are---   available, all of which can be useful in different situations.------     * 'S.Regular'------     * 'S.Catchup'------     * 'S.Persistent'-data Subscription a =-    Subscription-    { _subState    :: STM (SubState a)-    , _subSetState :: SubState a -> STM ()-    , _subRun      :: TMVar S.Running-    , _subStream   :: Text-    , _subProd     :: Production-    , _subInner    :: a-    }------------------------------------------------------------------------------------- | Gets the ID of the subscription.-getSubId :: Subscription a -> IO SubscriptionId-getSubId Subscription{..} = atomically $ do-    run <- readTMVar _subRun-    return $ SubId $ S.runningUUID run------------------------------------------------------------------------------------- | Gets the subscription stream name.-getSubStream :: Subscription a -> Text-getSubStream = _subStream------------------------------------------------------------------------------------- | Asynchronously unsubscribe from the the stream.-unsubscribe :: Subscription a -> IO ()-unsubscribe Subscription{..} = do-    run <- atomically $ readTMVar _subRun-    pushUnsubscribe _subProd run------------------------------------------------------------------------------------- | If the subscription is on the $all stream.-isSubscribedToAll :: Subscription a -> Bool-isSubscribedToAll = (== "") . getSubStream------------------------------------------------------------------------------------- | The last commit position seen on the subscription (if this a subscription---   to $all stream).-getSubLastCommitPos :: Subscription a -> IO Int64-getSubLastCommitPos Subscription{..} = atomically $ do-    run <- readTMVar _subRun-    return $ S.runningLastCommitPosition run------------------------------------------------------------------------------------- | The last event number seen on the subscription (if this is a subscription---   to a single stream).-getSubLastEventNumber :: Subscription a -> IO (Maybe Int32)-getSubLastEventNumber Subscription{..} = atomically $ do-    run <- readTMVar _subRun-    return $ S.runningLastEventNumber run------------------------------------------------------------------------------------- | Awaits for the next event.-nextEvent :: Subscription a -> IO ResolvedEvent-nextEvent sub = atomically $ do-    m <- _nextEventMaybe sub-    case m of-        Nothing -> retrySTM-        Just e  -> return e------------------------------------------------------------------------------------- | Non blocking version of 'nextEvent'.-nextEventMaybe :: Subscription a -> IO (Maybe ResolvedEvent)-nextEventMaybe = atomically . _nextEventMaybe------------------------------------------------------------------------------------- | Waits until the `Subscription` has been confirmed.-waitConfirmation :: Subscription a -> IO ()-waitConfirmation s = atomically $ do-    _ <- readTMVar $ _subRun s-    return ()-----------------------------------------------------------------------------------_nextEventMaybe :: Subscription a -> STM (Maybe ResolvedEvent)-_nextEventMaybe Subscription{..} = do-    st <- _subState-    case st of-        SubException e -> throwSTM e-        SubState sub close -> do-            run <- readTMVar _subRun-            let (res, nxt) = S.readNext sub-            case res of-                Nothing -> do-                    case close of-                      Nothing  -> return Nothing-                      Just err -> throwSTM $ SubscriptionClosed run err-                Just e -> do-                  _subSetState $ SubState nxt close-                  return $ Just e------------------------------------------------------------------------------------- | Acknowledges those event ids have been successfully processed.-notifyEventsProcessed :: Subscription S.Persistent -> [UUID] -> IO ()-notifyEventsProcessed Subscription{..} evts = do-    run <- atomically $ readTMVar _subRun-    pushAckPersist _subProd run evts------------------------------------------------------------------------------------- | Acknowledges that 'ResolvedEvent' has been successfully processed.-acknowledge :: Subscription S.Persistent -> ResolvedEvent -> IO ()-acknowledge sub e = notifyEventsProcessed sub [resolvedEventOriginalId e]------------------------------------------------------------------------------------- | Acknowledges those 'ResolvedEvent's have been successfully processed.-acknowledgeEvents :: Subscription S.Persistent -> [ResolvedEvent] -> IO ()-acknowledgeEvents sub = notifyEventsProcessed sub . fmap resolvedEventOriginalId------------------------------------------------------------------------------------- | Mark a message that has failed processing. The server will take action---   based upon the action parameter.-failed :: Subscription S.Persistent-       -> ResolvedEvent-       -> NakAction-       -> Maybe Text-       -> IO ()-failed sub e a r = notifyEventsFailed sub a r [resolvedEventOriginalId e]------------------------------------------------------------------------------------- | Mark messages that have failed processing. The server will take action---   based upon the action parameter.-eventsFailed :: Subscription S.Persistent-             -> [ResolvedEvent]-             -> NakAction-             -> Maybe Text-             -> IO ()-eventsFailed sub evts a r =-    notifyEventsFailed sub a r $ fmap resolvedEventOriginalId evts------------------------------------------------------------------------------------- | Acknowledges those event ids have failed to be processed successfully.-notifyEventsFailed :: Subscription S.Persistent-                   -> NakAction-                   -> Maybe Text-                   -> [UUID]-                   -> IO ()-notifyEventsFailed Subscription{..} act res evts = do-    run <- atomically $ readTMVar _subRun-    pushNakPersist _subProd run act res evts------------------------------------------------------------------------------------- | Modifies 'SubState' internal state machine, letting any 'SubDropReason'---   untouched.-modifySubSM :: (S.Subscription a -> S.Subscription a)-            -> SubState a-            -> SubState a-modifySubSM k (SubState sm r) = SubState (k sm) r-modifySubSM _ s = s------------------------------------------------------------------------------------- | This exception is raised when the user tries to get the next event from a---   'Subscription' that is already closed.-data SubscriptionClosed =-    SubscriptionClosed S.Running S.SubDropReason-    deriving (Show, Typeable)------------------------------------------------------------------------------------instance Exception SubscriptionClosed--------------------------------------------------------------------------------- -- | Sends a list of 'Event' to given stream. sendEvents :: Connection            -> Text             -- ^ Stream name@@ -664,33 +461,41 @@     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           -> Bool       -- ^ Resolve Link Tos-          -> IO (Subscription S.Regular)-subscribe Connection{..} stream_id res_lnk_tos = do-    mvar <- newEmptyTMVarIO-    var  <- newTVarIO $ SubState S.regularSubscription Nothing-    let mk r = putTMVar mvar r-        recv = readTVar var-        send = writeTVar var-        dropped r = do-            SubState sm _ <- readTVar var-            writeTVar var $ SubState sm (Just r)-        cb = createSubAsync mk recv send dropped-    pushConnectStream _prod cb stream_id res_lnk_tos-    let getState = readTVar var-        setState = writeTVar var-    return $ Subscription getState setState mvar stream_id _prod-      (S.Regular res_lnk_tos)+          -> IO (Subscription Regular)+subscribe conn streamId resLnkTos =+    regularSub (mkSubEnv conn) streamId resLnkTos  -------------------------------------------------------------------------------- -- | Subcribes to $all stream. subscribeToAll :: Connection                -> Bool       -- ^ Resolve Link Tos-               -> IO (Subscription S.Regular)-subscribeToAll conn res_lnk_tos = subscribe conn "" res_lnk_tos+               -> IO (Subscription Regular)+subscribeToAll conn = subscribe conn ""  -------------------------------------------------------------------------------- -- | Subscribes to given stream. If last checkpoint is defined, this will@@ -702,11 +507,11 @@               -> Bool        -- ^ Resolve Link Tos               -> Maybe Int32 -- ^ Last checkpoint               -> Maybe Int32 -- ^ Batch size-              -> IO (Subscription S.Catchup)-subscribeFrom conn stream_id res_lnk_tos last_chk_pt batch_m =-    subscribeFromCommon conn stream_id res_lnk_tos batch_m tpe+              -> IO (Subscription Catchup)+subscribeFrom conn streamId resLnkTos lastChkPt batch =+    subscribeFromCommon conn resLnkTos batch tpe   where-    tpe = Op.RegularCatchup stream_id (fromMaybe 0 last_chk_pt)+    tpe = Op.RegularCatchup streamId (fromMaybe 0 lastChkPt)  -------------------------------------------------------------------------------- -- | Same as 'subscribeFrom' but applied to $all stream.@@ -714,62 +519,26 @@                    -> Bool           -- ^ Resolve Link Tos                    -> Maybe Position -- ^ Last checkpoint                    -> Maybe Int32    -- ^ Batch size-                   -> IO (Subscription S.Catchup)-subscribeToAllFrom conn res_lnk_tos last_chk_pt batch_m =-    subscribeFromCommon conn "" res_lnk_tos batch_m tpe+                   -> IO (Subscription Catchup)+subscribeToAllFrom conn resLnkTos lastChkPt batch =+    subscribeFromCommon conn resLnkTos batch tpe   where-    Position c_pos p_pos = fromMaybe positionStart last_chk_pt-    tpe = Op.AllCatchup c_pos p_pos+    Position cPos pPos = fromMaybe positionStart lastChkPt+    tpe = Op.AllCatchup cPos pPos  -------------------------------------------------------------------------------- subscribeFromCommon :: Connection-                    -> Text                     -> Bool                     -> Maybe Int32                     -> Op.CatchupState-                    -> IO (Subscription S.Catchup)-subscribeFromCommon Connection{..} stream_id res_lnk_tos batch_m tpe = do-    mvarRun <- newEmptyTMVarIO-    mvarSub <- newEmptyTMVarIO-    let readFrom res =-            case res of-                -- We want to notify the user that something went wrong in the-                -- first phase of a catchup subscription (e.g. reading the-                -- stream forward until we catchup to stream's end). This-                -- prevents a deadlock on user side in case where the user calls-                -- `waitTillCatchup` on a stream that doesn't exist.-                Left e -> atomically $ do-                  isEmpty <- isEmptyTMVar mvarSub-                  if isEmpty-                      then putTMVar mvarSub (SubException e)-                      else () <$ swapTMVar mvarSub (SubException e)--                Right (xs, eos, chk) -> atomically $ do-                    -- When a catchup subscription receives events for the-                    -- first time.-                    whenM (isEmptyTMVar mvarSub) $ do-                        let initState = SubState S.catchupSubscription Nothing-                        putTMVar mvarSub initState--                    subState <- takeTMVar mvarSub-                    let nxtSubState =-                          modifySubSM (S.batchRead xs eos chk) subState--                    putTMVar mvarSub nxtSubState-        mk   = putTMVar mvarRun-        rcv  = readTMVar mvarSub-        send = \x -> () <$ swapTMVar mvarSub x-        dropped r = do-            SubState sm _ <- takeTMVar mvarSub-            putTMVar mvarSub $ SubState sm (Just r)-        op  = Op.catchup _settings tpe res_lnk_tos batch_m-        cb  = createSubAsync mk rcv send dropped--    pushOperation _prod readFrom op-    pushConnectStream _prod cb stream_id res_lnk_tos-    let getState = readTMVar mvarSub-        setState = \x -> () <$ swapTMVar mvarSub x-    return $ Subscription getState setState mvarRun stream_id _prod S.Catchup+                    -> IO (Subscription Catchup)+subscribeFromCommon conn resLnkTos batch tpe =+    catchupSub (mkSubEnv conn) params+  where+    params = CatchupParams { catchupResLnkTos = resLnkTos+                           , catchupState = tpe+                           , catchupBatchSize = batch+                           }  -------------------------------------------------------------------------------- -- | Asynchronously sets the metadata for a stream.@@ -799,7 +568,7 @@                              -> Text                              -> Text                              -> PersistentSubscriptionSettings-                             -> IO (Async (Maybe S.PersistActionException))+                             -> IO (Async (Maybe PersistActionException)) createPersistentSubscription Connection{..} group stream sett = do     mvar <- newEmptyTMVarIO     let _F res = atomically $@@ -815,7 +584,7 @@                              -> Text                              -> Text                              -> PersistentSubscriptionSettings-                             -> IO (Async (Maybe S.PersistActionException))+                             -> IO (Async (Maybe PersistActionException)) updatePersistentSubscription Connection{..} group stream sett = do     mvar <- newEmptyTMVarIO     let _F res = atomically $@@ -830,7 +599,7 @@ deletePersistentSubscription :: Connection                              -> Text                              -> Text-                             -> IO (Async (Maybe S.PersistActionException))+                             -> IO (Async (Maybe PersistActionException)) deletePersistentSubscription Connection{..} group stream = do     mvar <- newEmptyTMVarIO     let _F res = atomically $@@ -847,22 +616,9 @@                                 -> Text                                 -> Text                                 -> Int32-                                -> IO (Subscription S.Persistent)-connectToPersistentSubscription Connection{..} group stream bufSize = do-    mvar <- newEmptyTMVarIO-    var  <- newTVarIO $ SubState S.persistentSubscription Nothing-    let mk r = putTMVar mvar r-        recv = readTVar var-        send = writeTVar var-        dropped r = do-            SubState sm _ <- readTVar var-            writeTVar var $ SubState sm (Just r)-        cb = createSubAsync mk recv send dropped-    pushConnectPersist _prod cb group stream bufSize-    let getState = readTVar var-        setState = writeTVar var-    return $ Subscription getState setState mvar stream _prod-      (S.Persistent group)+                                -> IO (Subscription Persistent)+connectToPersistentSubscription conn group stream bufSize =+    persistentSub (mkSubEnv conn) group stream bufSize  -------------------------------------------------------------------------------- createOpAsync :: IO (Either OperationError a -> IO (), Async a)@@ -872,25 +628,3 @@         res <- readMVar mvar         either throwIO return res     return (putMVar mvar, as)-----------------------------------------------------------------------------------createSubAsync :: (S.Running -> STM ())-               -> STM (SubState a)-               -> (SubState a -> STM ())-               -> (S.SubDropReason -> STM ())-               -> (S.SubConnectEvent -> IO ())-createSubAsync mk rcv send quit = go-  where-    go (S.SubConfirmed run) = atomically $ mk run-    go (S.EventAppeared e) = atomically $ do-          st <- rcv-          case st of-              SubState sm close ->-                  let nxt = S.eventArrived e sm in-                  send $ SubState nxt close-              SubException _ ->-                  -- At this moment [07 October 2016], this can only happen during-                  -- the first phase of a catchup subscription where the user-                  -- asked for a subscription on a stream that doesn't exist.-                return ()-    go (S.Dropped r) = atomically $ quit r

@@ -13,21 +13,15 @@ module Database.EventStore.Internal.Command (Command(..)) where  ---------------------------------------------------------------------------------import Data.Word-import Numeric+import ClassyPrelude  ---------------------------------------------------------------------------------import ClassyPrelude+import Database.EventStore.Internal.Utils  -------------------------------------------------------------------------------- -- | Internal command representation. newtype Command = Command { cmdWord8 :: Word8 } deriving (Eq, Ord, Num)  ---------------------------------------------------------------------------------padding :: String -> String-padding [x] = ['0',x]-padding xs  = xs---------------------------------------------------------------------------------- instance Show Command where-    show (Command w) = "0x" ++ padding (showHex w "")+    show (Command w) = prettyWord8 w

@@ -22,6 +22,7 @@     , connRecv     , connIsClosed     , newConnection+    , connForceReconnect     ) where  --------------------------------------------------------------------------------@@ -36,6 +37,7 @@  -------------------------------------------------------------------------------- import Database.EventStore.Internal.Command+import Database.EventStore.Internal.EndPoint import Database.EventStore.Internal.Discovery import Database.EventStore.Internal.Types import Database.EventStore.Logging@@ -63,6 +65,7 @@     Close :: In ()     Send  :: Package -> In ()     Recv  :: In Package+    ForceReconnect :: NodeEndPoints -> In ()  -------------------------------------------------------------------------------- -- | Represents connection logic action to carry out.@@ -131,6 +134,11 @@         _      -> return False  --------------------------------------------------------------------------------+-- | Forces reconnection on given node.+connForceReconnect :: InternalConnection -> NodeEndPoints -> IO ()+connForceReconnect conn = execute conn . ForceReconnect++-------------------------------------------------------------------------------- -- Connection Logic -------------------------------------------------------------------------------- onlineLogic :: forall a. TMVar State@@ -175,6 +183,7 @@ handleInput _ conn Recv = recv conn handleInput uuid _ Id = return uuid handleInput _ conn Close = liftIO $ connectionClose conn+handleInput _ _ ForceReconnect{} = return ()  -------------------------------------------------------------------------------- -- | Main connection logic. It will automatically reconnect to the server when@@ -188,7 +197,10 @@         Errored e -> throwIO e         WithConnection uuid conn op -> handleInput uuid conn op         CreateConnection op -> do-            (uuid, conn) <- openConnection iconn+            (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 @@ -198,8 +210,14 @@ reachedMaxAttempt (AtMost n) cur = n <= cur  ---------------------------------------------------------------------------------openConnection :: InternalConnection -> IO (UUID, Connection)-openConnection InternalConnection{..} = attempt 1+isSsl :: Settings -> Bool+isSsl = isJust . s_ssl++--------------------------------------------------------------------------------+openConnection :: InternalConnection+               -> Maybe NodeEndPoints+               -> IO (UUID, Connection)+openConnection InternalConnection{..} nodeM = attempt 1   where     delay = s_reconnect_delay_secs _setts * secs @@ -212,17 +230,31 @@      attempt trialCount = do         _settingsLog _setts (Info $ Connecting trialCount)-        old <- readIORef _last-        ept_opt <- runDiscovery _disc old-        case ept_opt of-            Nothing -> handleFailure trialCount-            Just ept -> do-                let host = endPointIp ept+        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)  -------------------------------------------------------------------------------- secs :: Int

@@ -49,6 +49,9 @@ import System.Random  --------------------------------------------------------------------------------+import Database.EventStore.Internal.EndPoint++-------------------------------------------------------------------------------- data DnsDiscoveryException     = MaxDiscoveryAttemptReached ByteString     | DNSDiscoveryError DNSError@@ -56,18 +59,6 @@  -------------------------------------------------------------------------------- instance Exception DnsDiscoveryException------------------------------------------------------------------------------------- | Gathers both an IPv4 and a port.-data EndPoint =-    EndPoint-    { endPointIp   :: !String-    , endPointPort :: !Int-    } deriving Show-----------------------------------------------------------------------------------emptyEndPoint :: EndPoint-emptyEndPoint = EndPoint "" 0  -------------------------------------------------------------------------------- httpRequest :: EndPoint -> String -> IO Request

@@ -0,0 +1,34 @@+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Internal.EndPoint+-- 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.EndPoint where++--------------------------------------------------------------------------------+import ClassyPrelude++--------------------------------------------------------------------------------+-- | Gathers both an IPv4 and a port.+data EndPoint =+    EndPoint+    { endPointIp   :: !String+    , endPointPort :: !Int+    } deriving Show++--------------------------------------------------------------------------------+emptyEndPoint :: EndPoint+emptyEndPoint = EndPoint "" 0++--------------------------------------------------------------------------------+data NodeEndPoints =+    NodeEndPoints+    { tcpEndPoint :: !EndPoint+    , secureEndPoint :: !(Maybe EndPoint)+    } deriving Show

@@ -36,6 +36,7 @@     , pushNakPersist     , pushUnsubscribe     , prodWaitTillClosed+    , pushForceReconnect     ) where  --------------------------------------------------------------------------------@@ -50,14 +51,16 @@ -------------------------------------------------------------------------------- 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 hiding+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@@ -191,6 +194,7 @@           Text Text     | AckPersist Running [UUID]     | NakPersist Running NakAction (Maybe Text) [UUID]+    | ForceReconnect NodeEndPoints  -------------------------------------------------------------------------------- pushCmd :: Production -> Msg -> IO ()@@ -286,6 +290,10 @@ prodWaitTillClosed (Prod _ disposed) = atomically disposed  --------------------------------------------------------------------------------+pushForceReconnect :: Production -> NodeEndPoints -> IO ()+pushForceReconnect prod n = pushCmd prod (ForceReconnect n)++-------------------------------------------------------------------------------- newtype Job = Job (IO ())  --------------------------------------------------------------------------------@@ -374,6 +382,9 @@         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.@@ -390,6 +401,13 @@     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@@ -400,6 +418,7 @@     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@@ -517,6 +536,14 @@ 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@@ -535,15 +562,23 @@             case res of                 Left e -> do                     _settingsLog setts (Error $ UnexpectedException e)-                    case fromException e of-                        Just (_ :: ConnectionException) -> atomically $ do-                            writeTVar nxt_sub (raiseException e)-                            putTMVar disposed ()-                        _ -> do new_conn <- newConnection setts disc-                                writeIORef conn_ref new_conn-                                _ <- forkFinally (bootstrap env) handler-                                return ()-+                    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+                              writeIORef conn_ref new_conn+                              _ <- forkFinally (bootstrap env) handler+                              return ()+                        , Handler $ \(_ :: SomeException) -> do+                              new_conn <- newConnection setts disc+                              writeIORef conn_ref new_conn+                              _ <- forkFinally (bootstrap env) handler+                              return ()+                        ]                 _ -> atomically $ putTMVar disposed ()     _ <- forkFinally (bootstrap env) handler     return $ Prod nxt_sub $ do

@@ -77,6 +77,7 @@       -- ^ Asks for sending the given 'Package'.     | Await (Model r)       -- ^ waits for more input.+    | NotHandled MasterInfo (Transition r)  -------------------------------------------------------------------------------- -- | Main 'Operation' bookkeeping state machine.@@ -139,20 +140,40 @@  -------------------------------------------------------------------------------- runPackage :: Settings -> State r -> Package -> Maybe (Transition r)-runPackage setts st Package{..} = do+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 }-    if resp_cmd /= packageCmd-        then-            let r = cb $ Left $ InvalidServerResponse resp_cmd packageCmd in-            return $ Produce r (Await $ Model $ execute setts nxt_st)-        else-            case runGet decodeMessage packageData of-                Left e  ->+    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+                  Right m -> return $ runOperation setts cb op (cont m) nxt_st  -------------------------------------------------------------------------------- abortOperations :: Settings -> State r -> Transition r@@ -174,3 +195,10 @@ 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

@@ -1,220 +0,0 @@-{-# LANGUAGE GADTs               #-}-{-# LANGUAGE Rank2Types          #-}-{-# LANGUAGE ScopedTypeVariables #-}------------------------------------------------------------------------------------ |--- Module : Database.EventStore.Internal.Manager.Subscription--- Copyright : (C) 2015 Yorick Laupa--- License : (see the file LICENSE)------ Maintainer : Yorick Laupa <yo.eight@gmail.com>--- Stability : provisional--- Portability : non-portable------ Main subscription state machine declaration module. It also declares every--- functions required to drive a 'Subscription'.----------------------------------------------------------------------------------module Database.EventStore.Internal.Manager.Subscription-    ( module Database.EventStore.Internal.Manager.Subscription.Driver-    , Regular(..)-    , Persistent(..)-    , Catchup(..)-    , Subscription-    , Running(..)-    , Checkpoint(..)-    , regularSubscription-    , catchupSubscription-    , persistentSubscription-    , eventArrived-    , readNext-    , batchRead-    , hasCaughtUp-    , runningUUID-    , runningLastEventNumber-    , runningLastCommitPosition-    ) where-----------------------------------------------------------------------------------import Data.Int-----------------------------------------------------------------------------------import ClassyPrelude-import Data.Sequence (ViewL(..), viewl, dropWhileL)-----------------------------------------------------------------------------------import Database.EventStore.Internal.Manager.Subscription.Driver-import Database.EventStore.Internal.Manager.Subscription.Model-import Database.EventStore.Internal.Types------------------------------------------------------------------------------------- | Also referred as volatile subscription. For example, if a stream has 100---   events in it when a subscriber connects, the subscriber can expect to see---   event number 101 onwards until the time the subscription is closed or---   dropped.-data Regular = Regular { _subTos :: Bool }------------------------------------------------------------------------------------- | This kind of subscription specifies a starting point, in the form of an---   event number or transaction file position. The given function will be---   called for events from the starting point until the end of the stream, and---   then for subsequently written events.------   For example, if a starting point of 50 is specified when a stream has 100---   events in it, the subscriber can expect to see events 51 through 100, and---   then any events subsequently written until such time as the subscription is---   dropped or closed.-data Catchup = Catchup------------------------------------------------------------------------------------- | The server remembers the state of the subscription. This allows for many---   different modes of operations compared to a regular or catchup subscription---   where the client holds the subscription state.---   (Need EventStore >= v3.1.0).-data Persistent = Persistent { _perGroup  :: Text }------------------------------------------------------------------------------------- | Represents the different type of inputs a subscription state-machine can---   handle.-data Input t a where-    -- A event has written to the stream. Subscription state machine should-    -- store that event withing its state.-    Arrived :: ResolvedEvent -> Input t (Subscription t)-    -- The user asks for the next event coming from the server.-    ReadNext :: Input t (Maybe ResolvedEvent, Subscription t)-    -- A batch read has been made. It's only use for 'Catchup' subscription-    -- type. It gives the list of read events and indicates if it reaches the-    -- end of the stream along with the next checkpoint to point at.-    BatchRead :: [ResolvedEvent]-              -> Bool-              -> Checkpoint-              -> Input Catchup (Subscription Catchup)-    -- Used only for 'Catchup' subscription type. Asks if the subscription-    -- read every events up to the checkpoint given by the user.-    CaughtUp :: Input Catchup Bool------------------------------------------------------------------------------------- | Main subscription state machine.-newtype Subscription t = Subscription (forall a. Input t a -> a)------------------------------------------------------------------------------------- | Submit a new event to the subscription state machine. Internally,---   that event should be stored into the subscription buffer.-eventArrived :: ResolvedEvent -> Subscription t -> Subscription t-eventArrived e (Subscription k) = k (Arrived e)------------------------------------------------------------------------------------- | Reads the next available event. Returns 'Nothing' it there is any. When---   returning an event, it will be removed from the subscription buffer.-readNext :: Subscription t -> (Maybe ResolvedEvent, Subscription t)-readNext (Subscription k) = k ReadNext------------------------------------------------------------------------------------- | Submits a list of events read from a stream. It's only used by a 'Catchup'---   subscription.-batchRead :: [ResolvedEvent]-          -> Bool -- ^ If it reaches the end of the stream.-          -> Checkpoint-          -> Subscription Catchup-          -> Subscription Catchup-batchRead es eos nxt (Subscription k) = k (BatchRead es eos nxt)------------------------------------------------------------------------------------- | Indicates if the subscription caught up the end of the stream, meaning the---   subscription is actually live. Only used by 'Catchup' subscription.-hasCaughtUp :: Subscription Catchup -> Bool-hasCaughtUp (Subscription k) = k CaughtUp------------------------------------------------------------------------------------- | Main 'Regular' subscription state machine.-regularSubscription :: Subscription Regular-regularSubscription = baseSubscription------------------------------------------------------------------------------------- | Main 'Persistent' subscription state machine.-persistentSubscription :: Subscription Persistent-persistentSubscription = baseSubscription------------------------------------------------------------------------------------- | Represents the next checkpoint to reach on a catchup subscription. Wheither---   it's a regular stream or the $all stream, it either point to an 'Int32' or---   a 'Position'.-data Checkpoint = CheckpointNumber Int32 | CheckpointPosition Position------------------------------------------------------------------------------------- | Depending either if the subscription concerns a regular stream or $all,---  indicates if an event number (or 'Position') is lesser that the current the---  given 'CheckPoint'.-beforeChk :: Checkpoint -> ResolvedEvent -> Bool-beforeChk (CheckpointNumber num) re =-    recordedEventNumber (resolvedEventOriginal re) < num-beforeChk (CheckpointPosition pos) re =-    maybe False (< pos) $ resolvedEventPosition re------------------------------------------------------------------------------------- | That subscription state machine accumulates events coming from batch read---   and any real time change made on a stream. That state machine will not---   served any recent change made on the stream until it reaches the end of the---   stream. On every batch read, it makes sure events contained in that batch---   are deleted from the subscription buffer in order to avoid duplicates. That---   implemention has been chosen to avoid potential message lost between the---   moment with reach the end of the stream and the delay required by asking---   for a subscription.-catchupSubscription :: Subscription Catchup-catchupSubscription = Subscription $ catchingUp empty empty-  where-    catchingUp :: forall a. Seq ResolvedEvent-               -> Seq ResolvedEvent-               -> Input Catchup a-               -> a-    catchingUp b s (Arrived e) =-        Subscription $ catchingUp b (s `snoc` e)-    catchingUp b s ReadNext =-        case viewl b of-            EmptyL    -> (Nothing, Subscription $ catchingUp b s)-            e :< rest -> (Just e, Subscription $ catchingUp rest s)-    catchingUp b s (BatchRead es eos nxt_pt) =-        let nxt_b = foldl' snoc b es-            nxt_s = dropWhileL (beforeChk nxt_pt) s-            nxt   = if eos-                    then Subscription $ caughtUp nxt_b nxt_s-                    else Subscription $ catchingUp nxt_b nxt_s in-        nxt-    catchingUp _ _ CaughtUp = False--    caughtUp :: forall a. Seq ResolvedEvent-             -> Seq ResolvedEvent-             -> Input Catchup a-             -> a-    caughtUp  b s (Arrived e) = Subscription $ caughtUp b (s `snoc` e)-    caughtUp b s  ReadNext =-        case viewl b of-            EmptyL -> live s ReadNext-            e :< rest ->-                case viewl rest of-                    EmptyL -> (Just e, Subscription $ live s)-                    _      -> (Just e, Subscription $ caughtUp rest s)-    caughtUp b s (BatchRead _ _ _) = Subscription $ caughtUp b s-    caughtUp _ _ CaughtUp = False--    live :: forall a. Seq ResolvedEvent -> Input Catchup a -> a-    live s (Arrived e) = Subscription $ live (s `snoc` e)-    live s ReadNext =-        case viewl s of-            EmptyL    -> (Nothing, Subscription $ live s)-            e :< rest -> (Just e, Subscription $ live rest)-    live s (BatchRead _ _ _) = Subscription $ live s-    live _ CaughtUp = True------------------------------------------------------------------------------------- | Base subscription used for 'Regular' or 'Persistent' subscription.-baseSubscription :: forall t. Subscription t-baseSubscription = Subscription $ go empty-  where-    go :: forall a. Seq ResolvedEvent -> Input t a -> a-    go s (Arrived e) = Subscription $ go (s `snoc` e)-    go s ReadNext =-        case viewl s of-            EmptyL    -> (Nothing, Subscription $ go s)-            e :< rest -> (Just e, Subscription $ go rest)-    go _ BatchRead{} = error "impossible: base subscription"-    go _ CaughtUp    = error "impossible: base subscription"

@@ -0,0 +1,103 @@+--------------------------------------------------------------------------------+-- |+-- 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,8 +1,10 @@-{-# LANGUAGE DeriveDataTypeable  #-}-{-# LANGUAGE GADTs               #-}-{-# LANGUAGE Rank2Types          #-}-{-# LANGUAGE RecordWildCards     #-}-{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE DeriveDataTypeable         #-}+{-# LANGUAGE GADTs                      #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE OverloadedStrings          #-}+{-# LANGUAGE Rank2Types                 #-}+{-# LANGUAGE RecordWildCards            #-}+{-# LANGUAGE ScopedTypeVariables        #-} -------------------------------------------------------------------------------- -- | -- Module : Database.EventStore.Internal.Manager.Subscription.Driver@@ -41,16 +43,17 @@ import Data.Maybe  ---------------------------------------------------------------------------------import ClassyPrelude-import Data.Serialize-import Data.ProtocolBuffers+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  --------------------------------------------------------------------------------@@ -63,22 +66,7 @@     | SubConfirmed Running       -- ^ Subscription connection is confirmed. It means that subscription can       --   receive events from the server.------------------------------------------------------------------------------------- | 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.-    deriving (Show, Eq)+    | Unsubscribed  -------------------------------------------------------------------------------- -- | Enumerates all persistent action exceptions.@@ -245,14 +233,7 @@ updateRException UPS_AccessDenied = Just PersistActionAccessDenied  ---------------------------------------------------------------------------------toSubDropReason :: DropReason -> SubDropReason-toSubDropReason D_Unsubscribed                  = SubUnsubscribed-toSubDropReason D_NotFound                      = SubNotFound-toSubDropReason D_AccessDenied                  = SubAccessDenied-toSubDropReason D_PersistentSubscriptionDeleted = SubPersistDeleted- ----------------------------------------------------------------------------------------------------------------------------------------------------------------- -- | Type of inputs handled by the 'Subscription' driver. data In r a where     -- A command consists of receiving some parameters, updating the@@ -310,9 +291,15 @@       --   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 State r =-    State+data Internal r =+    Internal     { _model :: !Model       -- ^ Subscription model.     , _gen :: !Generator@@ -323,155 +310,241 @@     }  ---------------------------------------------------------------------------------initState :: Generator -> State r-initState gen = State newModel gen mempty+initInternal :: Generator -> Internal r+initInternal gen = Internal newModel gen mempty  -------------------------------------------------------------------------------- -- | Subscription driver state machine. newtype Driver r = Driver (forall a. In r a -> a)  ----------------------------------------------------------------------------------- | Creates a new subscription 'Driver' state machine.-newDriver :: forall r. Settings -> Generator -> Driver r-newDriver setts gen = Driver $ go (initState gen)-  where-    go :: forall a. State r -> In r a -> a-    go st@State{..} (Pkg Package{..}) = do-        elm <- lookup packageCorrelation _reg-        case packageCmd of-            0xC2 -> do-                _   <- querySubscription packageCorrelation _model-                msg <- maybeDecodeMessage packageData-                let e   = getField $ streamResolvedEvent msg-                    evt = newResolvedEventFromBuf e-                    app = EventAppeared evt-                    ConnectReg k _ _ = elm-                return (k app, Driver $ go st)+newtype DriverM r m a = DriverM (ReaderT Settings (StateT (Internal r) m) a)+    deriving ( Functor+             , Applicative+             , Monad+             , MonadReader Settings+             , MonadState (Internal r)+             ) -            0xC7 -> do-                _   <- querySubscription packageCorrelation _model-                msg <- maybeDecodeMessage packageData-                let e   = getField $ psseaEvt msg-                    evt = newResolvedEvent e-                    app = EventAppeared evt-                    ConnectPersist k _ _ _ = elm-                return (k app, Driver $ go st)+--------------------------------------------------------------------------------+instance MonadTrans (DriverM r) where+    lift m = DriverM $ lift $ lift m -            0xC1 -> do-                msg <- maybeDecodeMessage packageData-                let lcp  = getField $ subscribeLastCommitPos msg-                    len  = getField $ subscribeLastEventNumber msg-                    meta = RegularMeta lcp len-                    ConnectReg k _ _ = elm-                    nxt_m = confirmedSubscription packageCorrelation meta _model-                run   <- querySubscription packageCorrelation nxt_m-                let nxt_st = st { _model = nxt_m }-                    evt    = SubConfirmed run-                return (k evt, Driver $ go nxt_st)+--------------------------------------------------------------------------------+noop :: DriverM r Maybe a+noop = lift Nothing -            0xC6 -> do-                msg <- maybeDecodeMessage packageData-                let lcp  = getField $ pscLastCommitPos msg-                    sid  = getField $ pscId msg-                    len  = getField $ pscLastEvtNumber msg-                    meta = PersistMeta sid lcp len-                    ConnectPersist k _ _ _ = elm-                    nxt_m = confirmedSubscription packageCorrelation meta _model-                run   <- querySubscription packageCorrelation nxt_m-                let nxt_st = st { _model = nxt_m }-                    evt    = SubConfirmed run-                return (k evt, Driver $ go nxt_st)+--------------------------------------------------------------------------------+modelSubRunning :: UUID -> DriverM r Maybe Running+modelSubRunning uuid = do+    model <- gets _model+    lift $ querySubscription uuid $ model -            0xC9 -> confirmPAction elm (getField . cpscResult) createRException-            0xCF -> confirmPAction elm (getField . upscResult) updateRException-            0xCB -> confirmPAction elm (getField . dpscResult) deleteRException+--------------------------------------------------------------------------------+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 } -            0xC4 -> do-                run <- querySubscription packageCorrelation _model-                msg <- maybeDecodeMessage packageData-                let reason  = fromMaybe D_Unsubscribed $ getField-                                                       $ dropReason msg-                    nxt_m   = unsubscribed run _model-                    dreason = toSubDropReason reason-                    evt     = Dropped dreason-                    nxt_reg = deleteMap packageCorrelation _reg-                    nxt_st  = st { _model = nxt_m-                                 , _reg   = nxt_reg }-                case elm of-                  ConnectReg k _ _       -> return (k evt, Driver $ go nxt_st)-                  ConnectPersist k _ _ _ -> return (k evt, Driver $ go nxt_st)-                  _                      -> Nothing+--------------------------------------------------------------------------------+modelActionConfirmed :: Monad m => UUID -> DriverM r m ()+modelActionConfirmed uuid =+    modify $ \s -> s { _model = confirmedAction uuid $ _model s } -            _ -> Nothing-      where-        confirmPAction :: Decode m-                       => Cmd r-                       -> (m -> e)-                       -> (e -> Maybe PersistActionException)-                       -> Maybe (r, Driver r)-        confirmPAction (ApplyPersistAction k g n c) fd em = do-            msg <- maybeDecodeMessage packageData-            _   <- queryPersistentAction packageCorrelation _model-            let nxt_m  = confirmedAction packageCorrelation _model-                nxt_rg = deleteMap packageCorrelation _reg-                nxt_st = st { _model = nxt_m-                            , _reg   = nxt_rg-                            }-                evt    = ConfirmedAction packageCorrelation g n c-            case em $ fd msg of-                Just e  -> return (k $ Left e, Driver $ go nxt_st)-                Nothing -> return (k $ Right evt, Driver $ go nxt_st)-        confirmPAction _ _ _ = Nothing+--------------------------------------------------------------------------------+modelUnsubscribed :: UUID -> DriverM r Maybe ()+modelUnsubscribed uuid = do+    run <- modelSubRunning uuid+    model <- gets _model+    modify $ \s -> s { _model = unsubscribed run model } -    go st@State{..} (Cmd cmd) =-        case cmd of-            ConnectReg _ s tos ->-                let (u, nxt_g) = nextUUID _gen-                    pkg        = createConnectRegularPackage setts u s tos-                    nxt_m      = connectReg s tos u _model-                    nxt_st     = st { _model = nxt_m-                                    , _gen   = nxt_g-                                    , _reg   = insertMap u cmd _reg } in-                (pkg, Driver $ go nxt_st)-            ConnectPersist _ gn n b ->-                let (u, nxt_g) = nextUUID _gen-                    pkg        = createConnectPersistPackage setts u gn n b-                    nxt_m      = connectPersist gn n b u _model-                    nxt_st     = st { _model = nxt_m-                                    , _gen   = nxt_g-                                    , _reg   = insertMap u cmd _reg } in-                (pkg, Driver $ go nxt_st)-            Unsubscribe r ->-                let pkg    = createUnsubscribePackage setts $ runningUUID r in-                (pkg, Driver $ go st)-            ApplyPersistAction _ gn n a ->-                let (u, nxt_g) = nextUUID _gen-                    pkg        = createPersistActionPackage setts u gn n a-                    nxt_m      = persistAction gn n u a _model-                    nxt_st     = st { _model = nxt_m-                                    , _gen   = nxt_g-                                    , _reg   = insertMap u cmd _reg } in-                (pkg, Driver $ go nxt_st)-            PersistAck _ run evts ->-                let RunningPersist _ _ _ _ sid _ _ = run-                    u   = runningUUID run-                    pkg = createAckPackage setts u sid evts in-                (pkg, Driver $ go st)-            PersistNak _ run na r evts ->-                let RunningPersist _ _ _ _ sid _ _ = run-                    u   = runningUUID run-                    pkg = createNakPackage setts u sid na r evts  in-                (pkg, Driver $ go st)-    go 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 _                            = []+--------------------------------------------------------------------------------+registerDelete :: Monad m => UUID -> DriverM r m ()+registerDelete uuid = do+    reg <- gets _reg+    let nxtR = deleteMap uuid reg+    modify $ \s -> s { _reg = nxtR }  ---------------------------------------------------------------------------------maybeDecodeMessage :: Decode a => ByteString -> Maybe a-maybeDecodeMessage bytes =-    case runGet decodeMessage bytes of-        Right a -> Just a-        _       -> Nothing+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)

@@ -76,6 +76,7 @@     | D_AccessDenied     | D_NotFound     | D_PersistentSubscriptionDeleted+    | D_SubscriberMaxCountReached     deriving (Enum, Eq, Show)  --------------------------------------------------------------------------------

@@ -0,0 +1,40 @@+--------------------------------------------------------------------------------+-- |+-- 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)

@@ -50,6 +50,7 @@     | ProtobufDecodingError String     | ServerError (Maybe Text)                  -- ^ Reason     | InvalidOperation Text+    | NotAuthenticatedOp       -- ^ Invalid operation state. If happens, it's a driver bug.     | Aborted       -- ^ Occurs when the user asked to close the connection or if the

@@ -14,7 +14,10 @@ -------------------------------------------------------------------------------- module Database.EventStore.Internal.Operation.Catchup     ( CatchupState(..)+    , CatchupOpResult(..)+    , Checkpoint(..)     , catchup+    , catchupStreamName     ) where  --------------------------------------------------------------------------------@@ -25,7 +28,6 @@ import ClassyPrelude  ---------------------------------------------------------------------------------import Database.EventStore.Internal.Manager.Subscription (Checkpoint(..)) import Database.EventStore.Internal.Operation import Database.EventStore.Internal.Operation.Read.Common import Database.EventStore.Internal.Operation.ReadAllEvents@@ -34,6 +36,12 @@ import Database.EventStore.Internal.Types  --------------------------------------------------------------------------------+-- | Represents the next checkpoint to reach on a catchup subscription. Wheither+--   it's a regular stream or the $all stream, it either point to an 'Int32' or+--   a 'Position'.+data Checkpoint = CheckpointNumber Int32 | CheckpointPosition Position++-------------------------------------------------------------------------------- defaultBatchSize :: Int32 defaultBatchSize = 500 @@ -52,17 +60,24 @@       --   the $all stream.  ---------------------------------------------------------------------------------streamName :: CatchupState -> Text-streamName (RegularCatchup stream _) = stream-streamName _ = "$all"+catchupStreamName :: CatchupState -> Text+catchupStreamName (RegularCatchup stream _) = stream+catchupStreamName _ = "$all"  --------------------------------------------------------------------------------+data CatchupOpResult =+    CatchupOpResult { catchupReadEvents :: ![ResolvedEvent]+                    , catchupEndOfStream :: !Bool+                    , catchupCheckpoint :: !Checkpoint+                    }++-------------------------------------------------------------------------------- -- | Stream catching up operation. catchup :: Settings         -> CatchupState         -> Bool         -> Maybe Int32-        -> Operation ([ResolvedEvent], Bool, Checkpoint)+        -> Operation CatchupOpResult -- ([ResolvedEvent], Bool, Checkpoint) catchup setts init_tpe tos bat_siz = go init_tpe   where     batch = fromMaybe defaultBatchSize bat_siz@@ -85,15 +100,15 @@                         tmp_tpe = AllCatchup nxt_c nxt_p                         chk     = CheckpointPosition $ sliceNext as                     return (sliceEOS as, sliceEvents as, chk, tmp_tpe)-                Left rr -> fromReadResult (streamName tpe) rr $ \as ->+                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 (evts, eos, nchk)-            when (not eos) $ go nxt_tpe+            yield (CatchupOpResult evts eos nchk)+            unless eos $ go nxt_tpe  -------------------------------------------------------------------------------- fromReadResult :: Text

@@ -60,7 +60,7 @@ setMetaStream setts s v meta =     let stream = metaStream s         json   = streamMetadataJSON meta-        evt    = createEvent "$metadata" Nothing (withJson json)+        evt    = createEvent StreamMetadataType Nothing (withJson json)         inner  = writeEvents setts stream v [evt] in     foreach inner yield 

@@ -44,7 +44,7 @@                       evt_data_bytes                       evt_metadata_bytes   where-    evt_type           = eventType evt+    evt_type           = eventTypeText $ eventType evt     evt_id             = eventId evt     evt_data_bytes     = eventDataBytes $ eventData evt     evt_data_type      = eventDataType $ eventData evt

@@ -38,13 +38,15 @@ import Data.UUID  --------------------------------------------------------------------------------+import Database.EventStore.Internal.EndPoint import Database.EventStore.Internal.Generator import Database.EventStore.Internal.Operation hiding (SM(..)) import Database.EventStore.Internal.Types  -------------------------------------------------------------------------------- import qualified Database.EventStore.Internal.Manager.Operation.Model as Op-import qualified Database.EventStore.Internal.Manager.Subscription    as Sub+import qualified Database.EventStore.Internal.Manager.Subscription.Driver as Sub+import qualified Database.EventStore.Internal.Manager.Subscription.Model as Sub  -------------------------------------------------------------------------------- -- | Type of inputs handled by the 'Processor' driver.@@ -217,6 +219,7 @@       -- ^ Indicates to send the given 'Package'.     | Await (Processor r)       -- ^ Waits for more input.+    | ForceReconnectCmd NodeEndPoints (Transition r)  -------------------------------------------------------------------------------- -- | Processor state-machine.@@ -230,6 +233,9 @@     Transmit pkg (loopOpTransition st nxt) loopOpTransition st (Op.Await m) =     let nxt_st = st { _opModel = m } in Await $ Processor $ execute nxt_st+loopOpTransition st (Op.NotHandled info nxt) =+    let node = masterInfoNodeEndPoints info in+    ForceReconnectCmd node (loopOpTransition st nxt)  -------------------------------------------------------------------------------- abortTransition :: State r -> Op.Transition r -> [r] -> Transition r

@@ -0,0 +1,794 @@+{-# LANGUAGE DeriveDataTypeable  #-}+{-# LANGUAGE GADTs               #-}+{-# LANGUAGE OverloadedStrings   #-}+{-# LANGUAGE Rank2Types          #-}+{-# LANGUAGE RecordWildCards     #-}+{-# LANGUAGE ScopedTypeVariables #-}+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Internal.Subscription+-- Copyright : (C) 2015 Yorick Laupa+-- License : (see the file LICENSE)+--+-- Maintainer : Yorick Laupa <yo.eight@gmail.com>+-- Stability : provisional+-- Portability : non-portable+--+-- Main subscription state machine declaration module. It also declares every+-- functions required to drive a 'Subscription'.+--------------------------------------------------------------------------------+module Database.EventStore.Internal.Subscription+    ( Regular(..)+    , Persistent(..)+    , Catchup(..)+    , CatchupParams(..)+    , Running(..)+    , Checkpoint(..)+    , Subscription+    , SubscriptionId+    , SubscriptionClosed(..)+    , SubEnv(..)+    , PushCmd(..)+    , AckCmd(..)+    , catchupSub+    , regularSub+    , persistentSub+    , hasCaughtUp+    , getSubId+    , getSubStream+    , unsubscribe+    , isSubscribedToAll+    , getSubLastCommitPos+    , getSubLastEventNumber+    , nextEventMaybeSTM+    , nextEvent+    , nextEventMaybe+    , waitConfirmation+    , getSubResolveLinkTos+    , waitTillCatchup+    , hasCaughtUpSTM+    , notifyEventsProcessed+    , acknowledge+    , acknowledgeEvents+    , failed+    , eventsFailed+    , notifyEventsFailed+    , unsubscribeConfirmed+    , waitUnsubscribeConfirmed+    , unsubscribeConfirmedSTM+    ) where++--------------------------------------------------------------------------------+import Data.Int++--------------------------------------------------------------------------------+import ClassyPrelude+import Data.Sequence (ViewL(..), viewl, dropWhileL)+import Data.UUID++--------------------------------------------------------------------------------+import Database.EventStore.Internal.EndPoint+import Database.EventStore.Internal.Manager.Subscription.Driver hiding (unsubscribe)+import Database.EventStore.Internal.Manager.Subscription.Model+import Database.EventStore.Internal.Operation+import Database.EventStore.Internal.Operation.Catchup+import Database.EventStore.Internal.Types++--------------------------------------------------------------------------------+-- | Also referred as volatile subscription. For example, if a stream has 100+--   events in it when a subscriber connects, the subscriber can expect to see+--   event number 101 onwards until the time the subscription is closed or+--   dropped.+data Regular = Regular { _subTos :: Bool }++--------------------------------------------------------------------------------+-- | This kind of subscription specifies a starting point, in the form of an+--   event number or transaction file position. The given function will be+--   called for events from the starting point until the end of the stream, and+--   then for subsequently written events.+--+--   For example, if a starting point of 50 is specified when a stream has 100+--   events in it, the subscriber can expect to see events 51 through 100, and+--   then any events subsequently written until such time as the subscription is+--   dropped or closed.+data Catchup = Catchup++--------------------------------------------------------------------------------+-- | The server remembers the state of the subscription. This allows for many+--   different modes of operations compared to a regular or catchup subscription+--   where the client holds the subscription state.+--   (Need EventStore >= v3.1.0).+data Persistent = Persistent { _perGroup  :: Text }++--------------------------------------------------------------------------------+-- | Represents the different type of inputs a subscription state-machine can+--   handle.+data Input t a where+    -- A event has written to the stream. Subscription state machine should+    -- store that event withing its state.+    Arrived :: ResolvedEvent -> Input t (SubStateMachine t)+    -- The user asks for the next event coming from the server.+    ReadNext :: Input t (Maybe (ResolvedEvent, SubStateMachine t))+    -- A batch read has been made. It's only use for 'Catchup' subscription+    -- type. It gives the list of read events and indicates if it reaches the+    -- end of the stream along with the next checkpoint to point at.+    BatchRead :: [ResolvedEvent]+              -> Bool+              -> Checkpoint+              -> Input Catchup (SubStateMachine Catchup)+    -- Used only for 'Catchup' subscription type. Asks if the subscription+    -- read every events up to the checkpoint given by the user.+    CaughtUp :: Input Catchup Bool++    -- Returns the last event number read by the user.+    LastEventNum :: Input Catchup (Int32, Maybe Position)++--------------------------------------------------------------------------------+-- | Main subscription state machine.+newtype SubStateMachine t = SubStateMachine (forall a. Input t a -> a)++--------------------------------------------------------------------------------+type PushOp a = (Either OperationError a -> IO ()) -> Operation a -> IO ()++--------------------------------------------------------------------------------+type PushConnect = (SubConnectEvent -> IO ()) -> PushCmd -> IO ()++--------------------------------------------------------------------------------+data PushCmd+    = PushRegular Text Bool+    | PushPersistent Text Text Int32++--------------------------------------------------------------------------------+data AckCmd = AckCmd | NakCmd NakAction (Maybe Text)++--------------------------------------------------------------------------------+data SubEnv =+    SubEnv { subSettings :: Settings+           , subPushOp :: forall a. PushOp a+           , subPushConnect :: PushConnect+           , subPushUnsub :: Running -> IO ()+           , subAckCmd :: AckCmd -> Running -> [UUID] -> IO ()+           , subForceReconnect :: NodeEndPoints -> IO ()+           }++--------------------------------------------------------------------------------+data SubState t+    = SubOnline (SubStateMachine t)+    | SubDropped SubDropReason+    | forall e. Exception e => SubException e+    | SubUserUnsubscribed++--------------------------------------------------------------------------------+-- | Modifies 'SubState' internal state machine, letting any 'SubDropReason'+--   untouched.+modifySubSM :: (SubStateMachine a -> SubStateMachine a)+            -> SubState a+            -> SubState a+modifySubSM k (SubOnline sm) = SubOnline (k sm)+modifySubSM _ s = s++--------------------------------------------------------------------------------+data CatchupParams =+    CatchupParams { catchupResLnkTos :: !Bool+                  , catchupState :: !CatchupState+                  , catchupBatchSize :: !(Maybe Int32)+                  }++--------------------------------------------------------------------------------+-- | Represents a subscription life cycle.+data SubLifeCycle a =+    SubLifeCycle+    {+      onConfirm :: Running -> IO ()+      -- ^ When the server's confirmed this subscription's been created.+    , readState :: STM (SubState a)+      -- ^ Reads this subscription internal state.+    , writeState :: SubState a -> STM ()+      -- ^ Modifies this subscription internal state.+    , onError :: SubDropReason -> IO ()+      -- ^ When an error's occured.+    , onUserUnsubscribed :: IO ()+      -- ^ When the server confirmed the subscription is no longer live.+      --   This action is triggered because the user asks to unsubscribe.+    , retrySub :: IO ()+     -- ^ Retry the all subscription, this behavior is transparent to the user.+    }++--------------------------------------------------------------------------------+-- | It's possible to subscribe to a stream and be notified when new events are+--   written to that stream. There are three types of subscription which are+--   available, all of which can be useful in different situations.+--+--     * 'Regular'+--+--     * 'Catchup'+--+--     * 'Persistent'+data Subscription t =+    Subscription { subStream :: Text+                 , subLifeCycle :: SubLifeCycle t+                 , subEnv :: SubEnv+                 , subRun :: TMVar Running+                 , subType :: t+                 }++--------------------------------------------------------------------------------+-- | This exception is raised when the user tries to get the next event from a+--   'Subscription' that is already closed.+data SubscriptionClosed+    = SubscriptionClosed SubDropReason+    | SubscriptionUnsubscribedByUser+    deriving (Show, Typeable)++--------------------------------------------------------------------------------+instance Exception SubscriptionClosed++--------------------------------------------------------------------------------+catchupSub :: SubEnv -> CatchupParams -> IO (Subscription Catchup)+catchupSub env params = do+    mvarRun <- newEmptyTMVarIO+    mvarState <- newEmptyTMVarIO++    let streamId = catchupStreamName $ catchupState params+        pushCmd = PushRegular streamId (catchupResLnkTos params)+        lcycle =+            SubLifeCycle+            { onConfirm = atomically . putTMVar mvarRun+            , readState = readTMVar mvarState+            , writeState = \s -> () <$ swapTMVar mvarState s+            , onError = \e ->+                case e of+                    SubAborted ->+                        tryRetryCatcupSubscription pushCmd env mvarState+                            lcycle params+                    SubNotHandled reason infoM ->+                        subNotHandledMsg env lcycle reason infoM+                    _ -> atomically $ do+                        s <- takeTMVar mvarState+                        case s of+                            SubOnline{} -> putTMVar mvarState $ SubDropped e+                            _ -> putTMVar mvarState s+            , onUserUnsubscribed = atomically $ do+                  _ <- takeTMVar mvarState+                  putTMVar mvarState SubUserUnsubscribed+            , retrySub = tryRetryCatcupSubscription pushCmd env mvarState+                             lcycle params+            }++        op = createCatchupOperation env params++    subPushOp env (catchupOpEventHandler mvarState) op+    subPushConnect env (subEventHandler lcycle) pushCmd+    return $ Subscription streamId lcycle env mvarRun Catchup++--------------------------------------------------------------------------------+tryRetryCatcupSubscription :: PushCmd+                           -> SubEnv+                           -> TMVar (SubState Catchup)+                           -> SubLifeCycle Catchup+                           -> CatchupParams+                           -> IO ()+tryRetryCatcupSubscription pushCmd env mvarState lcycle params = do+    state <- atomically $ takeTMVar mvarState++    case state of+        -- In this case, we do our best to re-engage the+        -- catchup subscription where it was at before+        -- losing the connection with the server.+        SubOnline sm -> do+            let (num, posM) = lastEventNumSM sm+                newStart =+                  case catchupState params of+                      RegularCatchup stream _ ->+                          RegularCatchup stream num+                      AllCatchup{} ->+                          case posM of+                              Just (Position npc npp) ->+                                  AllCatchup npc npp+                              _ -> catchupState params++                newParams = params { catchupState = newStart }++                newOp = createCatchupOperation env newParams++            subPushOp env (catchupOpEventHandler mvarState)+                newOp+            subPushConnect env (subEventHandler lcycle)+                pushCmd+        _ -> atomically $ putTMVar mvarState state++--------------------------------------------------------------------------------+regularSub :: SubEnv -> Text -> Bool -> IO (Subscription Regular)+regularSub env streamId resLnkTos = do+    mvarRun <- newEmptyTMVarIO+    varState <- newTVarIO $ SubOnline regularSubscription+    let lcycle =+            SubLifeCycle+            { onConfirm = atomically . putTMVar mvarRun+            , readState = readTVar varState+            , writeState = writeTVar varState+            , onError = \r -> atomically $ do+                  s <- readTVar varState+                  case s of+                      SubOnline{} -> writeTVar varState $ SubDropped r+                      _ -> return ()+            , onUserUnsubscribed =+                  atomically $ writeTVar varState SubUserUnsubscribed+            , retrySub = do+                  atomically $ do+                      state <- readState lcycle+                      case state of+                          SubOnline{} -> return ()+                          _ -> writeState lcycle $ SubOnline regularSubscription+                  subPushConnect env (subEventHandler lcycle)+                                     (PushRegular streamId resLnkTos)+            }++    subPushConnect env (subEventHandler lcycle) (PushRegular streamId resLnkTos)+    return $ Subscription streamId lcycle env mvarRun (Regular resLnkTos)++--------------------------------------------------------------------------------+persistentSub :: SubEnv -> Text -> Text -> Int32 -> IO (Subscription Persistent)+persistentSub env grp stream bufSize = do+    mvarRun <- newEmptyTMVarIO+    varState <- newTVarIO $ SubOnline persistentSubscription+    let lcycle =+            SubLifeCycle+            { onConfirm = atomically . putTMVar mvarRun+            , readState = readTVar varState+            , writeState = writeTVar varState+            , onError = \r -> atomically $ do+                  s <- readTVar varState+                  case s of+                      SubOnline{} -> writeTVar varState $ SubDropped r+                      _ -> return ()+            , onUserUnsubscribed =+                  atomically $ writeTVar varState SubUserUnsubscribed+            , retrySub = do+                  atomically $ writeState lcycle+                             $ SubOnline persistentSubscription+                  subPushConnect env (subEventHandler lcycle) pushCmd+            }++        pushCmd = PushPersistent grp stream bufSize++    subPushConnect env (subEventHandler lcycle) pushCmd+    return $ Subscription stream lcycle env mvarRun (Persistent grp)++--------------------------------------------------------------------------------+subNotHandledMsg :: SubEnv+                 -> SubLifeCycle s+                 -> NotHandledReason+                 -> Maybe MasterInfo+                 -> IO ()+subNotHandledMsg env _ N_NotMaster (Just info) =+    subForceReconnect env $ masterInfoNodeEndPoints info+subNotHandledMsg _ lcycle N_NotMaster _ =+    atomically $ writeState lcycle+               $ SubDropped $ SubServerError (Just msg)+  where+    msg = "Been asked to connect to new master node \+          \ but no master info been sent."+subNotHandledMsg _ lcycle N_NotReady _ = retrySub lcycle+subNotHandledMsg _ lcycle N_TooBusy _ = retrySub lcycle++--------------------------------------------------------------------------------+createCatchupOperation :: SubEnv -> CatchupParams -> Operation CatchupOpResult+createCatchupOperation env params =+  catchup (subSettings env)+          (catchupState params)+          (catchupResLnkTos params)+          (catchupBatchSize params)++--------------------------------------------------------------------------------+-- | We want to notify the user that something went wrong in the first phase of+--   a catchup subscription (e.g. reading the stream forward until we catchup to+--   stream's end). This prevents a deadlock on user side in case where the user+--   calls `waitTillCatchup` on a stream that doesn't exist.+catchupOpEventHandler :: TMVar (SubState Catchup)+                      -> Either OperationError CatchupOpResult+                      -> IO ()+catchupOpEventHandler mvarState (Left e) = atomically $ do+    isEmpty <- isEmptyTMVar mvarState+    if isEmpty+        then putTMVar mvarState (SubException e)+        else () <$ swapTMVar mvarState (SubException e)+catchupOpEventHandler mvarState (Right res) = atomically $ do+    -- When a catchup subscription receives events for the+    -- first time.+    whenM (isEmptyTMVar mvarState) $ do+        let initState = SubOnline catchupSubscription+        putTMVar mvarState initState++    subState <- takeTMVar mvarState+    let cmd = batchReadSM (catchupReadEvents res)+                          (catchupEndOfStream res)+                          (catchupCheckpoint res)++        nxtSubState = modifySubSM cmd subState++    putTMVar mvarState nxtSubState++--------------------------------------------------------------------------------+-- | Subscription event handler. Used during a subscription lifetime.+subEventHandler :: SubLifeCycle a -> SubConnectEvent -> IO ()+subEventHandler lcycle (SubConfirmed run) = onConfirm lcycle run+subEventHandler lcycle (EventAppeared e) = atomically $ do+    st <- readState lcycle+    case st of+        SubOnline sm ->+            writeState lcycle $ SubOnline $ eventArrivedSM e sm+        SubException _ ->+            -- At this moment [07 October 2016], this can only happen during+            -- the first phase of a catchup subscription where the user+            -- asked for a subscription on a stream that doesn't exist.+            return ()+        _ -> error "Impossible: subEventHandler"+subEventHandler lcycle Unsubscribed = onUserUnsubscribed lcycle+subEventHandler lcycle (Dropped r) = onError lcycle r++--------------------------------------------------------------------------------+-- | Submit a new event to the subscription state machine. Internally,+--   that event should be stored into the subscription buffer.+eventArrivedSM :: ResolvedEvent -> SubStateMachine t -> SubStateMachine t+eventArrivedSM e (SubStateMachine k) = k (Arrived e)++--------------------------------------------------------------------------------+-- | Reads the next available event. Returns 'Nothing' it there is any. When+--   returning an event, it will be removed from the subscription buffer.+readNextSM :: SubStateMachine t -> Maybe (ResolvedEvent, SubStateMachine t)+readNextSM (SubStateMachine k) = k ReadNext++--------------------------------------------------------------------------------+-- | Submits a list of events read from a stream. It's only used by a 'Catchup'+--   subscription.+batchReadSM :: [ResolvedEvent]+            -> Bool -- ^ If it reaches the end of the stream.+            -> Checkpoint+            -> SubStateMachine Catchup+            -> SubStateMachine Catchup+batchReadSM es eos nxt (SubStateMachine k) = k (BatchRead es eos nxt)++--------------------------------------------------------------------------------+-- | Indicates if the subscription caught up the end of the stream, meaning the+--   subscription is actually live. Only used by 'Catchup' subscription.+hasCaughtUpSM :: SubStateMachine Catchup -> Bool+hasCaughtUpSM (SubStateMachine k) = k CaughtUp++--------------------------------------------------------------------------------+-- | Last event number read by the user.+lastEventNumSM :: SubStateMachine Catchup -> (Int32, Maybe Position)+lastEventNumSM (SubStateMachine k) = k LastEventNum++--------------------------------------------------------------------------------+-- | Main 'Regular' subscription state machine.+regularSubscription :: SubStateMachine Regular+regularSubscription = baseSubStateMachine++--------------------------------------------------------------------------------+-- | Main 'Persistent' subscription state machine.+persistentSubscription :: SubStateMachine Persistent+persistentSubscription = baseSubStateMachine++--------------------------------------------------------------------------------+-- | Depending either if the subscription concerns a regular stream or $all,+--  indicates if an event number (or 'Position') is lesser that the current the+--  given 'CheckPoint'.+beforeChk :: Checkpoint -> ResolvedEvent -> Bool+beforeChk (CheckpointNumber num) re =+    recordedEventNumber (resolvedEventOriginal re) < num+beforeChk (CheckpointPosition pos) re =+    maybe False (< pos) $ resolvedEventPosition re++--------------------------------------------------------------------------------+-- | This data structure is only used by catchup subscription state machine.+data CatchupSMState =+    CatchupSMState { csmReadSeq :: !(Seq ResolvedEvent)+                   -- ^ This sequence is used to pack events coming from reading+                   --   a stream forward.+                   , csmLiveSeq :: !(Seq ResolvedEvent)+                   -- ^ This sequence is used to pack events coming from live+                   --   subscription.+                   , csmLastNum :: !(Maybe Int32)+                   -- ^ Tracks the last event read.+                   , csmLastPos :: !(Maybe Position)+                   }++--------------------------------------------------------------------------------+initialCatchupSMState :: CatchupSMState+initialCatchupSMState = CatchupSMState empty empty Nothing Nothing++--------------------------------------------------------------------------------+insertReadEvents :: [ResolvedEvent]+                 -> Checkpoint+                 -> CatchupSMState+                 -> CatchupSMState+insertReadEvents es chp s =+    s { csmReadSeq = foldl' snoc (csmReadSeq s) es+      , csmLiveSeq = dropWhileL (beforeChk chp) (csmLiveSeq s)+      }++--------------------------------------------------------------------------------+insertLiveEvent :: ResolvedEvent -> CatchupSMState -> CatchupSMState+insertLiveEvent e s = s { csmLiveSeq = csmLiveSeq s `snoc` e }++--------------------------------------------------------------------------------+readNextFromBatchSeq :: CatchupSMState -> Maybe (ResolvedEvent, CatchupSMState)+readNextFromBatchSeq s =+    case viewl $ csmReadSeq s of+        EmptyL -> Nothing+        e :< rest ->+            let newLast = recordedEventNumber $ resolvedEventOriginal e+                nxtS = s { csmReadSeq = rest+                         , csmLastNum = Just newLast+                         , csmLastPos = resolvedEventPosition e+                         } in+            Just (e, nxtS)++--------------------------------------------------------------------------------+readNextFromLiveSeq :: CatchupSMState -> Maybe (ResolvedEvent, CatchupSMState)+readNextFromLiveSeq s =+    case viewl $ csmLiveSeq s of+        EmptyL -> Nothing+        e :< rest ->+            let newLast = recordedEventNumber $ resolvedEventOriginal e+                nxtS = s { csmLiveSeq = rest+                         , csmLastNum = Just newLast+                         , csmLastPos = resolvedEventPosition e+                         } in+            Just (e, nxtS)++--------------------------------------------------------------------------------+lastEventNumber :: CatchupSMState -> (Int32, Maybe Position)+lastEventNumber s = (fromMaybe 0 $ csmLastNum s, csmLastPos s)++--------------------------------------------------------------------------------+isBatchReqEmpty :: CatchupSMState -> Bool+isBatchReqEmpty s =+    case viewl $ csmReadSeq s of+        EmptyL -> True+        _ -> False++--------------------------------------------------------------------------------+-- | That subscription state machine accumulates events coming from batch read+--   and any real time change made on a stream. That state machine will not+--   served any recent change made on the stream until it reaches the end of the+--   stream. On every batch read, it makes sure events contained in that batch+--   are deleted from the subscription buffer in order to avoid duplicates. That+--   implemention has been chosen to avoid potential message lost between the+--   moment with reach the end of the stream and the delay required by asking+--   for a subscription.+catchupSubscription :: SubStateMachine Catchup+catchupSubscription = SubStateMachine $ catchingUp initialCatchupSMState+  where+    catchingUp :: forall a. CatchupSMState -> Input Catchup a -> a+    catchingUp s (Arrived e) =+        SubStateMachine $ catchingUp $ insertLiveEvent e s+    catchingUp s ReadNext =+        let _F (e, sm) = (e, SubStateMachine $ catchingUp sm) in+        fmap _F $ readNextFromBatchSeq s+    catchingUp s (BatchRead es eos chk) =+        let nxtS = insertReadEvents es chk s in+        SubStateMachine $+            if eos+            then caughtUp nxtS+            else catchingUp nxtS+    catchingUp _ CaughtUp = False+    catchingUp s LastEventNum = lastEventNumber s++    caughtUp :: forall a. CatchupSMState -> Input Catchup a -> a+    caughtUp s (Arrived e) = SubStateMachine $ caughtUp $ insertLiveEvent e s+    caughtUp s ReadNext =+        case readNextFromBatchSeq s of+            Nothing -> live s ReadNext+            Just (e, nxtS) ->+                if isBatchReqEmpty nxtS+                then Just (e, SubStateMachine $ live s)+                else Just (e, SubStateMachine $ caughtUp nxtS)+    caughtUp s BatchRead{} = SubStateMachine $ caughtUp s+    caughtUp _ CaughtUp = False+    caughtUp s LastEventNum = lastEventNumber s++    live :: forall a. CatchupSMState -> Input Catchup a -> a+    live s (Arrived e) = SubStateMachine $ live $ insertLiveEvent e s+    live s ReadNext =+        let _F (e, sm) = (e, SubStateMachine $ live sm) in+        fmap _F $ readNextFromLiveSeq s+    live s BatchRead{} = SubStateMachine $ live s+    live _ CaughtUp = True+    live s LastEventNum = lastEventNumber s++--------------------------------------------------------------------------------+-- | Base subscription used for 'Regular' or 'Persistent' subscription.+baseSubStateMachine :: forall t. SubStateMachine t+baseSubStateMachine = SubStateMachine $ go empty+  where+    go :: forall a. Seq ResolvedEvent -> Input t a -> a+    go s (Arrived e) = SubStateMachine $ go (s `snoc` e)+    go s ReadNext =+        case viewl s of+            EmptyL    -> Nothing+            e :< rest -> Just (e, SubStateMachine $ go rest)+    go _ _ = error "impossible: base subscription"++--------------------------------------------------------------------------------+-- Subscription API+--------------------------------------------------------------------------------+-- | Represents a subscription id.+newtype SubscriptionId = SubId UUID deriving (Eq, Ord, Show)++--------------------------------------------------------------------------------+-- | Gets the ID of the subscription.+getSubId :: Subscription a -> IO SubscriptionId+getSubId Subscription{..} = atomically $ do+    run <- readTMVar subRun+    return $ SubId $ runningUUID run++--------------------------------------------------------------------------------+-- | Gets the subscription stream name.+getSubStream :: Subscription a -> Text+getSubStream = subStream++--------------------------------------------------------------------------------+-- | Asynchronously unsubscribe from the the stream.+unsubscribe :: Subscription a -> IO ()+unsubscribe Subscription{..} = do+    run <- atomically $ readTMVar subRun+    subPushUnsub subEnv run++--------------------------------------------------------------------------------+-- | If the subscription is on the $all stream.+isSubscribedToAll :: Subscription a -> Bool+isSubscribedToAll = (== "") . getSubStream++--------------------------------------------------------------------------------+-- | The last commit position seen on the subscription (if this a subscription+--   to $all stream).+getSubLastCommitPos :: Subscription a -> IO Int64+getSubLastCommitPos Subscription{..} = atomically $ do+    run <- readTMVar subRun+    return $ runningLastCommitPosition run++--------------------------------------------------------------------------------+-- | The last event number seen on the subscription (if this is a subscription+--   to a single stream).+getSubLastEventNumber :: Subscription a -> IO (Maybe Int32)+getSubLastEventNumber Subscription{..} = atomically $ do+    run <- readTMVar subRun+    return $ runningLastEventNumber run++--------------------------------------------------------------------------------+-- | Asks for the next incoming event like 'nextEventMaybe' while still being+--   in the the 'STM'.+nextEventMaybeSTM :: Subscription a -> STM (Maybe ResolvedEvent)+nextEventMaybeSTM Subscription{..} = do+    st <- readState subLifeCycle+    case st of+        SubException e -> throwSTM e+        SubDropped r -> throwSTM $ SubscriptionClosed r+        SubOnline sub -> do+            case readNextSM sub of+                Just (e, nxt) ->+                    Just e <$ writeState subLifeCycle (SubOnline nxt)+                _ -> return Nothing+        SubUserUnsubscribed -> throwSTM SubscriptionUnsubscribedByUser++--------------------------------------------------------------------------------+-- | Awaits for the next event.+nextEvent :: Subscription a -> IO ResolvedEvent+nextEvent sub = atomically $ do+    m <- nextEventMaybeSTM sub+    case m of+        Nothing -> retrySTM+        Just e  -> return e++--------------------------------------------------------------------------------+-- | Non blocking version of 'nextEvent'.+nextEventMaybe :: Subscription a -> IO (Maybe ResolvedEvent)+nextEventMaybe = atomically . nextEventMaybeSTM++--------------------------------------------------------------------------------+-- | Waits until the `Subscription` has been confirmed.+waitConfirmation :: Subscription a -> IO ()+waitConfirmation s = atomically $ do+    _ <- readTMVar $ subRun s+    return ()++--------------------------------------------------------------------------------+-- | Determines whether or not any link events encontered in the stream will be+--   resolved.+getSubResolveLinkTos :: Subscription Regular -> Bool+getSubResolveLinkTos = _subTos . subType++--------------------------------------------------------------------------------+-- | Non blocking version of `waitTillCatchup`.+hasCaughtUp :: Subscription Catchup -> IO Bool+hasCaughtUp sub = atomically $ hasCaughtUpSTM sub++--------------------------------------------------------------------------------+-- | Waits until 'CatchupSubscription' subscription catch-up its stream.+waitTillCatchup :: Subscription Catchup -> IO ()+waitTillCatchup sub = atomically $+    unlessM (hasCaughtUpSTM sub) retrySTM++--------------------------------------------------------------------------------+-- | Like 'hasCaughtUp' but lives in 'STM' monad.+hasCaughtUpSTM :: Subscription Catchup -> STM Bool+hasCaughtUpSTM Subscription{..} = do+    res <- readState subLifeCycle+    case res of+        SubOnline sm -> return $ hasCaughtUpSM sm+        SubException e -> throwSTM e+        SubDropped r -> throwSTM $ SubscriptionClosed r+        SubUserUnsubscribed -> throwSTM SubscriptionUnsubscribedByUser++--------------------------------------------------------------------------------+-- | Like 'unsubscribeConfirmed' but lives in 'STM' monad.+unsubscribeConfirmedSTM :: Subscription a -> STM Bool+unsubscribeConfirmedSTM Subscription{..} = do+    res <- readState subLifeCycle+    case res of+        SubOnline _ -> return False+        SubException e -> throwSTM e+        SubDropped r -> throwSTM $ SubscriptionClosed r+        SubUserUnsubscribed -> return True++--------------------------------------------------------------------------------+-- | Non blocking version of `waitUnsubscribeConfirmed`.+unsubscribeConfirmed :: Subscription a -> IO Bool+unsubscribeConfirmed = atomically . unsubscribeConfirmedSTM++--------------------------------------------------------------------------------+-- | Wait until unsubscription has been confirmed by the server.+waitUnsubscribeConfirmed :: Subscription a -> IO ()+waitUnsubscribeConfirmed sub = atomically $+    unlessM (unsubscribeConfirmedSTM sub) retrySTM++--------------------------------------------------------------------------------+-- | Acknowledges those event ids have been successfully processed.+notifyEventsProcessed :: Subscription Persistent -> [UUID] -> IO ()+notifyEventsProcessed Subscription{..} evts = do+    run <- atomically $ readTMVar subRun+    subAckCmd subEnv AckCmd run evts++--------------------------------------------------------------------------------+-- | Acknowledges that 'ResolvedEvent' has been successfully processed.+acknowledge :: Subscription Persistent -> ResolvedEvent -> IO ()+acknowledge sub e = notifyEventsProcessed sub [resolvedEventOriginalId e]++--------------------------------------------------------------------------------+-- | Acknowledges those 'ResolvedEvent's have been successfully processed.+acknowledgeEvents :: Subscription Persistent -> [ResolvedEvent] -> IO ()+acknowledgeEvents sub = notifyEventsProcessed sub . fmap resolvedEventOriginalId++--------------------------------------------------------------------------------+-- | Mark a message that has failed processing. The server will take action+--   based upon the action parameter.+failed :: Subscription Persistent+       -> ResolvedEvent+       -> NakAction+       -> Maybe Text+       -> IO ()+failed sub e a r = notifyEventsFailed sub a r [resolvedEventOriginalId e]++--------------------------------------------------------------------------------+-- | Mark messages that have failed processing. The server will take action+--   based upon the action parameter.+eventsFailed :: Subscription Persistent+             -> [ResolvedEvent]+             -> NakAction+             -> Maybe Text+             -> IO ()+eventsFailed sub evts a r =+    notifyEventsFailed sub a r $ fmap resolvedEventOriginalId evts++--------------------------------------------------------------------------------+-- | Acknowledges those event ids have failed to be processed successfully.+notifyEventsFailed :: Subscription Persistent+                   -> NakAction+                   -> Maybe Text+                   -> [UUID]+                   -> IO ()+notifyEventsFailed Subscription{..} act res evts = do+    run <- atomically $ readTMVar subRun+    subAckCmd subEnv (NakCmd act res) run evts

@@ -36,6 +36,7 @@  -------------------------------------------------------------------------------- import Database.EventStore.Internal.Command+import Database.EventStore.Internal.EndPoint import Database.EventStore.Logging  --------------------------------------------------------------------------------@@ -57,18 +58,61 @@ -------------------------------------------------------------------------------- -- Event --------------------------------------------------------------------------------+-- | Constants for System event types.+data EventType+    = StreamDeletedType+      -- ^ Event type for stream deleted.+    | StatsCollectionType+      -- ^ Event type for statistics.+    | LinkToType+      -- ^ Event type for linkTo.+    | StreamMetadataType+      -- ^ Event type for stream metadata.+    | SettingsType+      -- ^ Event type for the system settings.+    | UserDefined Text+      -- ^ Event defined by the user.+    deriving Eq++--------------------------------------------------------------------------------+instance IsString EventType where+    fromString = eventTypeFromStr++--------------------------------------------------------------------------------+instance Show EventType where+    show = unpack . eventTypeText++--------------------------------------------------------------------------------+eventTypeText :: EventType -> Text+eventTypeText StreamDeletedType = "$streamDeleted"+eventTypeText StatsCollectionType = "$statsCollected"+eventTypeText LinkToType = "$>"+eventTypeText StreamMetadataType = "$metadata"+eventTypeText SettingsType = "$settings"+eventTypeText (UserDefined t) = t++--------------------------------------------------------------------------------+eventTypeFromStr :: String -> EventType+eventTypeFromStr "$streamDeleted" = StreamDeletedType+eventTypeFromStr "$statsCollected" = StatsCollectionType+eventTypeFromStr "$>" = LinkToType+eventTypeFromStr "$metadata" = StreamMetadataType+eventTypeFromStr "$settings" = SettingsType+eventTypeFromStr t = UserDefined $ pack t++-------------------------------------------------------------------------------- -- | Contains event information like its type and data. Only used for write --   queries. data Event     = Event-      { eventType :: !Text+      { eventType :: !EventType       , eventId   :: !(Maybe UUID)       , eventData :: !EventData       } deriving (Eq, Show)  -------------------------------------------------------------------------------- -- | Create an 'Event' meant to be persisted.-createEvent :: Text       -- ^ Event type+createEvent :: EventType  -- ^ Event type             -> Maybe UUID -- ^ Event ID, generated if 'Nothing'             -> EventData  -- ^ Event data             -> Event@@ -259,6 +303,78 @@ instance Decode ResolvedIndexedEvent  --------------------------------------------------------------------------------+data NotHandledReason+    = N_NotReady+    | N_TooBusy+    | N_NotMaster+    deriving (Enum, Eq, Show)++--------------------------------------------------------------------------------+data MasterInfoBuf+    = MasterInfoBuf+      { bufMasterExternalTcpAddr :: Required 1 (Value Text)+      , bufMasterExternalTcpPort :: Required 2 (Value Int32)+      , bufMasterExternalHttpAddr :: Required 3 (Value Text)+      , bufMasterExternalHttpPort :: Required 4 (Value Int32)+      , bufMasterExternalSecureTcpAddr :: Optional 5 (Value Text)+      , bufMasterExternalSecureTcpPort :: Optional 6 (Value Int32)+      } deriving (Generic, Show)++--------------------------------------------------------------------------------+instance Decode MasterInfoBuf++--------------------------------------------------------------------------------+data MasterInfo+    = MasterInfo+      { masterExternalTcpAddr :: !String+      , masterExternalTcpPort :: !Int+      , masterExternalHttpAddr :: !String+      , masterExternalHttpPort :: !Int+      , masterExternalSecureTcpAddr :: !(Maybe String)+      , masterExternalSecureTcpPort :: !(Maybe Int)+      } deriving (Show, Eq)++--------------------------------------------------------------------------------+masterInfo :: MasterInfoBuf -> MasterInfo+masterInfo buf =+    MasterInfo+    { masterExternalTcpAddr =+          unpack $ getField $ bufMasterExternalTcpAddr buf+    , masterExternalTcpPort =+          fromIntegral $ getField $ bufMasterExternalTcpPort buf+    , masterExternalHttpAddr =+          unpack $ getField $ bufMasterExternalHttpAddr buf+    , masterExternalHttpPort =+          fromIntegral $ getField $ bufMasterExternalHttpPort buf+    , masterExternalSecureTcpAddr =+          fmap unpack $ getField $ bufMasterExternalSecureTcpAddr buf+    , masterExternalSecureTcpPort =+          fmap fromIntegral $ getField $ bufMasterExternalSecureTcpPort buf+    }++--------------------------------------------------------------------------------+masterInfoNodeEndPoints :: MasterInfo -> NodeEndPoints+masterInfoNodeEndPoints info =+    NodeEndPoints+        (EndPoint (masterExternalTcpAddr info)+                  (fromIntegral $ masterExternalTcpPort info))+        securePoint+  where+    securePoint =+        EndPoint <$> masterExternalSecureTcpAddr info+                 <*> (fmap fromIntegral $ masterExternalSecureTcpPort info)++--------------------------------------------------------------------------------+data NotHandledBuf+    = NotHandledBuf+      { notHandledReason :: Required 1 (Enumeration NotHandledReason)+      , notHandledAdditionalInfo :: Optional 2 (Message MasterInfoBuf)+      } deriving (Generic, Show)++--------------------------------------------------------------------------------+instance Decode NotHandledBuf++-------------------------------------------------------------------------------- -- | Represents a serialized event sent by the server in a subscription context. data ResolvedEventBuf     = ResolvedEventBuf@@ -489,6 +605,13 @@       , packageCred        :: !(Maybe Credentials)       }     deriving Show++--------------------------------------------------------------------------------+packageDataAsText :: Package -> Maybe Text+packageDataAsText = go . decodeUtf8 . packageData+  where+    go "" = Nothing+    go t  = Just t  -------------------------------------------------------------------------------- -- | Constructs a heartbeat response given the 'UUID' of heartbeat request.

@@ -0,0 +1,27 @@+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Internal.Utils+-- 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.Utils (prettyWord8) where++--------------------------------------------------------------------------------+import Numeric++--------------------------------------------------------------------------------+import ClassyPrelude++--------------------------------------------------------------------------------+prettyWord8 :: Word8 -> String+prettyWord8 w = "0x" ++ padding (showHex w "")++--------------------------------------------------------------------------------+padding :: String -> String+padding [x] = ['0',x]+padding xs  = xs

@@ -5,6 +5,7 @@ [![Build Status](https://travis-ci.org/YoEight/eventstore.svg?branch=master)](https://travis-ci.org/YoEight/eventstore)  That driver supports 100% of EventStore features !+More information about the GetEventStore database can be found there: https://geteventstore.com/  Requirements ============

@@ -8,9 +8,9 @@ -- for standards guiding when and how versions should be incremented. -- http://www.haskell.org/haskellwiki/Package_versioning_policy -- PVP summary:      +-+------- breaking API changes---                   | | +----- non-breaking API additions---                   | | | +--- code changes with no API change-version:             0.13.1.7+--                   | |  +----- non-breaking API additions+--                   | |  | +--- code changes with no API change+version:             0.14.0.0  tested-with: GHC >= 7.8.3 && < 7.11 @@ -65,18 +65,22 @@   other-modules:   Database.EventStore.Internal.Command                    Database.EventStore.Internal.Connection                    Database.EventStore.Internal.Discovery+                   Database.EventStore.Internal.EndPoint                    Database.EventStore.Internal.Execution.Production                    Database.EventStore.Internal.Generator                    Database.EventStore.Internal.Operation                    Database.EventStore.Internal.Processor                    Database.EventStore.Internal.Stream+                   Database.EventStore.Internal.Subscription                    Database.EventStore.Internal.Types+                   Database.EventStore.Internal.Utils                    Database.EventStore.Internal.Manager.Operation.Model-                   Database.EventStore.Internal.Manager.Subscription+                   Database.EventStore.Internal.Manager.Subscription.Command                    Database.EventStore.Internal.Manager.Subscription.Driver                    Database.EventStore.Internal.Manager.Subscription.Message                    Database.EventStore.Internal.Manager.Subscription.Model                    Database.EventStore.Internal.Manager.Subscription.Packages+                   Database.EventStore.Internal.Manager.Subscription.Types                    Database.EventStore.Internal.Operations                    Database.EventStore.Internal.Operation.Catchup                    Database.EventStore.Internal.Operation.DeleteStream@@ -118,6 +122,7 @@                      , dotnet-timespan                      , connection ==0.2.*                      , classy-prelude ==1.*+                     , mtl    -- Directories containing source files.   -- hs-source-dirs:

@@ -237,6 +237,7 @@      loop alljss     unsubscribe sub+    waitUnsubscribeConfirmed sub     let action = do             _ <- nextEvent sub             return False