diff --git a/Changes.md b/Changes.md --- a/Changes.md +++ b/Changes.md @@ -1,3 +1,6 @@ +* 1.1.0.0 + - Adds bucket type as argument to many functions. + - Bugfix for exceptions being ignored (https://github.com/markhibberd/riak-haskell-client/pull/46) * 1.0 - No longer expected to work with riak <= 2 - CRDT functionality, including high-level interface diff --git a/riak.cabal b/riak.cabal --- a/riak.cabal +++ b/riak.cabal @@ -1,5 +1,5 @@ name: riak -version: 1.0.1.1 +version: 1.1.0.0 synopsis: A Haskell client for the Riak decentralized data store description: A Haskell client library for the Riak decentralized data @@ -41,7 +41,7 @@ category: Network build-type: Simple extra-source-files: - README.markdown + README.markdown Changes.md cabal-version: >=1.8 @@ -99,31 +99,32 @@ Network.Riak.Tag build-depends: - aeson >= 0.8 && < 0.12, + aeson >= 0.8 && < 0.12, attoparsec >= 0.12.1.6 && < 0.14, - base >= 3 && <5, + base >= 3 && <5, binary, - blaze-builder >= 0.3 && <= 0.5, + blaze-builder >= 0.3 && <= 0.5, bytestring, containers, data-default-class >= 0.0.1, deepseq >= 1.3, - enclosed-exceptions >= 1.0.1.1 && <= 1.1, - exceptions >= 0.8.0.2 && <= 0.9, + enclosed-exceptions >= 1.0.1.1 && <= 1.1, + exceptions >= 0.8.0.2 && <= 0.9, hashable >= 1.2.3, - transformers >= 0.3 && < 0.5, - mersenne-random-pure64 >= 0.2.0.4 && < 0.3, - monad-control >= 1.0.0.4 && < 1.1, + transformers >= 0.3 && < 0.6, + transformers-base == 0.4.*, + mersenne-random-pure64 >= 0.2.0.4 && < 0.3, + monad-control >= 1.0.0.4 && < 1.1, network >= 2.3, resource-pool == 0.2.*, - protocol-buffers >= 2.1.4 && < 2.3, + protocol-buffers >= 2.1.4 && < 2.5, pureMD5, random, - random-shuffle >= 0.0.4 && < 0.1, riak-protobuf == 0.21.*, semigroups >= 0.16, + stm == 2.4.*, text == 1.2.*, - time >= 1.4.2 && < 1.6, + time >= 1.4.2 && < 1.7, vector >= 0.10.12.3 && < 0.12, unordered-containers >= 0.2.5 @@ -155,6 +156,7 @@ build-depends: base, riak, + riak-protobuf, bytestring, containers, HUnit, diff --git a/src/Network/Riak/CRDT.hs b/src/Network/Riak/CRDT.hs --- a/src/Network/Riak/CRDT.hs +++ b/src/Network/Riak/CRDT.hs @@ -10,8 +10,8 @@ * Haskell values: 'Counter', 'Set' etc * ADT for operations: 'CounterOp', 'SetOp' etc - - * 'modify' to locally modify a value (matching riak behaviour) + + * 'modify' to locally modify a value (matching riak-side behaviour) * Riak-side diff --git a/src/Network/Riak/CRDT/Riak.hs b/src/Network/Riak/CRDT/Riak.hs --- a/src/Network/Riak/CRDT/Riak.hs +++ b/src/Network/Riak/CRDT/Riak.hs @@ -4,12 +4,23 @@ -- license: Apache -- -- CRDT operations -module Network.Riak.CRDT.Riak where +{-# LANGUAGE ScopedTypeVariables, MultiWayIf, OverloadedStrings #-} +module Network.Riak.CRDT.Riak (counterSendUpdate, + setSendUpdate, + mapSendUpdate, + get) + where + import Control.Applicative import qualified Network.Riak.CRDT.Types as CRDT import qualified Network.Riak.Connection as Conn import Network.Riak.Types +import qualified Network.Riak.Protocol.ErrorResponse as ER +import Data.Int +import Control.Applicative +import Control.Exception (catchJust) +import qualified Data.ByteString.Lazy as BS import qualified Network.Riak.CRDT.Request as Req import qualified Network.Riak.CRDT.Response as Resp @@ -17,19 +28,37 @@ counterSendUpdate :: Connection -> BucketType -> Bucket -> Key -> [CRDT.CounterOp] -> IO () -counterSendUpdate conn t b k ops = Conn.exchange_ conn (Req.counterUpdate ops t b k) +counterSendUpdate conn t b k ops = Conn.exchange_ conn $ Req.counterUpdate ops t b k setSendUpdate :: Connection -> BucketType -> Bucket -> Key -> [CRDT.SetOp] -> IO () -setSendUpdate conn t b k ops = Conn.exchange_ conn (Req.setUpdate ops t b k) +setSendUpdate conn t b k ops = handleEmpty . Conn.exchange_ conn $ Req.setUpdate ops t b k mapSendUpdate :: Connection -> BucketType -> Bucket -> Key -> [CRDT.MapOp] -> IO () -mapSendUpdate conn t b k ops = Conn.exchange_ conn (Req.mapUpdate ops t b k) +mapSendUpdate conn t b k ops = handleEmpty . Conn.exchange_ conn $ Req.mapUpdate ops t b k get :: Connection -> BucketType -> Bucket -> Key -> IO (Maybe CRDT.DataType) get conn t b k = Resp.get <$> Conn.exchange conn (Req.get t b k) + + +-- | Ignore a ‘not_present’ error on update. +-- +-- This is a bit hacky, but that's the behaviour we want. +-- +-- TODO: Add custom exceptions to riak-haskell-client and just catch a +-- NotPresent exception here +handleEmpty :: IO () -> IO () +handleEmpty act = catchJust + (\(e :: ER.ErrorResponse) -> + if | "{precondition,{not_present," + `BS.isPrefixOf` ER.errmsg e -> Just () + | otherwise -> Nothing + ) + act + pure + diff --git a/src/Network/Riak/Cluster.hs b/src/Network/Riak/Cluster.hs --- a/src/Network/Riak/Cluster.hs +++ b/src/Network/Riak/Cluster.hs @@ -11,26 +11,25 @@ , Riak.defaultClient ) where +import Control.Concurrent.STM (atomically) +import Control.Concurrent.STM.TMVar import Control.Exception import Control.Exception.Enclosed +import Control.Monad.Base (liftBase) import Control.Monad.Catch (MonadThrow (..)) import Control.Monad.Trans.Control (MonadBaseControl) import Data.Typeable -import Data.Vector (Vector) -import qualified Data.Vector as V import Network.Riak (Connection) import qualified Network.Riak as Riak import qualified Network.Riak.Connection.Pool as Riak import System.Random.Mersenne.Pure64 -import System.Random.Shuffle (shuffle') -- | Datatype holding connection-pool with all known cluster nodes data Cluster = Cluster - { clusterPools :: Vector Riak.Pool + { clusterPools :: [Riak.Pool] -- ^ Vector of connection pools to riak cluster nodes - , clusterGen :: PureMT + , clusterGen :: TMVar PureMT } - deriving (Show) -- | Error that gets thrown whenever operation couldn't succeed with -- any node. @@ -48,22 +47,33 @@ -- 'Riak.Pool' objects connectToClusterWithPools :: [Riak.Pool] -> IO Cluster connectToClusterWithPools pools = do - mt <- newPureMT - return (Cluster (V.fromList pools) mt) + gen <- newPureMT + mt <- atomically (newTMVar gen) + return (Cluster pools mt) -- | Tries to run some operation for a random riak node. If it fails, -- tries all other nodes. If all other nodes fail - throws -- 'InClusterError' exception. inCluster :: (MonadThrow m, MonadBaseControl IO m) => Cluster -> (Connection -> m a) -> m a -inCluster rc f = do - let pools = shuffle' (V.toList (clusterPools rc)) - (V.length (clusterPools rc)) - (clusterGen rc) - go pools [] +inCluster Cluster{clusterPools=pools, clusterGen=tMT} f = do + rnd <- liftBase $ atomically $ do + mt <- takeTMVar tMT + let (i, mt') = randomInt mt + putTMVar tMT mt' + return i + let n = if null pools then 0 else rnd `mod` length pools + -- we rotate pool vector by n + pools' = rotateL n pools + go pools' [] where go [] errors = throwM (InClusterError errors) go (p:ps) es = Riak.withConnectionM p $ \c -> do er <- tryAny (f c) either (\err -> go ps (err:es)) return er + +rotateL :: Int -> [a] -> [a] +rotateL i xs = right ++ left + where + (left, right) = splitAt i xs 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 @@ -1,4 +1,4 @@ -{-# LANGUAGE CPP, OverloadedStrings, RecordWildCards, ScopedTypeVariables #-} +{-# LANGUAGE CPP, OverloadedStrings, RecordWildCards, ScopedTypeVariables, FlexibleContexts #-} {-# OPTIONS_GHC -fno-warn-orphans #-} -- | @@ -40,7 +40,7 @@ import Control.Concurrent (forkIO) import Control.Concurrent.Chan (newChan, readChan, writeChan) import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar) -import Control.Exception (Exception, IOException, throw) +import Control.Exception (Exception, IOException, throwIO) import Control.Monad (forM_, replicateM, replicateM_) import Data.Binary.Put (Put, putWord32be, runPut) import Data.IORef (newIORef, readIORef, writeIORef) @@ -200,15 +200,15 @@ instance Exception ErrorResponse -throwError :: ErrorResponse -> a -throwError = throw +throwError :: ErrorResponse -> IO a +throwError = throwIO -getResponse :: (Response a) => T.MessageTag -> Get a -getResponse expected = do - tag <- getTag +getResponse :: Response a => Connection -> Int64 -> a -> T.MessageTag -> IO a +getResponse conn len v expected = do + tag <- recvGet conn getTag case undefined of - _| tag == expected -> messageGetM - | tag == T.ErrorResponse -> throwError `fmap` messageGetM + _| tag == expected -> recvGetN conn (len-1) messageGetM + | tag == T.ErrorResponse -> throwError =<< recvGetN conn (len-1) messageGetM | otherwise -> moduleError "getResponse" $ "received unexpected response: expected " ++ show expected ++ ", received " ++ show tag @@ -253,7 +253,7 @@ go :: Response b => b -> IO b go dummy = do len <- fromIntegral `fmap` recvGet conn getWord32be - recvGetN conn len (getResponse (messageTag dummy)) + getResponse conn len dummy (messageTag dummy) recvResponse_ :: Connection -> T.MessageTag -> IO () recvResponse_ conn expected = debugRecv show $ do @@ -269,14 +269,14 @@ let tag = messageTag dummy if len == 1 then recvCorrectTag "recvMaybeResponse" conn tag 1 Nothing - else Just `fmap` recvGetN conn len (getResponse tag) + else Just `fmap` getResponse conn len dummy tag recvCorrectTag :: String -> Connection -> T.MessageTag -> Int64 -> a -> IO a recvCorrectTag func conn expected len v = do tag <- recvGet conn getTag case undefined of _| tag == expected -> recvExactly conn (len-1) >> return v - | tag == T.ErrorResponse -> throwError `fmap` recvGetN conn len messageGetM + | tag == T.ErrorResponse -> throwError =<< recvGetN conn len messageGetM | otherwise -> moduleError func $ "received unexpected response: expected " ++ show expected ++ ", received " ++ show tag diff --git a/tests/Test.hs b/tests/Test.hs --- a/tests/Test.hs +++ b/tests/Test.hs @@ -1,6 +1,7 @@ {-# LANGUAGE CPP #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE ParallelListComp #-} +{-# LANGUAGE ScopedTypeVariables #-} module Main where @@ -13,10 +14,13 @@ import Data.Foldable (toList) import Data.Semigroup import Data.Text (Text) +import Control.Applicative import Control.Concurrent (threadDelay) +import Control.Exception import qualified Network.Riak as Riak import qualified Network.Riak.Basic as B import qualified Network.Riak.Content as B +import qualified Network.Riak.Content as B (binary,Content) import qualified Network.Riak.CRDT as C import qualified Network.Riak.CRDT.Riak as C import qualified Network.Riak.Search as S @@ -24,8 +28,10 @@ import qualified Network.Riak.JSON as J import Network.Riak.Resolvable (ResolvableMonoid (..)) import Network.Riak.Types hiding (key) +import qualified Network.Riak.Protocol.ErrorResponse as ER import qualified Properties import qualified CRDTProperties as CRDT +import Common import Test.Tasty import Test.Tasty.HUnit @@ -37,6 +43,7 @@ tests = testGroup "Tests" [properties, integrationalTests, ping'o'death, + exceptions, crdts, searches, bucketTypes @@ -202,3 +209,32 @@ valok :: Maybe (Seq.Seq B.Content, VClock) -> B.Content -> Bool valok (Just (rs,_)) o = B.value o `elem` map B.value (toList rs) valok _ _ = False + + +exceptions :: TestTree +exceptions = testGroup "exceptions" [ + testCase "correct put" . shouldBeOK . withSomeConnection $ put, + testCase "correct put_" . shouldBeOK . withSomeConnection $ put_, + testCase "invalid put" . shouldThrow . withSomeConnection $ putErr, + testCase "invalid put_" . shouldThrow . withSomeConnection $ put_Err + ] + where + put = putSome B.put btype + put_ = putSome B.put_ btype + putErr = putSome B.put noBtype + put_Err = putSome B.put_ noBtype + + putSome :: (Connection -> Maybe BucketType -> Bucket -> Key + -> Maybe VClock -> B.Content -> Quorum -> Quorum -> IO a) + -> Maybe BucketType -> Connection -> IO a + putSome f bt c = f c bt buck key Nothing val Default Default + + shouldBeOK act = act >> assertBool "ok" True + shouldThrow act = catch (act >> assertBool "exception" False) (\(e::ER.ErrorResponse) -> pure ()) + + btype = Just "untyped-1" + noBtype = Just "no such type" + buck = "xxx" + key = "0" + val = B.binary "" +