diff --git a/riak.cabal b/riak.cabal --- a/riak.cabal +++ b/riak.cabal @@ -1,5 +1,5 @@ name: riak -version: 0.1.0.0 +version: 0.2.0.0 synopsis: A Haskell client for the Riak decentralized data store description: A Haskell client library for the Riak decentralized data @@ -17,7 +17,7 @@ . [Network.Riak.JSON] JSON for storage, manual conflict resolution. . - [Network.Riak.Value.Monoid] More complex (but still automatic) + [Network.Riak.Value.Resolvable] More complex (but still automatic) storage, automatic conflict resolution. . [Network.Riak.Value] More complex (but still automatic) storage, @@ -58,17 +58,20 @@ exposed-modules: Network.Riak + Network.Riak.Basic Network.Riak.Connection + Network.Riak.Connection.Pool Network.Riak.Content Network.Riak.Debug + Network.Riak.Escape + Network.Riak.JSON + Network.Riak.JSON.Resolvable Network.Riak.Request + Network.Riak.Resolvable Network.Riak.Response - Network.Riak.Basic Network.Riak.Types - Network.Riak.JSON - Network.Riak.JSON.Monoid Network.Riak.Value - Network.Riak.Value.Monoid + Network.Riak.Value.Resolvable Network.Riak.Protocol.ServerInfo Network.Riak.Protocol.BucketProps Network.Riak.Protocol.DeleteRequest @@ -94,27 +97,32 @@ Network.Riak.Protocol.ListBucketsRequest other-modules: - Network.Riak.Monoid - Network.Riak.Protocol.Link Network.Riak.Connection.Internal Network.Riak.Connection.NoPush + Network.Riak.Protocol Network.Riak.Protocol.Content + Network.Riak.Protocol.Link + Network.Riak.Resolvable.Internal Network.Riak.Tag Network.Riak.Types.Internal - Network.Riak.Protocol build-depends: - aeson, - attoparsec >= 0.8.4.0, + aeson == 0.2.*, + attoparsec >= 0.8.5.0, base == 4.*, binary, + blaze-builder, bytestring, containers, + hashable >= 1.0.1.2, network >= 2.3, protocol-buffers >= 1.8.0, protocol-buffers-descriptor >= 1.8.1, pureMD5, - random + random, + stm, + time, + vector >= 0.7 if flag(debug) cpp-options: -DASSERTS -DDEBUG diff --git a/src/Network/Riak.hs b/src/Network/Riak.hs --- a/src/Network/Riak.hs +++ b/src/Network/Riak.hs @@ -18,13 +18,13 @@ -- use one of the following modules (ranked from easiest to most -- tricky to use): -- --- [Network.Riak.JSON.Monoid] JSON for storage, automatic conflict +-- [Network.Riak.JSON.Resolvable] JSON for storage, automatic conflict -- resolution. (This module actually re-exports its definitions.) -- This is the easiest module to work with. -- -- [Network.Riak.JSON] JSON for storage, manual conflict resolution. -- --- [Network.Riak.Value.Monoid] More complex (but still automatic) +-- [Network.Riak.Value.Resolvable] More complex (but still automatic) -- storage, automatic conflict resolution. -- -- [Network.Riak.Value] More complex (but still automatic) storage, @@ -67,4 +67,4 @@ import Network.Riak.Connection import Network.Riak.Types import Network.Riak.Basic hiding (get, put, put_) -import Network.Riak.JSON.Monoid (get, getMany, put, putMany) +import Network.Riak.JSON.Resolvable (get, getMany, put, putMany) diff --git a/src/Network/Riak/Connection/Internal.hs b/src/Network/Riak/Connection/Internal.hs --- a/src/Network/Riak/Connection/Internal.hs +++ b/src/Network/Riak/Connection/Internal.hs @@ -38,9 +38,9 @@ import Control.Concurrent import Control.Exception (Exception, IOException, throw) -import Control.Monad (forM_, replicateM, replicateM_, unless) +import Control.Monad (forM_, replicateM, replicateM_) import Data.Binary.Put (Put, putWord32be, runPut) -import Data.IORef (modifyIORef, newIORef, readIORef, writeIORef) +import Data.IORef (newIORef, readIORef, writeIORef) import Data.Int (Int64) import Network.Riak.Connection.NoPush (setNoPush) import Network.Riak.Debug as Debug @@ -120,9 +120,16 @@ sClose connSock writeIORef connBuffer L.empty -recvWith :: (L.ByteString -> IO L.ByteString) -> Connection -> Int64 - -> IO L.ByteString -recvWith onError Connection{..} n0 +-- | We use a larger receive buffer than we usually need, and +-- generally ask to receive more data than we know we'll need, in the +-- hope that we'll be able to buffer some of it and avoid future recv +-- system calls. +recvBufferSize :: Integral a => a +recvBufferSize = 16384 +{-# INLINE recvBufferSize #-} + +recvExactly :: Connection -> Int64 -> IO L.ByteString +recvExactly Connection{..} n0 | n0 <= 0 = return L.empty | otherwise = do bs <- readIORef connBuffer @@ -130,29 +137,30 @@ len = L.length h if len == n0 then writeIORef connBuffer t >> return h - else if len == 0 - then go [] n0 - else go (reverse (L.toChunks t)) (n0-len) + else go (reverse (L.toChunks h)) (n0-len) where maxInt = fromIntegral (maxBound :: Int) + go (s:acc) n + | n < 0 = do + let (h,t) = B.splitAt (B.length s + fromIntegral n) s + writeIORef connBuffer $! L.fromChunks [t] + return $ L.fromChunks (reverse (h:acc)) go acc n - | n <= 0 = return (L.fromChunks (reverse acc)) - | otherwise = do - let n' = min n maxInt - bs <- B.recv connSock (fromIntegral n') - let len = B.length bs - if len == 0 - then onError (L.fromChunks (reverse acc)) - else go (bs:acc) (n' - fromIntegral len) - -recvExactly :: Connection -> Int64 -> IO L.ByteString -recvExactly = recvWith $ \_ -> - moduleError "recvExactly" "short read from network" + | n == 0 = do + writeIORef connBuffer L.empty + return $ L.fromChunks (reverse acc) + | otherwise = do + let n' = max recvBufferSize $ min n maxInt + bs <- B.recv connSock (fromIntegral n') + let len = B.length bs + if len == 0 + then moduleError "recvExactly" "short read from network" + else go (bs:acc) (n - fromIntegral len) recvGet :: Connection -> Get a -> IO a recvGet Connection{..} get = do let refill = do - bs <- L.recv connSock 16384 + bs <- L.recv connSock recvBufferSize if L.null bs then shutdown connSock ShutdownReceive >> return Nothing else return (Just bs) @@ -171,13 +179,10 @@ recvGetN :: Connection -> Int64 -> Get a -> IO a recvGetN conn n get = do bs <- recvExactly conn n - let finish bs' r = do - unless (L.null bs') $ modifyIORef (connBuffer conn) (`L.append` bs') - return r case runGet get bs of - Finished bs' _ r -> finish bs' r + Finished _ _ r -> return r Partial k -> case k Nothing of - Finished bs' _ r -> finish bs' r + Finished _ _ r -> return r Failed _ err -> moduleError "recvGetN" err Partial _ -> moduleError "recvGetN" "parser wants more input!?" diff --git a/src/Network/Riak/Connection/NoPush.hsc b/src/Network/Riak/Connection/NoPush.hsc --- a/src/Network/Riak/Connection/NoPush.hsc +++ b/src/Network/Riak/Connection/NoPush.hsc @@ -25,7 +25,10 @@ noPush :: CInt #if defined(TCP_NOPUSH) -noPush = #const TCP_NOPUSH +-- TCP_NOPUSH is utterly fucked on OS X 10.6. It introduces a delay +-- of about 4.5 seconds per outbound packet train. What. The. Fuck. +noPush = 0 +--noPush = #const TCP_NOPUSH #elif defined(TCP_CORK) noPush = #const TCP_CORK #else diff --git a/src/Network/Riak/Connection/Pool.hs b/src/Network/Riak/Connection/Pool.hs new file mode 100644 --- /dev/null +++ b/src/Network/Riak/Connection/Pool.hs @@ -0,0 +1,211 @@ +{-# LANGUAGE DeriveDataTypeable, NamedFieldPuns, RecordWildCards, + ScopedTypeVariables #-} + +-- | +-- Module: Network.Riak.Connection.Pool +-- Copyright: (c) 2011 MailRank, Inc. +-- License: Apache +-- Maintainer: Bryan O'Sullivan +-- Stability: experimental +-- Portability: portable +-- +-- A high-performance striped pooling abstraction for managing +-- connections to a Riak cluster. +-- +-- \"Striped\" means that a single 'Pool' consists of several +-- sub-pools, each managed independently. A stripe size of 1 is fine +-- for many applications, and probably what you should choose by +-- default. Larger stripe sizes will lead to reduced contention in +-- high-performance multicore applications, at a trade-off of causing +-- the maximum number of simultaneous connections to grow. +module Network.Riak.Connection.Pool + ( + Pool + , client + , create + , idleTime + , maxConnections + , numStripes + , withConnection + ) where + +import Control.Applicative ((<$>)) +import Control.Concurrent (forkIO, killThread, myThreadId, threadDelay) +import Control.Concurrent.STM +import Control.Exception (SomeException, catch, onException) +import Control.Monad (forM_, forever, join, liftM2, unless, when) +import Data.Hashable (hash) +import Data.List (partition) +import Data.Time.Clock (NominalDiffTime, UTCTime, diffUTCTime, getCurrentTime) +import Data.Typeable (Typeable) +import Network.Riak.Connection.Internal (connect, disconnect, makeClientID) +import Network.Riak.Debug (debug) +import Network.Riak.Types (Client(clientID), Connection) +import Prelude hiding (catch) +import System.Mem.Weak (addFinalizer) +import qualified Data.Vector as V + +-- | A single connection pool entry. +data Entry = Entry { + connection :: Connection + , lastUse :: UTCTime + -- ^ Time of last return. + } + +-- | A single striped pool. +data LocalPool = LocalPool { + connected :: TVar Int + -- ^ Count of open connections (both idle and in use). + , entries :: TVar [Entry] + -- ^ Idle entries. + } + +-- | A pool of connections to a Riak server. +-- +-- This pool is \"striped\", i.e. it consists of several sub-pools +-- that are managed independently. +-- +-- The total number of connections that can possibly be open at once +-- is 'maxConnections' * 'numStripes'. +data Pool = Pool { + client :: Client + -- ^ Client specification. The client ID is ignored, and always + -- regenerated automatically for each new connection. + , numStripes :: Int + -- ^ Stripe count. The number of distinct sub-pools to maintain. + -- The smallest acceptable value is 1. + , idleTime :: NominalDiffTime + -- ^ Amount of time for which an unused connection is kept open. + -- The smallest acceptable value is 0.5 seconds. + -- + -- The elapsed time before closing may be a little longer than + -- requested, as the reaper thread wakes at 2-second intervals. + , maxConnections :: Int + -- ^ Maximum number of connections to keep open per stripe. The + -- smallest acceptable value is 1. + -- + -- Requests for connections will block if this limit is reached on + -- a single stripe, even if other stripes have idle connections + -- available. + , localPools :: V.Vector LocalPool + -- ^ Per-capability connection pools. + } deriving (Typeable) + +instance Show Pool where + show Pool{..} = "Pool { client = " ++ show client ++ ", " ++ + "numStripes = " ++ show numStripes ++ ", " ++ + "idleTime = " ++ show idleTime ++ ", " ++ + "maxConnections = " ++ show maxConnections ++ "}" + +instance Eq Pool where + a == b = client a == client b && numStripes a == numStripes b && + idleTime a == idleTime b && maxConnections a == maxConnections b + +-- | Create a new connection pool. +create :: Client + -- ^ Client configuration. The client ID is ignored, and + -- always regenerated automatically for each new connection. + -> Int + -- ^ Stripe count. The number of distinct sub-pools to + -- maintain. The smallest acceptable value is 1. + -> NominalDiffTime + -- ^ Amount of time for which an unused connection is kept + -- open. The smallest acceptable value is 0.5 seconds. + -- + -- The elapsed time before closing may be a little longer than + -- requested, as the reaper thread wakes at 2-second intervals. + -> Int + -- ^ Maximum number of connections to keep open per stripe. + -- The smallest acceptable value is 1. + -- + -- Requests for connections will block if this limit is reached + -- on a single stripe, even if other stripes have idle + -- connections available. + -> IO Pool +create client numStripes idleTime maxConnections = do + when (numStripes < 1) $ + modError "pool " $ "invalid stripe count " ++ show numStripes + when (idleTime < 0.5) $ + modError "pool " $ "invalid idle time " ++ show idleTime + when (maxConnections < 1) $ + modError "pool " $ "invalid maximum connection count " ++ + show maxConnections + localPools <- atomically . V.replicateM numStripes $ + liftM2 LocalPool (newTVar 0) (newTVar []) + reaperId <- forkIO $ reaper idleTime localPools + let p = Pool { + client + , numStripes + , idleTime + , maxConnections + , localPools + } + addFinalizer p $ killThread reaperId + return p + +-- | Periodically go through all pools, closing any connections that +-- have been left idle for too long. +reaper :: NominalDiffTime -> V.Vector LocalPool -> IO () +reaper idleTime pools = forever $ do + threadDelay (2 * 1000000) + now <- getCurrentTime + let isStale Entry{..} = now `diffUTCTime` lastUse > idleTime + V.forM_ pools $ \LocalPool{..} -> do + conns <- atomically $ do + (stale,fresh) <- partition isStale <$> readTVar entries + unless (null stale) $ do + writeTVar entries fresh + modifyTVar_ connected (subtract (length stale)) + return (map connection stale) + forM_ conns $ \conn -> do + debug "reaper" "closing idle connection" + disconnect conn `catch` \(_::SomeException) -> return () + +-- | Temporarily take a connection from a 'Pool', perform an action +-- with it, and return it to the pool afterwards. +-- +-- * If the pool has a connection available, it is used +-- immediately. +-- +-- * Otherwise, if the maximum number of connections has not been +-- reached, a new connection is created and used. +-- +-- * If the maximum number of connections has been reached, this +-- function blocks until a connection becomes available, then that +-- connection is used. +-- +-- If the action throws an exception of any type, the 'Connection' is +-- destroyed, and not returned to the pool. +-- +-- It probably goes without saying that you should never call +-- 'disconnect' on a connection, as doing so will cause a subsequent +-- user, expecting the connection to be valid, to throw an exception. +withConnection :: Pool -> (Connection -> IO a) -> IO a +withConnection Pool{..} act = do + i <- ((`mod` numStripes) . hash) <$> myThreadId + let LocalPool{..} = localPools V.! i + conn <- join . atomically $ do + ents <- readTVar entries + case ents of + (Entry{..}:es) -> writeTVar entries es >> return (return connection) + [] -> do + inUse <- readTVar connected + when (inUse == maxConnections) retry + writeTVar connected $! inUse + 1 + return $ do + cid <- makeClientID + connect client { clientID = cid } + `onException` atomically (modifyTVar_ connected (subtract 1)) + ret <- act conn `onException` do + disconnect conn `catch` \(_::SomeException) -> return () + atomically (modifyTVar_ connected (subtract 1)) + now <- getCurrentTime + atomically $ modifyTVar_ entries (Entry conn now:) + return ret + +modifyTVar_ :: TVar a -> (a -> a) -> STM () +modifyTVar_ v f = readTVar v >>= \a -> writeTVar v $! f a + +modError :: String -> String -> a +modError func msg = + error $ "Network.Riak.Connection.Pool." ++ func ++ ": " ++ msg diff --git a/src/Network/Riak/Escape.hs b/src/Network/Riak/Escape.hs new file mode 100644 --- /dev/null +++ b/src/Network/Riak/Escape.hs @@ -0,0 +1,77 @@ +{-# LANGUAGE OverloadedStrings #-} + +-- | +-- Module: Network.Riak.Connection +-- Copyright: (c) 2011 MailRank, Inc. +-- License: Apache +-- Maintainer: Bryan O'Sullivan +-- Stability: experimental +-- Portability: portable +-- +-- Support for REST-safe name handling. +-- +-- Riak's protocol buffer (PBC) API will accept unescaped bucket, +-- link, and key names. Its REST API does not unescape names, so it +-- is possible to use the PBC API to construct names that cannot be +-- accessed via the REST API (e.g. containing an embedded slash or +-- other URL-unsafe octet). + +module Network.Riak.Escape + ( + escape + , unescape + ) where + +import Blaze.ByteString.Builder (fromByteString, toByteString) +import Blaze.ByteString.Builder.Word (fromWord8) +import Control.Applicative ((<$>)) +import Data.Attoparsec as A +import Data.Bits ((.|.), (.&.), shiftL, shiftR) +import Data.ByteString (ByteString) +import Data.Monoid (mappend, mempty) +import qualified Data.ByteString as B +import qualified Data.ByteString.Unsafe as B + +-- | URL-escape a string. +escape :: ByteString -> ByteString +escape = toByteString . B.foldl step mempty + where + step acc 32 = acc `mappend` fromWord8 43 + step acc w | literal w = acc `mappend` fromWord8 w + | otherwise = acc `mappend` hex w + literal w = w >= 97 && w <= 122 || w >= 65 && w <= 90 || + w >= 48 && w <= 57 || w `B.elem` "$-.!*'()," + hex w = fromWord8 37 `mappend` d (w `shiftR` 4) `mappend` d (w .&. 0xf) + d n | n < 10 = fromWord8 (n + 48) + | otherwise = fromWord8 (n + 87) + +-- | URL-unescape a string. +unescapeP :: Parser ByteString +unescapeP = toByteString <$> go mempty + where + go acc = do + s <- A.takeWhile $ \w -> w /= 37 && w /= 43 + let rest = do + w <- anyWord8 + if w == 43 + then go (acc `mappend` fromByteString s `mappend` fromWord8 32) + else do + h <- A.take 2 + let hex b | b >= 48 && b <= 57 = b - 48 + | b >= 97 && b <= 102 = b - 87 + | b >= 65 && b <= 70 = b - 55 + | otherwise = 255 + hi = hex (B.unsafeIndex h 0) + lo = hex (B.unsafeIndex h 1) + if hi .|. lo == 255 + then fail "invalid hex escape" + else go (acc `mappend` fromByteString s `mappend` + fromWord8 (lo .|. (hi `shiftL` 4))) + done <- atEnd + if done + then return (acc `mappend` fromByteString s) + else rest + +-- | URL-unescape a string. +unescape :: ByteString -> Either String ByteString +unescape s0 = eitherResult $ parse unescapeP s0 `feed` B.empty diff --git a/src/Network/Riak/JSON.hs b/src/Network/Riak/JSON.hs --- a/src/Network/Riak/JSON.hs +++ b/src/Network/Riak/JSON.hs @@ -27,10 +27,10 @@ import Control.Applicative ((<$>)) import Control.Arrow (first) +import Data.Aeson.Types (FromJSON(..), ToJSON(..)) import Data.Monoid (Monoid) import Data.Typeable (Typeable) import Network.Riak.Types.Internal -import Data.Aeson.Types (FromJSON(..), ToJSON(..)) import qualified Network.Riak.Value as V newtype JSON a = J { @@ -48,8 +48,8 @@ {-# INLINE fmap #-} instance (FromJSON a, ToJSON a) => V.IsContent (JSON a) where - fromContent c = J `fmap` (V.fromContent c >>= fromJSON) - {-# INLINE fromContent #-} + parseContent c = J `fmap` (V.parseContent c >>= parseJSON) + {-# INLINE parseContent #-} toContent (J a) = V.toContent (toJSON a) {-# INLINE toContent #-} diff --git a/src/Network/Riak/JSON/Monoid.hs b/src/Network/Riak/JSON/Monoid.hs deleted file mode 100644 --- a/src/Network/Riak/JSON/Monoid.hs +++ /dev/null @@ -1,127 +0,0 @@ --- | --- Module: Network.Riak.JSON.Monoid --- Copyright: (c) 2011 MailRank, Inc. --- License: Apache --- Maintainer: Bryan O'Sullivan --- Stability: experimental --- Portability: portable --- --- This module allows storage and retrieval of JSON-encoded data. --- --- Functions automatically resolve conflicts using 'Monoid' instances. --- For instance, if a 'get' returns three siblings, a winner will be --- chosen using 'mconcat'. If a 'put' results in a conflict, a winner --- will be chosen using 'mconcat', and the winner will be 'put'; this --- will be repeated until no conflict occurs. - -module Network.Riak.JSON.Monoid - ( - get - , getMany - , put - , put_ - , putMany - , putMany_ - ) where - -import Data.Aeson.Types (FromJSON(..), ToJSON(..)) -import Data.Monoid (Dual(..), First(..), Last(..), Monoid) -import Network.Riak.Types.Internal hiding (MessageTag(..)) -import qualified Network.Riak.JSON as J -import qualified Network.Riak.Monoid as M - -instance ToJSON a => ToJSON (Dual a) where - toJSON = toJSON . getDual - {-# INLINE toJSON #-} - -instance FromJSON a => FromJSON (Dual a) where - fromJSON = fmap Dual . fromJSON - {-# INLINE fromJSON #-} - -instance ToJSON a => ToJSON (First a) where - toJSON = toJSON . getFirst - {-# INLINE toJSON #-} - -instance FromJSON a => FromJSON (First a) where - fromJSON = fmap First . fromJSON - {-# INLINE fromJSON #-} - -instance ToJSON a => ToJSON (Last a) where - toJSON = toJSON . getLast - {-# INLINE toJSON #-} - -instance FromJSON a => FromJSON (Last a) where - fromJSON = fmap Last . fromJSON - {-# INLINE fromJSON #-} - --- | Retrieve a single value. If conflicting values are returned, the --- 'Monoid' is used to choose a winner. -get :: (FromJSON c, ToJSON c, Monoid c) => - Connection -> Bucket -> Key -> R -> IO (Maybe (c, VClock)) -get = M.get J.get -{-# INLINE get #-} - --- | Retrieve multiple values. If conflicting values are returned for --- a key, the 'Monoid' is used to choose a winner. -getMany :: (FromJSON c, ToJSON c, Monoid c) - => Connection -> Bucket -> [Key] -> R -> IO [Maybe (c, VClock)] -getMany = M.getMany J.getMany -{-# INLINE getMany #-} - --- | Store a single value, automatically resolving any vector clock --- conflicts that arise. A single invocation of this function may --- involve several roundtrips to the server to resolve conflicts. --- --- If a conflict arises, a winner will be chosen using 'mconcat', and --- the winner will be stored; this will be repeated until no conflict --- occurs. --- --- The final value to be stored at the end of any conflict resolution --- is returned. -put :: (FromJSON c, ToJSON c, Monoid c) => - Connection -> Bucket -> Key -> Maybe VClock -> c -> W -> DW - -> IO (c, VClock) -put = M.put J.put -{-# INLINE put #-} - --- | Store a single value, automatically resolving any vector clock --- conflicts that arise. A single invocation of this function may --- involve several roundtrips to the server to resolve conflicts. --- --- If a conflict arises, a winner will be chosen using 'mconcat', and --- the winner will be stored; this will be repeated until no conflict --- occurs. -put_ :: (FromJSON c, ToJSON c, Monoid c) => - Connection -> Bucket -> Key -> Maybe VClock -> c -> W -> DW - -> IO () -put_ = M.put_ J.put -{-# INLINE put_ #-} - --- | Store multiple values, resolving any vector clock conflicts that --- arise. A single invocation of this function may involve several --- roundtrips to the server to resolve conflicts. --- --- If any conflicts arise, a winner will be chosen in each case using --- 'mconcat', and the winners will be stored; this will be repeated --- until no conflicts occur. --- --- For each original value to be stored, the final value that was --- stored at the end of any conflict resolution is returned. -putMany :: (FromJSON c, ToJSON c, Monoid c) => - Connection -> Bucket -> [(Key, Maybe VClock, c)] -> W -> DW - -> IO [(c, VClock)] -putMany = M.putMany J.putMany -{-# INLINE putMany #-} - --- | Store multiple values, resolving any vector clock conflicts that --- arise. A single invocation of this function may involve several --- roundtrips to the server to resolve conflicts. --- --- If any conflicts arise, a winner will be chosen in each case using --- 'mconcat', and the winners will be stored; this will be repeated --- until no conflicts occur. -putMany_ :: (FromJSON c, ToJSON c, Monoid c) => - Connection -> Bucket -> [(Key, Maybe VClock, c)] -> W -> DW - -> IO () -putMany_ = M.putMany_ J.putMany -{-# INLINE putMany_ #-} diff --git a/src/Network/Riak/JSON/Resolvable.hs b/src/Network/Riak/JSON/Resolvable.hs new file mode 100644 --- /dev/null +++ b/src/Network/Riak/JSON/Resolvable.hs @@ -0,0 +1,103 @@ +-- | +-- Module: Network.Riak.JSON.Resolvable +-- Copyright: (c) 2011 MailRank, Inc. +-- License: Apache +-- Maintainer: Bryan O'Sullivan +-- Stability: experimental +-- Portability: portable +-- +-- This module allows storage and retrieval of JSON-encoded data. +-- +-- Functions automatically resolve conflicts using 'Resolvable' instances. +-- For instance, if a 'get' returns three siblings, a winner will be +-- chosen using 'mconcat'. If a 'put' results in a conflict, a winner +-- will be chosen using 'mconcat', and the winner will be 'put'; this +-- will be repeated until no conflict occurs. + +module Network.Riak.JSON.Resolvable + ( + get + , getMany + , put + , put_ + , putMany + , putMany_ + ) where + +import Data.Aeson.Types (FromJSON(..), ToJSON(..)) +import Network.Riak.Resolvable.Internal (Resolvable) +import Network.Riak.Types.Internal hiding (MessageTag(..)) +import qualified Network.Riak.JSON as J +import qualified Network.Riak.Resolvable.Internal as R + +-- | Retrieve a single value. If conflicting values are returned, the +-- 'Resolvable' is used to choose a winner. +get :: (FromJSON c, ToJSON c, Resolvable c) => + Connection -> Bucket -> Key -> R -> IO (Maybe (c, VClock)) +get = R.get J.get +{-# INLINE get #-} + +-- | Retrieve multiple values. If conflicting values are returned for +-- a key, the 'Resolvable' is used to choose a winner. +getMany :: (FromJSON c, ToJSON c, Resolvable c) + => Connection -> Bucket -> [Key] -> R -> IO [Maybe (c, VClock)] +getMany = R.getMany J.getMany +{-# INLINE getMany #-} + +-- | Store a single value, automatically resolving any vector clock +-- conflicts that arise. A single invocation of this function may +-- involve several roundtrips to the server to resolve conflicts. +-- +-- If a conflict arises, a winner will be chosen using 'mconcat', and +-- the winner will be stored; this will be repeated until no conflict +-- occurs. +-- +-- The final value to be stored at the end of any conflict resolution +-- is returned. +put :: (Eq c, FromJSON c, ToJSON c, Resolvable c) => + Connection -> Bucket -> Key -> Maybe VClock -> c -> W -> DW + -> IO (c, VClock) +put = R.put J.put +{-# INLINE put #-} + +-- | Store a single value, automatically resolving any vector clock +-- conflicts that arise. A single invocation of this function may +-- involve several roundtrips to the server to resolve conflicts. +-- +-- If a conflict arises, a winner will be chosen using 'mconcat', and +-- the winner will be stored; this will be repeated until no conflict +-- occurs. +put_ :: (Eq c, FromJSON c, ToJSON c, Resolvable c) => + Connection -> Bucket -> Key -> Maybe VClock -> c -> W -> DW + -> IO () +put_ = R.put_ J.put +{-# INLINE put_ #-} + +-- | Store multiple values, resolving any vector clock conflicts that +-- arise. A single invocation of this function may involve several +-- roundtrips to the server to resolve conflicts. +-- +-- If any conflicts arise, a winner will be chosen in each case using +-- 'mconcat', and the winners will be stored; this will be repeated +-- until no conflicts occur. +-- +-- For each original value to be stored, the final value that was +-- stored at the end of any conflict resolution is returned. +putMany :: (Eq c, FromJSON c, ToJSON c, Resolvable c) => + Connection -> Bucket -> [(Key, Maybe VClock, c)] -> W -> DW + -> IO [(c, VClock)] +putMany = R.putMany J.putMany +{-# INLINE putMany #-} + +-- | Store multiple values, resolving any vector clock conflicts that +-- arise. A single invocation of this function may involve several +-- roundtrips to the server to resolve conflicts. +-- +-- If any conflicts arise, a winner will be chosen in each case using +-- 'mconcat', and the winners will be stored; this will be repeated +-- until no conflicts occur. +putMany_ :: (Eq c, FromJSON c, ToJSON c, Resolvable c) => + Connection -> Bucket -> [(Key, Maybe VClock, c)] -> W -> DW + -> IO () +putMany_ = R.putMany_ J.putMany +{-# INLINE putMany_ #-} diff --git a/src/Network/Riak/Monoid.hs b/src/Network/Riak/Monoid.hs deleted file mode 100644 --- a/src/Network/Riak/Monoid.hs +++ /dev/null @@ -1,83 +0,0 @@ --- | --- Module: Network.Riak.Monoid --- Copyright: (c) 2011 MailRank, Inc. --- License: Apache --- Maintainer: Bryan O'Sullivan --- Stability: experimental --- Portability: portable --- --- Storage and retrieval of monoidal data with automatic conflict resolution. - -module Network.Riak.Monoid - ( - get - , getMany - , put - , put_ - , putMany - , putMany_ - ) where - -import Control.Arrow (first, second) -import Data.Function (on) -import Data.List (partition, sortBy) -import Data.Monoid (Monoid(..)) -import Network.Riak.Types.Internal hiding (MessageTag(..)) -import qualified Data.IntMap as M - -get :: (Monoid c) => - (Connection -> Bucket -> Key -> R -> IO (Maybe ([c], VClock))) - -> (Connection -> Bucket -> Key -> R -> IO (Maybe (c, VClock))) -get doGet conn bucket key r = - fmap (first mconcat) `fmap` doGet conn bucket key r - -getMany :: (Monoid c) => - (Connection -> Bucket -> [Key] -> R -> IO [Maybe ([c], VClock)]) - -> Connection -> Bucket -> [Key] -> R -> IO [Maybe (c, VClock)] -getMany doGet conn b ks r = map (fmap (first mconcat)) `fmap` doGet conn b ks r - -put :: Monoid c => (Connection -> Bucket -> Key -> Maybe VClock -> c -> W -> DW - -> IO ([c], VClock)) - -> Connection -> Bucket -> Key -> Maybe VClock -> c -> W -> DW - -> IO (c, VClock) -put doPut conn bucket key mvclock0 val0 w dw = do - let go val mvclock1 = do - (xs, vclock) <- doPut conn bucket key mvclock1 val w dw - case xs of - [c] -> return (c, vclock) - _ -> go (mconcat xs) (Just vclock) - go val0 mvclock0 - -put_ :: Monoid c => (Connection -> Bucket -> Key -> Maybe VClock -> c -> W -> DW - -> IO ([c], VClock)) - -> Connection -> Bucket -> Key -> Maybe VClock -> c -> W -> DW - -> IO () -put_ doPut conn bucket key mvclock0 val0 w dw = - put doPut conn bucket key mvclock0 val0 w dw >> return () -{-# INLINE put_ #-} - -putMany :: (Monoid c) => - (Connection -> Bucket -> [(Key, Maybe VClock, c)] -> W -> DW - -> IO [([c], VClock)]) - -> Connection -> Bucket -> [(Key, Maybe VClock, c)] -> W -> DW - -> IO [(c, VClock)] -putMany doPut conn bucket puts0 w dw = go [] . zip [(0::Int)..] $ puts0 where - go acc [] = return . map snd . sortBy (compare `on` fst) $ acc - go acc puts = do - rs <- doPut conn bucket (map snd puts) w dw - let (conflicts, ok) = partition isConflict $ zip (map fst puts) rs - isConflict (_,(_:_:_,_)) = True - isConflict _ = False - go (map (second (first mconcat)) ok++acc) (map asPut conflicts) - asPut (i,(c,v)) = (i,(keys M.! i, Just v, mconcat c)) - keys = M.fromAscList (zip [(0::Int)..] (map fst3 puts0)) - fst3 (a,_,_) = a - -putMany_ :: (Monoid c) => - (Connection -> Bucket -> [(Key, Maybe VClock, c)] -> W -> DW - -> IO [([c], VClock)]) - -> Connection -> Bucket -> [(Key, Maybe VClock, c)] -> W -> DW - -> IO () -putMany_ doPut conn bucket puts0 w dw = - putMany doPut conn bucket puts0 w dw >> return () -{-# INLINE putMany_ #-} diff --git a/src/Network/Riak/Resolvable.hs b/src/Network/Riak/Resolvable.hs new file mode 100644 --- /dev/null +++ b/src/Network/Riak/Resolvable.hs @@ -0,0 +1,17 @@ +-- | +-- Module: Network.Riak.Resolvable +-- Copyright: (c) 2011 MailRank, Inc. +-- License: Apache +-- Maintainer: Bryan O'Sullivan +-- Stability: experimental +-- Portability: portable +-- +-- Storage and retrieval of data with automatic conflict resolution. + +module Network.Riak.Resolvable + ( + Resolvable(..) + , ResolvableMonoid(..) + ) where + +import Network.Riak.Resolvable.Internal (Resolvable(..), ResolvableMonoid(..)) diff --git a/src/Network/Riak/Resolvable/Internal.hs b/src/Network/Riak/Resolvable/Internal.hs new file mode 100644 --- /dev/null +++ b/src/Network/Riak/Resolvable/Internal.hs @@ -0,0 +1,131 @@ +{-# LANGUAGE DeriveDataTypeable, GeneralizedNewtypeDeriving #-} +-- | +-- Module: Network.Riak.Resolvable.Internal +-- Copyright: (c) 2011 MailRank, Inc. +-- License: Apache +-- Maintainer: Bryan O'Sullivan +-- Stability: experimental +-- Portability: portable +-- +-- Storage and retrieval of data with automatic conflict resolution. + +module Network.Riak.Resolvable.Internal + ( + Resolvable(..) + , ResolvableMonoid(..) + , get + , getMany + , put + , put_ + , putMany + , putMany_ + ) where + +import Control.Arrow (first) +import Control.Monad (unless) +import Data.Aeson.Types (FromJSON, ToJSON) +import Data.Data (Data) +import Data.Either (partitionEithers) +import Data.Function (on) +import Data.List (foldl', sortBy) +import Data.Monoid (Monoid(mappend)) +import Data.Typeable (Typeable) +import Network.Riak.Debug (debug) +import Network.Riak.Types.Internal hiding (MessageTag(..)) + +-- | A type that can automatically resolve a vector clock conflict +-- between two or more versions of a value. +-- +-- Instances must be symmetric in their behaviour, such that the +-- following law is obeyed: +-- +-- > resolve a b == resolve b a +-- +-- Otherwise, there are no restrictions on the behaviour of 'resolve'. +-- The result may be @a@, @b@, a value derived from @a@ and @b@, or +-- something else. +-- +-- If several conflicting siblings are found, 'resolve' will be +-- applied over all of them using a fold, to yield a single +-- \"winner\". +class Resolvable a where + -- | Resolve a conflict between two values. + resolve :: a -> a -> a + +-- | A newtype wrapper that uses the 'mappend' method of a type's +-- 'Monoid' instance to perform vector clock conflict resolution. +newtype ResolvableMonoid a = RM { unRM :: a } + deriving (Eq, Ord, Read, Show, Typeable, Data, Monoid, FromJSON, ToJSON) + +instance (Monoid a) => Resolvable (ResolvableMonoid a) where + resolve = mappend + {-# INLINE resolve #-} + +get :: (Resolvable a) => + (Connection -> Bucket -> Key -> R -> IO (Maybe ([a], VClock))) + -> (Connection -> Bucket -> Key -> R -> IO (Maybe (a, VClock))) +get doGet conn bucket key r = + fmap (first resolveMany) `fmap` doGet conn bucket key r + +getMany :: (Resolvable a) => + (Connection -> Bucket -> [Key] -> R -> IO [Maybe ([a], VClock)]) + -> Connection -> Bucket -> [Key] -> R -> IO [Maybe (a, VClock)] +getMany doGet conn b ks r = map (fmap (first resolveMany)) `fmap` doGet conn b ks r + +put :: (Eq a, Resolvable a) => + (Connection -> Bucket -> Key -> Maybe VClock -> a -> W -> DW + -> IO ([a], VClock)) + -> Connection -> Bucket -> Key -> Maybe VClock -> a -> W -> DW + -> IO (a, VClock) +put doPut conn bucket key mvclock0 val0 w dw = do + let go val mvclock1 = do + (xs, vclock) <- doPut conn bucket key mvclock1 val w dw + case xs of + [] -> return (val, vclock) -- not observed in the wild + [v] | v == val -> return (val, vclock) + ys -> do debug "put" "conflict" + go (resolveMany' val ys) (Just vclock) + go val0 mvclock0 + +put_ :: (Eq a, Resolvable a) => + (Connection -> Bucket -> Key -> Maybe VClock -> a -> W -> DW + -> IO ([a], VClock)) + -> Connection -> Bucket -> Key -> Maybe VClock -> a -> W -> DW + -> IO () +put_ doPut conn bucket key mvclock0 val0 w dw = + put doPut conn bucket key mvclock0 val0 w dw >> return () +{-# INLINE put_ #-} + +putMany :: (Eq a, Resolvable a) => + (Connection -> Bucket -> [(Key, Maybe VClock, a)] -> W -> DW + -> IO [([a], VClock)]) + -> Connection -> Bucket -> [(Key, Maybe VClock, a)] -> W -> DW + -> IO [(a, VClock)] +putMany doPut conn bucket puts0 w dw = go [] . zip [(0::Int)..] $ puts0 where + go acc [] = return . map snd . sortBy (compare `on` fst) $ acc + go acc puts = do + rs <- doPut conn bucket (map snd puts) w dw + let (conflicts, ok) = partitionEithers $ zipWith mush puts rs + unless (null conflicts) $ + debug "putMany" $ show (length conflicts) ++ " conflicts" + go (ok++acc) conflicts + mush (i,(k,_,c)) (cs,v) = + case cs of + [] -> Right (i,(c,v)) -- not observed in the wild + [x] | x == c -> Right (i,(c,v)) + _ -> Left (i,(k,Just v, resolveMany' c cs)) + +putMany_ :: (Eq a, Resolvable a) => + (Connection -> Bucket -> [(Key, Maybe VClock, a)] -> W -> DW + -> IO [([a], VClock)]) + -> Connection -> Bucket -> [(Key, Maybe VClock, a)] -> W -> DW -> IO () +putMany_ doPut conn bucket puts0 w dw = + putMany doPut conn bucket puts0 w dw >> return () +{-# INLINE putMany_ #-} + +resolveMany' :: (Resolvable a) => a -> [a] -> a +resolveMany' a as = foldl' resolve a as + +resolveMany :: (Resolvable a) => [a] -> a +resolveMany (a:as) = resolveMany' a as +resolveMany _ = error "resolveMany: empty list" diff --git a/src/Network/Riak/Types/Internal.hs b/src/Network/Riak/Types/Internal.hs --- a/src/Network/Riak/Types/Internal.hs +++ b/src/Network/Riak/Types/Internal.hs @@ -1,4 +1,4 @@ -{-# LANGUAGE DeriveDataTypeable, FunctionalDependencies, MultiParamTypeClasses, +{-# LANGUAGE DeriveDataTypeable, FunctionalDependencies, MultiParamTypeClasses, RecordWildCards #-} -- | diff --git a/src/Network/Riak/Value.hs b/src/Network/Riak/Value.hs --- a/src/Network/Riak/Value.hs +++ b/src/Network/Riak/Value.hs @@ -1,4 +1,4 @@ -{-# LANGUAGE OverloadedStrings, RecordWildCards #-} +{-# LANGUAGE GeneralizedNewtypeDeriving, OverloadedStrings, RecordWildCards, StandaloneDeriving #-} -- | -- Module: Network.Riak.Value @@ -17,6 +17,7 @@ module Network.Riak.Value ( IsContent(..) + , fromContent , get , getMany , put @@ -25,46 +26,57 @@ , putMany_ ) where -import Data.Attoparsec.Lazy (maybeResult, parse) +import Control.Applicative +import Data.Aeson.Types (Parser, Result(..), parse) import Data.Foldable (toList) import Network.Riak.Connection.Internal import Network.Riak.Protocol.Content (Content(..)) import Network.Riak.Protocol.GetResponse (GetResponse(..)) import Network.Riak.Protocol.PutResponse (PutResponse(..)) +import Network.Riak.Resolvable (ResolvableMonoid(..)) import Network.Riak.Types.Internal hiding (MessageTag(..)) import qualified Data.Aeson.Parser as Aeson import qualified Data.Aeson.Types as Aeson +import qualified Data.Attoparsec.Lazy as A import qualified Data.ByteString.Lazy as L import qualified Data.Sequence as Seq import qualified Network.Riak.Content as C import qualified Network.Riak.Request as Req +fromContent :: IsContent c => Content -> Maybe c +fromContent c = case parse parseContent c of + Success a -> Just a + Error _ -> Nothing + class IsContent c where - fromContent :: Content -> Maybe c + parseContent :: Content -> Parser c toContent :: c -> Content instance IsContent Content where - fromContent = Just - {-# INLINE fromContent #-} + parseContent = return + {-# INLINE parseContent #-} toContent v = v {-# INLINE toContent #-} instance IsContent () where - fromContent c | c == C.empty = Just () - | otherwise = Nothing - {-# INLINE fromContent #-} + parseContent c | c == C.empty = pure () + | otherwise = empty + {-# INLINE parseContent #-} toContent _ = C.empty {-# INLINE toContent #-} instance IsContent Aeson.Value where - fromContent c | content_type c == Just "application/json" = - maybeResult (parse Aeson.json (value c)) - | otherwise = Nothing - + parseContent c | content_type c == Just "application/json" = + case A.parse Aeson.json (value c) of + A.Done _ a -> return a + A.Fail _ _ err -> fail err + | otherwise = fail "non-JSON document" toContent = C.json {-# INLINE toContent #-} + +deriving instance (IsContent a) => IsContent (ResolvableMonoid a) put :: (IsContent c) => Connection -> Bucket -> Key -> Maybe VClock -> c -> W -> DW -> IO ([c], VClock) diff --git a/src/Network/Riak/Value/Monoid.hs b/src/Network/Riak/Value/Monoid.hs deleted file mode 100644 --- a/src/Network/Riak/Value/Monoid.hs +++ /dev/null @@ -1,105 +0,0 @@ --- | --- Module: Network.Riak.Value.Monoid --- Copyright: (c) 2011 MailRank, Inc. --- License: Apache --- Maintainer: Bryan O'Sullivan --- Stability: experimental --- Portability: portable --- --- This module allows storage and retrieval of data encoded using the --- 'V.IsContent' typeclass. This provides access to more of Riak's --- storage features than JSON, e.g. links. --- --- Functions automatically resolve conflicts using 'Monoid' instances. --- For instance, if a 'get' returns three siblings, a winner will be --- chosen using 'mconcat'. If a 'put' results in a conflict, a winner --- will be chosen using 'mconcat', and the winner will be 'put'; this --- will be repeated until no conflict occurs. - -module Network.Riak.Value.Monoid - ( - V.IsContent(..) - , get - , getMany - , put - , put_ - , putMany - , putMany_ - ) where - -import Data.Monoid (Monoid(..)) -import Network.Riak.Types.Internal hiding (MessageTag(..)) -import qualified Network.Riak.Monoid as M -import qualified Network.Riak.Value as V - --- | Retrieve a single value. If conflicting values are returned, the --- 'Monoid' is used to choose a winner. -get :: (Monoid c, V.IsContent c) => - Connection -> Bucket -> Key -> R -> IO (Maybe (c, VClock)) -get = M.get V.get -{-# INLINE get #-} - --- | Retrieve multiple values. If conflicting values are returned for --- a key, the 'Monoid' is used to choose a winner. -getMany :: (Monoid c, V.IsContent c) => Connection -> Bucket -> [Key] -> R - -> IO [Maybe (c, VClock)] -getMany = M.getMany V.getMany -{-# INLINE getMany #-} - --- | Store a single value, automatically resolving any vector clock --- conflicts that arise. A single invocation of this function may --- involve several roundtrips to the server to resolve conflicts. --- --- If a conflict arises, a winner will be chosen using 'mconcat', and --- the winner will be stored; this will be repeated until no conflict --- occurs. --- --- The final value to be stored at the end of any conflict resolution --- is returned. -put :: (Monoid c, V.IsContent c) => - Connection -> Bucket -> Key -> Maybe VClock -> c -> W -> DW - -> IO (c, VClock) -put = M.put V.put -{-# INLINE put #-} - --- | Store a single value, automatically resolving any vector clock --- conflicts that arise. A single invocation of this function may --- involve several roundtrips to the server to resolve conflicts. --- --- If a conflict arises, a winner will be chosen using 'mconcat', and --- the winner will be stored; this will be repeated until no conflict --- occurs. -put_ :: (Monoid c, V.IsContent c) => - Connection -> Bucket -> Key -> Maybe VClock -> c -> W -> DW - -> IO () -put_ = M.put_ V.put -{-# INLINE put_ #-} - --- | Store multiple values, resolving any vector clock conflicts that --- arise. A single invocation of this function may involve several --- roundtrips to the server to resolve conflicts. --- --- If any conflicts arise, a winner will be chosen in each case using --- 'mconcat', and the winners will be stored; this will be repeated --- until no conflicts occur. --- --- For each original value to be stored, the final value that was --- stored at the end of any conflict resolution is returned. -putMany :: (Monoid c, V.IsContent c) => - Connection -> Bucket -> [(Key, Maybe VClock, c)] -> W -> DW - -> IO [(c, VClock)] -putMany = M.putMany V.putMany -{-# INLINE putMany #-} - --- | Store multiple values, resolving any vector clock conflicts that --- arise. A single invocation of this function may involve several --- roundtrips to the server to resolve conflicts. --- --- If any conflicts arise, a winner will be chosen in each case using --- 'mconcat', and the winners will be stored; this will be repeated --- until no conflicts occur. -putMany_ :: (Monoid c, V.IsContent c) => - Connection -> Bucket -> [(Key, Maybe VClock, c)] -> W -> DW - -> IO () -putMany_ = M.putMany_ V.putMany -{-# INLINE putMany_ #-} diff --git a/src/Network/Riak/Value/Resolvable.hs b/src/Network/Riak/Value/Resolvable.hs new file mode 100644 --- /dev/null +++ b/src/Network/Riak/Value/Resolvable.hs @@ -0,0 +1,104 @@ +-- | +-- Module: Network.Riak.Value.Resolvable +-- Copyright: (c) 2011 MailRank, Inc. +-- License: Apache +-- Maintainer: Bryan O'Sullivan +-- Stability: experimental +-- Portability: portable +-- +-- This module allows storage and retrieval of data encoded using the +-- 'V.IsContent' typeclass. This provides access to more of Riak's +-- storage features than JSON, e.g. links. +-- +-- Functions automatically resolve conflicts using 'Resolvable' instances. +-- For instance, if a 'get' returns three siblings, a winner will be +-- chosen using 'mconcat'. If a 'put' results in a conflict, a winner +-- will be chosen using 'mconcat', and the winner will be 'put'; this +-- will be repeated until no conflict occurs. + +module Network.Riak.Value.Resolvable + ( + V.IsContent(..) + , get + , getMany + , put + , put_ + , putMany + , putMany_ + ) where + +import Network.Riak.Resolvable.Internal (Resolvable) +import Network.Riak.Types.Internal hiding (MessageTag(..)) +import qualified Network.Riak.Resolvable.Internal as R +import qualified Network.Riak.Value as V + +-- | Retrieve a single value. If conflicting values are returned, the +-- 'Resolvable' is used to choose a winner. +get :: (Resolvable a, V.IsContent a) => + Connection -> Bucket -> Key -> R -> IO (Maybe (a, VClock)) +get = R.get V.get +{-# INLINE get #-} + +-- | Retrieve multiple values. If conflicting values are returned for +-- a key, the 'Resolvable' is used to choose a winner. +getMany :: (Resolvable a, V.IsContent a) => Connection -> Bucket -> [Key] -> R + -> IO [Maybe (a, VClock)] +getMany = R.getMany V.getMany +{-# INLINE getMany #-} + +-- | Store a single value, automatically resolving any vector clock +-- conflicts that arise. A single invocation of this function may +-- involve several roundtrips to the server to resolve conflicts. +-- +-- If a conflict arises, a winner will be chosen using 'mconcat', and +-- the winner will be stored; this will be repeated until no conflict +-- occurs. +-- +-- The final value to be stored at the end of any conflict resolution +-- is returned. +put :: (Eq a, Resolvable a, V.IsContent a) => + Connection -> Bucket -> Key -> Maybe VClock -> a -> W -> DW + -> IO (a, VClock) +put = R.put V.put +{-# INLINE put #-} + +-- | Store a single value, automatically resolving any vector clock +-- conflicts that arise. A single invocation of this function may +-- involve several roundtrips to the server to resolve conflicts. +-- +-- If a conflict arises, a winner will be chosen using 'mconcat', and +-- the winner will be stored; this will be repeated until no conflict +-- occurs. +put_ :: (Eq a, Resolvable a, V.IsContent a) => + Connection -> Bucket -> Key -> Maybe VClock -> a -> W -> DW + -> IO () +put_ = R.put_ V.put +{-# INLINE put_ #-} + +-- | Store multiple values, resolving any vector clock conflicts that +-- arise. A single invocation of this function may involve several +-- roundtrips to the server to resolve conflicts. +-- +-- If any conflicts arise, a winner will be chosen in each case using +-- 'mconcat', and the winners will be stored; this will be repeated +-- until no conflicts occur. +-- +-- For each original value to be stored, the final value that was +-- stored at the end of any conflict resolution is returned. +putMany :: (Eq a, Resolvable a, V.IsContent a) => + Connection -> Bucket -> [(Key, Maybe VClock, a)] -> W -> DW + -> IO [(a, VClock)] +putMany = R.putMany V.putMany +{-# INLINE putMany #-} + +-- | Store multiple values, resolving any vector clock conflicts that +-- arise. A single invocation of this function may involve several +-- roundtrips to the server to resolve conflicts. +-- +-- If any conflicts arise, a winner will be chosen in each case using +-- 'mconcat', and the winners will be stored; this will be repeated +-- until no conflicts occur. +putMany_ :: (Eq a, Resolvable a, V.IsContent a) => + Connection -> Bucket -> [(Key, Maybe VClock, a)] -> W -> DW -> IO () +putMany_ = R.putMany_ V.putMany +{-# INLINE putMany_ #-}