cassandra-cql 0.4.0.1 → 0.5.0.0

raw patch · 3 files changed

+420/−169 lines, 3 filesdep +hsloggerdep +resource-poolnew-uploaderPVP ok

version bump matches the API change (PVP)

Dependencies added: hslogger, resource-pool

API changes (from Hackage documentation)

Files

• Database/Cassandra/CQL.hs +408/−163
• cassandra-cql.cabal +11/−5
• tests/example.hs +1/−1

@@ -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 = "<credentials>"-  sendFrame $ Frame [] 0 CREDENTIALS $ encodeElt $ authCredentials auth+  let qt = "<auth_response>"+  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 = "<startup>"     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" "<credentials>")-        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)

@@ -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

@@ -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: