riak 0.9.1.1 → 1.2.0.0
raw patch · 38 files changed
Files
- Changes.md +39/−0
- README.markdown +6/−6
- benchmarks/Main.hs +133/−0
- riak.cabal +109/−57
- src/Network/Riak.hs +2/−0
- src/Network/Riak/Basic.hs +54/−50
- src/Network/Riak/CRDT.hs +170/−0
- src/Network/Riak/CRDT/Ops.hs +107/−0
- src/Network/Riak/CRDT/Request.hs +50/−0
- src/Network/Riak/CRDT/Response.hs +63/−0
- src/Network/Riak/CRDT/Riak.hs +76/−0
- src/Network/Riak/CRDT/Types.hs +289/−0
- src/Network/Riak/Cluster.hs +25/−17
- src/Network/Riak/Connection.hs +1/−1
- src/Network/Riak/Connection/Internal.hs +93/−91
- src/Network/Riak/Connection/NoPush.hsc +6/−4
- src/Network/Riak/Content.hs +20/−30
- src/Network/Riak/Escape.hs +12/−17
- src/Network/Riak/Functions.hs +1/−7
- src/Network/Riak/JSON.hs +39/−34
- src/Network/Riak/JSON/Resolvable.hs +13/−13
- src/Network/Riak/Lens.hs +39/−0
- src/Network/Riak/Request.hs +196/−120
- src/Network/Riak/Resolvable/Internal.hs +41/−35
- src/Network/Riak/Response.hs +47/−30
- src/Network/Riak/Search.hs +64/−0
- src/Network/Riak/Tag.hs +224/−82
- src/Network/Riak/Types.hs +2/−0
- src/Network/Riak/Types/Internal.hs +72/−21
- src/Network/Riak/Value.hs +75/−69
- src/Network/Riak/Value/Resolvable.hs +12/−8
- tests/CRDTProperties.hs +316/−0
- tests/Internal.hs +32/−0
- tests/Properties.hs +39/−24
- tests/Test.hs +269/−29
- tests/Utils.hs +83/−0
- tests/dsl/Network/Riak/Admin/DSL.hs +56/−0
- tests/test.yaml +10/−0
+ Changes.md view
@@ -0,0 +1,39 @@+* 1.2.0.0+ - Bump `riak-protobuf` to `1.0.0`, which:+ - Switches the protocol buffer library from `protocol-buffers` to `proto-lens`+ - Retains the original protobuf message names that Basho defined, e.g.+ `Content` is now named `RpbContent`.+ - Uses strict `ByteString`s instead of lazy ones+ - Uses lists instead of `Seq`s+ - Bump `stm` upper bound.+ - Add `network >= 3.0` lower bound.+* 1.1.2.6+ - Regenerate protobuf for protocol-buffers 2.4.12. (https://github.com/riak-haskell-client/riak-haskell-client/issues/115)+ - Bump riak-protobuf version to 0.24.0.0.+ - Bump riak-protobuf-lens version to 0.24.0.0.+* 1.1.2.5+ - Bump aeson and exceptions upper bounds.+* 1.1.2.4+ - Bump async upper bounds.+* 1.1.2.2+ - Fix for GHC 8.2.2 support+* 1.1.2.1+ - PR #90. Add GHC 8.2.1 support+ - Bump time upper bound from <1.7 to <1.9+ - Bump aeson upper bound from <1.2 to <1.3+* 1.1.2.0+ - Fixes issue where exceptions were not handled properly with many threads (https://github.com/markhibberd/riak-haskell-client/pull/76)+ - Add / delete indexes+* 1.1.1.0+ - Fixes for 2 connection leaks on errors.+ - Bump upper bound on aeson to <1.1+* 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 and quickCheck tests+ - Basic yokozuna search functionality+ - CRDT benchmarking suite+ - Bugfixes+ - Tested on both 2.0.x and 2.1.x
README.markdown view
@@ -6,7 +6,7 @@ It uses Riak's [protobuf API](http://docs.basho.com/riak/latest/dev/references/protocol-buffers/) for optimal performance. -This project was originally the work of Bryan O'Sullivan (<bos@serpentine.com>), and then [Janrain, Inc.](http://janrain.com/), it is now being maintained by Mark Hibberd (<mark@hibberd.id.au>).+This project was originally the work of Bryan O'Sullivan (<bos@serpentine.com>), and then [Janrain, Inc.](http://janrain.com/), it is now being maintained by Mark Hibberd (<mark@hibberd.id.au>) and Tim McGilchrist (<timmcgil@gmail.com>). # Join in! @@ -14,15 +14,15 @@ and other improvements. Please report bugs via the-[github issue tracker](http://github.com/markhibberd/riak-haskell-client/issues).+[github issue tracker](http://github.com/riak-haskell-client/riak-haskell-client/issues). -Master [git repository](http://github.com/markhibberd/riak-haskell-client):+Master [git repository](http://github.com/riak-haskell-client/riak-haskell-client): -* `git clone git://github.com/markhibberd/riak-haskell-client.git`+* `git clone git://github.com/riak-haskell-client/riak-haskell-client.git` -Note the official repo is now <https://github.com/markhibberd/riak-haskell-client>.+Note the official repo is now <https://github.com/riak-haskell-client/riak-haskell-client>. # Build -[](https://travis-ci.org/markhibberd/riak-haskell-client)+[](https://travis-ci.org/riak-haskell-client/riak-haskell-client)
+ benchmarks/Main.hs view
@@ -0,0 +1,133 @@+{-# LANGUAGE OverloadedStrings #-}+module Main where++import Criterion.Main++import Network.Riak.Connection.Pool as Pool (Pool, create, withConnection)+import Network.Riak.Connection (defaultClient)+import Network.Riak.CRDT+import qualified Data.ByteString.Char8 as B+import Control.Applicative+import Control.Monad+import Data.Maybe+import Data.String+import qualified Data.List.NonEmpty as NEL+import Data.Semigroup+++main :: IO ()+main = benchmarks >>= defaultMain++bucket = "bench"++benchmarks = do+ pool <- Pool.create defaultClient 1 1 1+ already <- Pool.withConnection pool $ \c -> get c "counters" bucket "__setup"+ when (isNothing already) $ setup pool+ pure [crdt pool]++setup :: Pool -> IO ()+setup pool = Pool.withConnection pool $ \c -> do+ setupSetN c "10" 10+ setupSetN c "100" 100+ setupSetN c "1000" 1000++ setupMapN_elem c "1" 1+ setupMapN_elem c "10-elem" 10+ setupMapN_elem c "100-elem" 100+ setupMapN_elem c "1000-elem" 1000+ setupMapN_nest c "10-depth" 10+ setupMapN_nest c "100-depth" 100+ setupMapN_nest c "1000-depth" 1000++ sendModify c "counters" bucket "__setup" [CounterInc 1]+ putStrLn "Bench env set up."+++-- | * Common things++-- | Key names+setK n = fromString $ show n+mapElemK n = fromString $ show n <> "-elem"+mapDepthK n = fromString $ show n <> "-depth"++-- | Map things+nestedMapPath n = MapPath . NEL.fromList $ [ fromString (show i) | i <- [1..n] ]+++setupSetN c key n = sequence_ [+ sendModify c "sets" bucket key [SetAdd (B.pack $ show i)]+ | i <- [1..n]+ ]++-- | n-elem map with counters+setupMapN_elem c key n = sendModify c "maps" bucket key ops+ where ops = [MapUpdate (MapPath $ (fromString $ show i) :| [])+ (MapCounterOp $ CounterInc 1)+ | i <- [1..n]+ ]++-- | n-depth map with a counter+setupMapN_nest c key n = sendModify c "maps" bucket key [op]+ where op = MapUpdate (nestedMapPath n)+ (MapCounterOp $ CounterInc 1)++++crdt :: Pool -> Benchmark+crdt pool = bgroup "CRDT" [+ bgroup "get" [+ bench "get a non-existent counter" . nfIO $ pooled getEmpty+ ],++ bgroup "counters" [+ bench "get a counter" . nfIO $ pooled getCounter+ ],++ bgroup "sets" [+ bench "get a 10-elem set" . nfIO $ pooled getSet10,+ bench "get a 100-elem set" . nfIO $ pooled getSet100,+ bench "get a 1000-elem set" . nfIO $ pooled getSet1000,+ bench "1000-set add+remove an elem cycle" . nfIO $ pooled set1kAddRemove+ ],++ bgroup "maps" [+ bgroup "get" [+ bench "get a trivial map" . nfIO . pooled $ get1Map,+ bench "get a 10-elem map" . nfIO . pooled $ getNMap 10,+ bench "get a 100-elem map" . nfIO . pooled $ getNMap 100,+ bench "get a 1000-elem map" . nfIO . pooled $ getNMap 1000,+ bench "get a 10-depth map" . nfIO . pooled $ getDMap 10,+ bench "get a 100-depth map" . nfIO . pooled $ getDMap 100,+ bench "get a 1000-depth map" . nfIO . pooled $ getDMap 1000+ ],++ bgroup "update" [+ bench "update a counter inside 10-elem map" . nfIO . pooled $ updateNMap 10,+ bench "update a counter inside 100-elem map" . nfIO . pooled $ updateNMap 100,+ bench "update a counter inside 1000-elem map" . nfIO . pooled $ updateNMap 1000,+ bench "update a counter inside 10-depth map" . nfIO . pooled $ updateDMap 10,+ bench "update a counter inside 100-depth map" . nfIO . pooled $ updateDMap 100,+ bench "update a counter inside 1000-depth map" . nfIO . pooled $ updateDMap 1000+ ]+ ]+ ]+ where pooled = Pool.withConnection pool+++getEmpty c = get c "counters" "not here" "never was"+getCounter c = get c "counters" "xxx" "yyy"+getSet10 c = get c "sets" bucket "10"+getSet100 c = get c "sets" bucket "100"+getSet1000 c = get c "sets" bucket "1000"++set1kAddRemove c = sequence_ [ sendModify c "sets" bucket "1000" [o]+ | o <- [SetAdd "foo", SetRemove "foo"] ]++get1Map c = get c "maps" bucket "1"+getNMap n c = get c "maps" bucket (mapElemK n)+getDMap n c = get c "maps" bucket (mapDepthK n)++updateNMap n c = setupMapN_elem c (mapElemK n) n+updateDMap n c = setupMapN_nest c (mapDepthK n) n+
riak.cabal view
@@ -1,5 +1,5 @@ name: riak-version: 0.9.1.1+version: 1.2.0.0 synopsis: A Haskell client for the Riak decentralized data store description: A Haskell client library for the Riak decentralized data@@ -26,9 +26,13 @@ [Network.Riak.Basic] manual storage, manual conflict resolution. This is the most demanding module to work with, as you must encode and decode data yourself, and handle all conflict resolution- yourself.+ yourself+ .+ [Network.Riak.CRDT] CRDT operations. -homepage: http://github.com/markhibberd/riak-haskell-client++homepage: http://github.com/riak-haskell-client/riak-haskell-client+Bug-reports: http://github.com/riak-haskell-client/riak-haskell-client/issues license: OtherLicense license-file: LICENSE author: Bryan O'Sullivan <bos@serpentine.com>@@ -38,13 +42,16 @@ category: Network build-type: Simple extra-source-files:- README.markdown+ README.markdown+ Changes.md+ tests/test.yaml -cabal-version: >=1.8+cabal-version: >=1.10+tested-with: GHC==8.4.4, GHC==8.2.2, GHC == 8.6.3, GHC == 8.8.4, GHC==8.10.4 source-repository head type: git- location: https://github.com:markhibberd/riak-haskell-client.git+ location: https://github.com:riak-haskell-client/riak-haskell-client.git flag debug description: allow debug logging@@ -60,88 +67,133 @@ default: False library+ default-language: Haskell2010 hs-source-dirs: src- exposed-modules:- Network.Riak- Network.Riak.Basic- Network.Riak.Cluster- Network.Riak.Connection- Network.Riak.Connection.Pool- Network.Riak.Content- Network.Riak.Debug- Network.Riak.Escape- Network.Riak.Functions- Network.Riak.JSON- Network.Riak.JSON.Resolvable- Network.Riak.Request- Network.Riak.Resolvable- Network.Riak.Response- Network.Riak.Types- Network.Riak.Value- Network.Riak.Value.Resolvable+ Network.Riak+ Network.Riak.Basic+ Network.Riak.Cluster+ Network.Riak.Connection+ Network.Riak.Connection.Internal+ Network.Riak.Connection.Pool+ Network.Riak.Content+ Network.Riak.CRDT+ Network.Riak.CRDT.Ops+ Network.Riak.CRDT.Request+ Network.Riak.CRDT.Response+ Network.Riak.CRDT.Riak+ Network.Riak.CRDT.Types+ Network.Riak.Debug+ Network.Riak.Escape+ Network.Riak.Functions+ Network.Riak.JSON+ Network.Riak.JSON.Resolvable+ Network.Riak.Request+ Network.Riak.Resolvable+ Network.Riak.Response+ Network.Riak.Search+ Network.Riak.Types+ Network.Riak.Value+ Network.Riak.Value.Resolvable+ Network.Riak.Types.Internal other-modules:- Network.Riak.Connection.Internal- Network.Riak.Connection.NoPush- Network.Riak.Resolvable.Internal- Network.Riak.Tag- Network.Riak.Types.Internal+ Network.Riak.Connection.NoPush+ Network.Riak.Lens+ Network.Riak.Resolvable.Internal+ Network.Riak.Tag build-depends:- aeson >= 0.8 && < 0.11,+ aeson >= 0.8 && < 1.5.7,+ async >= 2.0.0.0 && < 2.3, attoparsec >= 0.12.1.6 && < 0.14,- base >= 3 && <5,+ base >= 3 && < 5,+ bifunctors >= 4.2 && < 6, binary,- blaze-builder >= 0.3 && <= 0.5,+ blaze-builder >= 0.3 && <= 0.5, bytestring, containers,- enclosed-exceptions >= 1.0.1.1 && <= 1.1,- exceptions >= 0.8.0.2 && <= 0.9,- transformers >= 0.3 && < 0.5,- mersenne-random-pure64 >= 0.2.0.4 && < 0.3,- monad-control >= 1.0.0.4 && < 1.1,- network >= 2.3,+ data-default-class >= 0.0.1,+ deepseq >= 1.3,+ enclosed-exceptions >= 1.0.1.1 && <= 1.1,+ exceptions >= 0.8.0.2 && < 0.11,+ hashable >= 1.2.3,+ 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 >= 3.0 && < 3.1, resource-pool == 0.2.*,- protocol-buffers >= 2.1.4 && < 2.2,+ proto-lens == 0.7.*, pureMD5, random,- random-shuffle >= 0.0.4 && < 0.1,- riak-protobuf == 0.20.*,+ riak-protobuf == 0.25.*,+ semigroups >= 0.16,+ stm >= 2.4 && < 2.6, text == 1.2.*,- time >= 1.4.2 && < 1.6,- vector >= 0.10.12.3 && < 0.12+ time >= 1.4.2 && < 1.10,+ vector >= 0.10.12.3 && < 0.13,+ unordered-containers >= 0.2.5 + default-extensions:+ DoAndIfThenElse+ if flag(debug) cpp-options: -DASSERTS -DDEBUG if flag(developer)- ghc-options: -Werror+ ghc-options: -Wall ghc-prof-options: -auto-all cpp-options: -DASSERTS -DDEBUG ghc-options: -Wall test-suite test+ default-language: Haskell2010 type: exitcode-stdio-1.0 main-is: Test.hs- hs-source-dirs: tests- ghc-options: -Wall+ hs-source-dirs: tests, tests/dsl+ ghc-options: -Wall -threaded if flag(test2i) cpp-options: -DTEST2I other-modules:- Properties+ CRDTProperties+ Internal+ Utils+ Network.Riak.Admin.DSL+ Properties build-depends:- base,- riak,- bytestring,- containers,- HUnit,- QuickCheck,- tasty,- tasty-hunit,- tasty-quickcheck,- text+ base,+ riak,+ riak-protobuf,+ aeson,+ bytestring,+ containers,+ HUnit,+ microlens,+ process,+ QuickCheck,+ tasty,+ tasty-hunit,+ tasty-quickcheck,+ template-haskell,+ text,+ mtl >= 2.1,+ semigroups >= 0.16,+ data-default-class >= 0.0.1,+ yaml >= 0.8.19++benchmark bench+ default-language: Haskell2010+ type: exitcode-stdio-1.0+ hs-source-dirs: benchmarks+ main-is: Main.hs+ build-depends:+ base,+ riak,+ criterion >= 1.1,+ bytestring >= 0.10,+ semigroups >= 0.16
src/Network/Riak.hs view
@@ -35,6 +35,8 @@ -- and decode data yourself, and handle all conflict resolution -- yourself. --+-- [Network.Riak.CRDT] CRDT operations.+-- -- A short getting started guide is available at <http://docs.basho.com/riak/latest/dev/taste-of-riak/haskell/> -- module Network.Riak
src/Network/Riak/Basic.hs view
@@ -1,4 +1,7 @@-{-# LANGUAGE BangPatterns, OverloadedStrings, RecordWildCards, CPP #-}+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE CPP #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-} -- | -- Module: Network.Riak.Basic@@ -39,25 +42,21 @@ , foldKeys , getBucket , setBucket+ , getBucketType -- * Map/reduce , mapReduce ) where #if __GLASGOW_HASKELL__ < 710-import Control.Applicative ((<$>))+import Control.Applicative ((<$>)) #endif-import Control.Monad.IO.Class-import Data.Maybe (fromMaybe)-import Network.Riak.Connection.Internal-import Network.Riak.Escape (unescape)-import Network.Riak.Protocol.BucketProps-import Network.Riak.Protocol.Content-import Network.Riak.Protocol.ListKeysResponse-import Network.Riak.Protocol.MapReduce as MapReduce-import Network.Riak.Protocol.ServerInfo-import Network.Riak.Types.Internal hiding (MessageTag(..))+import Control.Monad.IO.Class+import Network.Riak.Connection.Internal+import Network.Riak.Escape (unescape)+import Network.Riak.Lens+import Network.Riak.Types.Internal hiding (MessageTag(..)) import qualified Data.Foldable as F-import qualified Data.Sequence as Seq+import qualified Data.Riak.Proto as Proto import qualified Network.Riak.Request as Req import qualified Network.Riak.Response as Resp import qualified Network.Riak.Types.Internal as T@@ -71,82 +70,87 @@ getClientID conn = Resp.getClientID <$> exchange conn Req.getClientID -- | Retrieve information about the server.-getServerInfo :: Connection -> IO ServerInfo+getServerInfo :: Connection -> IO Proto.RpbGetServerInfoResp getServerInfo conn = exchange conn Req.getServerInfo -- | Retrieve a value. This may return multiple conflicting siblings. -- Choosing among them is your responsibility.-get :: Connection -> T.Bucket -> T.Key -> R- -> IO (Maybe (Seq.Seq Content, VClock))-get conn bucket key r = Resp.get <$> exchangeMaybe conn (Req.get bucket key r)+get :: Connection -> Maybe T.BucketType -> T.Bucket -> T.Key -> R+ -> IO (Maybe ([Proto.RpbContent], VClock))+get conn btype bucket key r = Resp.get <$> exchangeMaybe conn (Req.get btype bucket key r) -- | Store a single value. This may return multiple conflicting -- siblings. Choosing among them, and storing a new value, is your -- responsibility. -- -- You should /only/ supply 'Nothing' as a 'T.VClock' if you are sure--- that the given bucket+key combination does not already exist. If--- you omit a 'T.VClock' but the bucket+key /does/ exist, your value--- will not be stored.-put :: Connection -> T.Bucket -> T.Key -> Maybe T.VClock- -> Content -> W -> DW- -> IO (Seq.Seq Content, VClock)-put conn bucket key mvclock cont w dw =- Resp.put <$> exchange conn (Req.put bucket key mvclock cont w dw True)+-- that the given type+bucket+key combination does not already exist.+-- If you omit a 'T.VClock' but the type+bucket+key /does/ exist, your+-- value will not be stored.+put :: Connection -> Maybe T.BucketType -> T.Bucket -> T.Key -> Maybe T.VClock+ -> Proto.RpbContent -> W -> DW+ -> IO ([Proto.RpbContent], VClock)+put conn btype bucket key mvclock cont w dw =+ Resp.put <$> exchange conn (Req.put btype bucket key mvclock cont w dw True) -- | Store a single value, without the possibility of conflict -- resolution. -- -- You should /only/ supply 'Nothing' as a 'T.VClock' if you are sure--- that the given bucket+key combination does not already exist. If--- you omit a 'T.VClock' but the bucket+key /does/ exist, your value--- will not be stored, and you will not be notified.-put_ :: Connection -> T.Bucket -> T.Key -> Maybe T.VClock- -> Content -> W -> DW+-- that the given type+bucket+key combination does not already exist.+-- If you omit a 'T.VClock' but the type+bucket+key /does/ exist, your+-- value will not be stored, and you will not be notified.+put_ :: Connection -> Maybe T.BucketType -> T.Bucket -> T.Key -> Maybe T.VClock+ -> Proto.RpbContent -> W -> DW -> IO ()-put_ conn bucket key mvclock cont w dw =- exchange_ conn (Req.put bucket key mvclock cont w dw False)+put_ conn btype bucket key mvclock cont w dw =+ exchange_ conn (Req.put btype bucket key mvclock cont w dw False) -- | Delete a value.-delete :: Connection -> T.Bucket -> T.Key -> RW -> IO ()-delete conn bucket key rw = exchange_ conn $ Req.delete bucket key rw+delete :: Connection -> Maybe T.BucketType -> T.Bucket -> T.Key -> RW -> IO ()+delete conn btype bucket key rw = exchange_ conn $ Req.delete btype bucket key rw --- List the buckets in the cluster.+-- | List the buckets in the cluster. -- -- /Note/: this operation is expensive. Do not use it in production.-listBuckets :: Connection -> IO (Seq.Seq T.Bucket)-listBuckets conn = Resp.listBuckets <$> exchange conn Req.listBuckets+listBuckets :: Connection -> Maybe BucketType -> IO [T.Bucket]+listBuckets conn btype = Resp.listBuckets <$> exchange conn (Req.listBuckets btype) --- Fold over the keys in a bucket.+-- | Fold over the keys in a bucket. -- -- /Note/: this operation is expensive. Do not use it in production.-foldKeys :: (MonadIO m) => Connection -> T.Bucket -> (a -> Key -> m a) -> a -> m a-foldKeys conn bucket f z0 = do- liftIO $ sendRequest conn $ Req.listKeys bucket+foldKeys :: (MonadIO m) => Connection -> Maybe BucketType -> Bucket+ -> (a -> Key -> m a) -> a -> m a+foldKeys conn btype bucket f z0 = do+ liftIO $ sendRequest conn $ Req.listKeys btype bucket let g z = f z . unescape loop z = do- ListKeysResponse{..} <- liftIO $ recvResponse conn- z1 <- F.foldlM g z keys- if fromMaybe False done+ response <- liftIO $ (recvResponse conn :: IO Proto.RpbListKeysResp)+ z1 <- F.foldlM g z (response ^. Proto.keys)+ if response ^. Proto.done then return z1 else loop z1 loop z0 -- | Retrieve the properties of a bucket.-getBucket :: Connection -> T.Bucket -> IO BucketProps-getBucket conn bucket = Resp.getBucket <$> exchange conn (Req.getBucket bucket)+getBucket :: Connection -> Maybe BucketType -> Bucket -> IO Proto.RpbBucketProps+getBucket conn btype bucket = Resp.getBucket <$> exchange conn (Req.getBucket btype bucket) -- | Store new properties for a bucket.-setBucket :: Connection -> T.Bucket -> BucketProps -> IO ()-setBucket conn bucket props = exchange_ conn $ Req.setBucket bucket props+setBucket :: Connection -> Maybe BucketType -> Bucket -> Proto.RpbBucketProps -> IO ()+setBucket conn btype bucket props = exchange_ conn $ Req.setBucket btype bucket props +-- | Gets the bucket properties associated with a bucket type.+getBucketType :: Connection -> T.BucketType -> IO Proto.RpbBucketProps+getBucketType conn btype = Resp.getBucket <$> exchange conn (Req.getBucketType btype)+ -- | Run a 'MapReduce' job. Its result is consumed via a strict left -- fold.-mapReduce :: Connection -> Job -> (a -> MapReduce -> a) -> a -> IO a+mapReduce :: Connection -> Job -> (a -> Proto.RpbMapRedResp -> a) -> a -> IO a mapReduce conn job f z0 = loop z0 =<< (exchange conn . Req.mapReduce $ job) where loop z mr = do let !z' = f z mr- if fromMaybe False . MapReduce.done $ mr+ if mr ^. Proto.done then return z' else loop z' =<< recvResponse conn
+ src/Network/Riak/CRDT.hs view
@@ -0,0 +1,170 @@+{-# LANGUAGE CPP #-}+{-# LANGUAGE FunctionalDependencies #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE PatternGuards #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeFamilies #-}++{- | module: Network.Riak.CRDT+ copyright: (c) 2016 Sentenai+ author: Antonio Nikishaev <me@lelf.lu>+ license: Apache++CRDT operations++* Haskell-side++ * Haskell values: 'Counter', 'Set' etc++ * ADT for operations: 'CounterOp', 'SetOp' etc++ * 'modify' to locally modify a value (matching riak-side behaviour)++* Riak-side++ * 'get' to get a current value++ * 'sendModify' to ask Riak to apply modifications++TL;DR example++>>> let c = Counter 41+>>> let op = CounterInc 1+>>> modify op c+Counter 42+>>> get conn "counters" "bucket" "key"+Just (DTCounter (Counter 41))+>>> sendModify conn "counters" "bucket" "key" [op] >> get conn "counters" "bucket" "key"+Just (DTCounter (Counter 42))++-}++module Network.Riak.CRDT (+ module X+ , get+ , CRDT(..)+ ) where++import Data.Default.Class+import qualified Data.Map as M+import Data.Proxy+#if __GLASGOW_HASKELL__ < 804+import Data.Semigroup+#endif+import qualified Data.Set as S+import Network.Riak.CRDT.Ops+import Network.Riak.CRDT.Riak+import Network.Riak.CRDT.Types as X+import Network.Riak.Types++-- | Modify a counter by applying operations ops+modifyCounter :: CounterOp -> Counter -> Counter+modifyCounter op c = c <> Counter i+ where CounterInc i = op++-- | Modify a set by applying operations ops+modifySet :: SetOp -> Set -> Set+modifySet op (Set c) = Set (c `S.union` adds S.\\ rems)+ where SetOpsComb adds rems = toOpsComb op++modifyMap :: MapOp -> Map -> Map+modifyMap (MapRemove field) (Map mc) = Map $ M.delete field mc+modifyMap (MapUpdate path op) m = modifyMap1 path op m++modifyMap1 :: MapPath -> MapValueOp -> Map -> Map+modifyMap1 (MapPath (e :| [])) op m = modMap mf op m+ where mf = MapField (mapEntryTag op) e+modifyMap1 (MapPath (e :| (r:rs))) op (Map m')+ = Map $ M.alter (Just . f) (MapField MapMapTag e) m'+ where f :: Maybe MapEntry -> MapEntry+ f Nothing = f (Just $ MapMap def)+ f (Just (MapMap m)) = MapMap . modifyMap1 (MapPath (r :| rs)) op $ m+ f (Just z) = z++modMap :: MapField -> MapValueOp -> Map -> Map+modMap ix op (Map m) = Map $ M.alter (Just . modifyMapValue op) ix m++modifyMapValue :: MapValueOp -> Maybe MapEntry -> MapEntry+modifyMapValue (MapSetOp op) = modifyEntry (Proxy :: Proxy Set) op+modifyMapValue (MapCounterOp op) = modifyEntry (Proxy :: Proxy Counter) op+modifyMapValue (MapMapOp op) = modifyEntry (Proxy :: Proxy Map) op+modifyMapValue (MapFlagOp op) = modifyEntry (Proxy :: Proxy Flag) op+modifyMapValue (MapRegisterOp op) = modifyEntry (Proxy :: Proxy Register) op++modifyFlag :: FlagOp -> Flag -> Flag+modifyFlag (FlagSet x) = const (Flag x)++modifyRegister :: RegisterOp -> Register -> Register+modifyRegister (RegisterSet x) = const (Register x)++-- | Types that can be held inside 'Map'+class Default a => MapCRDT a where+ type MapOperation_ a :: *+ mapModify :: MapOperation_ a -> a -> a++ -- | modify a maybe-absent 'MapEntry'+ modifyEntry :: Proxy a -> MapOperation_ a -> Maybe MapEntry -> MapEntry+ modifyEntry _ op Nothing = toEntry . mapModify op $ (def :: a)+ modifyEntry _ op (Just e) | Just v <- fromEntry e = toEntry . mapModify op $ (v :: a)+ | otherwise = e+ toEntry :: a -> MapEntry+ fromEntry :: MapEntry -> Maybe a+++instance MapCRDT Flag where+ type MapOperation_ Flag = FlagOp+ mapModify = modifyFlag+ fromEntry (MapFlag f) = Just f+ fromEntry _ = Nothing+ toEntry = MapFlag++instance MapCRDT Set where+ type MapOperation_ Set = SetOp+ mapModify = modify+ fromEntry (MapSet s) = Just s+ fromEntry _ = Nothing+ toEntry = MapSet++instance MapCRDT Counter where+ type MapOperation_ Counter = CounterOp+ mapModify = modify+ fromEntry (MapCounter s) = Just s+ fromEntry _ = Nothing+ toEntry = MapCounter++instance MapCRDT Register where+ type MapOperation_ Register = RegisterOp+ mapModify = modifyRegister+ fromEntry (MapRegister s) = Just s+ fromEntry _ = Nothing+ toEntry = MapRegister++instance MapCRDT Map where+ type MapOperation_ Map = MapOp+ mapModify = modify+ fromEntry (MapMap s) = Just s+ fromEntry _ = Nothing+ toEntry = MapMap++-- | CRDT types+class MapCRDT a => CRDT a op | a -> op, op -> a where+ -- | Modify a value by applying an operation+ modify :: op -> a -> a++ -- | Request riak a modification+ sendModify :: Connection+ -> BucketType -> Bucket -> Key+ -> [op] -> IO ()++instance CRDT Counter CounterOp where+ modify = modifyCounter+ sendModify = counterSendUpdate++instance CRDT Set SetOp where+ modify = modifySet+ sendModify = setSendUpdate++instance CRDT Map MapOp where+ modify = modifyMap+ sendModify = mapSendUpdate
+ src/Network/Riak/CRDT/Ops.hs view
@@ -0,0 +1,107 @@+-- |+-- Module: Network.Riak.CRDT.Ops+-- Copyright: (c) 2016 Sentenai+-- Author: Antonio Nikishaev <me@lelf.lu>+-- License: Apache+-- Maintainer: Tim McGilchrist <timmcgil@gmail.com>, Mark Hibberd <mark@hibberd.id.au>+-- Stability: experimental+-- Portability: portable+--+--+-- Conversions of CRDT operations to 'PB.DtOp'+--+module Network.Riak.CRDT.Ops (+ counterUpdateOp+ , setUpdateOp+ , SetOpsComb(..)+ , toOpsComb+ , mapUpdateOp+ ) where++import Data.ByteString (ByteString)+import Data.Semigroup (Semigroup((<>)))+import qualified Data.Set as S++import qualified Data.Riak.Proto as Proto+import Network.Riak.CRDT.Types+import Network.Riak.Lens++counterUpdateOp :: [CounterOp] -> Proto.DtOp+counterUpdateOp ops = Proto.defMessage & Proto.counterOp .~ counterOpPB ops++counterOpPB :: [CounterOp] -> Proto.CounterOp+counterOpPB ops = Proto.defMessage & Proto.increment .~ i+ where CounterInc i = mconcat ops+++data SetOpsComb = SetOpsComb { setAdds :: S.Set ByteString,+ setRemoves :: S.Set ByteString }+ deriving (Show)++instance Semigroup SetOpsComb where+ (SetOpsComb a b) <> (SetOpsComb x y) = SetOpsComb (a<>x) (b<>y)++instance Monoid SetOpsComb where+ mempty = SetOpsComb mempty mempty+ (SetOpsComb a b) `mappend` (SetOpsComb x y) = SetOpsComb (a<>x) (b<>y)++toOpsComb :: SetOp -> SetOpsComb+toOpsComb (SetAdd s) = SetOpsComb (S.singleton s) S.empty+toOpsComb (SetRemove s) = SetOpsComb S.empty (S.singleton s)++++setUpdateOp :: [SetOp] -> Proto.DtOp+setUpdateOp ops = Proto.defMessage & Proto.setOp .~ setOpPB ops++setOpPB :: [SetOp] -> Proto.SetOp+setOpPB ops = Proto.defMessage & Proto.adds .~ S.toList adds+ & Proto.removes .~ S.toList rems+ where SetOpsComb adds rems = mconcat . map toOpsComb $ ops++flagOpPB :: FlagOp -> Proto.MapUpdate'FlagOp+flagOpPB (FlagSet True) = Proto.MapUpdate'ENABLE+flagOpPB (FlagSet False) = Proto.MapUpdate'DISABLE++registerOpPB :: RegisterOp -> ByteString+registerOpPB (RegisterSet x) = x++mapUpdateOp :: [MapOp] -> Proto.DtOp+mapUpdateOp ops = Proto.defMessage & Proto.mapOp .~ mapOpPB ops++mapOpPB :: [MapOp] -> Proto.MapOp+mapOpPB ops = Proto.defMessage & Proto.removes .~ rems+ & Proto.updates .~ updates+ where rems = [ toRemove f | MapRemove f <- ops ]+ updates = [ toUpdate f u | MapUpdate f u <- ops ]++toRemove :: MapField -> Proto.MapField+toRemove (MapField t name) = toField name t++toUpdate :: MapPath -> MapValueOp -> Proto.MapUpdate+toUpdate (MapPath (e :| [])) op = toUpdate' e (mapEntryTag op) op+toUpdate (MapPath (e :| (r:rs))) op = toUpdate' e MapMapTag op'+ where op' = MapMapOp (MapUpdate (MapPath (r:|rs)) op)++toUpdate' :: ByteString -> MapEntryTag -> MapValueOp -> Proto.MapUpdate+toUpdate' f t op = setSpecificOp op (updateNothing f t)++setSpecificOp :: MapValueOp -> Proto.MapUpdate -> Proto.MapUpdate+setSpecificOp (MapCounterOp cop) = Proto.counterOp .~ counterOpPB [cop]+setSpecificOp (MapSetOp sop) = Proto.setOp .~ setOpPB [sop]+setSpecificOp (MapRegisterOp rop) = Proto.registerOp .~ registerOpPB rop+setSpecificOp (MapFlagOp fop) = Proto.flagOp .~ flagOpPB fop+setSpecificOp (MapMapOp mop) = Proto.mapOp .~ mapOpPB [mop]+++updateNothing :: ByteString -> MapEntryTag -> Proto.MapUpdate+updateNothing f t = Proto.defMessage & Proto.field .~ toField f t++toField :: ByteString -> MapEntryTag -> Proto.MapField+toField name t = Proto.defMessage & Proto.name .~ name+ & Proto.type' .~ typ t+ where typ MapCounterTag = Proto.MapField'COUNTER+ typ MapSetTag = Proto.MapField'SET+ typ MapRegisterTag = Proto.MapField'REGISTER+ typ MapFlagTag = Proto.MapField'FLAG+ typ MapMapTag = Proto.MapField'MAP
+ src/Network/Riak/CRDT/Request.hs view
@@ -0,0 +1,50 @@+-- |+-- Module: Network.Riak.CRDT.Request+-- Copyright: (c) 2016 Sentenai+-- Author: Antonio Nikishaev <me@lelf.lu>+-- License: Apache+-- Maintainer: Tim McGilchrist <timmcgil@gmail.com>, Mark Hibberd <mark@hibberd.id.au>+-- Stability: experimental+-- Portability: portable+--+module Network.Riak.CRDT.Request (+ get+ , counterUpdate+ , setUpdate+ , mapUpdate+ ) where++import Data.ByteString (ByteString)+import qualified Data.Riak.Proto as Proto+import Network.Riak.CRDT.Ops+import qualified Network.Riak.CRDT.Types as CRDT+import Network.Riak.Lens+import Network.Riak.Types++counterUpdate :: [CRDT.CounterOp]+ -> BucketType -> Bucket -> Key+ -> Proto.DtUpdateReq+counterUpdate ops = update (counterUpdateOp ops)++setUpdate :: [CRDT.SetOp]+ -> BucketType -> Bucket -> Key+ -> Proto.DtUpdateReq+setUpdate ops = update (setUpdateOp ops)++mapUpdate :: [CRDT.MapOp]+ -> BucketType -> Bucket -> Key+ -> Proto.DtUpdateReq+mapUpdate ops = update (mapUpdateOp ops)++update :: Proto.DtOp -> BucketType -> Bucket -> Key -> Proto.DtUpdateReq+update op t b k = Proto.defMessage+ & Proto.bucket .~ b+ & Proto.key .~ k+ & Proto.op .~ op+ & Proto.type' .~ t++get :: ByteString -> ByteString -> ByteString -> Proto.DtFetchReq+get t b k = Proto.defMessage+ & Proto.bucket .~ b+ & Proto.key .~ k+ & Proto.type' .~ t
+ src/Network/Riak/CRDT/Response.hs view
@@ -0,0 +1,63 @@+-- |+-- Module: Network.Riak.CRDT.Response+-- Copyright: (c) 2016 Sentenai+-- Author: Antonio Nikishaev <me@lelf.lu>+-- License: Apache+-- Maintainer: Tim McGilchrist <timmcgil@gmail.com>, Mark Hibberd <mark@hibberd.id.au>+-- Stability: experimental+-- Portability: portable+--++{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE CPP #-}+module Network.Riak.CRDT.Response (get) where++#if __GLASGOW_HASKELL__ <= 708+import Control.Applicative ((<$>))+import Data.Traversable+#endif+import qualified Data.Map as Map+import Data.Maybe (catMaybes)+import qualified Data.Riak.Proto as Proto++import Network.Riak.CRDT.Types as CRDT+import Network.Riak.Lens++get :: Proto.DtFetchResp -> Maybe CRDT.DataType+get resp = case resp ^. Proto.type' of+ Proto.DtFetchResp'COUNTER ->+ DTCounter . Counter <$> ((^. Proto.maybe'counterValue) =<< (resp ^. Proto.maybe'value))+ Proto.DtFetchResp'SET ->+ DTSet . setFromList . (^. Proto.setValue) <$> (resp ^. Proto.maybe'value)+ Proto.DtFetchResp'MAP ->+ DTMap . deconstructMap . (^. Proto.mapValue) <$> (resp ^. Proto.maybe'value)++ -- We don't support hll or gset yet+ Proto.DtFetchResp'HLL -> Nothing+ Proto.DtFetchResp'GSET -> Nothing++deconstructMap :: [Proto.MapEntry] -> Map+deconstructMap = Map . Map.fromList . catMaybes . map f++f :: Proto.MapEntry -> Maybe (MapField, MapEntry)+f entry = sequenceA (MapField t (field ^. Proto.name), v)+ where field :: Proto.MapField+ field = entry ^. Proto.field+ type' :: Proto.MapField'MapFieldType+ type' = field ^. Proto.type'+ t :: MapEntryTag+ t = typeToTag type'+ v :: Maybe MapEntry+ v = case type' of+ Proto.MapField'COUNTER -> MapCounter . Counter <$> (entry ^. Proto.maybe'counterValue)+ Proto.MapField'SET -> Just . MapSet . setFromList $ (entry ^. Proto.setValue)+ Proto.MapField'REGISTER -> MapRegister . Register <$> (entry ^. Proto.maybe'registerValue)+ Proto.MapField'FLAG -> MapFlag . Flag <$> (entry ^. Proto.maybe'flagValue)+ Proto.MapField'MAP -> Just . MapMap . deconstructMap $ (entry ^. Proto.mapValue)++typeToTag :: Proto.MapField'MapFieldType -> MapEntryTag+typeToTag Proto.MapField'COUNTER = MapCounterTag+typeToTag Proto.MapField'SET = MapSetTag+typeToTag Proto.MapField'REGISTER = MapRegisterTag+typeToTag Proto.MapField'FLAG = MapFlagTag+typeToTag Proto.MapField'MAP = MapMapTag
+ src/Network/Riak/CRDT/Riak.hs view
@@ -0,0 +1,76 @@+-- |+-- Module: Network.Riak.CRDT.Riak+-- Copyright: (c) 2016 Sentenai+-- Author: Antonio Nikishaev <me@lelf.lu>+-- License: Apache+-- Maintainer: Tim McGilchrist <timmcgil@gmail.com>, Mark Hibberd <mark@hibberd.id.au>+-- Stability: experimental+-- Portability: portable+--+-- CRDT operations+--+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE MultiWayIf #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE CPP #-}++module Network.Riak.CRDT.Riak (+ counterSendUpdate+ , setSendUpdate+ , mapSendUpdate+ , get+ ) where++#if __GLASGOW_HASKELL__ <= 708+import Control.Applicative+import Data.Int+#endif++import Control.Exception (catchJust)++import qualified Data.ByteString as BS++import qualified Data.Riak.Proto as Proto+import qualified Network.Riak.CRDT.Request as Req+import qualified Network.Riak.CRDT.Response as Resp+import qualified Network.Riak.CRDT.Types as CRDT+import qualified Network.Riak.Connection as Conn+import Network.Riak.Lens+import Network.Riak.Types+++counterSendUpdate :: Connection -> BucketType -> Bucket -> Key+ -> [CRDT.CounterOp] -> IO ()+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 = handleEmpty . Conn.exchange_ conn $ Req.setUpdate ops t b k+++mapSendUpdate :: Connection -> BucketType -> Bucket -> Key+ -> [CRDT.MapOp] -> IO ()+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 :: Proto.RpbErrorResp) ->+ if | "{precondition,{not_present,"+ `BS.isPrefixOf` (e ^. Proto.errmsg) -> Just ()+ | otherwise -> Nothing+ )+ act+ pure
+ src/Network/Riak/CRDT/Types.hs view
@@ -0,0 +1,289 @@+-- |+-- Module: Network.Riak.CRDT.Types+-- Copyright: (c) 2016 Sentenai+-- Author: Antonio Nikishaev <me@lelf.lu>+-- License: Apache+-- Maintainer: Tim McGilchrist <timmcgil@gmail.com>, Mark Hibberd <mark@hibberd.id.au>+-- Stability: experimental+-- Portability: portable+--+-- Haskell-side view of CRDT+--+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE PatternGuards #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE DeriveGeneric #-}++module Network.Riak.CRDT.Types (+ -- * Types+ DataType(..)+ -- ** Counters+ , Counter(..), Count+ -- *** Modification+ , CounterOp(..)+ -- ** Sets+ , Set(..)+ -- *** Modification+ , SetOp(..)+ -- ** Maps+ , Map(..), MapContent+ , MapField(..)+ , MapEntry(..)+ -- *** Inspection+ , xlookup+ -- *** Modification+ , MapOp(..), MapPath(..), MapValueOp(..), mapUpdate, (-/)+ -- ** Registers+ , Register(..)+ -- *** Modification+ , RegisterOp(..)+ -- ** Flags+ , Flag(..)+ -- *** Modification+ , FlagOp(..)+ -- * Misc+ , NonEmpty(..), mapEntryTag, setFromList, MapEntryTag(..)+ ) where++import Control.DeepSeq (NFData)+import Data.ByteString (ByteString)+import Data.Default.Class+import Data.Int (Int64)+import Data.List.NonEmpty+import qualified Data.Map.Strict as M+import Data.Semigroup+import qualified Data.Set as S+import Data.String (IsString(..))+import GHC.Generics (Generic)+++-- | CRDT Map is indexed by MapField, which is a name tagged by a type+-- (there may be different entries with the same name, but different+-- types)+data MapField = MapField MapEntryTag ByteString deriving (Eq,Ord,Show,Generic)++instance NFData MapField++-- | CRDT Map is a Data.Map indexed by 'MapField' and holding+-- 'MapEntry'.+--+-- Maps are specials in a way that they can additionally+-- hold 'Flag's, 'Register's, and most importantly, other 'Map's.+newtype Map = Map MapContent deriving (Eq,Show,Generic)++instance NFData Map++type MapContent = M.Map MapField MapEntry++instance Default Map where+ def = Map M.empty++data MapEntryTag = MapCounterTag+ | MapSetTag+ | MapRegisterTag+ | MapFlagTag+ | MapMapTag+ deriving (Eq,Ord,Show,Generic)++instance NFData MapEntryTag++-- | CRDT Map holds values of type 'MapEntry'+data MapEntry = MapCounter !Counter+ | MapSet !Set+ | MapRegister !Register+ | MapFlag !Flag+ | MapMap !Map+ deriving (Eq,Show,Generic)++instance NFData MapEntry+++mapEntryTag :: MapValueOp -> MapEntryTag+mapEntryTag MapCounterOp{} = MapCounterTag+mapEntryTag MapSetOp{} = MapSetTag+mapEntryTag MapRegisterOp{} = MapRegisterTag+mapEntryTag MapFlagOp{} = MapFlagTag+mapEntryTag MapMapOp{} = MapMapTag+++-- | Selector (“xpath”) inside 'Map'+newtype MapPath = MapPath (NonEmpty ByteString) deriving (Eq,Show)+++-- | map operations+-- It's easier to use 'mapUpdate':+--+-- >>> "x" -/ "y" -/ "z" `mapUpdate` SetAdd "elem"+-- MapUpdate (MapPath ("x" :| ["y","z"])) (MapCounterOp (CounterInc 1))+data MapOp = MapRemove MapField -- ^ remove value in map+ | MapUpdate MapPath MapValueOp -- ^ update value on path by operation+ deriving (Eq,Show)+++-- | Polymprhic version of MapOp for nicer syntax+data MapOp_ op = MapRemove_ MapField+ | MapUpdate_ MapPath op+ deriving Show+++instance IsString MapPath where+ fromString s = MapPath (fromString s :| [])++(-/) :: ByteString -> MapPath -> MapPath+e -/ (MapPath p) = MapPath (e <| p)++infixr 6 -/++class IsMapOp op where toValueOp :: op -> MapValueOp+instance IsMapOp CounterOp where toValueOp = MapCounterOp+instance IsMapOp FlagOp where toValueOp = MapFlagOp+instance IsMapOp RegisterOp where toValueOp = MapRegisterOp+instance IsMapOp SetOp where toValueOp = MapSetOp+++mapUpdate :: IsMapOp o => MapPath -> o -> MapOp+p `mapUpdate` op = MapUpdate p (toValueOp op)++infixr 5 `mapUpdate`++++-- | Lookup a value of a given 'MapEntryTag' type on a given 'MapPath'+-- inside a map+--+-- >>> lookup ("a" -/ "b") MapFlagTag $ { "a"/Map: { "b"/Flag: Flag False } } -- pseudo+-- Just (MapFlag (Flag False))+xlookup :: MapPath -> MapEntryTag -> Map -> Maybe MapEntry+xlookup (MapPath (e :| [])) tag (Map m) = M.lookup (MapField tag e) m+xlookup (MapPath (e :| (r:rs))) tag (Map m)+ | Just (MapMap m') <- inner = xlookup (MapPath (r :| rs)) tag m'+ | otherwise = Nothing+ where inner = M.lookup (MapField MapMapTag e) m+++++-- | Registers can be set to a value+--+-- >>> RegisterSet "foo"+data RegisterOp = RegisterSet !ByteString deriving (Eq,Show)++-- | Flags can be enabled / disabled+--+-- >>> FlagSet True+data FlagOp = FlagSet !Bool deriving (Eq,Show)++-- | Flags can only be held as a 'Map' element.+--+-- Flag can be set or unset+--+-- >>> Flag False+newtype Flag = Flag Bool deriving (Eq,Ord,Show,Generic)++instance NFData Flag++-- | Last-wins monoid for 'Flag'+instance Monoid Flag where+ mempty = Flag False+ mappend = (<>)++-- | Last-wins semigroup for 'Flag'+instance Semigroup Flag where+ a <> b = getLast (Last a <> Last b)++instance Default Flag where+ def = mempty++-- | Registers can only be held as a 'Map' element.+--+-- Register holds a 'ByteString'.+newtype Register = Register ByteString deriving (Eq,Show,Generic)++instance NFData Register++-- | Last-wins monoid for 'Register'+instance Monoid Register where+ mempty = Register ""+ mappend = (<>)++instance Semigroup Register where+ a <> b = getLast (Last a <> Last b)++instance Default Register where+ def = mempty++++-- | Operations on map values+data MapValueOp = MapCounterOp !CounterOp+ | MapSetOp !SetOp+ | MapRegisterOp !RegisterOp+ | MapFlagOp !FlagOp+ | MapMapOp !MapOp+ deriving (Eq,Show)+++-- | CRDT ADT.+--+-- 'Network.Riak.CRDT.Riak.get' operations return value of this type+data DataType = DTCounter Counter+ | DTSet Set+ | DTMap Map+ deriving (Eq,Show,Generic)++instance NFData DataType++-- | CRDT Set is a Data.Set+--+-- >>> Set (Data.Set.fromList ["foo","bar"])+newtype Set = Set (S.Set ByteString) deriving (Eq,Ord,Show,Generic,Monoid)++instance NFData Set++instance Semigroup Set where+ Set a <> Set b = Set (a <> b)++instance Default Set where+ def = Set mempty++-- | CRDT Set operations+data SetOp = SetAdd ByteString -- ^ add element to the set+ --+ -- >>> SetAdd "foo"+ | SetRemove ByteString -- ^ remove element from the set+ --+ -- >>> SetRemove "bar"+ deriving (Eq,Show)++setFromList :: [ByteString] -> Set+setFromList = Set . S.fromList++-- | CRDT Counter hold a integer 'Count'+--+-- >>> Counter 42+newtype Counter = Counter Count deriving (Eq,Ord,Num,Show,Generic)+type Count = Int64++instance NFData Counter++instance Semigroup Counter where+ Counter a <> Counter b = Counter . getSum $ Sum a <> Sum b++instance Monoid Counter where+ mempty = Counter 0+ mappend = (<>)++instance Default Counter where+ def = mempty++-- | Counters can be incremented/decremented+--+-- >>> CounterInc 1+data CounterOp = CounterInc !Count deriving (Eq,Show)++instance Semigroup CounterOp where+ CounterInc x <> CounterInc y = CounterInc . getSum $ Sum x <> Sum y++instance Monoid CounterOp where+ mempty = CounterInc 0+ CounterInc x `mappend` CounterInc y = CounterInc . getSum $ Sum x <> Sum y
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,31 @@ -- '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+ go (p:ps) es =+ catchAny (Riak.withConnectionM p f) (\e -> go ps (e:es))++rotateL :: Int -> [a] -> [a]+rotateL i xs = right ++ left+ where+ (left, right) = splitAt i xs
src/Network/Riak/Connection.hs view
@@ -27,4 +27,4 @@ , pipeline_ ) where -import Network.Riak.Connection.Internal+import Network.Riak.Connection.Internal
src/Network/Riak/Connection/Internal.hs view
@@ -1,4 +1,4 @@-{-# LANGUAGE CPP, OverloadedStrings, RecordWildCards, ScopedTypeVariables #-}+{-# LANGUAGE CPP, OverloadedStrings, RecordWildCards, ScopedTypeVariables, FlexibleContexts, MultiWayIf #-} {-# OPTIONS_GHC -fno-warn-orphans #-} -- |@@ -37,29 +37,28 @@ , recvResponse_ ) where -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.Monad (forM_, replicateM, replicateM_)-import Data.Binary.Put (Put, putWord32be, runPut)+import Control.Concurrent.Async (async, waitBoth)+import Control.Exception (Exception, IOException, throwIO, bracketOnError)+import Control.Monad (forM_, replicateM)+import Data.Binary.Get (Get, Decoder(..), getWord32be, runGetIncremental)+import Data.Binary.Put (Put, putWord32be, runPut, putLazyByteString)+import Data.ByteString (ByteString) import Data.IORef (newIORef, readIORef, writeIORef)-import Data.Int (Int64) import Network.Riak.Connection.NoPush (setNoPush) import Network.Riak.Debug as Debug-import Network.Riak.Protocol.ErrorResponse-import Network.Riak.Protocol.SetClientIDRequest+import Network.Riak.Lens import Network.Riak.Tag (getTag, putTag) import Network.Riak.Types.Internal hiding (MessageTag(..)) import Network.Socket as Socket import Numeric (showHex)-import System.Mem.Weak (addFinalizer)+import Data.ProtoLens (buildMessage) import System.Random (randomIO)-import Text.ProtocolBuffers (messageGetM, messagePutM, messageSize)-import Text.ProtocolBuffers.Get (Get, Result(..), getWord32be, runGet) import qualified Control.Exception as E import qualified Data.ByteString as B+import qualified Data.ByteString.Char8 as B8+import qualified Data.ByteString.Builder as BB import qualified Data.ByteString.Lazy.Char8 as L+import qualified Data.Riak.Proto as Proto import qualified Network.Riak.Types.Internal as T import qualified Network.Socket.ByteString as B import qualified Network.Socket.ByteString.Lazy as L@@ -70,26 +69,26 @@ defaultClient = Client { host = "127.0.0.1" , port = "8087"- , clientID = L.empty+ , clientID = B.empty } -- | Tell the server our client ID. setClientID :: Connection -> ClientID -> IO () setClientID conn i = do- sendRequest conn $ SetClientIDRequest i+ sendRequest conn $ (Proto.defMessage & Proto.clientId .~ i :: Proto.RpbSetClientIdReq) recvResponse_ conn T.SetClientIDResponse -- | Generate a random client ID. makeClientID :: IO ClientID makeClientID = do r <- randomIO :: IO Int- return . L.append "hs_" . L.pack . showHex (abs r) $ ""+ return . B.append "hs_" . B8.pack . showHex (abs r) $ "" -- | Add a random 'ClientID' to a 'Client' if the 'Client' doesn't -- already have one. addClientID :: Client -> IO Client addClientID client- | L.null (clientID client) = do+ | B.null (clientID client) = do i <- makeClientID return client { clientID = i } | otherwise = return client@@ -103,28 +102,30 @@ , addrSocketType = Stream } debug "connect" $ "server " ++ host ++ ":" ++ port ++ ", client ID " ++- L.unpack clientID+ B8.unpack clientID ais <- getAddrInfo (Just hints) (Just host) (Just port) let ai = case ais of (a:_) -> a _ -> moduleError "connect" $ "could not look up server " ++ host ++ ":" ++ port- onIOException "connect" $ do- sock <- socket (addrFamily ai) (addrSocketType ai) (addrProtocol ai)- Socket.connect sock (addrAddress ai)- buf <- newIORef L.empty- let conn = Connection sock client buf- addFinalizer conn $ sClose sock -- Data.Pool doesn't guarantee our disconnect gets called...- setClientID conn clientID- return conn+ onIOException "connect" $+ bracketOnError+ (socket (addrFamily ai) (addrSocketType ai) (addrProtocol ai))+ close $+ \sock -> do+ Socket.connect sock (addrAddress ai)+ buf <- newIORef B.empty+ let conn = Connection sock client buf+ setClientID conn clientID+ return conn -- | Disconnect from a server. disconnect :: Connection -> IO () disconnect Connection{..} = onIOException "disconnect" $ do debug "disconnect" $ "server " ++ host connClient ++ ":" ++ port connClient ++- ", client ID " ++ L.unpack (clientID connClient)- sClose connSock- writeIORef connBuffer L.empty+ ", client ID " ++ B8.unpack (clientID connClient)+ close connSock+ writeIORef connBuffer B.empty -- | 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@@ -134,27 +135,27 @@ recvBufferSize = 16384 {-# INLINE recvBufferSize #-} -recvExactly :: Connection -> Int64 -> IO L.ByteString+recvExactly :: Connection -> Int -> IO ByteString recvExactly Connection{..} n0- | n0 <= 0 = return L.empty+ | n0 <= 0 = return B.empty | otherwise = do bs <- readIORef connBuffer- let (h,t) = L.splitAt n0 bs- len = L.length h+ let (h,t) = B.splitAt n0 bs+ len = B.length h if len == n0 then writeIORef connBuffer t >> return h- else go (reverse (L.toChunks h)) (n0-len)+ else go [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))+ writeIORef connBuffer $! t+ return $ B.concat (reverse (h:acc)) go acc n | n == 0 = do- writeIORef connBuffer L.empty- return $ L.fromChunks (reverse acc)+ writeIORef connBuffer B.empty+ return $ B.concat (reverse acc) | otherwise = do let n' = max recvBufferSize $ min n maxInt bs <- B.recv connSock (fromIntegral n')@@ -165,55 +166,58 @@ recvGet :: Connection -> Get a -> IO a recvGet Connection{..} get = do- let refill = do- bs <- L.recv connSock recvBufferSize- if L.null bs+ let refill :: IO (Maybe ByteString)+ refill = do+ bs <- B.recv connSock recvBufferSize+ if B.null bs then shutdown connSock ShutdownReceive >> return Nothing else return (Just bs)- step (Failed _ err) = moduleError "recvGet" err- step (Finished bs _ r) = writeIORef connBuffer bs >> return r- step (Partial k) = (step . k) =<< refill+ -- step :: Decoder a -> IO a+ step (Fail _ _ err) = moduleError "recvGet" err+ step (Done bs _ r) = writeIORef connBuffer bs >> return r+ step (Partial k) = (step . k) =<< refill mbs <- do buf <- readIORef connBuffer- if L.null buf+ if B.null buf then refill else return (Just buf) case mbs of- Just bs -> step $ runGet get bs+ Just bs ->+ case runGetIncremental get of+ Fail _ _ err -> moduleError "recvGet" err+ Done bs' _ r -> writeIORef connBuffer bs' >> return r+ Partial k -> step (k (Just bs)) Nothing -> moduleError "recvGet" "socket closed" -recvGetN :: Connection -> Int64 -> Get a -> IO a-recvGetN conn n get = do+recvGetN :: Proto.Message a => Connection -> Int -> IO a+recvGetN conn n = do bs <- recvExactly conn n- case runGet get bs of- Finished _ _ r -> return r- Partial k -> case k Nothing of- Finished _ _ r -> return r- Failed _ err -> moduleError "recvGetN" err- Partial _ -> moduleError "recvGetN"- "parser wants more input!?"- Failed _ err -> moduleError "recvGetN" err+ case Proto.decodeMessage bs of+ Left err -> moduleError "recvGetN" err+ Right r -> return r putRequest :: (Request req) => req -> Put putRequest req = do- putWord32be (fromIntegral (1 + messageSize req))+ putWord32be (fromIntegral (1 + L.length bytes)) putTag (messageTag req)- messagePutM req+ putLazyByteString bytes+ where+ bytes :: L.ByteString+ bytes = BB.toLazyByteString (buildMessage req) -instance Exception ErrorResponse+instance Exception Proto.RpbErrorResp -throwError :: ErrorResponse -> a-throwError = throw+throwError :: Proto.RpbErrorResp -> IO a+throwError = throwIO -getResponse :: (Response a) => T.MessageTag -> Get a-getResponse expected = do- tag <- getTag- case undefined of- _| tag == expected -> messageGetM- | tag == T.ErrorResponse -> throwError `fmap` messageGetM- | otherwise ->- moduleError "getResponse" $ "received unexpected response: expected " ++- show expected ++ ", received " ++ show tag+getResponse :: Response a => Connection -> Int -> a -> T.MessageTag -> IO a+getResponse conn len _ expected = do+ tag <- recvGet conn getTag+ if | tag == expected -> recvGetN conn (len-1)+ | tag == T.ErrorResponse -> throwError =<< recvGetN conn (len-1)+ | otherwise ->+ moduleError "getResponse" $ "received unexpected response: expected " +++ show expected ++ ", received " ++ show tag -- | Send a request to the server, and receive its response. exchange :: Exchange req resp => Connection -> req -> IO resp@@ -255,7 +259,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@@ -271,17 +275,16 @@ 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 :: String -> Connection -> T.MessageTag -> Int -> 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- | otherwise -> moduleError func $- "received unexpected response: expected " ++- show expected ++ ", received " ++ show tag+ if | tag == expected -> recvExactly conn (len-1) >> return v+ | tag == T.ErrorResponse -> throwError =<< recvGetN conn len+ | otherwise -> moduleError func $+ "received unexpected response: expected " +++ show expected ++ ", received " ++ show tag debugRecv :: (a -> String) -> IO a -> IO a #ifdef DEBUG@@ -294,20 +297,20 @@ {-# INLINE debugRecv #-} #endif -pipe :: (Request req, Show resp) =>+pipe :: (Request req) => (Connection -> IO resp) -> Connection -> [req] -> IO [resp] pipe _ _ [] = return [] pipe receive conn@Connection{..} reqs = do- ch <- newChan let numReqs = length reqs- _ <- forkIO . replicateM_ numReqs $ writeChan ch =<< receive conn let tag = show (messageTag (head reqs)) if Debug.level > 1 then forM_ reqs $ \req -> debug "pipe" $ ">>> " ++ showM req else debug "pipe" $ ">>> " ++ show numReqs ++ "x " ++ tag- onIOException ("pipe " ++ tag) .- sendAll connSock . runPut . mapM_ putRequest $ reqs- replicateM numReqs $ readChan ch+ receiveResps <- async . replicateM numReqs $ receive conn+ sendReqs <- async . sendAll connSock . runPut . mapM_ putRequest $ reqs+ (_, resps) <- onIOException ("pipe " ++ tag) $+ waitBoth sendReqs receiveResps+ return resps -- | Send a series of requests to the server, back to back, and -- receive a response for each request sent. The sending and@@ -330,16 +333,15 @@ pipeline_ :: (Request req) => Connection -> [req] -> IO () pipeline_ _ [] = return () pipeline_ conn@Connection{..} reqs = do- done <- newEmptyMVar- _ <- forkIO $ do- forM_ reqs (recvResponse_ conn . expectedResponse)- putMVar done ()+ receiveResps <- async $+ forM_ reqs (recvResponse_ conn . expectedResponse) if Debug.level > 1 then forM_ reqs $ \req -> debug "pipe" $ ">>> " ++ showM req else debug "pipe" $ ">>> " ++ show (length reqs) ++ "x " ++ show (messageTag (head reqs))- sendAll connSock . runPut . mapM_ putRequest $ reqs- takeMVar done+ sendReqs <- async . sendAll connSock . runPut . mapM_ putRequest $ reqs+ _ <- onIOException "pipeline_" $ waitBoth sendReqs receiveResps+ return () onIOException :: String -> IO a -> IO a onIOException func act =
src/Network/Riak/Connection/NoPush.hsc view
@@ -21,7 +21,7 @@ import Foreign.Marshal.Utils (with) import Foreign.Ptr (Ptr) import Foreign.Storable (sizeOf)-import Network.Socket (Socket(..))+import Network.Socket (Socket, fdSocket) noPush :: CInt #if defined(TCP_NOPUSH)@@ -37,11 +37,13 @@ setNoPush :: Socket -> Bool -> IO () setNoPush _ _ | noPush == 0 = return ()-setNoPush (MkSocket fd _ _ _ _) onOff = do+-- setNoPush (MkSocket fd _ _ _ _) onOff = do+setNoPush s onOff = do let v = if onOff then 1 else 0 with v $ \ptr ->- throwErrnoIfMinus1_ "setNoPush" $- c_setsockopt fd (#const IPPROTO_TCP) noPush ptr (fromIntegral (sizeOf v))+ throwErrnoIfMinus1_ "setNoPush" $ do+ a <- (fdSocket s)+ c_setsockopt a (#const IPPROTO_TCP) noPush ptr (fromIntegral (sizeOf v)) foreign import ccall unsafe "setsockopt" c_setsockopt :: CInt -> CInt -> CInt -> Ptr CInt -> CInt -> IO CInt
src/Network/Riak/Content.hs view
@@ -13,8 +13,8 @@ module Network.Riak.Content ( -- * Types- Content(..)- , Link.Link(..)+ Proto.RpbContent+ , Proto.RpbLink -- * Functions , empty , binary@@ -22,42 +22,32 @@ , link ) where -import Data.Aeson.Encode (encode)+import Data.Aeson (encode) import Data.Aeson.Types (ToJSON)-import Network.Riak.Protocol.Content (Content(..))+import Data.ByteString (ByteString)+import qualified Data.ByteString.Lazy as LazyByteString+import qualified Data.Riak.Proto as Proto+import Network.Riak.Lens import Network.Riak.Types.Internal (Bucket, Key, Tag)-import qualified Data.ByteString.Lazy.Char8 as L-import qualified Data.Sequence as Seq-import qualified Network.Riak.Protocol.Link as Link -- | Create a link.-link :: Bucket -> Key -> Tag -> Link.Link-link bucket key tag = Link.Link (Just bucket) (Just key) (Just tag)+link :: Bucket -> Key -> Tag -> Proto.RpbLink+link bucket key tag = Proto.defMessage+ & Proto.bucket .~ bucket+ & Proto.key .~ key+ & Proto.tag .~ tag {-# INLINE link #-} -- | An empty piece of content.-empty :: Content-empty = Content { value = L.empty- , content_type = Nothing- , charset = Nothing- , content_encoding = Nothing- , vtag = Nothing- , links = Seq.empty- , last_mod = Nothing- , last_mod_usecs = Nothing- , usermeta = Seq.empty- , indexes = Seq.empty- , deleted = Nothing- }+empty :: Proto.RpbContent+empty = Proto.defMessage -- | Content encoded as @application/octet-stream@.-binary :: L.ByteString -> Content-binary bs = empty { value = bs- , content_type = Just "application/octet-stream"- }+binary :: ByteString -> Proto.RpbContent+binary bs = empty & Proto.value .~ bs+ & Proto.contentType .~ "application/octet-stream" -- | Content encoded as @application/json@.-json :: ToJSON a => a -> Content-json j = empty { value = encode j- , content_type = Just "application/json"- }+json :: ToJSON a => a -> Proto.RpbContent+json j = empty & Proto.value .~ LazyByteString.toStrict (encode j)+ & Proto.contentType .~ "application/json"
src/Network/Riak/Escape.hs view
@@ -4,7 +4,7 @@ -- Module: Network.Riak.Connection -- Copyright: (c) 2011 MailRank, Inc. -- License: Apache--- Maintainer: Mark Hibberd <mark@hibberd.id.au>, Nathan Hunter <nhunter@janrain.com>+-- Maintainer: Tim McGilchrist <timmcgil@gmail.com>, Mark Hibberd <mark@hibberd.id.au>, Nathan Hunter <nhunter@janrain.com> -- Stability: experimental -- Portability: portable --@@ -28,7 +28,6 @@ import Control.Applicative ((<$>)) #endif import Data.Attoparsec.ByteString as A-import Data.Attoparsec.Lazy as AL import Data.Bits ((.|.), (.&.), shiftL, shiftR) import Data.ByteString (ByteString) #if __GLASGOW_HASKELL__ < 710@@ -36,7 +35,7 @@ #endif import Data.Text (Text) import Data.Word (Word8)-import Network.Riak.Functions (mapEither)+import Data.Bifunctor (second, first) import qualified Data.ByteString as B import qualified Data.ByteString.Lazy as L import qualified Data.ByteString.Unsafe as B@@ -49,51 +48,47 @@ -- unescaped. class Escape e where -- | URL-escape a string.- escape :: e -> L.ByteString+ escape :: e -> ByteString -- | URL-unescape a string.- unescape' :: L.ByteString -> Either String e+ unescape' :: ByteString -> Either String e -- | URL-unescape a string that is presumed to be properly escaped. -- If the string is invalid, an error will be thrown that cannot be -- caught from pure code.-unescape :: Escape e => L.ByteString -> e+unescape :: Escape e => ByteString -> e unescape bs = case unescape' bs of Left err -> error $ "Network.Riak.Escape.unescape: " ++ err Right v -> v {-# INLINE unescape #-} instance Escape ByteString where- escape = toLazyByteString . B.foldl escapeWord8 mempty+ escape = L.toStrict . toLazyByteString . B.foldl escapeWord8 mempty {-# INLINE escape #-}- unescape' = AL.eitherResult . AL.parse (toByteString <$> unescapeBS)+ unescape' = A.eitherResult . A.parse (toByteString <$> unescapeBS) {-# INLINE unescape' #-} instance Escape L.ByteString where- escape = toLazyByteString . L.foldl escapeWord8 mempty+ escape = L.toStrict . toLazyByteString . L.foldl escapeWord8 mempty {-# INLINE escape #-}- unescape' = AL.eitherResult . AL.parse (toLazyByteString <$> unescapeBS)+ unescape' = A.eitherResult . A.parse (toLazyByteString <$> unescapeBS) {-# INLINE unescape' #-} instance Escape Text where escape = escape . T.encodeUtf8 {-# INLINE escape #-}- unescape' lbs = case AL.parse (toByteString <$> unescapeBS) lbs of- AL.Done _ bs -> mapEither show id $ T.decodeUtf8' bs- AL.Fail _ _ err -> Left err+ unescape' = (>>= first show . T.decodeUtf8') . A.eitherResult . A.parse (toByteString <$> unescapeBS) {-# INLINE unescape' #-} instance Escape TL.Text where escape = escape . TL.encodeUtf8 {-# INLINE escape #-}- unescape' lbs = case AL.parse (toLazyByteString <$> unescapeBS) lbs of- AL.Done _ bs -> mapEither show id $ TL.decodeUtf8' bs- AL.Fail _ _ err -> Left err+ unescape' = (>>= first show . TL.decodeUtf8') . A.eitherResult . A.parse (toLazyByteString <$> unescapeBS) {-# INLINE unescape' #-} instance Escape [Char] where escape = escape . T.encodeUtf8 . T.pack {-# INLINE escape #-}- unescape' = mapEither id T.unpack . unescape'+ unescape' = second T.unpack . unescape' {-# INLINE unescape' #-} -- | URL-escape a byte from a bytestring.
src/Network/Riak/Functions.hs view
@@ -2,7 +2,7 @@ -- Module: Network.Riak.Functions -- Copyright: (c) 2011 MailRank, Inc. -- License: Apache--- Maintainer: Mark Hibberd <mark@hibberd.id.au>, Nathan Hunter <nhunter@janrain.com>+-- Maintainer: Tim McGilchrist <timmcgil@gmail.com>, Mark Hibberd <mark@hibberd.id.au>, Nathan Hunter <nhunter@janrain.com> -- Stability: experimental -- Portability: portable --@@ -12,7 +12,6 @@ ( strict , lazy- , mapEither ) where import qualified Data.ByteString as B@@ -27,8 +26,3 @@ lazy s | B.null s = L.Empty | otherwise = L.Chunk s L.Empty {-# INLINE lazy #-}--mapEither :: (a -> c) -> (b -> d) -> Either a b -> Either c d-mapEither f _ (Left l) = Left (f l)-mapEither _ g (Right r) = Right (g r)-{-# INLINE mapEither #-}
src/Network/Riak/JSON.hs view
@@ -1,4 +1,6 @@-{-# LANGUAGE DeriveDataTypeable, GeneralizedNewtypeDeriving, CPP #-}+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE CPP #-} -- | -- Module: Network.Riak.JSON@@ -34,17 +36,18 @@ #if __GLASGOW_HASKELL__ < 710 import Data.Monoid (Monoid) #endif+import Data.Semigroup (Semigroup) import Data.Typeable (Typeable) import Network.Riak.Types.Internal import qualified Network.Riak.Value as V newtype JSON a = J { plain :: a -- ^ Unwrap a 'JSON'-wrapped value.- } deriving (Eq, Ord, Show, Read, Bounded, Typeable, Monoid)+ } deriving (Eq, Ord, Show, Read, Bounded, Typeable, Semigroup, Monoid) -- | Wrap up a value so that it will be encoded and decoded as JSON -- when converted to/from 'Content'.-json :: (FromJSON a, ToJSON a) => a -> JSON a+json :: a -> JSON a json = J {-# INLINE json #-} @@ -61,75 +64,77 @@ -- | Retrieve a value. This may return multiple conflicting siblings. -- Choosing among them is your responsibility.-get :: (FromJSON c, ToJSON c) => Connection -> Bucket -> Key -> R+get :: (FromJSON c, ToJSON c) => Connection+ -> Maybe BucketType -> Bucket -> Key -> R -> IO (Maybe ([c], VClock))-get conn bucket key r = fmap convert <$> V.get conn bucket key r+get conn btype bucket' key' r = fmap convert <$> V.get conn btype bucket' key' r -getMany :: (FromJSON c, ToJSON c) => Connection -> Bucket -> [Key] -> R- -> IO [Maybe ([c], VClock)]-getMany conn bucket ks r = map (fmap convert) <$> V.getMany conn bucket ks r+getMany :: (FromJSON c, ToJSON c) => Connection+ -> Maybe BucketType -> Bucket -> [Key] -> R+ -> IO [Maybe ([c], VClock)]+getMany conn btype bucket' ks r = map (fmap convert) <$> V.getMany conn btype bucket' ks r -- | Store a single value. This may return multiple conflicting -- siblings. Choosing among them, and storing a new value, is your -- responsibility. -- -- You should /only/ supply 'Nothing' as a 'T.VClock' if you are sure--- that the given bucket+key combination does not already exist. If--- you omit a 'T.VClock' but the bucket+key /does/ exist, your value--- will not be stored.+-- that the given type+bucket+key combination does not already exist.+-- If you omit a 'T.VClock' but the type+bucket+key /does/ exist, your+-- value will not be stored. put :: (FromJSON c, ToJSON c) =>- Connection -> Bucket -> Key -> Maybe VClock -> c+ Connection -> Maybe BucketType -> Bucket -> Key -> Maybe VClock -> c -> W -> DW -> IO ([c], VClock)-put conn bucket key mvclock val w dw =- convert <$> V.put conn bucket key mvclock (json val) w dw+put conn btype bucket' key' mvclock val w dw =+ convert <$> V.put conn btype bucket' key' mvclock (json val) w dw -- | Store a single value indexed. putIndexed :: (FromJSON c, ToJSON c)- => Connection -> Bucket -> Key -> [IndexValue]+ => Connection -> Maybe BucketType -> Bucket -> Key -> [IndexValue] -> Maybe VClock -> c -> W -> DW -> IO ([c], VClock)-putIndexed conn bucket key ixs mvclock val w dw =- convert <$> V.putIndexed conn bucket key ixs mvclock (json val) w dw+putIndexed conn btype bucket' key' ixs mvclock val w dw =+ convert <$> V.putIndexed conn btype bucket' key' ixs mvclock (json val) w dw -- | Store a single value, without the possibility of conflict -- resolution. -- -- You should /only/ supply 'Nothing' as a 'T.VClock' if you are sure--- that the given bucket+key combination does not already exist. If--- you omit a 'T.VClock' but the bucket+key /does/ exist, your value--- will not be stored, and you will not be notified.+-- that the given type+bucket+key combination does not already exist.+-- If you omit a 'T.VClock' but the type+bucket+key /does/ exist, your+-- value will not be stored, and you will not be notified. put_ :: (FromJSON c, ToJSON c) =>- Connection -> Bucket -> Key -> Maybe VClock -> c+ Connection -> Maybe BucketType -> Bucket -> Key -> Maybe VClock -> c -> W -> DW -> IO ()-put_ conn bucket key mvclock val w dw =- V.put_ conn bucket key mvclock (json val) w dw+put_ conn btype bucket' key' mvclock val w dw =+ V.put_ conn btype bucket' key' mvclock (json val) w dw -- | Store many values. This may return multiple conflicting siblings -- for each value stored. Choosing among them, and storing a new -- value in each case, is your responsibility. -- -- You should /only/ supply 'Nothing' as a 'T.VClock' if you are sure--- that the given bucket+key combination does not already exist. If--- you omit a 'T.VClock' but the bucket+key /does/ exist, your value--- will not be stored.+-- that the given type+bucket+key combination does not already exist.+-- If you omit a 'T.VClock' but the type+bucket+key /does/ exist, your+-- value will not be stored. putMany :: (FromJSON c, ToJSON c) =>- Connection -> Bucket -> [(Key, Maybe VClock, c)]+ Connection -> Maybe BucketType -> Bucket -> [(Key, Maybe VClock, c)] -> W -> DW -> IO [([c], VClock)]-putMany conn bucket puts w dw =- map convert <$> V.putMany conn bucket (map f puts) w dw+putMany conn btype bucket' puts w dw =+ map convert <$> V.putMany conn btype bucket' (map f puts) w dw where f (k,v,c) = (k,v,json c) -- | Store many values, without the possibility of conflict -- resolution. -- -- You should /only/ supply 'Nothing' as a 'T.VClock' if you are sure--- that the given bucket+key combination does not already exist. If--- you omit a 'T.VClock' but the bucket+key /does/ exist, your value--- will not be stored, and you will not be notified.+-- that the given type+bucket+key combination does not already exist.+-- If you omit a 'T.VClock' but the type+bucket+key /does/ exist, your+-- value will not be stored, and you will not be notified. putMany_ :: (FromJSON c, ToJSON c) =>- Connection -> Bucket -> [(Key, Maybe VClock, c)]+ Connection -> Maybe BucketType -> Bucket -> [(Key, Maybe VClock, c)] -> W -> DW -> IO ()-putMany_ conn bucket puts w dw = V.putMany_ conn bucket (map f puts) w dw+putMany_ conn btype bucket' puts w dw = V.putMany_ conn btype bucket' (map f puts) w dw where f (k,v,c) = (k,v,json c) convert :: ([JSON a], VClock) -> ([a], VClock)
src/Network/Riak/JSON/Resolvable.hs view
@@ -41,14 +41,14 @@ -- | Retrieve a single value. If conflicting values are returned, -- 'resolve' is used to choose a winner. get :: (FromJSON c, ToJSON c, Resolvable c) =>- Connection -> Bucket -> Key -> R -> IO (Maybe (c, VClock))+ Connection -> Maybe BucketType -> 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, 'resolve' is used to choose a winner. getMany :: (FromJSON c, ToJSON c, Resolvable c)- => Connection -> Bucket -> [Key] -> R -> IO [Maybe (c, VClock)]+ => Connection -> Maybe BucketType -> Bucket -> [Key] -> R -> IO [Maybe (c, VClock)] getMany = R.getMany J.getMany {-# INLINE getMany #-} @@ -66,7 +66,7 @@ -- being stuck in a conflict resolution loop, it will throw a -- 'ResolutionFailure' exception. modify :: (FromJSON a, ToJSON a, Resolvable a) =>- Connection -> Bucket -> Key -> R -> W -> DW+ Connection -> Maybe BucketType -> Bucket -> Key -> R -> W -> DW -> (Maybe a -> IO (a,b)) -- ^ Modification function. Called with 'Just' the value if -- the key is present, 'Nothing' otherwise.@@ -86,7 +86,7 @@ -- being stuck in a conflict resolution loop, it will throw a -- 'ResolutionFailure' exception. modify_ :: (MonadIO m, FromJSON a, ToJSON a, Resolvable a) =>- Connection -> Bucket -> Key -> R -> W -> DW+ Connection -> Maybe BucketType -> Bucket -> Key -> R -> W -> DW -> (Maybe a -> m a) -> m a modify_ = R.modify_ J.get J.put {-# INLINE modify_ #-}@@ -104,20 +104,20 @@ -- conflict resolution loop, it will throw a 'ResolutionFailure' -- exception. put :: (FromJSON c, ToJSON c, Resolvable c) =>- Connection -> Bucket -> Key -> Maybe VClock -> c -> W -> DW+ Connection -> Maybe BucketType -> Bucket -> Key -> Maybe VClock -> c -> W -> DW -> IO (c, VClock) put = R.put J.put {-# INLINE put #-} -- | Store a single value indexed. putIndexed :: (FromJSON c, ToJSON c, Resolvable c)- => Connection -> Bucket -> Key -> [IndexValue]+ => Connection -> Maybe BucketType -> Bucket -> Key -> [IndexValue] -> Maybe VClock -> c -> W -> DW -> IO (c, VClock)-putIndexed c b k ixs vc cont w' dw' =- R.put (\conn bucket key mvclock val w dw ->- J.putIndexed conn bucket key ixs mvclock val w dw)- c b k vc cont w' dw'+putIndexed c bt b k ixs vc cont w' dw' =+ R.put (\conn btype bucket' key' mvclock val w dw ->+ J.putIndexed conn btype bucket' key' ixs mvclock val w dw)+ c bt b k vc cont w' dw' {-# INLINE putIndexed #-} -- | Store a single value, automatically resolving any vector clock@@ -133,7 +133,7 @@ -- conflict resolution loop, it will throw a 'ResolutionFailure' -- exception. put_ :: (FromJSON c, ToJSON c, Resolvable c) =>- Connection -> Bucket -> Key -> Maybe VClock -> c -> W -> DW+ Connection -> Maybe BucketType -> Bucket -> Key -> Maybe VClock -> c -> W -> DW -> IO () put_ = R.put_ J.put {-# INLINE put_ #-}@@ -153,7 +153,7 @@ -- If this function gives up due to apparently being stuck in a loop, -- it will throw a 'ResolutionFailure' exception. putMany :: (FromJSON c, ToJSON c, Resolvable c) =>- Connection -> Bucket -> [(Key, Maybe VClock, c)] -> W -> DW+ Connection -> Maybe BucketType -> Bucket -> [(Key, Maybe VClock, c)] -> W -> DW -> IO [(c, VClock)] putMany = R.putMany J.putMany {-# INLINE putMany #-}@@ -170,7 +170,7 @@ -- If this function gives up due to apparently being stuck in a loop, -- it will throw a 'ResolutionFailure' exception. putMany_ :: (FromJSON c, ToJSON c, Resolvable c) =>- Connection -> Bucket -> [(Key, Maybe VClock, c)] -> W -> DW+ Connection -> Maybe BucketType -> Bucket -> [(Key, Maybe VClock, c)] -> W -> DW -> IO () putMany_ = R.putMany_ J.putMany {-# INLINE putMany_ #-}
+ src/Network/Riak/Lens.hs view
@@ -0,0 +1,39 @@+{-# LANGUAGE RankNTypes #-}++module Network.Riak.Lens+ ( (&)+ , (^.)+ , (.~)+ , (%~)+ , mapped+ ) where++import Data.Function ((&))+import Data.Functor.Const+import Data.Functor.Identity++type Lens s a+ = forall f. Functor f => (a -> f a) -> (s -> f s)++type Setter s a+ = (a -> Identity a) -> (s -> Identity s)++infixl 8 ^.+(^.) :: s -> Lens s a -> a+s ^. l =+ getConst (l Const s)++infixr 4 .~+(.~) :: Setter s a -> a -> s -> s+(l .~ a) s =+ runIdentity (l (\_ -> Identity a) s)++infixr 4 %~+(%~) :: Setter s a -> (a -> a) -> (s -> s)+(l %~ f) s =+ runIdentity (l (\a -> Identity (f a)) s)+++mapped :: Functor f => Setter (f a) a+mapped f s =+ Identity (fmap (runIdentity . f) s)
src/Network/Riak/Request.hs view
@@ -1,4 +1,5 @@-{-# LANGUAGE OverloadedStrings, CPP #-}+{-# LANGUAGE CPP #-}+{-# LANGUAGE OverloadedStrings #-} -- | -- Module: Network.Riak.Request@@ -11,175 +12,250 @@ -- Smart constructors for Riak types. These functions correctly -- URL-escape bucket, key, and link names. You should thus use them -- in preference to the raw data constructors.- module Network.Riak.Request- (- -- * Connection management- PingRequest- , ping- , GetClientIDRequest- , getClientID- , GetServerInfoRequest- , getServerInfo+ ( -- * Connection management+ Proto.RpbPingReq,+ ping,+ Proto.RpbGetClientIdReq,+ getClientID,+ Proto.RpbGetServerInfoReq,+ getServerInfo,+ -- * Data management- , Get.GetRequest- , get- , getByIndex- , Index.IndexRequest- , Put.PutRequest- , put- , Del.DeleteRequest- , delete+ Proto.RpbGetReq,+ get,+ getByIndex,+ Proto.RpbIndexReq,+ Proto.RpbPutReq,+ put,+ Proto.RpbDelReq,+ delete,+ -- * Metadata- , Link.Link- , link- , ListBucketsRequest- , listBuckets- , Keys.ListKeysRequest- , listKeys- , GetBucket.GetBucketRequest- , getBucket- , SetBucket.SetBucketRequest- , setBucket+ Proto.RpbLink,+ link,+ Proto.RpbListBucketsReq,+ listBuckets,+ Proto.RpbListKeysReq,+ listKeys,+ Proto.RpbGetBucketReq,+ getBucket,+ Proto.RpbSetBucketReq,+ setBucket,+ getBucketType,+ -- * Map/reduce- , MapReduceRequest- , mapReduce- ) where+ Proto.RpbMapRedReq,+ mapReduce, + -- * Search+ search,+ getIndex,+ putIndex,+ deleteIndex,+ )+where+ #if __GLASGOW_HASKELL__ < 710 import Control.Applicative ((<$>)) #endif+import Data.ByteString (ByteString) import qualified Data.ByteString.Char8 as B8+#if __GLASGOW_HASKELL__ < 804 import Data.Monoid-import Network.Riak.Protocol.BucketProps hiding (r,rw)-import Network.Riak.Protocol.Content-import Network.Riak.Protocol.GetClientIDRequest-import Network.Riak.Protocol.GetServerInfoRequest-import Network.Riak.Protocol.ListBucketsRequest-import Network.Riak.Protocol.MapReduceRequest-import Network.Riak.Protocol.PingRequest-import Network.Riak.Types.Internal hiding (MessageTag(..))+#endif+import qualified Data.Riak.Proto as Proto import Network.Riak.Escape (escape)-import qualified Network.Riak.Protocol.DeleteRequest as Del-import qualified Network.Riak.Protocol.Link as Link-import qualified Network.Riak.Protocol.GetBucketRequest as GetBucket-import qualified Network.Riak.Protocol.GetRequest as Get-import qualified Network.Riak.Protocol.IndexRequest as Index-import qualified Network.Riak.Protocol.IndexRequest.IndexQueryType as IndexQueryType-import qualified Network.Riak.Protocol.ListKeysRequest as Keys-import qualified Network.Riak.Protocol.PutRequest as Put-import qualified Network.Riak.Protocol.SetBucketRequest as SetBucket+import Network.Riak.Lens+import Network.Riak.Types.Internal hiding (MessageTag (..)) -- | Create a ping request.-ping :: PingRequest-ping = PingRequest+ping :: Proto.RpbPingReq+ping = Proto.defMessage {-# INLINE ping #-} +{-+-- | Create a dtUpdate request.+dtUpdate :: DtUpdate.DtUpdateRequest+dtUpdate = DtUpdate.DtUpdateRequest+{-# INLINE dtUpdate #-}+-}+ -- | Create a client-ID request.-getClientID :: GetClientIDRequest-getClientID = GetClientIDRequest+getClientID :: Proto.RpbGetClientIdReq+getClientID = Proto.defMessage {-# INLINE getClientID #-} -- | Create a server-info request.-getServerInfo :: GetServerInfoRequest-getServerInfo = GetServerInfoRequest+getServerInfo :: Proto.RpbGetServerInfoReq+getServerInfo = Proto.defMessage {-# INLINE getServerInfo #-} -- | Create a get request. The bucket and key names are URL-escaped.-get :: Bucket -> Key -> R -> Get.GetRequest-get bucket key r = Get.GetRequest { Get.bucket = escape bucket- , Get.key = escape key- , Get.r = fromQuorum r- , Get.pr = Nothing- , Get.basic_quorum = Nothing- , Get.notfound_ok = Nothing- , Get.if_modified = Nothing- , Get.head = Nothing- , Get.deletedvclock = Nothing- , Get.timeout = Nothing- , Get.sloppy_quorum = Nothing- , Get.n_val = Nothing- }+get :: Maybe BucketType -> Bucket -> Key -> R -> Proto.RpbGetReq+get btype bucket key r =+ Proto.defMessage+ & Proto.bucket .~ escape bucket+ & Proto.key .~ escape key+ & Proto.maybe'r .~ fromQuorum r+ & Proto.maybe'type' .~ btype -- no escaping intentional+ -- TODO don't escape anything {-# INLINE get #-} -- | Create a secondary index request. Bucket, key and index names and -- values are URL-escaped.-getByIndex :: Bucket -> IndexQuery- -> Index.IndexRequest+getByIndex ::+ Bucket ->+ IndexQuery ->+ Proto.RpbIndexReq getByIndex bucket q =- case q of- IndexQueryExactInt index key ->- req (index <> "_int") (showIntKey key)- IndexQueryType.Eq Nothing Nothing- IndexQueryExactBin index key ->- req (index <> "_bin") (showBsKey $ key)- IndexQueryType.Eq Nothing Nothing- IndexQueryRangeInt index from to ->- req (index <> "_int") Nothing- IndexQueryType.Range (showIntKey from) (showIntKey to)- IndexQueryRangeBin index from to ->- req (index <> "_bin") Nothing- IndexQueryType.Range (showBsKey from) (showBsKey to)+ case q of+ IndexQueryExactInt index key ->+ req+ (index <> "_int")+ (showIntKey key)+ Proto.RpbIndexReq'Eq+ Nothing+ Nothing+ IndexQueryExactBin index key ->+ req+ (index <> "_bin")+ (showBsKey $ key)+ Proto.RpbIndexReq'Eq+ Nothing+ Nothing+ IndexQueryRangeInt index from to ->+ req+ (index <> "_int")+ Nothing+ Proto.RpbIndexReq'Range+ (showIntKey from)+ (showIntKey to)+ IndexQueryRangeBin index from to ->+ req+ (index <> "_bin")+ Nothing+ Proto.RpbIndexReq'Range+ (showBsKey from)+ (showBsKey to) where showIntKey = Just . escape . B8.pack . show showBsKey = Just . escape+ req ::+ ByteString ->+ Maybe ByteString ->+ Proto.RpbIndexReq'IndexQueryType ->+ Maybe ByteString ->+ Maybe ByteString ->+ Proto.RpbIndexReq req i k qt rmin rmax =- Index.IndexRequest { Index.bucket = escape bucket- , Index.index = escape i- , Index.qtype = qt- , Index.key = k- , Index.range_min = rmin- , Index.range_max = rmax- , Index.return_terms = Nothing- , Index.stream = Nothing- , Index.max_results = Nothing- , Index.continuation = Nothing- , Index.timeout = Nothing }+ Proto.defMessage+ & Proto.bucket .~ escape bucket+ & Proto.index .~ escape i+ & Proto.qtype .~ qt+ & Proto.maybe'key .~ k+ & Proto.maybe'rangeMin .~ rmin+ & Proto.maybe'rangeMax .~ rmax -- | Create a put request. The bucket and key names are URL-escaped. -- Any 'Link' values inside the 'Content' are assumed to have been -- constructed with the 'link' function, and hence /not/ escaped.-put :: Bucket -> Key -> Maybe VClock -> Content -> W -> DW -> Bool- -> Put.PutRequest-put bucket key mvclock cont mw mdw returnBody =- Put.PutRequest (escape bucket) (Just $ escape key) (fromVClock <$> mvclock)- cont (fromQuorum mw) (fromQuorum mdw) (Just returnBody)- Nothing Nothing Nothing Nothing Nothing Nothing Nothing Nothing+put ::+ Maybe BucketType ->+ Bucket ->+ Key ->+ Maybe VClock ->+ Proto.RpbContent ->+ W ->+ DW ->+ Bool ->+ Proto.RpbPutReq+put btype bucket key mvclock cont mw mdw returnBody =+ Proto.defMessage+ & Proto.bucket .~ escape bucket+ & Proto.key .~ escape key+ & Proto.maybe'vclock .~ (fromVClock <$> mvclock)+ & Proto.content .~ cont+ & Proto.maybe'w .~ fromQuorum mw+ & Proto.maybe'dw .~ fromQuorum mdw+ & Proto.returnBody .~ returnBody+ & Proto.maybe'type' .~ btype -- same as get {-# INLINE put #-} -- | Create a link. The bucket and key names are URL-escaped.-link :: Bucket -> Key -> Tag -> Link.Link-link bucket key = Link.Link (Just (escape bucket)) (Just (escape key)) . Just+link :: Bucket -> Key -> Tag -> Proto.RpbLink+link bucket key tag =+ Proto.defMessage+ & Proto.bucket .~ escape bucket+ & Proto.key .~ escape key+ & Proto.tag .~ tag {-# INLINE link #-} -- | Create a delete request. The bucket and key names are URL-escaped.-delete :: Bucket -> Key -> RW -> Del.DeleteRequest-delete bucket key rw = Del.DeleteRequest (escape bucket) (escape key)- (fromQuorum rw) Nothing Nothing Nothing- Nothing Nothing Nothing Nothing Nothing Nothing+delete :: Maybe BucketType -> Bucket -> Key -> RW -> Proto.RpbDelReq+delete btype bucket key rw =+ Proto.defMessage+ & Proto.bucket .~ escape bucket+ & Proto.key .~ escape key+ & Proto.maybe'rw .~ fromQuorum rw+ & Proto.maybe'type' .~ btype -- same as get {-# INLINE delete #-} -- | Create a list-buckets request.-listBuckets :: ListBucketsRequest-listBuckets = ListBucketsRequest Nothing Nothing+listBuckets :: Maybe BucketType -> Proto.RpbListBucketsReq+listBuckets btype = Proto.defMessage & Proto.maybe'type' .~ btype {-# INLINE listBuckets #-} --- | Create a list-keys request. The bucket name is URL-escaped.-listKeys :: Bucket -> Keys.ListKeysRequest-listKeys b = Keys.ListKeysRequest (escape b) Nothing+-- | Create a list-keys request. The bucket type and name are URL-escaped.+listKeys :: Maybe BucketType -> Bucket -> Proto.RpbListKeysReq+listKeys t b =+ Proto.defMessage & Proto.maybe'type' .~ (escape <$> t)+ & Proto.bucket .~ escape b {-# INLINE listKeys #-} --- | Create a get-bucket request. The bucket name is URL-escaped.-getBucket :: Bucket -> GetBucket.GetBucketRequest-getBucket = GetBucket.GetBucketRequest . escape+-- | Create a get-bucket request. The bucket type and name are URL-escaped.+getBucket :: Maybe BucketType -> Bucket -> Proto.RpbGetBucketReq+getBucket t b =+ Proto.defMessage & Proto.maybe'type' .~ (escape <$> t)+ & Proto.bucket .~ escape b {-# INLINE getBucket #-} --- | Create a set-bucket request. The bucket name is URL-escaped.-setBucket :: Bucket -> BucketProps -> SetBucket.SetBucketRequest-setBucket = SetBucket.SetBucketRequest . escape+-- | Create a set-bucket request. The bucket type and name are URL-escaped.+setBucket :: Maybe BucketType -> Bucket -> Proto.RpbBucketProps -> Proto.RpbSetBucketReq+setBucket t b ps =+ Proto.defMessage & Proto.bucket .~ escape b+ & Proto.maybe'type' .~ (escape <$> t)+ & Proto.props .~ ps {-# INLINE setBucket #-} +-- | Create a get-bucket-type request. The bucket type is URL-escaped.+getBucketType :: BucketType -> Proto.RpbGetBucketTypeReq+getBucketType t = Proto.defMessage & Proto.type' .~ escape t+ -- | Create a map-reduce request.-mapReduce :: Job -> MapReduceRequest-mapReduce (JSON bs) = MapReduceRequest bs "application/json"-mapReduce (Erlang bs) = MapReduceRequest bs "application/x-erlang-binary"+mapReduce :: Job -> Proto.RpbMapRedReq+mapReduce (JSON bs) =+ Proto.defMessage & Proto.request .~ bs+ & Proto.contentType .~ "application/json"+mapReduce (Erlang bs) =+ Proto.defMessage & Proto.request .~ bs+ & Proto.contentType .~ "application/x-erlang-binary"++-- | Create a search request+search :: SearchQuery -> Index -> Proto.RpbSearchQueryReq+search q ix =+ Proto.defMessage+ & Proto.q .~ q+ & Proto.index .~ escape ix++getIndex :: Maybe Index -> Proto.RpbYokozunaIndexGetReq+getIndex ix = Proto.defMessage & Proto.maybe'name .~ ix++putIndex :: IndexInfo -> Maybe Timeout -> Proto.RpbYokozunaIndexPutReq+putIndex info timeout =+ Proto.defMessage+ & Proto.index .~ info+ & Proto.maybe'timeout .~ timeout++deleteIndex :: Index -> Proto.RpbYokozunaIndexDeleteReq+deleteIndex ix = Proto.defMessage & Proto.name .~ ix
src/Network/Riak/Resolvable/Internal.hs view
@@ -45,6 +45,7 @@ #if __GLASGOW_HASKELL__ < 710 import Data.Monoid (Monoid(mappend)) #endif+import Data.Semigroup (Semigroup) import Data.Typeable (Typeable) import Network.Riak.Debug (debugValues) import Network.Riak.Types.Internal hiding (MessageTag(..))@@ -86,9 +87,9 @@ -- | 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)+ deriving (Eq, Ord, Read, Show, Typeable, Data, Semigroup, Monoid, FromJSON, ToJSON) -instance (Eq a, Show a, Monoid a) => Resolvable (ResolvableMonoid a) where+instance (Show a, Monoid a) => Resolvable (ResolvableMonoid a) where resolve = mappend {-# INLINE resolve #-} @@ -98,23 +99,25 @@ resolve _ b = b {-# INLINE resolve #-} -type Get a = Connection -> Bucket -> Key -> R -> IO (Maybe ([a], VClock))+type Get a = Connection -> Maybe BucketType -> Bucket -> Key -> R -> IO (Maybe ([a], VClock)) get :: (Resolvable a) => Get a- -> (Connection -> Bucket -> Key -> R -> IO (Maybe (a, VClock)))-get doGet conn bucket key r =- fmap (first resolveMany) `fmap` doGet conn bucket key r+ -> (Connection -> Maybe BucketType -> Bucket -> Key -> R -> IO (Maybe (a, VClock)))+get doGet conn btype bucket' key' r =+ fmap (first resolveMany) `fmap` doGet conn btype bucket' key' r {-# INLINE get #-} 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+ (Connection -> Maybe BucketType -> Bucket -> [Key]+ -> R -> IO [Maybe ([a], VClock)])+ -> Connection -> Maybe BucketType -> Bucket -> [Key]+ -> R -> IO [Maybe (a, VClock)]+getMany doGet conn bt b ks r =+ map (fmap (first resolveMany)) `fmap` doGet conn bt b ks r {-# INLINE getMany #-} -- If Riak receives a put request with no vclock, and the given--- bucket+key already exists, it will treat the missing vclock as+-- type+bucket+key already exists, it will treat the missing vclock as -- stale, ignore the put request, and send back whatever values it -- currently knows about. The same problem will arise if we send a -- vclock that really is stale, but that's much less likely to occur.@@ -122,17 +125,17 @@ -- of both put and putMany below, but we do not (can not?) handle the -- stale-vclock case. -type Put a = Connection -> Bucket -> Key -> Maybe VClock -> a -> W -> DW+type Put a = Connection -> Maybe BucketType -> Bucket -> Key -> Maybe VClock -> a -> W -> DW -> IO ([a], VClock) put :: (Resolvable a) => Put a- -> Connection -> Bucket -> Key -> Maybe VClock -> a -> W -> DW+ -> Connection -> Maybe BucketType -> Bucket -> Key -> Maybe VClock -> a -> W -> DW -> IO (a, VClock)-put doPut conn bucket key mvclock0 val0 w dw = do+put doPut conn btype bucket' key' mvclock0 val0 w dw = do let go !i val mvclock | i == maxRetries = throwIO RetriesExceeded | otherwise = do- (xs, vclock) <- doPut conn bucket key mvclock val w dw+ (xs, vclock) <- doPut conn btype bucket' key' mvclock val w dw case xs of [x] | i > 0 || isJust mvclock -> return (x, vclock) (_:_) -> do debugValues "put" "conflict" xs@@ -148,45 +151,47 @@ {-# INLINE maxRetries #-} put_ :: (Resolvable a) =>- (Connection -> Bucket -> Key -> Maybe VClock -> a -> W -> DW+ (Connection -> Maybe BucketType -> Bucket -> Key -> Maybe VClock -> a -> W -> DW -> IO ([a], VClock))- -> Connection -> Bucket -> Key -> Maybe VClock -> a -> W -> DW+ -> Connection -> Maybe BucketType -> 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 ()+put_ doPut conn btype bucket' key' mvclock0 val0 w dw =+ put doPut conn btype bucket' key' mvclock0 val0 w dw >> return () {-# INLINE put_ #-} modify :: (MonadIO m, Resolvable a) => Get a -> Put a- -> Connection -> Bucket -> Key -> R -> W -> DW -> (Maybe a -> m (a,b))+ -> Connection -> Maybe BucketType -> Bucket+ -> Key -> R -> W -> DW -> (Maybe a -> m (a,b)) -> m (a,b)-modify doGet doPut conn bucket key r w dw act = do- a0 <- liftIO $ get doGet conn bucket key r+modify doGet doPut conn btype bucket' key' r w dw act = do+ a0 <- liftIO $ get doGet conn btype bucket' key' r (a,b) <- act (fst <$> a0)- (a',_) <- liftIO $ put doPut conn bucket key (snd <$> a0) a w dw+ (a',_) <- liftIO $ put doPut conn btype bucket' key' (snd <$> a0) a w dw return (a',b) {-# INLINE modify #-} modify_ :: (MonadIO m, Resolvable a) => Get a -> Put a- -> Connection -> Bucket -> Key -> R -> W -> DW -> (Maybe a -> m a)+ -> Connection -> Maybe BucketType -> Bucket+ -> Key -> R -> W -> DW -> (Maybe a -> m a) -> m a-modify_ doGet doPut conn bucket key r w dw act = do- a0 <- liftIO $ get doGet conn bucket key r+modify_ doGet doPut conn btype bucket' key' r w dw act = do+ a0 <- liftIO $ get doGet conn btype bucket' key' r a <- act (fst <$> a0)- liftIO $ fst <$> put doPut conn bucket key (snd <$> a0) a w dw+ liftIO $ fst <$> put doPut conn btype bucket' key' (snd <$> a0) a w dw {-# INLINE modify_ #-} putMany :: (Resolvable a) =>- (Connection -> Bucket -> [(Key, Maybe VClock, a)] -> W -> DW+ (Connection -> Maybe BucketType -> Bucket -> [(Key, Maybe VClock, a)] -> W -> DW -> IO [([a], VClock)])- -> Connection -> Bucket -> [(Key, Maybe VClock, a)] -> W -> DW+ -> Connection -> Maybe BucketType -> Bucket -> [(Key, Maybe VClock, a)] -> W -> DW -> IO [(a, VClock)]-putMany doPut conn bucket puts0 w dw = go (0::Int) [] . zip [(0::Int)..] $ puts0+putMany doPut conn btype bucket' puts0 w dw = go (0::Int) [] . zip [(0::Int)..] $ puts0 where go _ acc [] = return . map snd . sortBy (compare `on` fst) $ acc go !i acc puts | i == maxRetries = throwIO RetriesExceeded | otherwise = do- rs <- doPut conn bucket (map snd puts) w dw+ rs <- doPut conn btype bucket' (map snd puts) w dw let (conflicts, ok) = partitionEithers $ zipWith mush puts rs unless (null conflicts) $ debugValues "putMany" "conflicts" conflicts@@ -200,11 +205,12 @@ {-# INLINE putMany #-} putMany_ :: (Resolvable a) =>- (Connection -> Bucket -> [(Key, Maybe VClock, a)] -> W -> DW+ (Connection -> Maybe BucketType -> 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 ()+ -> Connection -> Maybe BucketType -> Bucket+ -> [(Key, Maybe VClock, a)] -> W -> DW -> IO ()+putMany_ doPut conn btype bucket' puts0 w dw =+ putMany doPut conn btype bucket' puts0 w dw >> return () {-# INLINE putMany_ #-} resolveMany' :: (Resolvable a) => a -> [a] -> a
src/Network/Riak/Response.hs view
@@ -1,4 +1,4 @@-{-# LANGUAGE RecordWildCards, CPP #-}+{-# LANGUAGE NamedFieldPuns, RecordWildCards, CPP, OverloadedStrings #-} -- | -- Module: Network.Riak.Request@@ -24,57 +24,74 @@ , listBuckets , getBucket , unescapeLinks+ , search+ , getIndex ) where #if __GLASGOW_HASKELL__ < 710 import Control.Applicative ((<$>))+import Data.Semigroup+import Control.Arrow ((&&&))+import Control.Monad (join) #endif-import Data.Maybe (fromMaybe)+import qualified Data.Riak.Proto as Proto import Network.Riak.Escape (unescape)-import Network.Riak.Protocol.BucketProps-import Network.Riak.Protocol.Content-import Network.Riak.Protocol.GetBucketResponse-import Network.Riak.Protocol.GetClientIDResponse-import Network.Riak.Protocol.GetResponse-import Network.Riak.Protocol.Link-import Network.Riak.Protocol.ListBucketsResponse-import Network.Riak.Protocol.PutResponse+import Network.Riak.Lens import Network.Riak.Types.Internal hiding (MessageTag(..))-import qualified Network.Riak.Protocol.Link as Link-import qualified Data.ByteString.Lazy as L-import qualified Data.Sequence as Seq -getClientID :: GetClientIDResponse -> ClientID-getClientID = client_id+import Data.ByteString (ByteString)+import Data.Foldable (toList)++getClientID :: Proto.RpbGetClientIdResp -> ClientID+getClientID = (^. Proto.clientId) {-# INLINE getClientID #-} -- | Construct a get response. Bucket and key names in links are -- URL-unescaped.-get :: Maybe GetResponse -> Maybe (Seq.Seq Content, VClock)-get (Just (GetResponse content (Just vclock) _))- = Just (unescapeLinks <$> content, VClock vclock)-get _ = Nothing+get :: Maybe Proto.RpbGetResp -> Maybe ([Proto.RpbContent], VClock)+get mresponse = do+ response <- mresponse+ let content = response ^. Proto.content+ vclock <- response ^. Proto.maybe'vclock+ Just (unescapeLinks <$> content, VClock vclock) {-# INLINE get #-} -- | Construct a put response. Bucket and key names in links are -- URL-unescaped.-put :: PutResponse -> (Seq.Seq Content, VClock)-put PutResponse{..} = (unescapeLinks <$> content,- VClock (fromMaybe L.empty vclock))+put :: Proto.RpbPutResp -> ([Proto.RpbContent], VClock)+put response =+ ( unescapeLinks <$> (response ^. Proto.content)+ , VClock (response ^. Proto.vclock)+ ) {-# INLINE put #-} -- | Construct a list-buckets response. Bucket names are unescaped.-listBuckets :: ListBucketsResponse -> Seq.Seq Bucket-listBuckets = fmap unescape . buckets+listBuckets :: Proto.RpbListBucketsResp -> [Bucket]+listBuckets = fmap unescape . (^. Proto.buckets) {-# INLINE listBuckets #-} -getBucket :: GetBucketResponse -> BucketProps-getBucket = props+getBucket :: Proto.RpbGetBucketResp -> Proto.RpbBucketProps+getBucket = (^. Proto.props) {-# INLINE getBucket #-} -- | URL-unescape the names of keys and buckets in the links of a -- 'Content' value.-unescapeLinks :: Content -> Content-unescapeLinks c = c { links = go <$> links c }- where go l = l { bucket = unescape <$> bucket l- , Link.key = unescape <$> Link.key l }+unescapeLinks :: Proto.RpbContent -> Proto.RpbContent+unescapeLinks = Proto.links . mapped %~ go+ where go :: Proto.RpbLink -> Proto.RpbLink+ go l = l & Proto.bucket %~ unescape+ & Proto.key %~ unescape++search :: Proto.RpbSearchQueryResp -> SearchResult+search resp =+ SearchResult+ { docs = fmap (fmap unpair . (^. Proto.fields)) (resp ^. Proto.docs)+ , maxScore = resp ^. Proto.maybe'maxScore+ , numFound = resp ^. Proto.maybe'numFound+ }+ where+ unpair :: Proto.RpbPair -> (ByteString, Maybe ByteString)+ unpair pair = (pair ^. Proto.key, pair ^. Proto.maybe'value)++getIndex :: Proto.RpbYokozunaIndexGetResp -> [IndexInfo]+getIndex = toList . (^. Proto.index)
+ src/Network/Riak/Search.hs view
@@ -0,0 +1,64 @@+-- | module: Network.Riak.Search+-- copyright: (c) 2016 Sentenai+-- author: Antonio Nikishaev <me@lelf.lu>+-- license: Apache+--+-- Solr search+--+-- http://docs.basho.com/riak/2.1.3/dev/using/search/+{-# LANGUAGE CPP #-}+module Network.Riak.Search+ ( IndexInfo+ , SearchResult(..)+ , Score+ , indexInfo+ , getIndex+ , putIndex+ , deleteIndex+ , searchRaw+ ) where++#if __GLASGOW_HASKELL__ <= 708+import Control.Applicative+#endif+import qualified Data.Riak.Proto as Proto+import Network.Riak.Connection.Internal+import Network.Riak.Lens+import qualified Network.Riak.Request as Req+import qualified Network.Riak.Response as Resp+import Network.Riak.Types.Internal++-- | 'IndexInfo' smart constructor.+--+-- If 'Nothing', @schema@ defaults to @"_yz_default"@.+--+-- If 'Nothing', @n@ defaults to the default @n@ value for buckets that have not+-- explicitly set the property. In the default installation of @riak@, this is+-- 3 (see https://github.com/basho/riak_core/blob/develop/priv/riak_core.schema).+indexInfo :: Index -> Maybe Schema -> Maybe N -> IndexInfo+indexInfo ix schema nval = Proto.defMessage & Proto.name .~ ix+ & Proto.maybe'schema .~ schema+ & Proto.maybe'nVal .~ nval++-- | Get an index info for @Just index@, or get all indexes for @Nothing@.+--+-- https://docs.basho.com/riak/kv/2.1.4/developing/api/protocol-buffers/yz-index-get/+getIndex :: Connection -> Maybe Index -> IO [IndexInfo]+getIndex conn ix = Resp.getIndex <$> exchange conn (Req.getIndex ix)++-- | Create a new index or modify an existing index.+--+-- https://docs.basho.com/riak/kv/2.1.4/developing/api/protocol-buffers/yz-index-put/+putIndex :: Connection -> IndexInfo -> Maybe Timeout -> IO ([Proto.RpbContent], VClock)+putIndex conn info timeout = Resp.put <$> exchange conn (Req.putIndex info timeout)++-- | Delete an index.+--+-- https://docs.basho.com/riak/kv/2.1.4/developing/api/protocol-buffers/yz-index-delete/+deleteIndex :: Connection -> Index -> IO ()+deleteIndex conn ix = exchange_ conn (Req.deleteIndex ix)++-- | Search by raw 'SearchQuery' request (a 'Data.ByteString.Lazy.Bytestring')+-- using an 'Index'.+searchRaw :: Connection -> SearchQuery -> Index -> IO SearchResult+searchRaw conn q ix = Resp.search <$> exchange conn (Req.search q ix)
src/Network/Riak/Tag.hs view
@@ -1,4 +1,5 @@ {-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE CPP #-} {-# OPTIONS_GHC -fno-warn-orphans #-} -- |@@ -17,224 +18,365 @@ , getTag ) where +import Data.Binary.Get (Get, getWord8) import Data.Binary.Put (Put, putWord8)-import Network.Riak.Protocol.DeleteRequest-import Network.Riak.Protocol.ErrorResponse-import Network.Riak.Protocol.GetBucketRequest-import Network.Riak.Protocol.GetBucketResponse-import Network.Riak.Protocol.GetClientIDRequest-import Network.Riak.Protocol.GetClientIDResponse-import Network.Riak.Protocol.GetRequest-import Network.Riak.Protocol.GetResponse-import Network.Riak.Protocol.IndexRequest-import Network.Riak.Protocol.IndexResponse-import Network.Riak.Protocol.GetServerInfoRequest-import Network.Riak.Protocol.ListBucketsRequest-import Network.Riak.Protocol.ListBucketsResponse-import Network.Riak.Protocol.ListKeysRequest-import Network.Riak.Protocol.ListKeysResponse-import Network.Riak.Protocol.MapReduce-import Network.Riak.Protocol.MapReduceRequest-import Network.Riak.Protocol.PingRequest-import Network.Riak.Protocol.PutRequest-import Network.Riak.Protocol.PutResponse-import Network.Riak.Protocol.ServerInfo-import Network.Riak.Protocol.SetBucketRequest-import Network.Riak.Protocol.SetClientIDRequest+import Data.Word (Word8)+import qualified Data.HashMap.Strict as HM+#if __GLASGOW_HASKELL__ <= 708+import Control.Applicative+#endif+import Data.Riak.Proto+import Data.Tuple (swap) import Network.Riak.Types.Internal as Types-import Text.ProtocolBuffers.Get (Get, getWord8) -instance Tagged ErrorResponse where+instance Tagged RpbErrorResp where messageTag _ = Types.ErrorResponse {-# INLINE messageTag #-} -instance Response ErrorResponse+instance Response RpbErrorResp -instance Tagged PingRequest where+instance Tagged RpbPingReq where messageTag _ = Types.PingRequest {-# INLINE messageTag #-} -instance Request PingRequest where+instance Request RpbPingReq where expectedResponse _ = Types.PingResponse {-# INLINE expectedResponse #-} -instance Tagged GetClientIDRequest where+instance Tagged RpbGetClientIdReq where messageTag _ = Types.GetClientIDRequest {-# INLINE messageTag #-} -instance Request GetClientIDRequest where+instance Request RpbGetClientIdReq where expectedResponse _ = Types.GetClientIDResponse {-# INLINE expectedResponse #-} -instance Tagged GetClientIDResponse where+instance Tagged RpbGetClientIdResp where messageTag _ = Types.GetClientIDResponse {-# INLINE messageTag #-} -instance Response GetClientIDResponse+instance Response RpbGetClientIdResp -instance Exchange GetClientIDRequest GetClientIDResponse+instance Exchange RpbGetClientIdReq RpbGetClientIdResp -instance Tagged SetClientIDRequest where+instance Tagged RpbSetClientIdReq where messageTag _ = Types.SetClientIDRequest {-# INLINE messageTag #-} -instance Request SetClientIDRequest where+instance Request RpbSetClientIdReq where expectedResponse _ = Types.SetClientIDResponse {-# INLINE expectedResponse #-} -instance Tagged GetServerInfoRequest where+instance Tagged RpbGetServerInfoReq where messageTag _ = Types.GetServerInfoRequest {-# INLINE messageTag #-} -instance Request GetServerInfoRequest where+instance Request RpbGetServerInfoReq where expectedResponse _ = Types.GetServerInfoResponse {-# INLINE expectedResponse #-} -instance Tagged ServerInfo where+instance Tagged RpbGetServerInfoResp where messageTag _ = Types.GetServerInfoResponse {-# INLINE messageTag #-} -instance Response ServerInfo+instance Response RpbGetServerInfoResp -instance Exchange GetServerInfoRequest ServerInfo+instance Exchange RpbGetServerInfoReq RpbGetServerInfoResp -instance Tagged GetRequest where+instance Tagged RpbGetReq where messageTag _ = Types.GetRequest {-# INLINE messageTag #-} -instance Tagged IndexRequest where+instance Tagged RpbIndexReq where messageTag _ = Types.IndexRequest {-# INLINE messageTag #-} -instance Request GetRequest where+instance Request RpbGetReq where expectedResponse _ = Types.GetResponse {-# INLINE expectedResponse #-} -instance Request IndexRequest where+instance Request RpbIndexReq where expectedResponse _ = Types.IndexResponse {-# INLINE expectedResponse #-} -instance Tagged GetResponse where+instance Tagged RpbGetResp where messageTag _ = Types.GetResponse {-# INLINE messageTag #-} -instance Tagged IndexResponse where+instance Tagged RpbIndexResp where messageTag _ = Types.IndexResponse {-# INLINE messageTag #-} -instance Response GetResponse+instance Response RpbGetResp -instance Response IndexResponse+instance Response RpbIndexResp -instance Exchange GetRequest GetResponse+instance Exchange RpbGetReq RpbGetResp -instance Exchange IndexRequest IndexResponse+instance Exchange RpbIndexReq RpbIndexResp -instance Tagged PutRequest where+instance Tagged RpbPutReq where messageTag _ = Types.PutRequest {-# INLINE messageTag #-} -instance Request PutRequest where+instance Request RpbPutReq where expectedResponse _ = Types.PutResponse {-# INLINE expectedResponse #-} -instance Tagged PutResponse where+instance Tagged RpbPutResp where messageTag _ = Types.PutResponse {-# INLINE messageTag #-} -instance Response PutResponse+instance Response RpbPutResp -instance Exchange PutRequest PutResponse+instance Exchange RpbPutReq RpbPutResp -instance Tagged DeleteRequest where+instance Tagged RpbDelReq where messageTag _ = Types.DeleteRequest {-# INLINE messageTag #-} -instance Request DeleteRequest where+instance Request RpbDelReq where expectedResponse _ = Types.DeleteResponse {-# INLINE expectedResponse #-} -instance Tagged ListBucketsRequest where+instance Tagged RpbListBucketsReq where messageTag _ = Types.ListBucketsRequest {-# INLINE messageTag #-} -instance Request ListBucketsRequest where+instance Request RpbListBucketsReq where expectedResponse _ = Types.ListBucketsResponse {-# INLINE expectedResponse #-} -instance Tagged ListBucketsResponse where+instance Tagged RpbListBucketsResp where messageTag _ = Types.ListBucketsResponse {-# INLINE messageTag #-} -instance Response ListBucketsResponse+instance Response RpbListBucketsResp -instance Exchange ListBucketsRequest ListBucketsResponse+instance Exchange RpbListBucketsReq RpbListBucketsResp -instance Tagged ListKeysRequest where+instance Tagged RpbListKeysReq where messageTag _ = Types.ListKeysRequest {-# INLINE messageTag #-} -instance Request ListKeysRequest where+instance Request RpbListKeysReq where expectedResponse _ = Types.ListKeysResponse {-# INLINE expectedResponse #-} -instance Tagged ListKeysResponse where+instance Tagged RpbListKeysResp where messageTag _ = Types.ListKeysResponse {-# INLINE messageTag #-} -instance Response ListKeysResponse+instance Response RpbListKeysResp -instance Tagged GetBucketRequest where+instance Tagged RpbGetBucketReq where messageTag _ = Types.GetBucketRequest {-# INLINE messageTag #-} -instance Request GetBucketRequest where+instance Request RpbGetBucketReq where expectedResponse _ = Types.GetBucketResponse {-# INLINE expectedResponse #-} -instance Tagged GetBucketResponse where+instance Tagged RpbGetBucketResp where messageTag _ = Types.GetBucketResponse {-# INLINE messageTag #-} -instance Response GetBucketResponse+instance Response RpbGetBucketResp -instance Exchange GetBucketRequest GetBucketResponse+instance Exchange RpbGetBucketReq RpbGetBucketResp -instance Tagged SetBucketRequest where+instance Tagged RpbSetBucketReq where messageTag _ = Types.SetBucketRequest {-# INLINE messageTag #-} -instance Request SetBucketRequest where+instance Request RpbSetBucketReq where expectedResponse _ = Types.SetBucketResponse {-# INLINE expectedResponse #-} -instance Tagged MapReduceRequest where+instance Request RpbGetBucketTypeReq where+ expectedResponse _ = Types.GetBucketResponse++instance Tagged RpbGetBucketTypeReq where+ messageTag _ = Types.GetBucketTypeRequest++instance Exchange RpbGetBucketTypeReq RpbGetBucketResp++instance Tagged RpbMapRedReq where messageTag _ = Types.MapReduceRequest {-# INLINE messageTag #-} -instance Request MapReduceRequest where+instance Request RpbMapRedReq where expectedResponse _ = Types.MapReduceResponse {-# INLINE expectedResponse #-} -instance Tagged MapReduce where+instance Tagged RpbMapRedResp where messageTag _ = Types.MapReduceResponse {-# INLINE messageTag #-} -instance Response MapReduce -instance Exchange MapReduceRequest MapReduce+instance Response RpbMapRedResp +instance Exchange RpbMapRedReq RpbMapRedResp++instance Tagged DtFetchReq where+ messageTag _ = Types.DtFetchRequest+ {-# INLINE messageTag #-}++instance Tagged DtFetchResp where+ messageTag _ = Types.DtFetchResponse+ {-# INLINE messageTag #-}++instance Request DtFetchReq where+ expectedResponse _ = Types.DtFetchResponse+ {-# INLINE expectedResponse #-}++instance Response DtFetchResp++instance Exchange DtFetchReq DtFetchResp++instance Tagged DtUpdateReq where+ messageTag _ = Types.DtUpdateRequest+ {-# INLINE messageTag #-}++instance Tagged DtUpdateResp where+ messageTag _ = Types.DtUpdateResponse+ {-# INLINE messageTag #-}++instance Request DtUpdateReq where+ expectedResponse _ = Types.DtUpdateResponse+ {-# INLINE expectedResponse #-}++instance Response DtUpdateResp++instance Exchange DtUpdateReq DtUpdateResp++instance Tagged RpbSearchQueryReq where+ messageTag _ = Types.SearchQueryRequest+ {-# INLINE messageTag #-}++instance Request RpbSearchQueryReq where+ expectedResponse _ = Types.SearchQueryResponse+ {-# INLINE expectedResponse #-}++instance Tagged RpbSearchQueryResp where+ messageTag _ = Types.SearchQueryResponse+ {-# INLINE messageTag #-}++instance Response RpbSearchQueryResp++instance Exchange RpbSearchQueryReq RpbSearchQueryResp++instance Tagged RpbYokozunaIndexGetReq where+ messageTag _ = Types.YokozunaIndexGetRequest++instance Request RpbYokozunaIndexGetReq where+ expectedResponse _ = Types.YokozunaIndexGetResponse++instance Tagged RpbYokozunaIndexGetResp where+ messageTag _ = Types.YokozunaIndexGetResponse++instance Response RpbYokozunaIndexGetResp++instance Exchange RpbYokozunaIndexGetReq RpbYokozunaIndexGetResp++instance Request RpbYokozunaIndexPutReq where+ expectedResponse _ = Types.YokozunaIndexPutRequest++instance Tagged RpbYokozunaIndexPutReq where+ messageTag _ = Types.YokozunaIndexPutRequest++instance Exchange RpbYokozunaIndexPutReq RpbPutResp++instance Tagged RpbYokozunaIndexDeleteReq where+ messageTag _ = Types.YokozunaIndexDeleteRequest++instance Request RpbYokozunaIndexDeleteReq where+ expectedResponse _ = Types.DeleteResponse+ putTag :: MessageTag -> Put-putTag = putWord8 . fromIntegral . fromEnum+putTag m = putWord8 $ message2code HM.! m {-# INLINE putTag #-} getTag :: Get MessageTag getTag = do n <- getWord8- if n > 26- then moduleError "getTag" $ "invalid riak message code: " ++ show n- else return . toEnum . fromIntegral $ n+ maybe (err n) pure $ HM.lookup n code2message+ where+ err n = moduleError "getTag" $ "invalid riak message code: " ++ show n {-# INLINE getTag #-} moduleError :: String -> String -> a moduleError = netError "Network.Riak.Tag"+++code2message :: HM.HashMap Word8 MessageTag+code2message = HM.fromList messageCodes++message2code :: HM.HashMap MessageTag Word8+message2code = HM.fromList . map swap $ messageCodes++messageCodes :: [(Word8, MessageTag)]+messageCodes = [+ -- From riak-2.1.3/deps/riak_pb/src/riak_pb_messages.csv+ --+ -- This is a list of all known riak messages (with appropriate+ -- message codes). Most of them are described at+ -- http://docs.basho.com/riak/2.1.3/dev/references/protocol-buffers/+ --+ -- Commented ones are messages we don't use/support yet.+ (0, Types.ErrorResponse),+ (1, Types.PingRequest),+ (2, Types.PingResponse),+ (3, Types.GetClientIDResponse),+ (4, Types.GetClientIDResponse),+ (5, Types.SetClientIDRequest),+ (6, Types.SetClientIDResponse),+ (7, Types.GetServerInfoRequest),+ (8, Types.GetServerInfoResponse),+ (9, Types.GetRequest),+ (10, Types.GetResponse),+ (11, Types.PutRequest),+ (12, Types.PutResponse),+ (13, Types.DeleteRequest),+ (14, Types.DeleteResponse),+ (15, Types.ListBucketsRequest),+ (16, Types.ListBucketsResponse),+ (17, Types.ListKeysRequest),+ (18, Types.ListKeysResponse),+ (19, Types.GetBucketRequest),+ (20, Types.GetBucketResponse),+ (21, Types.SetBucketRequest),+ (22, Types.SetBucketResponse),+ (23, Types.MapReduceRequest),+ (24, Types.MapReduceResponse),+ (25, Types.IndexRequest),+ (26, Types.IndexResponse),+ (27, Types.SearchQueryRequest),+ (28, Types.SearchQueryResponse),+ -- (29,ResetBucketRequest),+ -- (30,ResetBucketResp),+ (31, Types.GetBucketTypeRequest),+ -- (32,SetBucketTypeReq),+ -- (33,GetBucketKeyPreflistReq),+ -- (34,GetBucketKeyPreflistResp),+ -- (40,CSBucketReq),+ -- (41,CSBucketResp),+ -- (50,CounterUpdateReq),+ -- (51,CounterUpdateResp),+ -- (52,CounterGetReq),+ -- (53,CounterGetResp),+ (54, Types.YokozunaIndexGetRequest),+ (55, Types.YokozunaIndexGetResponse),+ (56, Types.YokozunaIndexPutRequest),+ (57, Types.YokozunaIndexDeleteRequest),+ -- (58,YokozunaSchemaGetReq),+ -- (59,YokozunaSchemaGetResp),+ -- (60,YokozunaSchemaPutReq),+ (80, Types.DtFetchRequest),+ (81, Types.DtFetchResponse),+ (82, Types.DtUpdateRequest),+ (83, Types.DtUpdateResponse)+ -- (253,RpbAuthReq),+ -- (254,RpbAuthResp),+ -- (255,RpbStartTls)+ ]
src/Network/Riak/Types.hs view
@@ -18,6 +18,7 @@ -- * Errors , RiakException(excModule, excFunction, excMessage) -- * Data types+ , BucketType , Bucket , Key , Tag@@ -37,6 +38,7 @@ , Tagged(..) , IndexValue(..) , IndexQuery(..)+ , SearchResult(..) ) where import Network.Riak.Types.Internal
src/Network/Riak/Types/Internal.hs view
@@ -1,5 +1,5 @@-{-# LANGUAGE DeriveDataTypeable, FunctionalDependencies, MultiParamTypeClasses,- RecordWildCards #-}+{-# LANGUAGE BangPatterns, DeriveDataTypeable, FunctionalDependencies,+ MultiParamTypeClasses, RecordWildCards, DeriveGeneric #-} -- | -- Module: Network.Riak.Types.Internal@@ -25,13 +25,21 @@ , unexError -- * Data types , Bucket+ , BucketType , Key , Index+ , Schema , IndexQuery(..) , IndexValue(..) , Tag+ , SearchQuery+ , SearchResult(..)+ , Score+ , IndexInfo , VClock(..) , Job(..)+ , N+ , Timeout -- * Quorum management , Quorum(..) , DW@@ -48,15 +56,19 @@ , Tagged(..) ) where -import Control.Exception (Exception, throw)-import Data.ByteString.Lazy (ByteString)-import Data.Digest.Pure.MD5 (md5)-import Data.IORef (IORef)-import Data.Typeable (Typeable)-import Data.Word (Word32)-import Network.Socket (HostName, ServiceName, Socket)-import Text.ProtocolBuffers (ReflectDescriptor, Wire)+import Control.Exception (Exception, throw)+import Data.ByteString (ByteString)+import qualified Data.ByteString.Lazy as LazyByteString+import Data.Digest.Pure.MD5 (md5)+import Data.Hashable (Hashable)+import Data.IORef (IORef)+import qualified Data.Riak.Proto as Proto+import Data.Typeable (Typeable)+import Data.Word (Word32)+import GHC.Generics (Generic)+import Network.Socket (HostName, ServiceName, Socket) + -- | A client identifier. This is used by the Riak cluster when -- logging vector clock changes, and should be unique for each client. type ClientID = ByteString@@ -129,6 +141,10 @@ -- replicas, for instance). type Bucket = ByteString +-- | Bucket types is a riak >= 2.0 feature allowing groups of buckets+-- to share configuration details+type BucketType = ByteString+ -- | Keys are unique object identifiers in Riak and are scoped within -- buckets. type Key = ByteString@@ -136,6 +152,9 @@ -- | Name of a secondary index type Index = ByteString +-- | Name of an index schema+type Schema = ByteString+ -- | Index query. Can be exact or range, int or bin. Index name should -- not contain the "_bin" or "_int" part, since it's determined from -- data constructor.@@ -159,7 +178,31 @@ | Erlang ByteString deriving (Eq, Show, Typeable) --- | An identifier for an inbound or outbound message.+-- | Search request+type SearchQuery = ByteString++-- | Search result score+type Score = Double++-- | Search index info+type IndexInfo = Proto.RpbYokozunaIndex++-- | N value+--+-- http://docs.basho.com/riak/kv/2.1.4/learn/concepts/replication/+type N = Word32++-- | Timeout in milliseconds+type Timeout = Word32++-- | Solr search result+data SearchResult = SearchResult+ { docs :: ![[(ByteString, Maybe ByteString)]]+ , maxScore :: !(Maybe Float)+ , numFound :: !(Maybe Word32)+ } deriving (Eq, Ord, Show)++-- | List of (known to us) inbound or outbound message identifiers. data MessageTag = ErrorResponse | PingRequest | PingResponse@@ -183,12 +226,25 @@ | GetBucketResponse | SetBucketRequest | SetBucketResponse+ | GetBucketTypeRequest | MapReduceRequest | MapReduceResponse | IndexRequest | IndexResponse- deriving (Eq, Show, Enum, Typeable)+ | DtFetchRequest+ | DtFetchResponse+ | DtUpdateRequest+ | DtUpdateResponse+ | SearchQueryRequest+ | SearchQueryResponse+ | YokozunaIndexGetRequest+ | YokozunaIndexGetResponse+ | YokozunaIndexPutRequest+ | YokozunaIndexDeleteRequest+ deriving (Eq, Show, Generic) +instance Hashable MessageTag+ -- | Messages are tagged. class Tagged msg where messageTag :: msg -> MessageTag -- ^ Retrieve a message's tag.@@ -198,19 +254,14 @@ {-# INLINE messageTag #-} -- | A message representing a request from client to server.-class (Tagged msg, ReflectDescriptor msg, Show msg, Wire msg) => Request msg+class (Proto.Message msg, Show msg, Tagged msg) => Request msg where expectedResponse :: msg -> MessageTag -- | A message representing a response from server to client.-class (Tagged msg, ReflectDescriptor msg, Show msg, Wire msg) => Response msg+class (Proto.Message msg, Show msg, Tagged msg) => Response msg class (Request req, Response resp) => Exchange req resp- | req -> resp, resp -> req--instance (Tagged a, Tagged b) => Tagged (Either a b) where- messageTag (Left l) = messageTag l- messageTag (Right r) = messageTag r- {-# INLINE messageTag #-}+ | req -> resp -- | A wrapper that keeps Riak vector clocks opaque. newtype VClock = VClock {@@ -220,7 +271,7 @@ } deriving (Eq, Typeable) instance Show VClock where- show (VClock s) = "VClock " ++ show (md5 s)+ show (VClock s) = "VClock " ++ show (md5 (LazyByteString.fromStrict s)) -- | A read/write quorum. The quantity of replicas that must respond -- to a read or write request before it is considered successful. This
src/Network/Riak/Value.hs view
@@ -32,36 +32,30 @@ import Control.Applicative import Data.Aeson.Types (Parser, Result(..), parse)-import Data.Foldable (toList) import Data.Monoid ((<>)) import Network.Riak.Connection.Internal-import Network.Riak.Protocol.Content (Content(..))-import Network.Riak.Protocol.GetResponse (GetResponse(..))-import Network.Riak.Protocol.IndexResponse (IndexResponse(..))-import Network.Riak.Protocol.PutResponse (PutResponse(..))+import Network.Riak.Lens import Network.Riak.Resolvable (ResolvableMonoid(..)) import Network.Riak.Response (unescapeLinks) import Network.Riak.Types.Internal hiding (MessageTag(..))-import qualified Network.Riak.Protocol.Pair as Pair-import qualified Data.Aeson.Parser as Aeson+import qualified Data.Riak.Proto as Proto+import qualified Data.Aeson as Aeson (eitherDecodeStrict) import qualified Data.Aeson.Types as Aeson-import qualified Data.Attoparsec.Lazy as A-import qualified Data.ByteString.Lazy as L-import qualified Data.ByteString.Lazy.Char8 as CL8-import qualified Data.Sequence as Seq+import qualified Data.ByteString as L+import qualified Data.ByteString.Char8 as CL8 import qualified Network.Riak.Content as C import qualified Network.Riak.Request as Req -fromContent :: IsContent c => Content -> Maybe c+fromContent :: IsContent c => Proto.RpbContent -> Maybe c fromContent c = case parse parseContent c of Success a -> Just a Error _ -> Nothing class IsContent c where- parseContent :: Content -> Parser c- toContent :: c -> Content+ parseContent :: Proto.RpbContent -> Parser c+ toContent :: c -> Proto.RpbContent -instance IsContent Content where+instance IsContent Proto.RpbContent where parseContent = return {-# INLINE parseContent #-} @@ -77,10 +71,10 @@ {-# INLINE toContent #-} instance IsContent Aeson.Value where- 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+ parseContent c | c ^. Proto.maybe'contentType == Just "application/json" =+ case Aeson.eitherDecodeStrict (c ^. Proto.value) of+ Left err -> fail err+ Right a -> return a | otherwise = fail "non-JSON document" toContent = C.json {-# INLINE toContent #-}@@ -88,36 +82,41 @@ deriving instance (IsContent a) => IsContent (ResolvableMonoid a) -- | Add indexes to a content value for a further put request.-addIndexes :: [IndexValue] -> Content -> Content+addIndexes :: [IndexValue] -> Proto.RpbContent -> Proto.RpbContent addIndexes ix c =- c { C.indexes = Seq.fromList . map toPair $ ix }+ c & Proto.indexes .~ (map toPair ix) where- toPair (IndexInt k v) = Pair.Pair (k <> "_int")- (Just . CL8.pack . show $ v)- toPair (IndexBin k v) = Pair.Pair (k <> "_bin") (Just v)+ toPair :: IndexValue -> Proto.RpbPair+ toPair (IndexInt k v) = Proto.defMessage+ & Proto.key .~ (k <> "_int")+ & Proto.value .~ (CL8.pack . show $ v)+ toPair (IndexBin k v) = Proto.defMessage+ & Proto.key .~ (k <> "_bin")+ & Proto.value .~ v -- | Store a single value. This may return multiple conflicting -- siblings. Choosing among them, and storing a new value, is your -- responsibility. -- -- You should /only/ supply 'Nothing' as a 'T.VClock' if you are sure--- that the given bucket+key combination does not already exist. If--- you omit a 'T.VClock' but the bucket+key /does/ exist, your value--- will not be stored.-put :: (IsContent c) => Connection -> Bucket -> Key -> Maybe VClock -> c+-- that the given type+bucket+key combination does not already exist.+-- If you omit a 'T.VClock' but the type+bucket+key /does/ exist, your+-- value will not be stored.+put :: (IsContent c) => Connection+ -> Maybe BucketType -> Bucket -> Key -> Maybe VClock -> c -> W -> DW -> IO ([c], VClock)-put conn bucket key mvclock val w dw =+put conn btype bucket key mvclock val w dw = putResp =<< exchange conn- (Req.put bucket key mvclock (toContent val) w dw True)+ (Req.put btype bucket key mvclock (toContent val) w dw True) -- | Store an indexed value.-putIndexed :: (IsContent c) => Connection -> Bucket -> Key+putIndexed :: (IsContent c) => Connection -> Maybe BucketType -> Bucket -> Key -> [IndexValue] -> Maybe VClock -> c -> W -> DW -> IO ([c], VClock)-putIndexed conn b k inds mvclock val w dw =+putIndexed conn bt b k inds mvclock val w dw = putResp =<< exchange conn- (Req.put b k mvclock (addIndexes inds (toContent val))+ (Req.put bt b k mvclock (addIndexes inds (toContent val)) w dw True) -- | Store many values. This may return multiple conflicting siblings@@ -125,51 +124,54 @@ -- value in each case, is your responsibility. -- -- You should /only/ supply 'Nothing' as a 'T.VClock' if you are sure--- that the given bucket+key combination does not already exist. If--- you omit a 'T.VClock' but the bucket+key /does/ exist, your value--- will not be stored.-putMany :: (IsContent c) => Connection -> Bucket -> [(Key, Maybe VClock, c)]+-- that the given type+bucket+key combination does not already exist.+-- If you omit a 'T.VClock' but the type+bucket+key /does/ exist, your+-- value will not be stored.+putMany :: (IsContent c) => Connection+ -> Maybe BucketType -> Bucket -> [(Key, Maybe VClock, c)] -> W -> DW -> IO [([c], VClock)]-putMany conn b puts w dw =- mapM putResp =<< pipeline conn (map (\(k,v,c) -> Req.put b k v (toContent c) w dw True) puts)+putMany conn bt b puts w dw =+ mapM putResp =<< pipeline conn (map (\(k,v,c) -> Req.put bt b k v (toContent c) w dw True) puts) -putResp :: (IsContent c) => PutResponse -> IO ([c], VClock)-putResp PutResponse{..} = do- case vclock of+putResp :: (IsContent c) => Proto.RpbPutResp -> IO ([c], VClock)+putResp response = do+ case response ^. Proto.maybe'vclock of Nothing -> return ([], VClock L.empty) Just s -> do- c <- convert content+ c <- convert (response ^. Proto.content) return (c, VClock s) -- | Store a single value, without the possibility of conflict -- resolution. -- -- You should /only/ supply 'Nothing' as a 'T.VClock' if you are sure--- that the given bucket+key combination does not already exist. If--- you omit a 'T.VClock' but the bucket+key /does/ exist, your value--- will not be stored, and you will not be notified.-put_ :: (IsContent c) => Connection -> Bucket -> Key -> Maybe VClock -> c- -> W -> DW -> IO ()-put_ conn bucket key mvclock val w dw =- exchange_ conn (Req.put bucket key mvclock (toContent val) w dw False)+-- that the given type+bucket+key combination does not already exist.+-- If you omit a 'T.VClock' but the type+bucket+key /does/ exist, your+-- value will not be stored, and you will not be notified.+put_ :: (IsContent c) => Connection+ -> Maybe BucketType -> Bucket -> Key -> Maybe VClock -> c+ -> W -> DW -> IO ()+put_ conn btype bucket key mvclock val w dw =+ exchange_ conn (Req.put btype bucket key mvclock (toContent val) w dw False) -- | Store many values, without the possibility of conflict -- resolution. -- -- You should /only/ supply 'Nothing' as a 'T.VClock' if you are sure--- that the given bucket+key combination does not already exist. If--- you omit a 'T.VClock' but the bucket+key /does/ exist, your value--- will not be stored, and you will not be notified.-putMany_ :: (IsContent c) => Connection -> Bucket -> [(Key, Maybe VClock, c)]+-- that the given type+bucket+key combination does not already exist.+-- If you omit a 'T.VClock' but the type+bucket+key /does/ exist, your+-- value will not be stored, and you will not be notified.+putMany_ :: (IsContent c) => Connection+ -> Maybe BucketType -> Bucket -> [(Key, Maybe VClock, c)] -> W -> DW -> IO ()-putMany_ conn b puts w dw =- pipeline_ conn . map (\(k,v,c) -> Req.put b k v (toContent c) w dw False) $ puts+putMany_ conn bt b puts w dw =+ pipeline_ conn . map (\(k,v,c) -> Req.put bt b k v (toContent c) w dw False) $ puts -- | Retrieve a value. This may return multiple conflicting siblings. -- Choosing among them is your responsibility.-get :: (IsContent c) => Connection -> Bucket -> Key -> R+get :: (IsContent c) => Connection -> Maybe BucketType -> Bucket -> Key -> R -> IO (Maybe ([c], VClock))-get conn bucket key r = getResp =<< exchangeMaybe conn (Req.get bucket key r)+get conn btype bucket key r = getResp =<< exchangeMaybe conn (Req.get btype bucket key r) -- | Retrieve list of keys matching some index query. getByIndex :: Connection -> Bucket -> IndexQuery@@ -177,27 +179,31 @@ getByIndex conn b indq = getByIndexResp =<< exchangeMaybe conn (Req.getByIndex b indq) -getMany :: (IsContent c) => Connection -> Bucket -> [Key] -> R+getMany :: (IsContent c) => Connection+ -> Maybe BucketType -> Bucket -> [Key] -> R -> IO [Maybe ([c], VClock)]-getMany conn b ks r =- mapM getResp =<< pipelineMaybe conn (map (\k -> Req.get b k r) ks)+getMany conn bt b ks r =+ mapM getResp =<< pipelineMaybe conn (map (\k -> Req.get bt b k r) ks) -getResp :: (IsContent c) => Maybe GetResponse -> IO (Maybe ([c], VClock))+getResp :: (IsContent c) => Maybe Proto.RpbGetResp -> IO (Maybe ([c], VClock)) getResp resp = case resp of- Just (GetResponse content (Just s) _) -> do- c <- convert content+ Just resp' ->+ case resp' ^. Proto.maybe'vclock of+ Just s -> do+ c <- convert (resp' ^. Proto.content) return $ Just (c, VClock s)- _ -> return Nothing+ _ -> return Nothing+ _ -> return Nothing -getByIndexResp :: Maybe IndexResponse -> IO [Key]+getByIndexResp :: Maybe Proto.RpbIndexResp -> IO [Key] getByIndexResp resp = case resp of- Just (IndexResponse keys _ _ _) -> return (toList keys)+ Just resp' -> return (resp' ^. Proto.keys) Nothing -> return [] -convert :: IsContent v => Seq.Seq Content -> IO [v]-convert = go [] [] . toList+convert :: IsContent v => [Proto.RpbContent] -> IO [v]+convert = go [] [] where go cs vs (x:xs) = case fromContent y of Just v -> go cs (v:vs) xs _ -> go (y:cs) vs xs
src/Network/Riak/Value/Resolvable.hs view
@@ -41,13 +41,14 @@ -- | 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))+ Connection -> Maybe BucketType -> 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+getMany :: (Resolvable a, V.IsContent a) => Connection+ -> Maybe BucketType -> Bucket -> [Key] -> R -> IO [Maybe (a, VClock)] getMany = R.getMany V.getMany {-# INLINE getMany #-}@@ -66,7 +67,7 @@ -- being stuck in a conflict resolution loop, it will throw a -- 'ResolutionFailure' exception. modify :: (Resolvable a, V.IsContent a) =>- Connection -> Bucket -> Key -> R -> W -> DW+ Connection -> Maybe BucketType -> Bucket -> Key -> R -> W -> DW -> (Maybe a -> IO (a,b)) -- ^ Modification function. Called with 'Just' the value if -- the key is present, 'Nothing' otherwise.@@ -86,7 +87,7 @@ -- being stuck in a conflict resolution loop, it will throw a -- 'ResolutionFailure' exception. modify_ :: (Resolvable a, V.IsContent a) =>- Connection -> Bucket -> Key -> R -> W -> DW+ Connection -> Maybe BucketType -> Bucket -> Key -> R -> W -> DW -> (Maybe a -> IO a) -> IO a modify_ = R.modify_ V.get V.put {-# INLINE modify_ #-}@@ -104,7 +105,8 @@ -- conflict resolution loop, it will throw a 'ResolutionFailure' -- exception. put :: (Resolvable a, V.IsContent a) =>- Connection -> Bucket -> Key -> Maybe VClock -> a -> W -> DW+ Connection -> Maybe BucketType -> Bucket -> Key+ -> Maybe VClock -> a -> W -> DW -> IO (a, VClock) put = R.put V.put {-# INLINE put #-}@@ -122,7 +124,8 @@ -- conflict resolution loop, it will throw a 'ResolutionFailure' -- exception. put_ :: (Resolvable a, V.IsContent a) =>- Connection -> Bucket -> Key -> Maybe VClock -> a -> W -> DW+ Connection -> Maybe BucketType -> Bucket -> Key+ -> Maybe VClock -> a -> W -> DW -> IO () put_ = R.put_ V.put {-# INLINE put_ #-}@@ -142,7 +145,7 @@ -- If this function gives up due to apparently being stuck in a loop, -- it will throw a 'ResolutionFailure' exception. putMany :: (Resolvable a, V.IsContent a) =>- Connection -> Bucket -> [(Key, Maybe VClock, a)] -> W -> DW+ Connection -> Maybe BucketType -> Bucket -> [(Key, Maybe VClock, a)] -> W -> DW -> IO [(a, VClock)] putMany = R.putMany V.putMany {-# INLINE putMany #-}@@ -159,6 +162,7 @@ -- If this function gives up due to apparently being stuck in a loop, -- it will throw a 'ResolutionFailure' exception. putMany_ :: (Resolvable a, V.IsContent a) =>- Connection -> Bucket -> [(Key, Maybe VClock, a)] -> W -> DW -> IO ()+ Connection -> Maybe BucketType -> Bucket+ -> [(Key, Maybe VClock, a)] -> W -> DW -> IO () putMany_ = R.putMany_ V.putMany {-# INLINE putMany_ #-}
+ tests/CRDTProperties.hs view
@@ -0,0 +1,316 @@+-- | module: CRDTProperties+-- copyright: (c) 2016 Sentenai+-- author: Antonio Nikishaev <me@lelf.lu>+-- license: Apache+--+{-# LANGUAGE OverloadedStrings, FlexibleContexts, TupleSections, ScopedTypeVariables,+ GADTs, StandaloneDeriving, UndecidableInstances, PatternGuards, MultiParamTypeClasses, CPP #-}+{-# OPTIONS_GHC -fno-warn-orphans #-}+module CRDTProperties (prop_counters,+ prop_sets,+ prop_maps,+ tests) where++-- |+-- The idea: send arbitrary stream of commands to riak, collect each+-- command output to list :: [Maybe RiakReturnValue]. Then see it we+-- get the same list of results simulating Riak in this module.++#if __GLASGOW_HASKELL__ <= 708+import Control.Applicative+#endif+import Control.Monad.RWS+import Control.Monad.Fail+import Data.ByteString (ByteString)+import Data.Default.Class+import Data.List.NonEmpty+import qualified Data.Map as Map+import Data.Maybe+import Data.Proxy+import qualified Data.Set as Set+import Test.QuickCheck+import Test.QuickCheck.Monadic++import Utils+import Test.Tasty+import Test.Tasty.QuickCheck++import qualified Network.Riak.Basic as B+import qualified Network.Riak.CRDT as C++newtype BucketType = BucketType ByteString deriving (Show,Eq,Ord)+newtype Bucket = Bucket ByteString deriving (Show,Eq,Ord)+newtype Key = Key ByteString deriving (Show,Eq,Ord)+newtype Value = Value ByteString deriving Show++class Values a where values :: [a]++-- Not many. We want to hit each multiple times+instance Values BucketType where values = BucketType <$> ["sets","counters","maps"]+instance Values Bucket where values = Bucket <$> ["A","B"]+instance Values Key where values = Key <$> ["a","b","c"]+instance Values Value where values = Value <$> ["1","2","3","4"]++instance Arbitrary BucketType where arbitrary = elements values+instance Arbitrary Bucket where arbitrary = elements values+instance Arbitrary Key where arbitrary = elements values+instance Arbitrary Value where arbitrary = elements values++data Point = Point BucketType Bucket Key deriving (Show,Ord,Eq)+instance Values Point where+ values = [ Point t b k | t <- values, b <- values, k <- values ]++instance Arbitrary Point where arbitrary = elements values++type RiakState = Map.Map Point C.DataType+++-- | observe all current values we care about (instance 'Values') in+-- riak, gather them into a map.+--+-- As it turns out, observeRiak is not quite cheap operation after+-- /types/maps/… are populated. So first argument is proxy for the+-- (only) type we are interested in.+observeRiak :: Action a op => Proxy a -> IO RiakState+observeRiak p = Map.fromList . catMaybes <$> observeRiak' (BucketType $ bucketType p)++observeRiak' :: BucketType -> IO [Maybe (Point, C.DataType)]+observeRiak' bt@(BucketType t_) = withGlobalConn $ \c ->+ sequence [ do r <- C.get c t_ b_ k_+ pure . fmap (p,) $ r+ | b <- values, k <- values,+ let p@(Point _ (Bucket b_) (Key k_)) = Point bt b k+ ]+++-- | We will supply a list of these operations:+--+-- For each Action a op => a,+data Op a op = Get Bucket Key -- ^ we can get a value+ | Update Bucket Key op -- ^ we can update a value++deriving instance (Show op, C.CRDT a o) => Show (Op a op)++class (Show t, C.CRDT t op, Default t) => Action t op where+ -- | bucket type for this type (assumed/hardcoded)+ bucketType :: Proxy t -> ByteString+ -- | extract a value from 'C.DataType', or throw an error+ fromDT :: C.DataType -> t+ -- | pack a value into 'C.DataType'+ toDT :: t -> C.DataType+ -- | a kludge (or two), see 'update': if there's no value at the+ -- moment, having been provieded with an op, will riak create and+ -- operate on a empty value?+ updateCreates :: Proxy t -> Maybe C.DataType -> op -> Bool+ -- | is operation target there?+ targetThere :: Maybe C.DataType -> op -> Bool+++instance Action C.Counter C.CounterOp where+ bucketType _ = "counters"+ fromDT (C.DTCounter c) = c+ fromDT _ = error "expected counter" -- ok for tests+ toDT = C.DTCounter+ updateCreates _ _ _ = True+ targetThere _ _ = True++instance Action C.Set C.SetOp where+ bucketType _ = "sets"+ fromDT (C.DTSet c) = c+ fromDT _ = error "expected set"+ toDT = C.DTSet+ updateCreates _ _ C.SetAdd{} = True+ updateCreates _ _ C.SetRemove{} = False+ targetThere _ _ = True++instance Action C.Map C.MapOp where+ bucketType _ = "maps"+ fromDT (C.DTMap c) = c+ fromDT _ = error "expected map"+ toDT = C.DTMap+ -- Riak will operate on default value inside map if it's not there.+ -- Except while it's SetRemove.+ updateCreates _ _ (C.MapUpdate _ (C.MapSetOp C.SetRemove{})) = False+ updateCreates _ _ _ = True++ targetThere (Just (C.DTMap m)) (C.MapUpdate path _)+ | Just{} <- C.xlookup path C.MapSetTag m = True+ targetThere _ _ = False++-- | Many Arbitrary instances.+-- TODO: 'Generic' arbitrary++instance (Arbitrary a, Arbitrary op) => Arbitrary (Op a op) where+ arbitrary = oneof [+ Get <$> arbitrary <*> arbitrary,+ Update <$> arbitrary <*> arbitrary <*> arbitrary+ ]++instance Arbitrary C.Counter where+ arbitrary = C.Counter <$> arbitrary++instance Arbitrary C.CounterOp where+ arbitrary = C.CounterInc <$> choose (-16,16) -- https://github.com/basho/riak/issues/804++instance Arbitrary C.SetOp where+ arbitrary = oneof [+ C.SetAdd <$> arbitrary, C.SetRemove <$> arbitrary+ ]++instance Arbitrary C.FlagOp where+ arbitrary = C.FlagSet <$> arbitrary++instance Arbitrary C.Set where+ arbitrary = C.Set . Set.fromList <$> arbitrary++instance Arbitrary ByteString where+ arbitrary = elements [ "foo", "bar", "baz" ]++instance Arbitrary C.MapOp where+ arbitrary = C.MapUpdate <$> arbitrary <*> arbitrary++instance Arbitrary C.MapPath where+ arbitrary = (\a b -> C.MapPath (a :| b)) <$> arbitrary <*> arbitrary++instance Arbitrary C.MapField where+ arbitrary = C.MapField <$> arbitrary <*> arbitrary++instance Arbitrary C.MapEntryTag where+ arbitrary = elements [ C.MapCounterTag ]++instance Arbitrary C.MapValueOp where+ arbitrary = oneof [ C.MapCounterOp <$> arbitrary,+ C.MapSetOp <$> arbitrary,+ C.MapFlagOp <$> arbitrary ]++-- TODO Fix this+instance Arbitrary C.Map where+ arbitrary = pure (C.Map Map.empty)++-- | Abstract machine.+-- Yields a value on 'Get', modifies state on 'Update'.+machine :: (MonadFail m,+ MonadWriter [Maybe C.DataType] m,+ MonadState s m,+ Applicative m,+ Action t op)+ => Proxy t -> [Op t op]+ -> (Op t op -> s -> m (Either (Maybe C.DataType) s)) -> m ()++machine _ [] _ = pure ()++machine p (a@Get{} : as) onAct = do+ v <- get+ Left r <- onAct a v+ tell [r]+ machine p as onAct++machine p (a@Update{} : as) onAct = do+ v <- get+ Right r <- onAct a v+ put r+ machine p as onAct++++-- | Riak version of the 'machine'.+-- State is 'B.Connection', get/update are IO-requests to riak.+riak :: (MonadFail m,+ MonadWriter [Maybe C.DataType] m,+ MonadState B.Connection m,+ Applicative m, MonadIO m,+ Action t op)+ => Proxy t -> [Op t op] -> m ()++riak p ops = machine p ops onAct+ where onAct (Get (Bucket b) (Key k)) c+ = liftIO $ Left <$> C.get c bt b k+ onAct (Update (Bucket b) (Key k) op) c+ = do liftIO $ C.sendModify c bt b k [op]+ pure $ Right c+ bt = bucketType p++++-- | Haskell emulation version of the 'machine'.+-- State is 'RiakState', get/update try to match riak's behaviour.+pure_ :: (MonadFail m,+ MonadWriter [Maybe C.DataType] m,+ MonadState RiakState m,+ Applicative m,+ Action t op)+ => Proxy t -> [Op t op] -> m ()++pure_ p ops = machine p ops onAct+ where+ onAct (Get b k) v = pure . Left $ Map.lookup (point b k) v+ onAct (Update b k op) v = pure . Right $ Map.alter (update op) (point b k) v+ point b k = Point (BucketType (bucketType p)) b k++++update :: forall a op. (Action a op) =>+ op -> Maybe C.DataType -> Maybe C.DataType+update op dt+ -- | Dear diary, this is getting out of hand. It'd be far easier to+ -- not bother and assume Nothing ≡ Just mempty in test conditions.+ --+ -- …but let's continue to bother and gain more safety.+ | Just d <- dt, targetThere dt op+ = operate d+ | updateCreates (Proxy :: Proxy a) dt op+ -- it's sometimes ok to update even a non-set value in riak's mind+ = operate $ maybe surrogate id dt+ | otherwise+ = dt+ where fromDT' :: C.DataType -> a+ fromDT' = fromDT+ toDT' :: a -> C.DataType+ toDT' = toDT+ surrogate = toDT' def+ operate = Just . toDT . C.modify op . fromDT'++++doRiak :: Action a op =>+ Proxy a -> [Op a op] -> IO [Maybe C.DataType]+doRiak p ops = withGlobalConn $ \conn -> do+ --print ops+ (_,_,r) <- runRWST (riak p ops) () conn+ pure r++doPure :: (Action a op, Show op) =>+ RiakState -> Proxy a -> [Op a op] -> PropertyM IO [Maybe C.DataType]+doPure stat p ops = do (_,_,r) <- runRWST (pure_ p ops) () stat+ pure r+++++prop :: (Show op, Action a op) =>+ Proxy a -> [Op a op] -> Property+prop p ops = monadicIO $ do+ stat <- run $ observeRiak p+ r1 <- doPure stat p ops+ r2 <- run $ doRiak p ops+ run . when (r1/=r2) $ print (r1,r2)+ assert $ r1 == r2+++prop_counters :: [Op C.Counter C.CounterOp] -> Property+prop_counters = prop (Proxy :: Proxy C.Counter)++prop_sets :: [Op C.Set C.SetOp] -> Property+prop_sets = prop (Proxy :: Proxy C.Set)++prop_maps :: [Op C.Map C.MapOp] -> Property+prop_maps = prop (Proxy :: Proxy C.Map)++++tests :: TestTree+tests = testGroup "CRDT quickCheck" [+ testProperty "counters" prop_counters,+ testProperty "sets" prop_sets,+ testProperty "maps" prop_maps+ ]
+ tests/Internal.hs view
@@ -0,0 +1,32 @@+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE CPP #-}+-- | GHC stage restriction: have to write Lift in a separate module from where+-- TH is used (Utils.hs)++module Internal where++#if __GLASGOW_HASKELL__ < 710+import Control.Applicative+#endif+import Data.Aeson (FromJSON(parseJSON), genericParseJSON)+import Data.Aeson.Types (defaultOptions, fieldLabelModifier)+import Data.Char (toLower)+import GHC.Generics (Generic)+import Language.Haskell.TH.Syntax (Lift(lift))++data Config = Config+ { configHost :: String+ , configHttpPort :: Int+ , configProtoPort :: Int+ , configAdmin :: String+ } deriving Generic++instance Lift Config where+ lift (Config a b c d) = [| Config a b c d |]++instance FromJSON Config where+ parseJSON = genericParseJSON (defaultOptions { fieldLabelModifier = f })+ where+ f ('c':'o':'n':'f':'i':'g':x:xs) = toLower x : xs+ f _ = fail "invalid config"
tests/Properties.hs view
@@ -8,40 +8,55 @@ #if __GLASGOW_HASKELL__ < 710 import Control.Applicative ((<$>)) #endif-import qualified Data.ByteString.Lazy as L-import Data.IORef (IORef, modifyIORef, newIORef)+import Data.ByteString (ByteString)+import qualified Data.ByteString as BS+import Data.Maybe import qualified Network.Riak.Basic as B-import Network.Riak.Connection (defaultClient)-import Network.Riak.Connection.Pool (Pool, create, withConnection) import Network.Riak.Content (binary)-import Network.Riak.Types (Bucket, Key, Quorum (..))-import System.IO.Unsafe (unsafePerformIO)+import Network.Riak.Types as Riak import Test.QuickCheck.Monadic (assert, monadicIO, run) import Test.Tasty import Test.Tasty.QuickCheck+import Utils -instance Arbitrary L.ByteString where- arbitrary = L.pack `fmap` arbitrary+instance Arbitrary ByteString where+ arbitrary = BS.pack `fmap` arbitrary -cruft :: IORef [(Bucket, Key)]-{-# NOINLINE cruft #-}-cruft = unsafePerformIO $ newIORef []+newtype QCBucket = QCBucket Riak.Bucket deriving Show -pool :: Pool-{-# NOINLINE pool #-}-pool = unsafePerformIO $- create defaultClient 1 1 1+instance Arbitrary QCBucket where+ arbitrary = QCBucket <$> arbitrary `suchThat` (not . BS.null) -t_put_get :: Bucket -> Key -> L.ByteString -> Property-t_put_get b k v =- notempty b && notempty k ==> monadicIO $ assert . uncurry (==) =<< run act+newtype QCKey = QCKey Riak.Key deriving Show++instance Arbitrary QCKey where+ arbitrary = QCKey <$> arbitrary `suchThat` (not . BS.null)+++t_put_get :: QCBucket -> QCKey -> ByteString -> Property+t_put_get (QCBucket b) (QCKey k) v =+ monadicIO $ assert . uncurry (==) =<< run act where- act = withConnection pool $ \c -> do- modifyIORef cruft ((b,k):)- p <- Just <$> B.put c b k Nothing (binary v) Default Default- r <- B.get c b k Default+ act = withGlobalConn $ \c -> do+ p <- Just <$> B.put c btype b k Nothing (binary v) Default Default+ r <- B.get c btype b k Default return (p,r)- notempty = not . L.null+ btype = Nothing +put_delete_get :: QCBucket -> QCKey -> ByteString -> Property+put_delete_get (QCBucket b) (QCKey k) v+ = monadicIO $ do+ r <- run act+ assert $ isNothing r+ where+ act = withGlobalConn $ \c -> do+ _ <- B.put c bt b k Nothing (binary v) Default Default+ B.delete c bt b k Default+ B.get c bt b k Default+ bt = Nothing :: Maybe BucketType+ tests :: [TestTree]-tests = [ testProperty "t_put_get" t_put_get ]+tests = [+ testProperty "t_put_get" t_put_get,+ testProperty "put_delete_get" put_delete_get+ ]
tests/Test.hs view
@@ -1,59 +1,299 @@-{-# LANGUAGE CPP #-}-{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE CPP #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ParallelListComp #-}+{-# LANGUAGE ScopedTypeVariables #-} module Main where -import Control.Exception (finally)-import Control.Monad (forM_)-import Data.IORef-import qualified Data.Map as M-import Data.Text (Text)-import qualified Network.Riak as Riak-import qualified Network.Riak.Basic as B-import qualified Network.Riak.Cluster as Riak-import Network.Riak.Connection.Pool (withConnection)-import qualified Network.Riak.JSON as J-import Network.Riak.Resolvable (ResolvableMonoid (..))+import Control.Monad+#if __GLASGOW_HASKELL__ <= 708+import Control.Applicative+#endif+import qualified Data.Map as M+import qualified Data.Riak.Proto as Proto+import qualified Data.Set as S+import Data.List.NonEmpty (NonEmpty(..))+#if __GLASGOW_HASKELL__ < 804+import Data.Semigroup+#endif+import Data.Text (Text)+import Data.Function ((&))+import Control.Concurrent (threadDelay)+import Control.Exception+import Lens.Micro ((^.), (.~))+import qualified Network.Riak as Riak+import Network.Riak.Admin.DSL+import qualified Network.Riak.Basic as B+import qualified Network.Riak.Content as B (binary,RpbContent)+import Network.Riak.Connection (exchange)+import qualified Network.Riak.CRDT as C+import qualified Network.Riak.CRDT.Riak as C+import qualified Network.Riak.Request as Req+import qualified Network.Riak.Response as Resp+import qualified Network.Riak.Search as S+import qualified Network.Riak.JSON as J+import Network.Riak.Resolvable (ResolvableMonoid (..)) import Network.Riak.Types import qualified Properties+import qualified CRDTProperties as CRDT+import Utils import Test.Tasty import Test.Tasty.HUnit main :: IO ()-main = defaultMain tests `finally` cleanup- where- cleanup = withConnection Properties.pool $ \c -> do- bks <- readIORef Properties.cruft- forM_ bks $ \(b,k) -> B.delete c b k Default+main = do+ setup+ defaultMain tests -tests :: TestTree-tests = testGroup "Tests" [properties, integrationalTests]+setup :: IO ()+setup = do+ -- Create a "set-ix" index and wait for it to exist.+ let createIxUrl :: String+ createIxUrl = globalHost ++ ":" ++ show globalHttpPort ++ "/search/index/set-ix" + shell ("curl -sf -XPUT " ++ createIxUrl ++ " -H 'Content-Type: application/json'")+ let loop =+ try (shell ("curl -sf " ++ createIxUrl)) >>= \case+ Left (_ :: ShellFailure) -> do+ threadDelay (1*1000*1000)+ loop+ Right _ -> pure ()+ loop++ riakAdminWith globalAdmin+ [ waitForService "riak_kv" Nothing+ , waitForService "yokozuna" Nothing++ , bucketTypeCreate "maps" (Just "'{\"props\":{\"datatype\":\"map\"}}'")+ , bucketTypeCreate "sets" (Just "'{\"props\":{\"datatype\":\"set\",\"search_index\":\"set-ix\"}}'")+ , bucketTypeCreate "counters" (Just "'{\"props\":{\"datatype\":\"counter\"}}'")++ , bucketTypeActivate "maps"+ , bucketTypeActivate "sets"+ , bucketTypeActivate "counters"++ , bucketTypeCreate "untyped-1" Nothing+ , bucketTypeCreate "untyped-2" Nothing++ , bucketTypeActivate "untyped-1"+ , bucketTypeActivate "untyped-2"+ ]++tests :: TestTree+tests = testGroup "Tests" [properties,+ integrationalTests,+ ping'o'death,+ exceptions,+ crdts,+ searches,+ bucketTypes+ ] properties :: TestTree-properties = testGroup "Properties" Properties.tests+properties = testGroup "simple properties" Properties.tests integrationalTests :: TestTree-integrationalTests = testGroup "Integrational tests"+integrationalTests = testGroup "integrational tests" [ testClusterSimple #ifdef TEST2I , testIndexedPutGet #endif ] +crdts :: TestTree+crdts = testGroup "CRDT" [+ testGroup "simple" [counter, set, map_],+ CRDT.tests+ ]++searches :: TestTree+searches = testGroup "Search" [+ search1,+ search2,+ getIndex,+ putIndex,+ deleteIndex+ ]+ testClusterSimple :: TestTree-testClusterSimple = testCase "testClusterSimple" $ do- rc <- Riak.connectToCluster [Riak.defaultClient]- Riak.inCluster rc B.ping+testClusterSimple = testCase "testClusterSimple" $ withGlobalConn B.ping testIndexedPutGet :: TestTree testIndexedPutGet = testCase "testIndexedPutGet" $ do- rc <- Riak.connectToCluster [Riak.defaultClient]- let b = "riak-haskell-client-test"+ let bt = Nothing+ b = "riak-haskell-client-test" k = "test"- keys <- Riak.inCluster rc $ \c -> do- _ <- J.putIndexed c b k [(IndexInt "someindex" 135)] Nothing+ keys <- withGlobalConn $ \c -> do+ _ <- J.putIndexed c bt b k [(IndexInt "someindex" 135)] Nothing (RM (M.fromList [("somekey", "someval")] :: M.Map Text Text)) Default Default Riak.getByIndex c b (IndexQueryExactInt "someindex" 135) assertEqual "" ["test"] keys++ping'o'death :: TestTree+ping'o'death = testCase "ping'o'death" $ replicateM_ 23 ping+ where ping = withGlobalConn (\c -> replicateM_ 1024 (Riak.ping c))+++counter :: TestTree+counter = testCase "increment" $ withGlobalConn $ \conn -> do+ Just (C.DTCounter (C.Counter a)) <- act conn+ Just (C.DTCounter (C.Counter b)) <- act conn+ assertEqual "inc by 1" 1 (b-a)+ where+ act c = do C.counterSendUpdate c "counters" "xxx" "yyy" [C.CounterInc 1]+ C.get c "counters" "xxx" "yyy"++set :: TestTree+set = testCase "set add" $ withGlobalConn $ \conn -> do+ C.setSendUpdate conn btype buck key [C.SetRemove val]+ C.setSendUpdate conn btype buck key [C.SetAdd val]+ Just (C.DTSet (C.Set r)) <- C.get conn btype buck key+ assertBool "-foo +foo => contains foo" $ val `S.member` r+ where+ (btype,buck,key,val) = ("sets","xxx","yyy","foo")++map_ :: TestTree+map_ = testCase "map update" $ withGlobalConn $ \conn -> do+ Just (C.DTMap a) <- act conn -- do smth (increment), get+ Just (C.DTMap b) <- act conn -- increment, get+ assertEqual "map update" (C.modify mapOp a) b -- modify's behaviour should match+ assertEqual "mapUpdate sugar" mapOp' mapOp++ -- remove top-level field (X)+ C.mapSendUpdate conn btype buck key [C.MapRemove fieldX]+ Just (C.DTMap c) <- act conn+ assertEqual "update after delete" (Just updateCreates) $ C.xlookup "X" C.MapMapTag c++ -- remove nested field (X/Y)+ C.mapSendUpdate conn btype buck key [C.MapUpdate (C.MapPath $ "X" :| [])+ (C.MapMapOp (C.MapRemove fieldY))]+ Just (C.DTMap d) <- C.get conn btype buck key+ assertEqual "update after nested delete 1" Nothing+ $ C.xlookup ("X" C.-/ "Y") C.MapMapTag d+ assertEqual "update after nested delete 2" (Just (C.MapMap (C.Map mempty)))+ $ C.xlookup "X" C.MapMapTag d+ where+ act c = do C.mapSendUpdate c btype buck key [mapOp]+ C.get c btype buck key++ btype = "maps"+ fieldX = C.MapField C.MapMapTag "X"+ fieldY = C.MapField C.MapMapTag "Y"+ (buck,key) = ("xxx","yyy")++ updateCreates = C.MapMap (C.Map (M.fromList+ [(C.MapField C.MapMapTag "Y",+ C.MapMap (C.Map (M.fromList+ [(C.MapField C.MapCounterTag "Z",+ C.MapCounter (C.Counter 1))])))]))++ mapOp = "X" C.-/ "Y" C.-/ "Z" `C.mapUpdate` C.CounterInc 1++ mapOp' = C.MapUpdate (C.MapPath ("X" :| "Y" : "Z" : []))+ (C.MapCounterOp (C.CounterInc 1))+++search1 :: TestTree+search1 = testCase "basic searchRaw" $ withGlobalConn $ \conn -> do+ C.sendModify conn btype buck key [C.SetRemove kw]+ delay+ a <- query conn ("set:" <> kw)+ assertEqual "should not found non-existing" (S.SearchResult [] (Just 0.0) (Just 0)) a+ C.sendModify conn btype buck key [C.SetAdd kw]+ delay+ b <- query conn ("set:" <> kw)+ assertBool "searches specific" $ not (null (S.docs b))+ c <- query conn ("set:*")+ assertBool "searches *" $ not (null (S.docs c))+ where+ query conn q = S.searchRaw conn q "set-ix"+ (btype,buck,key) = ("sets","xxx","yyy")+ kw = "haskell"+ delay = threadDelay (1*5000*1000) -- http://docs.basho.com/riak/2.1.3/dev/using/search/#Indexing-Values++search2 :: TestTree+search2 = testCase "search with fl" $ withGlobalConn $ \conn -> do+ let req = (Req.search "set:haskell" "set-ix") & Proto.fl .~ ["_yz_rk"]+ resp <- Resp.search <$> exchange conn req+ assertEqual "only returns fl"+ ([[("_yz_rk", Just "yyy")]])+ (S.docs resp)+++getIndex :: TestTree+getIndex = testCase "getIndex" $ withGlobalConn $ \conn -> do+ all' <- S.getIndex conn Nothing+ one <- S.getIndex conn (Just "set-ix")+ assertBool "all indeces" $ not (null all')+ assertEqual "set index" 1 (length one)++putIndex :: TestTree+putIndex = testCase "putIndex" $ withGlobalConn $ \conn -> do+ _ <- S.putIndex conn (S.indexInfo "dummy-index" Nothing Nothing) Nothing+ threadDelay 5000000+ one <- S.getIndex conn (Just "dummy-index")+ assertEqual "index was created" 1 (length one)++deleteIndex :: TestTree+deleteIndex = testCase "deleteIndex" $ withGlobalConn $ \conn -> do+ S.deleteIndex conn "dummy-index"+ threadDelay (5*1000*1000)++ _ <- tryJust f (S.getIndex conn (Just "dummy-index"))+ pure ()++ where+ f :: Proto.RpbErrorResp -> Maybe ()+ f resp = guard (resp ^. Proto.errmsg == "notfound")++bucketTypes :: TestTree+bucketTypes = testCase "bucketTypes" $ withGlobalConn $ \conn -> do+ [p0,p1,p2] <- sequence [ B.put conn bt b k Nothing o Default Default+ | bt <- types | o <- [o0,o1,o2] ]+ [r0,r1,r2] <- sequence [ B.get conn bt b k Default | bt <- types ]++ assertBool "sound get Nothing" (valok r0 o0)+ assertBool "sound get untyped-1" (valok r1 o1)+ assertBool "sound get untyped-2" (valok r2 o2)++ assertEqual "put=get Nothing" (Just p0) r0+ assertEqual "put=get untyped-1" (Just p1) r1+ assertEqual "put=get untyped-2" (Just p2) r2+ where+ (b,k) = ("xxx","0") :: (Bucket,Key)+ types = [Nothing, Just "untyped-1", Just "untyped-2"] :: [Maybe BucketType]+ [o0,o1,o2] = B.binary <$> ["A","B","C"] :: [B.RpbContent]++ valok :: Maybe ([B.RpbContent], VClock) -> B.RpbContent -> Bool+ valok (Just (rs,_)) o = (o ^. Proto.value) `elem` map (^. Proto.value) rs+ valok _ _ = False+++exceptions :: TestTree+exceptions = testGroup "exceptions" [+ testCase "correct put" . shouldBeOK . withGlobalConn $ put,+ testCase "correct put_" . shouldBeOK . withGlobalConn $ put_,+ testCase "invalid put" . shouldThrow . withGlobalConn $ putErr,+ testCase "invalid put_" . shouldThrow . withGlobalConn $ 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.RpbContent -> 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::Proto.RpbErrorResp) -> pure ())++ btype = Just "untyped-1"+ noBtype = Just "no such type"+ buck = "xxx"+ key = "0"+ val = B.binary ""
+ tests/Utils.hs view
@@ -0,0 +1,83 @@+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE CPP #-}+module Utils+ ( globalAdmin+ , globalHost+ , globalHttpPort+ , shell+ , withGlobalConn+ , ShellFailure(..)+ ) where++#if __GLASGOW_HASKELL__ < 710+import Control.Applicative+#endif+import Control.Exception+import Control.Monad+import Data.Typeable+import Data.Yaml.TH (decodeFile)+import Internal+import System.Exit+import System.IO.Unsafe (unsafePerformIO)+import System.Timeout++import qualified Network.Riak as Riak+import qualified Network.Riak.Basic as B+import Network.Riak.Connection.Pool (Pool, create, withConnection)++import qualified System.Process as Process++config :: Config+config = $$(decodeFile "tests/test.yaml")++-- | The global riak-admin string, configured in test.yaml.+globalAdmin :: String+globalAdmin = configAdmin config++globalHost :: String+globalHost = configHost config++globalHttpPort :: Int+globalHttpPort = configHttpPort config++-- | Run action in some Riak connection+withGlobalConn :: (B.Connection -> IO a) -> IO a+withGlobalConn = withConnection pool++-- | The global riak pool that all tests share.+pool :: Pool+pool = unsafePerformIO (create client 1 1 1)+ where+ client = Riak.defaultClient+ { Riak.host = globalHost+ , Riak.port = show (configProtoPort config)+ }+{-# NOINLINE pool #-}++data ShellFailure+ = ShellFailure Int String+ | ShellTimeout String+ deriving (Show, Typeable)++instance Exception ShellFailure++-- | Run a shell command (inheriting stdin, stdout, and stderr), and throw an+-- exception if it fails. Time out after 30 seconds.+shell :: String -> IO ()+shell s =+ timeout (30*1000*1000) act >>= \case+ Nothing -> throw (ShellTimeout s)+ _ -> pure ()+ where+ act :: IO ()+ act =+ bracketOnError+ (do+ (_, _, _, h) <- Process.createProcess (Process.shell s)+ pure h)+ Process.terminateProcess+ (Process.waitForProcess >=> \case+ ExitSuccess -> pure ()+ ExitFailure n -> throw (ShellFailure n s))
+ tests/dsl/Network/Riak/Admin/DSL.hs view
@@ -0,0 +1,56 @@+-- | A simple "riak-admin DSL", which is nothing more than strings of stitched+-- together riak-admin shell commands.++module Network.Riak.Admin.DSL+ ( -- * Riak admin commands+ waitForService+ , bucketTypeCreate+ , bucketTypeActivate+ -- * Riak admin runners+ , riakAdmin+ , riakAdminWith+ ) where++import Utils+++-- | Bucket props string, e.g. "{\"props\":{\"n_val\":\"1\"}}"+type BucketProps = String++type BucketTypeName = String++-- | Erlang node name, e.g. "riak@172.17.0.2"+type NodeName = String++-- | riak-admin shell command+type RiakAdmin = String++-- | Riak service name, e.g. "yokozuna"+type ServiceName = String+++-- | riak-admin wait-for-service+waitForService :: ServiceName -> Maybe NodeName -> RiakAdmin+waitForService name node =+ "riak-admin wait-for-service " ++ name ++ maybe "" (" " ++) node++-- | riak-admin bucket-type create+bucketTypeCreate :: BucketTypeName -> Maybe BucketProps -> RiakAdmin+bucketTypeCreate name props =+ "riak-admin bucket-type create " ++ name ++ maybe "" (" " ++) props++-- | riak-admin bucket-type activate+bucketTypeActivate :: BucketTypeName -> RiakAdmin+bucketTypeActivate name = "riak-admin bucket-type activate " ++ name+++-- | Run a list of riak-admin commands locally, and throw an exception if any of+-- them fail.+riakAdmin :: [RiakAdmin] -> IO ()+riakAdmin = riakAdminWith ""++-- | Like 'riakAdmin', but prefix each command with the given 'String' (plus a+-- space).+riakAdminWith :: String -> [RiakAdmin] -> IO ()+riakAdminWith "" = mapM_ shell+riakAdminWith prefix = mapM_ (\s -> shell (prefix ++ " " ++ s))
+ tests/test.yaml view
@@ -0,0 +1,10 @@+# Riak host+host: localhost++# Riak ports+httpPort: 8098+protoPort: 8087++# String to prefix shell commands that exec riak-admin commands, e.g.+# "docker exec riak01"+admin: "sudo"