cql-io 0.9.7 → 0.10.1
raw patch · 7 files changed
+521/−76 lines, 7 filesdep +data-default-classdep −retry
Dependencies added: data-default-class
Dependencies removed: retry
Files
- CHANGELOG.md +5/−0
- README.md +3/−0
- cql-io.cabal +30/−26
- src/Control/Retry.hs +287/−0
- src/Database/CQL/IO.hs +28/−8
- src/Database/CQL/IO/Client.hs +90/−39
- src/Database/CQL/IO/Settings.hs +78/−3
CHANGELOG.md view
@@ -1,3 +1,8 @@+0.10.0+-----------------------------------------------------------------------------+- Add `MonadClient` type-class.+- Add retry settings.+ 0.9.7 ----------------------------------------------------------------------------- - Bugfix release.
README.md view
@@ -14,3 +14,6 @@ **Support for connection streams**. Requests can be multiplexed over a few connections.++**Customisable retry settings**. Support for default retry settings as well+as local overrides per query.
cql-io.cabal view
@@ -1,5 +1,5 @@ name: cql-io-version: 0.9.7+version: 0.10.1 synopsis: Cassandra CQL client. stability: experimental license: OtherLicense@@ -21,15 +21,18 @@ which implements Cassandra's CQL protocol and complements it with the neccessary I/O operations. The feature-set includes: .- * __Node discovery__. The driver discovers nodes automatically from a small+ * /Node discovery/. The driver discovers nodes automatically from a small set of bootstrap nodes. .- * __Customisable load-balancing policies__. In addition to pre-built LB+ * /Customisable load-balancing policies/. In addition to pre-built LB policies such as round-robin, users of this library can provide their own policies if desired. .- * __Support for connection streams__. Requests can be multiplexed over a+ * /Support for connection streams/. Requests can be multiplexed over a few connections.+ .+ * /Customisable retry settings/. Support for default retry settings as well+ as local overrides per query. source-repository head type: git@@ -45,6 +48,7 @@ Database.CQL.IO other-modules:+ Control.Retry Database.CQL.IO.Client Database.CQL.IO.Cluster.Discovery Database.CQL.IO.Cluster.Host@@ -62,25 +66,25 @@ Database.CQL.IO.Types build-depends:- async == 2.0.*- , auto-update == 0.1.*- , base >= 4.7 && < 5.0- , bytestring >= 0.10 && < 1.0- , containers >= 0.5 && < 1.0- , cql == 3.0.*- , exceptions >= 0.4 && < 1.0- , hashable >= 1.2 && < 2.0- , iproute >= 1.3 && < 1.4- , lens >= 4.4 && < 4.5- , mtl >= 2.1 && < 2.3- , mwc-random >= 0.13 && < 0.14- , network >= 2.4 && < 3.0- , retry >= 0.5 && < 0.6- , semigroups >= 0.15 && < 0.16- , stm >= 2.4 && < 3.0- , text >= 0.11 && < 2.0- , tinylog >= 0.8 && < 0.11- , time >= 1.4 && < 2.0- , transformers >= 0.3 && < 0.5- , uuid >= 1.2.6 && < 2.0- , vector >= 0.10 && < 1.0+ async == 2.0.*+ , auto-update == 0.1.*+ , base >= 4.7 && < 5.0+ , bytestring >= 0.10 && < 1.0+ , containers >= 0.5 && < 1.0+ , cql == 3.0.*+ , data-default-class+ , exceptions >= 0.4 && < 1.0+ , hashable >= 1.2 && < 2.0+ , iproute >= 1.3 && < 1.4+ , lens >= 4.4 && < 4.5+ , mtl >= 2.1 && < 2.3+ , mwc-random >= 0.13 && < 0.14+ , network >= 2.4 && < 3.0+ , semigroups >= 0.15 && < 0.16+ , stm >= 2.4 && < 3.0+ , text >= 0.11 && < 2.0+ , tinylog >= 0.8 && < 0.11+ , time >= 1.4 && < 2.0+ , transformers >= 0.3 && < 0.5+ , uuid >= 1.2.6 && < 2.0+ , vector >= 0.10 && < 1.0
+ src/Control/Retry.hs view
@@ -0,0 +1,287 @@+-- This is a copy of Ozgun Ataman's retry module version 0.5.1+-- available from http://hackage.haskell.org/package/retry+--+-- It includes changes introduced by pull request 17 in+-- https://github.com/Soostone/retry/pull/17+--+-- If PR 17 is merged into upstream it can be removed again.++{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE ScopedTypeVariables #-}++-----------------------------------------------------------------------------+-- |+-- Module : Control.Retry+-- Copyright : Ozgun Ataman <ozgun.ataman@soostone.com>+-- License : BSD3+--+-- Maintainer : Ozgun Ataman+-- Stability : provisional+--+-- This module exposes combinators that can wrap arbitrary monadic+-- actions. They run the action and potentially retry running it with+-- some configurable delay for a configurable number of times.+--+-- The express purpose of this library is to make it easier to work+-- with IO and especially network IO actions that often experience+-- temporary failure that warrant retrying of the original action. For+-- example, a database query may time out for a while, in which case+-- we should delay a bit and retry the query.+----------------------------------------------------------------------------+++module Control.Retry+ (+ -- * High Level Operation+ RetryPolicy (..)++ , retrying+ , recovering+ , recoverAll++ -- * Retry Policies+ , constantDelay+ , exponentialBackoff+ , fibonacciBackoff+ , limitRetries+ , capDelay++ -- * Re-export from Data.Monoid++ , (<>)++ ) where++-------------------------------------------------------------------------------+import Control.Concurrent+import Control.Monad.Catch+import Control.Monad.IO.Class+import Data.Default.Class+import Data.Monoid+-------------------------------------------------------------------------------+++-------------------------------------------------------------------------------+-- | A 'RetryPolicy' is a function that takes an iteration number and+-- possibly returns a delay in miliseconds. *Nothing* implies we have+-- reached the retry limit.+--+-- Please note that 'RetryPolicy' is a 'Monoid'. You can collapse+-- multiple strategies into one using 'mappend' or '<>'. The semantics+-- of this combination are as follows:+--+-- 1. If either policy returns 'Nothing', the combined policy returns+-- 'Nothing'. This can be used to @inhibit@ after a number of retries,+-- for example.+--+-- 2. If both policies return a delay, the larger delay will be used.+-- This is quite natural when combining multiple policies to achieve a+-- certain effect.+--+-- Example:+--+-- One can easily define an exponential backoff policy with a limited+-- number of retries:+--+-- >> limitedBackoff = exponentialBackoff 50 <> limitedRetries 5+--+-- Naturally, 'mempty' will retry immediately (delay 0) for an+-- unlimited number of retries, forming the identity for the 'Monoid'.+--+-- The default under 'def' implements a constant 50ms delay, up to 5 times:+--+-- >> def = constantDelay 50000 <> limitRetries 5+--+-- For anything more complex, just define your own 'RetryPolicy':+--+-- >> myPolicy = RetryPolicy $ \ n -> if n > 10 then Just 1000 else Just 10000+newtype RetryPolicy = RetryPolicy { getRetryPolicy :: Int -> Maybe Int }+++instance Default RetryPolicy where+ def = constantDelay 50000 <> limitRetries 5++instance Monoid RetryPolicy where+ mempty = RetryPolicy $ (const (Just 0))+ (RetryPolicy a) `mappend` (RetryPolicy b) = RetryPolicy $ \ n -> do+ a' <- a n+ b' <- b n+ return $! max a' b'+++-------------------------------------------------------------------------------+-- | Retry immediately, but only up to @n@ times.+limitRetries+ :: Int+ -- ^ Maximum number of retries.+ -> RetryPolicy+limitRetries i = RetryPolicy $ \ n -> if n >= i then Nothing else (Just 0)+++-------------------------------------------------------------------------------+-- | Implement a constant delay with unlimited retries.+constantDelay+ :: Int+ -- ^ Base delay in microseconds+ -> RetryPolicy+constantDelay delay = RetryPolicy (const (Just delay))+++-------------------------------------------------------------------------------+-- | Grow delay exponentially each iteration.+exponentialBackoff+ :: Int+ -- ^ Base delay in microseconds+ -> RetryPolicy+exponentialBackoff base = RetryPolicy $ \ n -> Just (2^n * base)+++-------------------------------------------------------------------------------+-- | Implement Fibonacci backoff.+fibonacciBackoff+ :: Int+ -- ^ Base delay in microseconds+ -> RetryPolicy+fibonacciBackoff base = RetryPolicy $ \ n -> Just $ fib (n + 1) (0, base)+ where+ fib 0 (a, _) = a+ fib !m (!a, !b) = fib (m-1) (b, a + b)+++-------------------------------------------------------------------------------+-- | Set an upperbound for any delays that may be directed by the+-- given policy.+capDelay+ :: Int+ -- ^ A maximum delay in microseconds+ -> RetryPolicy+ -> RetryPolicy+capDelay limit p = RetryPolicy $ \ n -> min limit `fmap` (getRetryPolicy p) n+++-------------------------------------------------------------------------------+-- | Retry combinator for actions that don't raise exceptions, but+-- signal in their type the outcome has failed. Examples are the+-- 'Maybe', 'Either' and 'EitherT' monads.+--+-- Let's write a function that always fails and watch this combinator+-- retry it 5 additional times following the initial run:+--+-- >>> import Data.Maybe+-- >>> let f = putStrLn "Running action" >> return Nothing+-- >>> retrying def isNothing f+-- Running action+-- Running action+-- Running action+-- Running action+-- Running action+-- Running action+-- Nothing+--+-- Note how the latest failing result is returned after all retries+-- have been exhausted.+retrying :: MonadIO m+ => RetryPolicy+ -> (Int -> b -> m Bool)+ -- ^ An action to check whether the result should be retried.+ -- If True, we delay and retry the operation.+ -> (Int -> m b)+ -- ^ Action to run+ -> m b+retrying (RetryPolicy policy) chk f = go 0+ where+ go n = do+ res <- f n+ chk' <- chk n res+ case chk' of+ True ->+ case (policy n) of+ Just delay -> do+ liftIO (threadDelay delay)+ go $! n+1+ Nothing -> return res+ False -> return res++++-------------------------------------------------------------------------------+-- | Retry ALL exceptions that may be raised. To be used with caution;+-- this matches the exception on 'SomeException'.+--+-- See how the action below is run once and retried 5 more times+-- before finally failing for good:+--+-- >>> let f = putStrLn "Running action" >> error "this is an error"+-- >>> recoverAll def f+-- Running action+-- Running action+-- Running action+-- Running action+-- Running action+-- Running action+-- *** Exception: this is an error+recoverAll :: (MonadIO m, MonadCatch m)+ => RetryPolicy+ -> (Int -> m a)+ -> m a+recoverAll set f = recovering set [h] f+ where+ h _ = Handler $ \ (_ :: SomeException) -> return True+++-------------------------------------------------------------------------------+-- | Run an action and recover from a raised exception by potentially+-- retrying the action a number of times.+recovering :: forall m a. (MonadIO m, MonadCatch m)+ => RetryPolicy+ -- ^ Just use 'def' faor default settings+ -> [(Int -> Handler m Bool)]+ -- ^ Should a given exception be retried? Action will be+ -- retried if this returns True.+ -> (Int -> m a)+ -- ^ Action to perform+ -> m a+recovering (RetryPolicy policy) hs f = go 0+ where++ -- | Convert a (e -> m Bool) handler into (e -> m a) so it can+ -- be wired into the 'catches' combinator.+ transHandler :: Int -> Handler m Bool -> Handler m a+ transHandler n (Handler h) = Handler $ \ e -> do+ chk <- h e+ case chk of+ True ->+ case policy n of+ Just delay -> do+ liftIO (threadDelay delay)+ go $! n+1+ Nothing -> throwM e+ False -> throwM e++ go n = f n `catches` map (transHandler n . ($ n)) hs+++++ ------------------+ -- Simple Tests --+ ------------------++++-- data TestException = TestException deriving (Show, Typeable)+-- data AnotherException = AnotherException deriving (Show, Typeable)++-- instance Exception TestException+-- instance Exception AnotherException+++-- test = retrying def [h1,h2] f+-- where+-- f = putStrLn "Running action" >> throwM AnotherException+-- h1 = Handler $ \ (e :: TestException) -> return False+-- h2 = Handler $ \ (e :: AnotherException) -> return True
src/Database/CQL/IO.hs view
@@ -47,17 +47,33 @@ , setProtocolVersion , setResponseTimeout , setSendTimeout+ , setRetrySettings + -- ** Retry Settings+ , RetrySettings+ , noRetry+ , retryForever+ , maxRetries+ , adjustConsistency+ , constDelay+ , expBackoff+ , fibBackoff+ , adjustSendTimeout+ , adjustResponseTimeout+ -- * Client monad , Client+ , MonadClient (..) , ClientState- , DebugInfo (..)+ , DebugInfo (..) , init , runClient+ , retry , shutdown , debugInfo , query+ , query1 , write , schema , batch@@ -95,7 +111,7 @@ import Control.Applicative import Control.Monad.Catch import Control.Monad (void)-import Data.Maybe (isJust)+import Data.Maybe (isJust, listToMaybe) import Database.CQL.Protocol import Database.CQL.IO.Client import Database.CQL.IO.Cluster.Host@@ -104,7 +120,7 @@ import Database.CQL.IO.Types import Prelude hiding (init) -runQuery :: (Tuple a, Tuple b) => QueryString k a b -> QueryParams a -> Client (Response k a b)+runQuery :: (MonadClient m, Tuple a, Tuple b) => QueryString k a b -> QueryParams a -> m (Response k a b) runQuery q p = do r <- request (RqQuery (Query q p)) case r of@@ -112,19 +128,23 @@ _ -> return r -- | Run a CQL read-only query against a Cassandra node.-query :: (Tuple a, Tuple b) => QueryString R a b -> QueryParams a -> Client [b]+query :: (MonadClient m, Tuple a, Tuple b) => QueryString R a b -> QueryParams a -> m [b] query q p = do r <- runQuery q p case r of RsResult _ (RowsResult _ b) -> return b _ -> throwM UnexpectedResponse +-- | Run a CQL read-only query against a Cassandra node.+query1 :: (MonadClient m, Tuple a, Tuple b) => QueryString R a b -> QueryParams a -> m (Maybe b)+query1 q p = listToMaybe <$> query q p+ -- | Run a CQL insert/update query against a Cassandra node.-write :: Tuple a => QueryString W a () -> QueryParams a -> Client ()+write :: (MonadClient m, Tuple a) => QueryString W a () -> QueryParams a -> m () write q p = void $ runQuery q p -- | Run a CQL schema query against a Cassandra node.-schema :: Tuple a => QueryString S a () -> QueryParams a -> Client (Maybe SchemaChange)+schema :: (MonadClient m, Tuple a) => QueryString S a () -> QueryParams a -> m (Maybe SchemaChange) schema x y = do r <- runQuery x y case r of@@ -133,7 +153,7 @@ _ -> throwM UnexpectedResponse -- | Run a batch query against a Cassandra node.-batch :: Batch -> Client ()+batch :: MonadClient m => Batch -> m () batch b = command (RqBatch b) -- | Return value of 'paginate'. Contains the actual result values as well@@ -158,7 +178,7 @@ -- Please note that -- as of Cassandra 2.1.0 -- if your requested page size -- is equal to the result size, 'hasMore' might be true and a subsequent -- 'nextPage' will return an empty list in 'result'.-paginate :: (Tuple a, Tuple b) => QueryString R a b -> QueryParams a -> Client (Page b)+paginate :: (MonadClient m, Tuple a, Tuple b) => QueryString R a b -> QueryParams a -> m (Page b) paginate q p = do let p' = p { pageSize = pageSize p <|> Just 10000 } r <- runQuery q p'
src/Database/CQL/IO/Client.hs view
@@ -9,13 +9,15 @@ module Database.CQL.IO.Client ( Client+ , MonadClient (..) , ClientState- , DebugInfo (..)+ , DebugInfo (..) , runClient , Database.CQL.IO.Client.init , shutdown , request , command+ , retry , debugInfo ) where @@ -52,6 +54,7 @@ import Network.Socket (SockAddr (..), PortNumber) import System.Logger.Class hiding (Settings, new, settings, create) +import qualified Control.Monad.Reader as Reader import qualified Data.List.NonEmpty as NE import qualified Data.Map.Strict as Map import qualified Database.CQL.IO.Connection as C@@ -71,10 +74,10 @@ } data Context = Context- { _settings :: Settings- , _logger :: Logger- , _timeouts :: TimeoutManager- , _sigMonit :: Signal HostEvent+ { _settings :: Settings+ , _logger :: Logger+ , _timeouts :: TimeoutManager+ , _sigMonit :: Signal HostEvent } -- | Opaque client state/environment.@@ -115,6 +118,18 @@ instance MonadLogger Client where log l m = view (context.logger) >>= \g -> Logger.log g l m +-- | Monads in which 'Client' actions may be embedded.+class (Functor m, Applicative m, Monad m, MonadIO m, MonadCatch m) => MonadClient m+ where+ -- | Lift a computation to the 'Client' monad.+ liftClient :: Client a -> m a+ -- | Execute an action with a modified 'ClientState'.+ localState :: (ClientState -> ClientState) -> m a -> m a++instance MonadClient Client where+ liftClient = id+ localState = Reader.local+ ----------------------------------------------------------------------------- -- API @@ -122,6 +137,10 @@ runClient :: MonadIO m => ClientState -> Client a -> m a runClient p a = liftIO $ runReaderT (client a) p +-- | Use given 'RetrySettings' during execution of some client action.+retry :: MonadClient m => RetrySettings -> m a -> m a+retry r = localState (set (context.settings.retrySettings) r)+ -- | Send a CQL 'Request' to the server and return a 'Response'. -- -- This function will first ask the clients load-balancing 'Policy' for@@ -131,49 +150,81 @@ -- If all available hosts are busy (i.e. their connection pools are fully -- utilised), the function will block until a connection becomes available -- or the maximum wait-queue length has been reached.-request :: (Tuple a, Tuple b) => Request k a b -> Client (Response k a b)-request a = do+request :: (MonadClient m, Tuple a, Tuple b) => Request k a b -> m (Response k a b)+request a = liftClient $ do s <- ask n <- liftIO $ hostCount (s^.policy)- start s n+ recovering (s^.context.settings.retrySettings.retryPolicy) recoverFrom $ \i ->+ if i == 0 then+ getResponse a s n+ else let s' = adjust s in+ getResponse (newRequest s) s' n where- start s n = do- h <- pickHost (s^.policy)- p <- Map.lookup h <$> readTVarIO' (s^.hostmap)- case p of- Just x -> action s n x `catches` handlers h- Nothing -> do- err $ msg (val "no pool for host " +++ h)- p' <- mkPool (s^.context) (h^.hostAddr)- atomically' $ modifyTVar' (s^.hostmap) (Map.alter (maybe (Just p') Just) h)- start s n+ adjust s =+ let x = s^.context.settings.retrySettings.sendTimeoutChange+ y = s^.context.settings.retrySettings.recvTimeoutChange+ in over (context.settings.connSettings.sendTimeout) (+ x)+ . over (context.settings.connSettings.responseTimeout) (+ y)+ $ s - action s n p = do- res <- tryWith p (go s)+ newRequest s =+ case s^.context.settings.retrySettings.reducedConsistency of+ Nothing -> a+ Just c ->+ case a of+ RqQuery (Query q p) -> RqQuery (Query q p { consistency = c })+ RqBatch b -> RqBatch b { batchConsistency = c }+ _ -> a++ recoverFrom =+ [ const $ Handler $ \e -> case e of+ ReadTimeout {} -> return True+ WriteTimeout {} -> return True+ Overloaded {} -> return True+ Unavailable {} -> return True+ ServerError {} -> return True+ _ -> return False+ , const $ Handler $ \(_ :: ConnectionError) -> return True+ ]++getResponse :: (Tuple a, Tuple b) => Request k a b -> ClientState -> Word -> Client (Response k a b)+getResponse a s n = do+ h <- pickHost (s^.policy)+ p <- Map.lookup h <$> readTVarIO' (s^.hostmap)+ case p of+ Just x -> action x `catches` handlers h+ Nothing -> do+ err $ msg (val "no pool for host " +++ h)+ p' <- mkPool (s^.context) (h^.hostAddr)+ atomically' $ modifyTVar' (s^.hostmap) (Map.alter (maybe (Just p') Just) h)+ getResponse a s n+ where+ action p = do+ res <- tryWith p go case res of Just r -> return r Nothing -> if n > 0 then- start s (n - 1)+ getResponse a s (n - 1) else case s^.context.settings.maxWaitQueue of- Nothing -> with p (transaction s)- Just q -> retry s q p+ Nothing -> with p transaction+ Just q -> again q p - go s h = do- atomically $ modifyTVar' (s^.failures) $ \n -> if n > 0 then n - 1 else 0- transaction s h+ go h = do+ atomically $ modifyTVar' (s^.failures) $ \x -> if x > 0 then x - 1 else 0+ transaction h - retry s q p = do- n <- atomically' $ do- n <- readTVar (s^.failures)- unless (n == q) $- writeTVar (s^.failures) (n + 1)- return n- unless (n < q) $+ again q p = do+ k <- atomically' $ do+ k <- readTVar (s^.failures)+ unless (k == q) $+ writeTVar (s^.failures) (k + 1)+ return k+ unless (k < q) $ throwM HostsBusy- with p (go s)+ with p go - transaction s c = do+ transaction c = do let x = s^.context.settings.connSettings.compression let v = s^.context.settings.protoVersion r <- parse x <$> C.request c (serialise v x a)@@ -187,7 +238,7 @@ pickHost p = maybe (throwM NoHostAvailable) return =<< liftIO (select p) -- | Like 'request' but not returning any result.-command :: Request k () () -> Client ()+command :: MonadClient m => Request k () () -> m () command = void . request data DebugInfo = DebugInfo@@ -205,8 +256,8 @@ . shows (policyInfo dbg) $ "" -debugInfo :: Client DebugInfo-debugInfo = do+debugInfo :: MonadClient m => m DebugInfo+debugInfo = liftClient $ do hosts <- Map.keys <$> (readTVarIO' =<< view hostmap) pols <- liftIO . display =<< view policy jbs <- Jobs.showJobs =<< view jobs@@ -353,7 +404,7 @@ else return Nothing maybe (liftIO . ignore . onEvent (e^.policy) $ HostDown (h^.hostAddr))- (liftIO . void . async . recovering reconnectPolicy reconnectHandlers . continue e)+ (liftIO . void . async . recovering reconnectPolicy reconnectHandlers . const . continue e) c Jobs.add (e^.jobs) (h^.hostAddr) $ monitor (e^.context) 0 30000000 h
src/Database/CQL/IO/Settings.hs view
@@ -8,7 +8,9 @@ module Database.CQL.IO.Settings where import Control.Lens hiding ((<|))+import Control.Retry import Data.List.NonEmpty (NonEmpty (..), (<|))+import Data.Monoid import Data.Time import Data.Word import Database.CQL.Protocol@@ -19,9 +21,17 @@ import Database.CQL.IO.Types (Milliseconds (..)) import Network.Socket (PortNumber (..)) +data RetrySettings = RetrySettings+ { _retryPolicy :: RetryPolicy+ , _reducedConsistency :: Maybe Consistency+ , _sendTimeoutChange :: Milliseconds+ , _recvTimeoutChange :: Milliseconds+ }+ data Settings = Settings { _poolSettings :: PoolSettings , _connSettings :: ConnectionSettings+ , _retrySettings :: RetrySettings , _protoVersion :: Version , _portnumber :: PortNumber , _contacts :: NonEmpty String@@ -29,6 +39,7 @@ , _policyMaker :: IO Policy } +makeLenses ''RetrySettings makeLenses ''Settings -- | Default settings:@@ -51,10 +62,13 @@ -- * no compression is applied to frame bodies -- -- * no default keyspace is used.+--+-- * no retries are done defSettings :: Settings defSettings = Settings P.defSettings C.defSettings+ noRetry V3 (fromInteger 9042) ("localhost" :| [])@@ -108,7 +122,7 @@ -- number of CPU cores this codes is running on. setPoolStripes :: Int -> Settings -> Settings setPoolStripes v s- | v < 1 = error "Database.CQL.IO.Settings: stripes must be greater than 0"+ | v < 1 = error "cql-io settings: stripes must be greater than 0" | otherwise = set (poolSettings.poolStripes) v s -- | When receiving a response times out, we can no longer use the stream of the@@ -131,8 +145,8 @@ -- to 32768 streams. setMaxStreams :: Int -> Settings -> Settings setMaxStreams v s = case s^.protoVersion of- V2 | v < 1 || v > 128 -> error "Database.CQL.IO.Settings: max. streams must be within [1, 128]"- V3 | v < 1 || v > 32768 -> error "Database.CQL.IO.Settings: max. streams must be within [1, 32768]"+ V2 | v < 1 || v > 128 -> error "cql-io settings: max. streams must be within [1, 128]"+ V3 | v < 1 || v > 32768 -> error "cql-io settings: max. streams must be within [1, 32768]" _ -> set (connSettings.maxStreams) v s -- | Set the connect timeout of a connection.@@ -154,3 +168,64 @@ -- initialised to use this keyspace. setKeyspace :: Keyspace -> Settings -> Settings setKeyspace v = set (connSettings.defKeyspace) (Just v)++-- | Set default retry settings to use.+setRetrySettings :: RetrySettings -> Settings -> Settings+setRetrySettings v = set retrySettings v++-----------------------------------------------------------------------------+-- Retry Settings++-- | Never retry.+noRetry :: RetrySettings+noRetry = RetrySettings (RetryPolicy $ const Nothing) Nothing 0 0++-- | Forever retry immediately.+retryForever :: RetrySettings+retryForever = RetrySettings mempty Nothing 0 0++-- | Limit number of retries.+maxRetries :: Word -> RetrySettings -> RetrySettings+maxRetries v = over retryPolicy (mappend (limitRetries $ fromIntegral v))++-- | When retrying a (batch-) query, change consistency to the given value.+adjustConsistency :: Consistency -> RetrySettings -> RetrySettings+adjustConsistency v = set reducedConsistency (Just v)++-- | Wait a constant time between retries.+constDelay :: NominalDiffTime -> RetrySettings -> RetrySettings+constDelay v = setDelayFn constantDelay v v++-- | Delay retries with exponential backoff.+expBackoff :: NominalDiffTime+ -- ^ Initial delay.+ -> NominalDiffTime+ -- ^ Maximum delay.+ -> RetrySettings+ -> RetrySettings+expBackoff = setDelayFn exponentialBackoff++-- | Delay retries using Fibonacci sequence as backoff.+fibBackoff :: NominalDiffTime+ -- ^ Initial delay.+ -> NominalDiffTime+ -- ^ Maximum delay.+ -> RetrySettings+ -> RetrySettings+fibBackoff = setDelayFn fibonacciBackoff++-- | On retry adjust the send timeout.+adjustSendTimeout :: NominalDiffTime -> RetrySettings -> RetrySettings+adjustSendTimeout v = set sendTimeoutChange (Ms $ round (1000 * v))++-- | On retry adjust the response timeout.+adjustResponseTimeout :: NominalDiffTime -> RetrySettings -> RetrySettings+adjustResponseTimeout v = set recvTimeoutChange (Ms $ round (1000 * v))++setDelayFn :: (Int -> RetryPolicy)+ -> NominalDiffTime+ -> NominalDiffTime+ -> RetrySettings+ -> RetrySettings+setDelayFn d v w = over retryPolicy+ (mappend $ capDelay (round (1000000 * w)) $ d (round (1000000 * v)))