cassy 0.4.0.1 → 0.5.1.0

raw patch · 8 files changed

+855/−52 lines, 8 filesdep +MonadCatchIO-transformersdep +asyncdep +cerealPVP ok

version bump matches the API change (PVP)

Dependencies added: MonadCatchIO-transformers, async, cereal, conduit, retry, safecopy

API changes (from Hackage documentation)

Files

• cassy.cabal +32/−10
• src/Database/Cassandra/Basic.hs +63/−3
• src/Database/Cassandra/JSON.hs +36/−11
• src/Database/Cassandra/Marshall.hs +470/−0
• src/Database/Cassandra/Pack.hs +131/−8
• src/Database/Cassandra/Pool.hs +22/−13
• src/Database/Cassandra/Types.hs +73/−4
• test/Test.hs +28/−3

@@ -1,16 +1,19 @@ Name:                cassy-Version:             0.4.0.1+Version:             0.5.1.0 Synopsis:            A high level driver for the Cassandra datastore License:             BSD3 License-file:        LICENSE Author:              Ozgun Ataman Maintainer:          ozataman@gmail.com Homepage:            http://github.com/ozataman/cassy-Category:            Data+Category:            Database Build-type:          Simple description:   The objective is to completely isolate away the thrift layer, providing-  a more idiomatic Haskell experience working with Cassandra.+  a more idiomatic Haskell experience working with Cassandra. Be sure+  to check out the README on Github for some more explanation and+  Release Notes, which is helpful in talking about what this library+  can do.   .   Certain parts of the API was inspired by pycassa (Python client) and   hscassandra (on Hackage).@@ -24,11 +27,19 @@     implementation of Cassandra interaction using the thrift API     underneath.   .-  * /Database.Cassandra.JSON/: A higher level API that operates on-    values with ToJSON and FromJSON isntances from the /aeson/-    library. This module has in part been inspired by Bryan-    O\'Sullivan\'s /riak/ client for Haskell.+  * /Database.Cassandra.Marshall/: Intended to be the main high level+    module that you should use, Marshall allows you to pick the+    serialization strategy you would like to use at each function+    call. We recommend using 'casSafeCopy' due to its support for+    evolving data types, although casJSON maybe another popular+    choice.   .+  * /Database.Cassandra.JSON/: (Now deprecated; use Marshall instead)+    A higher level API that operates on values with ToJSON and+    FromJSON isntances from the /aeson/ library. This module has in+    part been inspired by Bryan O\'Sullivan\'s /riak/ client for+    Haskell.+  .   * /Database.Cassandra.Pool/: Handles a /pool/ of connections to     multiple servers in a cluster, splitting the load among them.   .@@ -53,6 +64,7 @@   hs-source-dirs: src   Exposed-modules:     Database.Cassandra.Basic+    Database.Cassandra.Marshall     Database.Cassandra.JSON     Database.Cassandra.Pool     Database.Cassandra.Types@@ -62,6 +74,8 @@       base >= 4 && < 5     , bytestring     , binary+    , cereal+    , safecopy     , containers     , network     , time@@ -69,13 +83,17 @@     , stm     , syb     , text-    , attoparsec                >= 0.10    && < 0.11+    , attoparsec >= 0.10 && < 0.11     , aeson     , Thrift >= 0.6     , cassandra-thrift >= 0.8     , resource-pool-    , errors     , data-default+    , async+    , errors+    , MonadCatchIO-transformers >= 0.3+    , retry+    , conduit >= 0.5   test-suite test@@ -87,6 +105,8 @@       base >= 4 && < 5     , bytestring     , binary+    , cereal+    , safecopy     , containers     , network     , time@@ -99,8 +119,10 @@     , Thrift >= 0.6     , cassandra-thrift >= 0.8     , resource-pool-    , errors     , data-default+    , errors+    , MonadCatchIO-transformers >= 0.3+    , retry      , test-framework >= 0.6     , test-framework-quickcheck2 >= 0.2.12.2

@@ -7,6 +7,18 @@ {-# LANGUAGE PatternGuards              #-} {-# LANGUAGE RecordWildCards            #-} +-----------------------------------------------------------------------------+-- |+-- Module      :  Database.Cassandra.Basic+-- Copyright   :  Ozgun Ataman+-- License     :  BSD3+--+-- Maintainer  :  Ozgun Ataman+-- Stability   :  experimental+--+-- Low-level functionality for working with Cassandra at the most+-- basic level.+----------------------------------------------------------------------------  module Database.Cassandra.Basic @@ -24,6 +36,8 @@     , MonadCassandra (..)     , Cas (..)     , runCas+    , transCas+    , mapCassandra      -- * Cassandra Operations     , getCol@@ -32,9 +46,15 @@     , insert     , delete +    -- * Retrying Queries+    , retryCas+    , casRetryH+    , networkRetryH+     -- * Filtering     , Selector(..)     , range+    , boundless     , Order(..)     , reverseOrder     , KeySelector(..)@@ -81,15 +101,19 @@  ------------------------------------------------------------------------------- import           Control.Applicative+import           Control.Concurrent.Async import           Control.Exception import           Control.Monad+import qualified Control.Monad.CatchIO                      as MCIO import           Control.Monad.Reader+import           Control.Retry                              as R import           Data.ByteString.Lazy                       (ByteString) import           Data.Map                                   (Map) import qualified Data.Map                                   as M import           Data.Maybe                                 (mapMaybe)+import           Data.Traversable                           (Traversable) import qualified Database.Cassandra.Thrift.Cassandra_Client as C-import           Database.Cassandra.Thrift.Cassandra_Types  (ConsistencyLevel(..))+import           Database.Cassandra.Thrift.Cassandra_Types  (ConsistencyLevel (..)) import qualified Database.Cassandra.Thrift.Cassandra_Types  as T import           Prelude                                    hiding (catch) -------------------------------------------------------------------------------@@ -132,6 +156,15 @@   -------------------------------------------------------------------------------+-- | Run a list of cassandra computations in parallel using the async library+mapCassandra :: (Traversable t, MonadCassandra m) => t (Cas b) -> m (t b)+mapCassandra ms = do+    cp <- getCassandraPool+    let f m = runCas cp m+    liftIO $ mapConcurrently f ms+++------------------------------------------------------------------------------- withCassandraPool :: MonadCassandra m => (Cassandra -> IO b) -> m b withCassandraPool f = do   p <- getCassandraPool@@ -150,6 +183,19 @@ runCas = flip runReaderT  +-- | Unwrap a Cassandra action and return an IO continuation that can+-- then be run in a pure IO context.+--+-- This is useful when you design all your functions in a generic form+-- with 'MonadCassandra' m constraints and then one day need to feed+-- your function to a utility that can only run in an IO context. This+-- function is then your friendly utility for extracting an IO action.+transCas :: MonadCassandra m => Cas a -> m (IO a)+transCas m = do+    cp <- getCassandraPool+    return $ runCas cp m++ ------------------------------------------------------------------------------- instance (MonadIO m) => MonadCassandra (ReaderT CPool m) where     getCassandraPool = ask@@ -252,7 +298,7 @@ insert cf k cl row = withCassandraPool $ \ Cassandra{..} -> do   let insCol cp c = do         c' <- mkThriftCol c-        C.insert (cProto, cProto) k cp c' cl+        wrapException $ C.insert (cProto, cProto) k cp c' cl   forM_ row $ \ c -> do     case c of       Column{} -> do@@ -302,7 +348,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"+    Range{} -> error "delete: Range delete not implemented"   where     -- wipe out the entire row     cpAll = T.ColumnPath (Just cf) Nothing Nothing@@ -342,3 +388,17 @@   case res of     Left e -> throw e     Right a -> return a+++-- | 'retrying' with direct cassandra support. Server-related failures+-- will be retried.+--+-- 'UnavailableException', 'TimedOutException' and+-- 'SchemaDisagreementException' will be automatically retried.+retryCas :: MCIO.MonadCatchIO m+         => R.RetrySettings+         -- ^ For default settings, just use 'def'+         -> m a+         -- ^ Action to perform+         -> m a+retryCas set f = R.recovering set [casRetryH, networkRetryH] f

@@ -36,6 +36,8 @@     , MonadCassandra (..)     , Cas     , runCas+    , transCas+    , mapCassandra      -- * Cassandra Operations     , get@@ -43,6 +45,7 @@     , getCol     , getMulti     , insertCol+    , insertColTTL     , modify     , modify_     , delete@@ -58,6 +61,7 @@     -- * Filtering     , Selector (..)     , range+    , boundless     , Order(..)     , reverseOrder     , KeySelector (..)@@ -85,7 +89,7 @@ import           Control.Monad import           Data.Aeson                 as A import           Data.Aeson.Parser          (value)-import qualified Data.Attoparsec            as Atto (IResult(..), parse)+import qualified Data.Attoparsec            as Atto (IResult (..), parse) import qualified Data.ByteString.Char8      as B import           Data.ByteString.Lazy.Char8 (ByteString) import qualified Data.ByteString.Lazy.Char8 as LB@@ -94,7 +98,8 @@ import qualified Data.Map                   as M import           Prelude                    hiding (catch) --------------------------------------------------------------------------------import           Database.Cassandra.Basic   hiding (KeySelector(..), delete, get, getCol, getMulti)+import           Database.Cassandra.Basic   hiding (KeySelector (..), delete,+                                             get, getCol, getMulti) import qualified Database.Cassandra.Basic   as CB ------------------------------------------------------------------------------- @@ -153,17 +158,16 @@   -- ^ Return the decided 'ModifyOperation' and its execution outcome modify cf k cn rcl wcl f =   let-    k'  = toColKey k     cn' = encodeCas cn     execF prev = do       (fres, b) <- f prev       case fres of-        Update a  -> insert cf k' wcl [col cn' (marshallJSON' a)]-        Delete    -> CB.delete cf k' (ColNames [cn']) wcl+        Update a  -> insert cf k wcl [col cn' (marshallJSON' a)]+        Delete    -> CB.delete cf k (ColNames [cn']) wcl         DoNothing -> return ()       return b   in do-    res <- CB.getCol cf k' cn' rcl+    res <- CB.getCol cf k cn' rcl     case res of       Nothing              -> execF Nothing       Just Column{..}      -> execF (unMarshallJSON' colVal)@@ -212,9 +216,29 @@     -> a -- ^ Content     -> m () insertCol cf rk cn cl a =-    insert cf (toColKey rk) cl [packCol (cn, marshallJSON' a)]+    insert cf rk cl [packCol (cn, marshallJSON' a)]  ++-------------------------------------------------------------------------------+-- Simple insertion function making use of typeclasses+insertColTTL+    :: (MonadCassandra m, ToJSON a, CasType k)+    => ColumnFamily+    -> RowKey+    -> k+    -- ^ Column name. See 'CasType' for what you can use here.+    -> ConsistencyLevel+    -> a+    -- ^ Content+    -> Int32+    -- ^ TTL for this column+    -> m ()+insertColTTL cf rk cn cl a ttl = insert cf rk cl [column]+    where+      column = Column (packKey cn) (marshallJSON' a) Nothing (Just ttl)++ ------------------------------------------------------------------------------ -- | An arbitrary get operation - slice with 'Selector'. --@@ -231,7 +255,7 @@     -> m [(k, a)]     -- ^ List of key-value pairs. See 'CasType' for what key types you can use. get cf k s cl = do-  res <- CB.get cf (toColKey k) s cl+  res <- CB.get cf k s cl   return $ map col2val res  @@ -279,7 +303,7 @@   res <- CB.getMulti cf (ksToBasicKS ks) s cl   return . M.fromList . map conv . M.toList $ res   where-    conv (k, row) = (fromColKey' k, map col2val row)+    conv (k, row) = (k, map col2val row)   -------------------------------------------------------------------------------@@ -293,7 +317,7 @@     -> ConsistencyLevel     -> m (Maybe a) getCol cf rk ck cl = do-    res <- CB.getCol cf (toColKey rk) (encodeCas ck) cl+    res <- CB.getCol cf rk (encodeCas ck) cl     case res of       Nothing -> return Nothing       Just res' -> do@@ -315,7 +339,7 @@   -- ^ Columns to be deleted   -> ConsistencyLevel   -> m ()-delete cf k s cl = CB.delete cf (toColKey k) s cl+delete cf k s cl = CB.delete cf k s cl   ------------------------------------------------------------------------------@@ -334,6 +358,7 @@ -- | Lazy 'unMarshallJSON' unMarshallJSON' :: FromJSON a => ByteString -> Maybe a unMarshallJSON' = unMarshallJSON . B.concat . LB.toChunks+  ------------------------------------------------------------------------------ -- | Decode JSON

@@ -0,0 +1,470 @@+{-# LANGUAGE ExistentialQuantification #-}+{-# LANGUAGE FlexibleContexts          #-}+{-# LANGUAGE FlexibleInstances         #-}+{-# LANGUAGE MultiParamTypeClasses     #-}+{-# LANGUAGE NamedFieldPuns            #-}+{-# LANGUAGE OverloadedStrings         #-}+{-# LANGUAGE PatternGuards             #-}+{-# LANGUAGE RecordWildCards           #-}+{-# LANGUAGE ScopedTypeVariables       #-}+{-# LANGUAGE TypeSynonymInstances      #-}+++-----------------------------------------------------------------------------+-- |+-- Module      :+-- Copyright   :  (C) 2013 Ozgun Ataman+-- License     :  All Rights Reserved+--+-- Maintainer  :  Ozgun Ataman <oz@soostone.com>+-- Stability   :  experimental+--+-- Defines Cassandra operations for persistence of complex Haskell+-- data objects with custom-selected but implicitly performed+-- serialization.+--+-- The main design choice is to require a dictionary dictating+-- marshalling/serialization policy for every operation, rather than a+-- typeclass that can be instantiated once.+----------------------------------------------------------------------------++module Database.Cassandra.Marshall+    (++    -- * Connection+      CPool+    , Server+    , defServer+    , defServers+    , KeySpace+    , createCassandraPool++    -- * MonadCassandra Typeclass+    , MonadCassandra (..)+    , Cas+    , runCas+    , transCas+    , mapCassandra++    -- * Haskell Record Marshalling++    , Marshall (..)+    , casShow+    , casJSON+    -- , casBinary+    , casSerialize+    , casSafeCopy++    -- * Cassandra Operations+    , get+    , get_+    , getCol+    , getMulti+    , insertCol+    , insertColTTL+    , modify+    , modify_+    , delete++    -- * Retrying Queries+    , CB.retryCas+    , casRetryH++    -- * Necessary Types+    , RowKey+    , ColumnName+    , ModifyOperation (..)+    , ColumnFamily+    , ConsistencyLevel (..)+    , CassandraException (..)++    -- * Filtering+    , Selector (..)+    , range+    , boundless+    , Order(..)+    , reverseOrder+    , KeySelector (..)+    , KeyRangeType (..)+++    -- * Pagination+    , PageResult (..)+    , pIsDry+    , pIsDone+    , pHasMore+    , paginate+    , paginateSource+    , pageToSource++    -- * Helpers+    , CKey (..)+    , fromColKey'++    -- * Cassandra Column Key Types+    , module Database.Cassandra.Pack+    ) where++-------------------------------------------------------------------------------+import           Control.Error+import           Control.Monad+import           Control.Monad.CatchIO+import           Control.Monad.Trans+import           Control.Retry              as R+import qualified Data.Aeson                 as A+import qualified Data.Attoparsec            as Atto (IResult (..), parse)+import qualified Data.Binary                as BN+import qualified Data.Binary.Get            as BN+import qualified Data.ByteString.Char8      as B+import           Data.ByteString.Lazy.Char8 (ByteString)+import qualified Data.ByteString.Lazy.Char8 as LB+import           Data.Conduit+import qualified Data.Conduit.List          as C+import           Data.Int                   (Int32)+import           Data.Map                   (Map)+import qualified Data.Map                   as M+import qualified Data.SafeCopy              as SC+import qualified Data.Serialize             as SL+import           Prelude                    hiding (catch)+-------------------------------------------------------------------------------+import           Database.Cassandra.Basic   hiding (KeySelector (..), delete,+                                             get, getCol, getMulti)+import qualified Database.Cassandra.Basic   as CB+import           Database.Cassandra.Pack+import           Database.Cassandra.Types+-------------------------------------------------------------------------------+++-- | A Haskell dictionary containing a pair of encode/decode+-- functions.+--+-- This is the main design choice in this module. We require that each+-- operation takes an explicit marshalling policy rather than a+-- typeclass which makes it possible to do it in a single way per data+-- type.+--+-- You can create your own objects of this type with great ease. Just+-- look at one of the examples here ('casJSON', 'casSerialize', etc.)+data Marshall a = Marshall {+      marshallEncode :: a -> ByteString+    -- ^ An encoding function+    , marshallDecode :: ByteString -> Either String a+    -- ^ A decoding function+    }+++-- | Marshall data using JSON encoding. Good interoperability, but not+-- very efficient for data storage.+casJSON :: (A.ToJSON a, A.FromJSON a) => Marshall a+casJSON = Marshall A.encode A.eitherDecode+++-- | Marshall data using 'Show' and 'Read'. Not meant for serious production cases.+casShow :: (Show a, Read a) => Marshall a+casShow = Marshall+            (LB.pack . show)+            (readErr "casShow can't read cassandra value" . LB.unpack)+++-- -- | Marshall data using the 'Binary' instance. This is one of the+-- -- efficient methods available.+-- casBinary :: BN.Binary a => Marshall a+-- casBinary = Marshall BN.encode dec+--     where+--       dec bs = case BN.runGetOrFail BN.get bs of+--                  Left (_,_,err) -> Left err+--                  Right (_,_,a) -> Right a+++-- | Marshall data using the 'SafeCopy' instance. This is quite well+-- suited for production because it is both very efficient and+-- provides a systematic way to migrate your data types over time.+casSafeCopy :: SC.SafeCopy a => Marshall a+casSafeCopy = Marshall (SL.runPutLazy . SC.safePut) (SL.runGetLazy SC.safeGet)+++-- | Marshall data using the 'Serialize' instance. Like 'Binary',+-- 'Serialize' is very efficient.+casSerialize :: SL.Serialize a => Marshall a+casSerialize = Marshall SL.encodeLazy SL.decodeLazy+++------------------------------------------------------------------------------+-- | A modify function that will fetch a specific column, apply modification+-- function on it and save results back to Cassandra.+--+-- A 'b' side value is returned for computational convenience.+--+-- This is intended to be a workhorse function, in that you should be+-- able to do all kinds of relatively straightforward operations just+-- using this function.+--+-- This method may throw a 'CassandraException' for all exceptions other than+-- 'NotFoundException'.+modify+  :: (MonadCassandra m, CasType k)+  => Marshall a+  -- ^ A serialization methodology. Example: 'casJSON'+  -> ColumnFamily+  -> RowKey+  -> k+  -- ^ Column name; anything in CasType+  -> ConsistencyLevel+  -- ^ Read quorum+  -> ConsistencyLevel+  -- ^ Write quorum+  -> (Maybe a -> m (ModifyOperation a, b))+  -- ^ Modification function. Called with 'Just' the value if present,+  -- 'Nothing' otherwise.+  -> m b+  -- ^ Return the decided 'ModifyOperation' and its execution outcome+modify Marshall{..} cf k cn rcl wcl f =+  let+    cn' = encodeCas cn+    execF prev = do+      (fres, b) <- f prev+      case fres of+        Update a  -> insert cf k wcl [col cn' (marshallEncode a)]+        Delete    -> CB.delete cf k (ColNames [cn']) wcl+        DoNothing -> return ()+      return b+  in do+    res <- CB.getCol cf k cn' rcl+    case res of+      Nothing              -> execF Nothing+      Just Column{..}      -> execF (hush $ marshallDecode colVal)+      Just SuperColumn{..} -> throw $+        OperationNotSupported "modify not implemented for SuperColumn"+++------------------------------------------------------------------------------+-- | Same as 'modify' but does not offer a side value.+--+-- This method may throw a 'CassandraException' for all exceptions other than+-- 'NotFoundException'.+modify_+  :: (MonadCassandra m, CasType k)+  => Marshall a+  -> ColumnFamily+  -> RowKey+  -> k+  -- ^ Column name; anything in CasType+  -> ConsistencyLevel+  -- ^ Read quorum+  -> ConsistencyLevel+  -- ^ Write quorum+  -> (Maybe a -> m (ModifyOperation a))+  -- ^ Modification function. Called with 'Just' the value if present,+  -- 'Nothing' otherwise.+  -> m ()+modify_ m cf k cn rcl wcl f =+  let+    f' prev = do+      op <- f prev+      return (op, ())+  in do+      modify m cf k cn rcl wcl f'+      return ()+++-------------------------------------------------------------------------------+-- Simple insertion function making use of typeclasses+insertCol+    :: (MonadCassandra m, CasType k)+    => Marshall a+    -> ColumnFamily+    -> RowKey+    -> k+    -- ^ Column name. See 'CasType' for what you can use here.+    -> ConsistencyLevel+    -> a -- ^ Content+    -> m ()+insertCol Marshall{..} cf rk cn cl a =+    insert cf rk cl [packCol (cn, marshallEncode a)]++++-------------------------------------------------------------------------------+-- Simple insertion function making use of typeclasses+insertColTTL+    :: (MonadCassandra m, CasType k)+    => Marshall a+    -> ColumnFamily+    -> RowKey+    -> k+    -- ^ Column name. See 'CasType' for what you can use here.+    -> ConsistencyLevel+    -> a+    -- ^ Content+    -> Int32+    -- ^ TTL for this column+    -> m ()+insertColTTL Marshall{..} cf rk cn cl a ttl = insert cf rk cl [column]+    where+      column = Column (packKey cn) (marshallEncode a) Nothing (Just ttl)+++------------------------------------------------------------------------------+-- | An arbitrary get operation - slice with 'Selector'.+--+-- Internally based on Basic.get. Table is assumed to be a regular+-- ColumnFamily and contents of returned columns are cast into the+-- target type.+get+    :: (MonadCassandra m, CasType k)+    => Marshall a+    -> ColumnFamily+    -> RowKey+    -> Selector+    -- ^ A slice selector+    -> ConsistencyLevel+    -> m [(k, a)]+    -- ^ List of key-value pairs. See 'CasType' for what key types you can use.+get m cf k s cl = do+  res <- CB.get cf k s cl+  return $ map (col2val m) res+++-------------------------------------------------------------------------------+-- | A version of 'get' that discards the column names for the common+-- scenario. Useful because you would otherwise be forced to manually+-- supply type signatures to get rid of the 'CasType' ambiguity.+get_+    :: (MonadCassandra m)+    => Marshall a+    -> ColumnFamily+    -> RowKey+    -> Selector+    -- ^ A slice selector+    -> ConsistencyLevel+    -> m [a]+get_ m cf k s cl = do+    (res :: [(LB.ByteString, a)]) <- get m cf k s cl+    return $ map snd res+++-------------------------------------------------------------------------------+ksToBasicKS :: KeySelector -> CB.KeySelector+ksToBasicKS (Keys k) = CB.Keys $ map toColKey k+ksToBasicKS (KeyRange ty fr to i) = CB.KeyRange ty (toColKey fr) (toColKey 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)+    => Marshall a+    -> ColumnFamily+    -> KeySelector+    -> Selector+    -> ConsistencyLevel+    -> m (Map RowKey [(ColumnName, a)])+getMulti m 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) = (k, map (col2val m) row)+++-------------------------------------------------------------------------------+-- | Get a single column from a single row+getCol+    :: (MonadCassandra m, CasType k)+    => Marshall a+    -> ColumnFamily+    -> RowKey+    -> k+    -- ^ Column name; anything in 'CasType'+    -> ConsistencyLevel+    -> m (Maybe a)+getCol m cf rk ck cl = do+    res <- CB.getCol cf rk (encodeCas ck) cl+    case res of+      Nothing -> return Nothing+      Just res' -> do+          let (_ :: ByteString, x) = col2val m res'+          return $ Just x+++------------------------------------------------------------------------------+-- | Same as the 'delete' in the 'Cassandra.Basic' module, except that+-- it throws an exception rather than returning an explicit Either+-- value.+delete+  :: (MonadCassandra m)+  => ColumnFamily+  -- ^ In 'ColumnFamily'+  -> RowKey+  -- ^ Key to be deleted+  -> Selector+  -- ^ Columns to be deleted+  -> ConsistencyLevel+  -> m ()+delete cf k s cl = CB.delete cf k s cl+++-------------------------------------------------------------------------------+-- | Convert regular column to a key-value pair+col2val :: CasType k => Marshall a -> Column -> (k, a)+col2val Marshall{..} c = f $ unpackCol c+    where+      f (k, val) = (k, either err id $ marshallDecode val)+      err s = error $ "Cassandra Marshall: Value can't be decoded: " ++ s+col2val _ _ = error "col2val is not implemented for SuperColumns"++++-------------------------------------------------------------------------------+-- | Paginate over columns in a given key, repeatedly applying the+-- given 'Selector'. The 'Selector' must be a 'Range' selector, or+-- else this funtion will raise an exception.+paginate+  :: (MonadCassandra m, MonadCatchIO m, CasType k)+  => Marshall a+  -- ^ Serialization strategy+  -> ColumnFamily+  -> RowKey+  -- ^ Paginate columns of this row+  -> Selector+  -- ^ 'Range' selector to initially and repeatedly apply.+  -> ConsistencyLevel+  -> RetrySettings+  -- ^ Retry strategy for each underlying Cassandra call+  -> m (PageResult m (k, a))+paginate m cf k rng@(Range from to ord per) cl retry = do+  cs <- reverse `liftM` retryCas retry (get m cf k rng cl)+  case cs of+    [] -> return $ PDone []+    [a] -> return $ PDone [a]+    _ ->+      let cont = paginate m cf k (Range (Just cn) to ord per) cl retry+          (cn, _) = head cs+      in  return $ PMore (reverse (drop 1 cs)) cont+paginate _ _ _ _ _ _ = error "Must call paginate with a Range selector"++++-------------------------------------------------------------------------------+-- | Convenience layer: Convert a pagination scheme to a conduit 'Source'.+pageToSource :: (MonadCatchIO m, Monad m) => PageResult m a -> Source m a+pageToSource (PDone as) = C.sourceList as+pageToSource (PMore as m) = C.sourceList as >> lift m >>= pageToSource+++-------------------------------------------------------------------------------+-- | Just like 'paginate', but we instead return a conduit 'Source'.+paginateSource+    :: (CasType k, MonadCatchIO m, MonadCassandra m)+    => Marshall a+    -> ColumnFamily+    -> RowKey+    -> Selector+    -> ConsistencyLevel+    -> RetrySettings+    -> Source m (k, a)+paginateSource m cf k rng cl r = do+    buf <- lift $ paginate m cf k rng cl r+    pageToSource buf+

@@ -1,6 +1,7 @@-{-# LANGUAGE FlexibleInstances    #-}-{-# LANGUAGE OverlappingInstances #-}-{-# LANGUAGE RankNTypes           #-}+{-# LANGUAGE FlexibleInstances          #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE OverlappingInstances       #-}+{-# LANGUAGE RankNTypes                 #-}  {-| A Collection of utilities for binary packing values into Bytestring |-} @@ -14,8 +15,13 @@     , TUtf8 (..)     , TUUID (..)     , TLong (..)+    , TTimeStamp (..)+    , toTimeStamp+    , fromTimeStamp+     , Exclusive (..)     , Single (..)+    , SliceStart (..)     ) where  -------------------------------------------------------------------------------@@ -33,6 +39,9 @@ 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+import           Data.Time.Clock.POSIX -------------------------------------------------------------------------------  @@ -42,12 +51,32 @@ newtype TBytes = TBytes { getTBytes :: ByteString } deriving (Eq,Show,Read,Ord) newtype TCounter = TCounter { getCounter :: ByteString } deriving (Eq,Show,Read,Ord) newtype TInt32 = TInt32 { getInt32 :: Int32 } deriving (Eq,Show,Read,Ord)-newtype TInt = TInt { getInt :: Integer } deriving (Eq,Show,Read,Ord)+newtype TInt = TInt { getInt :: Integer } deriving (Eq,Show,Read,Ord,Enum,Real,Integral,Num) newtype TUUID = TUUID { getUUID :: ByteString } deriving (Eq,Show,Read,Ord)-newtype TLong = TLong { getLong :: Integer } deriving (Eq,Show,Read,Ord)+newtype TLong = TLong { getLong :: Integer } deriving (Eq,Show,Read,Ord,Enum,Real,Integral,Num) newtype TUtf8 = TUtf8 { getUtf8 :: Text } deriving (Eq,Show,Read,Ord)  +-- | Timestamp that stores micro-seconds since epoch as 'TLong' underneath.+newtype TTimeStamp = TTimeStamp { getTimeStamp :: TLong }+    deriving (Eq,Show,Read,Ord,Enum,Num,Real,Integral,CasType)+++-- | Convert commonly used 'UTCTime' to 'TTimeStamp'.+--+-- First converts to seconds since epoch (POSIX seconds), then+-- multiplies by a million and floors the resulting value. The value,+-- therefore, is in micro-seconds and is accurate to within a+-- microsecond.+toTimeStamp :: UTCTime -> TTimeStamp+toTimeStamp utc = fromIntegral . floor . (* 1e6) $ utcTimeToPOSIXSeconds utc+++fromTimeStamp :: TTimeStamp -> UTCTime+fromTimeStamp (TTimeStamp (TLong i)) =+    posixSecondsToUTCTime $ realToFrac $ fromIntegral i / (1e6)++ ------------------------------------------------------------------------------- -- | This typeclass defines and maps to haskell types that Cassandra -- natively knows about and uses in sorting and potentially validating@@ -131,12 +160,16 @@     encodeCas = runPut . putWord64be . fromIntegral . getInt     decodeCas = TInt . fromIntegral . runGet getWord64be ++------------------------------------------------------------------------------- instance CasType Int where     encodeCas = encodeCas . TInt . fromIntegral     decodeCas = fromIntegral . getInt . decodeCas + ---------------------------------------------------------------------------------- | Pack as an 8 byte unsigned number; negative signs are lost.+-- | Pack as an 8 byte unsigned number; negative signs are lost. Maps+-- to 'LongType'. instance CasType TLong where     encodeCas = runPut . putWord64be . fromIntegral . getLong     decodeCas = TLong . fromIntegral . runGet getWord64be@@ -150,14 +183,27 @@   -------------------------------------------------------------------------------+-- | Encode days as 'LongType' via 'TLong'.+instance CasType Day where+    encodeCas = encodeCas . TLong . toModifiedJulianDay+    decodeCas = ModifiedJulianDay . getLong . decodeCas+++-- | Via 'TTimeStamp', which is via 'TLong'+instance CasType UTCTime where+    encodeCas = encodeCas . toTimeStamp+    decodeCas = fromTimeStamp . decodeCas+++------------------------------------------------------------------------------- -- | Use the 'Single' wrapper when querying only with the first of a -- two or more field CompositeType. instance (CasType a) => CasType (Single a) where-    encodeCas a = runPut $ do-        putSegment a end+    encodeCas (Single a) = runPut $ putSegment a end      decodeCas bs = flip runGet bs $ Single <$> getSegment + ------------------------------------------------------------------------------- -- | Composite types - see Cassandra or pycassa docs to understand instance (CasType a, CasType b) => CasType (a,b) where@@ -196,6 +242,77 @@         <*> getSegment  +                              ------------------+                              -- Slice Starts --+                              ------------------++++instance (CasType a) => CasType (SliceStart (Single a)) where+    encodeCas (SliceStart (Single a)) = runPut $ do+        putSegment a exc+    decodeCas bs = flip runGet bs $ (SliceStart . Single) <$> getSegment+++-------------------------------------------------------------------------------+-- | Composite types - see Cassandra or pycassa docs to understand+instance (CasType a, CasType b) => CasType (SliceStart (a,b)) where+    encodeCas (SliceStart (a, b)) = runPut $ do+        putSegment a sep+        putSegment b exc++    decodeCas bs = SliceStart . flip runGet bs $ (,)+        <$> getSegment+        <*> getSegment+++instance (CasType a, CasType b, CasType c) => CasType (SliceStart (a,b,c)) where+    encodeCas (SliceStart (a, b, c)) = runPut $ do+        putSegment a sep+        putSegment b sep+        putSegment c exc++    decodeCas bs = SliceStart . flip runGet bs $ (,,)+        <$> getSegment+        <*> getSegment+        <*> getSegment+++instance (CasType a, CasType b, CasType c, CasType d) =>+    CasType (SliceStart (a,b,c,d)) where+    encodeCas (SliceStart (a, b, c, d)) = runPut $ do+        putSegment a sep+        putSegment b sep+        putSegment c sep+        putSegment d exc++    decodeCas bs = SliceStart . flip runGet bs $ (,,,)+        <$> getSegment+        <*> getSegment+        <*> getSegment+        <*> getSegment+++++++++++                            -----------------------+                            -- Exclusive Columns --+                            -----------------------+++instance CasType a => CasType (Exclusive (Single a)) where+    encodeCas (Exclusive (Single a)) = runPut $ do+        putSegment a exc++    decodeCas = Exclusive . decodeCas++ instance (CasType a, CasType b) => CasType (a, Exclusive b) where     encodeCas (a, Exclusive b) = runPut $ do         putSegment a sep@@ -249,6 +366,12 @@ -- | Use the Single wrapper when you want to refer only to the first -- coolumn of a CompositeType column. newtype Single a = Single a deriving (Eq,Show,Read,Ord)+++-------------------------------------------------------------------------------+-- | Wrap your composite columns in this type when you're starting an+-- inclusive column slice.+newtype SliceStart a = SliceStart a deriving (Eq,Show,Read,Ord)   -- | composite columns are a pain

@@ -95,27 +95,36 @@   -> IO CPool createCassandraPool servers numStripes perStripe maxIdle ks = do     sring <- newTVarIO $ mkRing servers-    pool <- createPool (cr sring) dest numStripes maxIdle perStripe-    forkIO (serverDiscoveryThread sring ks pool)+    pool <- createPool (cr 4 sring) dest numStripes maxIdle perStripe+    -- forkIO (serverDiscoveryThread sring ks pool)     return pool   where-    cr :: ServerRing -> IO Cassandra-    cr sring = do+    cr :: Int -> ServerRing -> IO Cassandra+    cr n sring = do       s@(host, p) <- atomically $ do         ring@Ring{..} <- readTVar sring         writeTVar sring (next ring)         return current -      handle (handler sring s) $ do+      handle (handler n sring s) $ do           (h,ft,proto) <- openThrift host p           C.set_keyspace (proto, proto) ks           return $ Cassandra h ft proto -    handler :: ServerRing -> Server -> SomeException -> IO Cassandra-    handler sring server e = do-      modifyServers sring (removeServer server)-      cr sring+    handler :: Int -> ServerRing -> Server -> SomeException -> IO Cassandra+    handler 0 _ _ e = error $ "Can't connect to cassandra after several tries: " ++ show e+    handler n sring server e = do +      -- we need a temporary removal system for servers; something+      -- with a TTL just removing them from ring is dangerous, what if+      -- the network is partitioned for a little while? +      +      -- modifyServers sring (removeServer server)++      -- wait 100ms to avoid crazy loops+      threadDelay 100000+      cr (n-1) sring+     dest h = hClose $ cHandle h  @@ -171,7 +180,7 @@  ------------------------------------------------------------------------------ mkRing [] = error "Can't make a ring from empty list"-mkRing (a:as) = Ring S.empty a [] as+mkRing all@(a:as) = Ring (S.fromList all) a [] as   ------------------------------------------------------------------------------@@ -187,11 +196,11 @@ ------------------------------------------------------------------------------ removeServer :: Ord a => a -> Ring a -> Ring a removeServer s r@Ring{..}-  | s `S.member` allItems = Ring (S.delete s allItems) current' used' upcoming'+  | s `S.member` allItems = Ring all' cur' [] up'   | otherwise             = r   where-    used' = filter (/=s) used-    (current':upcoming') = filter (/=s) (current:upcoming)+    all' = S.delete s allItems+    cur' : up' = S.toList all'   ------------------------------------------------------------------------------

@@ -5,14 +5,15 @@ {-# LANGUAGE OverloadedStrings         #-} {-# LANGUAGE PatternGuards             #-} {-# LANGUAGE RecordWildCards           #-}+{-# LANGUAGE ScopedTypeVariables       #-} {-# LANGUAGE TypeSynonymInstances      #-}  module Database.Cassandra.Types where  --------------------------------------------------------------------------------import           Control.Error import           Control.Exception import           Control.Monad+import qualified Control.Monad.CatchIO                     as MCIO import qualified Data.ByteString.Char8                     as B import           Data.ByteString.Lazy                      (ByteString) import qualified Data.ByteString.Lazy.Char8                as LB@@ -33,6 +34,15 @@ -------------------------------------------------------------------------------  +------------------------------------------------------------------------------+-- | Possible outcomes of a modify operation+data ModifyOperation a =+    Update a+  | Delete+  | DoNothing+  deriving (Eq,Show,Ord,Read)++ -- | A 'Key' range selector to use with 'getMulti'. data KeySelector =     Keys [Key]@@ -73,7 +83,7 @@   -- ^ When deleting specific columns in a super column   | forall a b. (CasType a, CasType b) => Range {       rangeStart :: Maybe a-    , rangeEnd :: Maybe b+    , rangeEnd   :: Maybe b     , rangeOrder :: Order     , rangeLimit :: Int32     }@@ -87,7 +97,11 @@ range = Range (Nothing :: Maybe ByteString) (Nothing :: Maybe ByteString) Regular 1024  +boundless :: Maybe ByteString+boundless = Nothing ++ instance Default Selector where     def = All @@ -111,7 +125,7 @@ mkPredicate :: Selector -> C.SlicePredicate mkPredicate s =   let-    allRange = C.SliceRange (Just "") (Just "") (Just False) (Just 5000)+    allRange = C.SliceRange (Just "") (Just "") (Just False) (Just 50000)   in case s of     All -> C.SlicePredicate Nothing (Just allRange)     ColNames ks -> C.SlicePredicate (Just (map encodeCas ks)) Nothing@@ -144,6 +158,7 @@   type Key = ByteString+type RowKey = Key   type ColumnName = ByteString@@ -160,7 +175,7 @@   | Column {       colKey :: ColumnName     , colVal :: Value-    , colTS :: Maybe Int64+    , colTS  :: Maybe Int64     -- ^ Last update timestamp; will be overridden during write/update ops     , colTTL :: Maybe Int32     -- ^ A TTL after which Cassandra will erase the column@@ -229,6 +244,23 @@ instance Exception CassandraException  +-- | Exception handler that returns @True@ for errors that may be+-- resolved after a retry. So they are good candidates for 'retrying'+-- queries.+casRetryH :: Monad m => MCIO.Handler m Bool+casRetryH = MCIO.Handler $ \ e -> return $+    case e of+      UnavailableException{} -> True+      TimedOutException{} -> True+      SchemaDisagreementException{} -> True+      _ -> False+++-- | 'IOException's should be retried+networkRetryH :: Monad m => MCIO.Handler m Bool+networkRetryH = MCIO.Handler $ \ (_ :: IOException) -> return True++ ------------------------------------------------------------------------------ -- | Cassandra is VERY sensitive to its timestamp values. As a convention, -- timestamps are always in microseconds@@ -236,6 +268,43 @@ getTime = do   t <- getPOSIXTime   return . fromIntegral . floor $ t * 1000000+++                               ----------------+                               -- Pagination --+                               ----------------+++-------------------------------------------------------------------------------+-- | Describes the result of a single pagination action+data PageResult m a+  = PDone { pCache :: [a] }+  -- ^ Done, this is all I have.+  | PMore { pCache :: [a], pMore :: m (PageResult m a) }+  -- ^ Here's a batch and there is more when you call the action.+++++-------------------------------------------------------------------------------+pIsDry x = pIsDone x && null (pCache x)++-------------------------------------------------------------------------------+pIsDone PDone{} = True+pIsDone _ = False+++-------------------------------------------------------------------------------+pHasMore PMore{} = True+pHasMore _ = False+++-------------------------------------------------------------------------------+instance Monad m => Functor (PageResult m) where+    fmap f (PDone as) = PDone (fmap f as)+    fmap f (PMore as m) = PMore (fmap f as) m'+      where+        m' = liftM (fmap f) m   

@@ -1,6 +1,7 @@ {-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE OverloadedStrings          #-} {-# LANGUAGE RecordWildCards            #-}+{-# LANGUAGE ScopedTypeVariables        #-} {-# LANGUAGE StandaloneDeriving         #-} {-# LANGUAGE TemplateHaskell            #-} @@ -15,10 +16,12 @@ import qualified Data.Map                                   as M import qualified Data.Text                                  as T import qualified Database.Cassandra.Thrift.Cassandra_Client as C-import           Database.Cassandra.Thrift.Cassandra_Types  (ConsistencyLevel(..))+import           Database.Cassandra.Thrift.Cassandra_Types +                                                             (ConsistencyLevel (..)) import           Database.Cassandra.Thrift.Cassandra_Types  as T import           System.IO.Unsafe-import           Test.Framework                             (defaultMain, testGroup)+import           Test.Framework                             (defaultMain,+                                                             testGroup) import           Test.Framework.Providers.HUnit import           Test.Framework.Providers.QuickCheck2       (testProperty) import           Test.HUnit@@ -47,7 +50,8 @@     , testProperty "cas type marshalling ascii" prop_casTypeInt32     , testProperty "cas type marshalling composite" prop_casTypeComp     , testCase "cas live test - composite get/set" test_composite_col-    , testProperty "cas live test read/write QC" (prop_composite_col_readWrite pool)+    , testCase "cas live - set comp + single col slice" test_composite_slice+    -- , testProperty "cas live test read/write QC" (prop_composite_col_readWrite pool)     ]  @@ -126,6 +130,26 @@   -------------------------------------------------------------------------------+test_composite_slice = do+    pool <- mkTestConn+    xs <- runCas pool $ do+        insert "testing" "row2" ONE [packCol (key, content)]+        get "testing" "row2" slice ONE+    let (res :: [((TLong, TBytes), LB.ByteString)]) = map unpackCol xs++    assertEqual "composite single col slice" content (snd . head $ res)+    where+      key = (TLong 125, TBytes "oklahoma")+      content = "asdf"+      slice = Range (Just (Exclusive (Single (TLong 125))))+                    (Just (Single (TLong 125)))+                    -- (Just (Single (TLong 127)))+                    Regular+                    100+      -- slice = Range (Just (TLong 125, TBytes "")) (Just (TLong 125, TBytes "zzzzz")) Regular 100+++------------------------------------------------------------------------------- -- | test quick-check generated pairs for composite column prop_composite_col_readWrite ::  CPool -> ((TLong, TBytes), LB.ByteString) -> Property prop_composite_col_readWrite pool content@(k@(TLong i, _),v) = i >= 0 ==>@@ -134,6 +158,7 @@       insert "testing" "row" ONE [packCol content]       getCol "testing" "row" (packKey k) ONE     return $ (Just content) == fmap unpackCol res+