diff --git a/Changes.md b/Changes.md
new file mode 100644
--- /dev/null
+++ b/Changes.md
@@ -0,0 +1,8 @@
+* 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
diff --git a/benchmarks/Main.hs b/benchmarks/Main.hs
new file mode 100644
--- /dev/null
+++ b/benchmarks/Main.hs
@@ -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.Lazy.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
+
diff --git a/riak.cabal b/riak.cabal
--- a/riak.cabal
+++ b/riak.cabal
@@ -1,5 +1,5 @@
name: riak
-version: 0.9.1.1
+version: 1.0.0.0
synopsis: A Haskell client for the Riak decentralized data store
description:
A Haskell client library for the Riak decentralized data
@@ -26,8 +26,11 @@
[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
license: OtherLicense
license-file: LICENSE
@@ -38,7 +41,8 @@
category: Network
build-type: Simple
extra-source-files:
- README.markdown
+ README.markdown
+ Changes.md
cabal-version: >=1.8
@@ -61,14 +65,20 @@
library
hs-source-dirs: src
-
exposed-modules:
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
@@ -77,16 +87,16 @@
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
build-depends:
aeson >= 0.8 && < 0.11,
@@ -96,27 +106,33 @@
blaze-builder >= 0.3 && <= 0.5,
bytestring,
containers,
+ data-default-class >= 0.0.1,
+ deepseq >= 1.3,
enclosed-exceptions >= 1.0.1.1 && <= 1.1,
exceptions >= 0.8.0.2 && <= 0.9,
+ hashable >= 1.2.3,
transformers >= 0.3 && < 0.5,
mersenne-random-pure64 >= 0.2.0.4 && < 0.3,
monad-control >= 1.0.0.4 && < 1.1,
network >= 2.3,
resource-pool == 0.2.*,
- protocol-buffers >= 2.1.4 && < 2.2,
+ protocol-buffers >= 2.1.4 && < 2.3,
pureMD5,
random,
random-shuffle >= 0.0.4 && < 0.1,
- riak-protobuf == 0.20.*,
+ riak-protobuf == 0.21.*,
+ semigroups >= 0.16,
text == 1.2.*,
time >= 1.4.2 && < 1.6,
- vector >= 0.10.12.3 && < 0.12
+ vector >= 0.10.12.3 && < 0.12,
+ unordered-containers >= 0.2.5
+
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
@@ -133,6 +149,8 @@
other-modules:
Properties
+ CRDTProperties
+ Common
build-depends:
base,
@@ -144,4 +162,22 @@
tasty,
tasty-hunit,
tasty-quickcheck,
- text
+ text,
+ mtl >= 2.1,
+ semigroups >= 0.16,
+ data-default-class >= 0.0.1
+
+
+benchmark bench
+ 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
+
+
+
diff --git a/src/Network/Riak.hs b/src/Network/Riak.hs
--- a/src/Network/Riak.hs
+++ b/src/Network/Riak.hs
@@ -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
--
module Network.Riak
diff --git a/src/Network/Riak/Basic.hs b/src/Network/Riak/Basic.hs
--- a/src/Network/Riak/Basic.hs
+++ b/src/Network/Riak/Basic.hs
@@ -39,6 +39,7 @@
, foldKeys
, getBucket
, setBucket
+ , getBucketType
-- * Map/reduce
, mapReduce
) where
@@ -114,15 +115,16 @@
-- 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 (Seq.Seq T.Bucket)
+listBuckets conn btype = Resp.listBuckets <$> exchange conn (Req.listBuckets btype)
-- 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
@@ -133,12 +135,16 @@
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 BucketProps
+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 -> BucketProps -> 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 BucketProps
+getBucketType conn btype = Resp.getBucket <$> exchange conn (Req.getBucketType btype)
-- | Run a 'MapReduce' job. Its result is consumed via a strict left
-- fold.
diff --git a/src/Network/Riak/CRDT.hs b/src/Network/Riak/CRDT.hs
new file mode 100644
--- /dev/null
+++ b/src/Network/Riak/CRDT.hs
@@ -0,0 +1,172 @@
+{- | module: Network.Riak.CRDT
+ copyright: (c) 2016 Sentenai
+ author: Antonio Nikishaev
+ 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 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))
+
+-}
+{-# LANGUAGE TypeFamilies, OverloadedStrings, ScopedTypeVariables, PatternGuards #-}
+{-# LANGUAGE MultiParamTypeClasses, FunctionalDependencies #-}
+
+module Network.Riak.CRDT (module Network.Riak.CRDT.Types,
+ get,
+ CRDT(..))
+ where
+
+
+import Data.Default.Class
+import qualified Data.Map as M
+import Data.Proxy
+import Data.Semigroup
+import qualified Data.Set as S
+import Network.Riak.CRDT.Ops
+import Network.Riak.CRDT.Riak
+import Network.Riak.CRDT.Types
+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
diff --git a/src/Network/Riak/CRDT/Ops.hs b/src/Network/Riak/CRDT/Ops.hs
new file mode 100644
--- /dev/null
+++ b/src/Network/Riak/CRDT/Ops.hs
@@ -0,0 +1,118 @@
+-- | module: Network.Riak.CRDT.Ops
+-- copyright: (c) 2016 Sentenai
+-- author: Antonio Nikishaev
+-- license: Apache
+--
+-- Conversions of CRDT operations to 'PB.DtOp'
+--
+module Network.Riak.CRDT.Ops (counterUpdateOp,
+ setUpdateOp, SetOpsComb(..), toOpsComb,
+ mapUpdateOp)
+ where
+
+import qualified Network.Riak.Protocol.DtOp as PB
+import qualified Network.Riak.Protocol.CounterOp as PB
+import qualified Network.Riak.Protocol.SetOp as PBSet
+
+import qualified Network.Riak.Protocol.MapOp as PBMap
+import qualified Network.Riak.Protocol.MapField as PBMap
+import qualified Network.Riak.Protocol.MapField.MapFieldType as PBMap
+import qualified Network.Riak.Protocol.MapUpdate as PBMap
+
+import qualified Network.Riak.Protocol.MapUpdate.FlagOp as PBFlag
+
+import Data.ByteString.Lazy (ByteString)
+import Data.Monoid
+import qualified Data.Sequence as Seq
+import qualified Data.Set as S
+import Network.Riak.CRDT.Types
+
+
+counterUpdateOp :: [CounterOp] -> PB.DtOp
+counterUpdateOp ops = PB.DtOp { PB.counter_op = Just $ counterOpPB ops,
+ PB.set_op = Nothing,
+ PB.map_op = Nothing
+ }
+
+counterOpPB :: [CounterOp] -> PB.CounterOp
+counterOpPB ops = PB.CounterOp (Just i)
+ where CounterInc i = mconcat ops
+
+
+data SetOpsComb = SetOpsComb { setAdds :: S.Set ByteString,
+ setRemoves :: S.Set ByteString }
+ deriving (Show)
+
+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] -> PB.DtOp
+setUpdateOp ops = PB.DtOp { PB.counter_op = Nothing,
+ PB.set_op = Just $ setOpPB ops,
+ PB.map_op = Nothing
+ }
+
+setOpPB :: [SetOp] -> PBSet.SetOp
+setOpPB ops = PBSet.SetOp (toSeq adds) (toSeq rems)
+ where SetOpsComb adds rems = mconcat . map toOpsComb $ ops
+ toSeq = Seq.fromList . S.toList
+
+flagOpPB :: FlagOp -> PBFlag.FlagOp
+flagOpPB (FlagSet True) = PBFlag.ENABLE
+flagOpPB (FlagSet False) = PBFlag.DISABLE
+
+registerOpPB :: RegisterOp -> ByteString
+registerOpPB (RegisterSet x) = x
+
+mapUpdateOp :: [MapOp] -> PB.DtOp
+mapUpdateOp ops = PB.DtOp { PB.counter_op = Nothing,
+ PB.set_op = Nothing,
+ PB.map_op = Just $ mapOpPB ops }
+
+mapOpPB :: [MapOp] -> PBMap.MapOp
+mapOpPB ops = PBMap.MapOp rems updates
+ where rems = Seq.fromList [ toRemove f | MapRemove f <- ops ]
+ updates = Seq.fromList [ toUpdate f u | MapUpdate f u <- ops ]
+
+toRemove :: MapField -> PBMap.MapField
+toRemove (MapField t name) = toField name t
+
+toUpdate :: MapPath -> MapValueOp -> PBMap.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 -> PBMap.MapUpdate
+toUpdate' f t op = setSpecificOp op (updateNothing f t)
+
+setSpecificOp :: MapValueOp -> PBMap.MapUpdate -> PBMap.MapUpdate
+setSpecificOp (MapCounterOp cop) u = u { PBMap.counter_op = Just $ counterOpPB [cop] }
+setSpecificOp (MapSetOp sop) u = u { PBMap.set_op = Just $ setOpPB [sop] }
+setSpecificOp (MapRegisterOp rop) u = u { PBMap.register_op = Just $ registerOpPB rop }
+setSpecificOp (MapFlagOp fop) u = u { PBMap.flag_op = Just $ flagOpPB fop }
+setSpecificOp (MapMapOp mop) u = u { PBMap.map_op = Just $ mapOpPB [mop] }
+
+
+updateNothing :: ByteString -> MapEntryTag -> PBMap.MapUpdate
+updateNothing f t = PBMap.MapUpdate { PBMap.field = toField f t,
+ PBMap.counter_op = Nothing,
+ PBMap.set_op = Nothing,
+ PBMap.register_op = Nothing,
+ PBMap.flag_op = Nothing,
+ PBMap.map_op = Nothing }
+
+toField :: ByteString -> MapEntryTag -> PBMap.MapField
+toField name t = PBMap.MapField { PBMap.name = name,
+ PBMap.type' = typ t }
+ where typ MapCounterTag = PBMap.COUNTER
+ typ MapSetTag = PBMap.SET
+ typ MapRegisterTag = PBMap.REGISTER
+ typ MapFlagTag = PBMap.FLAG
+ typ MapMapTag = PBMap.MAP
diff --git a/src/Network/Riak/CRDT/Request.hs b/src/Network/Riak/CRDT/Request.hs
new file mode 100644
--- /dev/null
+++ b/src/Network/Riak/CRDT/Request.hs
@@ -0,0 +1,63 @@
+-- | module: Network.Riak.CRDT.Request
+-- copyright: (c) 2016 Sentenai
+-- author: Antonio Nikishaev
+-- license: Apache
+--
+module Network.Riak.CRDT.Request
+ (get, counterUpdate, setUpdate, mapUpdate) where
+
+import Data.ByteString.Lazy (ByteString)
+import Network.Riak.CRDT.Ops
+import qualified Network.Riak.CRDT.Types as CRDT
+import qualified Network.Riak.Protocol.DtFetchRequest as DtFetch
+import qualified Network.Riak.Protocol.DtOp as DtOp
+import qualified Network.Riak.Protocol.DtUpdateRequest as DtUpdate
+import Network.Riak.Types
+
+counterUpdate :: [CRDT.CounterOp]
+ -> BucketType -> Bucket -> Key
+ -> DtUpdate.DtUpdateRequest
+counterUpdate ops = update (counterUpdateOp ops)
+
+setUpdate :: [CRDT.SetOp]
+ -> BucketType -> Bucket -> Key
+ -> DtUpdate.DtUpdateRequest
+setUpdate ops = update (setUpdateOp ops)
+
+mapUpdate :: [CRDT.MapOp]
+ -> BucketType -> Bucket -> Key
+ -> DtUpdate.DtUpdateRequest
+mapUpdate ops = update (mapUpdateOp ops)
+
+
+update :: DtOp.DtOp -> BucketType -> Bucket -> Key -> DtUpdate.DtUpdateRequest
+update op t b k = DtUpdate.DtUpdateRequest {
+ DtUpdate.bucket = b,
+ DtUpdate.key = Just k,
+ DtUpdate.type' = t,
+ DtUpdate.context = Nothing,
+ DtUpdate.op = op,
+ DtUpdate.w = Nothing,
+ DtUpdate.dw = Nothing,
+ DtUpdate.pw = Nothing,
+ DtUpdate.return_body = Nothing,
+ DtUpdate.timeout = Nothing,
+ DtUpdate.sloppy_quorum = Nothing,
+ DtUpdate.n_val = Nothing,
+ DtUpdate.include_context = Nothing
+ }
+
+get :: ByteString -> ByteString -> ByteString -> DtFetch.DtFetchRequest
+get t b k = DtFetch.DtFetchRequest {
+ DtFetch.bucket = b,
+ DtFetch.key = k,
+ DtFetch.type' = t,
+ DtFetch.r = Nothing,
+ DtFetch.pr = Nothing,
+ DtFetch.basic_quorum = Nothing,
+ DtFetch.notfound_ok = Nothing,
+ DtFetch.timeout = Nothing,
+ DtFetch.sloppy_quorum = Nothing,
+ DtFetch.n_val = Nothing,
+ DtFetch.include_context = Nothing
+ }
diff --git a/src/Network/Riak/CRDT/Response.hs b/src/Network/Riak/CRDT/Response.hs
new file mode 100644
--- /dev/null
+++ b/src/Network/Riak/CRDT/Response.hs
@@ -0,0 +1,56 @@
+-- | module: Network.Riak.CRDT.Response
+-- copyright: (c) 2016 Sentenai
+-- author: Antonio Nikishaev
+-- license: Apache
+--
+{-# LANGUAGE RecordWildCards #-}
+module Network.Riak.CRDT.Response (get) where
+
+import Control.Applicative ((<$>))
+import qualified Data.Sequence as Seq
+import qualified Data.Map as Map
+import Data.Foldable (toList)
+import Data.Traversable
+import Data.Maybe
+import Network.Riak.Protocol.DtFetchResponse (DtFetchResponse,value,type')
+import Network.Riak.Protocol.DtFetchResponse.DataType (DataType(..))
+import Network.Riak.Protocol.DtValue (counter_value,set_value,map_value)
+
+import qualified Network.Riak.Protocol.MapEntry as M (MapEntry(..))
+import qualified Network.Riak.Protocol.MapField as M
+import qualified Network.Riak.Protocol.MapField.MapFieldType as M
+
+import Network.Riak.CRDT.Types as CRDT
+
+get :: DtFetchResponse -> Maybe CRDT.DataType
+get resp = case type' resp of
+ COUNTER ->
+ DTCounter . Counter <$> (counter_value =<< value resp)
+ SET ->
+ DTSet . setFromSeq . set_value <$> value resp
+ MAP ->
+ DTMap . deconstructMap . map_value <$> value resp
+
+deconstructMap :: Seq.Seq M.MapEntry -> Map
+deconstructMap = Map . Map.fromList . catMaybes . map f . toList
+
+f :: M.MapEntry -> Maybe (MapField, MapEntry)
+f (M.MapEntry{..}) = sequenceA (MapField t name, v)
+ where M.MapField{..} = field
+ t :: MapEntryTag
+ t = typeToTag type'
+ v :: Maybe MapEntry
+ v = case type' of
+ M.COUNTER -> MapCounter . Counter <$> counter_value
+ M.SET -> Just . MapSet . setFromSeq $ set_value
+ M.REGISTER -> MapRegister . Register <$> register_value
+ M.FLAG -> MapFlag . Flag <$> flag_value
+ M.MAP -> Just . MapMap . deconstructMap $ map_value
+
+typeToTag :: M.MapFieldType -> MapEntryTag
+typeToTag M.COUNTER = MapCounterTag
+typeToTag M.SET = MapSetTag
+typeToTag M.REGISTER = MapRegisterTag
+typeToTag M.FLAG = MapFlagTag
+typeToTag M.MAP = MapMapTag
+
diff --git a/src/Network/Riak/CRDT/Riak.hs b/src/Network/Riak/CRDT/Riak.hs
new file mode 100644
--- /dev/null
+++ b/src/Network/Riak/CRDT/Riak.hs
@@ -0,0 +1,35 @@
+-- | module: Network.Riak.CRDT.Riak
+-- copyright: (c) 2016 Sentenai
+-- author: Antonio Nikishaev
+-- license: Apache
+--
+-- CRDT operations
+module Network.Riak.CRDT.Riak where
+
+import Control.Applicative
+import qualified Network.Riak.CRDT.Types as CRDT
+import qualified Network.Riak.Connection as Conn
+import Network.Riak.Types
+
+import qualified Network.Riak.CRDT.Request as Req
+import qualified Network.Riak.CRDT.Response as Resp
+
+
+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 = Conn.exchange_ conn (Req.setUpdate ops t b k)
+
+
+mapSendUpdate :: Connection -> BucketType -> Bucket -> Key
+ -> [CRDT.MapOp] -> IO ()
+mapSendUpdate conn t b k ops = Conn.exchange_ conn (Req.mapUpdate ops t b k)
+
+
+get :: Connection -> BucketType -> Bucket -> Key
+ -> IO (Maybe CRDT.DataType)
+get conn t b k = Resp.get <$> Conn.exchange conn (Req.get t b k)
diff --git a/src/Network/Riak/CRDT/Types.hs b/src/Network/Riak/CRDT/Types.hs
new file mode 100644
--- /dev/null
+++ b/src/Network/Riak/CRDT/Types.hs
@@ -0,0 +1,284 @@
+-- | module: Network.Riak.CRDT.Types
+-- copyright: (c) 2016 Sentenai
+-- author: Antonio Nikishaev
+-- license: Apache
+--
+-- Haskell-side view of CRDT
+--
+{-# LANGUAGE OverloadedStrings, PatternGuards, GeneralizedNewtypeDeriving, 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, setFromSeq, MapEntryTag(..))
+ where
+
+
+import qualified Data.Map.Strict as M
+import qualified Data.Set as S
+import qualified Data.Sequence as Seq
+import qualified Data.Foldable as F
+import Data.ByteString.Lazy (ByteString)
+import Data.Int (Int64)
+import Data.List.NonEmpty
+import Data.Semigroup
+import Data.Default.Class
+import Control.DeepSeq (NFData)
+import GHC.Generics (Generic)
+import Data.String
+
+
+-- | 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)
+
+setFromSeq :: Seq.Seq ByteString -> Set
+setFromSeq = Set . S.fromList . F.toList
+
+-- | 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 Monoid CounterOp where
+ mempty = CounterInc 0
+ CounterInc x `mappend` CounterInc y = CounterInc . getSum $ Sum x <> Sum y
+
diff --git a/src/Network/Riak/Connection/Internal.hs b/src/Network/Riak/Connection/Internal.hs
--- a/src/Network/Riak/Connection/Internal.hs
+++ b/src/Network/Riak/Connection/Internal.hs
@@ -53,7 +53,6 @@
import Network.Riak.Types.Internal hiding (MessageTag(..))
import Network.Socket as Socket
import Numeric (showHex)
-import System.Mem.Weak (addFinalizer)
import System.Random (randomIO)
import Text.ProtocolBuffers (messageGetM, messagePutM, messageSize)
import Text.ProtocolBuffers.Get (Get, Result(..), getWord32be, runGet)
@@ -114,7 +113,6 @@
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
diff --git a/src/Network/Riak/JSON.hs b/src/Network/Riak/JSON.hs
--- a/src/Network/Riak/JSON.hs
+++ b/src/Network/Riak/JSON.hs
@@ -63,11 +63,11 @@
-- Choosing among them is your responsibility.
get :: (FromJSON c, ToJSON c) => Connection -> Bucket -> Key -> R
-> IO (Maybe ([c], VClock))
-get conn bucket key r = fmap convert <$> V.get conn bucket key r
+get conn bucket' key' r = fmap convert <$> V.get conn 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 conn bucket' ks r = map (fmap convert) <$> V.getMany conn bucket' ks r
-- | Store a single value. This may return multiple conflicting
-- siblings. Choosing among them, and storing a new value, is your
@@ -80,16 +80,16 @@
put :: (FromJSON c, ToJSON c) =>
Connection -> 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 bucket' key' mvclock val w dw =
+ convert <$> V.put conn bucket' key' mvclock (json val) w dw
-- | Store a single value indexed.
putIndexed :: (FromJSON c, ToJSON c)
=> Connection -> 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 bucket' key' ixs mvclock val w dw =
+ convert <$> V.putIndexed conn bucket' key' ixs mvclock (json val) w dw
-- | Store a single value, without the possibility of conflict
-- resolution.
@@ -101,8 +101,8 @@
put_ :: (FromJSON c, ToJSON c) =>
Connection -> 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 bucket' key' mvclock val w dw =
+ V.put_ conn 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
@@ -115,8 +115,8 @@
putMany :: (FromJSON c, ToJSON c) =>
Connection -> 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 bucket' puts w dw =
+ map convert <$> V.putMany conn bucket' (map f puts) w dw
where f (k,v,c) = (k,v,json c)
-- | Store many values, without the possibility of conflict
@@ -129,7 +129,7 @@
putMany_ :: (FromJSON c, ToJSON c) =>
Connection -> 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 bucket' puts w dw = V.putMany_ conn bucket' (map f puts) w dw
where f (k,v,c) = (k,v,json c)
convert :: ([JSON a], VClock) -> ([a], VClock)
diff --git a/src/Network/Riak/JSON/Resolvable.hs b/src/Network/Riak/JSON/Resolvable.hs
--- a/src/Network/Riak/JSON/Resolvable.hs
+++ b/src/Network/Riak/JSON/Resolvable.hs
@@ -115,8 +115,8 @@
-> 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)
+ 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'
{-# INLINE putIndexed #-}
diff --git a/src/Network/Riak/Request.hs b/src/Network/Riak/Request.hs
--- a/src/Network/Riak/Request.hs
+++ b/src/Network/Riak/Request.hs
@@ -41,9 +41,12 @@
, getBucket
, SetBucket.SetBucketRequest
, setBucket
+ , getBucketType
-- * Map/reduce
, MapReduceRequest
, mapReduce
+ , search
+ , getIndex
) where
#if __GLASGOW_HASKELL__ < 710
@@ -51,14 +54,14 @@
#endif
import qualified Data.ByteString.Char8 as B8
import Data.Monoid
-import Network.Riak.Protocol.BucketProps hiding (r,rw)
+import Network.Riak.Protocol.BucketProps (BucketProps)
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(..))
+import Network.Riak.Types.Internal hiding (MessageTag(..),bucket,key)
import Network.Riak.Escape (escape)
import qualified Network.Riak.Protocol.DeleteRequest as Del
import qualified Network.Riak.Protocol.Link as Link
@@ -69,11 +72,20 @@
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 qualified Network.Riak.Protocol.GetBucketTypeRequest as GetBucketType
+import qualified Network.Riak.Protocol.SearchQueryRequest as SearchQueryRequest
+import qualified Network.Riak.Protocol.YzIndexGetRequest as YzIndex
-- | Create a ping request.
ping :: PingRequest
ping = PingRequest
{-# INLINE ping #-}
+{-
+-- | Create a dtUpdate request.
+dtUpdate :: DtUpdate.DtUpdateRequest
+dtUpdate = DtUpdate.DtUpdateRequest
+{-# INLINE dtUpdate #-}
+-}
-- | Create a client-ID request.
getClientID :: GetClientIDRequest
@@ -99,6 +111,7 @@
, Get.timeout = Nothing
, Get.sloppy_quorum = Nothing
, Get.n_val = Nothing
+ , Get.type' = Nothing
}
{-# INLINE get #-}
@@ -134,7 +147,11 @@
, Index.stream = Nothing
, Index.max_results = Nothing
, Index.continuation = Nothing
- , Index.timeout = Nothing }
+ , Index.timeout = Nothing
+ , Index.type' = Nothing
+ , Index.term_regex = Nothing
+ , Index.pagination_sort = Nothing
+ }
-- | Create a put request. The bucket and key names are URL-escaped.
-- Any 'Link' values inside the 'Content' are assumed to have been
@@ -144,7 +161,7 @@
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
+ Nothing Nothing Nothing Nothing Nothing Nothing Nothing Nothing Nothing
{-# INLINE put #-}
-- | Create a link. The bucket and key names are URL-escaped.
@@ -156,30 +173,56 @@
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
+ Nothing Nothing Nothing Nothing Nothing Nothing Nothing
{-# INLINE delete #-}
-- | Create a list-buckets request.
-listBuckets :: ListBucketsRequest
+listBuckets :: Maybe BucketType -> ListBucketsRequest
listBuckets = ListBucketsRequest Nothing Nothing
{-# 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 -> Keys.ListKeysRequest
+listKeys t b = Keys.ListKeysRequest (escape b) Nothing (escape <$> t)
{-# 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 -> GetBucket.GetBucketRequest
+getBucket t b = GetBucket.GetBucketRequest (escape b) (escape <$> t)
{-# 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 -> BucketProps -> SetBucket.SetBucketRequest
+setBucket t b ps = SetBucket.SetBucketRequest (escape b) ps (escape <$> t)
{-# INLINE setBucket #-}
+-- | Create a get-bucket-type request. The bucket type is URL-escaped.
+getBucketType :: BucketType -> GetBucketType.GetBucketTypeRequest
+getBucketType t = GetBucketType.GetBucketTypeRequest (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"
+
+-- | Create a search request
+search :: SearchQuery -> Index -> SearchQueryRequest.SearchQueryRequest
+search q ix = SearchQueryRequest.SearchQueryRequest {
+ SearchQueryRequest.q = q,
+ SearchQueryRequest.index = escape ix,
+ SearchQueryRequest.rows = Nothing,
+ SearchQueryRequest.start = Nothing,
+ SearchQueryRequest.sort = Nothing,
+ SearchQueryRequest.filter = Nothing,
+ SearchQueryRequest.df = Nothing,
+ SearchQueryRequest.op = Nothing,
+ SearchQueryRequest.fl = mempty,
+ SearchQueryRequest.presort = Nothing
+ }
+
+getIndex :: Maybe Index -> YzIndex.YzIndexGetRequest
+getIndex = YzIndex.YzIndexGetRequest
+
+
+
+
diff --git a/src/Network/Riak/Resolvable/Internal.hs b/src/Network/Riak/Resolvable/Internal.hs
--- a/src/Network/Riak/Resolvable/Internal.hs
+++ b/src/Network/Riak/Resolvable/Internal.hs
@@ -102,8 +102,8 @@
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
+get doGet conn bucket' key' r =
+ fmap (first resolveMany) `fmap` doGet conn bucket' key' r
{-# INLINE get #-}
getMany :: (Resolvable a) =>
@@ -128,11 +128,11 @@
put :: (Resolvable a) => Put a
-> Connection -> Bucket -> Key -> Maybe VClock -> a -> W -> DW
-> IO (a, VClock)
-put doPut conn bucket key mvclock0 val0 w dw = do
+put doPut conn 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 bucket' key' mvclock val w dw
case xs of
[x] | i > 0 || isJust mvclock -> return (x, vclock)
(_:_) -> do debugValues "put" "conflict" xs
@@ -152,27 +152,27 @@
-> IO ([a], VClock))
-> Connection -> Bucket -> Key -> Maybe VClock -> a -> W -> DW
-> IO ()
-put_ doPut conn bucket key mvclock0 val0 w dw =
- put doPut conn bucket key mvclock0 val0 w dw >> return ()
+put_ doPut conn bucket' key' mvclock0 val0 w dw =
+ put doPut conn 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))
-> 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 bucket' key' r w dw act = do
+ a0 <- liftIO $ get doGet conn 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 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)
-> 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 bucket' key' r w dw act = do
+ a0 <- liftIO $ get doGet conn bucket' key' r
a <- act (fst <$> a0)
- liftIO $ fst <$> put doPut conn bucket key (snd <$> a0) a w dw
+ liftIO $ fst <$> put doPut conn bucket' key' (snd <$> a0) a w dw
{-# INLINE modify_ #-}
putMany :: (Resolvable a) =>
@@ -180,13 +180,13 @@
-> IO [([a], VClock)])
-> Connection -> 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 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 bucket' (map snd puts) w dw
let (conflicts, ok) = partitionEithers $ zipWith mush puts rs
unless (null conflicts) $
debugValues "putMany" "conflicts" conflicts
@@ -203,8 +203,8 @@
(Connection -> Bucket -> [(Key, Maybe VClock, a)] -> W -> DW
-> IO [([a], VClock)])
-> Connection -> Bucket -> [(Key, Maybe VClock, a)] -> W -> DW -> IO ()
-putMany_ doPut conn bucket puts0 w dw =
- putMany doPut conn bucket puts0 w dw >> return ()
+putMany_ doPut conn bucket' puts0 w dw =
+ putMany doPut conn bucket' puts0 w dw >> return ()
{-# INLINE putMany_ #-}
resolveMany' :: (Resolvable a) => a -> [a] -> a
diff --git a/src/Network/Riak/Response.hs b/src/Network/Riak/Response.hs
--- a/src/Network/Riak/Response.hs
+++ b/src/Network/Riak/Response.hs
@@ -1,4 +1,4 @@
-{-# LANGUAGE RecordWildCards, CPP #-}
+{-# LANGUAGE RecordWildCards, CPP, OverloadedStrings #-}
-- |
-- Module: Network.Riak.Request
@@ -24,25 +24,36 @@
, listBuckets
, getBucket
, unescapeLinks
+ , search
+ , getIndex
) where
#if __GLASGOW_HASKELL__ < 710
import Control.Applicative ((<$>))
#endif
-import Data.Maybe (fromMaybe)
import Network.Riak.Escape (unescape)
-import Network.Riak.Protocol.BucketProps
+import Network.Riak.Protocol.BucketProps (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.Protocol.SearchQueryResponse
+import Network.Riak.Protocol.SearchDoc
+import qualified Network.Riak.Protocol.YzIndexGetResponse as Yz
import Network.Riak.Types.Internal hiding (MessageTag(..))
import qualified Network.Riak.Protocol.Link as Link
+import qualified Network.Riak.Protocol.Pair as Pair
+
import qualified Data.ByteString.Lazy as L
+import qualified Data.ByteString.Lazy.Char8 as LC
import qualified Data.Sequence as Seq
+import qualified Data.Map.Strict as M
+import Data.Maybe (fromMaybe)
+import Data.Semigroup
+import Control.Arrow ((&&&))
+import Data.Foldable (toList)
getClientID :: GetClientIDResponse -> ClientID
getClientID = client_id
@@ -76,5 +87,31 @@
-- 'Content' value.
unescapeLinks :: Content -> Content
unescapeLinks c = c { links = go <$> links c }
- where go l = l { bucket = unescape <$> bucket l
+ where go l = l { Link.bucket = unescape <$> Link.bucket l
, Link.key = unescape <$> Link.key l }
+
+search :: SearchQueryResponse -> [SearchResult]
+search = map toSearchResult . toList . docs
+
+toSearchResult :: SearchDoc -> SearchResult
+toSearchResult r = SearchResult {
+ bucketType = field "_yz_rt",
+ bucket = field "_yz_rb",
+ key = field "_yz_rk",
+ score = fmap (read . LC.unpack) =<< M.lookup "score" info
+ }
+ where
+ info :: M.Map L.ByteString (Maybe L.ByteString)
+ info = M.fromList . map (Pair.key &&& Pair.value) . toList . fields $ r
+
+ field :: L.ByteString -> L.ByteString
+ field name
+ = maybe (unexpected $ "field " <> show name <> " has empty value") id
+ . maybe (unexpected $ "no " <> show name <> " field") id
+ . M.lookup name $ info
+
+ unexpected = unexError "Network.Riak.Response" "search"
+
+
+getIndex :: Yz.YzIndexGetResponse -> [IndexInfo]
+getIndex = toList . Yz.index
diff --git a/src/Network/Riak/Search.hs b/src/Network/Riak/Search.hs
new file mode 100644
--- /dev/null
+++ b/src/Network/Riak/Search.hs
@@ -0,0 +1,28 @@
+-- | module: Network.Riak.Search
+-- copyright: (c) 2016 Sentenai
+-- author: Antonio Nikishaev
+-- license: Apache
+--
+-- Solr search
+--
+-- http://docs.basho.com/riak/2.1.3/dev/using/search/
+
+module Network.Riak.Search where
+
+import Network.Riak.Connection.Internal
+import Network.Riak.Types.Internal
+import qualified Network.Riak.Request as Req
+import qualified Network.Riak.Response as Resp
+import Control.Applicative
+
+-- | Get an index info for @Just index@, or get all indexes for
+-- @Nothing@.
+getIndex :: Connection -> Maybe Index -> IO [IndexInfo]
+getIndex conn ix = Resp.getIndex <$> exchange conn (Req.getIndex ix)
+
+
+-- | Search by raw 'SearchQuery' request (a bytestring) using an
+-- index.
+searchRaw :: Connection -> SearchQuery -> Index -> IO [SearchResult]
+searchRaw conn q ix = Resp.search <$> exchange conn (Req.search q ix)
+
diff --git a/src/Network/Riak/Tag.hs b/src/Network/Riak/Tag.hs
--- a/src/Network/Riak/Tag.hs
+++ b/src/Network/Riak/Tag.hs
@@ -18,9 +18,14 @@
) where
import Data.Binary.Put (Put, putWord8)
+import Data.Word (Word8)
+import qualified Data.HashMap.Strict as HM
+import Control.Applicative
+import Data.Tuple (swap)
import Network.Riak.Protocol.DeleteRequest
import Network.Riak.Protocol.ErrorResponse
import Network.Riak.Protocol.GetBucketRequest
+import Network.Riak.Protocol.GetBucketTypeRequest
import Network.Riak.Protocol.GetBucketResponse
import Network.Riak.Protocol.GetClientIDRequest
import Network.Riak.Protocol.GetClientIDResponse
@@ -41,6 +46,14 @@
import Network.Riak.Protocol.ServerInfo
import Network.Riak.Protocol.SetBucketRequest
import Network.Riak.Protocol.SetClientIDRequest
+import Network.Riak.Protocol.DtFetchRequest
+import Network.Riak.Protocol.DtFetchResponse
+import Network.Riak.Protocol.DtUpdateRequest
+import Network.Riak.Protocol.DtUpdateResponse
+import Network.Riak.Protocol.SearchQueryRequest
+import Network.Riak.Protocol.SearchQueryResponse
+import Network.Riak.Protocol.YzIndexGetRequest
+import Network.Riak.Protocol.YzIndexGetResponse
import Network.Riak.Types.Internal as Types
import Text.ProtocolBuffers.Get (Get, getWord8)
@@ -208,6 +221,14 @@
expectedResponse _ = Types.SetBucketResponse
{-# INLINE expectedResponse #-}
+instance Request GetBucketTypeRequest where
+ expectedResponse _ = Types.GetBucketResponse
+
+instance Tagged GetBucketTypeRequest where
+ messageTag _ = Types.GetBucketTypeRequest
+
+instance Exchange GetBucketTypeRequest GetBucketResponse
+
instance Tagged MapReduceRequest where
messageTag _ = Types.MapReduceRequest
{-# INLINE messageTag #-}
@@ -220,21 +241,157 @@
messageTag _ = Types.MapReduceResponse
{-# INLINE messageTag #-}
+
instance Response MapReduce
instance Exchange MapReduceRequest MapReduce
+instance Tagged DtFetchRequest where
+ messageTag _ = Types.DtFetchRequest
+ {-# INLINE messageTag #-}
+
+instance Tagged DtFetchResponse where
+ messageTag _ = Types.DtFetchResponse
+ {-# INLINE messageTag #-}
+
+instance Request DtFetchRequest where
+ expectedResponse _ = Types.DtFetchResponse
+ {-# INLINE expectedResponse #-}
+
+instance Response DtFetchResponse
+
+instance Exchange DtFetchRequest DtFetchResponse
+
+instance Tagged DtUpdateRequest where
+ messageTag _ = Types.DtUpdateRequest
+ {-# INLINE messageTag #-}
+
+instance Tagged DtUpdateResponse where
+ messageTag _ = Types.DtUpdateResponse
+ {-# INLINE messageTag #-}
+
+instance Request DtUpdateRequest where
+ expectedResponse _ = Types.DtUpdateResponse
+ {-# INLINE expectedResponse #-}
+
+instance Response DtUpdateResponse
+
+instance Exchange DtUpdateRequest DtUpdateResponse
+
+instance Tagged SearchQueryRequest where
+ messageTag _ = Types.SearchQueryRequest
+ {-# INLINE messageTag #-}
+
+instance Request SearchQueryRequest where
+ expectedResponse _ = Types.SearchQueryResponse
+ {-# INLINE expectedResponse #-}
+
+instance Tagged SearchQueryResponse where
+ messageTag _ = Types.SearchQueryResponse
+ {-# INLINE messageTag #-}
+
+instance Response SearchQueryResponse
+
+instance Exchange SearchQueryRequest SearchQueryResponse
+
+instance Tagged YzIndexGetRequest where
+ messageTag _ = Types.YokozunaIndexGetRequest
+
+instance Request YzIndexGetRequest where
+ expectedResponse _ = Types.YokozunaIndexGetResponse
+
+instance Tagged YzIndexGetResponse where
+ messageTag _ = Types.YokozunaIndexGetResponse
+
+instance Response YzIndexGetResponse
+
+instance Exchange YzIndexGetRequest YzIndexGetResponse
+
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,YokozunaIndexPutReq),
+ -- (57,YokozunaIndexDeleteReq),
+ -- (58,YokozunaSchemaGetReq),
+ -- (59,YokozunaSchemaGetResp),
+ -- (60,YokozunaSchemaPutReq),
+ (80, Types.DtFetchRequest),
+ (81, Types.DtFetchResponse),
+ (82, Types.DtUpdateRequest),
+ (83, Types.DtUpdateResponse)
+ -- (253,RpbAuthReq),
+ -- (254,RpbAuthResp),
+ -- (255,RpbStartTls)
+ ]
+
diff --git a/src/Network/Riak/Types.hs b/src/Network/Riak/Types.hs
--- a/src/Network/Riak/Types.hs
+++ b/src/Network/Riak/Types.hs
@@ -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
diff --git a/src/Network/Riak/Types/Internal.hs b/src/Network/Riak/Types/Internal.hs
--- a/src/Network/Riak/Types/Internal.hs
+++ b/src/Network/Riak/Types/Internal.hs
@@ -1,5 +1,5 @@
{-# LANGUAGE DeriveDataTypeable, FunctionalDependencies, MultiParamTypeClasses,
- RecordWildCards #-}
+ RecordWildCards, DeriveGeneric #-}
-- |
-- Module: Network.Riak.Types.Internal
@@ -25,11 +25,15 @@
, unexError
-- * Data types
, Bucket
+ , BucketType
, Key
, Index
, IndexQuery(..)
, IndexValue(..)
, Tag
+ , SearchQuery
+ , SearchResult(..)
+ , IndexInfo
, VClock(..)
, Job(..)
-- * Quorum management
@@ -48,15 +52,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.Lazy (ByteString)
+import Data.Digest.Pure.MD5 (md5)
+import Data.Hashable (Hashable)
+import Data.IORef (IORef)
+import Data.Typeable (Typeable)
+import Data.Word (Word32)
+import GHC.Generics (Generic)
+import qualified Network.Riak.Protocol.YzIndex as YzIndex
+import Network.Socket (HostName, ServiceName, Socket)
+import Text.ProtocolBuffers (ReflectDescriptor, Wire)
+
-- | 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 +137,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
@@ -159,7 +171,24 @@
| 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 = YzIndex.YzIndex
+
+-- | Solr search result
+data SearchResult = SearchResult {
+ bucketType :: BucketType, -- ^ bucket type
+ bucket :: Bucket, -- ^ bucket
+ key :: Key, -- ^ key
+ score :: Maybe Score -- ^ score, if provided
+ } deriving (Eq,Show)
+
+-- | List of (known to us) inbound or outbound message identifiers.
data MessageTag = ErrorResponse
| PingRequest
| PingResponse
@@ -183,12 +212,23 @@
| GetBucketResponse
| SetBucketRequest
| SetBucketResponse
+ | GetBucketTypeRequest
| MapReduceRequest
| MapReduceResponse
| IndexRequest
| IndexResponse
- deriving (Eq, Show, Enum, Typeable)
+ | DtFetchRequest
+ | DtFetchResponse
+ | DtUpdateRequest
+ | DtUpdateResponse
+ | SearchQueryRequest
+ | SearchQueryResponse
+ | YokozunaIndexGetRequest
+ | YokozunaIndexGetResponse
+ deriving (Eq, Show, Generic)
+instance Hashable MessageTag
+
-- | Messages are tagged.
class Tagged msg where
messageTag :: msg -> MessageTag -- ^ Retrieve a message's tag.
@@ -205,7 +245,7 @@
class (Tagged msg, ReflectDescriptor msg, Show msg, Wire msg) => Response msg
class (Request req, Response resp) => Exchange req resp
- | req -> resp, resp -> req
+ | req -> resp
instance (Tagged a, Tagged b) => Tagged (Either a b) where
messageTag (Left l) = messageTag l
diff --git a/tests/CRDTProperties.hs b/tests/CRDTProperties.hs
new file mode 100644
--- /dev/null
+++ b/tests/CRDTProperties.hs
@@ -0,0 +1,309 @@
+-- | module: CRDTProperties
+-- copyright: (c) 2016 Sentenai
+-- author: Antonio Nikishaev
+-- license: Apache
+--
+{-# LANGUAGE OverloadedStrings, FlexibleContexts, TupleSections, ScopedTypeVariables,
+ GADTs, StandaloneDeriving, UndecidableInstances, PatternGuards, MultiParamTypeClasses #-}
+{-# 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.
+
+import Control.Applicative
+import Control.Monad.RWS
+import Data.ByteString.Lazy (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 Common
+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_) = withSomeConnection $ \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
+
+-- | Abstract machine.
+-- Yields a value on 'Get', modifies state on 'Update'.
+machine :: (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 :: (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_ :: (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 = withSomeConnection $ \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
+ ]
diff --git a/tests/Common.hs b/tests/Common.hs
new file mode 100644
--- /dev/null
+++ b/tests/Common.hs
@@ -0,0 +1,17 @@
+module Common (withSomeConnection) where
+
+import qualified Network.Riak.Basic as B
+import Network.Riak.Connection.Pool (Pool, create, withConnection)
+import Network.Riak.Connection (defaultClient)
+import System.IO.Unsafe (unsafePerformIO)
+
+
+pool :: Pool
+{-# NOINLINE pool #-}
+pool = unsafePerformIO $
+ create defaultClient 1 1 1
+
+-- | run action in some riak connection
+withSomeConnection :: (B.Connection -> IO a) -> IO a
+withSomeConnection = withConnection pool
+
diff --git a/tests/Properties.hs b/tests/Properties.hs
--- a/tests/Properties.hs
+++ b/tests/Properties.hs
@@ -9,39 +9,56 @@
import Control.Applicative ((<$>))
#endif
import qualified Data.ByteString.Lazy as L
-import Data.IORef (IORef, modifyIORef, newIORef)
+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 Common
instance Arbitrary L.ByteString where
arbitrary = L.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 . L.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 . L.null)
+
+
+t_put_get :: QCBucket -> QCKey -> L.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):)
+ act = withSomeConnection $ \c -> do
p <- Just <$> B.put c b k Nothing (binary v) Default Default
r <- B.get c b k Default
return (p,r)
- notempty = not . L.null
+
+put_delete_get :: QCBucket -> QCKey -> L.ByteString -> Property
+put_delete_get (QCBucket b) (QCKey k) v
+ = monadicIO $ do
+ r <- run act
+ assert $ isNothing r
+ where
+ act = withSomeConnection $ \c -> do
+ _ <- B.put c b k Nothing (binary v) Default Default
+ B.delete c b k Default
+ B.get c b k Default
+
+
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
+ ]
+
+
+
diff --git a/tests/Test.hs b/tests/Test.hs
--- a/tests/Test.hs
+++ b/tests/Test.hs
@@ -3,43 +3,61 @@
module Main where
-import Control.Exception (finally)
-import Control.Monad (forM_)
-import Data.IORef
+import Control.Monad
import qualified Data.Map as M
+import qualified Data.Set as S
+import Data.List.NonEmpty (NonEmpty(..))
+import Data.Semigroup
import Data.Text (Text)
+import Control.Concurrent (threadDelay)
import qualified Network.Riak as Riak
import qualified Network.Riak.Basic as B
+import qualified Network.Riak.CRDT as C
+import qualified Network.Riak.CRDT.Riak as C
+import qualified Network.Riak.Search as S
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 Network.Riak.Types
+import Network.Riak.Types hiding (key)
import qualified Properties
+import qualified CRDTProperties as CRDT
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 = defaultMain tests
-tests :: TestTree
-tests = testGroup "Tests" [properties, integrationalTests]
+tests :: TestTree
+tests = testGroup "Tests" [properties,
+ integrationalTests,
+ ping'o'death,
+ crdts,
+ searches
+ ]
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" [
+ search,
+ getIndex
+ ]
+
testClusterSimple :: TestTree
testClusterSimple = testCase "testClusterSimple" $ do
rc <- Riak.connectToCluster [Riak.defaultClient]
@@ -57,3 +75,101 @@
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 = do
+ c <- Riak.connect Riak.defaultClient
+ replicateM_ 1024 $ Riak.ping c
+
+
+counter :: TestTree
+counter = testCase "increment" $ do
+ conn <- Riak.connect Riak.defaultClient
+ 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" $ do
+ conn <- Riak.connect Riak.defaultClient
+ 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" $ do
+ conn <- Riak.connect Riak.defaultClient
+ 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))
+
+
+search :: TestTree
+search = testCase "basic searchRaw" $ do
+ conn <- Riak.connect Riak.defaultClient
+ C.sendModify conn btype buck key [C.SetRemove kw]
+ delay
+ a <- query conn ("set:" <> kw)
+ assertEqual "should not found non-existing" [] a
+ C.sendModify conn btype buck key [C.SetAdd kw]
+ delay
+ b <- query conn ("set:" <> kw)
+ assertBool "searches specific" $ not (null b)
+ c <- query conn ("set:*")
+ assertBool "searches *" $ not (null 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
+
+
+getIndex :: TestTree
+getIndex = testCase "getIndex" $ do
+ conn <- Riak.connect Riak.defaultClient
+ all <- S.getIndex conn Nothing
+ one <- S.getIndex conn (Just "set-ix")
+ assertBool "all indeces" $ not (null all)
+ assertEqual "set index" 1 (length one)
+