diff --git a/CHANGELOG.txt b/CHANGELOG.txt --- a/CHANGELOG.txt +++ b/CHANGELOG.txt @@ -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 ------ diff --git a/README.asciidoc b/README.asciidoc --- a/README.asciidoc +++ b/README.asciidoc @@ -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). diff --git a/cql-io.cabal b/cql-io.cabal --- a/cql-io.cabal +++ b/cql-io.cabal @@ -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 -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 diff --git a/src/Control/Retry.hs b/src/Control/Retry.hs deleted file mode 100644 --- a/src/Control/Retry.hs +++ /dev/null @@ -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 --- 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 diff --git a/src/Database/CQL/IO.hs b/src/Database/CQL/IO.hs --- a/src/Database/CQL/IO.hs +++ b/src/Database/CQL/IO.hs @@ -75,6 +75,7 @@ , setSendTimeout , setRetrySettings , setMaxRecvBuffer + , setSSLContext -- ** Retry Settings , RetrySettings diff --git a/src/Database/CQL/IO/Client.hs b/src/Database/CQL/IO/Client.hs --- a/src/Database/CQL/IO/Client.hs +++ b/src/Database/CQL/IO/Client.hs @@ -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 () diff --git a/src/Database/CQL/IO/Connection.hs b/src/Database/CQL/IO/Connection.hs --- a/src/Database/CQL/IO/Connection.hs +++ b/src/Database/CQL/IO/Connection.hs @@ -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 diff --git a/src/Database/CQL/IO/Connection/Socket.hs b/src/Database/CQL/IO/Connection/Socket.hs new file mode 100644 --- /dev/null +++ b/src/Database/CQL/IO/Connection/Socket.hs @@ -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) diff --git a/src/Database/CQL/IO/Settings.hs b/src/Database/CQL/IO/Settings.hs --- a/src/Database/CQL/IO/Settings.hs +++ b/src/Database/CQL/IO/Settings.hs @@ -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 }