packages feed

cql-io 1.0.0 → 1.0.1

raw patch · 11 files changed

+150/−102 lines, 11 filesdep ~basePVP ok

version bump matches the API change (PVP)

Dependency ranges changed: base

API changes (from Hackage documentation)

Files

CHANGELOG view
@@ -1,3 +1,10 @@+1.0.1+------+- Address an issue whereby Cassandra 'Error' responses might be mistakenly+  thrown by the (internal) withRetries function, possibly leading to+  problems with the automatic (re-)preparation of prepared queries upon+  'Unprepared' server responses.+ 1.0.0 ------ - Add support for CQL V4 binary protocol.
cql-io.cabal view
@@ -1,5 +1,5 @@ name:                 cql-io-version:              1.0.0+version:              1.0.1 synopsis:             Cassandra CQL client. license:              MPL-2.0 license-file:         LICENSE@@ -79,7 +79,7 @@     build-depends:           async              >= 2.0         , auto-update        >= 0.1-        , base               >= 4.7   && < 5.0+        , base               >= 4.9   && < 5.0         , bytestring         >= 0.10         , containers         >= 0.5         , cql                >= 4.0
src/Database/CQL/IO.hs view
@@ -205,7 +205,10 @@  -- | A type which can be run as a query. class RunQ q where-    runQ :: (MonadClient m, Tuple a, Tuple b) => q k a b -> QueryParams a -> m (Response k a b)+    runQ :: (MonadClient m, Tuple a, Tuple b)+         => q k a b+         -> QueryParams a+         -> m (Response k a b)  instance RunQ QueryString where     runQ q p = request (RqQuery (Query q p))@@ -259,7 +262,7 @@     r <- runQ q p     getResult r >>= \case         RowsResult _ b -> return b-        _              -> throwM $ UnexpectedResponse' r+        _              -> throwM $ UnexpectedResponse r  -- | Run a CQL schema query, returning 'SchemaChange' information, if any. schema :: (MonadClient m, Tuple a, RunQ q) => q S a () -> QueryParams a -> m (Maybe SchemaChange)
src/Database/CQL/IO/Batch.hs view
@@ -23,6 +23,7 @@ import Control.Monad.Trans import Control.Monad.Trans.State.Strict import Database.CQL.IO.Client+import Database.CQL.IO.Cluster.Host import Database.CQL.IO.PrepQuery import Database.CQL.IO.Types import Database.CQL.Protocol@@ -38,9 +39,9 @@ batch m = do     b <- execStateT (unBatchM m) (Batch BatchLogged [] Quorum Nothing)     r <- executeWithPrepare Nothing (RqBatch b :: Raw Request)-    getResult r >>= \case+    getResult (hrResponse r) >>= \case         VoidResult -> return ()-        _          -> throwM $ UnexpectedResponse' r+        _          -> throwM $ UnexpectedResponse (hrResponse r)  -- | Add a query to this batch. addQuery :: (Show a, Tuple a, Tuple b) => QueryString W a b -> a -> BatchM ()
src/Database/CQL/IO/Client.hs view
@@ -54,7 +54,7 @@ #if MIN_VERSION_transformers(0,4,0) import Control.Monad.Trans.Except #endif-import Control.Retry (capDelay, exponentialBackoff, rsIterNumber, applyPolicy)+import Control.Retry (capDelay, exponentialBackoff, rsIterNumber) import Control.Retry (recovering) import Data.Foldable (for_, foldrM) import Data.List (find)@@ -235,12 +235,12 @@ request :: (MonadClient m, Tuple a, Tuple b) => Request k a b -> m (Response k a b) request a = liftClient $ do     n <- liftIO . hostCount =<< view policy-    snd <$> withRetries (requestN n) a+    hrResponse <$> withRetries (requestN n) a  -- | Invoke 'request1' up to @n@ times with different hosts if no -- connection is available. May return 'Nothing' if no connection -- is available on any of the tried hosts.-requestN :: (Tuple b, Tuple a) => Word -> Request k a b -> ClientState -> Client (Maybe (Host, Response k a b))+requestN :: (Tuple b, Tuple a) => Word -> Request k a b -> ClientState -> Client (Maybe (HostResponse k a b)) requestN !n a s = do     hst <- pickHost (s^.policy)     res <- request1 hst a s@@ -252,13 +252,13 @@  -- | Get 'Response' from a single 'Host'. -- May return 'Nothing' if no connection is available.-request1 :: (Tuple a, Tuple b) => Host -> Request k a b -> ClientState -> Client (Maybe (Host, Response k a b))+request1 :: (Tuple a, Tuple b) => Host -> Request k a b -> ClientState -> Client (Maybe (HostResponse k a b)) request1 h a s = do     p <- Map.lookup h <$> readTVarIO' (s^.hostmap)     case p of         Just x -> do             result <- with x transaction `catches` handlers-            for_ result $ \(_, r) ->+            for_ result $ \(HostResponse _ r) ->                 for_ (Cql.warnings r) $ \w ->                     warn $ msg (val "server warning: " +++ w)             return result@@ -272,7 +272,7 @@         let x = s^.context.settings.connSettings.compression         let v = s^.context.settings.protoVersion         r <- parse x <$> C.request c (serialise v x a)-        r `seq` return (h, r)+        r `seq` return (HostResponse h r)      handlers =         [ Handler $ \(e :: ConnectionError)  -> onConnectionError h e >> throwM e@@ -283,11 +283,11 @@ -- | Execute the given request. If an 'Unprepared' error is returned, this -- function will automatically try to re-prepare the query and re-execute -- the original request using the same host which was used for re-preparation.-executeWithPrepare :: (Tuple b, Tuple a) => Maybe Host -> Request k a b -> Client (Response k a b)+executeWithPrepare :: (Tuple b, Tuple a) => Maybe Host -> Request k a b -> Client (HostResponse k a b) executeWithPrepare h q = do     f <- selectAction h     r <- withRetries f q-    case snd r of+    case hrResponse r of         RsError _ _ (Unprepared _ i) -> do             pq <- preparedQueries             qs <- atomically' (PQ.lookupQueryString (QueryId i) pq)@@ -296,8 +296,7 @@                 Just  s -> do                     (g, _) <- prepare (Just LazyPrepare) (s :: Raw QueryString)                     executeWithPrepare (Just g) q-        RsError _ _ e -> throwM e-        x -> return x+        _ -> return r   where     selectAction Nothing  = view policy >>= liftIO . hostCount >>= return . requestN     selectAction (Just x) = return (request1 x)@@ -309,8 +308,8 @@ prepare (Just LazyPrepare) qs = do     s <- ask     n <- liftIO $ hostCount (s^.policy)-    (h, r) <- withRetries (requestN n) (RqPrepare (Prepare qs))-    (h,) <$> getPreparedQueryId r+    r <- withRetries (requestN n) (RqPrepare (Prepare qs))+    getPreparedQueryId r  prepare (Just EagerPrepare) qs = view policy     >>= liftIO . current@@ -318,8 +317,8 @@     >>= first   where     action rq h = do-        r <- snd <$> withRetries (request1 h) rq-        (h,) <$> getPreparedQueryId r+        r <- withRetries (request1 h) rq+        getPreparedQueryId r      first (x:_) = return x     first []    = throwM NoHostAvailable@@ -334,7 +333,7 @@     pq <- view prepQueries     maybe (new pq) (run Nothing) =<< atomically' (PQ.lookupQueryId q pq)   where-    run h i = executeWithPrepare h (RqExecute (Execute i p))+    run h i = hrResponse <$> executeWithPrepare h (RqExecute (Execute i p))     new pq  = do         (h, i) <- prepare (Just LazyPrepare) (PQ.queryString q)         atomically' (PQ.insert q i pq)@@ -500,24 +499,31 @@ ----------------------------------------------------------------------------- -- Exception handling +-- [Note: Error responses]+-- Cassandra error responses are locally thrown as 'ResponseError's to achieve+-- a unified handling of retries in the context of a single retry policy,+-- together with other recoverable (i.e. retryable) exceptions. However, this+-- is just an internal technicality for handling retries - generally error+-- responses must not escape this function as exceptions. Deciding if and when+-- to actually throw a 'ResponseError' upon inspection of the 'HostResponse'+-- must be left to the caller. withRetries     :: (Tuple a, Tuple b)-    => (Request k a b -> ClientState -> Client (Maybe (Host, Response k a b)))+    => (Request k a b -> ClientState -> Client (Maybe (HostResponse k a b)))     -> Request k a b-    -> Client (Host, Response k a b)+    -> Client (HostResponse k a b) withRetries fn a = do     s <- ask     let p = s^.context.settings.retrySettings.retryPolicy-    recovering p recoverFrom $ \i -> do+    r <- try $ recovering p canRetry $ \i -> do         r <- if rsIterNumber i == 0                  then fn a s                  else fn (newRequest s) (adjust s)         case r of             Nothing -> throwM HostsBusy-            Just hr -> case snd hr of-                RsError _ _ e -> applyPolicy p i-                             >>= maybe (return hr) (const (throwM e))-                _             -> return hr+            -- [Note: Error responses]+            Just hr -> maybe (return hr) throwM (toResponseError hr)+    return $ either fromResponseError id r   where     adjust s =         let x = s^.context.settings.retrySettings.sendTimeoutChange@@ -536,14 +542,14 @@                     RqBatch b               -> RqBatch b { batchConsistency = c }                     _                       -> a -    recoverFrom =-        [ const $ Handler $ \e -> return $ case e of-            ReadTimeout  {} -> True-            WriteTimeout {} -> True-            Overloaded   {} -> True-            Unavailable  {} -> True-            ServerError  {} -> True-            _               -> False+    canRetry =+        [ const $ Handler $ \(e :: ResponseError) -> case reCause e of+            ReadTimeout  {} -> return True+            WriteTimeout {} -> return True+            Overloaded   {} -> return True+            Unavailable  {} -> return True+            ServerError  {} -> return True+            _               -> return False         , const $ Handler $ \(_ :: ConnectionError)  -> return True         , const $ Handler $ \(_ :: IOException)      -> return True         , const $ Handler $ \(_ :: HostError)        -> return True@@ -664,15 +670,15 @@ getResult :: MonadThrow m => Response k a b -> m (Result k a b) getResult (RsResult _ _ r) = return r getResult (RsError  _ _ e) = throwM e-getResult r                = throwM $ UnexpectedResponse r+getResult hr               = throwM (UnexpectedResponse hr) {-# INLINE getResult #-} -getPreparedQueryId :: MonadThrow m => Response k a b -> m (QueryId k a b)-getPreparedQueryId r = do-    rs <- getResult r-    case rs of-        PreparedResult i _ _ -> return i-        _                    -> throwM $ UnexpectedResponse r+getPreparedQueryId :: MonadThrow m => HostResponse k a b -> m (Host, QueryId k a b)+getPreparedQueryId hr = do+    r <- getResult (hrResponse hr)+    case r of+        PreparedResult i _ _ -> return (hrHost hr, i)+        _                    -> throwM $ UnexpectedResponse (hrResponse hr) {-# INLINE getPreparedQueryId #-}  peer2Host :: PortNumber -> Peer -> Host
src/Database/CQL/IO/Cluster/Host.hs view
@@ -2,16 +2,33 @@ -- 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 CPP               #-} {-# LANGUAGE OverloadedStrings #-}  module Database.CQL.IO.Cluster.Host where  import Control.Lens ((^.), Lens') import Data.ByteString.Lazy.Char8 (unpack)+import Database.CQL.Protocol (Response (..))+import Data.IP import Data.Text (Text)-import Database.CQL.IO.Types (InetAddr)+import Network.Socket (SockAddr (..), PortNumber) import System.Logger.Message +-- | Host representation.+data Host = Host+    { _hostAddr   :: !InetAddr+    , _dataCentre :: !Text+    , _rack       :: !Text+    } deriving (Eq, Ord)++-- | A response that is known to originate from a specific+-- host of a cluster.+data HostResponse k a b = HostResponse+    { hrHost     :: !Host+    , hrResponse :: !(Response k a b)+    } deriving (Show)+ -- | This event will be passed to a 'Policy' to inform it about -- cluster changes. data HostEvent@@ -20,13 +37,6 @@     | HostUp   !InetAddr -- ^ a host has been started     | HostDown !InetAddr -- ^ a host has been stopped --- | Host representation.-data Host = Host-    { _hostAddr   :: !InetAddr-    , _dataCentre :: !Text-    , _rack       :: !Text-    } deriving (Eq, Ord)- -- | The IP address and port number of this host. hostAddr :: Lens' Host InetAddr hostAddr f ~(Host a c r) = fmap (\x -> Host x c r) (f a)@@ -47,3 +57,43 @@  instance ToBytes Host where     bytes h = h^.dataCentre +++ val ":" +++ h^.rack +++ val ":" +++ h^.hostAddr++-----------------------------------------------------------------------------+-- InetAddr++newtype InetAddr = InetAddr { sockAddr :: SockAddr } deriving (Eq, Ord)++instance Show InetAddr where+    show (InetAddr (SockAddrInet p a)) =+        let i = fromIntegral p :: Int in+        shows (fromHostAddress a) . showString ":" . shows i $ ""+    show (InetAddr (SockAddrInet6 p _ a _)) =+        let i = fromIntegral p :: Int in+        shows (fromHostAddress6 a) . showString ":" . shows i $ ""+    show (InetAddr (SockAddrUnix unix)) = unix+#if MIN_VERSION_network(2,6,1) && !MIN_VERSION_network(3,0,0)+    show (InetAddr (SockAddrCan int32)) = show int32+#endif++instance ToBytes InetAddr where+    bytes (InetAddr (SockAddrInet p a)) =+        let i = fromIntegral p :: Int in+        show (fromHostAddress a) +++ val ":" +++ i+    bytes (InetAddr (SockAddrInet6 p _ a _)) =+        let i = fromIntegral p :: Int in+        show (fromHostAddress6 a) +++ val ":" +++ i+    bytes (InetAddr (SockAddrUnix unix)) = bytes unix+#if MIN_VERSION_network(2,6,1) && !MIN_VERSION_network(3,0,0)+    bytes (InetAddr (SockAddrCan int32)) = bytes int32+#endif++ip2inet :: PortNumber -> IP -> InetAddr+ip2inet p (IPv4 a) = InetAddr $ SockAddrInet p (toHostAddress a)+ip2inet p (IPv6 a) = InetAddr $ SockAddrInet6 p 0 (toHostAddress6 a) 0++inet2ip :: InetAddr -> IP+inet2ip (InetAddr (SockAddrInet _ a))      = IPv4 (fromHostAddress a)+inet2ip (InetAddr (SockAddrInet6 _ _ a _)) = IPv6 (fromHostAddress6 a)+inet2ip _                                  = error "inet2Ip: not IP4/IP6 address"++
src/Database/CQL/IO/Cluster/Policies.hs view
@@ -17,7 +17,6 @@ import Data.Map.Strict (Map) import Data.Word import Database.CQL.IO.Cluster.Host-import Database.CQL.IO.Types (InetAddr) import System.Random.MWC import Prelude
src/Database/CQL/IO/Connection.hs view
@@ -43,6 +43,7 @@ import Data.Unique import Data.Vector (Vector, (!)) import Database.CQL.Protocol+import Database.CQL.IO.Cluster.Host import Database.CQL.IO.Connection.Socket (Socket) import Database.CQL.IO.Connection.Settings import Database.CQL.IO.Hexdump@@ -272,7 +273,7 @@     res <- request c enc     case parse (c^.settings.compression) res :: Raw Response of         RsReady _ _ Ready -> c^.eventSig |-> f-        other           -> throwM (UnexpectedResponse' other)+        other             -> throwM (UnexpectedResponse' other)  validateSettings :: MonadIO m => Connection -> m () validateSettings c = liftIO $ do@@ -287,7 +288,7 @@     res <- request c (serialise (c^.protocol) noCompression options)     case parse noCompression res :: Raw Response of         RsSupported _ _ x -> return x-        other           -> throwM (UnexpectedResponse' other)+        other             -> throwM (UnexpectedResponse' other)  useKeyspace :: MonadIO m => Connection -> Keyspace -> m () useKeyspace c ks = liftIO $ do@@ -298,7 +299,7 @@     res <- request c (serialise (c^.protocol) cmp req)     case parse cmp res :: Raw Response of         RsResult _ _ (SetKeyspaceResult _) -> return ()-        other                            -> throwM (UnexpectedResponse' other)+        other                              -> throwM (UnexpectedResponse' other)  query :: forall k a b m. (Tuple a, Tuple b, Show b, MonadIO m)       => Connection@@ -312,7 +313,7 @@     res <- request c enc     case parse (c^.settings.compression) res :: Response k a b of         RsResult _ _ (RowsResult _ b) -> return b-        other                       -> throwM (UnexpectedResponse' other)+        other                         -> throwM (UnexpectedResponse' other)   where     params = QueryParams cons False p Nothing Nothing Nothing Nothing
src/Database/CQL/IO/Connection/Settings.hs view
@@ -37,6 +37,7 @@ import Data.HashMap.Strict (HashMap) import Data.Int import Database.CQL.Protocol+import Database.CQL.IO.Cluster.Host import Database.CQL.IO.Types import OpenSSL.Session (SSLContext) import Prelude
src/Database/CQL/IO/Connection/Socket.hs view
@@ -23,6 +23,7 @@ import Data.ByteString.Builder import Data.Maybe (isJust) import Data.Monoid+import Database.CQL.IO.Cluster.Host import Database.CQL.IO.Types import Foreign.C.Types (CInt (..)) import Network.Socket hiding (Stream, Socket, connect, close, recv, send, shutdown)
src/Database/CQL/IO/Types.hs view
@@ -2,7 +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 CPP                        #-} {-# LANGUAGE DeriveDataTypeable         #-} {-# LANGUAGE GADTs                      #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-}@@ -13,14 +12,13 @@  import Control.Monad.Catch import Data.Hashable-import Data.IP import Data.String import Data.Text (Text) import Data.Typeable import Data.Unique+import Data.UUID+import Database.CQL.IO.Cluster.Host import Database.CQL.Protocol-import Network.Socket (SockAddr (..), PortNumber)-import System.Logger.Message  import qualified Data.Text.Lazy as Lazy @@ -39,44 +37,6 @@     hashWithSalt _ (ConnId u) = hashUnique u  -------------------------------------------------------------------------------- InetAddr--newtype InetAddr = InetAddr { sockAddr :: SockAddr } deriving (Eq, Ord)--instance Show InetAddr where-    show (InetAddr (SockAddrInet p a)) =-        let i = fromIntegral p :: Int in-        shows (fromHostAddress a) . showString ":" . shows i $ ""-    show (InetAddr (SockAddrInet6 p _ a _)) =-        let i = fromIntegral p :: Int in-        shows (fromHostAddress6 a) . showString ":" . shows i $ ""-    show (InetAddr (SockAddrUnix unix)) = unix-#if MIN_VERSION_network(2,6,1) && !MIN_VERSION_network(3,0,0)-    show (InetAddr (SockAddrCan int32)) = show int32-#endif--instance ToBytes InetAddr where-    bytes (InetAddr (SockAddrInet p a)) =-        let i = fromIntegral p :: Int in-        show (fromHostAddress a) +++ val ":" +++ i-    bytes (InetAddr (SockAddrInet6 p _ a _)) =-        let i = fromIntegral p :: Int in-        show (fromHostAddress6 a) +++ val ":" +++ i-    bytes (InetAddr (SockAddrUnix unix)) = bytes unix-#if MIN_VERSION_network(2,6,1) && !MIN_VERSION_network(3,0,0)-    bytes (InetAddr (SockAddrCan int32)) = bytes int32-#endif--ip2inet :: PortNumber -> IP -> InetAddr-ip2inet p (IPv4 a) = InetAddr $ SockAddrInet p (toHostAddress a)-ip2inet p (IPv6 a) = InetAddr $ SockAddrInet6 p 0 (toHostAddress6 a) 0--inet2ip :: InetAddr -> IP-inet2ip (InetAddr (SockAddrInet _ a))      = IPv4 (fromHostAddress a)-inet2ip (InetAddr (SockAddrInet6 _ _ a _)) = IPv6 (fromHostAddress6 a)-inet2ip _                                  = error "inet2Ip: not IP4/IP6 address"------------------------------------------------------------------------------- -- InvalidSettings  data InvalidSettings@@ -102,6 +62,25 @@     show (InternalError e) = "cql-io: internal error: " ++ show e  -----------------------------------------------------------------------------+-- ResponseError++data ResponseError = ResponseError+    { reHost  :: !Host+    , reTrace :: !(Maybe UUID)+    , reWarn  :: ![Text]+    , reCause :: !Error+    } deriving (Show, Typeable)++instance Exception ResponseError++toResponseError :: HostResponse k a b -> Maybe ResponseError+toResponseError (HostResponse h (RsError t w c)) = Just (ResponseError h t w c)+toResponseError _                                = Nothing++fromResponseError :: ResponseError -> HostResponse k a b+fromResponseError (ResponseError h t w c) = HostResponse h (RsError t w c)++----------------------------------------------------------------------------- -- HostError  data HostError@@ -147,8 +126,8 @@ data NoShow = NoShow deriving Show  data UnexpectedResponse where-    UnexpectedResponse  :: !(Response k a b) -> UnexpectedResponse-    UnexpectedResponse' :: Show b => !(Response k a b) -> UnexpectedResponse+    UnexpectedResponse     :: !(Response k a b) -> UnexpectedResponse+    UnexpectedResponse'    :: Show b => !(Response k a b) -> UnexpectedResponse  deriving instance Typeable UnexpectedResponse instance Exception UnexpectedResponse@@ -156,8 +135,8 @@ instance Show UnexpectedResponse where     show x = showString "cql-io: unexpected response: "            . case x of-                UnexpectedResponse  r  -> shows (f r)-                UnexpectedResponse' r  -> shows r+                UnexpectedResponse  r -> shows (f r)+                UnexpectedResponse' r -> shows r            $ ""       where         f :: Response k a b -> Response k a NoShow