cql-io 1.0.1.1 → 1.1.0

raw patch · 49 files changed

+4303/−3582 lines, 49 filesdep +cryptonitedep +primesdep +unliftio-coredep −cryptohashdep −monad-controldep −tinylogdep ~asyncdep ~basedep ~cqlsetup-changedPVP ok

version bump matches the API change (PVP)

Dependencies added: cryptonite, primes, unliftio-core

Dependencies removed: cryptohash, monad-control, tinylog, transformers-base

Dependency ranges changed: async, base, cql, iproute, network, raw-strings-qq, uuid

API changes (from Hackage documentation)

Files

• CHANGELOG +11/−0
• README.md +19/−0
• Setup.hs +0/−2
• cql-io.cabal +35/−16
• src/Database/CQL/IO.hs +0/−345
• src/Database/CQL/IO/Batch.hs +0/−81
• src/Database/CQL/IO/Client.hs +0/−702
• src/Database/CQL/IO/Cluster/Discovery.hs +0/−32
• src/Database/CQL/IO/Cluster/Host.hs +0/−99
• src/Database/CQL/IO/Cluster/Policies.hs +0/−153
• src/Database/CQL/IO/Connection.hs +0/−325
• src/Database/CQL/IO/Connection/Settings.hs +0/−141
• src/Database/CQL/IO/Connection/Socket.hs +0/−100
• src/Database/CQL/IO/Hexdump.hs +0/−63
• src/Database/CQL/IO/Jobs.hs +0/−69
• src/Database/CQL/IO/Pool.hs +0/−217
• src/Database/CQL/IO/PrepQuery.hs +0/−143
• src/Database/CQL/IO/Protocol.hs +0/−39
• src/Database/CQL/IO/Settings.hs +0/−269
• src/Database/CQL/IO/Signal.hs +0/−36
• src/Database/CQL/IO/Sync.hs +0/−56
• src/Database/CQL/IO/Tickets.hs +0/−47
• src/Database/CQL/IO/Timeouts.hs +0/−76
• src/Database/CQL/IO/Types.hs +0/−207
• src/api/Database/CQL/IO.hs +344/−0
• src/lib/Database/CQL/IO/Batch.hs +79/−0
• src/lib/Database/CQL/IO/Client.hs +806/−0
• src/lib/Database/CQL/IO/Cluster/Discovery.hs +33/−0
• src/lib/Database/CQL/IO/Cluster/Host.hs +102/−0
• src/lib/Database/CQL/IO/Cluster/Policies.hs +152/−0
• src/lib/Database/CQL/IO/Connection.hs +365/−0
• src/lib/Database/CQL/IO/Connection/Settings.hs +163/−0
• src/lib/Database/CQL/IO/Connection/Socket.hs +116/−0
• src/lib/Database/CQL/IO/Exception.hs +247/−0
• src/lib/Database/CQL/IO/Hexdump.hs +72/−0
• src/lib/Database/CQL/IO/Jobs.hs +139/−0
• src/lib/Database/CQL/IO/Log.hs +85/−0
• src/lib/Database/CQL/IO/Pool.hs +218/−0
• src/lib/Database/CQL/IO/PrepQuery.hs +144/−0
• src/lib/Database/CQL/IO/Protocol.hs +52/−0
• src/lib/Database/CQL/IO/Settings.hs +380/−0
• src/lib/Database/CQL/IO/Signal.hs +34/−0
• src/lib/Database/CQL/IO/Sync.hs +56/−0
• src/lib/Database/CQL/IO/Tickets.hs +47/−0
• src/lib/Database/CQL/IO/Timeouts.hs +77/−0
• src/test/Main.hs +14/−0
• src/test/Test/Database/CQL/IO.hs +365/−0
• src/test/Test/Database/CQL/IO/Jobs.hs +148/−0
• test/Main.hs +0/−364

@@ -1,3 +1,14 @@+1.1.0+-----+- Improve configuration of retry settings and refactor exceptions+  (https://gitlab.com/twittner/cql-io/issues/13).+- Document and export the 'getResult' function alongside the low-level+  query API.+- Replace monad-control with unliftio.+- Remove dependency on tinylog, introducing a minimal logging interface that can+  be hooked up to any logging library. Tinylog integration has moved to+  the new cql-io-tinylog library.+ 1.0.1.1 ------- - Add more documentation on queries.

@@ -32,3 +32,22 @@  Client to node communication can optionally use transport layer security (using HsOpenSSL).++License+=======++See [LICENSE](./LICENSE).++Cassandra Logo License+======================++Copyright © 2018 Apache Software Foundation++Licensed under the Apache License, Version 2.0 (the "License"); you may not use+this file except in compliance with the License. You may obtain a copy of the+License at https://www.apache.org/licenses/LICENSE-2.0. Unless required by+applicable law or agreed to in writing, software distributed under the License+is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY+KIND, either express or implied. See the License for the specific language+governing permissions and limitations under the License.+

@@ -1,2 +0,0 @@-import Distribution.Simple-main = defaultMain

@@ -1,5 +1,5 @@ name:                 cql-io-version:              1.0.1.1+version:              1.1.0 synopsis:             Cassandra CQL client. license:              MPL-2.0 license-file:         LICENSE@@ -11,7 +11,7 @@ bug-reports:          https://gitlab.com/twittner/cql-io/issues category:             Database build-type:           Simple-cabal-version:        >= 1.10+cabal-version:        >= 2.0 extra-source-files:   README.md                       CHANGELOG                       AUTHORS@@ -49,13 +49,26 @@  library     default-language: Haskell2010-    hs-source-dirs:   src+    hs-source-dirs:   src/api     ghc-options:      -Wall -O2 -fwarn-tabs      exposed-modules:         Database.CQL.IO -    other-modules:+    reexported-modules:+        Database.CQL.IO.Hexdump++    build-depends:+          base+        , cql-io-lib+        , cql++library cql-io-lib+    default-language: Haskell2010+    hs-source-dirs: src/lib+    ghc-options: -Wall -O2 -fwarn-tabs++    exposed-modules:         Database.CQL.IO.Batch         Database.CQL.IO.Client         Database.CQL.IO.Cluster.Discovery@@ -64,8 +77,10 @@         Database.CQL.IO.Connection         Database.CQL.IO.Connection.Socket         Database.CQL.IO.Connection.Settings+        Database.CQL.IO.Exception         Database.CQL.IO.Hexdump         Database.CQL.IO.Jobs+        Database.CQL.IO.Log         Database.CQL.IO.Pool         Database.CQL.IO.PrepQuery         Database.CQL.IO.Protocol@@ -74,23 +89,21 @@         Database.CQL.IO.Sync         Database.CQL.IO.Tickets         Database.CQL.IO.Timeouts-        Database.CQL.IO.Types      build-depends:-          async              >= 2.0+          async              >= 2.2         , auto-update        >= 0.1         , base               >= 4.9   && < 5.0         , bytestring         >= 0.10         , containers         >= 0.5         , cql                >= 4.0-        , cryptohash         >= 0.11+        , cryptonite         >= 0.13         , data-default-class         , exceptions         >= 0.4         , hashable           >= 1.2         , iproute            >= 1.3         , HsOpenSSL          >= 0.11         , lens               >= 4.4-        , monad-control      >= 0.3         , mtl                >= 2.1         , mwc-random         >= 0.13         , retry              >= 0.7@@ -98,10 +111,9 @@         , semigroups         >= 0.15         , stm                >= 2.4         , text               >= 0.11-        , tinylog            >= 0.8         , time               >= 1.4         , transformers       >= 0.3-        , transformers-base  >= 0.4+        , unliftio-core      >= 0.1.1         , unordered-containers >= 0.2         , uuid               >= 1.2.6         , vector             >= 0.10@@ -110,20 +122,27 @@     type:               exitcode-stdio-1.0     default-language:   Haskell2010     main-is:            Main.hs-    hs-source-dirs:     test+    hs-source-dirs:     src/test     ghc-options:        -threaded -Wall -O2 -fwarn-tabs     build-depends:-          base           >= 4.7+          base+        , async         , containers         , cql         , cql-io+        , cql-io-lib         , Decimal-        , iproute        >= 1.7+        , iproute         , mtl         , tasty          >= 0.11         , tasty-hunit    >= 0.9         , text-        , raw-strings-qq >= 1.1+        , primes+        , raw-strings-qq         , time-        , tinylog-        , uuid           >= 1.3+        , uuid++    other-modules:+        Test.Database.CQL.IO+        Test.Database.CQL.IO.Jobs+

@@ -1,345 +0,0 @@--- 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/.---- | This driver operates on some state which must be initialised prior to--- executing client operations and terminated eventually. The library uses--- <http://hackage.haskell.org/package/tinylog tinylog> for its logging--- output and expects a 'Logger'.------ For example (here using the @OverloadedStrings@ extension) :------ @--- > import Data.Text (Text)--- > import Data.Functor.Identity--- > import Database.CQL.IO as Client--- > import qualified System.Logger as Logger--- >--- > g <- Logger.new Logger.defSettings--- > c <- Client.init g defSettings--- > let q = "SELECT cql_version from system.local" :: QueryString R () (Identity Text)--- > let p = defQueryParams One ()--- > runClient c (query q p)--- [Identity "3.4.4"]--- > shutdown c--- @-----{-# LANGUAGE DeriveFunctor #-}-{-# LANGUAGE LambdaCase    #-}--module Database.CQL.IO-    ( -- * Client Settings-      Settings-    , S.defSettings-    , addContact-    , setCompression-    , setConnectTimeout-    , setContacts-    , setIdleTimeout-    , setKeyspace-    , setMaxConnections-    , setMaxStreams-    , setMaxTimeouts-    , setPolicy-    , setPoolStripes-    , setPortNumber-    , PrepareStrategy (..)-    , setPrepareStrategy-    , setProtocolVersion-    , setResponseTimeout-    , setSendTimeout-    , setRetrySettings-    , setMaxRecvBuffer-    , setSSLContext--      -- ** Authentication-    , setAuthentication-    , Authenticator (..)-    , AuthContext-    , ConnId-    , authConnId-    , authHost-    , AuthMechanism (..)-    , AuthUser      (..)-    , AuthPass      (..)-    , passwordAuthenticator--      -- ** Retry Settings-    , RetrySettings-    , noRetry-    , retryForever-    , maxRetries-    , adjustConsistency-    , constDelay-    , expBackoff-    , fibBackoff-    , adjustSendTimeout-    , adjustResponseTimeout--      -- ** Load-balancing-    , Policy (..)-    , random-    , roundRobin--      -- *** Hosts-    , Host-    , HostEvent (..)-    , InetAddr  (..)-    , hostAddr-    , dataCentre-    , rack--      -- * Client Monad-    , Client-    , MonadClient (..)-    , ClientState-    , DebugInfo   (..)-    , init-    , runClient-    , shutdown-    , debugInfo--      -- * Queries-      -- $queries-    , R, W, S-    , QueryParams       (..)-    , defQueryParams-    , Consistency       (..)-    , SerialConsistency (..)-    , Identity          (..)--      -- ** Basic Queries-    , QueryString (..)-    , query-    , query1-    , write-    , schema--      -- ** Prepared Queries-    , PrepQuery-    , prepared-    , queryString--      -- ** Paging-    , Page (..)-    , emptyPage-    , paginate--      -- ** Lightweight Transactions-    , Row-    , fromRow-    , trans--      -- ** Batch Queries-    , BatchM-    , addQuery-    , addPrepQuery-    , setType-    , setConsistency-    , setSerialConsistency-    , batch--      -- ** Retries-    , retry-    , once--      -- ** Low-Level Queries-      -- $low-level-queries-    , RunQ (..)-    , request--      -- * Exceptions-    , InvalidSettings     (..)-    , InternalError       (..)-    , HostError           (..)-    , ConnectionError     (..)-    , UnexpectedResponse  (..)-    , Timeout             (..)-    , HashCollision       (..)-    , AuthenticationError (..)-    ) where--import Control.Applicative-import Control.Monad.Catch-import Data.Functor.Identity-import Data.Maybe (isJust, listToMaybe)-import Database.CQL.Protocol-import Database.CQL.IO.Batch hiding (batch)-import Database.CQL.IO.Client-import Database.CQL.IO.Cluster.Host-import Database.CQL.IO.Cluster.Policies-import Database.CQL.IO.Connection.Settings as C-import Database.CQL.IO.PrepQuery-import Database.CQL.IO.Settings as S-import Database.CQL.IO.Types-import Prelude hiding (init)--import qualified Database.CQL.IO.Batch as B---- $queries------ Queries are defined either as 'QueryString's or 'PrepQuery's.--- Both types carry three phantom type parameters used to describe--- the query, input and output types, respectively, as follows:------   * @__k__@ is one of 'R'ead, 'W'rite or 'S'chema.---   * @__a__@ is the tuple type for the input, i.e. for the---     parameters bound by positional (@?@) or named (@:foo@) placeholders.---   * @__b__@ is the tuple type for the outputs, i.e. for the---     columns selected in a query.------ Thereby every type used in an input or output tuple must be an instance--- of the 'Cql' typeclass. It is the responsibility of user code--- that the type ascription of a query matches the order, number and types of--- the parameters. For example:------ @--- myQuery :: QueryString R (Identity UUID) (Text, Int, Maybe UTCTime)--- myQuery = "select name, age, birthday from user where id = ?"--- @------ In this example, the query is declared as a 'R'ead with a single--- input (id) and three outputs (name, age and birthday).------ Note that a single input or output type needs to be wrapped--- in the 'Identity' newtype, for which there is a `Cql` instance,--- in order to avoid overlapping instances.------ It is a common strategy to use additional @newtype@s with derived--- @Cql@ instances for additional type safety, e.g.------ @--- newtype UserId = UserId UUID deriving (Eq, Show, Cql)--- @------ The input and output tuples can further be automatically--- converted from and to records via the 'Database.CQL.Protocol.Record'--- typeclass, whose instances can be generated via @TemplateHaskell@,--- if desired.------ __Note on null values__------ In principle, any column in Cassandra is /nullable/, i.e. may be--- be set to @null@ as a result of row operations. It is therefore--- important that any output type of a query that may be null--- is wrapped in the 'Maybe' type constructor.--- It is a common pitfall that a column is assumed to never contain--- null values, when in fact partial updates or deletions on a row,--- including via the use of TTLs, may result in null values and thus--- runtime errors when processing the responses.---- $low-level-queries------ /Note/: Use of the these functions may require additional imports from--- @Database.CQL.Protocol@ or its submodules in order to construct--- 'Request's and evaluate 'Response's.---- | 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)--instance RunQ QueryString where-    runQ q p = request (RqQuery (Query q p))--instance RunQ PrepQuery where-    runQ q = liftClient . execute q---- | Construct default 'QueryParams' for the given consistency--- and bound values. In particular, no page size, paging state--- or serial consistency will be set.-defQueryParams :: Consistency -> a -> QueryParams a-defQueryParams c a = QueryParams-    { consistency       = c-    , values            = a-    , skipMetaData      = False-    , pageSize          = Nothing-    , queryPagingState  = Nothing-    , serialConsistency = Nothing-    , enableTracing     = Nothing-    }---- | Run a CQL read-only query returning a list of results.-query :: (MonadClient m, Tuple a, Tuple b, RunQ q) => q R a b -> QueryParams a -> m [b]-query q p = do-    r <- runQ q p-    getResult r >>= \case-        RowsResult _ b -> return b-        _              -> throwM $ UnexpectedResponse r---- | Run a CQL read-only query returning a single result.-query1 :: (MonadClient m, Tuple a, Tuple b, RunQ q) => q R a b -> QueryParams a -> m (Maybe b)-query1 q p = listToMaybe <$> query q p---- | Run a CQL write-only query (e.g. insert\/update\/delete),--- returning no result.------ /Note: If the write operation is conditional, i.e. is in fact a "lightweight--- transaction" returning a result, 'trans' must be used instead./-write :: (MonadClient m, Tuple a, RunQ q) => q W a () -> QueryParams a -> m ()-write q p = do-    r <- runQ q p-    getResult r >>= \case-        VoidResult -> return ()-        _          -> throwM $ UnexpectedResponse r---- | Run a CQL conditional write query (e.g. insert\/update\/delete) as a--- "lightweight transaction", returning the result 'Row's describing the--- outcome.-trans :: (MonadClient m, Tuple a, RunQ q) => q W a Row -> QueryParams a -> m [Row]-trans q p = do-    r <- runQ q p-    getResult r >>= \case-        RowsResult _ b -> return b-        _              -> 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)-schema q p = do-    r <- runQ q p-    getResult r >>= \case-        SchemaChangeResult s -> return $ Just s-        VoidResult           -> return Nothing-        _                    -> throwM $ UnexpectedResponse r---- | Run a batch query against a Cassandra node.-batch :: MonadClient m => BatchM () -> m ()-batch = liftClient . B.batch---- | Return value of 'paginate'. Contains the actual result values as well--- as an indication of whether there is more data available and the actual--- action to fetch the next page.-data Page a = Page-    { hasMore  :: !Bool-    , result   :: [a]-    , nextPage :: Client (Page a)-    } deriving (Functor)---- | A page with an empty result list.-emptyPage :: Page a-emptyPage = Page False [] (return emptyPage)---- | Run a CQL read-only query against a Cassandra node.------ This function is like 'query', but limits the result size to 10000--- (default) unless there is an explicit size restriction given in--- 'QueryParams'. The returned 'Page' can be used to continue the query.------ 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 :: (MonadClient m, Tuple a, Tuple b, RunQ q) => q R a b -> QueryParams a -> m (Page b)-paginate q p = do-    let p' = p { pageSize = pageSize p <|> Just 10000 }-    r <- runQ q p'-    getResult r >>= \case-        RowsResult m b ->-            if isJust (pagingState m) then-                return $ Page True b (paginate q p' { queryPagingState = pagingState m })-            else-                return $ Page False b (return emptyPage)-        _ -> throwM $ UnexpectedResponse r-

@@ -1,81 +0,0 @@--- 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 CPP                        #-}-{-# LANGUAGE GeneralizedNewtypeDeriving #-}-{-# LANGUAGE LambdaCase                 #-}--module Database.CQL.IO.Batch-    ( BatchM-    , batch-    , addQuery-    , addPrepQuery-    , setType-    , setConsistency-    , setSerialConsistency-    ) where--import Control.Applicative-import Control.Concurrent.STM (atomically)-import Control.Monad.Catch-import Control.Monad.IO.Class-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-import Prelude---- | 'Batch' construction monad.-newtype BatchM a = BatchM-    { unBatchM :: StateT Batch Client a-    } deriving (Functor, Applicative, Monad)---- | Execute the complete 'Batch' statement.-batch :: BatchM a -> Client ()-batch m = do-    b <- execStateT (unBatchM m) (Batch BatchLogged [] Quorum Nothing)-    r <- executeWithPrepare Nothing (RqBatch b :: Raw Request)-    getResult (hrResponse r) >>= \case-        VoidResult -> return ()-        _          -> throwM $ UnexpectedResponse (hrResponse r)---- | Add a query to this batch.-addQuery :: (Show a, Tuple a, Tuple b) => QueryString W a b -> a -> BatchM ()-addQuery q p = BatchM $ modify' $ \b ->-    b { batchQuery = BatchQuery q p : batchQuery b }---- | Add a prepared query to this batch.-addPrepQuery :: (Show a, Tuple a, Tuple b) => PrepQuery W a b -> a -> BatchM ()-addPrepQuery q p = BatchM $ do-    pq <- lift preparedQueries-    maybe (fresh pq) add =<< liftIO (atomically (lookupQueryId q pq))-  where-    fresh pq = do-        i <- snd <$> lift (prepare Nothing (queryString q))-        liftIO $ atomically (insert q i pq)-        add i--    add i = modify' $ \b -> b { batchQuery = BatchPrepared i p : batchQuery b }---- | Set the type of this batch.-setType :: BatchType -> BatchM ()-setType t = BatchM $ modify' $ \b -> b { batchType = t }---- | Set 'Batch' consistency level.-setConsistency :: Consistency -> BatchM ()-setConsistency c = BatchM $ modify' $ \b -> b { batchConsistency = c }---- | Set 'Batch' serial consistency.-setSerialConsistency :: SerialConsistency -> BatchM ()-setSerialConsistency c = BatchM $ modify' $ \b -> b { batchSerialConsistency = Just c }--#if ! MIN_VERSION_transformers(0,4,0)-modify' :: Monad m => (s -> s) -> StateT s m ()-modify' f = do-    s <- get-    put $! f s-#endif

@@ -1,702 +0,0 @@--- 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 FlexibleContexts           #-}-{-# LANGUAGE FlexibleInstances          #-}-{-# LANGUAGE GeneralizedNewtypeDeriving #-}-{-# LANGUAGE MultiParamTypeClasses      #-}-{-# LANGUAGE OverloadedStrings          #-}-{-# LANGUAGE ScopedTypeVariables        #-}-{-# LANGUAGE TemplateHaskell            #-}-{-# LANGUAGE TupleSections              #-}-{-# LANGUAGE TypeFamilies               #-}-{-# LANGUAGE UndecidableInstances       #-}-{-# LANGUAGE ViewPatterns               #-}--module Database.CQL.IO.Client-    ( Client-    , MonadClient (..)-    , ClientState-    , DebugInfo   (..)-    , runClient-    , init-    , shutdown-    , request-    , requestN-    , request1-    , execute-    , executeWithPrepare-    , prepare-    , retry-    , once-    , debugInfo-    , preparedQueries-    , withPrepareStrategy-    , getResult-    ) where--import Control.Applicative-import Control.Concurrent (threadDelay)-import Control.Concurrent.Async (async, wait)-import Control.Concurrent.STM hiding (retry)-import Control.Exception (IOException)-import Control.Lens hiding ((.=), Context)-import Control.Monad (void, when)-import Control.Monad.Base (MonadBase (..))-import Control.Monad.Catch-import Control.Monad.IO.Class-import Control.Monad.Reader (ReaderT (..), runReaderT, MonadReader, ask)-import Control.Monad.Trans.Class-import Control.Monad.Trans.Control (MonadBaseControl (..))-#if MIN_VERSION_transformers(0,4,0)-import Control.Monad.Trans.Except-#endif-import Control.Retry (capDelay, exponentialBackoff, rsIterNumber)-import Control.Retry (recovering)-import Data.Foldable (for_, foldrM)-import Data.List (find)-import Data.List.NonEmpty (NonEmpty (..))-import Data.Map.Strict (Map)-import Data.Maybe (fromMaybe, listToMaybe)-import Data.Text (Text)-import Data.Word-import Database.CQL.IO.Cluster.Discovery as Discovery-import Database.CQL.IO.Cluster.Host-import Database.CQL.IO.Cluster.Policies-import Database.CQL.IO.Connection hiding (request)-import Database.CQL.IO.Connection.Settings-import Database.CQL.IO.Jobs (Jobs)-import Database.CQL.IO.Pool-import Database.CQL.IO.PrepQuery (PrepQuery, PreparedQueries)-import Database.CQL.IO.Protocol-import Database.CQL.IO.Settings-import Database.CQL.IO.Signal-import Database.CQL.IO.Timeouts (TimeoutManager)-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 hiding (init)--import qualified Control.Monad.Reader       as Reader-import qualified Control.Monad.State.Strict as S-import qualified Control.Monad.State.Lazy   as LS-import qualified Data.List.NonEmpty         as NE-import qualified Data.Map.Strict            as Map-import qualified Database.CQL.IO.Connection as C-import qualified Database.CQL.IO.Jobs       as Jobs-import qualified Database.CQL.IO.PrepQuery  as PQ-import qualified Database.CQL.IO.Timeouts   as TM-import qualified Database.CQL.Protocol      as Cql-import qualified System.Logger              as Logger--data ControlState-    = Disconnected-    | Connected-    | Reconnecting-    deriving (Eq, Ord, Show)--data Control = Control-    { _state      :: !ControlState-    , _connection :: !Connection-    }--data Context = Context-    { _settings      :: !Settings-    , _logger        :: !Logger-    , _timeouts      :: !TimeoutManager-    , _sigMonit      :: !(Signal HostEvent)-    }---- | Opaque client state/environment.-data ClientState = ClientState-    { _context     :: !Context-    , _policy      :: !Policy-    , _prepQueries :: !PreparedQueries-    , _control     :: !(TVar Control)-    , _hostmap     :: !(TVar (Map Host Pool))-    , _jobs        :: !(Jobs InetAddr)-    }--makeLenses ''Control-makeLenses ''Context-makeLenses ''ClientState---- | The Client monad.------ A simple reader monad around some internal state. Prior to executing--- this monad via 'runClient', its state must be initialised through--- 'Database.CQL.IO.Client.init' and after finishing operation it should be--- terminated with 'shutdown'.------ To lift 'Client' actions into another monad, see 'MonadClient'.-newtype Client a = Client-    { client :: ReaderT ClientState IO a-    } deriving ( Functor-               , Applicative-               , Monad-               , MonadIO-               , MonadThrow-               , MonadMask-               , MonadCatch-               , MonadReader ClientState-               , MonadBase IO-               )--instance MonadLogger Client where-    log l m = view (context.logger) >>= \g -> Logger.log g l m--#if MIN_VERSION_monad_control(1,0,0)-instance MonadBaseControl IO Client where-    type StM Client a = StM (ReaderT ClientState IO) a-    liftBaseWith f = Client $ liftBaseWith $ \run -> f (run . client)-    restoreM = Client . restoreM-#else-instance MonadBaseControl IO Client where-    newtype StM Client a = ClientStM-        { unClientStM :: StM (ReaderT ClientState IO) a }--    liftBaseWith f =-        Client $ liftBaseWith $ \run -> f (fmap ClientStM . run . client)--    restoreM = Client . restoreM . unClientStM-#endif---- | 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 from 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--instance MonadClient m => MonadClient (ReaderT r m) where-    liftClient     = lift . liftClient-    localState f m = ReaderT (localState f . runReaderT m)--instance MonadClient m => MonadClient (S.StateT s m) where-    liftClient     = lift . liftClient-    localState f m = S.StateT (localState f . S.runStateT m)--instance MonadClient m => MonadClient (LS.StateT s m) where-    liftClient     = lift . liftClient-    localState f m = LS.StateT (localState f . LS.runStateT m)--#if MIN_VERSION_transformers(0,4,0)-instance MonadClient m => MonadClient (ExceptT e m) where-    liftClient     = lift . liftClient-    localState f m = ExceptT $ localState f (runExceptT m)-#endif---------------------------------------------------------------------------------- API---- | Execute the client monad.-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)---- | Execute a client action once, without retries, i.e.------ @once action = retry noRetry action@.------ Primarily for use in applications where global 'RetrySettings'--- are configured and need to be selectively disabled for individual--- queries.-once :: MonadClient m => m a -> m a-once = retry noRetry---- | Change the default 'PrepareStrategy' for the given client action.-withPrepareStrategy :: MonadClient m => PrepareStrategy -> m a -> m a-withPrepareStrategy s = localState (set (context.settings.prepStrategy) s)---- | Send a 'Request' to the server and return a 'Response'.------ This function will first ask the clients load-balancing 'Policy' for--- some host and use its connection pool to acquire a connection for--- request transmission.------ 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.------ The request is retried according to the configured 'RetrySettings'.-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-    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 (HostResponse k a b))-requestN !n a s = do-    hst <- pickHost (s^.policy)-    res <- request1 hst a s-    case res of-        Just  _ -> return res-        Nothing -> if n > 0 then requestN (n - 1) a s else return Nothing-  where-    pickHost p = maybe (throwM NoHostAvailable) return =<< liftIO (select p)---- | 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 (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 $ \(HostResponse _ r) ->-                for_ (Cql.warnings r) $ \w ->-                    warn $ msg (val "server warning: " +++ w)-            return result-        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)-            request1 h a s-  where-    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)-        r `seq` return (HostResponse h r)--    handlers =-        [ 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--- 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 (HostResponse k a b)-executeWithPrepare h q = do-    f <- selectAction h-    r <- withRetries f q-    case hrResponse r of-        RsError _ _ (Unprepared _ i) -> do-            pq <- preparedQueries-            qs <- atomically' (PQ.lookupQueryString (QueryId i) pq)-            case qs of-                Nothing -> throwM $ InternalError "Unknown QueryID returned from server"-                Just  s -> do-                    (g, _) <- prepare (Just LazyPrepare) (s :: Raw QueryString)-                    executeWithPrepare (Just g) q-        _ -> return r-  where-    selectAction Nothing  = view policy >>= liftIO . hostCount >>= return . requestN-    selectAction (Just x) = return (request1 x)---- | Prepare the given query according to the given--- 'PrepareStrategy', returning the resulting 'QueryId'--- and 'Host' which was used for preparation.-prepare :: (Tuple b, Tuple a) => Maybe PrepareStrategy -> QueryString k a b -> Client (Host, QueryId k a b)-prepare (Just LazyPrepare) qs = do-    s <- ask-    n <- liftIO $ hostCount (s^.policy)-    r <- withRetries (requestN n) (RqPrepare (Prepare qs))-    getPreparedQueryId r--prepare (Just EagerPrepare) qs = view policy-    >>= liftIO . current-    >>= mapM (action (RqPrepare (Prepare qs)))-    >>= first-  where-    action rq h = do-        r <- withRetries (request1 h) rq-        getPreparedQueryId r--    first (x:_) = return x-    first []    = throwM NoHostAvailable--prepare Nothing qs = do-    ps <- view (context.settings.prepStrategy)-    prepare (Just ps) qs---- | Execute a prepared query (transparently re-preparing if necessary).-execute :: (Tuple b, Tuple a) => PrepQuery k a b -> QueryParams a -> Client (Response k a b)-execute q p = do-    pq <- view prepQueries-    maybe (new pq) (run Nothing) =<< atomically' (PQ.lookupQueryId q pq)-  where-    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)-        run (Just h) i--data DebugInfo = DebugInfo-    { policyInfo :: String     -- ^ 'Policy' string representation-    , jobInfo    :: [InetAddr] -- ^ hosts currently checked for reachability-    , hostInfo   :: [Host]     -- ^ all known hosts-    }--instance Show DebugInfo where-    show dbg = showString "running jobs: "-             . shows (jobInfo dbg)-             . showString "\nknown hosts: "-             . shows (hostInfo dbg)-             . showString "\npolicy info: "-             . shows (policyInfo dbg)-             $ ""--debugInfo :: MonadClient m => m DebugInfo-debugInfo = liftClient $ do-    hosts <- Map.keys <$> (readTVarIO' =<< view hostmap)-    pols  <- liftIO . display =<< view policy-    jbs   <- Jobs.showJobs =<< view jobs-    return $ DebugInfo pols jbs hosts--preparedQueries :: Client PreparedQueries-preparedQueries = view prepQueries---------------------------------------------------------------------------------- Initialisation---- | Initialise client state with the given 'Settings' using the provided--- 'Logger' for all it's logging output.-init :: MonadIO m => Logger -> Settings -> m ClientState-init g s = liftIO $ do-    t <- TM.create 250-    c <- tryAll (s^.contacts) (mkConnection t)-    e <- Context s g t <$> signal-    p <- s^.policyMaker-    x <- ClientState e-            <$> pure p-            <*> PQ.new-            <*> newTVarIO (Control Connected c)-            <*> newTVarIO Map.empty-            <*> Jobs.new-    e^.sigMonit |-> onEvent p-    runClient x (initialise c) `onException` liftIO (C.close c)-    return x-  where-    mkConnection t h = do-        as <- C.resolve h (s^.portnumber)-        NE.fromList as `tryAll` doConnect t--    doConnect t a = do-        Logger.debug g $ msg (val "connecting to " +++ a)-        c <- C.connect (s^.connSettings) t (s^.protoVersion) g a-        Logger.info g $ msg (val "control connection: " +++ c)-        return c--initialise :: Connection -> Client ()-initialise c = do-    startup c-    env <- ask-    pol <- view policy-    ctx <- view context-    l <- local2Host (c^.address) . listToMaybe <$> query c One Discovery.local ()-    r <- discoverPeers ctx c-    (u, d) <- mkHostMap ctx pol (l:r)-    m <- view hostmap-    let h = Map.union u d-    atomically' $ writeTVar m h-    liftIO $ setup pol (Map.keys u) (Map.keys d)-    register c allEventTypes (runClient env . onCqlEvent)-    info $ msg (val "known hosts: " +++ show (Map.keys h))-    j <- view jobs-    for_ (Map.keys d) $ \down ->-        Jobs.add j (down^.hostAddr) True $ monitor ctx 1000000 60000000 down--discoverPeers :: MonadIO m => Context -> Connection -> m [Host]-discoverPeers ctx c = liftIO $ do-    let p = ctx^.settings.portnumber-    map (peer2Host p . asRecord) <$> query c One peers ()--mkHostMap :: Context -> Policy -> [Host] -> Client (Map Host Pool, Map Host Pool)-mkHostMap c p = liftIO . foldrM checkHost (Map.empty, Map.empty)-  where-    checkHost h (up, down) = do-        okay <- acceptable p h-        if okay then do-            isUp <- C.ping (h^.hostAddr)-            if isUp then do-                up' <- Map.insert h <$> mkPool c (h^.hostAddr) <*> pure up-                return (up', down)-            else do-                down' <- Map.insert h <$> mkPool c (h^.hostAddr) <*> pure down-                return (up, down')-        else-            return (up, down)--mkPool :: MonadIO m => Context -> InetAddr -> m Pool-mkPool ctx i = liftIO $ do-    let s = ctx^.settings-    let m = s^.connSettings.maxStreams-    create (connOpen s) connClose (ctx^.logger) (s^.poolSettings) m-  where-    connOpen s = do-        let g = ctx^.logger-        c <- C.connect (s^.connSettings) (ctx^.timeouts) (s^.protoVersion) g i-        Logger.debug g $ "client.connect" .= c-        connInit c `onException` connClose c-        return c--    connInit con = do-        C.startup con-        for_ (ctx^.settings.connSettings.defKeyspace) $-            C.useKeyspace con--    connClose con = do-        Logger.debug (ctx^.logger) $ "client.close" .= con-        C.close con---------------------------------------------------------------------------------- Termination---- | Terminate client state, i.e. end all running background checks and--- shutdown all connection pools.  Once this is entered, the client--- will eventually be shut down, though an asynchronous exception can--- interrupt the wait for that to occur.-shutdown :: MonadIO m => ClientState -> m ()-shutdown s = liftIO $ asyncShutdown >>= wait-  where-    asyncShutdown = async $ do-        TM.destroy (s^.context.timeouts) True-        Jobs.destroy (s^.jobs)-        ignore $ C.close . view connection =<< readTVarIO (s^.control)-        mapM_ destroy . Map.elems =<< readTVarIO (s^.hostmap)---------------------------------------------------------------------------------- Monitoring--monitor :: Context -> Int -> Int -> Host -> IO ()-monitor ctx initial upperBound h = do-    threadDelay initial-    Logger.info (ctx^.logger) $ msg (val "monitoring: " +++ h)-    hostCheck 0 maxN-  where-    hostCheck :: Int -> Int -> IO ()-    hostCheck n mx = do-        threadDelay (2^(min n mx) * 50000)-        isUp <- C.ping (h^.hostAddr)-        if isUp then do-            ctx^.sigMonit $$ HostUp (h^.hostAddr)-            Logger.info (ctx^.logger) $ msg (val "reachable: " +++ h)-        else do-            Logger.info (ctx^.logger) $ msg (val "unreachable: " +++ h)-            hostCheck (n + 1) mx--    maxN :: Int-    maxN = floor . logBase 2 $ (fromIntegral (upperBound `div` 50000) :: Double)---------------------------------------------------------------------------------- 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 (HostResponse k a b)))-    -> Request k a b-    -> Client (HostResponse k a b)-withRetries fn a = do-    s <- ask-    let p = s^.context.settings.retrySettings.retryPolicy-    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-            -- [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-            y = s^.context.settings.retrySettings.recvTimeoutChange-        in over (context.settings.connSettings.sendTimeout)     (+ x)-         . over (context.settings.connSettings.responseTimeout) (+ y)-         $ 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 })-                    RqExecute (Execute q p) -> RqExecute (Execute q p { consistency = c })-                    RqBatch b               -> RqBatch b { batchConsistency = c }-                    _                       -> a--    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-        , const $ Handler $ \(_ :: SomeSSLException) -> return True-        ]--onConnectionError :: Exception e => Host -> e -> Client ()-onConnectionError h exc = do-    warn $ "exception" .= show exc-    e <- ask-    c <- atomically' $ do-        ctrl <- readTVar (e^.control)-        let a = ctrl^.connection.address-        if ctrl^.state == Connected && a == h^.hostAddr then do-            writeTVar (e^.control) (set state Reconnecting ctrl)-            return $ Just (ctrl^.connection)-        else-            return Nothing-    maybe (liftIO . ignore . onEvent (e^.policy) $ HostDown (h^.hostAddr))-          (liftIO . void . async . recovering reconnectPolicy reconnectHandlers . const . continue e)-          c-    Jobs.add (e^.jobs) (h^.hostAddr) True $-        monitor (e^.context) 0 30000000 h-  where-    continue e conn = do-        Jobs.destroy (e^.jobs)-        ignore $ C.close conn-        ignore $ onEvent (e^.policy) (HostDown (h^.hostAddr))-        x <- NE.nonEmpty . map (view hostAddr) . Map.keys <$> readTVarIO (e^.hostmap)-        case x of-            Just  a -> a `tryAll` (runClient e . replaceControl) `onException` reconnect e a-            Nothing -> do-                atomically $ modifyTVar' (e^.control) (set state Disconnected)-                Logger.fatal (e^.context.logger) $ "error-handler" .= val "no host available"--    reconnect e a = do-        Logger.info (e^.context.logger) $ msg (val "reconnecting control ...")-        a `tryAll` (runClient e . replaceControl)--    reconnectPolicy = capDelay 5000000 (exponentialBackoff 5000)--    reconnectHandlers =-        [ const (Handler $ \(_ :: IOException)      -> return True)-        , const (Handler $ \(_ :: ConnectionError)  -> return True)-        , const (Handler $ \(_ :: HostError)        -> return True)-        , const (Handler $ \(_ :: SomeSSLException) -> return True)-        ]--replaceControl :: InetAddr -> Client ()-replaceControl a = do-    ctx <- view context-    ctl <- view control-    let s = ctx^.settings-    c <- C.connect (s^.connSettings) (ctx^.timeouts) (s^.protoVersion) (ctx^.logger) a-    initialise c `onException` liftIO (C.close c)-    atomically' $ writeTVar ctl (Control Connected c)-    info $ msg (val "new control connection: " +++ c)---------------------------------------------------------------------------------- Event handling--onCqlEvent :: Event -> Client ()-onCqlEvent x = do-    info $ "client.event" .= show x-    pol <- view policy-    prt <- view (context.settings.portnumber)-    case x of-        StatusEvent Down (mapAddr prt -> a) ->-            liftIO $ onEvent pol (HostDown a)-        TopologyEvent RemovedNode (mapAddr prt -> a) -> do-            hmap <- view hostmap-            atomically' $-                modifyTVar' hmap (Map.filterWithKey (\h _ -> h^.hostAddr /= a))-            liftIO $ onEvent pol $ HostGone a-        StatusEvent Up (mapAddr prt -> a) -> do-            s <- ask-            startMonitor s a-        TopologyEvent NewNode (mapAddr prt -> a) -> do-            s <- ask-            let ctx  = s^.context-            let hmap = s^.hostmap-            ctrl <- readTVarIO' (s^.control)-            let c = ctrl^.connection-            h    <- fromMaybe (Host a "" "") . find ((a == ) . view hostAddr) <$> discoverPeers' ctx c-            okay <- liftIO $ acceptable pol h-            when okay $ do-                p <- mkPool ctx (h^.hostAddr)-                atomically' $ modifyTVar' hmap (Map.alter (maybe (Just p) Just) h)-                liftIO $ onEvent pol (HostNew h)-                Jobs.add (s^.jobs) a False $ runClient s (prepareAllQueries h)-        SchemaEvent _ -> return ()-  where-    mapAddr i (SockAddrInet _ a)      = InetAddr (SockAddrInet i a)-    mapAddr i (SockAddrInet6 _ f a b) = InetAddr (SockAddrInet6 i f a b)-    mapAddr _ unix                    = InetAddr unix--    discoverPeers' ctx c = discoverPeers ctx c `catchAll` const (return [])--    startMonitor s a = do-        hmp <- readTVarIO' (s^.hostmap)-        case find ((a ==) . view hostAddr) (Map.keys hmp) of-            Just h -> Jobs.add (s^.jobs) a False $ do-                monitor (s^.context) 3000000 60000000 h-                runClient s (prepareAllQueries h)-            Nothing -> return ()--prepareAllQueries :: Host -> Client ()-prepareAllQueries h = do-    pq <- view prepQueries-    qs <- atomically' $ PQ.queryStrings pq-    for_ qs $ \q ->-        let qry = QueryString q :: Raw QueryString in-        withRetries (request1 h) (RqPrepare (Prepare qry))---------------------------------------------------------------------------------- Utilities--getResult :: MonadThrow m => Response k a b -> m (Result k a b)-getResult (RsResult _ _ r) = return r-getResult (RsError  _ _ e) = throwM e-getResult hr               = throwM (UnexpectedResponse hr)-{-# INLINE getResult #-}--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-peer2Host i p = Host (ip2inet i (peerRPC p)) (peerDC p) (peerRack p)--local2Host :: InetAddr -> Maybe (Text, Text) -> Host-local2Host i (Just (dc, rk)) = Host i dc rk-local2Host i Nothing         = Host i "" ""--allEventTypes :: [EventType]-allEventTypes = [TopologyChangeEvent, StatusChangeEvent, SchemaChangeEvent]--tryAll :: NonEmpty a -> (a -> IO b) -> IO b-tryAll (a :| []) f = f a-tryAll (a :| aa) f = f a `catchAll` const (tryAll (NE.fromList aa) f)--atomically' :: STM a -> Client a-atomically' = liftIO . atomically--readTVarIO' :: TVar a -> Client a-readTVarIO' = liftIO . readTVarIO

@@ -1,32 +0,0 @@--- 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 OverloadedStrings #-}-{-# LANGUAGE TemplateHaskell   #-}-{-# LANGUAGE TypeFamilies      #-}--module Database.CQL.IO.Cluster.Discovery where--import Data.Functor.Identity (Identity)-import Data.IP-import Data.Text (Text)-import Database.CQL.Protocol--data Peer = Peer-    { peerAddr :: !IP-    , peerRPC  :: !IP-    , peerDC   :: !Text-    , peerRack :: !Text-    } deriving Show--recordInstance ''Peer--peers :: QueryString R () (IP, IP, Text, Text)-peers = "SELECT peer, rpc_address, data_center, rack FROM system.peers"--peer :: QueryString R (Identity IP) (IP, IP, Text, Text)-peer = "SELECT peer, rpc_address, data_center, rack FROM system.peers where peer = ?"--local :: QueryString R () (Text, Text)-local = "SELECT data_center, rack FROM system.local WHERE key='local'"

@@ -1,99 +0,0 @@--- 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 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 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-    = HostNew  !Host     -- ^ a new host has been added to the cluster-    | HostGone !InetAddr -- ^ a host has been removed from the cluster-    | HostUp   !InetAddr -- ^ a host has been started-    | HostDown !InetAddr -- ^ a host has been stopped---- | 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)-{-# INLINE hostAddr #-}---- | The data centre name (may be an empty string).-dataCentre :: Lens' Host Text-dataCentre f ~(Host a c r) = fmap (\x -> Host a x r) (f c)-{-# INLINE dataCentre #-}---- | The rack name (may be an empty string).-rack :: Lens' Host Text-rack f ~(Host a c r) = fmap (\x -> Host a c x) (f r)-{-# INLINE rack #-}--instance Show Host where-    show = unpack . eval . bytes--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"--

@@ -1,153 +0,0 @@--- 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 TemplateHaskell #-}--module Database.CQL.IO.Cluster.Policies-    ( Policy (..)-    , random-    , roundRobin-    ) where--import Control.Applicative-import Control.Concurrent.STM-import Control.Lens ((^.), view, over, makeLenses)-import Control.Monad-import Data.Map.Strict (Map)-import Data.Word-import Database.CQL.IO.Cluster.Host-import System.Random.MWC-import Prelude--import qualified Data.Map.Strict as Map---- | A policy defines a load-balancing strategy and generally--- handles host visibility.-data Policy = Policy-    { setup :: [Host] -> [Host] -> IO ()-      -- ^ Initialise the policy with two sets of hosts. The first-      -- parameter are hosts known to be available, the second are other-      -- nodes.-      -- Please note that a policy may be re-initialised at any point-      -- through this method.-    , onEvent :: HostEvent -> IO ()-      -- ^ Event handler. Policies will be informed about cluster changes-      -- through this function.-    , select :: IO (Maybe Host)-      -- ^ Host selection. The driver will ask for a host to use in a query-      -- through this function. A policy which has no available nodes may-      -- return Nothing.-    , current :: IO [Host]-      -- ^ Return all currently alive hosts.-    , acceptable :: Host -> IO Bool-      -- ^ During startup and node discovery, the driver will ask the-      -- policy if a dicovered host should be ignored.-    , hostCount :: IO Word-      -- ^ During query processing, the driver will ask the policy for-      -- a rough esitimate of alive hosts. The number is used to repeatedly-      -- invoke 'select' (with the underlying assumption that the policy-      -- returns mostly different hosts).-    , display :: IO String-      -- ^ Like having an effectful 'Show' instance for this policy.-    }--type HostMap = TVar Hosts--data Hosts = Hosts-    { _alive :: !(Map InetAddr Host)-    , _other :: !(Map InetAddr Host)-    } deriving Show--makeLenses ''Hosts---- | Iterate over hosts one by one.-roundRobin :: IO Policy-roundRobin = do-    h <- newTVarIO emptyHosts-    c <- newTVarIO 0-    return $ Policy (defSetup h) (defOnEvent h) (pickHost h c)-                    (defCurrent h) defAcceptable (defHostCount h)-                    (defDisplay h)-  where-    pickHost h c = atomically $ do-        m <- view alive <$> readTVar h-        if Map.null m then-            return Nothing-        else do-            k <- readTVar c-            writeTVar c $ succ k `mod` Map.size m-            return . Just . snd $ Map.elemAt (k `mod` Map.size m) m---- | Return hosts in random order.-random :: IO Policy-random = do-    h <- newTVarIO emptyHosts-    g <- createSystemRandom-    return $ Policy (defSetup h) (defOnEvent h) (pickHost h g)-                    (defCurrent h) defAcceptable (defHostCount h)-                    (defDisplay h)-  where-    pickHost h g = do-        m <- view alive <$> readTVarIO h-        if Map.null m then-            return Nothing-        else do-            let i = uniformR (0, Map.size m - 1) g-            Just . snd . flip Map.elemAt m <$> i---------------------------------------------------------------------------------- Defaults--emptyHosts :: Hosts-emptyHosts = Hosts Map.empty Map.empty--defDisplay :: HostMap -> IO String-defDisplay h = show <$> readTVarIO h--defAcceptable :: Host -> IO Bool-defAcceptable = const $ return True--defSetup :: HostMap -> [Host] -> [Host] -> IO ()-defSetup r a b = do-    let ha = Map.fromList $ zip (map (view hostAddr) a) a-    let hb = Map.fromList $ zip (map (view hostAddr) b) b-    let hosts = Hosts ha hb-    atomically $ writeTVar r hosts--defHostCount :: HostMap -> IO Word-defHostCount r = fromIntegral . Map.size . view alive <$> readTVarIO r--defCurrent :: HostMap -> IO [Host]-defCurrent r = Map.elems . view alive <$> readTVarIO r--defOnEvent :: HostMap -> HostEvent -> IO ()-defOnEvent r (HostNew h) = atomically $ do-    m <- readTVar r-    when (Nothing == get (h^.hostAddr) m) $-        writeTVar r (over alive (Map.insert (h^.hostAddr) h) m)-defOnEvent r (HostGone a) = atomically $ do-    m <- readTVar r-    if Map.member a (m^.alive) then-        writeTVar r (over alive (Map.delete a) m)-    else-        writeTVar r (over other (Map.delete a) m)-defOnEvent r (HostUp a) = atomically $ do-    m <- readTVar r-    case get a m of-        Nothing -> return ()-        Just  h -> writeTVar r-            $ over alive (Map.insert a h)-            . over other (Map.delete a)-            $ m-defOnEvent r (HostDown a) = atomically $ do-    m <- readTVar r-    case get a m of-        Nothing -> return ()-        Just  h -> writeTVar r-            $ over other (Map.insert a h)-            . over alive (Map.delete a)-            $ m--get :: InetAddr -> Hosts -> Maybe Host-get a m = Map.lookup a (m^.alive) <|> Map.lookup a (m^.other)

@@ -1,325 +0,0 @@--- 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 LambdaCase          #-}-{-# LANGUAGE OverloadedStrings   #-}-{-# LANGUAGE ScopedTypeVariables #-}-{-# LANGUAGE TemplateHaskell     #-}-{-# LANGUAGE ViewPatterns        #-}--module Database.CQL.IO.Connection-    ( Connection-    , ConnId-    , resolve-    , ping-    , connect-    , close-    , request-    , startup-    , register-    , query-    , useKeyspace-    , address-    , protocol-    , eventSig-    ) where--import Control.Applicative-import Control.Concurrent (myThreadId, forkIOWithUnmask)-import Control.Concurrent.Async-import Control.Concurrent.MVar-import Control.Concurrent.STM-import Control.Exception (throwTo, AsyncException (ThreadKilled))-import Control.Lens ((^.), makeLenses, view)-import Control.Monad-import Control.Monad.Catch-import Control.Monad.IO.Class-import Data.ByteString.Lazy (ByteString)-import Data.Int-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.Cluster.Host-import Database.CQL.IO.Connection.Socket (Socket)-import Database.CQL.IO.Connection.Settings-import Database.CQL.IO.Hexdump-import Database.CQL.IO.Protocol-import Database.CQL.IO.Signal hiding (connect)-import Database.CQL.IO.Sync (Sync)-import Database.CQL.IO.Types-import Database.CQL.IO.Tickets (Pool, toInt, markAvailable)-import Database.CQL.IO.Timeouts (TimeoutManager, withTimeout)-import Network.Socket hiding (Socket, close, connect, send)-import System.IO (nativeNewline, Newline (..))-import System.Logger hiding (Settings, close, defSettings, settings)-import System.Timeout-import Prelude--import qualified Data.ByteString.Lazy              as L-import qualified Data.ByteString.Lazy.Char8        as Char8-import qualified Data.HashMap.Strict               as HashMap-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--type Streams = Vector (Sync (Header, ByteString))--data Connection = Connection-    { _settings :: !ConnectionSettings-    , _address  :: !InetAddr-    , _tmanager :: !TimeoutManager-    , _protocol :: !Version-    , _sock     :: !Socket-    , _status   :: !(TVar Bool)-    , _streams  :: !Streams-    , _wLock    :: !(MVar ())-    , _reader   :: !(Async ())-    , _tickets  :: !Pool-    , _logger   :: !Logger-    , _eventSig :: !(Signal Event)-    , _ident    :: !ConnId-    }--makeLenses ''Connection--instance Eq Connection where-    a == b = a^.ident == b^.ident--instance Show Connection where-    show = Char8.unpack . eval . bytes--instance ToBytes Connection where-    bytes c = bytes (c^.address) +++ val "#" +++ c^.sock--resolve :: String -> PortNumber -> IO [InetAddr]-resolve host port =-    map (InetAddr . addrAddress) <$> getAddrInfo (Just hints) (Just host) (Just (show port))-  where-    hints = defaultHints { addrFlags = [AI_ADDRCONFIG], addrSocketType = Stream }--connect :: MonadIO m => ConnectionSettings -> TimeoutManager -> Version -> Logger -> InetAddr -> m Connection-connect t m v g a = liftIO $ 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 ()-        sta <- newTVarIO True-        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 . ConnId <$> newUnique-    validateSettings c `onException` close c-    return c--ping :: MonadIO m => InetAddr -> m Bool-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))--readLoop :: Version-         -> Logger-         -> ConnectionSettings-         -> Pool-         -> InetAddr-         -> Socket-         -> Streams-         -> Signal Event-         -> TVar Bool-         -> MVar ()-         -> IO ()-readLoop v g set tck i sck syn s sref wlck =-    run `catch` logException `finally` cleanup-  where-    run = forever $ do-        x <- readSocket v g i sck (set^.maxRecvBuffer)-        case fromStreamId $ streamId (fst x) of-            -1 ->-                case parse (set^.compression) x :: Raw Response of-                    RsError _ _ e -> throwM e-                    RsEvent _ _ e -> emit s e-                    r             -> throwM (UnexpectedResponse' r)-            sid -> do-                ok <- Sync.put x (syn ! sid)-                unless ok $-                    markAvailable tck sid--    cleanup = uninterruptibleMask_ $ do-        isOpen <- atomically $ swapTVar sref False-        when isOpen $ do-            Tickets.close (ConnectionClosed i) tck-            Vector.mapM_ (Sync.close (ConnectionClosed i)) syn-            void $ forkIOWithUnmask $ \unmask -> unmask $ do-                Socket.shutdown sck ShutdownReceive-                withMVar wlck (const $ Socket.close sck)--    logException :: SomeException -> IO ()-    logException e = case fromException e of-        Just ThreadKilled -> return ()-        _                 -> warn g $ msg i ~~ msg (val "read-loop: " +++ show e)--close :: Connection -> IO ()-close = cancel . view reader--request :: Connection -> (Int -> ByteString) -> IO (Header, ByteString)-request c f = send >>= receive-  where-    send = withTimeout (c^.tmanager) (c^.settings.sendTimeout) (close c) $ do-        i <- toInt <$> Tickets.get (c^.tickets)-        let req = f i-        trace (c^.logger) $ msg c-            ~~ "stream" .= i-            ~~ "type"   .= val "request"-            ~~ msg' (hexdump (L.take 160 req))-        withMVar (c^.wLock) $ const $ do-            isOpen <- readTVarIO (c^.status)-            if isOpen then-                Socket.send (c^.sock) req-            else-                throwM $ ConnectionClosed (c^.address)-        return i--    receive i = do-        let e = TimeoutRead (show c ++ ":" ++ show i)-        tid <- myThreadId-        withTimeout (c^.tmanager) (c^.settings.responseTimeout) (throwTo tid e) $ do-            x <- Sync.get (view streams c ! i) `onException` Sync.kill e (view streams c ! i)-            markAvailable (c^.tickets) i-            return x--readSocket :: Version -> Logger -> InetAddr -> Socket -> Int -> IO (Header, ByteString)-readSocket v g i s n = do-    b <- Socket.recv n i s 9-    h <- case header v b of-            Left  e -> throwM $ InternalError ("response header reading: " ++ e)-            Right h -> return h-    case headerType h of-        RqHeader -> throwM $ InternalError "unexpected request header"-        RsHeader -> do-            let len = lengthRepr (bodyLength h)-            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)---------------------------------------------------------------------------------- Operations--startup :: MonadIO m => Connection -> m ()-startup c = liftIO $ do-    let cmp = c^.settings.compression-    let req = RqStartup (Startup Cqlv300 (algorithm cmp))-    let enc = serialise (c^.protocol) cmp (req :: Raw Request)-    res <- request c enc-    case parse cmp res :: Raw Response of-        RsReady _ _ Ready       -> checkAuth c-        RsAuthenticate _ _ auth -> authenticate c auth-        RsError _ _ e           -> throwM e-        other                   -> throwM $ UnexpectedResponse' other--checkAuth :: Connection -> IO ()-checkAuth c = unless (null (c^.settings.authenticators)) $-    warn (_logger c) $ msg $ val-        "Authentication configured but none required by server."--authenticate :: (MonadIO m, MonadThrow m) => Connection -> Authenticate -> m ()-authenticate c (Authenticate (AuthMechanism -> m)) =-    case HashMap.lookup m (c^.settings.authenticators) of-        Nothing -> throwM $ AuthenticationRequired m-        Just Authenticator {-            authOnRequest   = onR-          , authOnChallenge = onC-          , authOnSuccess   = onS-        } -> liftIO $ do-            (rs, s) <- onR context-            case onC of-                Just  f -> loop f onS (rs, s)-                Nothing -> authResponse c rs >>= either-                    (throwM . UnexpectedAuthenticationChallenge m)-                    (onS s)-  where-    context = AuthContext (c^.ident) (c^.address)--    loop onC onS (rs, s) =-        authResponse c rs >>= either-            (onC s >=> loop onC onS)-            (onS s)--authResponse :: MonadIO m-             => Connection-             -> AuthResponse-             -> m (Either AuthChallenge AuthSuccess)-authResponse c resp = liftIO $ do-    let cmp = c^.settings.compression-    let req = RqAuthResp resp-    let enc = serialise (c^.protocol) cmp (req :: Raw Request)-    res <- request c enc-    case parse cmp res :: Raw Response of-        RsAuthSuccess _ _ success     -> return $ Right success-        RsAuthChallenge _ _ challenge -> return $ Left challenge-        RsError _ _ e                 -> throwM e-        other                         -> throwM $ UnexpectedResponse' other--register :: MonadIO m => Connection -> [EventType] -> EventHandler -> m ()-register c e f = liftIO $ do-    let req = RqRegister (Register e) :: Raw Request-    let enc = serialise (c^.protocol) (c^.settings.compression) req-    res <- request c enc-    case parse (c^.settings.compression) res :: Raw Response of-        RsReady _ _ Ready -> c^.eventSig |-> f-        other             -> throwM (UnexpectedResponse' other)--validateSettings :: MonadIO m => Connection -> m ()-validateSettings c = liftIO $ do-    Supported ca _ <- supportedOptions c-    let x = algorithm (c^.settings.compression)-    unless (x == None || x `elem` ca) $-        throwM $ UnsupportedCompression ca--supportedOptions :: MonadIO m => Connection -> m Supported-supportedOptions c = liftIO $ do-    let options = RqOptions Options :: Raw Request-    res <- request c (serialise (c^.protocol) noCompression options)-    case parse noCompression res :: Raw Response of-        RsSupported _ _ x -> return x-        other             -> throwM (UnexpectedResponse' other)--useKeyspace :: MonadIO m => Connection -> Keyspace -> m ()-useKeyspace c ks = liftIO $ do-    let cmp    = c^.settings.compression-        params = QueryParams One False () Nothing Nothing Nothing Nothing-        kspace = quoted (fromStrict $ unKeyspace ks)-        req    = RqQuery (Query (QueryString $ "use " <> kspace) params)-    res <- request c (serialise (c^.protocol) cmp req)-    case parse cmp res :: Raw Response of-        RsResult _ _ (SetKeyspaceResult _) -> return ()-        other                              -> throwM (UnexpectedResponse' other)--query :: forall k a b m. (Tuple a, Tuple b, Show b, MonadIO m)-      => Connection-      -> Consistency-      -> QueryString k a b-      -> a-      -> m [b]-query c cons q p = liftIO $ do-    let req = RqQuery (Query q params) :: Request k a b-    let enc = serialise (c^.protocol) (c^.settings.compression) req-    res <- request c enc-    case parse (c^.settings.compression) res :: Response k a b of-        RsResult _ _ (RowsResult _ b) -> return b-        other                         -> throwM (UnexpectedResponse' other)-  where-    params = QueryParams cons False p Nothing Nothing Nothing Nothing---- logging helpers:--msg' :: ByteString -> Msg -> Msg-msg' x = msg $ case nativeNewline of-    LF   -> val "\n"   +++ x-    CRLF -> val "\r\n" +++ x

@@ -1,141 +0,0 @@--- 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 ExistentialQuantification  #-}-{-# LANGUAGE GeneralizedNewtypeDeriving #-}-{-# LANGUAGE OverloadedStrings          #-}-{-# LANGUAGE ScopedTypeVariables        #-}-{-# LANGUAGE TemplateHaskell            #-}--module Database.CQL.IO.Connection.Settings-    ( ConnectionSettings-    , defSettings-    , connectTimeout-    , sendTimeout-    , responseTimeout-    , maxStreams-    , compression-    , defKeyspace-    , maxRecvBuffer-    , tlsContext-    , authenticators--      -- * Authentication-    , AuthMechanism (..)-    , Authenticator (..)-    , AuthContext   (..)-    , authConnId-    , authHost-    , passwordAuthenticator-    , AuthUser (..)-    , AuthPass (..)-    ) where--import Control.Lens (makeLenses)-import Control.Monad-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--import qualified Data.ByteString.Lazy.Char8 as Char8-import qualified Data.HashMap.Strict        as HashMap-import qualified Data.Text.Lazy             as Lazy-import qualified Data.Text.Lazy.Encoding    as Lazy--data ConnectionSettings = ConnectionSettings-    { _connectTimeout  :: !Milliseconds-    , _sendTimeout     :: !Milliseconds-    , _responseTimeout :: !Milliseconds-    , _maxStreams      :: !Int-    , _compression     :: !Compression-    , _defKeyspace     :: !(Maybe Keyspace)-    , _maxRecvBuffer   :: !Int-    , _tlsContext      :: !(Maybe SSLContext)-    , _authenticators  :: !(HashMap AuthMechanism Authenticator)-    }---- | Context information given to 'Authenticator's when--- the server requests authentication on a connection.--- See 'authOnRequest'.-data AuthContext = AuthContext-    { _authConnId :: !ConnId-    , _authHost   :: !InetAddr-    }---- | A client authentication handler.------ The fields of an 'Authenticator' must implement the client-side--- of an (SASL) authentication mechanism as follows:------    * When a Cassandra server requests authentication on a new connection,---      'authOnRequest' is called with the 'AuthContext' of the---      connection.------    * If additional challenges are posed by the server,---      'authOnChallenge' is called, if available, otherwise an---      'AuthenticationError' is thrown, i.e. every challenge must be---      answered.------    * Upon successful authentication 'authOnSuccess' is called.------ The existential type @s@ is chosen by an implementation and can--- be used to thread arbitrary state through the sequence of callback--- invocations during an authentication exchange.------ See also:--- <https://tools.ietf.org/html/rfc4422 RFC4422>--- <https://docs.datastax.com/en/cassandra/latest/cassandra/configuration/secureInternalAuthenticationTOC.html Authentication>-data Authenticator = forall s. Authenticator-    { authMechanism :: !AuthMechanism-        -- ^ The (unique) name of the (SASL) mechanism that the callbacks-        -- implement.-    , authOnRequest :: AuthContext -> IO (AuthResponse, s)-        -- ^ Callback for initiating an authentication exchange.-    , authOnChallenge :: Maybe (s -> AuthChallenge -> IO (AuthResponse, s))-        -- ^ Optional callback for additional challenges posed by the server.-        -- If the authentication mechanism does not require additional-        -- challenges, it should be set to 'Nothing'. Otherwise every-        -- challenge must be answered with a response.-    , authOnSuccess :: s -> AuthSuccess -> IO ()-        -- ^ Callback for successful completion of an authentication exchange.-    }--makeLenses ''AuthContext-makeLenses ''ConnectionSettings--newtype AuthUser = AuthUser Lazy.Text-newtype AuthPass = AuthPass Lazy.Text---- | A password authentication handler for use with Cassandra's--- @PasswordAuthenticator@.------ See: <https://docs.datastax.com/en/cassandra/latest/cassandra/configuration/secureConfigNativeAuth.html Configuring Authentication>-passwordAuthenticator :: AuthUser -> AuthPass -> Authenticator-passwordAuthenticator (AuthUser u) (AuthPass p) = Authenticator-    { authMechanism   = "org.apache.cassandra.auth.PasswordAuthenticator"-    , authOnChallenge = Nothing-    , authOnSuccess   = \() _ -> return ()-    , authOnRequest   = \_ctx ->-        let user = Lazy.encodeUtf8 u-            pass = Lazy.encodeUtf8 p-            resp = AuthResponse (Char8.concat ["\0", user, "\0", pass])-        in return (resp, ())-    }--defSettings :: ConnectionSettings-defSettings =-    ConnectionSettings 5000          -- connect timeout-                       3000          -- send timeout-                       10000         -- response timeout-                       128           -- max streams per connection-                       noCompression -- compression-                       Nothing       -- keyspace-                       16384         -- receive buffer size-                       Nothing       -- no tls by default-                       HashMap.empty -- no authentication-

@@ -1,100 +0,0 @@--- 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.Cluster.Host-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) && !MIN_VERSION_network(3,0,0)-    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)

@@ -1,63 +0,0 @@--- 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 OverloadedStrings #-}--module Database.CQL.IO.Hexdump (hexdump, hexdumpBuilder) where--import Data.ByteString.Builder-import Data.ByteString.Lazy (ByteString)-import Data.Int-import Data.Monoid-import Data.Word-import System.IO (nativeNewline, Newline (..))--import qualified Data.ByteString.Lazy as L-import qualified Data.List            as List--width, groups :: Int64-width  = 16-groups = 4--hexdump :: ByteString -> ByteString-hexdump = toLazyByteString . hexdumpBuilder--hexdumpBuilder :: ByteString -> Builder-hexdumpBuilder = mconcat-    . List.intersperse newline-    . map toLine-    . zipWith (,) [1 ..]-    . chunks width--chunks :: Int64 -> ByteString -> [ByteString]-chunks n b = List.unfoldr step b-  where-    step "" = Nothing-    step c  = Just $! L.splitAt n c--toLine :: (Word16, ByteString) -> Builder-toLine (n, b) = let k = L.length b in-       word16HexFixed n-    <> ":  "-    <> mconcat (List.intersperse space (map toGroup (chunks groups b)))-    <> spaces ((width - k) * (groups - 1) + pad k + 2)-    <> lazyByteString (toAscii b)--toGroup :: ByteString -> Builder-toGroup = L.foldr (\x y -> word8HexFixed x <> space <> y) mempty--toAscii :: ByteString -> ByteString-toAscii = L.map (\w -> if w > 0x1F && w < 0x7F then w else 0x2E)--space, newline :: Builder-space   = byteString " "-newline = case nativeNewline of-    LF   -> byteString "\n"-    CRLF -> byteString "\r\n"--spaces :: Int64 -> Builder-spaces n = lazyByteString $ L.replicate n 0x20--pad :: Int64 -> Int64-pad n = (width - n) `div` groups

@@ -1,69 +0,0 @@--- 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/.--module Database.CQL.IO.Jobs-    ( Jobs-    , new-    , add-    , destroy-    , showJobs-    ) where--import Control.Applicative-import Control.Concurrent.Async-import Control.Concurrent.STM-import Control.Monad-import Control.Monad.Catch-import Control.Monad.IO.Class-import Database.CQL.IO.Types-import Data.Map.Strict (Map)-import Data.Unique-import Prelude--import qualified Data.Map.Strict as Map--data Job-    = Reserved-    | Running !Unique !(Async ())--newtype Jobs k = Jobs (TVar (Map k Job))--new :: MonadIO m => m (Jobs k)-new = liftIO $ Jobs <$> newTVarIO Map.empty--add :: (MonadIO m, Ord k) => Jobs k -> k -> Bool -> IO () -> m ()-add (Jobs d) k replace j = liftIO $ do-    (ok, prev) <- atomically $ do-        m <- readTVar d-        case Map.lookup k m of-            Nothing -> do-                modifyTVar' d (Map.insert k Reserved)-                return (True, Nothing)-            Just (Running _ a) | replace -> do-                modifyTVar' d (Map.insert k Reserved)-                return (True, Just a)-            _ -> return (False, Nothing)-    when ok $ do-        maybe (return ()) (ignore . cancel) prev-        u <- newUnique-        a <- async $ j `finally` remove u-        atomically $-            modifyTVar' d (Map.insert k (Running u a))-  where-    remove u = atomically $-        modifyTVar' d $ flip Map.update k $ \a ->-            case a of-                Running u' _ | u == u' -> Nothing-                _                      -> Just a--destroy :: MonadIO m => Jobs k -> m ()-destroy (Jobs d) = liftIO $ do-    items <- Map.elems <$> atomically (swapTVar d Map.empty)-    mapM_ f items-  where-    f (Running _ a) = ignore (cancel a)-    f _             = return ()--showJobs :: MonadIO m => Jobs k -> m [k]-showJobs (Jobs d) = liftIO $ Map.keys <$> readTVarIO d

@@ -1,217 +0,0 @@--- 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 MultiWayIf          #-}-{-# LANGUAGE OverloadedStrings   #-}-{-# LANGUAGE ScopedTypeVariables #-}-{-# LANGUAGE TemplateHaskell     #-}--module Database.CQL.IO.Pool-    ( Pool-    , create-    , destroy-    , purge-    , with--    , PoolSettings-    , defSettings-    , idleTimeout-    , maxConnections-    , maxTimeouts-    , poolStripes-    ) where--import Control.Applicative-import Control.AutoUpdate-import Control.Concurrent-import Control.Concurrent.Async-import Control.Concurrent.STM-import Control.Exception-import Control.Lens ((^.), makeLenses, view)-import Control.Monad.IO.Class-import Control.Monad hiding (forM_, mapM_)-import Data.Foldable (forM_, mapM_, find)-import Data.Function (on)-import Data.Hashable-import Data.IORef-import Prelude hiding (mapM_)-import Data.Sequence (Seq, ViewL (..), (|>), (><))-import Data.Time.Clock (UTCTime, NominalDiffTime, getCurrentTime, diffUTCTime)-import Data.Vector (Vector, (!))-import Database.CQL.IO.Connection (Connection)-import Database.CQL.IO.Types (Timeout, ignore)-import System.Logger hiding (create, defSettings, settings)--import qualified Data.Sequence as Seq-import qualified Data.Vector   as Vec---------------------------------------------------------------------------------- API--data PoolSettings = PoolSettings-    { _idleTimeout    :: !NominalDiffTime-    , _maxConnections :: !Int-    , _maxTimeouts    :: !Int-    , _poolStripes    :: !Int-    }--data Pool = Pool-    { _createFn    :: !(IO Connection)-    , _destroyFn   :: !(Connection -> IO ())-    , _logger      :: !Logger-    , _settings    :: !PoolSettings-    , _maxRefs     :: !Int-    , _currentTime :: !(IO UTCTime)-    , _stripes     :: !(Vector Stripe)-    , _finaliser   :: !(IORef ())-    }--data Resource = Resource-    { tstamp   :: !UTCTime-    , refcnt   :: !Int-    , timeouts :: !Int-    , value    :: !Connection-    } deriving Show--data Box-    = New  !(IO Resource)-    | Used !Resource-    | Empty--data Stripe = Stripe-    { conns :: !(TVar (Seq Resource))-    , inUse :: !(TVar Int)-    }--makeLenses ''PoolSettings-makeLenses ''Pool--defSettings :: PoolSettings-defSettings = PoolSettings-    60 -- idle timeout-    2  -- max connections per stripe-    16 -- max timeouts per connection-    4  -- max stripes--create :: IO Connection -> (Connection -> IO ()) -> Logger -> PoolSettings -> Int -> IO Pool-create mk del g s k = do-    p <- Pool mk del g s k-            <$> mkAutoUpdate defaultUpdateSettings { updateAction = getCurrentTime }-            <*> Vec.replicateM (s^.poolStripes) (Stripe <$> newTVarIO Seq.empty <*> newTVarIO 0)-            <*> newIORef ()-    r <- async $ reaper p-    void $ mkWeakIORef (p^.finaliser) (cancel r >> destroy p)-    return p--destroy :: Pool -> IO ()-destroy = purge--with :: MonadIO m => Pool -> (Connection -> IO a) -> m (Maybe a)-with p f = liftIO $ do-    s <- stripe p-    mask $ \restore -> do-        r <- take1 p s-        case r of-            Just  v -> do-                x <- restore (f (value v)) `catches` handlers p s v-                put p s v id-                return (Just x)-            Nothing -> return Nothing--purge :: Pool -> IO ()-purge p = Vec.forM_ (p^.stripes) $ \s ->-    atomically (swapTVar (conns s) Seq.empty) >>= mapM_ (ignore . view destroyFn p . value)---------------------------------------------------------------------------------- Internal--handlers :: Pool -> Stripe -> Resource -> [Handler a]-handlers p s r =-    [ Handler $ \(x :: Timeout)       -> onTimeout      >> throwIO x-    , Handler $ \(x :: SomeException) -> destroyR p s r >> throwIO x-    ]-  where-    onTimeout =-        if timeouts r > p^.settings.maxTimeouts-            then do-                info (p^.logger) $ msg (show (value r) +++ val " has too many timeouts.")-                destroyR p s r-            else put p s r incrTimeouts--take1 :: Pool -> Stripe -> IO (Maybe Resource)-take1 p s = do-    r <- atomically $ do-        c <- readTVar (conns s)-        u <- readTVar (inUse s)-        let n = Seq.length c-        check (u == n)-        let r :< rr = Seq.viewl $ Seq.unstableSortBy (compare `on` refcnt) c-        if | u < p^.settings.maxConnections -> do-                writeTVar (inUse s) $! u + 1-                mkNew p-           | n > 0 && refcnt r < p^.maxRefs -> use s r rr-           | otherwise                      -> return Empty-    case r of-        New io -> do-            x <- io `onException` atomically (modifyTVar' (inUse s) (subtract 1))-            atomically (modifyTVar' (conns s) (|> x))-            return (Just x)-        Used x -> return (Just x)-        Empty  -> return Nothing--use :: Stripe -> Resource -> Seq Resource -> STM Box-use s r rr = do-    writeTVar (conns s) $! rr |> r { refcnt = refcnt r + 1 }-    return (Used r)-{-# INLINE use #-}--mkNew :: Pool -> STM Box-mkNew p = return (New $ Resource <$> p^.currentTime <*> pure 1 <*> pure 0 <*> p^.createFn)-{-# INLINE mkNew #-}--put :: Pool -> Stripe -> Resource -> (Resource -> Resource) -> IO ()-put p s r f = do-    now <- p^.currentTime-    let updated x = f x { tstamp = now, refcnt = refcnt x - 1 }-    atomically $ do-        rs <- readTVar (conns s)-        let (xs, rr) = Seq.breakl ((value r ==) . value) rs-        case Seq.viewl rr of-            EmptyL  -> writeTVar (conns s) $! xs         |> updated r-            y :< ys -> writeTVar (conns s) $! (xs >< ys) |> updated y--destroyR :: Pool -> Stripe -> Resource -> IO ()-destroyR p s r = do-    atomically $ do-        rs <- readTVar (conns s)-        case find ((value r ==) . value) rs of-            Nothing -> return ()-            Just  _ -> do-                modifyTVar' (inUse s) (subtract 1)-                writeTVar (conns s) $! Seq.filter ((value r /=) . value) rs-    ignore $ p^.destroyFn $ value r--reaper :: Pool -> IO ()-reaper p = forever $ do-    threadDelay 1000000-    now <- p^.currentTime-    let isStale r = refcnt r == 0 && now `diffUTCTime` tstamp r > p^.settings.idleTimeout-    Vec.forM_ (p^.stripes) $ \s -> do-        x <- atomically $ do-                (stale, okay) <- Seq.partition isStale <$> readTVar (conns s)-                unless (Seq.null stale) $ do-                    writeTVar   (conns s) okay-                    modifyTVar' (inUse s) (subtract (Seq.length stale))-                return stale-        forM_ x $ \v -> ignore $ do-            trace (p^.logger) $ "reap" .= show (value v)-            p^.destroyFn $ (value v)--stripe :: Pool -> IO Stripe-stripe p = ((p^.stripes) !) <$> ((`mod` (p^.settings.poolStripes)) . hash) <$> myThreadId-{-# INLINE stripe #-}--incrTimeouts :: Resource -> Resource-incrTimeouts r = r { timeouts = timeouts r + 1 }-{-# INLINE incrTimeouts #-}

@@ -1,143 +0,0 @@--- 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/.--module Database.CQL.IO.PrepQuery-    ( PrepQuery-    , prepared-    , queryString--    , PreparedQueries-    , new-    , lookupQueryId-    , lookupQueryString-    , insert-    , delete-    , queryStrings-    ) where--import Control.Applicative-import Control.Concurrent.STM-import Control.Monad-import Crypto.Hash.SHA1-import Data.ByteString (ByteString)-import Data.Text.Lazy (Text)-import Data.Text.Lazy.Encoding (encodeUtf8)-import Data.Foldable (for_)-import Data.Map.Strict (Map)-import Data.String-import Database.CQL.Protocol hiding (Map)-import Database.CQL.IO.Types (HashCollision (..))-import Prelude--import qualified Data.Map.Strict as M---------------------------------------------------------------------------------- Prepared Query---- | Representation of a prepared 'QueryString'. A prepared query is--- executed in two stages:------   1. The query string is sent to a server without parameters for---      preparation. The server responds with a 'QueryId'.---   2. The prepared query is executed by sending the 'QueryId'---      and parameters to the server.------ Thereby step 1 is only performed when the query has not yet been prepared--- with the host (coordinator) used for query execution. Thus, prepared--- queries enhance performance by avoiding the repeated sending and parsing--- of query strings.------ Query preparation is handled transparently by the client.--- See 'Database.CQL.IO.setPrepareStrategy'.------ __Note__------ Prepared statements are fully supported but rely on some--- assumptions beyond the scope of the CQL binary protocol--- specification (spec):------ (1) The spec scopes the 'QueryId' to the node the query has---     been prepared with. The spec does not state anything---     about the format of the 'QueryId'. However the official---     Java driver assumes that any given 'QueryString' yields---     the same 'QueryId' on every node. This client make the---     same assumption.--- (2) In case a node does not know a given 'QueryId' an 'Unprepared'---     error is returned. We assume that it is always safe to---     transparently re-prepare the corresponding 'QueryString' and---     to re-execute the original request against the same node.------ Besides these assumptions there is also a potential tradeoff in--- regards to /eager/ vs. /lazy/ query preparation.--- We understand /eager/ to mean preparation against all current nodes of--- a cluster and /lazy/ to mean preparation against a single node on demand,--- i.e. upon receiving an 'Unprepared' error response. Which strategy to--- choose depends on the scope of query reuse and the size of the cluster.--- The global default can be changed through the 'Settings' module as well--- as locally using 'withPrepareStrategy'.-data PrepQuery k a b = PrepQuery-    { pqStr :: !(QueryString k a b)-    , pqId  :: !PrepQueryId-    }--instance IsString (PrepQuery k a b) where-    fromString = prepared . fromString--newtype PrepQueryId = PrepQueryId ByteString deriving (Eq, Ord)--prepared :: QueryString k a b -> PrepQuery k a b-prepared q = PrepQuery q $ PrepQueryId (hashlazy . encodeUtf8 . unQueryString $ q)--queryString :: PrepQuery k a b -> QueryString k a b-queryString = pqStr---------------------------------------------------------------------------------- Map of prepared queries to their query ID and query string--newtype QST = QST { unQST :: Text }-newtype QID = QID { unQID :: ByteString } deriving (Eq, Ord)--data PreparedQueries = PreparedQueries-    { queryMap :: !(TVar (Map PrepQueryId (QID, QST)))-    , qid2Str  :: !(TVar (Map QID QST))-    }--new :: IO PreparedQueries-new = PreparedQueries <$> newTVarIO M.empty <*> newTVarIO M.empty--lookupQueryId :: PrepQuery k a b -> PreparedQueries -> STM (Maybe (QueryId k a b))-lookupQueryId q m = do-    qm <- readTVar (queryMap m)-    return $ QueryId . unQID . fst <$> M.lookup (pqId q) qm--lookupQueryString :: QueryId k a b -> PreparedQueries -> STM (Maybe (QueryString k a b))-lookupQueryString q m = do-    qm <- readTVar (qid2Str m)-    return $ QueryString . unQST <$> M.lookup (QID $ unQueryId q) qm--insert :: PrepQuery k a b -> QueryId k a b -> PreparedQueries -> STM ()-insert q i m = do-    qq <- M.lookup (pqId q) <$> readTVar (queryMap m)-    for_ qq (verify . snd)-    modifyTVar' (queryMap m) $-        M.insert (pqId q) (QID $ unQueryId i, QST $ unQueryString (pqStr q))-    modifyTVar' (qid2Str  m) $-        M.insert (QID $ unQueryId i) (QST $ unQueryString (pqStr q))-  where-    verify qs =-        unless (unQST qs == unQueryString (pqStr q)) $ do-            let a = unQST qs-            let b = unQueryString (pqStr q)-            throwSTM (HashCollision a b)--delete :: PrepQuery k a b -> PreparedQueries -> STM ()-delete q m = do-    qid <- M.lookup (pqId q) <$> readTVar (queryMap m)-    modifyTVar' (queryMap m) $ M.delete (pqId q)-    case qid of-        Nothing -> return ()-        Just  i -> modifyTVar' (qid2Str m) $ M.delete (fst i)--queryStrings :: PreparedQueries -> STM [Text]-queryStrings m = map (unQST . snd) . M.elems <$> readTVar (queryMap m)

@@ -1,39 +0,0 @@--- 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 OverloadedStrings #-}--module Database.CQL.IO.Protocol where--import Control.Exception (throw)-import Data.ByteString.Lazy (ByteString)-import Data.Maybe (fromMaybe)-import Data.Monoid ((<>))-import Database.CQL.Protocol-import Database.CQL.IO.Types--import qualified Data.Text.Lazy as LT--parse :: (Tuple a, Tuple b) => Compression -> (Header, ByteString) -> Response k a b-parse x (h, a) =-    case unpack x h a of-        Left  e -> throw $ InternalError ("response body reading: " ++ e)-        Right r -> r--serialise :: Tuple a => Version -> Compression -> Request k a b -> Int -> ByteString-serialise v f r i =-    let c = case getOpCode r of-                OcStartup -> noCompression-                OcOptions -> noCompression-                _         -> f-        s = mkStreamId i-    in either (throw $ InternalError "request creation") id (pack v c (isTracing r) s r)-  where-    isTracing :: Request k a b -> Bool-    isTracing (RqQuery (Query _ p))     = fromMaybe False $ enableTracing p-    isTracing (RqExecute (Execute _ p)) = fromMaybe False $ enableTracing p-    isTracing _                         = False--quoted :: LT.Text -> LT.Text-quoted s = "\"" <> LT.replace "\"" "\"\"" s <> "\""

@@ -1,269 +0,0 @@--- 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 OverloadedStrings #-}-{-# LANGUAGE RankNTypes        #-}-{-# LANGUAGE TemplateHaskell   #-}--module Database.CQL.IO.Settings where--import Control.Lens hiding ((<|))-import Control.Retry hiding (retryPolicy)-import Data.List.NonEmpty (NonEmpty (..), (<|))-import Data.Monoid-import Data.Time-import Data.Word-import Database.CQL.Protocol-import Database.CQL.IO.Cluster.Policies (Policy, random)-import Database.CQL.IO.Connection.Settings as C-import Database.CQL.IO.Pool as P-import Database.CQL.IO.Types (Milliseconds (..))-import Network.Socket (PortNumber (..))-import OpenSSL.Session (SSLContext)-import Prelude--import qualified Data.HashMap.Strict as HashMap---- | Strategy for the execution of 'PrepQuery's.-data PrepareStrategy-    = EagerPrepare -- ^ cluster-wide preparation-    | LazyPrepare  -- ^ on-demand per node preparation-    deriving (Eq, Ord, Show)--data RetrySettings = RetrySettings-    { _retryPolicy        :: !(forall m. Monad m => RetryPolicyM m)-    , _reducedConsistency :: !(Maybe Consistency)-    , _sendTimeoutChange  :: !Milliseconds-    , _recvTimeoutChange  :: !Milliseconds-    }--data Settings = Settings-    { _poolSettings  :: !PoolSettings-    , _connSettings  :: !ConnectionSettings-    , _retrySettings :: !RetrySettings-    , _protoVersion  :: !Version-    , _portnumber    :: !PortNumber-    , _contacts      :: !(NonEmpty String)-    , _policyMaker   :: !(IO Policy)-    , _prepStrategy  :: !PrepareStrategy-    }--makeLenses ''RetrySettings-makeLenses ''Settings---- | Default settings:------ * contact point is \"localhost\" port 9042------ * load-balancing policy is 'random'------ * binary protocol version is 3------ * connection idle timeout is 60s------ * the connection pool uses 4 stripes to mitigate thread contention------ * connections use a connect timeout of 5s, a send timeout of 3s and--- a receive timeout of 10s------ * 128 streams per connection are used------ * 16k receive buffer size------ * no compression is applied to frame bodies------ * no default keyspace is used.------ * no retries are done------ * lazy prepare strategy-defSettings :: Settings-defSettings = Settings-    P.defSettings-    C.defSettings-    noRetry-    V3-    9042-    ("localhost" :| [])-    random-    LazyPrepare---------------------------------------------------------------------------------- Settings---- | Set the binary protocol version to use.-setProtocolVersion :: Version -> Settings -> Settings-setProtocolVersion v = set protoVersion v---- | Set the initial contact points (hosts) from which node discovery will--- start.-setContacts :: String -> [String] -> Settings -> Settings-setContacts v vv = set contacts (v :| vv)---- | Add an additional host to the contact list.-addContact :: String -> Settings -> Settings-addContact v = over contacts (v <|)---- | Set the portnumber to use to connect on /every/ node of the cluster.-setPortNumber :: PortNumber -> Settings -> Settings-setPortNumber v = set portnumber v---- | Set the load-balancing policy.-setPolicy :: IO Policy -> Settings -> Settings-setPolicy v = set policyMaker v---- | Set strategy to use for preparing statements.-setPrepareStrategy :: PrepareStrategy -> Settings -> Settings-setPrepareStrategy v = set prepStrategy v---------------------------------------------------------------------------------- Pool Settings---- | Set the connection idle timeout. Connections in a pool will be closed--- if not in use for longer than this timeout.-setIdleTimeout :: NominalDiffTime -> Settings -> Settings-setIdleTimeout v = set (poolSettings.idleTimeout) v---- | Maximum connections per pool /stripe/.-setMaxConnections :: Int -> Settings -> Settings-setMaxConnections v = set (poolSettings.maxConnections) v---- | Set the number of pool stripes to use. A good setting is equal to the--- number of CPU cores this codes is running on.-setPoolStripes :: Int -> Settings -> Settings-setPoolStripes v s-    | 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--- connection that was used to make the request as it is uncertain if--- a response will arrive later. Thus the bandwith of a connection will be--- decreased. This settings defines a threshold after which we close the--- connection to get a new one with all streams available.-setMaxTimeouts :: Int -> Settings -> Settings-setMaxTimeouts v = set (poolSettings.maxTimeouts) v---------------------------------------------------------------------------------- Connection Settings---- | Set the compression to use for frame body compression.-setCompression :: Compression -> Settings -> Settings-setCompression v = set (connSettings.compression) v---- | Set the maximum number of streams per connection. In version 2 of the--- binary protocol at most 128 streams can be used. Version 3 supports up--- to 32768 streams.-setMaxStreams :: Int -> Settings -> Settings-setMaxStreams v s-    | v < 1 || v > 32768 = error "cql-io settings: max. streams must be within [1, 32768]"-    | otherwise          = set (connSettings.maxStreams) v s---- | Set the connect timeout of a connection.-setConnectTimeout :: NominalDiffTime -> Settings -> Settings-setConnectTimeout v = set (connSettings.connectTimeout) (Ms $ round (1000 * v))---- | Set the send timeout of a connection. Request exceeding the send will--- cause the connection to be closed and fail with 'ConnectionClosed'--- exception.-setSendTimeout :: NominalDiffTime -> Settings -> Settings-setSendTimeout v = set (connSettings.sendTimeout) (Ms $ round (1000 * v))---- | Set the receive timeout of a connection. Requests exceeding the--- receive timeout will fail with a 'Timeout' exception.-setResponseTimeout :: NominalDiffTime -> Settings -> Settings-setResponseTimeout v = set (connSettings.responseTimeout) (Ms $ round (1000 * v))---- | Set the default keyspace to use. Every new connection will be--- 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---- | Set maximum receive buffer size.------ The actual buffer size used will be the minimum of the CQL response size--- and the value set here.-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)---- | Set the supported authentication mechanisms.------ When a Cassandra server requests authentication on a connection,--- it specifies the requested 'AuthMechanism'. The client 'Authenticator'--- is chosen based that name. If no authenticator with a matching--- name is configured, an 'AuthenticationError' is thrown.-setAuthentication :: [C.Authenticator] -> Settings -> Settings-setAuthentication = set (connSettings.authenticators)-                  . HashMap.fromList-                  . map (\a -> (authMechanism a, a))---------------------------------------------------------------------------------- Retry Settings---- | Never retry.-noRetry :: RetrySettings-noRetry = RetrySettings (RetryPolicyM $ const (return 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 s =-    s { _retryPolicy = limitRetries (fromIntegral v) <> _retryPolicy s }---- | 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 f v w s =-    let a = round (1000000 * w)-        b = round (1000000 * v)-    in-        s { _retryPolicy = capDelay a (f b) <> _retryPolicy s }

@@ -1,36 +0,0 @@--- 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/.--module Database.CQL.IO.Signal-    ( Signal-    , signal-    , connect-    , emit-    , (|->)-    , ($$)-    ) where--import Control.Applicative-import Control.Concurrent (forkIO)-import Data.IORef-import Prelude--newtype Signal a = Sig (IORef [a -> IO ()])--signal :: IO (Signal a)-signal = Sig <$> newIORef []--connect :: Signal a -> (a -> IO ()) -> IO ()-connect (Sig s) f = modifyIORef s (f:)--infixl 2 |->-(|->) :: Signal a -> (a -> IO ()) -> IO ()-(|->) = connect--emit :: Signal a -> a -> IO ()-emit (Sig s) a = readIORef s >>= mapM_ (forkIO . ($ a))--infixr 1 $$-($$) :: Signal a -> a -> IO ()-($$) = emit

@@ -1,56 +0,0 @@--- 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/.--module Database.CQL.IO.Sync-    ( Sync-    , create-    , get-    , put-    , kill-    , close-    ) where--import Control.Applicative-import Control.Concurrent.STM-import Control.Exception (SomeException, Exception, toException)-import Prelude--data State a-    = Empty-    | Value  !a-    | Killed !SomeException-    | Closed !SomeException--newtype Sync a = Sync (TVar (State a))--create :: IO (Sync a)-create = Sync <$> newTVarIO Empty--get :: Sync a -> IO a-get (Sync s) = atomically $ do-    v <- readTVar s-    case v of-        Empty    -> retry-        Value  a -> writeTVar s Empty >> return a-        Closed x -> throwSTM x-        Killed x -> throwSTM x--put :: a -> Sync a -> IO Bool-put a (Sync s) = atomically $ do-    v <- readTVar s-    case v of-        Empty    -> writeTVar s (Value a) >> return True-        Closed _ -> return True-        _        -> writeTVar s Empty     >> return False--kill :: Exception e => e -> Sync a -> IO ()-kill x (Sync s) = atomically $ do-    v <- readTVar s-    case v of-        Closed _ -> return ()-        _        -> writeTVar s (Killed $ toException x)--close :: Exception e => e -> Sync a -> IO ()-close x (Sync s) = atomically $ writeTVar s (Closed $ toException x)-

@@ -1,47 +0,0 @@--- 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/.--module Database.CQL.IO.Tickets-    ( Ticket-    , toInt-    , Pool-    , pool-    , close-    , get-    , markAvailable-    ) where--import Control.Applicative-import Control.Concurrent.STM-import Control.Exception (SomeException, Exception, toException)-import Data.Set (Set)-import Prelude--import qualified Data.Set as Set--newtype Ticket = Ticket { toInt :: Int } deriving (Eq, Ord, Show)--newtype Pool = Pool (TVar (Either SomeException (Set Ticket)))--pool :: Int -> IO Pool-pool n = Pool <$> newTVarIO (Right . Set.fromList $ map Ticket [0 .. n-1])--close :: Exception e => e -> Pool -> IO ()-close x (Pool p) = atomically $ writeTVar p (Left $ toException x)--get :: Pool -> IO Ticket-get (Pool p) = atomically $ readTVar p >>= popHead-  where-    popHead (Left x) = throwSTM x-    popHead (Right x)-        | Set.null x = retry-        | otherwise  = do-            let (t, tt) = Set.deleteFindMin x-            writeTVar p (Right tt)-            return t--markAvailable :: Pool -> Int -> IO ()-markAvailable (Pool p) t =-    atomically $ modifyTVar' p (fmap (Set.insert (Ticket t)))-

@@ -1,76 +0,0 @@--- 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/.--module Database.CQL.IO.Timeouts-    ( TimeoutManager-    , create-    , destroy-    , Action-    , Milliseconds (..)-    , add-    , cancel-    , withTimeout-    ) where--import Control.Applicative-import Control.Concurrent.STM-import Control.Exception (mask_, bracket)-import Control.Reaper-import Control.Monad-import Database.CQL.IO.Types (Milliseconds (..), ignore)-import Prelude--data TimeoutManager = TimeoutManager-    { roundtrip :: !Int-    , reaper    :: !(Reaper [Action] Action)-    }--data Action = Action-    { action :: !(IO ())-    , state  :: !(TVar State)-    }--data State = Running !Int | Canceled--create :: Milliseconds -> IO TimeoutManager-create (Ms n) = TimeoutManager n <$> mkReaper defaultReaperSettings-    { reaperAction = mkListAction prune-    , reaperDelay  = n * 1000-    }-  where-    prune a = do-        s <- atomically $ do-            x <- readTVar (state a)-            writeTVar (state a) (newState x)-            return x-        case s of-            Running 0 -> do-                ignore (action a)-                return Nothing-            Canceled -> return Nothing-            _        -> return $ Just a--    newState (Running k) = Running (k - 1)-    newState s           = s--destroy :: TimeoutManager -> Bool -> IO ()-destroy tm exec = mask_ $ do-    a <- reaperStop (reaper tm)-    when exec $ mapM_ f a-  where-    f e = readTVarIO (state e) >>= \s -> case s of-        Running _ -> ignore (action e)-        Canceled  -> return ()--add :: TimeoutManager -> Milliseconds -> IO () -> IO Action-add tm (Ms n) a = do-    r <- Action a <$> newTVarIO (Running $ n `div` roundtrip tm)-    reaperAdd (reaper tm) r-    return r--cancel :: Action -> IO ()-cancel a = atomically $ writeTVar (state a) Canceled--withTimeout :: TimeoutManager -> Milliseconds -> IO () -> IO a -> IO a-withTimeout tm m x a = bracket (add tm m x) cancel $ const a

@@ -1,207 +0,0 @@--- 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 DeriveDataTypeable         #-}-{-# LANGUAGE GADTs                      #-}-{-# LANGUAGE GeneralizedNewtypeDeriving #-}-{-# LANGUAGE OverloadedStrings          #-}-{-# LANGUAGE StandaloneDeriving         #-}--module Database.CQL.IO.Types where--import Control.Monad.Catch-import Data.Hashable-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 qualified Data.Text.Lazy as Lazy--type EventHandler = Event -> IO ()--newtype Milliseconds = Ms { ms :: Int } deriving (Eq, Show, Num)--type Raw a = a () () ()---------------------------------------------------------------------------------- ConnId--newtype ConnId = ConnId Unique deriving (Eq, Ord)--instance Hashable ConnId where-    hashWithSalt _ (ConnId u) = hashUnique u---------------------------------------------------------------------------------- InvalidSettings--data InvalidSettings-    = UnsupportedCompression [CompressionAlgorithm]-    | InvalidCacheSize-    deriving Typeable--instance Exception InvalidSettings--instance Show InvalidSettings where-    show (UnsupportedCompression cc) = "cql-io: unsupported compression: " ++ show cc-    show InvalidCacheSize            = "cql-io: invalid cache size"---------------------------------------------------------------------------------- InternalError--newtype InternalError = InternalError String-    deriving Typeable--instance Exception InternalError--instance Show InternalError where-    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-    = NoHostAvailable-    | HostsBusy-    deriving Typeable--instance Exception HostError--instance Show HostError where-    show NoHostAvailable = "cql-io: no host available"-    show HostsBusy       = "cql-io: hosts busy"---------------------------------------------------------------------------------- ConnectionError--data ConnectionError-    = ConnectionClosed !InetAddr-    | ConnectTimeout   !InetAddr-    deriving Typeable--instance Exception ConnectionError--instance Show ConnectionError where-    show (ConnectionClosed i) = "cql-io: connection closed: " ++ show i-    show (ConnectTimeout   i) = "cql-io: connect timeout: " ++ show i---------------------------------------------------------------------------------- Timeout--newtype Timeout = TimeoutRead String-    deriving Typeable--instance Exception Timeout--instance Show Timeout where-    show (TimeoutRead e) = "cql-io: read timeout: " ++ e---------------------------------------------------------------------------------- UnexpectedResponse---- | Placeholder for parts of a 'Response' that are not 'Show'able.-data NoShow = NoShow deriving Show--data UnexpectedResponse where-    UnexpectedResponse     :: !(Response k a b) -> UnexpectedResponse-    UnexpectedResponse'    :: Show b => !(Response k a b) -> UnexpectedResponse--deriving instance Typeable UnexpectedResponse-instance Exception UnexpectedResponse--instance Show UnexpectedResponse where-    show x = showString "cql-io: unexpected response: "-           . case x of-                UnexpectedResponse  r -> shows (f r)-                UnexpectedResponse' r -> shows r-           $ ""-      where-        f :: Response k a b -> Response k a NoShow-        f (RsError         a b c) = RsError a b c-        f (RsReady         a b c) = RsReady a b c-        f (RsAuthenticate  a b c) = RsAuthenticate a b c-        f (RsAuthChallenge a b c) = RsAuthChallenge a b c-        f (RsAuthSuccess   a b c) = RsAuthSuccess a b c-        f (RsSupported     a b c) = RsSupported a b c-        f (RsResult        a b c) = RsResult a b (g c)-        f (RsEvent         a b c) = RsEvent a b c--        g :: Result k a b -> Result k a NoShow-        g VoidResult                       = VoidResult-        g (RowsResult              a  b  ) = RowsResult a (map (const NoShow) b)-        g (SetKeyspaceResult       a     ) = SetKeyspaceResult a-        g (SchemaChangeResult      a     ) = SchemaChangeResult a-        g (PreparedResult (QueryId a) b c) = PreparedResult (QueryId a) b c---------------------------------------------------------------------------------- HashCollision--data HashCollision = HashCollision !Lazy.Text !Lazy.Text-    deriving Typeable--instance Exception HashCollision--instance Show HashCollision where-    show (HashCollision a b) = showString "cql-io: hash collision: "-                             . shows a-                             . showString " "-                             . shows b-                             $ ""---------------------------------------------------------------------------------- Authentication---- | The (unique) name of a SASL authentication mechanism.------ In the case of Cassandra, this is currently always the fully-qualified--- Java class name of the configured server-side @IAuthenticator@--- implementation.-newtype AuthMechanism = AuthMechanism Text-    deriving (Eq, Ord, Show, IsString, Hashable)--data AuthenticationError-    = AuthenticationRequired !AuthMechanism-    | UnexpectedAuthenticationChallenge !AuthMechanism !AuthChallenge--instance Exception AuthenticationError--instance Show AuthenticationError where-    show (AuthenticationRequired a)-        = showString "cql-io: authentication required: "-        . shows a-        $ ""--    show (UnexpectedAuthenticationChallenge n c)-        = showString "cql-io: unexpected authentication challenge: '"-        . shows c-        . showString "' using mechanism '"-        . shows n-        . showString "'"-        $ ""--ignore :: IO () -> IO ()-ignore a = catchAll a (const $ return ())-{-# INLINE ignore #-}

@@ -0,0 +1,344 @@+-- 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/.++-- | This driver operates on some state which must be initialised prior to+-- executing client operations and terminated eventually. The library uses+-- <http://hackage.haskell.org/package/tinylog tinylog> for its logging+-- output and expects a 'Logger'.+--+-- For example (here using the @OverloadedStrings@ extension) :+--+-- @+-- > import Data.Text (Text)+-- > import Data.Functor.Identity+-- > import Database.CQL.IO as Client+-- > import qualified System.Logger as Logger+-- >+-- > g <- Logger.new Logger.defSettings+-- > c <- Client.init g defSettings+-- > let q = "SELECT cql_version from system.local" :: QueryString R () (Identity Text)+-- > let p = defQueryParams One ()+-- > runClient c (query q p)+-- [Identity "3.4.4"]+-- > shutdown c+-- @+--++{-# LANGUAGE DeriveFunctor #-}+{-# LANGUAGE LambdaCase    #-}++module Database.CQL.IO+    ( -- * Client Settings+      Settings+    , S.defSettings+    , addContact+    , setCompression+    , setConnectTimeout+    , setContacts+    , setIdleTimeout+    , setKeyspace+    , setMaxConnections+    , setMaxStreams+    , setMaxTimeouts+    , setPolicy+    , setPoolStripes+    , setPortNumber+    , PrepareStrategy (..)+    , setPrepareStrategy+    , setProtocolVersion+    , setResponseTimeout+    , setSendTimeout+    , setRetrySettings+    , setMaxRecvBuffer+    , setSSLContext++      -- ** Logging+    , Logger (..)+    , LogLevel (..)+    , setLogger+    , nullLogger+    , stdoutLogger++      -- ** Authentication+    , setAuthentication+    , Authenticator (..)+    , AuthContext+    , ConnId+    , authConnId+    , authHost+    , AuthMechanism (..)+    , AuthUser      (..)+    , AuthPass      (..)+    , passwordAuthenticator++      -- ** Retry Settings+    , RetrySettings+    , noRetry+      -- *** Default+    , defRetrySettings+    , defRetryPolicy+    , defRetryHandlers+      -- *** Eager+    , eagerRetrySettings+    , eagerRetryPolicy+    , eagerRetryHandlers+      -- *** Configuration+    , setRetryPolicy+    , setRetryHandlers+    , adjustConsistency+    , adjustSendTimeout+    , adjustResponseTimeout++      -- ** Load-balancing+    , Policy (..)+    , random+    , roundRobin++      -- *** Hosts+    , Host+    , HostEvent (..)+    , InetAddr  (..)+    , hostAddr+    , dataCentre+    , rack++      -- * Client Monad+    , Client+    , MonadClient (..)+    , ClientState+    , DebugInfo   (..)+    , init+    , runClient+    , shutdown+    , debugInfo++      -- * Queries+      -- $queries+    , R, W, S+    , QueryParams       (..)+    , defQueryParams+    , Consistency       (..)+    , SerialConsistency (..)+    , Identity          (..)++      -- ** Basic Queries+    , QueryString (..)+    , query+    , query1+    , write+    , schema++      -- ** Prepared Queries+    , PrepQuery+    , prepared+    , queryString++      -- ** Paging+    , Page (..)+    , emptyPage+    , paginate++      -- ** Lightweight Transactions+    , Row+    , fromRow+    , trans++      -- ** Batch Queries+    , BatchM+    , addQuery+    , addPrepQuery+    , setType+    , setConsistency+    , setSerialConsistency+    , batch++      -- ** Retries+    , retry+    , once++      -- ** Low-Level Queries+      -- $low-level-queries+    , RunQ (..)+    , HostResponse (..)+    , request+    , getResult++      -- * Exceptions+    , ProtocolError       (..)+    , HostError           (..)+    , ConnectionError     (..)+    , ResponseError       (..)+    , AuthenticationError (..)+    , HashCollision       (..)+    ) where++import Control.Applicative+import Data.Functor.Identity+import Data.Maybe (isJust, listToMaybe)+import Database.CQL.Protocol+import Database.CQL.IO.Batch hiding (batch)+import Database.CQL.IO.Client+import Database.CQL.IO.Cluster.Host+import Database.CQL.IO.Cluster.Policies+import Database.CQL.IO.Connection.Settings as C+import Database.CQL.IO.Exception+import Database.CQL.IO.Log+import Database.CQL.IO.PrepQuery+import Database.CQL.IO.Settings as S+import Prelude hiding (init)++import qualified Database.CQL.IO.Batch as B++-- $queries+--+-- Queries are defined either as 'QueryString's or 'PrepQuery's.+-- Both types carry three phantom type parameters used to describe+-- the query, input and output types, respectively, as follows:+--+--   * @__k__@ is one of 'R'ead, 'W'rite or 'S'chema.+--   * @__a__@ is the tuple type for the input, i.e. for the+--     parameters bound by positional (@?@) or named (@:foo@) placeholders.+--   * @__b__@ is the tuple type for the outputs, i.e. for the+--     columns selected in a query.+--+-- Thereby every type used in an input or output tuple must be an instance+-- of the 'Cql' typeclass. It is the responsibility of user code+-- that the type ascription of a query matches the order, number and types of+-- the parameters. For example:+--+-- @+-- myQuery :: QueryString R (Identity UUID) (Text, Int, Maybe UTCTime)+-- myQuery = "select name, age, birthday from user where id = ?"+-- @+--+-- In this example, the query is declared as a 'R'ead with a single+-- input (id) and three outputs (name, age and birthday).+--+-- Note that a single input or output type needs to be wrapped+-- in the 'Identity' newtype, for which there is a `Cql` instance,+-- in order to avoid overlapping instances.+--+-- It is a common strategy to use additional @newtype@s with derived+-- @Cql@ instances for additional type safety, e.g.+--+-- @+-- newtype UserId = UserId UUID deriving (Eq, Show, Cql)+-- @+--+-- The input and output tuples can further be automatically+-- converted from and to records via the 'Database.CQL.Protocol.Record'+-- typeclass, whose instances can be generated via @TemplateHaskell@,+-- if desired.+--+-- __Note on null values__+--+-- In principle, any column in Cassandra is /nullable/, i.e. may be+-- be set to @null@ as a result of row operations. It is therefore+-- important that any output type of a query that may be null+-- is wrapped in the 'Maybe' type constructor.+-- It is a common pitfall that a column is assumed to never contain+-- null values, when in fact partial updates or deletions on a row,+-- including via the use of TTLs, may result in null values and thus+-- runtime errors when processing the responses.++-- $low-level-queries+--+-- /Note/: Use of the these functions may require additional imports from+-- @Database.CQL.Protocol@ or its submodules in order to construct+-- 'Request's and evaluate 'Response's.++-- | 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 (HostResponse k a b)++instance RunQ QueryString where+    runQ q p = request (RqQuery (Query q p))++instance RunQ PrepQuery where+    runQ q = liftClient . execute q++-- | Run a CQL read-only query returning a list of results.+query :: (MonadClient m, Tuple a, Tuple b, RunQ q) => q R a b -> QueryParams a -> m [b]+query q p = do+    r <- runQ q p+    getResult r >>= \case+        RowsResult _ b -> return b+        _              -> unexpected r++-- | Run a CQL read-only query returning a single result.+query1 :: (MonadClient m, Tuple a, Tuple b, RunQ q) => q R a b -> QueryParams a -> m (Maybe b)+query1 q p = listToMaybe <$> query q p++-- | Run a CQL write-only query (e.g. insert\/update\/delete),+-- returning no result.+--+-- /Note: If the write operation is conditional, i.e. is in fact a "lightweight+-- transaction" returning a result, 'trans' must be used instead./+write :: (MonadClient m, Tuple a, RunQ q) => q W a () -> QueryParams a -> m ()+write q p = do+    r <- runQ q p+    getResult r >>= \case+        VoidResult -> return ()+        _          -> unexpected r++-- | Run a CQL conditional write query (e.g. insert\/update\/delete) as a+-- "lightweight transaction", returning the result 'Row's describing the+-- outcome.+trans :: (MonadClient m, Tuple a, RunQ q) => q W a Row -> QueryParams a -> m [Row]+trans q p = do+    r <- runQ q p+    getResult r >>= \case+        RowsResult _ b -> return b+        _              -> unexpected 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)+schema q p = do+    r <- runQ q p+    getResult r >>= \case+        SchemaChangeResult s -> return $ Just s+        VoidResult           -> return Nothing+        _                    -> unexpected r++-- | Run a batch query against a Cassandra node.+batch :: MonadClient m => BatchM () -> m ()+batch = liftClient . B.batch++-- | Return value of 'paginate'. Contains the actual result values as well+-- as an indication of whether there is more data available and the actual+-- action to fetch the next page.+data Page a = Page+    { hasMore  :: !Bool+    , result   :: [a]+    , nextPage :: Client (Page a)+    } deriving (Functor)++-- | A page with an empty result list.+emptyPage :: Page a+emptyPage = Page False [] (return emptyPage)++-- | Run a CQL read-only query against a Cassandra node.+--+-- This function is like 'query', but limits the result size to 10000+-- (default) unless there is an explicit size restriction given in+-- 'QueryParams'. The returned 'Page' can be used to continue the query.+--+-- 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 :: (MonadClient m, Tuple a, Tuple b, RunQ q) => q R a b -> QueryParams a -> m (Page b)+paginate q p = do+    let p' = p { pageSize = pageSize p <|> Just 10000 }+    r <- runQ q p'+    getResult r >>= \case+        RowsResult m b ->+            if isJust (pagingState m) then+                return $ Page True b (paginate q p' { queryPagingState = pagingState m })+            else+                return $ Page False b (return emptyPage)+        _ -> unexpected r+

@@ -0,0 +1,79 @@+-- 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 CPP                        #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE LambdaCase                 #-}++module Database.CQL.IO.Batch+    ( BatchM+    , batch+    , addQuery+    , addPrepQuery+    , setType+    , setConsistency+    , setSerialConsistency+    ) where++import Control.Applicative+import Control.Concurrent.STM (atomically)+import Control.Monad.IO.Class+import Control.Monad.Trans+import Control.Monad.Trans.State.Strict+import Database.CQL.IO.Client+import Database.CQL.IO.Connection (Raw)+import Database.CQL.IO.PrepQuery+import Database.CQL.Protocol+import Prelude++-- | 'Batch' construction monad.+newtype BatchM a = BatchM+    { unBatchM :: StateT Batch Client a+    } deriving (Functor, Applicative, Monad)++-- | Execute the complete 'Batch' statement.+batch :: BatchM a -> Client ()+batch m = do+    b <- execStateT (unBatchM m) (Batch BatchLogged [] Quorum Nothing)+    r <- executeWithPrepare Nothing (RqBatch b :: Raw Request)+    getResult r >>= \case+        VoidResult -> return ()+        _          -> unexpected r++-- | Add a query to this batch.+addQuery :: (Show a, Tuple a, Tuple b) => QueryString W a b -> a -> BatchM ()+addQuery q p = BatchM $ modify' $ \b ->+    b { batchQuery = BatchQuery q p : batchQuery b }++-- | Add a prepared query to this batch.+addPrepQuery :: (Show a, Tuple a, Tuple b) => PrepQuery W a b -> a -> BatchM ()+addPrepQuery q p = BatchM $ do+    pq <- lift preparedQueries+    maybe (fresh pq) add =<< liftIO (atomically (lookupQueryId q pq))+  where+    fresh pq = do+        i <- snd <$> lift (prepare Nothing (queryString q))+        liftIO $ atomically (insert q i pq)+        add i++    add i = modify' $ \b -> b { batchQuery = BatchPrepared i p : batchQuery b }++-- | Set the type of this batch.+setType :: BatchType -> BatchM ()+setType t = BatchM $ modify' $ \b -> b { batchType = t }++-- | Set 'Batch' consistency level.+setConsistency :: Consistency -> BatchM ()+setConsistency c = BatchM $ modify' $ \b -> b { batchConsistency = c }++-- | Set 'Batch' serial consistency.+setSerialConsistency :: SerialConsistency -> BatchM ()+setSerialConsistency c = BatchM $ modify' $ \b -> b { batchSerialConsistency = Just c }++#if ! MIN_VERSION_transformers(0,4,0)+modify' :: Monad m => (s -> s) -> StateT s m ()+modify' f = do+    s <- get+    put $! f s+#endif

@@ -0,0 +1,806 @@+-- 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 GeneralizedNewtypeDeriving #-}+{-# LANGUAGE LambdaCase                 #-}+{-# LANGUAGE OverloadedStrings          #-}+{-# LANGUAGE ScopedTypeVariables        #-}+{-# LANGUAGE TemplateHaskell            #-}+{-# LANGUAGE TupleSections              #-}+{-# LANGUAGE ViewPatterns               #-}++module Database.CQL.IO.Client+    ( Client+    , MonadClient (..)+    , ClientState+    , DebugInfo (..)+    , ControlState (..)+    , runClient+    , init+    , shutdown+    , request+    , requestN+    , request1+    , execute+    , executeWithPrepare+    , prepare+    , retry+    , once+    , debugInfo+    , preparedQueries+    , withPrepareStrategy+    , getResult+    , unexpected+    , C.defQueryParams+    ) where++import Control.Applicative+import Control.Concurrent (threadDelay, forkIO)+import Control.Concurrent.Async (async, wait)+import Control.Concurrent.STM (STM, atomically)+import Control.Concurrent.STM.TVar+import Control.Exception (IOException, SomeAsyncException (..))+import Control.Lens (makeLenses, (^.), set, over, view)+import Control.Monad (when, unless)+import Control.Monad.Catch+import Control.Monad.IO.Class+import Control.Monad.IO.Unlift+import Control.Monad.Reader (ReaderT (..), runReaderT, MonadReader, ask)+import Control.Monad.Trans.Class+import Control.Monad.Trans.Except+import Control.Retry (capDelay, exponentialBackoff, rsIterNumber)+import Control.Retry (recovering)+import Data.Foldable (for_, foldrM)+import Data.List (find)+import Data.Map.Strict (Map)+import Data.Maybe (fromMaybe, listToMaybe)+import Data.Semigroup+import Data.Text.Encoding (encodeUtf8)+import Data.Word+import Database.CQL.IO.Cluster.Host+import Database.CQL.IO.Cluster.Policies+import Database.CQL.IO.Connection (Connection, host, Raw)+import Database.CQL.IO.Connection.Settings+import Database.CQL.IO.Exception+import Database.CQL.IO.Jobs+import Database.CQL.IO.Log+import Database.CQL.IO.Pool (Pool)+import Database.CQL.IO.PrepQuery (PrepQuery, PreparedQueries)+import Database.CQL.IO.Settings+import Database.CQL.IO.Signal+import Database.CQL.IO.Timeouts (TimeoutManager)+import Database.CQL.Protocol hiding (Map)+import OpenSSL.Session (SomeSSLException)+import Prelude hiding (init)++import qualified Control.Monad.Reader              as Reader+import qualified Control.Monad.State.Strict        as S+import qualified Control.Monad.State.Lazy          as LS+import qualified Data.List.NonEmpty                as NE+import qualified Data.Map.Strict                   as Map+import qualified Database.CQL.IO.Cluster.Discovery as Disco+import qualified Database.CQL.IO.Connection        as C+import qualified Database.CQL.IO.Pool              as Pool+import qualified Database.CQL.IO.PrepQuery         as PQ+import qualified Database.CQL.IO.Timeouts          as TM+import qualified Database.CQL.Protocol             as Cql++data ControlState+    = Connected+    | Reconnecting+    | Disconnected+    deriving (Eq, Ord, Show)++data Control = Control+    { _state      :: !ControlState+    , _connection :: !Connection+    }++data Context = Context+    { _settings :: !Settings+    , _timeouts :: !TimeoutManager+    , _sigMonit :: !(Signal HostEvent)+    }++-- | Opaque client state/environment.+data ClientState = ClientState+    { _context     :: !Context+    , _policy      :: !Policy+    , _prepQueries :: !PreparedQueries+    , _control     :: !(TVar Control)+    , _hostmap     :: !(TVar (Map Host Pool))+    , _jobs        :: !(Jobs InetAddr)+    }++makeLenses ''Control+makeLenses ''Context+makeLenses ''ClientState++-- | The Client monad.+--+-- A simple reader monad on `IO` around some internal state. Prior to executing+-- this monad via 'runClient', its state must be initialised through+-- 'Database.CQL.IO.Client.init' and after finishing operation it should be+-- terminated with 'shutdown'.+--+-- To lift 'Client' actions into another monad, see 'MonadClient'.+newtype Client a = Client+    { client :: ReaderT ClientState IO a+    } deriving ( Functor+               , Applicative+               , Monad+               , MonadIO+               , MonadUnliftIO+               , MonadThrow+               , MonadCatch+               , MonadMask+               , MonadReader ClientState+               )++-- | Monads in which 'Client' actions may be embedded.+class (MonadIO m, MonadThrow m) => MonadClient m+  where+    -- | Lift a computation from 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++instance MonadClient m => MonadClient (ReaderT r m) where+    liftClient     = lift . liftClient+    localState f m = ReaderT (localState f . runReaderT m)++instance MonadClient m => MonadClient (S.StateT s m) where+    liftClient     = lift . liftClient+    localState f m = S.StateT (localState f . S.runStateT m)++instance MonadClient m => MonadClient (LS.StateT s m) where+    liftClient     = lift . liftClient+    localState f m = LS.StateT (localState f . LS.runStateT m)++instance MonadClient m => MonadClient (ExceptT e m) where+    liftClient     = lift . liftClient+    localState f m = ExceptT $ localState f (runExceptT m)++-----------------------------------------------------------------------------+-- API++-- | Execute the client monad.+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)++-- | Execute a client action once, without retries, i.e.+--+-- @once action = retry noRetry action@.+--+-- Primarily for use in applications where global 'RetrySettings'+-- are configured and need to be selectively disabled for individual+-- queries.+once :: MonadClient m => m a -> m a+once = retry noRetry++-- | Change the default 'PrepareStrategy' for the given client action.+withPrepareStrategy :: MonadClient m => PrepareStrategy -> m a -> m a+withPrepareStrategy s = localState (set (context.settings.prepStrategy) s)++-- | Send a 'Request' to the server and return a 'Response'.+--+-- This function will first ask the clients load-balancing 'Policy' for+-- some host and use its connection pool to acquire a connection for+-- request transmission.+--+-- 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.+--+-- The request is retried according to the configured 'RetrySettings'.+request :: (MonadClient m, Tuple a, Tuple b) => Request k a b -> m (HostResponse k a b)+request a = liftClient $ do+    n <- liftIO . hostCount =<< view policy+    withRetries (requestN n) a++-- | Send a request to a host chosen by the configured host policy.+--+-- Tries up to @max(1,n)@ hosts. If no host can execute the request,+-- a 'HostError' is thrown. Specifically:+--+--   * If no host is available from the 'Policy', 'NoHostAvailable' is thrown.+--   * If no host can execute the request, e.g. because all streams+--     on all connections are occupied, 'HostsBusy' is thrown.+requestN :: (Tuple b, Tuple a)+    => Word+    -> Request k a b+    -> ClientState+    -> Client (HostResponse k a b)+requestN !n a s = liftIO (select (s^.policy)) >>= \case+    Nothing -> replaceControl >> throwM NoHostAvailable+    Just  h -> tryRequest1 h a s >>= \case+        Just hr -> return hr+        Nothing -> if n > 1+            then requestN (n - 1) a s+            else throwM HostsBusy++-- | Send a 'Request' to a specific 'Host'.+--+-- If the request cannot be executed on the given host, e.g.+-- because all connections are occupied, 'HostsBusy' is thrown.+request1 :: (Tuple a, Tuple b)+    => Host+    -> Request k a b+    -> ClientState+    -> Client (HostResponse k a b)+request1 h r s = do+    rs <- tryRequest1 h r s+    maybe (throwM HostsBusy) return rs++-- | Try to send a 'Request' to a specific 'Host'.+--+-- If the request cannot be executed on the given host, e.g.+-- because all connections are occupied, 'Nothing' is returned.+tryRequest1 :: (Tuple a, Tuple b)+    => Host+    -> Request k a b+    -> ClientState+    -> Client (Maybe (HostResponse k a b))+tryRequest1 h a s = do+    pool <- Map.lookup h <$> readTVarIO' (s^.hostmap)+    case pool of+        Just p -> do+            result <- Pool.with p exec `catches` handlers+            for_ result $ \(HostResponse _ r) ->+                for_ (Cql.warnings r) $ \w ->+                    logWarn' $ "Server warning: " <> byteString (encodeUtf8 w)+            return result+        Nothing -> do+            logError' $ "No pool for host: " <> string8 (show h)+            p' <- mkPool (s^.context) h+            atomically' $ modifyTVar' (s^.hostmap) (Map.alter (maybe (Just p') Just) h)+            tryRequest1 h a s+  where+    exec c = do+        r <- C.request c a+        return $ HostResponse h r++    handlers =+        [ Handler $ \(e :: ConnectionError)  -> onConnectionError e+        , Handler $ \(e :: IOException)      -> onConnectionError e+        , Handler $ \(e :: SomeSSLException) -> onConnectionError e+        ]++    onConnectionError exc = do+        e <- ask+        logWarn' (string8 (show exc))+        -- Tell the policy that the host is down until monitoring confirms+        -- it is still up, which will be signalled by a subsequent 'HostUp'+        -- event.+        liftIO $ ignore $ onEvent (e^.policy) (HostDown (h^.hostAddr))+        runJob_ (e^.jobs) (h^.hostAddr) $+            runClient e $ monitor (Ms 0) (Ms 30000) h+        -- Any connection error may indicate a problem with the+        -- control connection, if it uses the same host.+        ch <- fmap (view (connection.host)) . readTVarIO' =<< view control+        when (h == ch) $ do+            ok <- checkControl+            unless ok replaceControl+        throwM exc++------------------------------------------------------------------------------+-- Prepared queries++-- | 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 (HostResponse k a b)+executeWithPrepare mh rq+    | Just h <- mh = exec (request1 h)+    | otherwise    = do+        p <- view policy+        n <- liftIO $ hostCount p+        exec (requestN n)+  where+    exec action = do+        r <- withRetries action rq+        case hrResponse r of+            RsError _ _ (Unprepared _ i) -> do+                pq <- preparedQueries+                qs <- atomically' (PQ.lookupQueryString (QueryId i) pq)+                case qs of+                    Nothing -> throwM $ UnexpectedQueryId (QueryId i)+                    Just  s -> do+                        (h, _) <- prepare (Just LazyPrepare) (s :: Raw QueryString)+                        executeWithPrepare (Just h) rq+            _ -> return r++-- | Prepare the given query according to the given 'PrepareStrategy',+-- returning the resulting 'QueryId' and 'Host' which was used for+-- preparation.+prepare :: (Tuple b, Tuple a) => Maybe PrepareStrategy -> QueryString k a b -> Client (Host, QueryId k a b)+prepare (Just LazyPrepare) qs = do+    s <- ask+    n <- liftIO $ hostCount (s^.policy)+    r <- withRetries (requestN n) (RqPrepare (Prepare qs))+    getPreparedQueryId r++prepare (Just EagerPrepare) qs = view policy+    >>= liftIO . current+    >>= mapM (action (RqPrepare (Prepare qs)))+    >>= first+  where+    action rq h = withRetries (request1 h) rq >>= getPreparedQueryId++    first (x:_) = return x+    first []    = replaceControl >> throwM NoHostAvailable++prepare Nothing qs = do+    ps <- view (context.settings.prepStrategy)+    prepare (Just ps) qs++-- | Execute a prepared query (transparently re-preparing if necessary).+execute :: (Tuple b, Tuple a) => PrepQuery k a b -> QueryParams a -> Client (HostResponse k a b)+execute q p = do+    pq <- view prepQueries+    maybe (new pq) (exec Nothing) =<< atomically' (PQ.lookupQueryId q pq)+  where+    exec h i = executeWithPrepare h (RqExecute (Execute i p))+    new pq = do+        (h, i) <- prepare (Just LazyPrepare) (PQ.queryString q)+        atomically' (PQ.insert q i pq)+        exec (Just h) i++prepareAllQueries :: Host -> Client ()+prepareAllQueries h = do+    pq <- view prepQueries+    qs <- atomically' $ PQ.queryStrings pq+    for_ qs $ \q ->+        let qry = QueryString q :: Raw QueryString in+        withRetries (request1 h) (RqPrepare (Prepare qry))++------------------------------------------------------------------------------+-- Debug info++data DebugInfo = DebugInfo+    { policyInfo :: String+        -- ^ Host 'Policy' string representation.+    , jobInfo :: [InetAddr]+        -- ^ Hosts with running background jobs (e.g. monitoring of hosts+        -- currently considered down).+    , hostInfo :: [Host]+        -- ^ All known hosts.+    , controlInfo :: (Host, ControlState)+        -- ^ Control connection information.+    }++instance Show DebugInfo where+    show dbg = showString "running jobs: "+             . shows (jobInfo dbg)+             . showString "\nknown hosts: "+             . shows (hostInfo dbg)+             . showString "\npolicy info: "+             . shows (policyInfo dbg)+             . showString "\ncontrol host: "+             . shows (controlInfo dbg)+             $ ""++debugInfo :: MonadClient m => m DebugInfo+debugInfo = liftClient $ do+    hosts <- Map.keys <$> (readTVarIO' =<< view hostmap)+    pols  <- liftIO . display =<< view policy+    jbs   <- listJobKeys =<< view jobs+    ctrl  <- (\(Control s c) -> (c^.host, s)) <$> (readTVarIO' =<< view control)+    return $ DebugInfo pols jbs hosts ctrl++preparedQueries :: Client PreparedQueries+preparedQueries = view prepQueries++-----------------------------------------------------------------------------+-- Initialisation++-- | Initialise client state with the given 'Settings' using the provided+-- 'Logger' for all it's logging output.+init :: MonadIO m => Settings -> m ClientState+init s = liftIO $ do+    tom <- TM.create (Ms 250)+    ctx <- Context s tom <$> signal+    bracketOnError (mkContact ctx) C.close $ \con -> do+        pol <- s^.policyMaker+        cst <- ClientState ctx+                <$> pure pol+                <*> PQ.new+                <*> newTVarIO (Control Connected con)+                <*> newTVarIO Map.empty+                <*> newJobs+        ctx^.sigMonit |-> onEvent pol+        runClient cst (setupControl con)+        return cst++-- | Try to establish a connection to one of the initial contacts.+mkContact :: Context -> IO Connection+mkContact (Context s t _) = tryAll (s^.contacts) mkConnection+  where+    mkConnection h = do+        as <- C.resolve h (s^.portnumber)+        NE.fromList as `tryAll` doConnect++    doConnect a = do+        logDebug (s^.logger) $ "Connecting to " <> string8 (show a)+        c <- C.connect (s^.connSettings) t (s^.protoVersion) (s^.logger) (Host a "" "")+        return c++discoverPeers :: MonadIO m => Context -> Connection -> m [Host]+discoverPeers ctx c = liftIO $ do+    let p = ctx^.settings.portnumber+    map (peer2Host p . asRecord) <$> C.query c One Disco.peers ()++mkPool :: MonadIO m => Context -> Host -> m Pool+mkPool ctx h = liftIO $ do+    let s = ctx^.settings+    let m = s^.connSettings.maxStreams+    Pool.create (connOpen s) connClose (ctx^.settings.logger) (s^.poolSettings) m+  where+    lgr = ctx^.settings.logger++    connOpen s = do+        c <- C.connect (s^.connSettings) (ctx^.timeouts) (s^.protoVersion) lgr h+        logDebug lgr $ "Connection established: " <> string8 (show c)+        return c++    connClose c = do+        C.close c+        logDebug lgr $ "Connection closed: " <> string8 (show c)++-----------------------------------------------------------------------------+-- Termination++-- | Terminate client state, i.e. end all running background checks and+-- shutdown all connection pools. Once this is entered, the client+-- will eventually be shut down, though an asynchronous exception can+-- interrupt the wait for that to occur.+shutdown :: MonadIO m => ClientState -> m ()+shutdown s = liftIO $ asyncShutdown >>= wait+  where+    asyncShutdown = async $ do+        TM.destroy (s^.context.timeouts) True+        cancelJobs (s^.jobs)+        ignore $ C.close . view connection =<< readTVarIO (s^.control)+        mapM_ Pool.destroy . Map.elems =<< readTVarIO (s^.hostmap)++-----------------------------------------------------------------------------+-- Monitoring++-- | @monitor initialDelay maxDelay host@ tries to establish a connection+-- to @host@ after @initialDelay@. If the connection attempt fails, it is+-- retried with exponentially increasing delays, up to a maximum delay of+-- @maxDelay@. When a connection attempt suceeds, a 'HostUp' event is+-- signalled.+--+-- The function returns when one of the following conditions is met:+--+--   1. The connection attempt suceeds.+--   2. The host is no longer found to be in the client's known host map.+--+-- I.e. as long as the host is still known to the client and is unreachable, the+-- connection attempts continue. Both @initialDelay@ and @maxDelay@ are bounded+-- by a limit of 5 minutes.+monitor :: Milliseconds -> Milliseconds -> Host -> Client ()+monitor initial maxDelay h = do+    liftIO $ threadDelay (toMicros initial)+    logInfo' $ "Monitoring: " <> string8 (show h)+    hostCheck 0+  where+    hostCheck :: Int -> Client ()+    hostCheck !n = do+        hosts <- liftIO . readTVarIO =<< view hostmap+        when (Map.member h hosts) $ do+            isUp <- C.canConnect h+            if isUp then do+                sig <- view (context.sigMonit)+                liftIO $ sig $$ (HostUp (h^.hostAddr))+                logInfo' $ "Reachable: " <> string8 (show h)+            else do+                logInfo' $ "Unreachable: " <> string8 (show h)+                liftIO $ threadDelay (2^n * minDelay)+                hostCheck (min (n + 1) maxExp)++    -- Bounded to 5min+    toMicros :: Milliseconds -> Int+    toMicros (Ms s) = min (s * 1000) (5 * 60 * 1000000)++    minDelay :: Int+    minDelay = 50000 -- 50ms++    maxExp :: Int+    maxExp = let steps = fromIntegral (toMicros maxDelay `div` minDelay) :: Double+              in floor (logBase 2 steps)++-----------------------------------------------------------------------------+-- 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 (HostResponse k a b))+    -> Request k a b+    -> Client (HostResponse k a b)+withRetries fn a = do+    s <- ask+    let how = s^.context.settings.retrySettings.retryPolicy+    let what = s^.context.settings.retrySettings.retryHandlers+    r <- try $ recovering how what $ \i -> do+        hr <- if rsIterNumber i == 0+                 then fn a s+                 else fn (newRequest s) (adjust s)+        -- [Note: Error responses]+        maybe (return hr) throwM (toResponseError hr)+    return $ either fromResponseError id r+  where+    adjust s =+        let Ms x = s^.context.settings.retrySettings.sendTimeoutChange+            Ms y = s^.context.settings.retrySettings.recvTimeoutChange+        in over (context.settings.connSettings.sendTimeout)     (Ms . (+ x) . ms)+         . over (context.settings.connSettings.responseTimeout) (Ms . (+ y) . ms)+         $ 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 })+                    RqExecute (Execute q p) -> RqExecute (Execute q p { consistency = c })+                    RqBatch b               -> RqBatch b { batchConsistency = c }+                    _                       -> a++------------------------------------------------------------------------------+-- Control connection handling+--+-- The control connection is dedicated to maintaining the client's+-- view of the cluster topology. There is a single control connection in a+-- client's 'ClientState' at any particular time.++-- | Setup and install the given connection as the new control+-- connection, replacing the current one.+setupControl :: Connection -> Client ()+setupControl c = do+    env <- ask+    pol <- view policy+    ctx <- view context+    l <- updateHost (c^.host) . listToMaybe <$> C.query c One Disco.local ()+    r <- discoverPeers ctx c+    (up, down) <- mkHostMap ctx pol (l:r)+    m <- view hostmap+    let h = Map.union up down+    atomically' $ writeTVar m h+    liftIO $ setup pol (Map.keys up) (Map.keys down)+    C.register c C.allEventTypes (runClient env . onCqlEvent)+    logInfo' $ "Known hosts: " <> string8 (show (Map.keys h))+    j <- view jobs+    for_ (Map.keys down) $ \d ->+        runJob j (d^.hostAddr) $+            runClient env $ monitor (Ms 1000) (Ms 60000) d+    ctl <- view control+    let c' = set C.host l c+    atomically' $ writeTVar ctl (Control Connected c')+    logInfo' $ "New control connection: " <> string8 (show c')++-- | Initialise connection pools for the given hosts, checking for+-- acceptability with the host policy and separating them by reachability.+mkHostMap :: Context -> Policy -> [Host] -> Client (Map Host Pool, Map Host Pool)+mkHostMap c p = liftIO . foldrM checkHost (Map.empty, Map.empty)+  where+    checkHost h (up, down) = do+        okay <- acceptable p h+        if okay then do+            isUp <- C.canConnect h+            if isUp then do+                up' <- Map.insert h <$> mkPool c h <*> pure up+                return (up', down)+            else do+                down' <- Map.insert h <$> mkPool c h <*> pure down+                return (up, down')+        else+            return (up, down)++-- | Check if the control connection is healthy.+checkControl :: Client Bool+checkControl = do+    cc <- view connection <$> (readTVarIO' =<< view control)+    rs <- liftIO $ C.requestRaw cc (RqOptions Options)+    return $ case rs of+        RsSupported {} -> True+        _              -> False+  `recover`+    False++-- | Asynchronously replace the control connection.+--+-- Invariants:+--+--   1) When the control connection is in state 'Reconnecting' there+--      is a thread running that attempts to establish a new control+--      connection.+--+--   2) There is only one thread performing a reconnect at a time.+--+-- To that end, the 'ControlState' acts as a mutex that is acquired+-- in state 'Reconnecting' and must eventually be released by either+-- a successful reconnect with state 'Connected' or a (fatal) failure+-- with state 'Disconnected'. In the latter case, further failing+-- requests may trigger another recovery attempt of the control+-- connection.+replaceControl :: Client ()+replaceControl = do+    e <- ask+    let l = e^.context.settings.logger+    liftIO $ mask $ \restore -> do+        cc <- setReconnecting e+        for_ cc $ \c -> forkIO $+            restore $ do+                ignore (C.close c)+                reconnect e l+              `catchAll` \ex -> do+                logError l $ "Control connection reconnect aborted: " <> string8 (show ex)+                atomically $ modifyTVar' (e^.control) (set state Disconnected)+  where+    setReconnecting e = atomically $ do+        ctrl <- readTVar (e^.control)+        if ctrl^.state /= Reconnecting+            then do+                writeTVar (e^.control) (set state Reconnecting ctrl)+                return $ Just (ctrl^.connection)+            else+                return Nothing++    reconnect e l = recovering adInf (onExc l) $ \_ -> do+        hosts <- NE.nonEmpty . Map.keys <$> readTVarIO (e^.hostmap)+        case hosts of+            Just hs -> hs `tryAll` (runClient e . renewControl)+                `catch` \x -> case fromException x of+                    Just (SomeAsyncException _) -> throwM x+                    Nothing                     -> do+                        logError l "All known hosts unreachable."+                        runClient e rebootControl+            Nothing -> do+                logError l "No known hosts."+                runClient e rebootControl++    adInf = capDelay 5000000 (exponentialBackoff 5000)++    onExc l =+        [ const $ Handler $ \(_ :: SomeAsyncException) -> return False+        , const $ Handler $ \(e :: SomeException)      -> do+            logError l $ "Replacement of control connection failed with: "+                <> string8 (show e)+                <> ". Retrying ..."+            return True+        ]++-- | Create a new connection to a known host and set it up+-- as the new control connection.+renewControl :: Host -> Client ()+renewControl h = do+    ctx <- view context+    logInfo' "Renewing control connection with known host ..."+    let s = ctx^.settings+    bracketOnError+        (C.connect (s^.connSettings) (ctx^.timeouts) (s^.protoVersion) (s^.logger) h)+        (liftIO . C.close)+        setupControl++-- | Create a new connection to one of the initial contacts+-- and set it up as the new control connection.+rebootControl :: Client ()+rebootControl = do+    e <- ask+    logInfo' "Renewing control connection with initial contacts ..."+    bracketOnError+        (liftIO (mkContact (e^.context)))+        (liftIO . C.close)+        setupControl++-----------------------------------------------------------------------------+-- Event handling++onCqlEvent :: Event -> Client ()+onCqlEvent x = do+    logInfo' $ "Event: " <> string8 (show x)+    pol <- view policy+    prt <- view (context.settings.portnumber)+    case x of+        StatusEvent Down (sock2inet prt -> a) ->+            liftIO $ onEvent pol (HostDown a)+        TopologyEvent RemovedNode (sock2inet prt -> a) -> do+            hmap <- view hostmap+            atomically' $+                modifyTVar' hmap (Map.filterWithKey (\h _ -> h^.hostAddr /= a))+            liftIO $ onEvent pol (HostGone a)+        StatusEvent Up (sock2inet prt -> a) -> do+            s <- ask+            startMonitor s a+        TopologyEvent NewNode (sock2inet prt -> a) -> do+            s <- ask+            let ctx  = s^.context+            let hmap = s^.hostmap+            ctrl <- readTVarIO' (s^.control)+            let c = ctrl^.connection+            peers <- liftIO $ discoverPeers ctx c `recover` []+            let h = fromMaybe (Host a "" "") $ find ((a == ) . view hostAddr) peers+            okay <- liftIO $ acceptable pol h+            when okay $ do+                p <- mkPool ctx h+                atomically' $ modifyTVar' hmap (Map.alter (maybe (Just p) Just) h)+                liftIO $ onEvent pol (HostNew h)+                tryRunJob_ (s^.jobs) a $ runClient s (prepareAllQueries h)+        SchemaEvent _ -> return ()+  where+    startMonitor s a = do+        hmp <- readTVarIO' (s^.hostmap)+        case find ((a ==) . view hostAddr) (Map.keys hmp) of+            Just h -> tryRunJob_ (s^.jobs) a $ runClient s $ do+                monitor (Ms 3000) (Ms 60000) h+                prepareAllQueries h+            Nothing -> return ()++-----------------------------------------------------------------------------+-- Utilities++-- | Get the 'Result' out of a 'HostResponse'. If the response is an 'RsError',+-- a 'ResponseError' is thrown. If the response is neither+-- 'RsResult' nor 'RsError', an 'UnexpectedResponse' is thrown.+getResult :: MonadThrow m => HostResponse k a b -> m (Result k a b)+getResult (HostResponse _ (RsResult _ _ r)) = return r+getResult (HostResponse h (RsError  t w e)) = throwM (ResponseError h t w e)+getResult hr                                = unexpected hr+{-# INLINE getResult #-}++getPreparedQueryId :: MonadThrow m => HostResponse k a b -> m (Host, QueryId k a b)+getPreparedQueryId hr = getResult hr >>= \case+    PreparedResult i _ _ -> return (hrHost hr, i)+    _                    -> unexpected hr+{-# INLINE getPreparedQueryId #-}++unexpected :: MonadThrow m => HostResponse k a b -> m c+unexpected (HostResponse h r) = throwM $ UnexpectedResponse h r++atomically' :: STM a -> Client a+atomically' = liftIO . atomically++readTVarIO' :: TVar a -> Client a+readTVarIO' = liftIO . readTVarIO++logInfo' :: Builder -> Client ()+logInfo' m = do+    l <- view (context.settings.logger)+    liftIO $ logInfo l m+{-# INLINE logInfo' #-}++logWarn' :: Builder -> Client ()+logWarn' m = do+    l <- view (context.settings.logger)+    liftIO $ logWarn l m+{-# INLINE logWarn' #-}++logError' :: Builder -> Client ()+logError' m = do+    l <- view (context.settings.logger)+    liftIO $ logError l m+{-# INLINE logError' #-}+

@@ -0,0 +1,33 @@+-- 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 OverloadedStrings #-}+{-# LANGUAGE TemplateHaskell   #-}+{-# LANGUAGE TypeFamilies      #-}++module Database.CQL.IO.Cluster.Discovery where++import Data.Functor.Identity (Identity)+import Data.IP+import Data.Text (Text)+import Database.CQL.Protocol++data Peer = Peer+    { peerAddr :: !IP+    , peerRPC  :: !IP+        -- ^ The address for the client to connect to.+    , peerDC   :: !Text+    , peerRack :: !Text+    } deriving Show++recordInstance ''Peer++peers :: QueryString R () (IP, IP, Text, Text)+peers = "SELECT peer, rpc_address, data_center, rack FROM system.peers"++peer :: QueryString R (Identity IP) (IP, IP, Text, Text)+peer = "SELECT peer, rpc_address, data_center, rack FROM system.peers where peer = ?"++local :: QueryString R () (Text, Text)+local = "SELECT data_center, rack FROM system.local WHERE key='local'"

@@ -0,0 +1,102 @@+-- 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 CPP               #-}+{-# LANGUAGE OverloadedStrings #-}++module Database.CQL.IO.Cluster.Host where++import Control.Lens (Lens')+import Database.CQL.Protocol (Response (..))+import Database.CQL.IO.Cluster.Discovery+import Data.IP+import Data.Text (Text, unpack)+import Network.Socket (SockAddr (..), PortNumber)++-- | A Cassandra host known to the client.+data Host = Host+    { _hostAddr   :: !InetAddr+    , _dataCentre :: !Text+    , _rack       :: !Text+    }++instance Eq Host where+    a == b = _hostAddr a == _hostAddr b++instance Ord Host where+    compare a b = compare (_hostAddr a) (_hostAddr b)++peer2Host :: PortNumber -> Peer -> Host+peer2Host i p = Host (ip2inet i (peerRPC p)) (peerDC p) (peerRack p)++updateHost :: Host -> Maybe (Text, Text) -> Host+updateHost h (Just (dc, rk)) = h { _dataCentre = dc, _rack = rk }+updateHost h Nothing         = h++-- | A response that is known to originate from a specific 'Host'.+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+    = HostNew  !Host     -- ^ a new host has been added to the cluster+    | HostGone !InetAddr -- ^ a host has been removed from the cluster+    | HostUp   !InetAddr -- ^ a host has been started+    | HostDown !InetAddr -- ^ a host has been stopped++-- | The IP address and port number of a host.+hostAddr :: Lens' Host InetAddr+hostAddr f ~(Host a c r) = fmap (\x -> Host x c r) (f a)+{-# INLINE hostAddr #-}++-- | The data centre name (may be an empty string).+dataCentre :: Lens' Host Text+dataCentre f ~(Host a c r) = fmap (\x -> Host a x r) (f c)+{-# INLINE dataCentre #-}++-- | The rack name (may be an empty string).+rack :: Lens' Host Text+rack f ~(Host a c r) = fmap (\x -> Host a c x) (f r)+{-# INLINE rack #-}++instance Show Host where+    show h = showString (unpack (_dataCentre h))+           . showString ":"+           . showString (unpack (_rack h))+           . showString ":"+           . shows (_hostAddr h)+           $ ""++-----------------------------------------------------------------------------+-- 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++-- | Map a 'SockAddr' into an 'InetAddr', using the given port number.+sock2inet :: PortNumber -> SockAddr -> InetAddr+sock2inet i (SockAddrInet _ a)      = InetAddr (SockAddrInet i a)+sock2inet i (SockAddrInet6 _ f a b) = InetAddr (SockAddrInet6 i f a b)+sock2inet _ unix                    = InetAddr unix++-- | Map an 'IP' into an 'InetAddr', using the given port number.+ip2inet :: PortNumber -> IP -> InetAddr+ip2inet p (IPv4 a) = InetAddr $ SockAddrInet p (toHostAddress a)+ip2inet p (IPv6 a) = InetAddr $ SockAddrInet6 p 0 (toHostAddress6 a) 0+

@@ -0,0 +1,152 @@+-- 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 TemplateHaskell #-}++module Database.CQL.IO.Cluster.Policies+    ( Policy (..)+    , random+    , roundRobin+    ) where++import Control.Applicative+import Control.Concurrent.STM+import Control.Lens ((^.), view, over, makeLenses)+import Control.Monad+import Data.Map.Strict (Map)+import Data.Word+import Database.CQL.IO.Cluster.Host+import System.Random.MWC+import Prelude++import qualified Data.Map.Strict as Map++-- | A policy defines a load-balancing strategy and generally+-- handles host visibility.+data Policy = Policy+    { setup :: [Host] -> [Host] -> IO ()+      -- ^ Initialise the policy with two sets of hosts. The first+      -- parameter are hosts known to be available, the second are other+      -- nodes. Note that a policy may be re-initialised at any point+      -- through this function.+    , onEvent :: HostEvent -> IO ()+      -- ^ Event handler. Policies will be informed about cluster changes+      -- through this function.+    , select :: IO (Maybe Host)+      -- ^ Host selection. The driver will ask for a host to use in a query+      -- through this function. A policy which has no available nodes may+      -- return Nothing.+    , current :: IO [Host]+      -- ^ Return all currently alive hosts.+    , acceptable :: Host -> IO Bool+      -- ^ During startup and node discovery, the driver will ask the+      -- policy if a dicovered host should be ignored.+    , hostCount :: IO Word+      -- ^ During query processing, the driver will ask the policy for+      -- a rough estimate of alive hosts. The number is used to repeatedly+      -- invoke 'select' (with the underlying assumption that the policy+      -- returns mostly different hosts).+    , display :: IO String+      -- ^ Like having an effectful 'Show' instance for this policy.+    }++type HostMap = TVar Hosts++data Hosts = Hosts+    { _alive :: !(Map InetAddr Host)+    , _other :: !(Map InetAddr Host)+    } deriving Show++makeLenses ''Hosts++-- | Iterate over hosts one by one.+roundRobin :: IO Policy+roundRobin = do+    h <- newTVarIO emptyHosts+    c <- newTVarIO 0+    return $ Policy (defSetup h) (defOnEvent h) (pickHost h c)+                    (defCurrent h) defAcceptable (defHostCount h)+                    (defDisplay h)+  where+    pickHost h c = atomically $ do+        m <- view alive <$> readTVar h+        if Map.null m then+            return Nothing+        else do+            k <- readTVar c+            writeTVar c $ succ k `mod` Map.size m+            return . Just . snd $ Map.elemAt (k `mod` Map.size m) m++-- | Return hosts in random order.+random :: IO Policy+random = do+    h <- newTVarIO emptyHosts+    g <- createSystemRandom+    return $ Policy (defSetup h) (defOnEvent h) (pickHost h g)+                    (defCurrent h) defAcceptable (defHostCount h)+                    (defDisplay h)+  where+    pickHost h g = do+        m <- view alive <$> readTVarIO h+        if Map.null m then+            return Nothing+        else do+            let i = uniformR (0, Map.size m - 1) g+            Just . snd . flip Map.elemAt m <$> i++-----------------------------------------------------------------------------+-- Defaults++emptyHosts :: Hosts+emptyHosts = Hosts Map.empty Map.empty++defDisplay :: HostMap -> IO String+defDisplay h = show <$> readTVarIO h++defAcceptable :: Host -> IO Bool+defAcceptable = const $ return True++defSetup :: HostMap -> [Host] -> [Host] -> IO ()+defSetup r a b = do+    let ha = Map.fromList $ zip (map (view hostAddr) a) a+    let hb = Map.fromList $ zip (map (view hostAddr) b) b+    let hosts = Hosts ha hb+    atomically $ writeTVar r hosts++defHostCount :: HostMap -> IO Word+defHostCount r = fromIntegral . Map.size . view alive <$> readTVarIO r++defCurrent :: HostMap -> IO [Host]+defCurrent r = Map.elems . view alive <$> readTVarIO r++defOnEvent :: HostMap -> HostEvent -> IO ()+defOnEvent r (HostNew h) = atomically $ do+    m <- readTVar r+    when (Nothing == get (h^.hostAddr) m) $+        writeTVar r (over alive (Map.insert (h^.hostAddr) h) m)+defOnEvent r (HostGone a) = atomically $ do+    m <- readTVar r+    if Map.member a (m^.alive) then+        writeTVar r (over alive (Map.delete a) m)+    else+        writeTVar r (over other (Map.delete a) m)+defOnEvent r (HostUp a) = atomically $ do+    m <- readTVar r+    case get a m of+        Nothing -> return ()+        Just  h -> writeTVar r+            $ over alive (Map.insert a h)+            . over other (Map.delete a)+            $ m+defOnEvent r (HostDown a) = atomically $ do+    m <- readTVar r+    case get a m of+        Nothing -> return ()+        Just  h -> writeTVar r+            $ over other (Map.insert a h)+            . over alive (Map.delete a)+            $ m++get :: InetAddr -> Hosts -> Maybe Host+get a m = Map.lookup a (m^.alive) <|> Map.lookup a (m^.other)

@@ -0,0 +1,365 @@+-- 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 LambdaCase        #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE TemplateHaskell   #-}+{-# LANGUAGE ViewPatterns      #-}++module Database.CQL.IO.Connection+    ( Connection+    , ConnId+    , ident+    , host++    -- * Lifecycle+    , connect+    , canConnect+    , close++    -- * Requests+    , request+    , Raw+    , requestRaw++    -- ** Queries+    , query+    , defQueryParams++    -- ** Events+    , EventHandler+    , allEventTypes+    , register++    -- * Re-exports+    , Socket.resolve+    ) where++import Control.Concurrent (myThreadId, forkIOWithUnmask)+import Control.Concurrent.Async+import Control.Concurrent.MVar+import Control.Concurrent.STM+import Control.Exception (throwTo)+import Control.Lens ((^.), makeLenses, view, set)+import Control.Monad+import Control.Monad.Catch+import Control.Monad.IO.Class+import Data.ByteString.Builder+import Data.Foldable (for_)+import Data.Semigroup ((<>))+import Data.Text.Lazy (fromStrict)+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.Exception+import Database.CQL.IO.Log+import Database.CQL.IO.Protocol+import Database.CQL.IO.Signal (Signal, signal, (|->), emit)+import Database.CQL.IO.Sync (Sync)+import Database.CQL.IO.Timeouts (TimeoutManager, withTimeout)++import qualified Data.HashMap.Strict               as HashMap+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++-- | The streams of a connection are a vector of slots, each+-- containing the last received CQL protocol frame on that stream.+type Streams = Vector (Sync Frame)++-- | A connection to a 'Host' in a Cassandra cluster.+data Connection = Connection+    { _settings :: !ConnectionSettings+    , _host     :: !Host+    , _tmanager :: !TimeoutManager+    , _protocol :: !Version+    , _sock     :: !Socket+    , _status   :: !(TVar Bool)+    , _streams  :: !Streams+    , _wLock    :: !(MVar ())+    , _reader   :: !(Async ())+    , _tickets  :: !Tickets.Pool+    , _logger   :: !Logger+    , _eventSig :: !(Signal Event)+    , _ident    :: !ConnId+    }++makeLenses ''Connection++instance Eq Connection where+    a == b = a^.ident == b^.ident++instance Show Connection where+    show c = shows (c^.host) . showString "#" . shows (c^.sock) $ ""++------------------------------------------------------------------------------+-- Lifecycle++-- | Establish and initialise a new connection to a Cassandra host.+connect :: MonadIO m+    => ConnectionSettings+    -> TimeoutManager+    -> Version+    -> Logger+    -> Host+    -> m Connection+connect t m v g h = liftIO $ do+    c <- bracketOnError sockOpen Socket.close $ \s -> do+        tck <- Tickets.pool (t^.maxStreams)+        syn <- Vector.replicateM (t^.maxStreams) Sync.create+        lck <- newMVar ()+        sta <- newTVarIO True+        sig <- signal+        rdr <- async (readLoop v g t tck h s syn sig sta lck)+        Connection t h m v s sta syn lck rdr tck g sig . ConnId <$> newUnique+    initialise c+    return c+  where+    sockOpen = Socket.open (t^.connectTimeout) (h^.hostAddr) (t^.tlsContext)++    initialise c = do+        validateSettings c+        startup c+        for_ (t^.defKeyspace) $+            useKeyspace c+      `onException`+        close c++    validateSettings c = do+        Supported ca _ <- supportedOptions c+        let x = algorithm (c^.settings.compression)+        unless (x == None || x `elem` ca) $+            throwM $ UnsupportedCompression x ca++    supportedOptions c = do+        let req = RqOptions Options+        let c' = set (settings.compression) noCompression c+        requestRaw c' req >>= \case+            RsSupported _ _ x -> return x+            rs                -> unhandled c rs++-- | Check the connectivity of a Cassandra host on a new connection.+canConnect :: MonadIO m => Host -> m Bool+canConnect h = liftIO $ reachable `recover` False+  where+    reachable = bracket (Socket.open (Ms 5000) (h^.hostAddr) Nothing)+                        Socket.close+                        (const (return True))++-- Note: The socket is closed when the 'readLoop' exits.+close :: Connection -> IO ()+close = cancel . view reader++------------------------------------------------------------------------------+-- Low-level operations++type Raw a = a () () ()++request :: (Tuple a, Tuple b) => Connection -> Request k a b -> IO (Response k a b)+request c rq = send >>= receive+  where+    send = withTimeout (c^.tmanager) (c^.settings.sendTimeout) (close c) $ do+        i <- Tickets.toInt <$> Tickets.get (c^.tickets)+        req <- serialise (c^.protocol) (c^.settings.compression) rq i+        logRequest (c^.logger) req+        withMVar (c^.wLock) $ const $ do+            isOpen <- readTVarIO (c^.status)+            if isOpen then+                Socket.send (c^.sock) req+            else+                throwM $ ConnectionClosed (c^.host.hostAddr)+        return i++    receive i = do+        let rt = ResponseTimeout (c^.host.hostAddr)+        tid <- myThreadId+        r <- withTimeout (c^.tmanager) (c^.settings.responseTimeout) (throwTo tid rt) $ do+            r <- Sync.get (view streams c ! i)+                `onException` Sync.kill rt (view streams c ! i)+            Tickets.markAvailable (c^.tickets) i+            return r+        parse (c^.settings.compression) r++requestRaw :: Connection -> Raw Request -> IO (Raw Response)+requestRaw = request++-----------------------------------------------------------------------------+-- High-level operations++startup :: MonadIO m => Connection -> m ()+startup c = liftIO $ do+    let cmp = c^.settings.compression+    let req = RqStartup (Startup Cqlv300 (algorithm cmp))+    requestRaw c req >>= \case+        RsReady _ _ Ready       -> checkAuth c+        RsAuthenticate _ _ auth -> authenticate c auth+        rs                      -> unhandled c rs++checkAuth :: Connection -> IO ()+checkAuth c = unless (null (c^.settings.authenticators)) $+    logWarn' (c^.logger) (c^.host) $+        "Authentication configured but none required by the server."++authenticate :: Connection -> Authenticate -> IO ()+authenticate c (Authenticate (AuthMechanism -> m)) =+    case HashMap.lookup m (c^.settings.authenticators) of+        Nothing -> throwM $ AuthenticationRequired m+        Just Authenticator {+            authOnRequest   = onR+          , authOnChallenge = onC+          , authOnSuccess   = onS+        } -> do+            (rs, s) <- onR context+            case onC of+                Just  f -> loop f onS (rs, s)+                Nothing -> authResponse c rs >>= either+                    (throwM . UnexpectedAuthenticationChallenge m)+                    (onS s)+  where+    context = AuthContext (c^.ident) (c^.host.hostAddr)++    loop onC onS (rs, s) =+        authResponse c rs >>= either+            (onC s >=> loop onC onS)+            (onS s)++authResponse :: Connection -> AuthResponse -> IO (Either AuthChallenge AuthSuccess)+authResponse c resp = liftIO $ do+    let req = RqAuthResp resp+    requestRaw c req >>= \case+        RsAuthSuccess _ _ success -> return $ Right success+        RsAuthChallenge _ _ chall -> return $ Left chall+        rs                        -> unhandled c rs++useKeyspace :: MonadIO m => Connection -> Keyspace -> m ()+useKeyspace c ks = liftIO $ do+    let params = defQueryParams One ()+        kspace = quoted (fromStrict $ unKeyspace ks)+        req    = RqQuery (Query (QueryString $ "use " <> kspace) params)+    requestRaw c req >>= \case+        RsResult _ _ (SetKeyspaceResult _) -> return ()+        rs                                 -> unhandled c rs++------------------------------------------------------------------------------+-- Queries++query :: (Tuple a, Tuple b, MonadIO m)+      => Connection+      -> Consistency+      -> QueryString k a b+      -> a+      -> m [b]+query c cons q p = liftIO $ do+    let req = RqQuery (Query q (defQueryParams cons p))+    request c req >>= \case+        RsResult _ _ (RowsResult _ b) -> return b+        rs                            -> unhandled c rs++-- | Construct default 'QueryParams' for the given consistency+-- and bound values. In particular, no page size, paging state+-- or serial consistency will be set.+defQueryParams :: Consistency -> a -> QueryParams a+defQueryParams c a = QueryParams+    { consistency       = c+    , values            = a+    , skipMetaData      = False+    , pageSize          = Nothing+    , queryPagingState  = Nothing+    , serialConsistency = Nothing+    , enableTracing     = Nothing+    }++------------------------------------------------------------------------------+-- Events++type EventHandler = Event -> IO ()++allEventTypes :: [EventType]+allEventTypes = [TopologyChangeEvent, StatusChangeEvent, SchemaChangeEvent]++register :: MonadIO m => Connection -> [EventType] -> EventHandler -> m ()+register c ev f = liftIO $ do+    let req = RqRegister (Register ev)+    requestRaw c req >>= \case+        RsReady _ _ Ready -> c^.eventSig |-> f+        rs                -> unhandled c rs++------------------------------------------------------------------------------+-- Read loop++-- Note: The read loop owns the socket given and is responsible+-- for closing it, when it gets interrupted.+readLoop :: Version+         -> Logger+         -> ConnectionSettings+         -> Tickets.Pool+         -> Host+         -> Socket+         -> Streams+         -> Signal Event+         -> TVar Bool+         -> MVar ()+         -> IO ()+readLoop v g cset tck h sck syn sig sref wlck =+    run `catch` logException `finally` cleanup+  where+    run = forever $ do+        f@(Frame hd _) <- readFrame v g h sck (cset^.maxRecvBuffer)+        case fromStreamId (streamId hd) of+            -1 -> do+                r <- parse (cset^.compression) f :: IO (Raw Response)+                case r of+                    RsEvent _ _ e -> emit sig e+                    _             -> throwM (UnexpectedResponse h r)+            sid -> do+                ok <- Sync.put f (syn ! sid)+                unless ok $+                    Tickets.markAvailable tck sid++    cleanup = uninterruptibleMask_ $ do+        isOpen <- atomically $ swapTVar sref False+        when isOpen $ do+            let ex = ConnectionClosed (h^.hostAddr)+            Tickets.close ex tck+            Vector.mapM_ (Sync.close ex) syn+            -- Try to shut down the socket gracefully, now allowing+            -- interruptions (i.e. all exceptions) but make sure+            -- the socket gets closed eventually.+            void $ forkIOWithUnmask $ \unmask -> unmask (do+                Socket.shutdown sck Socket.ShutdownReceive+                withMVar wlck (const $ Socket.close sck)+              ) `onException` Socket.close sck++    logException e = case fromException e of+        Just AsyncCancelled -> return ()+        _                   -> logWarn' g h ("read-loop: " <> string8 (show e))++readFrame :: Version -> Logger -> Host -> Socket -> Int -> IO Frame+readFrame v g h s n = do+    b <- Socket.recv n (h^.hostAddr) s 9+    case header v b of+       Left    e -> throwM $ ParseError ("response header reading: " ++ e)+       Right hdr -> case headerType hdr of+           RqHeader -> throwM $ ParseError "unexpected header"+           RsHeader -> do+               let len = lengthRepr (bodyLength hdr)+               dat <- Socket.recv n (h^.hostAddr) s (fromIntegral len)+               logResponse g (b <> dat)+               return $ Frame hdr dat++unhandled :: Connection -> Response k a b -> IO c+unhandled c r = case r of+    RsError t w e -> throwM (ResponseError (c^.host) t w e)+    rs            -> unexpected c rs++unexpected :: Connection -> Response k a b -> IO c+unexpected c r = throwM $ UnexpectedResponse (c^.host) r++logWarn' :: Logger -> Host -> Builder -> IO ()+logWarn' l h m = logWarn l $ string8 (show h) <> string8 ": " <> m+

@@ -0,0 +1,163 @@+-- 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 ExistentialQuantification  #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE OverloadedStrings          #-}+{-# LANGUAGE ScopedTypeVariables        #-}+{-# LANGUAGE TemplateHaskell            #-}++module Database.CQL.IO.Connection.Settings+    ( ConnectionSettings+    , ConnId (..)+    , defSettings+    , defKeyspace+    , compression+    , tlsContext++    -- * Timeouts+    , Milliseconds (..)+    , connectTimeout+    , sendTimeout+    , responseTimeout++    -- * Limits+    , maxStreams+    , maxRecvBuffer++    -- * Authentication+    , authenticators+    , AuthMechanism (..)+    , Authenticator (..)+    , AuthContext   (..)+    , authConnId+    , authHost+    , passwordAuthenticator+    , AuthUser (..)+    , AuthPass (..)+    ) where++import Control.Lens (makeLenses)+import Data.Hashable+import Data.HashMap.Strict (HashMap)+import Data.String+import Data.Text (Text)+import Data.Unique+import Database.CQL.Protocol+import Database.CQL.IO.Cluster.Host+import OpenSSL.Session (SSLContext)++import qualified Data.ByteString.Lazy.Char8 as Char8+import qualified Data.HashMap.Strict        as HashMap+import qualified Data.Text.Lazy             as Lazy+import qualified Data.Text.Lazy.Encoding    as Lazy++newtype Milliseconds = Ms { ms :: Int }+    deriving (Eq, Show)++newtype ConnId = ConnId Unique deriving (Eq, Ord)++instance Hashable ConnId where+    hashWithSalt _ (ConnId u) = hashUnique u++data ConnectionSettings = ConnectionSettings+    { _connectTimeout  :: !Milliseconds+    , _sendTimeout     :: !Milliseconds+    , _responseTimeout :: !Milliseconds+    , _maxStreams      :: !Int+    , _compression     :: !Compression+    , _defKeyspace     :: !(Maybe Keyspace)+    , _maxRecvBuffer   :: !Int+    , _tlsContext      :: !(Maybe SSLContext)+    , _authenticators  :: !(HashMap AuthMechanism Authenticator)+    }++-- | Context information given to 'Authenticator's when+-- the server requests authentication on a connection.+-- See 'authOnRequest'.+data AuthContext = AuthContext+    { _authConnId :: !ConnId+    , _authHost   :: !InetAddr+    }++-- | The (unique) name of a SASL authentication mechanism.+--+-- In the case of Cassandra, this is currently always the fully-qualified+-- Java class name of the configured server-side @IAuthenticator@+-- implementation.+newtype AuthMechanism = AuthMechanism Text+    deriving (Eq, Ord, Show, IsString, Hashable)++-- | A client authentication handler.+--+-- The fields of an 'Authenticator' must implement the client-side+-- of an (SASL) authentication mechanism as follows:+--+--    * When a Cassandra server requests authentication on a new connection,+--      'authOnRequest' is called with the 'AuthContext' of the+--      connection.+--+--    * If additional challenges are posed by the server,+--      'authOnChallenge' is called, if available, otherwise an+--      'AuthenticationError' is thrown, i.e. every challenge must be+--      answered.+--+--    * Upon successful authentication 'authOnSuccess' is called.+--+-- The existential type @s@ is chosen by an implementation and can+-- be used to thread arbitrary state through the sequence of callback+-- invocations during an authentication exchange.+--+-- See also:+-- <https://tools.ietf.org/html/rfc4422 RFC4422>+-- <https://docs.datastax.com/en/cassandra/latest/cassandra/configuration/secureInternalAuthenticationTOC.html Authentication>+data Authenticator = forall s. Authenticator+    { authMechanism :: !AuthMechanism+        -- ^ The (unique) name of the (SASL) mechanism that the callbacks+        -- implement.+    , authOnRequest :: AuthContext -> IO (AuthResponse, s)+        -- ^ Callback for initiating an authentication exchange.+    , authOnChallenge :: Maybe (s -> AuthChallenge -> IO (AuthResponse, s))+        -- ^ Optional callback for additional challenges posed by the server.+        -- If the authentication mechanism does not require additional+        -- challenges, it should be set to 'Nothing'. Otherwise every+        -- challenge must be answered with a response.+    , authOnSuccess :: s -> AuthSuccess -> IO ()+        -- ^ Callback for successful completion of an authentication exchange.+    }++makeLenses ''AuthContext+makeLenses ''ConnectionSettings++newtype AuthUser = AuthUser Lazy.Text+newtype AuthPass = AuthPass Lazy.Text++-- | A password authentication handler for use with Cassandra's+-- @PasswordAuthenticator@.+--+-- See: <https://docs.datastax.com/en/cassandra/latest/cassandra/configuration/secureConfigNativeAuth.html Configuring Authentication>+passwordAuthenticator :: AuthUser -> AuthPass -> Authenticator+passwordAuthenticator (AuthUser u) (AuthPass p) = Authenticator+    { authMechanism   = "org.apache.cassandra.auth.PasswordAuthenticator"+    , authOnChallenge = Nothing+    , authOnSuccess   = \() _ -> return ()+    , authOnRequest   = \_ctx ->+        let user = Lazy.encodeUtf8 u+            pass = Lazy.encodeUtf8 p+            resp = AuthResponse (Char8.concat ["\0", user, "\0", pass])+        in return (resp, ())+    }++defSettings :: ConnectionSettings+defSettings =+    ConnectionSettings (Ms 5000)     -- connect timeout+                       (Ms 3000)     -- send timeout+                       (Ms 10000)    -- response timeout+                       128           -- max streams per connection+                       noCompression -- compression+                       Nothing       -- keyspace+                       16384         -- receive buffer size+                       Nothing       -- no tls by default+                       HashMap.empty -- no authentication+

@@ -0,0 +1,116 @@+-- 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 #-}++-- | A thin wrapper of the Network.Socket API.+module Database.CQL.IO.Connection.Socket+    ( Socket+    , resolve+    , open+    , send+    , recv+    , close+    , shutdown++    -- Re-exports+    , HostName+    , PortNumber+    , ShutdownCmd (..)+    ) 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.Cluster.Host+import Database.CQL.IO.Exception (ConnectionError (..))+import Database.CQL.IO.Timeouts (Milliseconds (..))+import Foreign.C.Types (CInt (..))+import Network.Socket (HostName, PortNumber, SockAddr (..), ShutdownCmd (..))+import Network.Socket (Family (..), AddrInfo (..), AddrInfoFlag (..))+import Network.Socket.ByteString.Lazy (sendAll)+import OpenSSL.Session (SSL, SSLContext)+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 Show Socket where+    show s = show $ case s of+        Stream x -> fd x+        Tls  x _ -> fd x+      where+        fd x = let CInt n = S.fdSocket x in n++resolve :: HostName -> PortNumber -> IO [InetAddr]+resolve h p = do+    ais <- S.getAddrInfo (Just hints) (Just h) (Just (show p))+    return $ map (InetAddr . addrAddress) ais+  where+    hints = S.defaultHints { addrFlags = [AI_ADDRCONFIG], addrSocketType = S.Stream }++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 S.defaultProtocol+  where+    familyOf (SockAddrInet  _ _)     = AF_INET+    familyOf (SockAddrInet6 _ _ _ _) = AF_INET6+    familyOf (SockAddrUnix  _)       = AF_UNIX+#if MIN_VERSION_network(2,6,1) && !MIN_VERSION_network(3,0,0)+    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)++

@@ -0,0 +1,247 @@+-- 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 GADTs                      #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE OverloadedStrings          #-}+{-# LANGUAGE StandaloneDeriving         #-}+{-# LANGUAGE ScopedTypeVariables        #-}++module Database.CQL.IO.Exception where++import Control.Exception (SomeAsyncException (..))+import Control.Monad.Catch+import Data.List.NonEmpty (NonEmpty ((:|)))+import Data.Text (Text)+import Data.Typeable+import Data.UUID+import Database.CQL.IO.Cluster.Host+import Database.CQL.IO.Connection.Settings+import Database.CQL.Protocol++import qualified Data.List.NonEmpty as NonEmpty+import qualified Data.Text.Lazy as Lazy++-----------------------------------------------------------------------------+-- ResponseError++-- | The server responded with an 'Error'.+--+-- Most of these errors are either not retryable or only safe to retry+-- for idempotent queries. For more details of which errors may be safely+-- retried under which circumstances, see also the documentation of the+-- <https://docs.datastax.com/en/developer/java-driver/latest/manual/retries/ Java driver>.+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++-- | An error during host selection prior to query execution.+--+-- These errors are always safe to retry but may indicate an overload+-- situation and thus suggest a review of the client and cluster+-- configuration (number of hosts, pool sizes, connections per host,+-- streams per connection, ...).+data HostError+    = NoHostAvailable+        -- ^ There is currently not a single host available to the+        -- client according to the configured 'Policy'.+    | HostsBusy+        -- ^ All streams on all connections are currently in use.+    deriving Typeable++instance Exception HostError++instance Show HostError where+    show NoHostAvailable = "cql-io: no host available"+    show HostsBusy       = "cql-io: hosts busy"++-----------------------------------------------------------------------------+-- ConnectionError++-- | An error while establishing or using a connection to send a+-- request or receive a response.+data ConnectionError+    = ConnectionClosed !InetAddr+        -- ^ The connection was suddenly closed.+        -- Retries are only safe for idempotent queries.+    | ConnectTimeout   !InetAddr+        -- ^ A timeout occurred while establishing a connection.+        -- See also 'setConnectTimeout'. Retries are always safe.+    | ResponseTimeout  !InetAddr+        -- ^ A timeout occurred while waiting for a response.+        -- See also 'setResponseTimeout'. Retries are only+        -- safe for idempotent queries.+    deriving Typeable++instance Exception ConnectionError++instance Show ConnectionError where+    show (ConnectionClosed i) = "cql-io: connection closed: " ++ show i+    show (ConnectTimeout   i) = "cql-io: connect timeout: " ++ show i+    show (ResponseTimeout  i) = "cql-io: response timeout: " ++ show i++-----------------------------------------------------------------------------+-- ProtocolError++-- | A protocol error indicates a problem related to the client-server+-- communication protocol. The cause may either be misconfiguration+-- on the client or server, or an implementation bug. In the latter case+-- it should be reported. In either case these errors are not recoverable+-- and should never be retried.+data ProtocolError where+    -- | The client received an unexpected response for a request.+    -- This indicates a problem with the communication protocol+    -- and should be reported.+    UnexpectedResponse :: Host -> Response k a b -> ProtocolError+    -- | The client received an unexpected query ID in an 'Unprepared'+    -- server response upon executing a prepared query. This indicates+    -- a problem with the communication protocol and should be reported.+    UnexpectedQueryId :: QueryId k a b -> ProtocolError+    -- | The client tried to use a compression algorithm that+    -- is not supported by the server. The first argument is the offending+    -- algorithm and the second argument the list of supported algorithms+    -- as reported by the server. This indicates a client or server-side+    -- configuration error.+    UnsupportedCompression :: CompressionAlgorithm -> [CompressionAlgorithm] -> ProtocolError+    -- | An error occurred during the serialisation of a request.+    -- This indicates a problem with the wire protocol and should+    -- be reported.+    SerialiseError :: String -> ProtocolError+    -- | An error occurred during parsing of a response. This indicates+    -- a problem with the wire protocol and should be reported.+    ParseError :: String -> ProtocolError++deriving instance Typeable ProtocolError+instance Exception ProtocolError++instance Show ProtocolError where+    show e = showString "cql-io: protocol error: " . case e of+        ParseError x ->+            showString "parse error: " . showString x+        SerialiseError x ->+            showString "serialise error: " . showString x+        UnsupportedCompression x cc ->+            showString "unsupported compression: " . shows x .+            showString ", expected one of " . shows cc+        UnexpectedQueryId i ->+            showString "unexpected query ID: " . shows i+        UnexpectedResponse h r -> showString "unexpected response: " .+            shows h . showString ": " . shows (f r)+        $ ""+      where+        f :: Response k a b -> Response k a NoShow+        f (RsError         a b c) = RsError a b c+        f (RsReady         a b c) = RsReady a b c+        f (RsAuthenticate  a b c) = RsAuthenticate a b c+        f (RsAuthChallenge a b c) = RsAuthChallenge a b c+        f (RsAuthSuccess   a b c) = RsAuthSuccess a b c+        f (RsSupported     a b c) = RsSupported a b c+        f (RsResult        a b c) = RsResult a b (g c)+        f (RsEvent         a b c) = RsEvent a b c++        g :: Result k a b -> Result k a NoShow+        g VoidResult                       = VoidResult+        g (RowsResult              a  b  ) = RowsResult a (map (const NoShow) b)+        g (SetKeyspaceResult       a     ) = SetKeyspaceResult a+        g (SchemaChangeResult      a     ) = SchemaChangeResult a+        g (PreparedResult (QueryId a) b c) = PreparedResult (QueryId a) b c++-- | Placeholder for parts of a 'Response' that are not 'Show'able.+data NoShow = NoShow deriving Show++-----------------------------------------------------------------------------+-- HashCollision++-- | An unexpected hash collision occurred for a prepared query string.+-- This indicates a problem with the implementation of prepared queries+-- and should be reported.+data HashCollision = HashCollision !Lazy.Text !Lazy.Text+    deriving Typeable++instance Exception HashCollision++instance Show HashCollision where+    show (HashCollision a b) = showString "cql-io: hash collision: "+                             . shows a+                             . showString " "+                             . shows b+                             $ ""++-----------------------------------------------------------------------------+-- AuthenticationError++-- | An error occurred during the authentication phase while+-- initialising a new connection. This indicates a configuration+-- error or a faulty 'Authenticator'.+data AuthenticationError+    = AuthenticationRequired !AuthMechanism+        -- ^ The server demanded authentication but none was provided+        -- by the client.+    | UnexpectedAuthenticationChallenge !AuthMechanism !AuthChallenge+        -- ^ The server presented an additional authentication challenge+        -- that the configured 'Authenticator' did not respond to.++instance Exception AuthenticationError++instance Show AuthenticationError where+    show (AuthenticationRequired a)+        = showString "cql-io: authentication required: "+        . shows a+        $ ""++    show (UnexpectedAuthenticationChallenge n c)+        = showString "cql-io: unexpected authentication challenge: '"+        . shows c+        . showString "' using mechanism '"+        . shows n+        . showString "'"+        $ ""++-----------------------------------------------------------------------------+-- Utilities++-- | Recover from all (synchronous) exceptions raised by a+-- computation with a fixed value.+recover :: forall m a. MonadCatch m => m a -> a -> m a+recover io val = try io >>= either fallback return+  where+    fallback :: SomeException -> m a+    fallback e = case fromException e of+        Just (SomeAsyncException _) -> throwM e+        Nothing                     -> return val+{-# INLINE recover #-}++-- | Ignore all (synchronous) exceptions raised by a+-- computation that produces no result, i.e. is only run for+-- its (side-)effects.+ignore :: MonadCatch m => m () -> m ()+ignore io = recover io ()+{-# INLINE ignore #-}++-- | Try a computation on a non-empty list of values, recovering+-- from (synchronous) exceptions for all but the last value.+tryAll :: forall m a b. MonadCatch m => NonEmpty a -> (a -> m b) -> m b+tryAll (a :| []) f = f a+tryAll (a :| aa) f = try (f a) >>= either next return+  where+    next :: SomeException -> m b+    next e = case fromException e of+        Just (SomeAsyncException _) -> throwM e+        Nothing                     -> tryAll (NonEmpty.fromList aa) f+

@@ -0,0 +1,72 @@+-- 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 OverloadedStrings #-}++-- | Create hexadecimal-encoded (lazy) 'ByteString's in a pretty-printed+-- format common to *nix hexdump programs. For example:+--+-- >>> import Database.CQL.IO.Hexdump+-- >>> import qualified Data.ByteString.Lazy.Char8 as Char8+-- >>> Char8.putStrLn $ hexdump "GET /foo/bar?x=y HTTP/1.1\r\nHost: foo.com\r\n\r\n"+-- 0001:  47 45 54 20  2f 66 6f 6f  2f 62 61 72  3f 78 3d 79   GET /foo/bar?x=y+-- 0002:  20 48 54 54  50 2f 31 2e  31 0d 0a 48  6f 73 74 3a    HTTP/1.1..Host:+-- 0003:  20 66 6f 6f  2e 63 6f 6d  0d 0a 0d 0a                 foo.com....+module Database.CQL.IO.Hexdump (hexdump, hexdumpBuilder) where++import Data.ByteString.Builder+import Data.ByteString.Lazy (ByteString)+import Data.Int+import Data.Semigroup ((<>))+import Data.Word+import System.IO (nativeNewline, Newline (..))++import qualified Data.ByteString.Lazy as L+import qualified Data.List            as List++width, groups :: Int64+width  = 16+groups = 4++hexdump :: ByteString -> ByteString+hexdump = toLazyByteString . hexdumpBuilder++hexdumpBuilder :: ByteString -> Builder+hexdumpBuilder = mconcat+    . List.intersperse newline+    . map toLine+    . zipWith (,) [1 ..]+    . chunks width++chunks :: Int64 -> ByteString -> [ByteString]+chunks n b = List.unfoldr step b+  where+    step "" = Nothing+    step c  = Just $! L.splitAt n c++toLine :: (Word16, ByteString) -> Builder+toLine (n, b) = let k = L.length b in+       word16HexFixed n+    <> ":  "+    <> mconcat (List.intersperse space (map toGroup (chunks groups b)))+    <> spaces ((width - k) * (groups - 1) + pad k + 2)+    <> lazyByteString (toAscii b)++toGroup :: ByteString -> Builder+toGroup = L.foldr (\x y -> word8HexFixed x <> space <> y) mempty++toAscii :: ByteString -> ByteString+toAscii = L.map (\w -> if w > 0x1F && w < 0x7F then w else 0x2E)++space, newline :: Builder+space   = byteString " "+newline = case nativeNewline of+    LF   -> byteString "\n"+    CRLF -> byteString "\r\n"++spaces :: Int64 -> Builder+spaces n = lazyByteString $ L.replicate n 0x20++pad :: Int64 -> Int64+pad n = (width - n) `div` groups

@@ -0,0 +1,139 @@+-- 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/.++module Database.CQL.IO.Jobs+    ( Jobs+    , JobReplaced (..)+    , newJobs+    , runJob+    , runJob_+    , tryRunJob+    , tryRunJob_+    , cancelJobs+    , listJobs+    , listJobKeys+    ) where++import Control.Concurrent+import Control.Concurrent.Async+import Control.Exception (asyncExceptionFromException, asyncExceptionToException)+import Control.Monad+import Control.Monad.Catch+import Control.Monad.IO.Class+import Data.IORef+import Data.Map.Strict (Map)+import Data.Typeable+import Data.Unique++import qualified Data.Map.Strict as Map++-- | A registry for asynchronous computations ("jobs") associated with keys+-- of type @k@, with only a single job running at a time for a particular key.+newtype Jobs k = Jobs (IORef (Map k Job))++-- | Internal representation of a job. The 'Unique' value ensures that+-- a job that finishes or is aborted never accidentily removes a newly+-- registered job for the same key from the 'Jobs' registry.+data Job = Job+    { _jobUniq :: !Unique+    , jobAsync :: !(Async ())+    }++-- | The asynchronous exception used to cancel a job if it is replaced+-- by another job.+data JobReplaced = JobReplaced deriving (Eq, Show, Typeable)+instance Exception JobReplaced where+    toException   = asyncExceptionToException+    fromException = asyncExceptionFromException++newJobs :: MonadIO m => m (Jobs k)+newJobs = liftIO $ Jobs <$> newIORef Map.empty++-- | 'runJob' and ignore the result.+runJob_ :: (MonadIO m, Ord k) => Jobs k -> k -> IO () -> m ()+runJob_ j k = void . runJob j k++-- | Run an asynchronous job for a key. If there is a running job for the same+-- key, it is replaced and cancelled with a 'JobReplaced' exception.+runJob :: (MonadIO m, Ord k) => Jobs k -> k -> IO () -> m (Async ())+runJob j@(Jobs ref) k = runJobWith addJob j k+  where+    addJob new = atomicModifyIORef' ref $ \jobs ->+        let jobs' = Map.insert k new jobs+            old   = Map.lookup k jobs+            val   = jobAsync new+        in (jobs', (True, val, old))++-- | 'tryRunJob' and ignore the result.+tryRunJob_ :: (MonadIO m, Ord k) => Jobs k -> k -> IO () -> m ()+tryRunJob_ j k = void . tryRunJob j k++-- | Try to run an asynchronous job for a key. If there is a running job for+-- the same key, 'Nothing' is returned and the job will not run.+tryRunJob :: (MonadIO m, Ord k) => Jobs k -> k -> IO () -> m (Maybe (Async ()))+tryRunJob j@(Jobs ref) k = runJobWith addJob j k+  where+    addJob new = atomicModifyIORef' ref $ \jobs ->+        if Map.member k jobs+            then (jobs, (False, Nothing, Nothing))+            else+                let jobs' = Map.insert k new jobs+                    val   = Just (jobAsync new)+                in (jobs', (True, val, Nothing))++-- | Cancel all running jobs.+cancelJobs :: MonadIO m => Jobs k -> m ()+cancelJobs (Jobs d) = liftIO $ do+    jobs <- Map.elems <$> atomicModifyIORef' d (\m -> (Map.empty, m))+    mapM_ (cancel . jobAsync) jobs++-- | List all running jobs.+listJobs :: MonadIO m => Jobs k -> m [(k, Async ())]+listJobs (Jobs j) = liftIO $ Map.foldrWithKey f [] <$> readIORef j+  where+    f k a b = (k, jobAsync a) : b++-- | List the keys of all running jobs.+listJobKeys :: MonadIO m => Jobs k -> m [k]+listJobKeys (Jobs j) = liftIO $ Map.keys <$> readIORef j++------------------------------------------------------------------------------+-- Internal++runJobWith :: (MonadIO m, Ord k)+    => (Job -> IO (Bool, a, Maybe Job))+    -> Jobs k+    -> k+    -> IO ()+    -> m a+runJobWith addJob (Jobs ref) k io = liftIO $ do+    u <- newUnique+    l <- newEmptyMVar+    -- Once the async is created and waiting on the latch @l@, it must+    -- either be unblocked by putMVar or cancelled, hence masking+    -- between 'async' and 'run' (nb. 'takeMVar' is interruptible).+    mask $ \restore -> do+        new <- async $ do+            takeMVar l+            restore io+            remove u+          `catches`+            [ Handler $ \x@JobReplaced     -> throwM x+            , Handler $ \x@SomeException{} -> remove u >> throwM x+            ]+        restore (run u l new) `onException` cancel new+  where+    run u l new = do+        (ok, a, old) <- addJob (Job u new)+        mapM_ ((`cancelWith` JobReplaced) . jobAsync) old+        if ok then putMVar l () else cancel new+        return a++    remove u = atomicModifyIORef' ref $ \jobs ->+        let update = Map.update $ \a ->+                        case a of+                            Job u' _ | u == u' -> Nothing+                            _                  -> Just a+        in (update k jobs, ())+

@@ -0,0 +1,85 @@+{-# LANGUAGE OverloadedStrings #-}++module Database.CQL.IO.Log+    ( Logger (..)+    , LogLevel (..)+    , nullLogger+    , stdoutLogger+    , logDebug+    , logInfo+    , logWarn+    , logError+    , module BB+    ) where++import Control.Monad (when)+import Data.ByteString.Builder as BB+import Data.ByteString.Lazy (ByteString)+import Database.CQL.IO.Hexdump+import Data.Semigroup ((<>))++import qualified Data.ByteString.Lazy.Char8 as Char8++-- | A 'Logger' provides functions for logging textual messages as well as+-- binary CQL protocol requests and responses emitted by the client.+data Logger = Logger+    { logMessage  :: LogLevel -> Builder -> IO ()+    , logRequest  :: ByteString -> IO ()+    , logResponse :: ByteString -> IO ()+    }++-- | Log levels used by the client.+data LogLevel+    = LogDebug+        -- ^ Verbose debug information that should not be enabled in+        -- production environments.+    | LogInfo+        -- ^ General information concerning client and cluster state.+    | LogWarn+        -- ^ Warnings of potential problems that should be investigated.+    | LogError+        -- ^ Errors that should be investigated and monitored.+    deriving (Eq, Ord, Show, Read)++-- | A logger that discards all log messages.+nullLogger :: Logger+nullLogger = Logger+    { logMessage  = \_ _ -> return ()+    , logRequest  = \_   -> return ()+    , logResponse = \_   -> return ()+    }++-- | A logger that writes all log messages to stdout, discarding log messages+-- whose level is less than the given level. Requests and responses are+-- logged on debug level, formatted in hexadecimal blocks.+stdoutLogger :: LogLevel -> Logger+stdoutLogger l = Logger+    { logMessage  = \l' m -> when (l <= l') $+        Char8.putStrLn (withLevel l' (toLazyByteString m))+    , logRequest  = \rq   -> when (l <= LogDebug) $+        Char8.putStrLn (hexdump rq)+    , logResponse = \rs   -> when (l <= LogDebug) $+        Char8.putStrLn (hexdump rs)+    }+  where+    withLevel LogDebug m = "[Debug] " <> m+    withLevel LogInfo  m = "[Info]  " <> m+    withLevel LogWarn  m = "[Warn]  " <> m+    withLevel LogError m = "[Error] " <> m++logDebug :: Logger -> Builder -> IO ()+logDebug l = logMessage l LogDebug+{-# INLINE logDebug #-}++logInfo :: Logger -> Builder -> IO ()+logInfo l = logMessage l LogInfo+{-# INLINE logInfo #-}++logWarn :: Logger -> Builder -> IO ()+logWarn l = logMessage l LogWarn+{-# INLINE logWarn #-}++logError :: Logger -> Builder -> IO ()+logError l = logMessage l LogError+{-# INLINE logError #-}+

@@ -0,0 +1,218 @@+-- 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 MultiWayIf          #-}+{-# LANGUAGE OverloadedStrings   #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TemplateHaskell     #-}++module Database.CQL.IO.Pool+    ( Pool+    , create+    , destroy+    , purge+    , with++    , PoolSettings+    , defSettings+    , idleTimeout+    , maxConnections+    , maxTimeouts+    , poolStripes+    ) where++import Control.AutoUpdate+import Control.Concurrent+import Control.Concurrent.Async+import Control.Concurrent.STM+import Control.Exception+import Control.Lens ((^.), makeLenses, view)+import Control.Monad.IO.Class+import Control.Monad+import Data.Foldable (forM_, mapM_, find)+import Data.Function (on)+import Data.Hashable+import Data.IORef+import Data.Sequence (Seq, ViewL (..), (|>), (><))+import Data.Semigroup ((<>))+import Data.Time.Clock (UTCTime, NominalDiffTime, getCurrentTime, diffUTCTime)+import Data.Vector (Vector, (!))+import Database.CQL.IO.Connection (Connection)+import Database.CQL.IO.Exception (ConnectionError (..), ignore)+import Database.CQL.IO.Log++import qualified Data.Sequence as Seq+import qualified Data.Vector   as Vec++-----------------------------------------------------------------------------+-- API++data PoolSettings = PoolSettings+    { _idleTimeout    :: !NominalDiffTime+    , _maxConnections :: !Int+    , _maxTimeouts    :: !Int+    , _poolStripes    :: !Int+    }++data Pool = Pool+    { _createFn    :: !(IO Connection)+    , _destroyFn   :: !(Connection -> IO ())+    , _logger      :: !Logger+    , _settings    :: !PoolSettings+    , _maxRefs     :: !Int+    , _currentTime :: !(IO UTCTime)+    , _stripes     :: !(Vector Stripe)+    , _finaliser   :: !(IORef ())+    }++data Resource = Resource+    { tstamp   :: !UTCTime+    , refcnt   :: !Int+    , timeouts :: !Int+    , value    :: !Connection+    } deriving Show++data Box+    = New  !(IO Resource)+    | Used !Resource+    | Empty++data Stripe = Stripe+    { conns :: !(TVar (Seq Resource))+    , inUse :: !(TVar Int)+    }++makeLenses ''PoolSettings+makeLenses ''Pool++defSettings :: PoolSettings+defSettings = PoolSettings+    60 -- idle timeout+    2  -- max connections per stripe+    16 -- max timeouts per connection+    4  -- max stripes++create :: IO Connection -> (Connection -> IO ()) -> Logger -> PoolSettings -> Int -> IO Pool+create mk del g s k = do+    p <- Pool mk del g s k+            <$> mkAutoUpdate defaultUpdateSettings { updateAction = getCurrentTime }+            <*> Vec.replicateM (s^.poolStripes) (Stripe <$> newTVarIO Seq.empty <*> newTVarIO 0)+            <*> newIORef ()+    r <- async $ reaper p+    void $ mkWeakIORef (p^.finaliser) (cancel r >> destroy p)+    return p++destroy :: Pool -> IO ()+destroy = purge++with :: MonadIO m => Pool -> (Connection -> IO a) -> m (Maybe a)+with p f = liftIO $ do+    s <- stripe p+    mask $ \restore -> do+        r <- take1 p s+        case r of+            Just  v -> do+                x <- restore (f (value v)) `catch` cleanup p s v+                put p s v id+                return (Just x)+            Nothing -> return Nothing++purge :: Pool -> IO ()+purge p = Vec.forM_ (p^.stripes) $ \s -> do+    cs <- atomically (swapTVar (conns s) Seq.empty)+    mapM_ (ignore . view destroyFn p . value) cs++-----------------------------------------------------------------------------+-- Internal++cleanup :: Pool -> Stripe -> Resource -> SomeException -> IO a+cleanup p s r x = do+    case fromException x of+        Just (ResponseTimeout {}) -> onTimeout+        _                         -> destroyR p s r+    throwIO x+  where+    onTimeout =+        if timeouts r > p^.settings.maxTimeouts+            then do+                logInfo (p^.logger) $ string8 (show (value r)) <> ": Too many timeouts."+                destroyR p s r+            else put p s r incrTimeouts++take1 :: Pool -> Stripe -> IO (Maybe Resource)+take1 p s = do+    r <- atomically $ do+        c <- readTVar (conns s)+        u <- readTVar (inUse s)+        let n = Seq.length c+        check (u == n)+        let r :< rr = Seq.viewl $ Seq.unstableSortBy (compare `on` refcnt) c+        if | u < p^.settings.maxConnections -> do+                writeTVar (inUse s) $! u + 1+                mkNew p+           | n > 0 && refcnt r < p^.maxRefs -> use s r rr+           | otherwise                      -> return Empty+    case r of+        New io -> do+            x <- io `onException` atomically (modifyTVar' (inUse s) (subtract 1))+            atomically (modifyTVar' (conns s) (|> x))+            return (Just x)+        Used x -> return (Just x)+        Empty  -> return Nothing++use :: Stripe -> Resource -> Seq Resource -> STM Box+use s r rr = do+    writeTVar (conns s) $! rr |> r { refcnt = refcnt r + 1 }+    return (Used r)+{-# INLINE use #-}++mkNew :: Pool -> STM Box+mkNew p = return (New $ Resource <$> p^.currentTime <*> pure 1 <*> pure 0 <*> p^.createFn)+{-# INLINE mkNew #-}++put :: Pool -> Stripe -> Resource -> (Resource -> Resource) -> IO ()+put p s r f = do+    now <- p^.currentTime+    let updated x = f x { tstamp = now, refcnt = refcnt x - 1 }+    atomically $ do+        rs <- readTVar (conns s)+        let (xs, rr) = Seq.breakl ((value r ==) . value) rs+        case Seq.viewl rr of+            EmptyL  -> writeTVar (conns s) $! xs         |> updated r+            y :< ys -> writeTVar (conns s) $! (xs >< ys) |> updated y++destroyR :: Pool -> Stripe -> Resource -> IO ()+destroyR p s r = do+    atomically $ do+        rs <- readTVar (conns s)+        case find ((value r ==) . value) rs of+            Nothing -> return ()+            Just  _ -> do+                modifyTVar' (inUse s) (subtract 1)+                writeTVar (conns s) $! Seq.filter ((value r /=) . value) rs+    ignore $ p^.destroyFn $ value r++reaper :: Pool -> IO ()+reaper p = forever $ do+    threadDelay 1000000+    now <- p^.currentTime+    let isStale r = refcnt r == 0 && now `diffUTCTime` tstamp r > p^.settings.idleTimeout+    Vec.forM_ (p^.stripes) $ \s -> do+        x <- atomically $ do+                (stale, okay) <- Seq.partition isStale <$> readTVar (conns s)+                unless (Seq.null stale) $ do+                    writeTVar   (conns s) okay+                    modifyTVar' (inUse s) (subtract (Seq.length stale))+                return stale+        forM_ x $ \v -> ignore $ do+            logDebug (p^.logger) $ "Reaping idle connection: " <> string8 (show (value v))+            p^.destroyFn $ (value v)++stripe :: Pool -> IO Stripe+stripe p = ((p^.stripes) !) <$> ((`mod` (p^.settings.poolStripes)) . hash) <$> myThreadId+{-# INLINE stripe #-}++incrTimeouts :: Resource -> Resource+incrTimeouts r = r { timeouts = timeouts r + 1 }+{-# INLINE incrTimeouts #-}

@@ -0,0 +1,144 @@+-- 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/.++module Database.CQL.IO.PrepQuery+    ( PrepQuery+    , prepared+    , queryString++    , PreparedQueries+    , new+    , lookupQueryId+    , lookupQueryString+    , insert+    , delete+    , queryStrings+    ) where++import Control.Applicative+import Control.Concurrent.STM+import Control.Monad+import Crypto.Hash+import Crypto.Hash.Algorithms (SHA1)+import Data.ByteString (ByteString)+import Data.Text.Lazy (Text)+import Data.Text.Lazy.Encoding (encodeUtf8)+import Data.Foldable (for_)+import Data.Map.Strict (Map)+import Data.String+import Database.CQL.Protocol hiding (Map)+import Database.CQL.IO.Exception (HashCollision (..))+import Prelude++import qualified Data.Map.Strict as M++-----------------------------------------------------------------------------+-- Prepared Query++-- | Representation of a prepared 'QueryString'. A prepared query is+-- executed in two stages:+--+--   1. The query string is sent to a server without parameters for+--      preparation. The server responds with a 'QueryId'.+--   2. The prepared query is executed by sending the 'QueryId'+--      and parameters to the server.+--+-- Thereby step 1 is only performed when the query has not yet been prepared+-- with the host (coordinator) used for query execution. Thus, prepared+-- queries enhance performance by avoiding the repeated sending and parsing+-- of query strings.+--+-- Query preparation is handled transparently by the client.+-- See 'Database.CQL.IO.setPrepareStrategy'.+--+-- __Note__+--+-- Prepared statements are fully supported but rely on some+-- assumptions beyond the scope of the CQL binary protocol+-- specification (spec):+--+-- (1) The spec scopes the 'QueryId' to the node the query has+--     been prepared with. The spec does not state anything+--     about the format of the 'QueryId'. However the official+--     Java driver assumes that any given 'QueryString' yields+--     the same 'QueryId' on every node. This client make the+--     same assumption.+-- (2) In case a node does not know a given 'QueryId' an 'Unprepared'+--     error is returned. We assume that it is always safe to+--     transparently re-prepare the corresponding 'QueryString' and+--     to re-execute the original request against the same node.+--+-- Besides these assumptions there is also a potential tradeoff in+-- regards to /eager/ vs. /lazy/ query preparation.+-- We understand /eager/ to mean preparation against all current nodes of+-- a cluster and /lazy/ to mean preparation against a single node on demand,+-- i.e. upon receiving an 'Unprepared' error response. Which strategy to+-- choose depends on the scope of query reuse and the size of the cluster.+-- The global default can be changed through the 'Settings' module as well+-- as locally using 'withPrepareStrategy'.+data PrepQuery k a b = PrepQuery+    { pqStr :: !(QueryString k a b)+    , pqId  :: !PrepQueryId+    }++instance IsString (PrepQuery k a b) where+    fromString = prepared . fromString++newtype PrepQueryId = PrepQueryId (Digest SHA1) deriving (Eq, Ord)++prepared :: QueryString k a b -> PrepQuery k a b+prepared q = PrepQuery q $ PrepQueryId (hashlazy . encodeUtf8 . unQueryString $ q)++queryString :: PrepQuery k a b -> QueryString k a b+queryString = pqStr++-----------------------------------------------------------------------------+-- Map of prepared queries to their query ID and query string++newtype QST = QST { unQST :: Text }+newtype QID = QID { unQID :: ByteString } deriving (Eq, Ord)++data PreparedQueries = PreparedQueries+    { queryMap :: !(TVar (Map PrepQueryId (QID, QST)))+    , qid2Str  :: !(TVar (Map QID QST))+    }++new :: IO PreparedQueries+new = PreparedQueries <$> newTVarIO M.empty <*> newTVarIO M.empty++lookupQueryId :: PrepQuery k a b -> PreparedQueries -> STM (Maybe (QueryId k a b))+lookupQueryId q m = do+    qm <- readTVar (queryMap m)+    return $ QueryId . unQID . fst <$> M.lookup (pqId q) qm++lookupQueryString :: QueryId k a b -> PreparedQueries -> STM (Maybe (QueryString k a b))+lookupQueryString q m = do+    qm <- readTVar (qid2Str m)+    return $ QueryString . unQST <$> M.lookup (QID $ unQueryId q) qm++insert :: PrepQuery k a b -> QueryId k a b -> PreparedQueries -> STM ()+insert q i m = do+    qq <- M.lookup (pqId q) <$> readTVar (queryMap m)+    for_ qq (verify . snd)+    modifyTVar' (queryMap m) $+        M.insert (pqId q) (QID $ unQueryId i, QST $ unQueryString (pqStr q))+    modifyTVar' (qid2Str  m) $+        M.insert (QID $ unQueryId i) (QST $ unQueryString (pqStr q))+  where+    verify qs =+        unless (unQST qs == unQueryString (pqStr q)) $ do+            let a = unQST qs+            let b = unQueryString (pqStr q)+            throwSTM (HashCollision a b)++delete :: PrepQuery k a b -> PreparedQueries -> STM ()+delete q m = do+    qid <- M.lookup (pqId q) <$> readTVar (queryMap m)+    modifyTVar' (queryMap m) $ M.delete (pqId q)+    case qid of+        Nothing -> return ()+        Just  i -> modifyTVar' (qid2Str m) $ M.delete (fst i)++queryStrings :: PreparedQueries -> STM [Text]+queryStrings m = map (unQST . snd) . M.elems <$> readTVar (queryMap m)

@@ -0,0 +1,52 @@+-- 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 OverloadedStrings #-}++module Database.CQL.IO.Protocol where++import Control.Monad.Catch+import Data.ByteString.Lazy (ByteString)+import Data.Maybe (fromMaybe)+import Data.Monoid ((<>))+import Database.CQL.Protocol+import Database.CQL.IO.Exception++import qualified Data.Text.Lazy as LT++data Frame = Frame !Header !ByteString++-- | Parse a CQL protocol frame into a 'Response'.+parse :: (Tuple a, Tuple b, MonadThrow m)+    => Compression+    -> Frame+    -> m (Response k a b)+parse x (Frame h b) =+    case unpack x h b of+        Left  e -> throwM $ ParseError ("response body reading: " ++ e)+        Right r -> return r++-- | Serialise a 'Request' into a complete CQL protocol frame,+-- including header, length and body.+serialise :: (Tuple a, MonadThrow m)+    => Version+    -> Compression+    -> Request k a b+    -> Int+    -> m ByteString+serialise v f r i =+    let c = case getOpCode r of+                OcStartup -> noCompression+                OcOptions -> noCompression+                _         -> f+        s = mkStreamId i+    in either (throwM . SerialiseError) return (pack v c (isTracing r) s r)+  where+    isTracing :: Request k a b -> Bool+    isTracing (RqQuery (Query _ p))     = fromMaybe False $ enableTracing p+    isTracing (RqExecute (Execute _ p)) = fromMaybe False $ enableTracing p+    isTracing _                         = False++quoted :: LT.Text -> LT.Text+quoted s = "\"" <> LT.replace "\"" "\"\"" s <> "\""

@@ -0,0 +1,380 @@+-- 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 RankNTypes          #-}+{-# LANGUAGE RecordWildCards     #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE StrictData          #-}+{-# LANGUAGE TemplateHaskell     #-}++module Database.CQL.IO.Settings where++import Control.Exception (IOException)+import Control.Lens (makeLenses, set, over)+import Control.Monad.Catch+import Control.Retry hiding (retryPolicy)+import Data.List.NonEmpty (NonEmpty (..), (<|))+import Data.Semigroup ((<>))+import Data.Time+import Database.CQL.Protocol+import Database.CQL.IO.Cluster.Policies (Policy, random)+import Database.CQL.IO.Connection.Socket (PortNumber)+import Database.CQL.IO.Connection.Settings as C+import Database.CQL.IO.Exception+import Database.CQL.IO.Log+import Database.CQL.IO.Pool as P+import Database.CQL.IO.Timeouts (Milliseconds (..))+import OpenSSL.Session (SSLContext, SomeSSLException)++import qualified Data.HashMap.Strict as HashMap++data Settings = Settings+    { _poolSettings  :: PoolSettings+    , _connSettings  :: ConnectionSettings+    , _retrySettings :: RetrySettings+    , _logger        :: Logger+    , _protoVersion  :: Version+    , _portnumber    :: PortNumber+    , _contacts      :: NonEmpty String+    , _policyMaker   :: IO Policy+    , _prepStrategy  :: PrepareStrategy+    }++-- | Strategy for the execution of 'PrepQuery's.+data PrepareStrategy+    = EagerPrepare -- ^ cluster-wide preparation+    | LazyPrepare  -- ^ on-demand per node preparation+    deriving (Eq, Ord, Show)++-- | Retry settings control if and how retries are performed+-- by the client upon encountering errors during query execution.+--+-- There are three aspects to the retry settings:+--+--   1. /What/ to retry. Determined by the retry handlers ('setRetryHandlers').+--   2. /How/ to perform the retries. Determined by the retry policy+--      ('setRetryPolicy').+--   3. Configuration adjustments to be performed before retrying. Determined by+--      'adjustConsistency', 'adjustSendTimeout' and 'adjustResponseTimeout'.+--      These adjustments are performed /once/ before the first retry and are+--      scoped to the retries only.+--+-- Retry settings can be scoped to a client action by 'Database.CQL.IO.Client.retry',+-- thus locally overriding the \"global\" retry settings configured by+-- 'setRetrySettings'.+data RetrySettings = RetrySettings+    { _retryPolicy        :: forall m. Monad m => RetryPolicyM m+    , _reducedConsistency :: (Maybe Consistency)+    , _sendTimeoutChange  :: Milliseconds+    , _recvTimeoutChange  :: Milliseconds+    , _retryHandlers      :: forall m. Monad m => [RetryStatus -> Handler m Bool]+    }++makeLenses ''RetrySettings+makeLenses ''Settings++-- | Default settings:+--+-- * The initial contact point is \"localhost\" on port 9042.+--+-- * The load-balancing policy is 'random'.+--+-- * The binary protocol version is 3.+--+-- * The connection idle timeout is 60s.+--+-- * The connection pool uses 4 stripes to mitigate thread contention.+--+-- * Connections use a connect timeout of 5s, a send timeout of 3s and+--   a receive timeout of 10s.+--+-- * 128 streams per connection are used.+--+-- * 16k receive buffer size.+--+-- * No compression is applied to frame bodies.+--+-- * No default keyspace is used.+--+-- * A single, immediate retry is performed for errors that are always safe to+--   retry and are known to have good chances of succeeding on a retry.+--   See 'defRetrySettings'.+--+-- * Query preparation is done lazily. See 'PrepareStrategy'.+defSettings :: Settings+defSettings = Settings+    P.defSettings+    C.defSettings+    defRetrySettings+    nullLogger+    V3+    9042+    ("localhost" :| [])+    random+    LazyPrepare++-----------------------------------------------------------------------------+-- Settings++-- | Set the binary protocol version to use.+setProtocolVersion :: Version -> Settings -> Settings+setProtocolVersion v = set protoVersion v++-- | Set the initial contact points (hosts) from which node discovery will+-- start.+setContacts :: String -> [String] -> Settings -> Settings+setContacts v vv = set contacts (v :| vv)++-- | Add an additional host to the contact list.+addContact :: String -> Settings -> Settings+addContact v = over contacts (v <|)++-- | Set the portnumber to use to connect on /every/ node of the cluster.+setPortNumber :: PortNumber -> Settings -> Settings+setPortNumber v = set portnumber v++-- | Set the load-balancing policy.+setPolicy :: IO Policy -> Settings -> Settings+setPolicy v = set policyMaker v++-- | Set strategy to use for preparing statements.+setPrepareStrategy :: PrepareStrategy -> Settings -> Settings+setPrepareStrategy v = set prepStrategy v++-- | Set the 'Logger' to use for processing log messages emitted by the client.+setLogger :: Logger -> Settings -> Settings+setLogger v = set logger v++-----------------------------------------------------------------------------+-- Pool Settings++-- | Set the connection idle timeout. Connections in a pool will be closed+-- if not in use for longer than this timeout.+setIdleTimeout :: NominalDiffTime -> Settings -> Settings+setIdleTimeout v = set (poolSettings.idleTimeout) v++-- | Maximum connections per pool /stripe/.+setMaxConnections :: Int -> Settings -> Settings+setMaxConnections v = set (poolSettings.maxConnections) v++-- | Set the number of pool stripes to use. A good setting is equal to the+-- number of CPU cores this codes is running on.+setPoolStripes :: Int -> Settings -> Settings+setPoolStripes v s+    | 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+-- connection that was used to make the request as it is uncertain if+-- a response will arrive later. Thus the bandwith of a connection will be+-- decreased. This settings defines a threshold after which we close the+-- connection to get a new one with all streams available.+setMaxTimeouts :: Int -> Settings -> Settings+setMaxTimeouts v = set (poolSettings.maxTimeouts) v++-----------------------------------------------------------------------------+-- Connection Settings++-- | Set the compression to use for frame body compression.+setCompression :: Compression -> Settings -> Settings+setCompression v = set (connSettings.compression) v++-- | Set the maximum number of streams per connection. In version 2 of the+-- binary protocol at most 128 streams can be used. Version 3 supports up+-- to 32768 streams.+setMaxStreams :: Int -> Settings -> Settings+setMaxStreams v s+    | v < 1 || v > 32768 = error "cql-io settings: max. streams must be within [1, 32768]"+    | otherwise          = set (connSettings.maxStreams) v s++-- | Set the connect timeout of a connection.+setConnectTimeout :: NominalDiffTime -> Settings -> Settings+setConnectTimeout v = set (connSettings.connectTimeout) (Ms $ round (1000 * v))++-- | Set the send timeout of a connection. Requests exceeding the send+-- timeout will cause the connection to be closed and fail with a+-- 'ConnectionClosed' exception.+setSendTimeout :: NominalDiffTime -> Settings -> Settings+setSendTimeout v = set (connSettings.sendTimeout) (Ms $ round (1000 * v))++-- | Set the response timeout of a connection. Requests exceeding the+-- response timeout will fail with a 'ResponseTimeout' exception.+setResponseTimeout :: NominalDiffTime -> Settings -> Settings+setResponseTimeout v = set (connSettings.responseTimeout) (Ms $ round (1000 * v))++-- | Set the default keyspace to use. Every new connection will be+-- initialised to use this keyspace.+setKeyspace :: Keyspace -> Settings -> Settings+setKeyspace v = set (connSettings.defKeyspace) (Just v)++-- | Set the retry settings to use.+setRetrySettings :: RetrySettings -> Settings -> Settings+setRetrySettings v = set retrySettings v++-- | Set maximum receive buffer size.+--+-- The actual buffer size used will be the minimum of the CQL response size+-- and the value set here.+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)++-- | Set the supported authentication mechanisms.+--+-- When a Cassandra server requests authentication on a connection,+-- it specifies the requested 'AuthMechanism'. The client 'Authenticator'+-- is chosen based that name. If no authenticator with a matching+-- name is configured, an 'AuthenticationError' is thrown.+setAuthentication :: [C.Authenticator] -> Settings -> Settings+setAuthentication = set (connSettings.authenticators)+                  . HashMap.fromList+                  . map (\a -> (authMechanism a, a))++-----------------------------------------------------------------------------+-- Retry Settings++-- | Never retry.+noRetry :: RetrySettings+noRetry = RetrySettings+    { _retryPolicy        = RetryPolicyM $ const (return Nothing)+    , _reducedConsistency = Nothing+    , _sendTimeoutChange  = Ms 0+    , _recvTimeoutChange  = Ms 0+    , _retryHandlers      = []+    }++-- | Default retry settings, combining 'defRetryHandlers' with 'defRetryPolicy'.+-- Consistency is never reduced on retries and timeout values remain unchanged.+defRetrySettings :: RetrySettings+defRetrySettings = RetrySettings+    { _retryPolicy        = defRetryPolicy+    , _reducedConsistency = Nothing+    , _sendTimeoutChange  = Ms 0+    , _recvTimeoutChange  = Ms 0+    , _retryHandlers      = defRetryHandlers+    }++-- | Eager retry settings, combining 'eagerRetryHandlers' with+-- 'eagerRetryPolicy'. Consistency is never reduced on retries and timeout+-- values remain unchanged.+eagerRetrySettings :: RetrySettings+eagerRetrySettings = RetrySettings+    { _retryPolicy        = eagerRetryPolicy+    , _reducedConsistency = Nothing+    , _sendTimeoutChange  = Ms 0+    , _recvTimeoutChange  = Ms 0+    , _retryHandlers      = eagerRetryHandlers+    }++-- | The default retry policy permits a single, immediate retry.+defRetryPolicy :: RetryPolicy+defRetryPolicy = limitRetries 1++-- | The eager retry policy permits 5 retries with exponential+-- backoff (base-2) with an initial delay of 100ms, i.e. the+-- retries will be performed with 100ms, 200ms, 400ms, 800ms+-- and 1.6s delay, respectively, for a maximum delay of ~3s.+eagerRetryPolicy :: RetryPolicy+eagerRetryPolicy = limitRetries 5 <> exponentialBackoff 100000++-- | The default retry handlers permit a retry for the following errors:+--+--   * A 'HostError', since it always occurs before a query has been+--     sent to the server.+--+--   * A 'ConnectionError' that is a 'ConnectTimeout', since it always+--     occurs before a query has been sent to the server.+--+--   * A 'ResponseError' that is one of the following:+--+--       * 'Unavailable', since that is an error response from a coordinator+--         before the query is actually executed.+--       * A 'ReadTimeout' that indicates that the required consistency+--         level could be achieved but the data was unfortunately chosen+--         by the coordinator to be returned from a replica that turned+--         out to be unavailable. A retry has a good chance of getting the data+--         from one of the other replicas.+--       * A 'WriteTimeout' for a write to the batch log failed. The batch log+--         is written prior to execution of the statements of the batch and+--         hence these errors are safe to retry.+--+defRetryHandlers :: Monad m => [RetryStatus -> Handler m Bool]+defRetryHandlers =+    [ const $ Handler $ \(e :: ConnectionError) -> case e of+        ConnectTimeout {} -> return True+        _                 -> return False+    , const $ Handler $ \(e :: ResponseError) -> return $ case reCause e of+        Unavailable  {}   -> True+        ReadTimeout  {..} -> rTimeoutNumAck >= rTimeoutNumRequired &&+                             not rTimeoutDataPresent+        WriteTimeout {..} -> wTimeoutWriteType == WriteBatchLog+        _                 -> False+    , const $ Handler $ \(_ :: HostError)        -> return True+    , const $ Handler $ \(_ :: SomeSSLException) -> return True+    ]++-- | The eager retry handlers permit a superset of the errors+-- of 'defRetryHandlers', namely:+--+--   * Any 'ResponseError' that is a 'ReadTimeout', 'WriteTimeout',+--     'Overloaded', 'Unavailable' or 'ServerError'.+--+--   * Any 'ConnectionError'.+--+--   * Any 'IOException'.+--+--   * Any 'HostError'.+--+--   * Any 'SomeSSLException' (if an SSL context is configured).+--+-- Notably, these retry handlers are only safe to use for idempotent+-- queries, or if a duplicate write has no severe consequences in+-- the context of the application's data model.+eagerRetryHandlers :: Monad m => [RetryStatus -> Handler m Bool]+eagerRetryHandlers =+    [ 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+    , const $ Handler $ \(_ :: SomeSSLException) -> return True+    ]++-- | Set the 'RetryPolicy' to apply on retryable exceptions,+-- which determines the number and distribution of retries over time,+-- i.e. /how/ retries are performed. Configuring a retry policy+-- does not specify /what/ errors should actually be retried.+-- See 'setRetryHandlers'.+setRetryPolicy :: RetryPolicy -> RetrySettings -> RetrySettings+setRetryPolicy v s = s { _retryPolicy = v }++-- | Set the exception handlers that decide whether a request can be+-- retried by the client, i.e. /what/ errors are permissible to retry.+-- For configuring /how/ the retries are performed, see 'setRetryPolicy'.+setRetryHandlers :: (forall m. Monad m => [RetryStatus -> Handler m Bool])+    -> RetrySettings -> RetrySettings+setRetryHandlers v s = s { _retryHandlers = v }++-- | On retry, change the consistency to the given value.+adjustConsistency :: Consistency -> RetrySettings -> RetrySettings+adjustConsistency v = set reducedConsistency (Just v)++-- | On retry adjust the send timeout. See 'setSendTimeout'.+adjustSendTimeout :: NominalDiffTime -> RetrySettings -> RetrySettings+adjustSendTimeout v = set sendTimeoutChange (Ms $ round (1000 * v))++-- | On retry adjust the response timeout. See 'setResponseTimeout'.+adjustResponseTimeout :: NominalDiffTime -> RetrySettings -> RetrySettings+adjustResponseTimeout v = set recvTimeoutChange (Ms $ round (1000 * v))+

@@ -0,0 +1,34 @@+-- 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/.++module Database.CQL.IO.Signal+    ( Signal+    , signal+    , connect+    , emit+    , (|->)+    , ($$)+    ) where++import Control.Concurrent (forkIO)+import Data.IORef++newtype Signal a = Sig (IORef [a -> IO ()])++signal :: IO (Signal a)+signal = Sig <$> newIORef []++connect :: Signal a -> (a -> IO ()) -> IO ()+connect (Sig s) f = modifyIORef s (f:)++infixl 2 |->+(|->) :: Signal a -> (a -> IO ()) -> IO ()+(|->) = connect++emit :: Signal a -> a -> IO ()+emit (Sig s) a = readIORef s >>= mapM_ (forkIO . ($ a))++infixr 1 $$+($$) :: Signal a -> a -> IO ()+($$) = emit

@@ -0,0 +1,56 @@+-- 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/.++module Database.CQL.IO.Sync+    ( Sync+    , create+    , get+    , put+    , kill+    , close+    ) where++import Control.Applicative+import Control.Concurrent.STM+import Control.Exception (SomeException, Exception, toException)+import Prelude++data State a+    = Empty+    | Value  !a+    | Killed !SomeException+    | Closed !SomeException++newtype Sync a = Sync (TVar (State a))++create :: IO (Sync a)+create = Sync <$> newTVarIO Empty++get :: Sync a -> IO a+get (Sync s) = atomically $ do+    v <- readTVar s+    case v of+        Empty    -> retry+        Value  a -> writeTVar s Empty >> return a+        Closed x -> throwSTM x+        Killed x -> throwSTM x++put :: a -> Sync a -> IO Bool+put a (Sync s) = atomically $ do+    v <- readTVar s+    case v of+        Empty    -> writeTVar s (Value a) >> return True+        Closed _ -> return True+        _        -> writeTVar s Empty     >> return False++kill :: Exception e => e -> Sync a -> IO ()+kill x (Sync s) = atomically $ do+    v <- readTVar s+    case v of+        Closed _ -> return ()+        _        -> writeTVar s (Killed $ toException x)++close :: Exception e => e -> Sync a -> IO ()+close x (Sync s) = atomically $ writeTVar s (Closed $ toException x)+

@@ -0,0 +1,47 @@+-- 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/.++module Database.CQL.IO.Tickets+    ( Ticket+    , toInt+    , Pool+    , pool+    , close+    , get+    , markAvailable+    ) where++import Control.Applicative+import Control.Concurrent.STM+import Control.Exception (SomeException, Exception, toException)+import Data.Set (Set)+import Prelude++import qualified Data.Set as Set++newtype Ticket = Ticket { toInt :: Int } deriving (Eq, Ord, Show)++newtype Pool = Pool (TVar (Either SomeException (Set Ticket)))++pool :: Int -> IO Pool+pool n = Pool <$> newTVarIO (Right . Set.fromList $ map Ticket [0 .. n-1])++close :: Exception e => e -> Pool -> IO ()+close x (Pool p) = atomically $ writeTVar p (Left $ toException x)++get :: Pool -> IO Ticket+get (Pool p) = atomically $ readTVar p >>= popHead+  where+    popHead (Left x) = throwSTM x+    popHead (Right x)+        | Set.null x = retry+        | otherwise  = do+            let (t, tt) = Set.deleteFindMin x+            writeTVar p (Right tt)+            return t++markAvailable :: Pool -> Int -> IO ()+markAvailable (Pool p) t =+    atomically $ modifyTVar' p (fmap (Set.insert (Ticket t)))+

@@ -0,0 +1,77 @@+-- 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 GeneralizedNewtypeDeriving #-}++module Database.CQL.IO.Timeouts+    ( TimeoutManager+    , Milliseconds (..)+    , create+    , destroy+    , Action+    , add+    , cancel+    , withTimeout+    ) where++import Control.Concurrent.STM+import Control.Exception (mask_, bracket)+import Control.Reaper+import Control.Monad+import Database.CQL.IO.Exception (ignore)+import Database.CQL.IO.Connection.Settings (Milliseconds (..))++data TimeoutManager = TimeoutManager+    { roundtrip :: !Int+    , reaper    :: !(Reaper [Action] Action)+    }++data Action = Action+    { action :: !(IO ())+    , state  :: !(TVar State)+    }++data State = Running !Int | Canceled++create :: Milliseconds -> IO TimeoutManager+create (Ms n) = TimeoutManager n <$> mkReaper defaultReaperSettings+    { reaperAction = mkListAction prune+    , reaperDelay  = n * 1000+    }+  where+    prune a = do+        s <- atomically $ do+            x <- readTVar (state a)+            writeTVar (state a) (newState x)+            return x+        case s of+            Running 0 -> do+                ignore (action a)+                return Nothing+            Canceled -> return Nothing+            _        -> return $ Just a++    newState (Running k) = Running (k - 1)+    newState s           = s++destroy :: TimeoutManager -> Bool -> IO ()+destroy tm exec = mask_ $ do+    a <- reaperStop (reaper tm)+    when exec $ mapM_ f a+  where+    f e = readTVarIO (state e) >>= \s -> case s of+        Running _ -> ignore (action e)+        Canceled  -> return ()++add :: TimeoutManager -> Milliseconds -> IO () -> IO Action+add tm (Ms n) a = do+    r <- Action a <$> newTVarIO (Running $ n `div` roundtrip tm)+    reaperAdd (reaper tm) r+    return r++cancel :: Action -> IO ()+cancel a = atomically $ writeTVar (state a) Canceled++withTimeout :: TimeoutManager -> Milliseconds -> IO () -> IO a -> IO a+withTimeout tm m x a = bracket (add tm m x) cancel $ const a

@@ -0,0 +1,14 @@+module Main (main) where++import Test.Tasty+import Test.Database.CQL.IO+import Test.Database.CQL.IO.Jobs++main :: IO ()+main = do+    tree <- sequence+        [ Test.Database.CQL.IO.tests+        , pure Test.Database.CQL.IO.Jobs.tests+        ]+    defaultMain $ testGroup "cql-io" tree+

@@ -0,0 +1,365 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE QuasiQuotes       #-}+{-# LANGUAGE TypeFamilies      #-}++module Test.Database.CQL.IO (tests) where++import Control.Monad+import Control.Monad.Identity+import Control.Monad.IO.Class+import Data.Decimal+import Data.Int+import Data.IP+import Data.List (sort)+import Data.Maybe+import Data.Text (Text)+import Data.Time+import Data.UUID+import Database.CQL.Protocol+import Database.CQL.IO as Client+import System.Environment+import Test.Tasty+import Test.Tasty.HUnit+import Text.RawString.QQ++import qualified Data.Set as Set++-----------------------------------------------------------------------------+-- Test Setup++type TestHost = String++tests :: IO TestTree+tests = do+    h <- fromMaybe "localhost" <$> lookupEnv "CASSANDRA_HOST"+    initSchema h+    fmap (testGroup "Database.CQL.IO") $+        forM versions (\v -> do+            c <- Client.init (settings h v)+            return $ testGroup (show v) (cqlTests c))++versions :: [Version]+versions = [V3, V4]++settings :: TestHost -> Version -> Settings+settings h v = setContacts h []+             . setProtocolVersion v+             . setLogger (stdoutLogger LogInfo)+             $ defSettings++initSchema :: TestHost -> IO ()+initSchema h = do+    c <- Client.init (settings h V4)+    runClient c $ do+        dropKeyspace+        createKeyspace+        createTables+    shutdown c++test :: ClientState -> String -> Client () -> TestTree+test c name runTest =+    testCase name $+        runClient c $ do+            truncateTables+            runTest++-----------------------------------------------------------------------------+-- Test Schema++-- Columns of cqltest.table1+type Ty1 =+    ( Int64+    , Ascii+    , Blob+    , Bool+    , Decimal+    , Double+    , Float+    , Int32+    , UTCTime+    , UUID+    , Text+    , Integer+    , TimeUuid+    , IP+    )++-- Columns of cqltest.table2+type Ty2 =+    ( Int64+    , [Int32]+    , Set Ascii+    , Map Ascii Int32+    , Maybe Int32+    , (Bool, Ascii, Int32)+    , Map Int32 (Map Int32 (Set Ascii))+    )++createKeyspace :: Client ()+createKeyspace = void $ schema cql (params ())+  where+    cql :: QueryString S () ()+    cql = [r| create keyspace if not exists cqltest+                with replication = {+                    'class': 'SimpleStrategy',+                    'replication_factor': '1'+                } |]++dropKeyspace :: Client ()+dropKeyspace = void $ schema cql (params ())+  where+    cql :: QueryString S () ()+    cql = "drop keyspace if exists cqltest"++createTables :: Client ()+createTables = forM_ [cql1, cql2, cql3] $ \q ->+    void $ schema q (params ())+  where+    cql1, cql2, cql3 :: QueryString S () ()+    cql1 = [r|+        create table if not exists cqltest.test1+            ( a bigint+            , b ascii+            , c blob+            , d boolean+            , e decimal+            , f double+            , g float+            , h int+            , i timestamp+            , j uuid+            , k varchar+            , l varint+            , m timeuuid+            , n inet+            , primary key (a)+            ) |]++    cql2 = [r|+        create table if not exists cqltest.test2+            ( a bigint+            , b list<int>+            , c set<ascii>+            , d map<ascii,int>+            , e int+            , f tuple<boolean,ascii,int>+            , g map<int,frozen<map<int,set<ascii>>>>+            , primary key (a)+            ) |]++    cql3 = [r|+        create table if not exists cqltest.counters+            ( a bigint+            , n counter+            , primary key (a)+            ) |]++truncateTables :: Client ()+truncateTables = forM_ [cql1, cql2, cql3] $ \q ->+    void $ schema q (params ())+  where+    cql1, cql2, cql3 :: QueryString S () ()+    cql1 = "truncate table cqltest.test1"+    cql2 = "truncate table cqltest.test2"+    cql3 = "truncate table cqltest.counters"++-----------------------------------------------------------------------------+-- Tests++cqlTests :: ClientState -> [TestTree]+cqlTests c =+    [ test c "write-read" testWriteRead+    , test c "write-read-ttl" testWriteReadTtl+    , test c "trans" testTrans+    , test c "paging" testPaging+    , test c "batch" testBatch+    , test c "batch-counter" testBatchCounter+    ]++testWriteRead :: Client ()+testWriteRead = do+    t <- liftIO $ fmap (\x -> x { utctDayTime = secondsToDiffTime 3600 }) getCurrentTime+    let a = ( 4835637638+            , "hello world"+            , Blob "blooooooooooooooooooooooob"+            , False+            , 1.2342342342423423423423423442+            , 433243.13+            , 1.23+            , 2342342+            , t+            , fromJust (fromString "af93aafe-dea5-4427-bea4-8d7872507efb")+            , "sdfsdžȢぴせそぼξλж҈Ҵאבג"+            , 8763847563478568734687345683765873458734+            , TimeUuid . fromJust $ fromString "559ab19e-52d8-11e3-a847-270bf6910c08"+            , read "127.0.0.1"+            )+    let b = ( 4835637638+            , [1,2,3]+            , Set ["peter", "paul", "mary"]+            , Map [("peter", 1), ("paul", 2), ("mary", 3)]+            , Just 42+            , (True, "ascii", 42)+            , Map [(1, Map [(1, Set ["ascii"])])+                  ,(2, Map [(2, Set ["ascii", "text"])])+                  ]+            )+    write ins1 (params a)+    write ins2 (params b)+    x <- fromJust <$> query1 get1 (params (Identity 4835637638))+    y <- fromJust <$> query1 get2 (params (Identity 4835637638))+    liftIO $ do+        a @=? x+        b @=? y+  where+    ins1 :: PrepQuery W Ty1 ()+    ins1 = [r|+        insert into cqltest.test1+            (a,b,c,d,e,f,g,h,i,j,k,l,m,n)+        values+            (?,?,?,?,?,?,?,?,?,?,?,?,?,?) |]++    ins2 :: PrepQuery W Ty2 ()+    ins2 = [r|+        insert into cqltest.test2+            (a,b,c,d,e,f,g)+        values+            (?,?,?,?,?,?,?) |]++    get1 :: PrepQuery R (Identity Int64) Ty1+    get1 = "select a,b,c,d,e,f,g,h,i,j,k,l,m,n from cqltest.test1 where a = ?"++    get2 :: PrepQuery R (Identity Int64) Ty2+    get2 = "select a,b,c,d,e,f,g from cqltest.test2 where a = ?"++testWriteReadTtl :: Client ()+testWriteReadTtl = do+    write ins (params (1000, True))+    (d, ttl) <- fromJust <$> query1 get (params (Identity 1000))+    liftIO $ do+        assertBool "d == False" d+        assertBool "TTL > 0" (ttl > Just 0)+  where+    ins :: PrepQuery W (Int64, Bool) ()+    ins = "insert into cqltest.test1 (a,d) values (?,?) using ttl 3600"++    get :: PrepQuery R (Identity Int64) (Bool, Maybe Int32)+    get = "select d, ttl(d) from cqltest.test1 where a = ?"++testTrans :: Client ()+testTrans = do+    -- 1st insert (success)+    _row <- head <$> trans ins (params (1, "ascii-1"))+    assertApplied _row++    -- 2nd insert (conflict)+    _row <- head <$> trans ins (params (1, "ascii-1"))+    liftIO $ do+        rowLength _row @?= 15 -- [applied] + full existing row+        fromRow 0 _row @?= Right (Just False)             -- [applied]+        fromRow 1 _row @?= Right (Just (1 :: Int64))      -- a+        fromRow 2 _row @?= Right (Just (Ascii "ascii-1")) -- b+        -- remaining columns with null values (since none were inserted)+        let vnull = Nothing :: Maybe Blob -- type irrelevant+        map (($ _row) . fromRow) [3..14] @?= replicate 12 (Right vnull)++    -- 1st update (success)+    _row <- head <$> trans upd (params ("ascii-2", 1, "ascii-1"))+    assertApplied _row++    -- 2nd update (conflict)+    _row <- head <$> trans upd (params ("ascii-2", 1, "ascii-1"))+    liftIO $ do+        rowLength _row @?= 2 -- [applied] + conflicting value+        fromRow 0 _row @?= Right (Just False)             -- [applied]+        fromRow 1 _row @?= Right (Just (Ascii "ascii-2")) -- b+  where+    ins :: PrepQuery W (Int64, Text) Row+    ins = "insert into cqltest.test1 (a,b) values (?,?) if not exists"++    upd :: PrepQuery W (Text, Int64, Text) Row+    upd = "update cqltest.test1 set b = ? where a = ? if b = ?"++    assertApplied row = liftIO $ do+        rowLength row @?= 1 -- [applied]+        fromRow 0 row @?= Right (Just True)++testPaging :: Client ()+testPaging = do+    let dat = zip [1..101] (repeat "b")+    mapM_ (write ins . params) dat+    p <- paginate qry $ (params ()) { pageSize = Just 10 }+    assertPages 11 (Set.fromList dat) p+  where+    ins :: PrepQuery W (Int64, Ascii) ()+    ins = "insert into cqltest.test1 (a,b) values (?,?)"++    qry :: PrepQuery R () (Int64, Ascii)+    qry = "select a,b from cqltest.test1"++testBatch :: Client ()+testBatch = do+    exec $ setType BatchLogged+    exec $ setType BatchUnLogged+    exec $ setSerialConsistency SerialConsistency+    exec $ setSerialConsistency LocalSerialConsistency+  where+    exec configure = do+        batch $ configure >> forM_ dat (addQuery ins)+        rs <- query qry (params ())+        liftIO $ sort rs @?= dat+        truncateTables++    dat :: [(Int64, Ascii)]+    dat = [(1, "1"), (2, "2"), (3, "3")]++    ins :: QueryString W (Int64, Ascii) ()+    ins = "insert into cqltest.test1 (a,b) values (?,?)"++    qry :: PrepQuery R () (Int64, Ascii)+    qry = "select a,b from cqltest.test1"++testBatchCounter :: Client ()+testBatchCounter = exec >> total 3 >> exec >> total 6+  where+    exec = batch $ do+        setType BatchCounter+        addQuery upd (Identity 1)+        addQuery upd (Identity 2)+        addQuery upd (Identity 3)++    total n = do+        rs <- query qry (params ())+        let n' = sum (map (fromCounter . runIdentity) rs)+        liftIO $ n @=? n'++    upd :: QueryString W (Identity Int64) ()+    upd = "update cqltest.counters set n = n + 1 where a = ?"++    qry :: PrepQuery R () (Identity Counter)+    qry = "select n from cqltest.counters"++-----------------------------------------------------------------------------+-- Utilities++assertPages :: (Ord a, Show a) => Int -> Set.Set a -> Page a -> Client ()+assertPages numPages expected p = do+    let got = Set.fromList (result p)+    let remaining = Set.difference expected got+    liftIO $ hasMore p @?= numPages > 1+    liftIO $ got `Set.isSubsetOf` expected @?= True+    if numPages > 1+        then nextPage p >>= assertPages (numPages - 1) remaining+        else liftIO $ remaining @?= Set.empty++params :: Tuple a => a -> QueryParams a+params p = QueryParams+    { consistency       = One+    , skipMetaData      = False+    , values            = p+    , pageSize          = Nothing+    , queryPagingState  = Nothing+    , serialConsistency = Nothing+    , enableTracing     = Nothing+    }+

@@ -0,0 +1,148 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RankNTypes        #-}++module Test.Database.CQL.IO.Jobs (tests) where++import Control.Concurrent+import Control.Concurrent.Async+import Control.Exception+import Data.IORef+import Data.Either+import Data.Maybe+import Data.Numbers.Primes+import Test.Tasty+import Test.Tasty.HUnit++import Database.CQL.IO.Jobs++tests :: TestTree+tests = testGroup "Database.CQL.IO.Jobs"+    [ testCase "run-sequential"     testRunJobSequential+    , testCase "run-concurrent"     testRunJobConcurrent+    , testCase "try-run-sequential" testTryRunJobSequential+    , testCase "try-run-concurrent" testTryRunJobConcurrent+    , testCase "try-run-replace"    testTryRunJobReplace+    ]++-----------------------------------------------------------------------------+-- Tests++-- 1. Sequentially run jobs with the same key, each waiting on a latch.+-- 2. Release all latches.+-- 3. Expect only the last job to complete and all others to be replaced.+testRunJobSequential :: IO ()+testRunJobSequential = do+    jobs <- newJobs+    iref <- newIORef (1 :: Int)+    let prepare = prepareRun jobs . mulJob iref+    (fails, succs) <- execute mapM prepare (take 100 primes)+    val <- readIORef iref+    map fromException fails @?= replicate 99 (Just JobReplaced)+    succs                   @?= [()]+    val                     @?= 541 -- 100th prime => 100th job++-- 1. Concurrently run jobs with the same key, each waiting on a latch.+-- 2. Release all latches.+-- 3. Expect only one (non-deterministic) job to complete and all others to+--    be replaced.+testRunJobConcurrent :: IO ()+testRunJobConcurrent = do+    jobs <- newJobs+    iref <- newIORef (0 :: Int)+    let prepare = prepareRun jobs . incJob iref+    (fails, succs) <- execute mapConcurrently prepare [1..100 :: Int]+    val <- readIORef iref+    map fromException fails @?= replicate 99 (Just JobReplaced)+    succs                   @?= [()]+    val                     @?= 1++-- 1. Try to run jobs for the same key, each waiting on a latch.+-- 2. Release all latches.+-- 3. Expect only the first job to run.+testTryRunJobSequential :: IO ()+testTryRunJobSequential = do+    jobs <- newJobs+    iref <- newIORef (1 :: Int)+    let prepare = prepareTryRun jobs . mulJob iref+    (fails, succs) <- execute mapM prepare (take 100 primes)+    val <- readIORef iref+    length fails @?= 0+    succs        @?= [()]+    val          @?= 2++-- 1. Concurrently try to run jobs for the same key, each waiting on a latch.+-- 2. Release all latches.+-- 3. Expect only one job to run.+testTryRunJobConcurrent :: IO ()+testTryRunJobConcurrent = do+    jobs <- newJobs+    iref <- newIORef (0 :: Int)+    let prepare = prepareTryRun jobs . incJob iref+    (fails, succs) <- execute mapConcurrently prepare [1..100]+    val <- readIORef iref+    length fails @?= 0+    succs        @?= [()]+    val          @?= 1++-- 1. Concurrently try to run jobs for the same key, each waiting on+--    a latch. One of the jobs is run unconditionally and should+--    thus always take precedence.+-- 2. Release all latches.+-- 3. Expect only the unconditionally run job to complete,+--    replacing one job. All others are never run.+testTryRunJobReplace :: IO ()+testTryRunJobReplace = do+    jobs <- newJobs+    iref <- newIORef (1 :: Int)+    let prepare = choose jobs (mulJob iref)+    (Just a1, l1)  <- prepare 2 -- there should always be a job to replace+    (fails, succs) <- execute mapConcurrently prepare (take 99 (drop 1 primes))+    putMVar l1 ()+    r1  <- either Just (const Nothing) <$> waitCatch a1+    val <- readIORef iref+    (fromException =<< r1) @?= Just JobReplaced+    length fails           @?= 0+    succs                  @?= [()]+    val                    @?= 229 -- 50th's prime is unconditionally run+  where+    choose jobs run i+        | i == 229  = prepareRun    jobs (run i)+        | otherwise = prepareTryRun jobs (run i)++-----------------------------------------------------------------------------+-- Test Jobs++type PrepareJob a = a -> IO (Maybe (Async ()), MVar ())+type ExecuteJob a = IORef a -> a -> IO ()++jobKey :: Int+jobKey = 1++prepareRun :: Jobs Int -> IO () -> IO (Maybe (Async ()), MVar ())+prepareRun jobs run = do+    l <- newEmptyMVar+    a <- runJob jobs jobKey $ takeMVar l >> run+    return (Just a, l)++prepareTryRun :: Jobs Int -> IO () -> IO (Maybe (Async ()), MVar ())+prepareTryRun jobs run = do+    l <- newEmptyMVar+    a <- tryRunJob jobs jobKey $ takeMVar l >> run+    return (a, l)++execute+    :: (forall a b. (a -> IO b) -> [a] -> IO [b])+    -> PrepareJob i+    -> [i]+    -> IO ([SomeException], [()])+execute runIO prepJob input = do+    js <- runIO prepJob input+    as <- runIO (\(a, l) -> putMVar l () *> pure a) js+    partitionEithers <$> mapM waitCatch (catMaybes as)++incJob :: ExecuteJob Int+incJob iref _ = atomicModifyIORef' iref (\x -> (x + 1, ()))++mulJob :: ExecuteJob Int+mulJob iref i = atomicModifyIORef' iref (\x -> (x * i, ()))+