cql-io 0.14.5 → 0.15.2

raw patch · 9 files changed

+186/−420 lines, 9 filesdep +HsOpenSSLdep +retrydep ~asyncPVP ok

version bump matches the API change (PVP)

Dependencies added: HsOpenSSL, retry

Dependency ranges changed: async

API changes (from Hackage documentation)

Files

• CHANGELOG.txt +13/−1
• README.asciidoc +5/−0
• cql-io.cabal +9/−4
• src/Control/Retry.hs +0/−341
• src/Database/CQL/IO.hs +1/−0
• src/Database/CQL/IO/Client.hs +16/−10
• src/Database/CQL/IO/Connection.hs +25/−58
• src/Database/CQL/IO/Connection/Socket.hs +99/−0
• src/Database/CQL/IO/Settings.hs +18/−6

@@ -1,6 +1,18 @@+0.15.2+------+- Update `async` dependency++0.15.1+------+- Use `retry >= 0.7` instead of internal module++0.15.0+------+- Add experimental TLS support+ 0.14.5 -------- Add `trans` to execute "lightweight transactions".+- Add `trans` to execute "lightweight transactions"  0.14.4 ------

@@ -29,3 +29,8 @@ Prepared queries are an optimisation which parse and prepare a query only once on Cassandra nodes but execute it many times with different concrete values.++.TLS support++Client to node communication can optionally use transport layer security+(using HsOpenSSL).

@@ -1,12 +1,12 @@ name:                 cql-io-version:              0.14.5+version:              0.15.2 synopsis:             Cassandra CQL client. stability:            experimental license:              MPL-2.0 license-file:         LICENSE.txt author:               Toralf Wittner maintainer:           Toralf Wittner <tw@dtex.org>-copyright:            (C) 2014-2015 Toralf Wittner+copyright:            (C) 2014-2016 Toralf Wittner homepage:             https://github.com/twittner/cql-io/ bug-reports:          https://github.com/twittner/cql-io/issues category:             Database@@ -39,6 +39,9 @@     * /Prepared queries/. Prepared queries are an optimisation which parse     and prepare a query only once on Cassandra nodes but execute it many     times with different concrete values.+    .+    * /TLS support/. Client to node communication can optionally use transport+    layer security (using HsOpenSSL).  source-repository head     type:             git@@ -53,13 +56,13 @@         Database.CQL.IO      other-modules:-        Control.Retry         Database.CQL.IO.Batch         Database.CQL.IO.Client         Database.CQL.IO.Cluster.Discovery         Database.CQL.IO.Cluster.Host         Database.CQL.IO.Cluster.Policies         Database.CQL.IO.Connection+        Database.CQL.IO.Connection.Socket         Database.CQL.IO.Hexdump         Database.CQL.IO.Jobs         Database.CQL.IO.Pool@@ -73,7 +76,7 @@         Database.CQL.IO.Types      build-depends:-          async              == 2.0.*+          async              >= 2.0         , auto-update        == 0.1.*         , base               >= 4.7     && < 5.0         , bytestring         >= 0.10    && < 1.0@@ -84,10 +87,12 @@         , exceptions         >= 0.4     && < 1.0         , hashable           >= 1.2     && < 2.0         , iproute            >= 1.3     && < 2.0+        , HsOpenSSL          >= 0.11    && < 1.0         , lens               >= 4.4     && < 5.0         , monad-control      >= 0.3     && < 2.0         , mtl                >= 2.1     && < 2.5         , mwc-random         >= 0.13    && < 0.14+        , retry              >= 0.7     && < 1.0         , network            >= 2.4     && < 3.0         , semigroups         >= 0.15    && < 0.20         , stm                >= 2.4     && < 3.0

@@ -1,341 +0,0 @@--- This is a copy of Ozgun Ataman's retry module version 0.6--- 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 CPP                   #-}-{-# LANGUAGE FlexibleContexts      #-}-{-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE RankNTypes            #-}-{-# LANGUAGE RecordWildCards       #-}-{-# LANGUAGE ScopedTypeVariables   #-}-{-# LANGUAGE ViewPatterns          #-}---------------------------------------------------------------------------------- |--- 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-    , logRetries--    -- * Retry Policies-    , constantDelay-    , exponentialBackoff-    , fibonacciBackoff-    , limitRetries-    , limitRetriesByDelay-    , 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 microseconds. *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 <> limitRetries 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)------------------------------------------------------------------------------------- | Add an upperbound to a policy such that once the given time-delay--- amount has been reached or exceeded, the policy will stop retrying--- and fail.-limitRetriesByDelay-    :: Int-    -- ^ Time-delay limit in microseconds.-    -> RetryPolicy-    -> RetryPolicy-limitRetriesByDelay i p = RetryPolicy $ \ n -> getRetryPolicy p n >>= limit-  where-    limit delay = if delay >= i then Nothing else Just delay------------------------------------------------------------------------------------- | 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 a time-upperbound for any delays that may be directed by the--- given policy.  This function does not terminate the retrying.  The policy--- `capDelay maxDelay (exponentialBackoff n)` will never stop retrying.  It--- will reach a state where it retries forever with a delay of `maxDelay`--- between each one.  To get termination you need to use one of the--- 'limitRetries' function variants.-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 (const $ return . 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-#if MIN_VERSION_exceptions(0, 6, 0)-         :: (MonadIO m, MonadMask m)-#else-         :: (MonadIO m, MonadCatch m)-#endif-         => 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-#if MIN_VERSION_exceptions(0, 6, 0)-           :: (MonadIO m, MonadMask m)-#else-           :: (MonadIO m, MonadCatch m)-#endif-           => RetryPolicy-           -- ^ Just use 'def' for 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 = mask $ \restore -> go restore 0-    where-      go restore = loop-        where-          loop n = do-            r <- try $ restore (f n)-            case r of-              Right x -> return x-              Left e -> recover (e :: SomeException) hs-            where-              recover e [] = throwM e-              recover e ((($ n) -> Handler h) : hs')-                | Just e' <- fromException e = do-                    chk <- h e'-                    if chk-                      then case policy n of-                        Just delay -> do-                          liftIO $ threadDelay delay-                          loop $! n+1-                        Nothing -> throwM e'-                      else throwM e'-                | otherwise = recover e hs'------------------------------------------------------------------------------------- | Helper function for constructing handler functions of the form required--- by 'recovering'.-logRetries-    :: (Monad m, Show e, Exception e)-    => (e -> m Bool)-    -- ^ Test for whether action is to be retried-    -> (String -> m ())-    -- ^ How to report the generated warning message.-    -> Int-    -- ^ Retry number-    -> Handler m Bool-logRetries f report n = Handler $ \ e -> do-    res <- f e-    let msg = "[retry:" <> show n <> "] Encountered " <> show e <> ". " <>-              if res then "Retrying." else "Crashing."-    report msg-    return res---                              -------------------                              -- 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

@@ -75,6 +75,7 @@     , setSendTimeout     , setRetrySettings     , setMaxRecvBuffer+    , setSSLContext        -- ** Retry Settings     , RetrySettings

@@ -51,7 +51,7 @@ #if MIN_VERSION_transformers(0,4,0) import Control.Monad.Trans.Except #endif-import Control.Retry+import Control.Retry (recovering, capDelay, exponentialBackoff, rsIterNumber) import Data.Foldable (for_, foldrM) import Data.List (find) import Data.List.NonEmpty (NonEmpty (..))@@ -73,6 +73,7 @@ import Database.CQL.IO.Types import Database.CQL.Protocol hiding (Map) import Network.Socket (SockAddr (..), PortNumber)+import OpenSSL.Session (SomeSSLException) import System.Logger.Class hiding (Settings, new, settings, create) import Prelude @@ -226,7 +227,9 @@ mkRequest fn a = do     s <- ask     recovering (s^.context.settings.retrySettings.retryPolicy) recoverFrom $ \i -> do-        r <- if i == 0 then fn a s else fn (newRequest s) (adjust s)+        r <- if rsIterNumber i == 0+                 then fn a s+                 else fn (newRequest s) (adjust s)         maybe (throwM HostsBusy) return r   where     adjust s =@@ -254,9 +257,10 @@             Unavailable  {} -> return True             ServerError  {} -> return True             _               -> return False-        , const $ Handler $ \(_ :: ConnectionError) -> return True-        , const $ Handler $ \(_ :: IOException)     -> return True-        , const $ Handler $ \(_ :: HostError)       -> return True+        , const $ Handler $ \(_ :: ConnectionError)  -> return True+        , const $ Handler $ \(_ :: IOException)      -> return True+        , const $ Handler $ \(_ :: HostError)        -> return True+        , const $ Handler $ \(_ :: SomeSSLException) -> return True         ]  -- | Invoke 'request1' up to @n@ times with different hosts if no@@ -292,8 +296,9 @@         r `seq` return (h, r)      handlers =-        [ Handler $ \(e :: ConnectionError) -> onConnectionError h e >> throwM e-        , Handler $ \(e :: IOException)     -> onConnectionError h e >> throwM e+        [ Handler $ \(e :: ConnectionError)  -> onConnectionError h e >> throwM e+        , Handler $ \(e :: IOException)      -> onConnectionError h e >> throwM e+        , Handler $ \(e :: SomeSSLException) -> onConnectionError h e >> throwM e         ]  -- | Execute the given request. If an 'Unprepared' error is returned, this@@ -555,9 +560,10 @@     reconnectPolicy = capDelay 5000000 (exponentialBackoff 5000)      reconnectHandlers =-        [ const (Handler $ \(_ :: IOException)     -> return True)-        , const (Handler $ \(_ :: ConnectionError) -> return True)-        , const (Handler $ \(_ :: HostError)       -> return True)+        [ const (Handler $ \(_ :: IOException)      -> return True)+        , const (Handler $ \(_ :: ConnectionError)  -> return True)+        , const (Handler $ \(_ :: HostError)        -> return True)+        , const (Handler $ \(_ :: SomeSSLException) -> return True)         ]  replaceControl :: InetAddr -> Client ()

@@ -2,8 +2,6 @@ -- License, v. 2.0. If a copy of the MPL was not distributed with this -- file, You can obtain one at http://mozilla.org/MPL/2.0/. -{-# LANGUAGE BangPatterns        #-}-{-# LANGUAGE CPP                 #-} {-# LANGUAGE OverloadedStrings   #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE TemplateHaskell     #-}@@ -32,6 +30,7 @@     , compression     , defKeyspace     , maxRecvBuffer+    , tlsContext     ) where  import Control.Applicative@@ -44,15 +43,15 @@ import Control.Monad import Control.Monad.Catch import Control.Monad.IO.Class-import Data.ByteString.Builder import Data.ByteString.Lazy (ByteString) import Data.Int-import Data.Maybe (isJust, fromMaybe)+import Data.Maybe (fromMaybe) import Data.Monoid import Data.Text.Lazy (fromStrict) import Data.Unique import Data.Vector (Vector, (!)) import Database.CQL.Protocol+import Database.CQL.IO.Connection.Socket (Socket) import Database.CQL.IO.Hexdump import Database.CQL.IO.Protocol import Database.CQL.IO.Signal hiding (connect)@@ -60,23 +59,20 @@ import Database.CQL.IO.Types import Database.CQL.IO.Tickets (Pool, toInt, markAvailable) import Database.CQL.IO.Timeouts (TimeoutManager, withTimeout)-import Foreign.C.Types (CInt (..))-import Network-import Network.Socket hiding (connect, close, recv, send)-import Network.Socket.ByteString.Lazy (sendAll)+import Network.Socket hiding (Socket, close, connect, send)+import OpenSSL.Session (SSLContext) import System.IO (nativeNewline, Newline (..)) import System.Logger hiding (Settings, close, defSettings, settings) import System.Timeout import Prelude -import qualified Data.ByteString            as B-import qualified Data.ByteString.Lazy       as L-import qualified Data.ByteString.Lazy.Char8 as Char8-import qualified Data.Vector                as Vector-import qualified Database.CQL.IO.Sync       as Sync-import qualified Database.CQL.IO.Tickets    as Tickets-import qualified Network.Socket             as S-import qualified Network.Socket.ByteString  as NB+import qualified Data.ByteString.Lazy              as L+import qualified Data.ByteString.Lazy.Char8        as Char8+import qualified Data.Vector                       as Vector+import qualified Database.CQL.IO.Connection.Socket as Socket+import qualified Database.CQL.IO.Sync              as Sync+import qualified Database.CQL.IO.Tickets           as Tickets+import qualified Network.Socket                    as S  data ConnectionSettings = ConnectionSettings     { _connectTimeout  :: !Milliseconds@@ -86,6 +82,7 @@     , _compression     :: !Compression     , _defKeyspace     :: !(Maybe Keyspace)     , _maxRecvBuffer   :: !Int+    , _tlsContext      :: !(Maybe SSLContext)     }  type Streams = Vector (Sync (Header, ByteString))@@ -116,7 +113,7 @@     show = Char8.unpack . eval . bytes  instance ToBytes Connection where-    bytes c = bytes (c^.address) +++ val "#" +++ fd (c^.sock)+    bytes c = bytes (c^.address) +++ val "#" +++ c^.sock  defSettings :: ConnectionSettings defSettings =@@ -127,6 +124,7 @@                        noCompression -- compression                        Nothing       -- keyspace                        16384         -- receive buffer size+                       Nothing       -- no tls by default  resolve :: String -> PortNumber -> IO [InetAddr] resolve host port =@@ -136,15 +134,7 @@  connect :: MonadIO m => ConnectionSettings -> TimeoutManager -> Version -> Logger -> InetAddr -> m Connection connect t m v g a = liftIO $ do-    c <- bracketOnError (mkSock a) S.close $ \s -> do-        ok <- timeout (ms (t^.connectTimeout) * 1000) (S.connect s (sockAddr a))-        unless (isJust ok) $-            throwM (ConnectTimeout a)-        open s-    validateSettings c `onException` close c-    return c-  where-    open s = do+    c <- bracketOnError (Socket.open (t^.connectTimeout) a (t^.tlsContext)) Socket.close $ \s -> do         tck <- Tickets.pool (t^.maxStreams)         syn <- Vector.replicateM (t^.maxStreams) Sync.create         lck <- newMVar ()@@ -152,19 +142,11 @@         sig <- signal         rdr <- async (readLoop v g t tck a s syn sig sta lck)         Connection t a m v s sta syn lck rdr tck g sig <$> newUnique--mkSock :: InetAddr -> IO Socket-mkSock (InetAddr a) = socket (familyOf a) Stream defaultProtocol-  where-    familyOf (SockAddrInet  _ _)     = AF_INET-    familyOf (SockAddrInet6 _ _ _ _) = AF_INET6-    familyOf (SockAddrUnix  _)       = AF_UNIX-#if MIN_VERSION_network(2,6,1)-    familyOf (SockAddrCan   _      ) = AF_CAN-#endif+    validateSettings c `onException` close c+    return c  ping :: MonadIO m => InetAddr -> m Bool-ping a = liftIO $ bracket (mkSock a) S.close $ \s ->+ping a = liftIO $ bracket (Socket.mkSock a) S.close $ \s ->     fromMaybe False <$> timeout 5000000         ((S.connect s (sockAddr a) >> return True) `catchAll` const (return False)) @@ -201,8 +183,8 @@             Tickets.close (ConnectionClosed i) tck             Vector.mapM_ (Sync.close (ConnectionClosed i)) syn             void $ forkIOWithUnmask $ \unmask -> unmask $ do-                shutdown sck ShutdownReceive-                withMVar wlck (const $ S.close sck)+                Socket.shutdown sck ShutdownReceive+                withMVar wlck (const $ Socket.close sck)      logException :: SomeException -> IO ()     logException e = case fromException e of@@ -225,7 +207,7 @@         withMVar (c^.wLock) $ const $ do             isOpen <- readTVarIO (c^.status)             if isOpen then-                sendAll (c^.sock) req+                Socket.send (c^.sock) req             else                 throwM $ ConnectionClosed (c^.address)         return i@@ -240,7 +222,7 @@  readSocket :: Version -> Logger -> InetAddr -> Socket -> Int -> IO (Header, ByteString) readSocket v g i s n = do-    b <- recv n i s (if v == V3 then 9 else 8)+    b <- Socket.recv n i s (if v == V3 then 9 else 8)     h <- case header v b of             Left  e -> throwM $ InternalError ("response header reading: " ++ e)             Right h -> return h@@ -248,25 +230,13 @@         RqHeader -> throwM $ InternalError "unexpected request header"         RsHeader -> do             let len = lengthRepr (bodyLength h)-            x <- recv n i s (fromIntegral len)-            trace g $ msg (i +++ val "#" +++ fd s)+            x <- Socket.recv n i s (fromIntegral len)+            trace g $ msg (i +++ val "#" +++ s)                 ~~ "stream" .= fromStreamId (streamId h)                 ~~ "type"   .= val "response"                 ~~ msg' (hexdump $ L.take 160 (b <> x))             return (h, x) -recv :: Int -> InetAddr -> Socket -> Int -> IO ByteString-recv _ _ _ 0 = return L.empty-recv x i s n = toLazyByteString <$> go n mempty-  where-    go !k !bb = do-        a <- NB.recv s (k `min` x)-        when (B.null a) $-            throwM (ConnectionClosed i)-        let b = bb <> byteString a-        let m = k - B.length a-        if m > 0 then go m b else return b- ----------------------------------------------------------------------------- -- Operations @@ -330,9 +300,6 @@     params = QueryParams cons False p Nothing Nothing Nothing  -- logging helpers:--fd :: Socket -> Int32-fd !s = let CInt !n = fdSocket s in n  msg' :: ByteString -> Msg -> Msg msg' x = msg $ case nativeNewline of

@@ -0,0 +1,99 @@+-- This Source Code Form is subject to the terms of the Mozilla Public+-- License, v. 2.0. If a copy of the MPL was not distributed with this+-- file, You can obtain one at http://mozilla.org/MPL/2.0/.++{-# LANGUAGE BangPatterns    #-}+{-# LANGUAGE CPP             #-}+{-# LANGUAGE TemplateHaskell #-}++module Database.CQL.IO.Connection.Socket+    ( Socket+    , mkSock+    , open+    , send+    , recv+    , close+    , shutdown+    ) where++import Control.Applicative+import Control.Monad+import Control.Monad.Catch+import Data.ByteString (ByteString)+import Data.ByteString.Builder+import Data.Maybe (isJust)+import Data.Monoid+import Database.CQL.IO.Types+import Foreign.C.Types (CInt (..))+import Network.Socket hiding (Stream, Socket, connect, close, recv, send, shutdown)+import Network.Socket.ByteString.Lazy (sendAll)+import OpenSSL.Session (SSL, SSLContext)+import System.Logger (ToBytes (..))+import System.Timeout+import Prelude++import qualified Data.ByteString            as Bytes+import qualified Data.ByteString.Lazy       as Lazy+import qualified Network.Socket             as S+import qualified Network.Socket.ByteString  as NB+import qualified OpenSSL.Session            as SSL++data Socket = Stream !S.Socket | Tls !S.Socket !SSL++instance ToBytes Socket where+    bytes s = bytes $ case s of+        Stream x -> fd x+        Tls  x _ -> fd x+      where+        fd x = let CInt n = S.fdSocket x in n++open :: Milliseconds -> InetAddr -> Maybe SSLContext -> IO Socket+open to a ctx = do+    bracketOnError (mkSock a) S.close $ \s -> do+        ok <- timeout (ms to * 1000) (S.connect s (sockAddr a))+        unless (isJust ok) $+            throwM (ConnectTimeout a)+        case ctx of+            Nothing  -> return (Stream s)+            Just set -> do+                c <- SSL.connection set s+                SSL.connect c+                return (Tls s c)++mkSock :: InetAddr -> IO S.Socket+mkSock (InetAddr a) = S.socket (familyOf a) S.Stream defaultProtocol+  where+    familyOf (SockAddrInet  _ _)     = AF_INET+    familyOf (SockAddrInet6 _ _ _ _) = AF_INET6+    familyOf (SockAddrUnix  _)       = AF_UNIX+#if MIN_VERSION_network(2,6,1)+    familyOf (SockAddrCan   _      ) = AF_CAN+#endif++close :: Socket -> IO ()+close (Stream s) = S.close s+close (Tls s  c) = SSL.shutdown c SSL.Bidirectional >> S.close s++shutdown :: Socket -> ShutdownCmd -> IO ()+shutdown (Stream s) cmd = S.shutdown s cmd+shutdown _          _   = return ()++recv :: Int -> InetAddr -> Socket -> Int -> IO Lazy.ByteString+recv x a (Stream s) n = receive x a (NB.recv s) n+recv x a (Tls _  c) n = receive x a (SSL.read c) n++receive :: Int -> InetAddr -> (Int -> IO ByteString) -> Int -> IO Lazy.ByteString+receive _ _ _ 0 = return Lazy.empty+receive x i f n = toLazyByteString <$> go n mempty+  where+    go !k !bb = do+        a <- f (k `min` x)+        when (Bytes.null a) $+            throwM (ConnectionClosed i)+        let b = bb <> byteString a+        let m = k - Bytes.length a+        if m > 0 then go m b else return b++send :: Socket -> Lazy.ByteString -> IO ()+send (Stream s) b = sendAll s b+send (Tls _  c) b = mapM_ (SSL.write c) (Lazy.toChunks b)

@@ -3,12 +3,13 @@ -- file, You can obtain one at http://mozilla.org/MPL/2.0/.  {-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RankNTypes        #-} {-# LANGUAGE TemplateHaskell   #-}  module Database.CQL.IO.Settings where  import Control.Lens hiding ((<|))-import Control.Retry+import Control.Retry hiding (retryPolicy) import Data.List.NonEmpty (NonEmpty (..), (<|)) import Data.Monoid import Data.Time@@ -20,6 +21,7 @@ import Database.CQL.IO.Pool as P import Database.CQL.IO.Types (Milliseconds (..)) import Network.Socket (PortNumber (..))+import OpenSSL.Session (SSLContext) import Prelude  data PrepareStrategy@@ -28,7 +30,7 @@     deriving (Eq, Ord, Show)  data RetrySettings = RetrySettings-    { _retryPolicy        :: !RetryPolicy+    { _retryPolicy        :: !(forall m. Monad m => RetryPolicyM m)     , _reducedConsistency :: !(Maybe Consistency)     , _sendTimeoutChange  :: !Milliseconds     , _recvTimeoutChange  :: !Milliseconds@@ -187,12 +189,18 @@ setMaxRecvBuffer :: Int -> Settings -> Settings setMaxRecvBuffer v = set (connSettings.maxRecvBuffer) v +-- | Set a fully configured SSL context.+--+-- This will make client server queries use TLS.+setSSLContext :: SSLContext -> Settings -> Settings+setSSLContext v = set (connSettings.tlsContext) (Just v)+ ----------------------------------------------------------------------------- -- Retry Settings  -- | Never retry. noRetry :: RetrySettings-noRetry = RetrySettings (RetryPolicy $ const Nothing) Nothing 0 0+noRetry = RetrySettings (RetryPolicyM $ const (return Nothing)) Nothing 0 0  -- | Forever retry immediately. retryForever :: RetrySettings@@ -200,7 +208,8 @@  -- | Limit number of retries. maxRetries :: Word -> RetrySettings -> RetrySettings-maxRetries v = over retryPolicy (mappend (limitRetries $ fromIntegral v))+maxRetries v s =+    s { _retryPolicy = limitRetries (fromIntegral v) <> _retryPolicy s }  -- | When retrying a (batch-) query, change consistency to the given value. adjustConsistency :: Consistency -> RetrySettings -> RetrySettings@@ -241,5 +250,8 @@            -> NominalDiffTime            -> RetrySettings            -> RetrySettings-setDelayFn d v w = over retryPolicy-    (mappend $ capDelay (round (1000000 * w)) $ d (round (1000000 * v)))+setDelayFn f v w s =+    let a = round (1000000 * w)+        b = round (1000000 * v)+    in+        s { _retryPolicy = capDelay a (f b) <> _retryPolicy s }