riak 0.9.1.1 → 1.0.0.0

raw patch · 25 files changed

+1797/−120 lines, 25 filesdep +criteriondep +data-default-classdep +deepseqdep ~basedep ~bytestringdep ~protocol-buffersPVP ok

version bump matches the API change (PVP)

Dependencies added: criterion, data-default-class, deepseq, hashable, mtl, semigroups, unordered-containers

Dependency ranges changed: base, bytestring, protocol-buffers, riak-protobuf

API changes (from Hackage documentation)

Files

• Changes.md +8/−0
• benchmarks/Main.hs +133/−0
• riak.cabal +47/−11
• src/Network/Riak.hs +2/−0
• src/Network/Riak/Basic.hs +15/−9
• src/Network/Riak/CRDT.hs +172/−0
• src/Network/Riak/CRDT/Ops.hs +118/−0
• src/Network/Riak/CRDT/Request.hs +63/−0
• src/Network/Riak/CRDT/Response.hs +56/−0
• src/Network/Riak/CRDT/Riak.hs +35/−0
• src/Network/Riak/CRDT/Types.hs +284/−0
• src/Network/Riak/Connection/Internal.hs +0/−2
• src/Network/Riak/JSON.hs +11/−11
• src/Network/Riak/JSON/Resolvable.hs +2/−2
• src/Network/Riak/Request.hs +58/−15
• src/Network/Riak/Resolvable/Internal.hs +16/−16
• src/Network/Riak/Response.hs +42/−5
• src/Network/Riak/Search.hs +28/−0
• src/Network/Riak/Tag.hs +161/−4
• src/Network/Riak/Types.hs +2/−0
• src/Network/Riak/Types/Internal.hs +52/−12
• tests/CRDTProperties.hs +309/−0
• tests/Common.hs +17/−0
• tests/Properties.hs +36/−19
• tests/Test.hs +130/−14

@@ -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

@@ -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+

@@ -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+++

@@ -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

@@ -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.

@@ -0,0 +1,172 @@+{- | 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 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

@@ -0,0 +1,118 @@+-- | module:    Network.Riak.CRDT.Ops+--   copyright: (c) 2016 Sentenai+--   author:    Antonio Nikishaev <me@lelf.lu>+--   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

@@ -0,0 +1,63 @@+-- |   module:    Network.Riak.CRDT.Request+--     copyright: (c) 2016 Sentenai+--     author:    Antonio Nikishaev <me@lelf.lu>+--     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+            }

@@ -0,0 +1,56 @@+-- |   module:    Network.Riak.CRDT.Response+--     copyright: (c) 2016 Sentenai+--     author:    Antonio Nikishaev <me@lelf.lu>+--     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+

@@ -0,0 +1,35 @@+-- |   module:    Network.Riak.CRDT.Riak+--     copyright: (c) 2016 Sentenai+--     author:    Antonio Nikishaev <me@lelf.lu>+--     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)

@@ -0,0 +1,284 @@+-- |   module:    Network.Riak.CRDT.Types+--     copyright: (c) 2016 Sentenai+--     author:    Antonio Nikishaev <me@lelf.lu>+--     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+

@@ -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 

@@ -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)

@@ -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 #-} 

@@ -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++++

@@ -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

@@ -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

@@ -0,0 +1,28 @@+-- | 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/++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)+

@@ -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)+ ]+

@@ -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

@@ -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

@@ -0,0 +1,309 @@+-- |   module:    CRDTProperties+--     copyright: (c) 2016 Sentenai+--     author:    Antonio Nikishaev <me@lelf.lu>+--     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+        ]

@@ -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+

@@ -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+ ]+++

@@ -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)+