diff --git a/CHANGELOG.markdown b/CHANGELOG.markdown new file mode 100644 --- /dev/null +++ b/CHANGELOG.markdown @@ -0,0 +1,3 @@ +0.6.0.0 +------- +* Support `keepRetrying` reconnection strategy. diff --git a/Database/EventStore.hs b/Database/EventStore.hs --- a/Database/EventStore.hs +++ b/Database/EventStore.hs @@ -22,6 +22,9 @@ , Connection , Credentials , Settings(..) + , Retry + , atMost + , keepRetrying , credentials , defaultSettings , connect diff --git a/Database/EventStore/Internal/Processor.hs b/Database/EventStore/Internal/Processor.hs --- a/Database/EventStore/Internal/Processor.hs +++ b/Database/EventStore/Internal/Processor.hs @@ -83,10 +83,8 @@ -------------------------------------------------------------------------------- data State = Offline - { _maxAttempt :: !Int } | Online { _uuidCon :: !UUID - , _maxAttempt :: !Int , _packageCount :: !Int , _host :: !HostName , _port :: !Int @@ -94,10 +92,6 @@ } -------------------------------------------------------------------------------- -initState :: Int -> State -initState max_at = Offline max_at - --------------------------------------------------------------------------------- -- Event -------------------------------------------------------------------------------- data Connect = Connect HostName Int @@ -133,12 +127,12 @@ fmap reconnected onReconnected <> fmap received onReceived - stateB <- accum (initState $ s_maxRetries sett) stateE + stateB <- accum Offline stateE let heartbeatP pkg = packageCmd pkg == heartbeatRequestCmd onlyHeartbeats = filterE heartbeatP onReceived - con_snap = snapshot connectSnapshot onConnect stateB + con_snap = fmap connectSnapshot onConnect reco_snap = snapshot reconnectSnapshot onReconnect stateB clean_snap = snapshot cleanupSnapshot onCleanup stateB @@ -153,21 +147,21 @@ push_con_io = pushAsync4 $ \h p u c -> pushConnected $ Connected h p u c - _ <- listen con_snap $ \(ConnectionSnapshot max_a host port) -> - connection push_recv_io + _ <- listen con_snap $ \(ConnectionSnapshot host port) -> + connection sett + push_recv_io (push_con_io host port) push_reco_io onSend - max_a host port - _ <- listen reco_snap $ \(ConnectionSnapshot max_a host port) -> - connection push_recv_io + _ <- listen reco_snap $ \(ConnectionSnapshot host port) -> + connection sett + push_recv_io push_recod_io push_reco_io onSend - max_a host port @@ -191,43 +185,50 @@ -------------------------------------------------------------------------------- data ConnectionSnapshot = ConnectionSnapshot - { _conMax :: !Int - , _conHost :: !HostName + { _conHost :: !HostName , _conPort :: !Int } -------------------------------------------------------------------------------- -connectSnapshot :: Connect -> State -> ConnectionSnapshot -connectSnapshot (Connect host port) s = +connectSnapshot :: Connect -> ConnectionSnapshot +connectSnapshot (Connect host port) = ConnectionSnapshot - { _conMax = _maxAttempt s - , _conHost = host + { _conHost = host , _conPort = port } -------------------------------------------------------------------------------- -reconnectDelay :: Int -reconnectDelay = 500000 +secs :: Int +secs = 1000000 -------------------------------------------------------------------------------- -connection :: (Package -> IO ()) +connection :: Settings + -> (Package -> IO ()) -> (UUID -> IO () -> IO ()) -> IO () -> Event Package - -> Int -> HostName -> Int -> IO () -connection push_pkg push_con push_reco evt_pkg max_a host port = loop 1 +connection sett push_pkg push_con push_reco evt_pkg host port = go where - loop att - | max_a == att = - throwIO $ MaxAttempt host port max_a - | otherwise = - catch (doConnect att) $ \(_ :: SomeException) -> do - threadDelay reconnectDelay - loop (att + 1) + go = + case s_retry sett of + AtMost n -> + let loop i = + catch (doConnect i) $ \(_ :: SomeException) -> do + threadDelay delay + if n <= i then throwIO $ MaxAttempt host port n + else loop (i + 1) in + loop 1 + KeepRetrying -> + let endlessly i = + catch (doConnect i) $ \(_ :: SomeException) -> + threadDelay delay >> endlessly (i + 1) in + endlessly (1 :: Int) + delay = (s_reconnect_delay_secs sett) * secs + doConnect att = do printf "Connecting...Attempt %d\n" att hdl <- connectTo host (PortNumber $ fromIntegral port) @@ -264,9 +265,8 @@ reconnectSnapshot :: Reconnect -> State -> ConnectionSnapshot reconnectSnapshot _ s = ConnectionSnapshot - { _conMax = _maxAttempt s - , _conHost = _host s - , _conPort = _port s + { _conHost = _host s + , _conPort = _port s } -------------------------------------------------------------------------------- @@ -287,10 +287,9 @@ connected :: Connected -> State -> State connected (Connected host port uuid cl) s = case s of - Offline {} + Offline -> Online { _uuidCon = uuid - , _maxAttempt = _maxAttempt s , _packageCount = 0 , _host = host , _port = port diff --git a/Database/EventStore/Internal/Types.hs b/Database/EventStore/Internal/Types.hs --- a/Database/EventStore/Internal/Types.hs +++ b/Database/EventStore/Internal/Types.hs @@ -553,25 +553,45 @@ -------------------------------------------------------------------------------- -- Settings -------------------------------------------------------------------------------- +-- | Represents reconnection strategy. +data Retry + = AtMost Int + | KeepRetrying + +-------------------------------------------------------------------------------- +-- | Indicates how many times we should try to reconnect to the server. A value +-- less than or equal to 0 means no retry. +atMost :: Int -> Retry +atMost = AtMost + +-------------------------------------------------------------------------------- +-- | Indicates we should try to reconnect to the server until the end of the +-- Universe. +keepRetrying :: Retry +keepRetrying = KeepRetrying + +-------------------------------------------------------------------------------- -- | Global 'Connection' settings data Settings = Settings - { s_heartbeatInterval :: NominalDiffTime - , s_heartbeatTimeout :: NominalDiffTime - , s_requireMaster :: Bool - , s_credentials :: Maybe Credentials - , s_maxRetries :: Int + { s_heartbeatInterval :: NominalDiffTime + , s_heartbeatTimeout :: NominalDiffTime + , s_requireMaster :: Bool + , s_credentials :: Maybe Credentials + , s_retry :: Retry + , s_reconnect_delay_secs :: Int -- ^ In seconds } -------------------------------------------------------------------------------- -- | Default global settings. defaultSettings :: Settings defaultSettings = Settings - { s_heartbeatInterval = msDiffTime 750 -- 750ms - , s_heartbeatTimeout = msDiffTime 1500 -- 1500ms - , s_requireMaster = True - , s_credentials = Nothing - , s_maxRetries = 3 + { s_heartbeatInterval = msDiffTime 750 -- 750ms + , s_heartbeatTimeout = msDiffTime 1500 -- 1500ms + , s_requireMaster = True + , s_credentials = Nothing + , s_retry = atMost 3 + , s_reconnect_delay_secs = 3 } -------------------------------------------------------------------------------- diff --git a/eventstore.cabal b/eventstore.cabal --- a/eventstore.cabal +++ b/eventstore.cabal @@ -10,7 +10,7 @@ -- PVP summary: +-+------- breaking API changes -- | | +----- non-breaking API additions -- | | | +--- code changes with no API change -version: 0.5.0.1 +version: 0.6.0.0 -- A short (one-line) description of the package. synopsis: EventStore TCP Client @@ -35,6 +35,7 @@ -- copyright: homepage: http://github.com/YoEight/eventstore +homepage: http://github.com/YoEight/eventstore/issues category: Database build-type: Simple @@ -42,6 +43,7 @@ -- Extra files to be distributed with the package, such as examples or a -- README. extra-source-files: README.md + CHANGELOG.markdown -- Constraint on the version of Cabal needed to build this package. cabal-version: >=1.10