riak 1.0.1.1 → 1.1.0.0
raw patch · 7 files changed
+124/−44 lines, 7 filesdep +stmdep +transformers-basedep −random-shuffledep ~protocol-buffersdep ~riak-protobufdep ~time
Dependencies added: stm, transformers-base
Dependencies removed: random-shuffle
Dependency ranges changed: protocol-buffers, riak-protobuf, time, transformers
Files
- Changes.md +3/−0
- riak.cabal +15/−13
- src/Network/Riak/CRDT.hs +2/−2
- src/Network/Riak/CRDT/Riak.hs +33/−4
- src/Network/Riak/Cluster.hs +23/−13
- src/Network/Riak/Connection/Internal.hs +12/−12
- tests/Test.hs +36/−0
Changes.md view
@@ -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
riak.cabal view
@@ -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,
src/Network/Riak/CRDT.hs view
@@ -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
src/Network/Riak/CRDT/Riak.hs view
@@ -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+
src/Network/Riak/Cluster.hs view
@@ -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
src/Network/Riak/Connection/Internal.hs view
@@ -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
tests/Test.hs view
@@ -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 ""+