diff --git a/Database/Cassandra/CQL.hs b/Database/Cassandra/CQL.hs --- a/Database/Cassandra/CQL.hs +++ b/Database/Cassandra/CQL.hs @@ -1,6 +1,6 @@ {-# LANGUAGE OverloadedStrings, GeneralizedNewtypeDeriving, ScopedTypeVariables, FlexibleInstances, DeriveDataTypeable, UndecidableInstances, - BangPatterns, OverlappingInstances, DataKinds, GADTs, KindSignatures #-} + BangPatterns, OverlappingInstances, DataKinds, GADTs, KindSignatures, NamedFieldPuns #-} -- | Haskell client for Cassandra's CQL protocol -- -- For examples, take a look at the /tests/ directory in the source archive. @@ -75,8 +75,9 @@ -- -- * 'Rows' for selects that give a list of rows in response. -- --- The functions to use for these query types are 'executeSchema', 'executeWrite' and --- 'executeRows' or 'executeRow' respectively. +-- The functions to use for these query types are 'executeSchema', +-- 'executeWrite', 'executeTrans' and 'executeRows' or 'executeRow' +-- respectively. -- -- The following pattern seems to work very well, especially along with your own 'CasType' -- instances, because it gives you a place to neatly add marshalling details that keeps @@ -96,11 +97,9 @@ -- -- /To do/ -- --- * Add credentials. --- --- * Improve connection pooling. --- -- * Add the ability to easily run queries in parallel. +-- * Add support for batch queries. +-- * Add support for query paging. module Database.Cassandra.CQL ( -- * Initialization @@ -108,6 +107,8 @@ Keyspace(..), Pool, newPool, + newPool', + defaultConfig, -- * Cassandra monad MonadCassandra(..), Cas, @@ -128,6 +129,7 @@ executeWrite, executeRows, executeRow, + executeTrans, -- * Value types Blob(..), Counter(..), @@ -143,27 +145,26 @@ Metadata(..), CType(..), Table(..), - PreparedQueryID(..) + PreparedQueryID(..), + serverStats, + ServerStat(..), + PoolConfig(..), ) where import Control.Applicative -import Control.Concurrent +import Control.Concurrent (threadDelay, forkIO) import Control.Concurrent.STM -import Control.Exception (IOException, SomeException) +import Control.Exception (IOException, SomeException, MaskingState(..), throwIO, getMaskingState, mask) import Control.Monad.CatchIO import Control.Monad.Reader import Control.Monad.State hiding (get, put) -import qualified Control.Monad.State as State -import qualified Control.Monad.Reader import qualified Control.Monad.RWS -import qualified Control.Monad.State -import qualified Control.Monad.State.Strict import qualified Control.Monad.Error import qualified Control.Monad.Writer -import Control.Monad.Trans import Crypto.Hash (hash, Digest, SHA1) import Data.Bits import Data.ByteString (ByteString) +import qualified Data.ByteString.Char8 as C8BS import qualified Data.ByteString as B import qualified Data.ByteString.Lazy as L import Data.Data @@ -174,50 +175,111 @@ import Data.Map (Map) import qualified Data.Map as M import Data.Maybe +import qualified Data.Foldable as F import Data.Monoid (Monoid) -import Data.Sequence (Seq, (|>)) import qualified Data.Sequence as Seq import Data.Serialize hiding (Result) import Data.Set (Set) import qualified Data.Set as S -import Foreign.Storable +import qualified Data.Pool as P import Data.String import Data.Text (Text) import qualified Data.Text as T import qualified Data.Text.Encoding as T import Data.Time.Calendar import Data.Time.Clock -import Data.Typeable +import Data.Typeable () import Data.UUID (UUID) import qualified Data.UUID as UUID import Data.Word import Network.Socket (Socket, HostName, ServiceName, getAddrInfo, socket, AddrInfo(..), connect, sClose, SockAddr(..), SocketType(..), defaultHints) -import Network.Socket.ByteString (send, sendAll, recv) +import Network.Socket.ByteString (sendAll, recv) import Numeric import Unsafe.Coerce ---import qualified Data.ByteString.Char8 as C ---import Text.Hexdump +import Data.Function (on) +import Data.Monoid ((<>)) +import Data.Fixed (Pico) +import System.Timeout (timeout) +import System.Log.Logger (debugM, warningM) +defaultConnectionTimeout :: NominalDiffTime +defaultConnectionTimeout = 10 + +defaultIoTimeout :: NominalDiffTime +defaultIoTimeout = 300 + +defaultSessionCreateTimeout :: NominalDiffTime +defaultSessionCreateTimeout = 20 + +defaultBackoffOnError :: NominalDiffTime +defaultBackoffOnError = 60 + +defaultMaxSessionIdleTime :: NominalDiffTime +defaultMaxSessionIdleTime = 60 + +defaultMaxSessions :: Int +defaultMaxSessions = 20 + + type Server = (HostName, ServiceName) data ActiveSession = ActiveSession { + actServer :: Server, actSocket :: Socket, + actIoTimeout :: NominalDiffTime, actQueryCache :: Map QueryID PreparedQuery } + data Session = Session { - sesServer :: Server, - sesActive :: Maybe ActiveSession + sessServerIndex :: Int, + sessServer :: Server, + sessSocket :: Socket } --- | A handle for the state of the connection pool. -data Pool = Pool { - piKeyspace :: Keyspace, - piSessions :: TVar (Seq Session), - piAuth :: Maybe Authentication + +data ServerState = ServerState { + ssServer :: Server, + ssOrdinal :: Int, + ssSessionCount :: Int, + ssLastError :: Maybe UTCTime, + ssAvailable :: Bool + } deriving (Show, Eq) + +instance Ord ServerState where + compare = + let compareCount = compare `on` ssSessionCount + tieBreaker = compare `on` ssOrdinal + in compareCount <> tieBreaker + + +data PoolConfig = PoolConfig { + piServers :: [Server], + piKeyspace :: Keyspace, + piAuth :: Maybe Authentication, + piSessionCreateTimeout :: NominalDiffTime, + piConnectionTimeout :: NominalDiffTime, + piIoTimeout :: NominalDiffTime, + piBackoffOnError :: NominalDiffTime, + piMaxSessionIdleTime :: NominalDiffTime, + piMaxSessions :: Int } +data PoolState = PoolState { + psConfig :: PoolConfig, + psServers :: TVar (Seq.Seq ServerState) +} + +-- | Exported stats for a server. +data ServerStat = ServerStat { + statServer :: Server, + statSessionCount :: Int, + statAvailable :: Bool +} deriving (Show) + +newtype Pool = Pool (PoolState, P.Pool Session) + class MonadCatchIO m => MonadCassandra m where getCassandraPool :: m Pool @@ -236,61 +298,211 @@ instance (MonadCassandra m, Monoid w) => MonadCassandra (Control.Monad.RWS.RWST r w s m) where getCassandraPool = lift getCassandraPool + + +defaultConfig :: [Server] -> Keyspace -> Maybe Authentication -> PoolConfig +defaultConfig servers keyspace auth = PoolConfig { + piServers = servers, + piKeyspace = keyspace, + piAuth = auth, + piSessionCreateTimeout = defaultSessionCreateTimeout, + piConnectionTimeout = defaultConnectionTimeout, + piIoTimeout = defaultIoTimeout, + piBackoffOnError = defaultBackoffOnError, + piMaxSessionIdleTime = defaultMaxSessionIdleTime, + piMaxSessions = defaultMaxSessions + } + + -- | Construct a pool of Cassandra connections. newPool :: [Server] -> Keyspace -> Maybe Authentication -> IO Pool -newPool svrs ks auth = do - let sessions = map (\svr -> Session svr Nothing) svrs - sess <- atomically $ newTVar (Seq.fromList sessions) - return $ Pool { - piKeyspace = ks, - piSessions = sess, - piAuth = auth - } +newPool servers keyspace auth = newPool' $ defaultConfig servers keyspace auth -takeSession :: Pool -> IO Session -takeSession pool = atomically $ do - sess <- readTVar (piSessions pool) - if Seq.null sess - then retry - else do - let ses = sess `Seq.index` 0 - writeTVar (piSessions pool) (Seq.drop 1 sess) - return ses +newPool' :: PoolConfig -> IO Pool +newPool' config@PoolConfig { piServers, piMaxSessions, piMaxSessionIdleTime } = do + when (null piServers) $ throwIO $ userError "at least one server required" -putSession :: Pool -> Session -> IO () -putSession pool ses = atomically $ modifyTVar (piSessions pool) (|> ses) + -- TODO: Shuffle ordinals + let servers = Seq.fromList $ map (\(s, idx) -> ServerState s idx 0 Nothing True) $ zip piServers [0..] + servers' <- atomically $ newTVar servers -connectIfNeeded :: Pool -> Maybe Authentication -> Session -> IO Session -connectIfNeeded pool auth session = - if isJust (sesActive session) - then return session - else do - let hints = defaultHints { addrSocketType = Stream } - let (host, service) = sesServer session - ais <- getAddrInfo (Just hints) (Just host) (Just service) - mSocket <- foldM (\mSocket ai -> do - case mSocket of - Nothing -> do - s <- socket (addrFamily ai) (addrSocketType ai) (addrProtocol ai) - do - connect s (addrAddress ai) - return (Just s) - `catch` \(exc :: IOException) -> do - sClose s - return Nothing - Just _ -> return mSocket - ) Nothing ais - case mSocket of - Just socket -> do - let active = ActiveSession { - actSocket = socket, - actQueryCache = M.empty - } - active' <- execStateT (introduce pool auth) active - return $ session { sesActive = Just active' } - Nothing -> - return session + let poolState = PoolState { + psConfig = config, + psServers = servers' + } + sessions <- P.createPool (newSession poolState) (destroySession poolState) 1 piMaxSessionIdleTime piMaxSessions + + let pool = Pool (poolState, sessions) + + _ <- forkIO $ poolWatch pool + + return pool + + +poolWatch :: Pool -> IO () +poolWatch (Pool (PoolState { psConfig, psServers }, _)) = do + + let loop = do + cutoff <- (piBackoffOnError psConfig `addUTCTime`) <$> getCurrentTime + + debugM "Database.Cassandra.CQL.poolWatch" "starting" + sleepTil <- atomically $ do + servers <- readTVar psServers + + let availableAgain = filter (((&&) <$> (not . ssAvailable) <*> (maybe False (<= cutoff) . ssLastError)) . snd) (zip [0..] $ F.toList servers) + servers' = F.foldr' (\(idx, server) accum -> Seq.update idx server { ssAvailable = True } accum) servers availableAgain + nextWakeup = F.foldr' (\s nwu -> if not (ssAvailable s) && maybe False (<= nwu) (ssLastError s) + then fromJust . ssLastError $ s + else nwu) cutoff servers' + + writeTVar psServers servers' + + return nextWakeup + + + delay <- (sleepTil `diffUTCTime`) <$> getCurrentTime + + statusDump <- atomically $ readTVar psServers + debugM "Database.Cassandra.CQL.poolWatch" $ "completed : delaying for " ++ show delay ++ ", server states : " ++ show statusDump + + threadDelay (floor $ delay * 1000000) + loop + + loop + + +serverStats :: Pool -> IO [ServerStat] +serverStats (Pool (PoolState { psServers }, _)) = atomically $ do + servers <- readTVar psServers + return $ map (\ServerState { ssServer, ssSessionCount, ssAvailable } -> ServerStat { statServer = ssServer, statSessionCount = ssSessionCount, statAvailable = ssAvailable }) (F.toList servers) + + + + +newSession :: PoolState -> IO Session +newSession poolState@PoolState { psConfig, psServers } = do + debugM "Database.Cassandra.CQL.nextSession" "starting" + + maskingState <- getMaskingState + when (maskingState == Unmasked) $ throwIO $ userError "caller MUST mask async exceptions before attempting to create a session" + + startTime <- getCurrentTime + + let giveUpAt = piSessionCreateTimeout psConfig `addUTCTime` startTime + + loop = do + timeLeft <- (giveUpAt `diffUTCTime`) <$> getCurrentTime + + when (timeLeft <= 0) $ throwIO NoAvailableServers + + debugM "Database.Cassandra.CQL.newSession" "starting attempt to create a new session" + sessionZ <- timeout ((floor $ timeLeft * 1000000) :: Int) makeSession + `catches` [ Handler $ (\(e :: CassandraCommsError) -> do + warningM "Database.Cassandra.CQL.newSession" $ "failed to create a session due to temporary error (will retry) : " ++ show e + return Nothing), + Handler $ (\(e :: SomeException) -> do + warningM "Database.Cassandra.CQL.newSession" $ "failed to create a session due to permanent error (will rethrow) : " ++ show e + throwIO e) + ] + + case sessionZ of + Just session -> return session + Nothing -> loop + + makeSession = bracketOnError chooseServer restoreCount setup + + chooseServer = atomically $ do + servers <- readTVar psServers + + let available = filter (ssAvailable . snd) (zip [0..] $ F.toList servers) + + if null available + then retry + else do + let (idx, best @ ServerState { ssSessionCount }) = minimumBy (compare `on` snd) available + updatedBest = best { ssSessionCount = ssSessionCount + 1 } + + modifyTVar' psServers (Seq.update idx updatedBest) + return (updatedBest, idx) + + + restoreCount (_, idx) = do + now <- getCurrentTime + atomically $ modifyTVar' psServers (Seq.adjust (\s -> s { ssSessionCount = ssSessionCount s - 1, ssLastError = Just now, ssAvailable = False }) idx) + + + setup (ServerState { ssServer }, idx) = setupConnection poolState idx ssServer + + loop + + + +destroySession :: PoolState -> Session -> IO () +destroySession PoolState { psServers } Session { sessSocket, sessServerIndex } = mask $ \restore -> do + atomically $ modifyTVar' psServers (Seq.adjust (\s -> s { ssSessionCount = ssSessionCount s - 1 }) sessServerIndex) + restore (sClose sessSocket) + + + +setupConnection :: PoolState -> Int -> Server -> IO Session +setupConnection PoolState { psConfig } serverIndex server = do + let hints = defaultHints { addrSocketType = Stream } + (host, service) = server + + debugM "Database.Cassandra.CQL.setupConnection" $ "attempting to connect to " ++ host + + startTime <- getCurrentTime + + ais <- getAddrInfo (Just hints) (Just host) (Just service) + + bracketOnError (connectSocket startTime ais) (maybe (return ()) sClose) buildSession + + where connectSocket startTime ais = + foldM (\mSocket ai -> do + case mSocket of + + Nothing -> do + + let tryConnect = do + debugM "Database.Cassandra.CQL.setupConnection" $ "trying address " ++ show ai + + -- No need to use 'bracketOnError' here because we are already masked. + s <- socket (addrFamily ai) (addrSocketType ai) (addrProtocol ai) + mConn <- timeout ((floor $ (piConnectionTimeout psConfig) * 1000000) :: Int) (connect s (addrAddress ai)) `onException` sClose s + case mConn of + Nothing -> sClose s >> return Nothing + Just _ -> return $ Just s + + now <- getCurrentTime + + if now `diffUTCTime` startTime >= piConnectionTimeout psConfig + then return Nothing + else tryConnect `catch` (\ (e :: SomeException) -> do + debugM "Database.Cassandra.CQL.setupConnection" $ "failed to connect to address " ++ show ai ++ " : " ++ show e + return Nothing + ) + + Just _ -> return mSocket + ) Nothing ais + + + buildSession (Just s) = do + debugM "Database.Cassandra.CQL.setupConnection" $ "made connection, now attempting setup for socket " ++ show s + + let active = Session { + sessServerIndex = serverIndex, + sessServer = server, + sessSocket = s + } + + evalStateT (introduce psConfig) (activeSession psConfig active) + + return active + + buildSession Nothing = throwIO NoAvailableServers + + data Flag = Compression | Tracing deriving Show @@ -310,69 +522,80 @@ 3 -> [Compression, Tracing] _ -> error "recvFrame impossible" -data Opcode = ERROR | STARTUP | READY | AUTHENTICATE | CREDENTIALS | OPTIONS | - SUPPORTED | QUERY | RESULT | PREPARE | EXECUTE | REGISTER | EVENT +data Opcode = ERROR | STARTUP | READY | AUTHENTICATE | OPTIONS | SUPPORTED + | QUERY | RESULT | PREPARE | EXECUTE | REGISTER | EVENT | BATCH + | AUTH_CHALLENGE | AUTH_RESPONSE | AUTH_SUCCESS deriving (Eq, Show) instance Serialize Opcode where put op = putWord8 $ case op of - ERROR -> 0x00 - STARTUP -> 0x01 - READY -> 0x02 - AUTHENTICATE -> 0x03 - CREDENTIALS -> 0x04 - OPTIONS -> 0x05 - SUPPORTED -> 0x06 - QUERY -> 0x07 - RESULT -> 0x08 - PREPARE -> 0x09 - EXECUTE -> 0x0a - REGISTER -> 0x0b - EVENT -> 0x0c + ERROR -> 0x00 + STARTUP -> 0x01 + READY -> 0x02 + AUTHENTICATE -> 0x03 + OPTIONS -> 0x05 + SUPPORTED -> 0x06 + QUERY -> 0x07 + RESULT -> 0x08 + PREPARE -> 0x09 + EXECUTE -> 0x0a + REGISTER -> 0x0b + EVENT -> 0x0c + BATCH -> 0x0d + AUTH_CHALLENGE -> 0x0e + AUTH_RESPONSE -> 0x0f + AUTH_SUCCESS -> 0x10 get = do w <- getWord8 case w of - 0x00 -> return $ ERROR - 0x01 -> return $ STARTUP - 0x02 -> return $ READY - 0x03 -> return $ AUTHENTICATE - 0x04 -> return $ CREDENTIALS - 0x05 -> return $ OPTIONS - 0x06 -> return $ SUPPORTED - 0x07 -> return $ QUERY - 0x08 -> return $ RESULT - 0x09 -> return $ PREPARE - 0x0a -> return $ EXECUTE - 0x0b -> return $ REGISTER - 0x0c -> return $ EVENT + 0x00 -> return ERROR + 0x01 -> return STARTUP + 0x02 -> return READY + 0x03 -> return AUTHENTICATE + 0x05 -> return OPTIONS + 0x06 -> return SUPPORTED + 0x07 -> return QUERY + 0x08 -> return RESULT + 0x09 -> return PREPARE + 0x0a -> return EXECUTE + 0x0b -> return REGISTER + 0x0c -> return EVENT + 0x0d -> return BATCH + 0x0e -> return AUTH_CHALLENGE + 0x0f -> return AUTH_RESPONSE + 0x10 -> return AUTH_SUCCESS _ -> fail $ "unknown opcode 0x"++showHex w "" data Frame a = Frame { - frFlags :: [Flag], - frStream :: Int8, - frOpcode :: Opcode, - frBody :: a + _frFlags :: [Flag], + _frStream :: Int8, + frOpcode :: Opcode, + frBody :: a } deriving Show -recvAll :: Socket -> Int -> IO ByteString -recvAll s n = do +timeout' :: NominalDiffTime -> IO a -> IO a +timeout' to = timeout (floor $ to * 1000000) >=> maybe (throwIO CoordinatorTimeout) return + +recvAll :: NominalDiffTime -> Socket -> Int -> IO ByteString +recvAll ioTimeout s n = timeout' ioTimeout $ do bs <- recv s n when (B.null bs) $ throw ShortRead let left = n - B.length bs if left == 0 then return bs else do - bs' <- recvAll s left + bs' <- recvAll ioTimeout s left return (bs `B.append` bs') protocolVersion :: Word8 -protocolVersion = 1 +protocolVersion = 2 recvFrame :: Text -> StateT ActiveSession IO (Frame ByteString) recvFrame qt = do s <- gets actSocket - hdrBs <- liftIO $ recvAll s 8 + ioTimeout <- gets actIoTimeout + hdrBs <- liftIO $ recvAll ioTimeout s 8 case runGet parseHeader hdrBs of Left err -> throw $ LocalProtocolError ("recvFrame: " `T.append` T.pack err) qt Right (ver0, flags, stream, opcode, length) -> do @@ -381,7 +604,7 @@ throw $ LocalProtocolError ("unexpected version " `T.append` T.pack (show ver)) qt body <- if length == 0 then pure B.empty - else liftIO $ recvAll s (fromIntegral length) + else liftIO $ recvAll ioTimeout s (fromIntegral length) --liftIO $ putStrLn $ hexdump 0 (C.unpack $ hdrBs `B.append` body) return $ Frame flags stream opcode body `catch` \exc -> throw $ CassandraIOException exc @@ -395,7 +618,7 @@ return (ver, flags, stream, opcode, length) sendFrame :: Frame ByteString -> StateT ActiveSession IO () -sendFrame fr@(Frame flags stream opcode body) = do +sendFrame (Frame flags stream opcode body) = do let bs = runPut $ do putWord8 protocolVersion putFlags flags @@ -405,7 +628,8 @@ putByteString body --liftIO $ putStrLn $ hexdump 0 (C.unpack bs) s <- gets actSocket - liftIO $ sendAll s bs + ioTimeout <- gets actIoTimeout + liftIO $ timeout' ioTimeout $ sendAll s bs `catch` \exc -> throw $ CassandraIOException exc class ProtoElt a where @@ -519,6 +743,8 @@ | ValueMarshallingException TransportDirection Text Text | CassandraIOException IOException | ShortRead + | NoAvailableServers + | CoordinatorTimeout deriving (Show, Typeable) instance Exception CassandraCommsError @@ -559,38 +785,38 @@ _ -> fail $ "unknown error code 0x"++showHex code "" -type User = T.Text -type Password = T.Text -data Authentication = PasswordAuthenticator User Password -type Credentials = [(Text, Text)] +type UserId = String +type Password = String +data Authentication = PasswordAuthenticator UserId Password +type Credentials = Long ByteString authCredentials :: Authentication -> Credentials -authCredentials (PasswordAuthenticator user password) = [("username", user), ("password", password)] +authCredentials (PasswordAuthenticator user password) = Long $ C8BS.pack $ "\0" ++ user ++ "\0" ++ password authenticate :: Authentication -> StateT ActiveSession IO () authenticate auth = do - let qt = "" - sendFrame $ Frame [] 0 CREDENTIALS $ encodeElt $ authCredentials auth + let qt = "" + sendFrame $ Frame [] 0 AUTH_RESPONSE $ encodeElt $ authCredentials auth fr2 <- recvFrame qt case frOpcode fr2 of - READY -> return () + AUTH_SUCCESS -> return () ERROR -> throwError qt (frBody fr2) op -> throw $ LocalProtocolError ("introduce: unexpected opcode " `T.append` T.pack (show op)) qt -introduce :: Pool -> Maybe Authentication -> StateT ActiveSession IO () -introduce pool auth = do +introduce :: PoolConfig -> StateT ActiveSession IO () +introduce PoolConfig { piKeyspace, piAuth } = do let qt = "" sendFrame $ Frame [] 0 STARTUP $ encodeElt $ ([("CQL_VERSION", "3.0.0")] :: [(Text, Text)]) fr <- recvFrame qt case frOpcode fr of AUTHENTICATE -> maybe (throw $ MissingAuthenticationError "introduce: server expects auth but none provided" "") - authenticate auth + authenticate piAuth READY -> return () ERROR -> throwError qt (frBody fr) op -> throw $ LocalProtocolError ("introduce: unexpected opcode " `T.append` T.pack (show op)) qt - let Keyspace ksName = piKeyspace pool + let Keyspace ksName = piKeyspace let q = query $ "USE " `T.append` ksName :: Query Rows () () res <- executeInternal q () ONE case res of @@ -599,32 +825,29 @@ withSession :: MonadCassandra m => (Pool -> StateT ActiveSession IO a) -> m a withSession code = do - pool <- getCassandraPool - let auth = piAuth pool - mA <- liftIO $ do - session <- connectIfNeeded pool auth =<< takeSession pool - case sesActive session of - Just active -> do - (a, active') <- runStateT (code pool) active - putSession pool $ session { sesActive = Just active' } - return (Just a) - `catches` [ - -- Close the session if we get any IOException - Handler $ \(exc :: IOException) -> do - sClose (actSocket active) - putSession pool $ session { sesActive = Nothing } - throw exc, - Handler $ \(exc :: SomeException) -> do - putSession pool session - throw exc - ] - Nothing -> do - putSession pool session - return Nothing - case mA of - Just a -> return a - Nothing -> withSession code -- Try again until we succeed + pool@(Pool (PoolState { psConfig }, sessions)) <- getCassandraPool + liftIO $ mask $ \restore -> do + (session, local') <- P.takeResource sessions + + a <- restore (evalStateT (code pool) (activeSession psConfig session)) + `catches` + [ Handler $ \(exc :: CassandraException) -> P.putResource local' session >> throwIO exc, + Handler $ \(exc :: SomeException) -> P.destroyResource sessions local' session >> throwIO exc + ] + + P.putResource local' session + + return a + +activeSession :: PoolConfig -> Session -> ActiveSession +activeSession poolConfig session = ActiveSession { + actServer = sessServer session, + actSocket = sessSocket session, + actIoTimeout = piIoTimeout poolConfig, + actQueryCache = M.empty + } + -- | The name of a Cassandra keyspace. See the Cassandra documentation for more -- information. newtype Keyspace = Keyspace Text @@ -844,10 +1067,10 @@ instance CasType UTCTime where getCas = do ms <- getWord64be - let difft = realToFrac $ (fromIntegral ms :: Double) / 1000 + let difft = realToFrac $ (fromIntegral ms :: Pico) / 1000 return $ addUTCTime difft epoch putCas utc = do - let seconds = realToFrac $ diffUTCTime utc epoch :: Double + let seconds = realToFrac $ diffUTCTime utc epoch :: Pico ms = round (seconds * 1000) :: Word64 putWord64be ms casType _ = CTimestamp @@ -1308,7 +1531,7 @@ (a, b, c, d, e, f, g, h, i, j, k, l, m, n, o, p, q, r, s, t)) <$> decodeNested 0 vs -- | Cassandra consistency level. See the Cassandra documentation for an explanation. -data Consistency = ANY | ONE | TWO | THREE | QUORUM | ALL | LOCAL_QUORUM | EACH_QUORUM +data Consistency = ANY | ONE | TWO | THREE | QUORUM | ALL | LOCAL_QUORUM | EACH_QUORUM | SERIAL | LOCAL_SERIAL | LOCAL_ONE deriving (Eq, Ord, Show, Bounded, Enum) instance ProtoElt Consistency where @@ -1321,6 +1544,9 @@ ALL -> 0x0005 LOCAL_QUORUM -> 0x0006 EACH_QUORUM -> 0x0007 + SERIAL -> 0x0008 + LOCAL_SERIAL -> 0x0009 + LOCAL_ONE -> 0x000A getElt = do w <- getWord16be case w of @@ -1332,6 +1558,9 @@ 0x0005 -> pure ALL 0x0006 -> pure LOCAL_QUORUM 0x0007 -> pure EACH_QUORUM + 0x0008 -> pure SERIAL + 0x0009 -> pure LOCAL_SERIAL + 0x000A -> pure LOCAL_ONE _ -> fail $ "unknown consistency value 0x"++showHex w "" -- | A low-level function in case you need some rarely-used capabilities. @@ -1342,12 +1571,14 @@ executeInternal :: CasValues values => Query style any_i any_o -> values -> Consistency -> StateT ActiveSession IO (Result [Maybe ByteString]) executeInternal query i cons = do - pq@(PreparedQuery pqid queryMeta) <- prepare query + (PreparedQuery pqid queryMeta) <- prepare query values <- case encodeValues i (metadataTypes queryMeta) of Left err -> throw $ ValueMarshallingException TransportSending (T.pack $ show err) (queryText query) Right values -> return values sendFrame $ Frame [] 0 EXECUTE $ runPut $ do putElt pqid + putElt cons + putWord8 0x01 putWord16be (fromIntegral $ length values) forM_ values $ \mValue -> case mValue of @@ -1356,7 +1587,6 @@ let enc = encodeCas value putWord32be (fromIntegral $ B.length enc) putByteString enc - putElt cons fr <- recvFrame (queryText query) case frOpcode fr of RESULT -> decodeEltM "RESULT" (frBody fr) (queryText query) @@ -1365,7 +1595,7 @@ -- | Execute a query that returns rows. executeRows :: (MonadCassandra m, CasValues i, CasValues o) => - Consistency + Consistency -- ^ Consistency level of the operation -> Query Rows i o -- ^ CQL query to execute -> i -- ^ Input values substituted in the query -> m [o] @@ -1375,10 +1605,25 @@ RowsResult meta rows -> decodeRows q meta rows _ -> throw $ LocalProtocolError ("expected Rows, but got " `T.append` T.pack (show res)) (queryText q) +-- | Execute a lightweight transaction. The consistency level is implicit and +-- is SERIAL. +executeTrans :: (MonadCassandra m, CasValues i) => + Query Write i () -- ^ CQL query to execute + -> i -- ^ Input values substituted in the query + -> m Bool +executeTrans q i = do + res <- executeRaw q i SERIAL + case res of + RowsResult _ ((el:row):rows) -> + case decodeCas $ fromJust el of + Left s -> error $ "executeTrans: decode result failure=" ++ s + Right b -> return b + _ -> throw $ LocalProtocolError ("expected Rows, but got " `T.append` T.pack (show res)) (queryText q) + -- | Helper for 'executeRows' useful in situations where you are only expecting one row -- to be returned. executeRow :: (MonadCassandra m, CasValues i, CasValues o) => - Consistency + Consistency -- ^ Consistency level of the operation -> Query Rows i o -- ^ CQL query to execute -> i -- ^ Input values substituted in the query -> m (Maybe o) @@ -1397,7 +1642,7 @@ -- | Execute a write operation that returns void. executeWrite :: (MonadCassandra m, CasValues i) => - Consistency + Consistency -- ^ Consistency level of the operation -> Query Write i () -- ^ CQL query to execute -> i -- ^ Input values substituted in the query -> m () @@ -1409,7 +1654,7 @@ -- | Execute a schema change, such as creating or dropping a table. executeSchema :: (MonadCassandra m, CasValues i) => - Consistency + Consistency -- ^ Consistency level of the operation -> Query Schema i () -- ^ CQL query to execute -> i -- ^ Input values substituted in the query -> m (Change, Keyspace, Table) diff --git a/cassandra-cql.cabal b/cassandra-cql.cabal --- a/cassandra-cql.cabal +++ b/cassandra-cql.cabal @@ -1,16 +1,19 @@ name: cassandra-cql -version: 0.4.0.1 +version: 0.5.0.0 synopsis: Haskell client for Cassandra's CQL protocol description: Haskell client for Cassandra's CQL protocol + . + Reivision history: . - Revision history: 0.3.0.1 Fix socket issue on Mac. - 0.4.0.1 Add PasswordAuthenticator (thanks Curtis Carter) & accept ghc-7.8 + * 0.5.0.0 Upgrade to CQL Binary Protocol v2. Support Cassandra Lightweight transactions. + * 0.4.0.1 Add PasswordAuthenticator (thanks Curtis Carter) & accept ghc-7.8 + * 0.3.0.1 Fix socket issue on Mac. license: BSD3 license-file: LICENSE author: Stephen Blackheath maintainer: http://blacksapphire.com/antispam/ -copyright: (c) Stephen Blackheath 2013-2014 +copyright: (c) Stephen Blackheath 2013-2015 category: Database build-type: Simple stability: alpha @@ -45,7 +48,10 @@ stm >= 2.4.0, uuid >= 1.2.0, time >= 1.4.0.0, - Decimal >= 0.3.0 + Decimal >= 0.3.0, + resource-pool >= 0.2.3, + hslogger >= 1.2 + ghc-options: -Wall -fno-warn-name-shadowing -fno-warn-unused-matches -fno-warn-missing-signatures -fno-warn-orphans -fno-warn-unused-imports -fno-warn-unused-binds diff --git a/tests/example.hs b/tests/example.hs --- a/tests/example.hs +++ b/tests/example.hs @@ -33,7 +33,7 @@ _ -> throw exc main = do - --let auth = Just (PasswordAuthenticator "cassandra" "cassandra") + -- let auth = Just (PasswordAuthenticator "cassandra" "cassandra") let auth = Nothing {- Assuming a 'test' keyspace already exists. Here's some CQL to create it: