cql-io (empty) → 0.9.6

raw patch · 20 files changed

+2549/−0 lines, 20 filesdep +asyncdep +auto-updatedep +basesetup-changed

Dependencies added: async, auto-update, base, bytestring, containers, cql, exceptions, hashable, iproute, lens, mtl, mwc-random, network, retry, semigroups, stm, text, time, tinylog, transformers, uuid, vector

Files

• LICENSE +373/−0
• README.md +16/−0
• Setup.hs +2/−0
• cql-io.cabal +86/−0
• src/Database/CQL/IO.hs +171/−0
• src/Database/CQL/IO/Client.hs +460/−0
• src/Database/CQL/IO/Cluster/Discovery.hs +32/−0
• src/Database/CQL/IO/Cluster/Host.hs +49/−0
• src/Database/CQL/IO/Cluster/Policies.hs +146/−0
• src/Database/CQL/IO/Connection.hs +311/−0
• src/Database/CQL/IO/Hexdump.hs +63/−0
• src/Database/CQL/IO/Jobs.hs +68/−0
• src/Database/CQL/IO/Pool.hs +239/−0
• src/Database/CQL/IO/Protocol.hs +33/−0
• src/Database/CQL/IO/Settings.hs +156/−0
• src/Database/CQL/IO/Signal.hs +35/−0
• src/Database/CQL/IO/Sync.hs +55/−0
• src/Database/CQL/IO/Tickets.hs +46/−0
• src/Database/CQL/IO/Timeouts.hs +72/−0
• src/Database/CQL/IO/Types.hs +136/−0

@@ -0,0 +1,373 @@+Mozilla Public License Version 2.0+==================================++1. Definitions+--------------++1.1. "Contributor"+    means each individual or legal entity that creates, contributes to+    the creation of, or owns Covered Software.++1.2. "Contributor Version"+    means the combination of the Contributions of others (if any) used+    by a Contributor and that particular Contributor's Contribution.++1.3. "Contribution"+    means Covered Software of a particular Contributor.++1.4. "Covered Software"+    means Source Code Form to which the initial Contributor has attached+    the notice in Exhibit A, the Executable Form of such Source Code+    Form, and Modifications of such Source Code Form, in each case+    including portions thereof.++1.5. "Incompatible With Secondary Licenses"+    means++    (a) that the initial Contributor has attached the notice described+        in Exhibit B to the Covered Software; or++    (b) that the Covered Software was made available under the terms of+        version 1.1 or earlier of the License, but not also under the+        terms of a Secondary License.++1.6. "Executable Form"+    means any form of the work other than Source Code Form.++1.7. "Larger Work"+    means a work that combines Covered Software with other material, in+    a separate file or files, that is not Covered Software.++1.8. "License"+    means this document.++1.9. "Licensable"+    means having the right to grant, to the maximum extent possible,+    whether at the time of the initial grant or subsequently, any and+    all of the rights conveyed by this License.++1.10. "Modifications"+    means any of the following:++    (a) any file in Source Code Form that results from an addition to,+        deletion from, or modification of the contents of Covered+        Software; or++    (b) any new file in Source Code Form that contains any Covered+        Software.++1.11. "Patent Claims" of a Contributor+    means any patent claim(s), including without limitation, method,+    process, and apparatus claims, in any patent Licensable by such+    Contributor that would be infringed, but for the grant of the+    License, by the making, using, selling, offering for sale, having+    made, import, or transfer of either its Contributions or its+    Contributor Version.++1.12. "Secondary License"+    means either the GNU General Public License, Version 2.0, the GNU+    Lesser General Public License, Version 2.1, the GNU Affero General+    Public License, Version 3.0, or any later versions of those+    licenses.++1.13. "Source Code Form"+    means the form of the work preferred for making modifications.++1.14. "You" (or "Your")+    means an individual or a legal entity exercising rights under this+    License. For legal entities, "You" includes any entity that+    controls, is controlled by, or is under common control with You. For+    purposes of this definition, "control" means (a) the power, direct+    or indirect, to cause the direction or management of such entity,+    whether by contract or otherwise, or (b) ownership of more than+    fifty percent (50%) of the outstanding shares or beneficial+    ownership of such entity.++2. License Grants and Conditions+--------------------------------++2.1. Grants++Each Contributor hereby grants You a world-wide, royalty-free,+non-exclusive license:++(a) under intellectual property rights (other than patent or trademark)+    Licensable by such Contributor to use, reproduce, make available,+    modify, display, perform, distribute, and otherwise exploit its+    Contributions, either on an unmodified basis, with Modifications, or+    as part of a Larger Work; and++(b) under Patent Claims of such Contributor to make, use, sell, offer+    for sale, have made, import, and otherwise transfer either its+    Contributions or its Contributor Version.++2.2. Effective Date++The licenses granted in Section 2.1 with respect to any Contribution+become effective for each Contribution on the date the Contributor first+distributes such Contribution.++2.3. Limitations on Grant Scope++The licenses granted in this Section 2 are the only rights granted under+this License. No additional rights or licenses will be implied from the+distribution or licensing of Covered Software under this License.+Notwithstanding Section 2.1(b) above, no patent license is granted by a+Contributor:++(a) for any code that a Contributor has removed from Covered Software;+    or++(b) for infringements caused by: (i) Your and any other third party's+    modifications of Covered Software, or (ii) the combination of its+    Contributions with other software (except as part of its Contributor+    Version); or++(c) under Patent Claims infringed by Covered Software in the absence of+    its Contributions.++This License does not grant any rights in the trademarks, service marks,+or logos of any Contributor (except as may be necessary to comply with+the notice requirements in Section 3.4).++2.4. Subsequent Licenses++No Contributor makes additional grants as a result of Your choice to+distribute the Covered Software under a subsequent version of this+License (see Section 10.2) or under the terms of a Secondary License (if+permitted under the terms of Section 3.3).++2.5. Representation++Each Contributor represents that the Contributor believes its+Contributions are its original creation(s) or it has sufficient rights+to grant the rights to its Contributions conveyed by this License.++2.6. Fair Use++This License is not intended to limit any rights You have under+applicable copyright doctrines of fair use, fair dealing, or other+equivalents.++2.7. Conditions++Sections 3.1, 3.2, 3.3, and 3.4 are conditions of the licenses granted+in Section 2.1.++3. Responsibilities+-------------------++3.1. Distribution of Source Form++All distribution of Covered Software in Source Code Form, including any+Modifications that You create or to which You contribute, must be under+the terms of this License. You must inform recipients that the Source+Code Form of the Covered Software is governed by the terms of this+License, and how they can obtain a copy of this License. You may not+attempt to alter or restrict the recipients' rights in the Source Code+Form.++3.2. Distribution of Executable Form++If You distribute Covered Software in Executable Form then:++(a) such Covered Software must also be made available in Source Code+    Form, as described in Section 3.1, and You must inform recipients of+    the Executable Form how they can obtain a copy of such Source Code+    Form by reasonable means in a timely manner, at a charge no more+    than the cost of distribution to the recipient; and++(b) You may distribute such Executable Form under the terms of this+    License, or sublicense it under different terms, provided that the+    license for the Executable Form does not attempt to limit or alter+    the recipients' rights in the Source Code Form under this License.++3.3. Distribution of a Larger Work++You may create and distribute a Larger Work under terms of Your choice,+provided that You also comply with the requirements of this License for+the Covered Software. If the Larger Work is a combination of Covered+Software with a work governed by one or more Secondary Licenses, and the+Covered Software is not Incompatible With Secondary Licenses, this+License permits You to additionally distribute such Covered Software+under the terms of such Secondary License(s), so that the recipient of+the Larger Work may, at their option, further distribute the Covered+Software under the terms of either this License or such Secondary+License(s).++3.4. Notices++You may not remove or alter the substance of any license notices+(including copyright notices, patent notices, disclaimers of warranty,+or limitations of liability) contained within the Source Code Form of+the Covered Software, except that You may alter any license notices to+the extent required to remedy known factual inaccuracies.++3.5. Application of Additional Terms++You may choose to offer, and to charge a fee for, warranty, support,+indemnity or liability obligations to one or more recipients of Covered+Software. However, You may do so only on Your own behalf, and not on+behalf of any Contributor. You must make it absolutely clear that any+such warranty, support, indemnity, or liability obligation is offered by+You alone, and You hereby agree to indemnify every Contributor for any+liability incurred by such Contributor as a result of warranty, support,+indemnity or liability terms You offer. You may include additional+disclaimers of warranty and limitations of liability specific to any+jurisdiction.++4. Inability to Comply Due to Statute or Regulation+---------------------------------------------------++If it is impossible for You to comply with any of the terms of this+License with respect to some or all of the Covered Software due to+statute, judicial order, or regulation then You must: (a) comply with+the terms of this License to the maximum extent possible; and (b)+describe the limitations and the code they affect. Such description must+be placed in a text file included with all distributions of the Covered+Software under this License. Except to the extent prohibited by statute+or regulation, such description must be sufficiently detailed for a+recipient of ordinary skill to be able to understand it.++5. Termination+--------------++5.1. The rights granted under this License will terminate automatically+if You fail to comply with any of its terms. However, if You become+compliant, then the rights granted under this License from a particular+Contributor are reinstated (a) provisionally, unless and until such+Contributor explicitly and finally terminates Your grants, and (b) on an+ongoing basis, if such Contributor fails to notify You of the+non-compliance by some reasonable means prior to 60 days after You have+come back into compliance. Moreover, Your grants from a particular+Contributor are reinstated on an ongoing basis if such Contributor+notifies You of the non-compliance by some reasonable means, this is the+first time You have received notice of non-compliance with this License+from such Contributor, and You become compliant prior to 30 days after+Your receipt of the notice.++5.2. If You initiate litigation against any entity by asserting a patent+infringement claim (excluding declaratory judgment actions,+counter-claims, and cross-claims) alleging that a Contributor Version+directly or indirectly infringes any patent, then the rights granted to+You by any and all Contributors for the Covered Software under Section+2.1 of this License shall terminate.++5.3. In the event of termination under Sections 5.1 or 5.2 above, all+end user license agreements (excluding distributors and resellers) which+have been validly granted by You or Your distributors under this License+prior to termination shall survive termination.++************************************************************************+*                                                                      *+*  6. Disclaimer of Warranty                                           *+*  -------------------------                                           *+*                                                                      *+*  Covered Software is provided under this License on an "as is"       *+*  basis, without warranty of any kind, either expressed, implied, or  *+*  statutory, including, without limitation, warranties that the       *+*  Covered Software is free of defects, merchantable, fit for a        *+*  particular purpose or non-infringing. The entire risk as to the     *+*  quality and performance of the Covered Software is with You.        *+*  Should any Covered Software prove defective in any respect, You     *+*  (not any Contributor) assume the cost of any necessary servicing,   *+*  repair, or correction. This disclaimer of warranty constitutes an   *+*  essential part of this License. No use of any Covered Software is   *+*  authorized under this License except under this disclaimer.         *+*                                                                      *+************************************************************************++************************************************************************+*                                                                      *+*  7. Limitation of Liability                                          *+*  --------------------------                                          *+*                                                                      *+*  Under no circumstances and under no legal theory, whether tort      *+*  (including negligence), contract, or otherwise, shall any           *+*  Contributor, or anyone who distributes Covered Software as          *+*  permitted above, be liable to You for any direct, indirect,         *+*  special, incidental, or consequential damages of any character      *+*  including, without limitation, damages for lost profits, loss of    *+*  goodwill, work stoppage, computer failure or malfunction, or any    *+*  and all other commercial damages or losses, even if such party      *+*  shall have been informed of the possibility of such damages. This   *+*  limitation of liability shall not apply to liability for death or   *+*  personal injury resulting from such party's negligence to the       *+*  extent applicable law prohibits such limitation. Some               *+*  jurisdictions do not allow the exclusion or limitation of           *+*  incidental or consequential damages, so this exclusion and          *+*  limitation may not apply to You.                                    *+*                                                                      *+************************************************************************++8. Litigation+-------------++Any litigation relating to this License may be brought only in the+courts of a jurisdiction where the defendant maintains its principal+place of business and such litigation shall be governed by laws of that+jurisdiction, without reference to its conflict-of-law provisions.+Nothing in this Section shall prevent a party's ability to bring+cross-claims or counter-claims.++9. Miscellaneous+----------------++This License represents the complete agreement concerning the subject+matter hereof. If any provision of this License is held to be+unenforceable, such provision shall be reformed only to the extent+necessary to make it enforceable. Any law or regulation which provides+that the language of a contract shall be construed against the drafter+shall not be used to construe this License against a Contributor.++10. Versions of the License+---------------------------++10.1. New Versions++Mozilla Foundation is the license steward. Except as provided in Section+10.3, no one other than the license steward has the right to modify or+publish new versions of this License. Each version will be given a+distinguishing version number.++10.2. Effect of New Versions++You may distribute the Covered Software under the terms of the version+of the License under which You originally received the Covered Software,+or under the terms of any subsequent version published by the license+steward.++10.3. Modified Versions++If you create software not governed by this License, and you want to+create a new license for such software, you may create and use a+modified version of this License if you rename the license and remove+any references to the name of the license steward (except to note that+such modified license differs from this License).++10.4. Distributing Source Code Form that is Incompatible With Secondary+Licenses++If You choose to distribute Source Code Form that is Incompatible With+Secondary Licenses under the terms of this version of the License, the+notice described in Exhibit B of this License must be attached.++Exhibit A - Source Code Form License Notice+-------------------------------------------++  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/.++If it is not possible or desirable to put the notice in a particular+file, then You may include the notice in a location (such as a LICENSE+file in a relevant directory) where a recipient would be likely to look+for such a notice.++You may add additional accurate notices of copyright ownership.++Exhibit B - "Incompatible With Secondary Licenses" Notice+---------------------------------------------------------++  This Source Code Form is "Incompatible With Secondary Licenses", as+  defined by the Mozilla Public License, v. 2.0.

@@ -0,0 +1,16 @@+CQL Cassandra driver+--------------------++This library uses the [cql](http://hackage.haskell.org/package/cql) library+which implements Cassandra's CQL protocol and complements it with the+neccessary I/O operations. The feature-set includes:++**Node discovery**. The driver discovers nodes automatically from a small+set of bootstrap nodes.++**Customisable load-balancing policies**. In addition to pre-built LB+policies such as round-robin, users of this library can provide their+own policies if desired.++**Support for connection streams**. Requests can be multiplexed over a+few connections.

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

@@ -0,0 +1,86 @@+name:                 cql-io+version:              0.9.6+synopsis:             Cassandra CQL client.+stability:            experimental+license:              OtherLicense+license-file:         LICENSE+author:               Toralf Wittner+maintainer:           Toralf Wittner <tw@dtex.org>+copyright:            (c) 2014 Toralf Wittner+homepage:             https://github.com/twittner/cql-io/+bug-reports:          https://github.com/twittner/cql-io/issues+category:             Database+build-type:           Simple+cabal-version:        >= 1.10+extra-source-files:   README.md++description:+    CQL Cassandra driver supporting native protocol versions 2 and 3.+    .+    This library uses the <http://hackage.haskell.org/package/cql cql> library+    which implements Cassandra's CQL protocol and complements it with the+    neccessary I/O operations. The feature-set includes:+    .+    * __Node discovery__. The driver discovers nodes automatically from a small+    set of bootstrap nodes.+    .+    * __Customisable load-balancing policies__. In addition to pre-built LB+    policies such as round-robin, users of this library can provide their+    own policies if desired.+    .+    * __Support for connection streams__. Requests can be multiplexed over a+    few connections.++source-repository head+    type:             git+    location:         git@github.com:twittner/cql-io.git++library+    default-language: Haskell2010+    hs-source-dirs:   src+    ghc-options:      -Wall -O2 -fwarn-tabs+    ghc-prof-options: -prof -auto-all++    exposed-modules:+        Database.CQL.IO++    other-modules:+        Database.CQL.IO.Client+        Database.CQL.IO.Cluster.Discovery+        Database.CQL.IO.Cluster.Host+        Database.CQL.IO.Cluster.Policies+        Database.CQL.IO.Connection+        Database.CQL.IO.Hexdump+        Database.CQL.IO.Jobs+        Database.CQL.IO.Pool+        Database.CQL.IO.Protocol+        Database.CQL.IO.Settings+        Database.CQL.IO.Signal+        Database.CQL.IO.Sync+        Database.CQL.IO.Tickets+        Database.CQL.IO.Timeouts+        Database.CQL.IO.Types++    build-depends:+          async            == 2.0.*+        , auto-update      == 0.1.*+        , base             >= 4.7     && < 5.0+        , bytestring       >= 0.10    && < 1.0+        , containers       >= 0.5     && < 1.0+        , cql              == 3.0.*+        , exceptions       >= 0.4     && < 1.0+        , hashable         >= 1.2     && < 2.0+        , iproute          >= 1.3     && < 1.4+        , lens             >= 4.4     && < 4.5+        , mtl              >= 2.1     && < 2.3+        , mwc-random       >= 0.13    && < 0.14+        , network          >= 2.4     && < 3.0+        , retry            >= 0.5     && < 0.6+        , semigroups       >= 0.15    && < 0.16+        , stm              >= 2.4     && < 3.0+        , text             >= 0.11    && < 2.0+        , tinylog          >= 0.8     && < 0.11+        , time             >= 1.4     && < 2.0+        , transformers     >= 0.3     && < 0.5+        , uuid             >= 1.2.6   && < 2.0+        , vector           >= 0.10    && < 1.0

@@ -0,0 +1,171 @@+-- 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:+--+-- @+-- > import Data.Text (Text)+-- > import Data.Functor.Identity+-- > import Database.CQL.IO as Client+-- > import Database.CQL.Protocol+-- > import qualified System.Logger as Logger+-- >+-- > g <- Logger.new Logger.defSettings+-- > c <- Client.init g defSettings+-- > let p = QueryParams One False () Nothing Nothing Nothing+-- > runClient c $ query ("SELECT cql_version from system.local" :: QueryString R () (Identity Text)) p+-- [Identity "3.2.0"]+-- > shutdown c+-- @+--++{-# LANGUAGE DeriveFunctor #-}++module Database.CQL.IO+    ( -- * Client settings+      Settings+    , defSettings+    , addContact+    , setCompression+    , setConnectTimeout+    , setContacts+    , setIdleTimeout+    , setKeyspace+    , setMaxConnections+    , setMaxStreams+    , setMaxTimeouts+    , setMaxWaitQueue+    , setPolicy+    , setPoolStripes+    , setPortNumber+    , setProtocolVersion+    , setResponseTimeout+    , setSendTimeout++      -- * Client monad+    , Client+    , ClientState+    , DebugInfo (..)+    , init+    , runClient+    , shutdown+    , debugInfo++    , query+    , write+    , schema+    , batch++    , Page      (..)+    , emptyPage+    , paginate++      -- ** low-level+    , request+    , command++      -- * Policies+    , Policy (..)+    , random+    , roundRobin++      -- ** Hosts+    , Host+    , HostEvent (..)+    , InetAddr  (..)+    , hostAddr+    , dataCentre+    , rack++    -- * Exceptions+    , InvalidSettings    (..)+    , InternalError      (..)+    , HostError          (..)+    , ConnectionError    (..)+    , UnexpectedResponse (..)+    , Timeout            (..)+    ) where++import Control.Applicative+import Control.Monad.Catch+import Control.Monad (void)+import Data.Maybe (isJust)+import Database.CQL.Protocol+import Database.CQL.IO.Client+import Database.CQL.IO.Cluster.Host+import Database.CQL.IO.Cluster.Policies+import Database.CQL.IO.Settings+import Database.CQL.IO.Types+import Prelude hiding (init)++runQuery :: (Tuple a, Tuple b) => QueryString k a b -> QueryParams a -> Client (Response k a b)+runQuery q p = do+    r <- request (RqQuery (Query q p))+    case r of+        RsError _ e -> throwM e+        _           -> return r++-- | Run a CQL read-only query against a Cassandra node.+query :: (Tuple a, Tuple b) => QueryString R a b -> QueryParams a -> Client [b]+query q p = do+    r <- runQuery q p+    case r of+        RsResult _ (RowsResult _ b) -> return b+        _                           -> throwM UnexpectedResponse++-- | Run a CQL insert/update query against a Cassandra node.+write :: Tuple a => QueryString W a () -> QueryParams a -> Client ()+write q p = void $ runQuery q p++-- | Run a CQL schema query against a Cassandra node.+schema :: Tuple a => QueryString S a () -> QueryParams a -> Client (Maybe SchemaChange)+schema x y = do+    r <- runQuery x y+    case r of+        RsResult _ (SchemaChangeResult s) -> return $ Just s+        RsResult _ VoidResult             -> return Nothing+        _                                 -> throwM UnexpectedResponse++-- | Run a batch query against a Cassandra node.+batch :: Batch -> Client ()+batch b = command (RqBatch b)++-- | 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 :: (Tuple a, Tuple b) => QueryString R a b -> QueryParams a -> Client (Page b)+paginate q p = do+    let p' = p { pageSize = pageSize p <|> Just 10000 }+    r <- runQuery q p'+    case r of+        RsResult _ (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

@@ -0,0 +1,460 @@+-- 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 #-}+{-# LANGUAGE OverloadedStrings          #-}+{-# LANGUAGE ScopedTypeVariables        #-}+{-# LANGUAGE TemplateHaskell            #-}++module Database.CQL.IO.Client+    ( Client+    , ClientState+    , DebugInfo (..)+    , runClient+    , Database.CQL.IO.Client.init+    , shutdown+    , request+    , command+    , debugInfo+    ) where++import Control.Applicative+import Control.Concurrent (threadDelay)+import Control.Concurrent.Async (async)+import Control.Concurrent.STM hiding (retry)+import Control.Exception (IOException)+import Control.Lens hiding ((.=), Context)+import Control.Monad (unless, void, when)+import Control.Monad.Catch+import Control.Monad.IO.Class+import Control.Monad.Reader (ReaderT, runReaderT, MonadReader, ask)+import Control.Retry+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.Jobs (Jobs)+import Database.CQL.IO.Pool+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 System.Logger.Class hiding (Settings, new, settings, create)++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.Timeouts   as TM+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+    , _control  :: TVar Control+    , _failures :: TVar Word64+    , _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'.+--+-- Actual CQL queries are handled by invoking 'request' oder 'command'.+-- Additionally 'debugInfo' returns an internal cluster view.+newtype Client a = Client+    { client :: ReaderT ClientState IO a+    } deriving ( Functor+               , Applicative+               , Monad+               , MonadIO+               , MonadThrow+               , MonadMask+               , MonadCatch+               , MonadReader ClientState+               )++instance MonadLogger Client where+    log l m = view (context.logger) >>= \g -> Logger.log g l m++-----------------------------------------------------------------------------+-- API++-- | Execute the client monad.+runClient :: MonadIO m => ClientState -> Client a -> m a+runClient p a = liftIO $ runReaderT (client a) p++-- | Send a CQL '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.+request :: (Tuple a, Tuple b) => Request k a b -> Client (Response k a b)+request a = do+    s <- ask+    n <- liftIO $ hostCount (s^.policy)+    start s n+  where+    start s n = do+        h <- pickHost (s^.policy)+        p <- Map.lookup h <$> readTVarIO' (s^.hostmap)+        case p of+            Just  x -> action s n x `catches` handlers h+            Nothing -> do+                err $ msg (val "no pool for host " +++ h)+                p' <- mkPool (s^.context) (h^.hostAddr)+                atomically' $ modifyTVar (s^.hostmap) (Map.alter (maybe (Just p') Just) h)+                start s n++    action s n p = do+        res <- tryWith p (go s)+        case res of+            Just  r -> return r+            Nothing ->+                if n > 0 then+                    start s (n - 1)+                else case s^.context.settings.maxWaitQueue of+                    Nothing -> with p (transaction s)+                    Just  q -> retry s q p++    go s h = do+        atomically $ modifyTVar (s^.failures) $ \n -> if n > 0 then n - 1 else 0+        transaction s h++    retry s q p = do+        n <- atomically' $ do+            n <- readTVar (s^.failures)+            unless (n == q) $+                writeTVar (s^.failures) (n + 1)+            return n+        unless (n < q) $+            throwM HostsBusy+        with p (go s)++    transaction s 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 r++    handlers h =+        [ Handler $ \(e :: ConnectionError) -> onConnectionError h e >> throwM e+        , Handler $ \(e :: IOException)     -> onConnectionError h e >> throwM e+        ]++    pickHost p = maybe (throwM NoHostAvailable) return =<< liftIO (select p)++-- | Like 'request' but not returning any result.+command :: Request k () () -> Client ()+command = void . request++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 :: Client DebugInfo+debugInfo = do+    hosts <- Map.keys <$> (readTVarIO' =<< view hostmap)+    pols  <- liftIO . display =<< view policy+    jbs   <- Jobs.showJobs =<< view jobs+    return $ DebugInfo pols jbs hosts++-----------------------------------------------------------------------------+-- 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+            <*> newTVarIO (Control Connected c)+            <*> newTVarIO 0+            <*> newTVarIO Map.empty+            <*> Jobs.new+    e^.sigMonit |-> onEvent p+    runClient x (initialise c)+    return x+  where+    mkConnection t h = do+        a <- C.resolve h (s^.portnumber)+        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) $ 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.+shutdown :: MonadIO m => ClientState -> m ()+shutdown s = liftIO $ 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++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 . continue e)+          c+    Jobs.add (e^.jobs) (h^.hostAddr) $+        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)+        ]++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+    atomically' $ writeTVar ctl (Control Connected c)+    info $ msg (val "new control connection: " +++ c)++onCqlEvent :: Event -> Client ()+onCqlEvent x = do+    info $ "client.event" .= show x+    pol <- view policy+    prt <- view (context.settings.portnumber)+    case x of+        StatusEvent Down sa -> do+            liftIO $ onEvent pol $ HostDown (InetAddr $ mapPort prt sa)+        TopologyEvent RemovedNode sa -> do+            let a = InetAddr $ mapPort prt sa+            hmap <- view hostmap+            atomically' $+                modifyTVar hmap (Map.filterWithKey (\h _ -> h^.hostAddr /= a))+            liftIO $ onEvent pol $ HostGone a+        StatusEvent Up sa ->+            startMonitor $ (InetAddr $ mapPort prt sa)+        TopologyEvent NewNode sa -> do+            ctx  <- view context+            hmap <- view hostmap+            ctrl <- readTVarIO' =<< view control+            let a = InetAddr $ mapPort prt sa+            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+        SchemaEvent _ -> return ()+  where+    mapPort i (SockAddrInet _ a)      = SockAddrInet i a+    mapPort i (SockAddrInet6 _ f a b) = SockAddrInet6 i f a b+    mapPort _ unix                    = unix++    discoverPeers' ctx c = discoverPeers ctx c `catchAll` (const $ return [])++    startMonitor a = do+        hmp <- readTVarIO' =<< view hostmap+        case find ((a ==) . view hostAddr) (Map.keys hmp) of+            Just h  -> do+                ctx <- view context+                jbs <- view jobs+                Jobs.add jbs a $ monitor ctx 3000000 60000000 h+            Nothing -> return ()++-----------------------------------------------------------------------------+-- Utilities++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

@@ -0,0 +1,32 @@+-- 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'"

@@ -0,0 +1,49 @@+-- 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.Cluster.Host where++import Control.Lens ((^.), Lens')+import Data.ByteString.Lazy.Char8 (unpack)+import Data.Text (Text)+import Database.CQL.IO.Types (InetAddr)+import System.Logger.Message++-- | 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++-- | Host representation.+data Host = Host+    { _hostAddr   :: !InetAddr+    , _dataCentre :: !Text+    , _rack       :: !Text+    } deriving (Eq, Ord)++-- | The IP address and port number of this host.+hostAddr :: Lens' Host InetAddr+hostAddr f ~(Host a c r) = fmap (\x -> Host x c r) (f a)+{-# 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

@@ -0,0 +1,146 @@+-- 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 Database.CQL.IO.Types (InetAddr)+import System.Random.MWC++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 my+      -- return Nothing.+    , 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)+                    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)+                    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++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,311 @@+-- 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 OverloadedStrings   #-}+{-# LANGUAGE RecordWildCards     #-}+{-# LANGUAGE TemplateHaskell     #-}+{-# LANGUAGE ScopedTypeVariables #-}++module Database.CQL.IO.Connection+    ( Connection+    , resolve+    , ping+    , connect+    , close+    , request+    , startup+    , register+    , query+    , useKeyspace+    , address+    , protocol+    , eventSig++    , ConnectionSettings+    , defSettings+    , connectTimeout+    , sendTimeout+    , responseTimeout+    , maxStreams+    , compression+    , defKeyspace+    ) where++import Control.Applicative+import Control.Concurrent (myThreadId)+import Control.Concurrent.Async+import Control.Concurrent.MVar+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.Builder+import Data.ByteString.Lazy (ByteString)+import Data.Int+import Data.Maybe (isJust, fromMaybe)+import Data.Monoid+import Data.Text.Lazy (fromStrict)+import Data.Unique+import Data.Vector (Vector, (!))+import Database.CQL.Protocol+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 Foreign.C.Types (CInt (..))+import Network+import Network.Socket hiding (connect, close, recv, send)+import Network.Socket.ByteString.Lazy (sendAll)+import System.IO (nativeNewline, Newline (..))+import System.Logger hiding (Settings, close, defSettings, settings)+import System.Timeout++import qualified Data.ByteString            as B+import qualified Data.ByteString.Lazy       as L+import qualified Data.ByteString.Lazy.Char8 as Char8+import qualified Data.Vector                as Vector+import qualified Database.CQL.IO.Sync       as Sync+import qualified Database.CQL.IO.Tickets    as Tickets+import qualified Network.Socket             as S+import qualified Network.Socket.ByteString  as NB++data ConnectionSettings = ConnectionSettings+    { _connectTimeout  :: Milliseconds+    , _sendTimeout     :: Milliseconds+    , _responseTimeout :: Milliseconds+    , _maxStreams      :: Int+    , _compression     :: Compression+    , _defKeyspace     :: Maybe Keyspace+    }++type Streams = Vector (Sync (Header, ByteString))++data Connection = Connection+    { _settings :: ConnectionSettings+    , _address  :: InetAddr+    , _tmanager :: TimeoutManager+    , _protocol :: Version+    , _sock     :: Socket+    , _streams  :: Streams+    , _wLock    :: MVar ()+    , _reader   :: Async ()+    , _tickets  :: Pool+    , _logger   :: Logger+    , _eventSig :: Signal Event+    , _ident    :: Unique+    }++makeLenses ''ConnectionSettings+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 "#" +++ fd (c^.sock)++defSettings :: ConnectionSettings+defSettings =+    ConnectionSettings 5000          -- connect timeout+                       3000          -- send timeout+                       10000         -- response timeout+                       128           -- max streams per connection+                       noCompression -- compression+                       Nothing       -- keyspace++resolve :: String -> PortNumber -> IO InetAddr+resolve host port =+    InetAddr . addrAddress . head <$> 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 $+    bracketOnError (mkSock a) S.close $ \s -> do+        ok <- timeout (ms (t^.connectTimeout) * 1000) (S.connect s (sockAddr a))+        unless (isJust ok) $+            throwM (ConnectTimeout a)+        c <- open s+        validateSettings c+        return c+  where+    open s = do+        tck <- Tickets.pool (t^.maxStreams)+        syn <- Vector.replicateM (t^.maxStreams) Sync.create+        lck <- newMVar ()+        sig <- signal+        rdr <- async (readLoop v g (t^.compression) tck a s syn sig)+        Connection t a m v s syn lck rdr tck g sig <$> newUnique++mkSock :: InetAddr -> IO Socket+mkSock (InetAddr a) = socket (familyOf a) Stream defaultProtocol+  where+    familyOf (SockAddrInet  {..}) = AF_INET+    familyOf (SockAddrInet6 {..}) = AF_INET6+    familyOf (SockAddrUnix  {..}) = AF_UNIX++ping :: MonadIO m => InetAddr -> m Bool+ping a = liftIO $ bracket (mkSock a) S.close $ \s ->+    fromMaybe False <$> timeout 5000000+        ((S.connect s (sockAddr a) >> return True) `catchAll` const (return False))++readLoop :: Version -> Logger -> Compression -> Pool -> InetAddr -> Socket -> Streams -> Signal Event -> IO ()+readLoop v g cmp tck i sck syn s = run `catch` logException `finally` cleanup+  where+    run = forever $ do+        x <- readSocket v g i sck+        case fromStreamId $ streamId (fst x) of+            -1 ->+                case parse cmp x :: 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 = do+        Tickets.close (ConnectionClosed i) tck+        Vector.mapM_ (Sync.close (ConnectionClosed i)) syn+        S.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 $ sendAll (c^.sock) req+        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 -> IO (Header, ByteString)+readSocket v g i s = do+    b <- recv i (if v == V3 then 9 else 8) s+    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 <- recv i (fromIntegral len) s+            trace g $ msg (i +++ val "#" +++ fd s)+                ~~ "stream" .= fromStreamId (streamId h)+                ~~ "type"   .= val "response"+                ~~ msg' (hexdump $ L.take 160 (b <> x))+            return (h, x)++recv :: InetAddr -> Int -> Socket -> IO ByteString+recv _ 0 _ = return L.empty+recv i n c = toLazyByteString <$> go 0 mempty+  where+    go !k !bb = do+        a <- NB.recv c (n - k)+        when (B.null a) $+            throwM (ConnectionClosed i)+        let b = bb <> byteString a+            m = B.length a + k+        if m < n then go m b else return b++-----------------------------------------------------------------------------+-- 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 :: Request () () ())+    res <- request c enc+    (parse cmp res :: Response () () ()) `seq` return ()++register :: MonadIO m => Connection -> [EventType] -> EventHandler -> m ()+register c e f = liftIO $ do+    let req = RqRegister (Register e) :: Request () () ()+    let enc = serialise (c^.protocol) (c^.settings.compression) req+    res <- request c enc+    case parse (c^.settings.compression) res :: 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 :: Request () () ()+    res <- request c (serialise (c^.protocol) noCompression options)+    case parse noCompression res :: 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+        kspace = quoted (fromStrict $ unKeyspace ks)+        req    = RqQuery (Query (QueryString $ "use " <> kspace) params)+    res <- request c (serialise (c^.protocol) cmp req)+    case parse cmp res :: 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++-- logging helpers:++fd :: Socket -> Int32+fd !s = let CInt !n = fdSocket s in n++msg' :: ByteString -> Msg -> Msg+msg' x = msg $ case nativeNewline of+    LF   -> val "\n"   +++ x+    CRLF -> val "\r\n" +++ x

@@ -0,0 +1,63 @@+-- 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

@@ -0,0 +1,68 @@+-- 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 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 -> IO () -> m ()+add (Jobs d) k j = liftIO $ do+    (ok, prev) <- atomically $ do+        m <- readTVar d+        case Map.lookup k m of+            Just Reserved      -> return (False, Nothing)+            Just (Running _ a) -> do+                modifyTVar d (Map.insert k Reserved)+                return (True, Just a)+            Nothing          -> do+                modifyTVar d (Map.insert k Reserved)+                return (True, 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

@@ -0,0 +1,239 @@+-- 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+    , tryWith++    , 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 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 a+with p f = liftIO $ do+    s <- stripe p+    mask $ \restore -> do+        r <- take1 p s+        x <- restore (f (value r)) `catches` handlers p s r+        put p s r id+        return x++tryWith :: MonadIO m => Pool -> (Connection -> IO a) -> m (Maybe a)+tryWith p f = liftIO $ do+    s <- stripe p+    mask $ \restore -> do+        r <- tryTake1 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 Resource+take1 p s = do+    r <- join . atomically $ do+        c <- readTVar (conns s)+        u <- readTVar (inUse s)+        let n       = Seq.length c+        let r :< rr = Seq.viewl $ Seq.unstableSortBy (compare `on` refcnt) c+        if | u < p^.settings.maxConnections -> mkNew p s u+           | n > 0 && refcnt r < p^.maxRefs -> use s r rr+           | otherwise                      -> retry+    case r of+        Left  x -> do+            atomically (modifyTVar (conns s) (|> x))+            return x+        Right x -> return x++tryTake1 :: Pool -> Stripe -> IO (Maybe Resource)+tryTake1 p s = do+    r <- join . atomically $ do+        c <- readTVar (conns s)+        u <- readTVar (inUse s)+        let n       = Seq.length c+        let r :< rr = Seq.viewl $ Seq.unstableSortBy (compare `on` refcnt) c+        if | u < p^.settings.maxConnections -> fmap Just <$> mkNew p s u+           | n > 0 && refcnt r < p^.maxRefs -> fmap Just <$> use s r rr+           | otherwise                      -> return (return Nothing)+    case r of+        Just (Left  x) -> do+            atomically (modifyTVar (conns s) (|> x))+            return (Just x)+        Just (Right x) -> return (Just x)+        Nothing        -> return Nothing++use :: Stripe -> Resource -> Seq Resource -> STM (IO (Either Resource Resource))+use s r rr = do+    writeTVar (conns s) $! rr |> r { refcnt = refcnt r + 1 }+    return (return (Right r))+{-# INLINE use #-}++mkNew :: Pool -> Stripe -> Int -> STM (IO (Either Resource Resource))+mkNew p s u = do+    writeTVar (inUse s) $! u + 1+    return $ Left <$> onException+        (Resource <$> p^.currentTime <*> pure 1 <*> pure 0 <*> p^.createFn)+        (atomically (modifyTVar (inUse s) (subtract 1)))+{-# 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 #-}+

@@ -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 #-}++module Database.CQL.IO.Protocol where++import Control.Exception (throw)+import Data.ByteString.Lazy (ByteString)+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 False s r)++quoted :: LT.Text -> LT.Text+quoted s = "\"" <> LT.replace "\"" "\"\"" s <> "\""

@@ -0,0 +1,156 @@+-- 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   #-}++module Database.CQL.IO.Settings where++import Control.Lens hiding ((<|))+import Data.List.NonEmpty (NonEmpty (..), (<|))+import Data.Time+import Data.Word+import Database.CQL.Protocol+import Database.CQL.IO.Connection+import Database.CQL.IO.Cluster.Policies (Policy, random)+import Database.CQL.IO.Connection as C+import Database.CQL.IO.Pool as P+import Database.CQL.IO.Types (Milliseconds (..))+import Network.Socket (PortNumber (..))++data Settings = Settings+    { _poolSettings  :: PoolSettings+    , _connSettings  :: ConnectionSettings+    , _protoVersion  :: Version+    , _portnumber    :: PortNumber+    , _contacts      :: NonEmpty String+    , _maxWaitQueue  :: Maybe Word64+    , _policyMaker   :: IO Policy+    }++makeLenses ''Settings++-- | Default settings:+--+-- * contact point is \"localhost\" port 9042+--+-- * load-balancing policy is 'random'+--+-- * binary protocol version is 3 (supported by Cassandra >= 2.1.0)+--+-- * 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+--+-- * no compression is applied to frame bodies+--+-- * no default keyspace is used.+defSettings :: Settings+defSettings = Settings+    P.defSettings+    C.defSettings+    V3+    (fromInteger 9042)+    ("localhost" :| [])+    Nothing+    random++-----------------------------------------------------------------------------+-- 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 the maximum length of the wait queue which is used if+-- connection pools of all nodes in a cluster are busy. Once the maximum+-- queue size has been reached, queries will throw a 'HostsBusy' exception+-- immediatly.+setMaxWaitQueue :: Word64 -> Settings -> Settings+setMaxWaitQueue v = set maxWaitQueue (Just 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 "Database.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 = case s^.protoVersion of+    V2 | v < 1 || v > 128   -> error "Database.CQL.IO.Settings: max. streams must be within [1, 128]"+    V3 | v < 1 || v > 32768 -> error "Database.CQL.IO.Settings: max. streams must be within [1, 32768]"+    _                       -> 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)

@@ -0,0 +1,35 @@+-- 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++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,55 @@+-- 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)++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,46 @@+-- 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 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,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/.++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)++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_ (ignore . action) a++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,136 @@+-- 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 GeneralizedNewtypeDeriving #-}+{-# LANGUAGE GADTs                      #-}+{-# LANGUAGE OverloadedStrings          #-}+{-# LANGUAGE StandaloneDeriving         #-}++module Database.CQL.IO.Types where++import Control.Monad.Catch+import Data.IP+import Data.Typeable+import Database.CQL.Protocol (Event, Response, CompressionAlgorithm)+import Network.Socket (SockAddr (..), PortNumber)+import System.Logger.Message++type EventHandler = Event -> IO ()++newtype Milliseconds = Ms { ms :: Int } deriving (Eq, Show, Num)++-----------------------------------------------------------------------------+-- 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++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++ip2inet :: PortNumber -> IP -> InetAddr+ip2inet p (IPv4 a) = InetAddr $ SockAddrInet p (toHostAddress a)+ip2inet p (IPv6 a) = InetAddr $ SockAddrInet6 p 0 (toHostAddress6 a) 0++inet2ip :: InetAddr -> IP+inet2ip (InetAddr (SockAddrInet _ a))      = IPv4 (fromHostAddress a)+inet2ip (InetAddr (SockAddrInet6 _ _ a _)) = IPv6 (fromHostAddress6 a)+inet2ip _                                  = error "inet2Ip: not IP4/IP6 address"++-----------------------------------------------------------------------------+-- InvalidSettings++data InvalidSettings+    = 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++data InternalError = InternalError String+    deriving Typeable++instance Exception InternalError++instance Show InternalError where+    show (InternalError e) = "cql-io: internal error: " ++ show e++-----------------------------------------------------------------------------+-- 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++data Timeout = TimeoutRead !String+    deriving Typeable++instance Exception Timeout++instance Show Timeout where+    show (TimeoutRead e) = "cql-io: read timeout: " ++ e++-----------------------------------------------------------------------------+-- UnexpectedResponse++data UnexpectedResponse where+    UnexpectedResponse  :: UnexpectedResponse+    UnexpectedResponse' :: Show b => Response k a b -> UnexpectedResponse++deriving instance Typeable UnexpectedResponse+instance Exception UnexpectedResponse++instance Show UnexpectedResponse where+    show UnexpectedResponse      = "cql-io: unexpected response"+    show (UnexpectedResponse' r) = "cql-io: unexpected response: " ++ show r++ignore :: IO () -> IO ()+ignore a = catchAll a (const $ return ())+{-# INLINE ignore #-}