cassy 0.2.0.3 → 0.3.2

raw patch · 6 files changed

+432/−391 lines, 6 filesdep +binarydep +resource-poolPVP ok

version bump matches the API change (PVP)

Dependencies added: binary, resource-pool

API changes (from Hackage documentation)

Files

• cassy.cabal +4/−1
• src/Database/Cassandra/Basic.hs +177/−134
• src/Database/Cassandra/JSON.hs +140/−130
• src/Database/Cassandra/Pack.hs +17/−0
• src/Database/Cassandra/Pool.hs +49/−121
• src/Database/Cassandra/Types.hs +45/−5

@@ -1,5 +1,5 @@ Name:                cassy-Version:             0.2.0.3+Version:             0.3.2 Synopsis:            A high level driver for the Cassandra datastore License:             BSD3 License-file:        LICENSE@@ -52,11 +52,13 @@     Database.Cassandra.JSON     Database.Cassandra.Pool     Database.Cassandra.Types+    Database.Cassandra.Pack      -- Packages needed in order to build this package.   Build-depends:       base >= 4 && < 5     , bytestring+    , binary     , containers     , network     , time@@ -68,6 +70,7 @@     , aeson     , Thrift >= 0.6     , cassandra-thrift >= 0.8+    , resource-pool    Extensions:       TypeSynonymInstances

@@ -1,124 +1,175 @@ {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE PatternGuards, NamedFieldPuns, RecordWildCards #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE FlexibleContexts #-}   module Database.Cassandra.Basic  -(-  -- * Basic Types-    ColumnFamily(..)-  , Key(..)-  , ColumnName(..)-  , Value(..)-  , Column(..)-  , col-  , Row(..)-  , ConsistencyLevel(..)+    ( -  -- * Filtering-  , Selector(..)-  , Order(..)-  , KeySelector(..)-  , KeyRangeType(..)+    -- * Connection+      CPool+    , Server+    , defServer+    , defServers+    , KeySpace+    , createCassandraPool+    +    -- * MonadCassandra Typeclass+    , MonadCassandra (..)+    , Cas (..)+    , runCas -  -- * Exceptions-  , CassandraException(..)+    -- * Cassandra Operations+    , getCol+    , get+    , getMulti+    , insert+    , delete -  -- * Connection-  , CPool-  , Server(..)-  , defServer-  , defServers-  , KeySpace(..)-  , createCassandraPool+    -- * Filtering+    , Selector(..)+    , Order(..)+    , KeySelector(..)+    , KeyRangeType(..) -  -- * Cassandra Operations-  , getCol-  , get-  , getMulti-  , insert-  , delete+    -- * Exceptions+    , CassandraException(..) -  -- * Utility-  , getTime-  , throwing-) where+    -- * Utility+    , getTime+    , throwing+    , wrapException +    -- * Basic Types+    , ColumnFamily+    , Key+    , ColumnName+    , Value+    , Column(..)+    , col+    , Row+    , ConsistencyLevel(..) +    -- * Helpers+    , CKey (..)+    , packLong+    ) where+++-------------------------------------------------------------------------------+import           Control.Applicative import           Control.Exception import           Control.Monad-import           Data.ByteString.Lazy (ByteString)+import           Control.Monad.IO.Class+import           Control.Monad.Trans.Reader+import           Data.ByteString.Lazy                       (ByteString)+import           Data.Map                                   (Map)+import qualified Data.Map                                   as M+import           Data.Maybe                                 (mapMaybe) import qualified Database.Cassandra.Thrift.Cassandra_Client as C-import qualified Database.Cassandra.Thrift.Cassandra_Types as T-import           Database.Cassandra.Thrift.Cassandra_Types -                  (ConsistencyLevel(..))-import           Data.Map (Map)-import qualified Data.Map as M-import           Data.Maybe (mapMaybe)-import           Network-import           Prelude hiding (catch)-+import           Database.Cassandra.Thrift.Cassandra_Types  (ConsistencyLevel (..))+import qualified Database.Cassandra.Thrift.Cassandra_Types  as T+import           Prelude                                    hiding (catch)+-------------------------------------------------------------------------------+import           Database.Cassandra.Pack import           Database.Cassandra.Pool import           Database.Cassandra.Types+-------------------------------------------------------------------------------  -test = do-  pool <- createCassandraPool [("127.0.0.1", PortNumber 9160)] 3 300 "Keyspace1"-  withPool pool $ \ Cassandra{..} -> do-    let cp = T.ColumnParent (Just "CF1") Nothing-    let sr = Just $ T.SliceRange (Just "") (Just "") (Just False) (Just 100)-    let ks = Just ["eben"]-    let sp = T.SlicePredicate Nothing sr-    C.get_slice (cProto, cProto) "darak" cp sp ONE-  get pool "darak" "CF1" All ONE-  getCol pool "CF1" "darak" "eben" ONE-  insert pool "CF1" "test1" ONE [col "col1" "val1", col "col2" "val2"] -  get pool "test1" "CF1" All ONE >>= putStrLn . show-  delete pool "test1" "CF1" (ColNames ["col2"]) ONE-  get pool "test1" "CF1" (Range Nothing Nothing Reversed 1) ONE >>= putStrLn . show+-- test = do+--   pool <- createCassandraPool [("127.0.0.1", 9160)] 3 300 "Keyspace1"+--   withResource pool $ \ Cassandra{..} -> do+--     let cp = T.ColumnParent (Just "CF1") Nothing+--     let sr = Just $ T.SliceRange (Just "") (Just "") (Just False) (Just 100)+--     let ks = Just ["eben"]+--     let sp = T.SlicePredicate Nothing sr+--     C.get_slice (cProto, cProto) "darak" cp sp ONE+--   flip runCas pool $ do+--     get "CF1" "CF1" All ONE+--     getCol "CF1" "darak" "eben" ONE+--     insert "CF1" "test1" ONE [col "col1" "val1", col "col2" "val2"] +--     get  "CF1" "CF1" All ONE >>= liftIO . print+--     get  "CF1" "not here" All ONE >>= liftIO . print+--     delete  "CF1" "CF1" (ColNames ["col2"]) ONE+--     get  "CF1" "CF1" (Range Nothing Nothing Reversed 1) ONE >>= liftIO . putStrLn . show  +-------------------------------------------------------------------------------+-- | All Cassy operations are designed to run inside 'MonadCassandra'+-- context.+--+-- We provide a default concrete 'Cas' datatype, but you can simply+-- make your own application monads an instance of 'MonadCassandra'+-- for conveniently using all operations of this package.+--+-- Please keep in mind that all Cassandra operations may raise+-- 'CassandraException's at any point in time.+class (MonadIO m) => MonadCassandra m where+    getCassandraPool :: m CPool+++-------------------------------------------------------------------------------+withCassandraPool :: MonadCassandra m => (Cassandra -> IO b) -> m b+withCassandraPool f = do+  p <- getCassandraPool+  liftIO $ withResource p f+++-------------------------------------------------------------------------------+newtype Cas a = Cas { unCas :: ReaderT CPool IO a }+    deriving (Functor,Applicative,Monad,MonadIO)+++-------------------------------------------------------------------------------+-- | Main running function when using the ad-hoc Cas monad. Just write+-- your cassandra actions within the 'Cas' monad and supply them with+-- a 'CPool' to execute.+runCas :: Cas a -> CPool -> IO a+runCas f p = runReaderT (unCas f) p +++-------------------------------------------------------------------------------+instance MonadCassandra Cas where+    getCassandraPool = Cas ask++ --------------------------------------------------------------------------------- | Get a single key-column value+-- | Get a single key-column value. getCol -  :: CPool-  -> ColumnFamily +  :: (MonadCassandra m)+  => ColumnFamily    -> Key    -- ^ Row key   -> ColumnName   -- ^ Column/SuperColumn name   -> ConsistencyLevel    -- ^ Read quorum-  -> IO (Either CassandraException Column)-getCol p cf k cn cl = do-  c <- get p cf k (ColNames [cn]) cl-  case c of-    Left e -> return $ Left e-    Right [] -> return $ Left NotFoundException-    Right (x:_) -> return $ Right x+  -> m (Maybe Column)+getCol cf k cn cl = do+    res <- get cf k (ColNames [cn]) cl+    case res of+      [] -> return Nothing+      x:_ -> return $ Just x   ------------------------------------------------------------------------------ -- | An arbitrary get operation - slice with 'Selector' get -  :: CPool-  -> ColumnFamily +  :: (MonadCassandra m)+  => ColumnFamily    -- ^ in ColumnFamily   -> Key    -- ^ Row key to get   -> Selector    -- ^ Slice columns with selector   -> ConsistencyLevel -  -> IO (Either CassandraException Row)-get p cf k s cl = withPool p $ \ Cassandra{..} -> do+  -> m Row+get cf k s cl = withCassandraPool $ \ Cassandra{..} -> do   res <- wrapException $ C.get_slice (cProto, cProto) k cp (mkPredicate s) cl-  case res of-    Left e -> return $ Left e-    Right xs -> return $ do-      cs <- mapM castColumn xs-      case cs of-        [] -> Left NotFoundException-        _ -> Right $ cs+  throwing . return $ mapM castColumn res   where     cp = T.ColumnParent (Just cf) Nothing @@ -126,36 +177,27 @@ ------------------------------------------------------------------------------ -- | Do multiple 'get's in one DB hit getMulti -  :: CPool-  -> ColumnFamily +  :: (MonadCassandra m)+  => ColumnFamily    -> KeySelector   -- ^ A selection of rows to fetch in one hit   -> Selector    -- ^ Subject to column selector conditions   -> ConsistencyLevel -  -> IO (Either CassandraException (Map ByteString Row))+  -> m (Map ByteString Row)   -- ^ A Map from Row keys to 'Row's is returned-getMulti p cf ks s cl = withPool p $ \ Cassandra{..} -> do+getMulti cf ks s cl = withCassandraPool $ \ Cassandra{..} -> do   case ks of     Keys xs -> do       res <- wrapException $ C.multiget_slice (cProto, cProto) xs cp (mkPredicate s) cl-      case res of-        Left e -> return $ Left e-        Right m -> return . Right $ M.mapMaybe f m+      return $ M.mapMaybe f res     KeyRange {} -> do       res <- wrapException $          C.get_range_slices (cProto, cProto) cp (mkPredicate s) (mkKeyRange ks) cl-      case res of-        Left e -> return $ Left e-        Right res' -> return . Right $ collectKeySlices res'+      return $ collectKeySlices res   where     collectKeySlices :: [T.KeySlice] -> Map ByteString Row-    collectKeySlices ks = M.fromList $ mapMaybe collectKeySlice ks-      where -        f (k, Just x) = True-        f _ = False-        g (k, Just x) = (k,x)-+    collectKeySlices xs = M.fromList $ mapMaybe collectKeySlice xs      collectKeySlice (T.KeySlice (Just k) (Just xs)) =        case mapM castColumn xs of@@ -173,46 +215,44 @@ ------------------------------------------------------------------------------ -- | Insert an entire row into the db. ----- This will do as many round-trips as necessary to insert the full row.+-- This will do as many round-trips as necessary to insert the full+-- row. Please keep in mind that each column and each column of each+-- super-column is sent to the server one by one. insert-  :: CPool-  -> ColumnFamily+  :: (MonadCassandra m)+  => ColumnFamily   -> Key   -> ConsistencyLevel   -> Row-  -> IO (Either CassandraException ())-insert p cf k cl row = withPool p $ \ Cassandra{..} -> do-  let iOne c = do-                c' <- mkThriftCol c-                wrapException $ C.insert (cProto, cProto) k cp c' cl -  res <- sequenceE $ map iOne row-  return $ res >> return ()-  where-    cp = T.ColumnParent (Just cf) Nothing+  -> m ()+insert cf k cl row = withCassandraPool $ \ Cassandra{..} -> do+  let insCol cp c = do+        c' <- mkThriftCol c+        C.insert (cProto, cProto) k cp c' cl+  forM_ row $ \ c -> do+    case c of+      Column{} -> do+        let cp = T.ColumnParent (Just cf) Nothing+        insCol cp c+      SuperColumn cn cols -> do+        let cp = T.ColumnParent (Just cf) (Just cn)+        mapM_ (insCol cp) cols      -sequenceE :: [IO (Either a b)] -> IO (Either a [b])-sequenceE [] = return (return [])-sequenceE (a:as) = do-  r1 <- a-  rr <- sequenceE as-  return $ liftM2 (:) r1 rr-- ------------------------------------------------------------------------------ -- | Delete an entire row, specific columns or a specific sub-set of columns -- within a SuperColumn. delete -  ::  CPool-  -> ColumnFamily+  ::  (MonadCassandra m)+  => ColumnFamily   -- ^ In 'ColumnFamily'   -> Key   -- ^ Key to be deleted   -> Selector   -- ^ Columns to be deleted   -> ConsistencyLevel-  -> IO (Either CassandraException ())-delete p cf k s cl = withPool p $ \ Cassandra {..} -> do+  -> m ()+delete cf k s cl = withCassandraPool $ \ Cassandra {..} -> do   now <- getTime   wrapException $ case s of     All -> C.remove (cProto, cProto) k cpAll now cl@@ -220,6 +260,7 @@       C.remove (cProto, cProto) k (cpCol c) now cl     SupNames sn cs -> forM_ cs $ \c -> do       C.remove (cProto, cProto) k (cpSCol sn c) now cl+    Range _ _ _ _ -> error "delete: Range delete not implemented"   where     -- wipe out the entire row     cpAll = T.ColumnPath (Just cf) Nothing Nothing@@ -233,24 +274,26 @@   --------------------------------------------------------------------------------- | Wrap exceptions into an explicit type-wrapException :: IO a -> IO (Either CassandraException a)-wrapException a = -  (a >>= return . Right)-  `catch` (\(T.NotFoundException) -> return $ Left NotFoundException)-  `catch` (\(T.InvalidRequestException e) -> -            return . Left . InvalidRequestException $ maybe "" id e)-  `catch` (\T.UnavailableException -> return $ Left UnavailableException)-  `catch` (\T.TimedOutException -> return $ Left TimedOutException)-  `catch` (\(T.AuthenticationException e) -> -            return . Left . AuthenticationException $ maybe "" id e)-  `catch` (\(T.AuthorizationException e) -> -            return . Left . AuthorizationException $ maybe "" id e)-  `catch` (\T.SchemaDisagreementException -> return $ Left SchemaDisagreementException)+-- | Wrap exceptions of the underlying thrift library into the+-- exception types defined here.+wrapException :: IO a -> IO a+wrapException a = f+    where +      f = a+        `catch` (\ (T.NotFoundException) -> throw NotFoundException)+        `catch` (\ (T.InvalidRequestException e) -> +                  throw . InvalidRequestException $ maybe "" id e)+        `catch` (\ T.UnavailableException -> throw UnavailableException)+        `catch` (\ T.TimedOutException -> throw TimedOutException)+        `catch` (\ (T.AuthenticationException e) -> +                  throw . AuthenticationException $ maybe "" id e)+        `catch` (\ (T.AuthorizationException e) -> +                  throw . AuthorizationException $ maybe "" id e)+        `catch` (\ T.SchemaDisagreementException -> throw SchemaDisagreementException)   ---------------------------------------------------------------------------------- | Make exceptions implicit+-- | Make exceptions implicit. throwing :: IO (Either CassandraException a) -> IO a throwing f = do   res <- f

@@ -4,53 +4,66 @@              TypeSynonymInstances,               ScopedTypeVariables,              FlexibleInstances,+             ExistentialQuantification,              RecordWildCards #-}  {-|     A higher level module for working with Cassandra.     -    Row and Column keys can be any string-like type implementing the-    CKey typeclass. You can add your own types by defining new instances +    All row and column keys are standardized to be of strict types.+    Row keys are Text, while Column keys are ByteString. This might change+    in the future and we may revert to entirely ByteString keys.++         Serialization and de-serialization of Column values are taken care of     automatically using the ToJSON and FromJSON typeclasses.-    -    Also, this module currently attempts to reduce verbosity by-    throwing errors instead of returning Either types as in the-    'Database.Cassandra.Basic' module.  -}  module Database.Cassandra.JSON -( +    (    -  -- * Connection-    CPool-  , Server(..)-  , defServer-  , defServers-  , KeySpace(..)-  , createCassandraPool---  -- * Necessary Types-  , CKey(..)-  , ModifyOperation(..)-  , ColumnFamily(..)-  , ConsistencyLevel(..)-  , CassandraException(..)+    -- * Connection+      CPool+    , Server (..)+    , defServer+    , defServers+    , KeySpace (..)+    , createCassandraPool -  -- * Cassandra Operations-  , get-  , getCol  -  , insertCol-  , modify-  , modify_-  , delete+    -- * MonadCassandra Typeclass+    , MonadCassandra (..)+    , Cas (..)+    , runCas +    -- * Cassandra Operations+    , get+    , getCol  +    , getMulti+    , insertCol+    , modify+    , modify_+    , delete -) where+    -- * Necessary Types+    , RowKey+    , ColumnName+    , ModifyOperation (..)+    , ColumnFamily (..)+    , ConsistencyLevel (..)+    , CassandraException (..)+    , Selector (..)+    , KeySelector (..)+    , KeyRangeType (..)+    +    -- * Helpers+    , CKey (..)+    , packLong+    +    ) where +------------------------------------------------------------------------------- import           Control.Exception import           Control.Monad import           Data.Aeson                 as A@@ -59,57 +72,30 @@ import qualified Data.ByteString.Char8      as B import           Data.ByteString.Lazy.Char8 (ByteString) import qualified Data.ByteString.Lazy.Char8 as LB+import           Data.Int                   (Int32, Int64) import           Data.Map                   (Map) import qualified Data.Map                   as M import qualified Data.Text                  as T+import           Data.Text                  (Text) import qualified Data.Text.Encoding         as T import qualified Data.Text.Lazy             as LT import qualified Data.Text.Lazy.Encoding    as LT import           Network import           Prelude                    hiding (catch)--import           Database.Cassandra.Basic   hiding (get, getCol, delete)+-------------------------------------------------------------------------------+import           Database.Cassandra.Basic   hiding (get, getCol, delete, KeySelector (..), +                                                    getMulti) import qualified Database.Cassandra.Basic   as CB import           Database.Cassandra.Pool-import           Database.Cassandra.Types--------------------------------------------------------------------------------------- CKey Typeclass+import           Database.Cassandra.Types   hiding (KeySelector (..), ColumnName)+import           Database.Cassandra.Pack -------------------------------------------------------------------------------  ---------------------------------------------------------------------------------- | A typeclass to enable using any string-like type for row and column keys-class CKey a where-  toBS    :: a -> ByteString-  fromBS  :: ByteString -> a -instance CKey String where-    toBS = LB.pack-    fromBS = LB.unpack--instance CKey LT.Text where-    toBS = LT.encodeUtf8-    fromBS = LT.decodeUtf8--instance CKey T.Text where-    toBS = toBS . LT.fromChunks . return-    fromBS = T.concat . LT.toChunks . fromBS--instance CKey B.ByteString where-    toBS = LB.fromChunks . return-    fromBS = B.concat . LB.toChunks . fromBS--instance CKey ByteString where-    toBS = id-    fromBS = id--- ------------------------------------------------------------------------------- -- | Convert regular column to a key-value pair-col2val :: (CKey colKey, FromJSON a) => Column -> (colKey, a)+col2val :: (FromJSON a) => Column -> (ColumnName, a) col2val (Column nm val _ _) =  (fromBS nm, maybe err id $ unMarshallJSON' val)     where err = error "Value can't be parsed from JSON." col2val _ = error "col2val is not implemented for SuperColumns"@@ -125,6 +111,10 @@   deriving (Eq,Show,Ord,Read)  +-------------------------------------------------------------------------------+type RowKey = Text++ ------------------------------------------------------------------------------ -- | A modify function that will fetch a specific column, apply modification -- function on it and save results back to Cassandra. @@ -138,42 +128,38 @@ -- This method may throw a 'CassandraException' for all exceptions other than -- 'NotFoundException'. modify-  :: (CKey rowKey, CKey colKey, ToJSON a, FromJSON a)-  => CPool-  -> ColumnFamily-  -> rowKey-  -> colKey+  :: (MonadCassandra m, ToJSON a, FromJSON a)+  => ColumnFamily+  -> RowKey+  -> ColumnName   -> ConsistencyLevel   -- ^ Read quorum   -> ConsistencyLevel   -- ^ Write quorum-  -> (Maybe a -> IO (ModifyOperation a, b))+  -> (Maybe a -> m (ModifyOperation a, b))   -- ^ Modification function. Called with 'Just' the value if present,   -- 'Nothing' otherwise.-  -> IO b+  -> m b   -- ^ Return the decided 'ModifyOperation' and its execution outcome-modify cp cf k cn rcl wcl f = +modify cf k cn rcl wcl f =    let     k' = toBS k     cn' = toBS cn     execF prev = do       (fres, b) <- f prev-      dbres <- case fres of+      case fres of         (Update a) ->-          insert cp cf k' wcl [col cn' (marshallJSON' a)]+          insert cf k' wcl [col cn' (marshallJSON' a)]         (Delete) ->-          CB.delete cp cf k' (ColNames [cn']) wcl-        (DoNothing) -> return $ Right ()-      case dbres of-        Left e -> throw e -- Modify op returned error; throw it-        Right _ -> return b+          CB.delete cf k' (ColNames [cn']) wcl+        (DoNothing) -> return ()+      return b   in do-    res <- CB.getCol cp cf k' cn' rcl+    res <- CB.getCol cf k' cn' rcl     case res of-      Left NotFoundException -> execF Nothing-      Left e -> throw e-      Right Column{..} -> execF (unMarshallJSON' colVal)-      Right SuperColumn{..} -> throw $ +      Nothing -> execF Nothing+      Just Column{..} -> execF (unMarshallJSON' colVal)+      Just SuperColumn{..} -> throw $          OperationNotSupported "modify not implemented for SuperColumn"  @@ -183,41 +169,40 @@ -- This method may throw a 'CassandraException' for all exceptions other than -- 'NotFoundException'. modify_-  :: (CKey rowKey, CKey colKey, ToJSON a, FromJSON a)-  => CPool-  -> ColumnFamily-  -> rowKey-  -> colKey+  :: (MonadCassandra m, ToJSON a, FromJSON a)+  => ColumnFamily+  -> RowKey+  -> ColumnName   -> ConsistencyLevel   -- ^ Read quorum   -> ConsistencyLevel   -- ^ Write quorum-  -> (Maybe a -> IO (ModifyOperation a))+  -> (Maybe a -> m (ModifyOperation a))   -- ^ Modification function. Called with 'Just' the value if present,   -- 'Nothing' otherwise.-  -> IO ()-modify_ cp cf k cn rcl wcl f = +  -> m ()+modify_ cf k cn rcl wcl f =    let     f' prev = do       op <- f prev       return (op, ())   in do-  modify cp cf k cn rcl wcl f'-  return ()+      modify cf k cn rcl wcl f'+      return ()   ------------------------------------------------------------------------------- -- Simple insertion function making use of typeclasses insertCol-    :: (CKey rowKey, CKey colKey, ToJSON a)-    => CPool -> ColumnFamily -    -> rowKey-    -> colKey+    :: (MonadCassandra m, ToJSON a)+    => ColumnFamily +    -> RowKey+    -> ColumnName     -> ConsistencyLevel     -> a -- ^ Content-    -> IO ()-insertCol cp cf k cn cl a = -  throwing $ insert cp cf (toBS k) cl [col (toBS cn) (marshallJSON' a)]+    -> m ()+insertCol cf k cn cl a = +    insert cf (toBS k) cl [col (toBS cn) (marshallJSON' a)]   ------------------------------------------------------------------------------@@ -227,35 +212,62 @@ -- ColumnFamily and contents of returned columns are cast into the -- target type. get-    :: (CKey rowKey, CKey colKey, FromJSON a)-    => CPool -> ColumnFamily-    -> rowKey+    :: (MonadCassandra m, FromJSON a)+    => ColumnFamily+    -> RowKey     -> Selector-   -> ConsistencyLevel-    -> IO [(colKey, a)]-get cp cf k s cl = do-  res <- throwing $ CB.get cp cf (toBS k) s cl+    -> ConsistencyLevel+    -> m [(ColumnName, a)]+get cf k s cl = do+  res <- CB.get cf (toBS k) s cl   return $ map col2val res   -------------------------------------------------------------------------------+data KeySelector +    = Keys [RowKey]+    | KeyRange KeyRangeType RowKey RowKey Int32+++-------------------------------------------------------------------------------+ksToBasicKS (Keys k) = CB.Keys $ map toBS k+ksToBasicKS (KeyRange ty fr to i) = CB.KeyRange ty (toBS fr) (toBS to) i+++-------------------------------------------------------------------------------+-- | Get a slice of columns from multiple rows at once. Note that+-- since we are auto-serializing from JSON, all the columns must be of+-- the same data type.+getMulti +    :: (MonadCassandra m, FromJSON a)+    => ColumnFamily+    -> KeySelector +    -> Selector +    -> ConsistencyLevel+    -> m (Map RowKey [(ColumnName, a)])+getMulti cf ks s cl = do+  res <- CB.getMulti cf (ksToBasicKS ks) s cl+  return . M.fromList . map conv . M.toList $ res+  where+    conv (k, row) = (fromBS k, map col2val row)+  ++------------------------------------------------------------------------------- -- | Get a single column from a single row getCol-    :: (CKey rowKey, CKey colKey, FromJSON a)-    => CPool -> ColumnFamily-    -> rowKey-    -> colKey+    :: (MonadCassandra m, FromJSON a)+    => ColumnFamily+    -> RowKey+    -> ColumnName     -> ConsistencyLevel-    -> IO (Maybe a)-getCol cp cf rk ck cl = do-  res <- CB.getCol cp cf (toBS rk) (toBS ck) cl-  case res of-    Left NotFoundException -> return Nothing-    Left e -> throw e-    Right a -> -        let (_ :: ByteString, x) = col2val a-        in return $ Just x-+    -> m (Maybe a)+getCol cf rk ck cl = do+    res <- CB.getCol cf (toBS rk) (toBS ck) cl +    case res of+      Nothing -> return Nothing+      Just res' -> do+          let (_, x) = col2val res'+          return $ Just x   ------------------------------------------------------------------------------@@ -263,18 +275,16 @@ -- it throws an exception rather than returning an explicit Either -- value. delete -  :: (CKey rowKey)-  => CPool-  -- ^ Cassandra connection-  -> ColumnFamily+  :: (MonadCassandra m)+  =>ColumnFamily   -- ^ In 'ColumnFamily'-  -> rowKey+  -> RowKey   -- ^ Key to be deleted   -> Selector   -- ^ Columns to be deleted   -> ConsistencyLevel-  -> IO ()-delete p cf k s cl = throwing $ CB.delete p cf (toBS k) s cl+  -> m ()+delete cf k s cl = CB.delete cf (toBS k) s cl   ------------------------------------------------------------------------------

@@ -0,0 +1,17 @@+{-| A Collection of utilities for binary packing values into Bytestring |-}++module Database.Cassandra.Pack+    ( packLong+    ) where++-------------------------------------------------------------------------------+import qualified Data.Binary.Put            as BN+import           Data.ByteString.Lazy.Char8 (ByteString)+-------------------------------------------------------------------------------+++-------------------------------------------------------------------------------+-- | Pack any integral value into LongType+packLong :: Integral a => a -> ByteString+packLong = BN.runPut . BN.putWord64be . fromIntegral+

@@ -1,10 +1,18 @@ {-# LANGUAGE PatternGuards, NamedFieldPuns, RecordWildCards #-}---module Database.Cassandra.Pool where-+{-# LANGUAGE PackageImports #-} +module Database.Cassandra.Pool +    ( CPool+    , Server+    , defServer+    , defServers+    , KeySpace+    , Cassandra (..)+    , createCassandraPool+    , withResource+    ) where +------------------------------------------------------------------------------- import Control.Applicative ((<$>)) import Control.Concurrent.STM import Control.Exception (SomeException, catch, onException)@@ -12,30 +20,33 @@ import Control.Monad.IO.Class (liftIO) import Data.ByteString (ByteString) import Data.List (partition)+import "resource-pool" Data.Pool import Data.Time.Clock (NominalDiffTime, UTCTime, diffUTCTime, getCurrentTime)+import qualified Database.Cassandra.Thrift.Cassandra_Client as C+import Network import Prelude hiding (catch)-import System.Mem.Weak (addFinalizer) import System.IO (hClose, Handle(..))---import qualified Database.Cassandra.Thrift.Cassandra_Client as C+import System.Mem.Weak (addFinalizer)+import Thrift.Protocol.Binary import Thrift.Transport-import Thrift.Transport.Handle import Thrift.Transport.Framed-import Thrift.Protocol.Binary-import Network+import Thrift.Transport.Handle+------------------------------------------------------------------------------- + ------------------------------------------------------------------------------ -- | A round-robin pool of cassandra connections-type CPool = Pool Cassandra Server+type CPool = Pool Cassandra  -type Server = (HostName, PortID)+-------------------------------------------------------------------------------+-- | A (ServerName, Port) tuple+type Server = (HostName, Int)   -- | A localhost server with default configuration defServer :: Server-defServer = ("127.0.0.1", PortNumber 9160)+defServer = ("127.0.0.1", 9160)   -- | A single localhost server with default configuration@@ -43,9 +54,11 @@ defServers = [defServer]  +------------------------------------------------------------------------------- type KeySpace = String  +------------------------------------------------------------------------------- data Cassandra = Cassandra {     cHandle :: Handle   , cFramed :: FramedTransport Handle@@ -62,17 +75,29 @@   :: [Server]   -- ^ List of servers to connect to   -> Int-  -- ^ Max connections per server (n)+  -- ^ Number of stripes to maintain+  -> Int+  -- ^ Max connections per stripe   -> NominalDiffTime   -- ^ Kill each connection after this many seconds   -> KeySpace   -- ^ Each pool operates on a single KeySpace   -> IO CPool-createCassandraPool servers n maxIdle ks = createPool cr dest n maxIdle servers+createCassandraPool servers numStripes perStripe maxIdle ks = do+    sring <- newTVarIO $ mkRing servers+    createPool (cr sring) dest numStripes maxIdle perStripe   where-    cr :: Server -> IO Cassandra-    cr (host, p) = do-      h <- hOpen (host, p)+    cr :: ServerRing -> IO Cassandra+    cr sring = do+      server <- atomically $ do+        ring@Ring{..} <- readTVar sring+        writeTVar sring $ next ring+        return current+      crCon server+        +    crCon :: Server -> IO Cassandra+    crCon (host, p) = do+      h <- hOpen (host, PortNumber (fromIntegral p))       ft <- openFramedTransport h       let p = BinaryProtocol ft       C.set_keyspace (p,p) ks@@ -80,36 +105,12 @@     dest h = hClose $ cHandle h  ---------------------------------------------------------------------------------- Generic pool functionality - might want to factor out one day---------------------------------------------------------------------------------- -newtype Pool a s = Pool { stripes :: TVar (Ring (Stripe a s)) }---createPool cr dest n maxIdle servers = do-  when (maxIdle < 0.5) $-    modError "pool " $ "invalid idle time " ++ show maxIdle-  when (n < 1) $-    modError "pool " $ "invalid maximum resource count " ++ show n-  stripes' <- mapM (createStripe cr dest n maxIdle) servers-  -- reaperId <- forkIO $ reaper destroy idleTime localPools-  -- addFinalizer p $ killThread reaperId-  tv <- atomically $ newTVar (mkRing stripes')-  return $ Pool tv----withPool :: Pool a s -> (a -> IO b) -> IO b-withPool Pool{..} f = do-  Ring{..} <- atomically $ do-    r <- readTVar stripes-    writeTVar stripes $ next r-    return r-  withStripe current f+-------------------------------------------------------------------------------+type ServerRing = TVar (Ring Server)  +------------------------------------------------------------------------------- data Ring a = Ring {     current :: !a   , used :: [a]@@ -117,10 +118,12 @@   }  +------------------------------------------------------------------------------- mkRing [] = error "Can't make a ring from empty list" mkRing (a:as) = Ring a [] as  +------------------------------------------------------------------------------- next :: Ring a -> Ring a next Ring{..}    | (n:rest) <- upcoming@@ -128,79 +131,4 @@ next Ring{..}    | (n:rest) <- reverse (current : used)   = Ring n [] rest---data Stripe a s = Stripe {-    idle :: TVar [Connection a]-  -- ^ FIFO buffer of idle connections-  , inUse :: TVar Int-  -- ^ Set of in-use connections-  , server :: s-  -- ^ Server this strip is connected to-  , create :: s -> IO a-  -- ^ Create action-  , destroy :: (a -> IO ())-  -- ^ Destroy action-  , cxns :: Int-  -- ^ Max connections-  , ttl :: NominalDiffTime-  -- ^ TTL for each connection-  }---createStripe -  :: (s -> IO a)-  -> (a -> IO ())-  -> Int-  -> NominalDiffTime-  -> s-  -> IO (Stripe a s)-createStripe cr dest n maxIdle s = atomically $ do-  idles <- newTVar []-  used <- newTVar 0-  return $ Stripe {-    idle = idles-  , inUse = used-  , server = s-  , create = cr-  , destroy = dest-  , cxns = n-  , ttl = maxIdle-  }---withStripe :: Stripe a s -> (a -> IO b) -> IO b-withStripe Stripe{..} f = do-  res <- join . atomically $ do-    cs <- readTVar idle-    case cs of-      (Connection{..}:rest) -> writeTVar idle rest >> return (return cxn)-      [] -> do-        used <- readTVar inUse-        when (used == cxns) retry-        writeTVar inUse $! used + 1-        return $ create server -          `onException` atomically (modifyTVar_ inUse (subtract 1))-  ret <- f res `onException` (destroy res `onException` return ())-  now <- getCurrentTime-  atomically $ modifyTVar_ idle (Connection res now : ) -  return ret----data Connection a = Connection {-    cxn :: a-  , lastUse :: UTCTime-  }----modifyTVar_ :: TVar a -> (a -> a) -> STM ()-modifyTVar_ v f = readTVar v >>= \a -> writeTVar v $! f a---modError :: String -> String -> a-modError func msg =-    error $ "Data.Pool." ++ func ++ ": " ++ msg- 

@@ -5,17 +5,23 @@  module Database.Cassandra.Types where -import           Control.Monad+------------------------------------------------------------------------------- import           Control.Exception-import           Data.Generics-import           Data.ByteString.Lazy (ByteString)+import           Control.Monad import qualified Data.ByteString.Char8 as B+import           Data.ByteString.Lazy (ByteString) import qualified Data.ByteString.Lazy.Char8 as LB+import           Data.Generics import           Data.Int (Int32, Int64)+import           Data.Text                  (Text)+import qualified Data.Text                  as T+import qualified Data.Text.Encoding         as T+import qualified Data.Text.Lazy             as LT+import qualified Data.Text.Lazy.Encoding    as LT import           Data.Time import           Data.Time.Clock.POSIX- import qualified Database.Cassandra.Thrift.Cassandra_Types as C+-------------------------------------------------------------------------------   -- | A 'Key' range selector to use with 'getMulti'.@@ -124,13 +130,14 @@ mkThriftCol Column{..} = do   now <- getTime   return $ C.Column (Just colKey) (Just colVal) (Just now) colTTL+mkThriftCol _ = error "mkThriftCol can only process regular columns."   castColumn :: C.ColumnOrSuperColumn -> Either CassandraException Column castColumn x | Just c <- C.f_ColumnOrSuperColumn_column x = castCol c              | Just c <- C.f_ColumnOrSuperColumn_super_column x = castSuperCol c castColumn _ = -  Left $ ConversionException "Unsupported/unexpected ColumnOrSuperColumn type"+  Left $ ConversionException "castColumn: Unsupported/unexpected ColumnOrSuperColumn type"   castCol :: C.Column -> Either CassandraException Column@@ -175,4 +182,37 @@ getTime = do   t <- getPOSIXTime   return . fromIntegral . floor $ t * 1000000++++                             --------------------+                             -- CKey Typeclass --+                             --------------------++------------------------------------------------------------------------------+-- | A typeclass to enable using any string-like type for row and column keys+class CKey a where+  toBS    :: a -> ByteString+  fromBS  :: ByteString -> a++instance CKey String where+    toBS = LB.pack+    fromBS = LB.unpack++instance CKey LT.Text where+    toBS = LT.encodeUtf8+    fromBS = LT.decodeUtf8++instance CKey T.Text where+    toBS = toBS . LT.fromChunks . return+    fromBS = T.concat . LT.toChunks . fromBS++instance CKey B.ByteString where+    toBS = LB.fromChunks . return+    fromBS = B.concat . LB.toChunks . fromBS++instance CKey ByteString where+    toBS = id+    fromBS = id+