diff --git a/Changes.md b/Changes.md --- a/Changes.md +++ b/Changes.md @@ -1,3 +1,12 @@ +* 1.2.0.0 + - Bump `riak-protobuf` to `1.0.0`, which: + - Switches the protocol buffer library from `protocol-buffers` to `proto-lens` + - Retains the original protobuf message names that Basho defined, e.g. + `Content` is now named `RpbContent`. + - Uses strict `ByteString`s instead of lazy ones + - Uses lists instead of `Seq`s + - Bump `stm` upper bound. + - Add `network >= 3.0` lower bound. * 1.1.2.6 - Regenerate protobuf for protocol-buffers 2.4.12. (https://github.com/riak-haskell-client/riak-haskell-client/issues/115) - Bump riak-protobuf version to 0.24.0.0. @@ -23,8 +32,7 @@ - Bugfix for exceptions being ignored (https://github.com/markhibberd/riak-haskell-client/pull/46) * 1.0 - No longer expected to work with riak <= 2 - - CRDT functionality, including high-level interface - and quickCheck tests + - CRDT functionality, including high-level interface and quickCheck tests - Basic yokozuna search functionality - CRDT benchmarking suite - Bugfixes diff --git a/benchmarks/Main.hs b/benchmarks/Main.hs --- a/benchmarks/Main.hs +++ b/benchmarks/Main.hs @@ -6,7 +6,7 @@ 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 qualified Data.ByteString.Char8 as B import Control.Applicative import Control.Monad import Data.Maybe diff --git a/riak.cabal b/riak.cabal --- a/riak.cabal +++ b/riak.cabal @@ -1,5 +1,5 @@ name: riak -version: 1.1.2.6 +version: 1.2.0.0 synopsis: A Haskell client for the Riak decentralized data store description: A Haskell client library for the Riak decentralized data @@ -46,8 +46,9 @@ Changes.md tests/test.yaml -cabal-version: >=1.8 -tested-with: GHC==8.4.4, GHC==8.2.2, GHC==8.0.2 +cabal-version: >=1.10 +tested-with: GHC==8.4.4, GHC==8.2.2, GHC == 8.6.3, GHC == 8.8.4, GHC==8.10.4 + source-repository head type: git location: https://github.com:riak-haskell-client/riak-haskell-client.git @@ -66,6 +67,7 @@ default: False library + default-language: Haskell2010 hs-source-dirs: src exposed-modules: Network.Riak @@ -97,11 +99,12 @@ other-modules: Network.Riak.Connection.NoPush + Network.Riak.Lens Network.Riak.Resolvable.Internal Network.Riak.Tag build-depends: - aeson >= 0.8 && < 1.4, + aeson >= 0.8 && < 1.5.7, async >= 2.0.0.0 && < 2.3, attoparsec >= 0.12.1.6 && < 0.14, base >= 3 && < 5, @@ -119,20 +122,22 @@ transformers-base == 0.4.*, mersenne-random-pure64 >= 0.2.0.4 && < 0.3, monad-control >= 1.0.0.4 && < 1.1, - network >= 2.3, + network >= 3.0 && < 3.1, resource-pool == 0.2.*, - protocol-buffers >= 2.1.4 && < 2.5, + proto-lens == 0.7.*, pureMD5, random, - riak-protobuf == 0.24.*, + riak-protobuf == 0.25.*, semigroups >= 0.16, stm >= 2.4 && < 2.6, text == 1.2.*, - time >= 1.4.2 && < 1.9, + time >= 1.4.2 && < 1.10, vector >= 0.10.12.3 && < 0.13, unordered-containers >= 0.2.5 - + default-extensions: + DoAndIfThenElse + if flag(debug) cpp-options: -DASSERTS -DDEBUG @@ -144,6 +149,7 @@ ghc-options: -Wall test-suite test + default-language: Haskell2010 type: exitcode-stdio-1.0 main-is: Test.hs hs-source-dirs: tests, tests/dsl @@ -167,6 +173,7 @@ bytestring, containers, HUnit, + microlens, process, QuickCheck, tasty, @@ -180,6 +187,7 @@ yaml >= 0.8.19 benchmark bench + default-language: Haskell2010 type: exitcode-stdio-1.0 hs-source-dirs: benchmarks main-is: Main.hs 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 @@ -51,20 +51,15 @@ import Control.Applicative ((<$>)) #endif import Control.Monad.IO.Class -import qualified Data.Foldable as F -import Data.Maybe (fromMaybe) -import qualified Data.Sequence as Seq import Network.Riak.Connection.Internal -import Network.Riak.Escape (unescape) -import Network.Riak.Protocol.BucketProps -import Network.Riak.Protocol.Content -import Network.Riak.Protocol.ListKeysResponse -import Network.Riak.Protocol.MapReduce as MapReduce -import Network.Riak.Protocol.ServerInfo -import qualified Network.Riak.Request as Req -import qualified Network.Riak.Response as Resp -import Network.Riak.Types.Internal hiding (MessageTag (..)) -import qualified Network.Riak.Types.Internal as T +import Network.Riak.Escape (unescape) +import Network.Riak.Lens +import Network.Riak.Types.Internal hiding (MessageTag(..)) +import qualified Data.Foldable as F +import qualified Data.Riak.Proto as Proto +import qualified Network.Riak.Request as Req +import qualified Network.Riak.Response as Resp +import qualified Network.Riak.Types.Internal as T -- | Check to see if the connection to the server is alive. ping :: Connection -> IO () @@ -75,13 +70,13 @@ getClientID conn = Resp.getClientID <$> exchange conn Req.getClientID -- | Retrieve information about the server. -getServerInfo :: Connection -> IO ServerInfo +getServerInfo :: Connection -> IO Proto.RpbGetServerInfoResp getServerInfo conn = exchange conn Req.getServerInfo -- | Retrieve a value. This may return multiple conflicting siblings. -- Choosing among them is your responsibility. get :: Connection -> Maybe T.BucketType -> T.Bucket -> T.Key -> R - -> IO (Maybe (Seq.Seq Content, VClock)) + -> IO (Maybe ([Proto.RpbContent], VClock)) get conn btype bucket key r = Resp.get <$> exchangeMaybe conn (Req.get btype bucket key r) -- | Store a single value. This may return multiple conflicting @@ -93,8 +88,8 @@ -- If you omit a 'T.VClock' but the type+bucket+key /does/ exist, your -- value will not be stored. put :: Connection -> Maybe T.BucketType -> T.Bucket -> T.Key -> Maybe T.VClock - -> Content -> W -> DW - -> IO (Seq.Seq Content, VClock) + -> Proto.RpbContent -> W -> DW + -> IO ([Proto.RpbContent], VClock) put conn btype bucket key mvclock cont w dw = Resp.put <$> exchange conn (Req.put btype bucket key mvclock cont w dw True) @@ -106,7 +101,7 @@ -- If you omit a 'T.VClock' but the type+bucket+key /does/ exist, your -- value will not be stored, and you will not be notified. put_ :: Connection -> Maybe T.BucketType -> T.Bucket -> T.Key -> Maybe T.VClock - -> Content -> W -> DW + -> Proto.RpbContent -> W -> DW -> IO () put_ conn btype bucket key mvclock cont w dw = exchange_ conn (Req.put btype bucket key mvclock cont w dw False) @@ -118,7 +113,7 @@ -- | List the buckets in the cluster. -- -- /Note/: this operation is expensive. Do not use it in production. -listBuckets :: Connection -> Maybe BucketType -> IO (Seq.Seq T.Bucket) +listBuckets :: Connection -> Maybe BucketType -> IO [T.Bucket] listBuckets conn btype = Resp.listBuckets <$> exchange conn (Req.listBuckets btype) -- | Fold over the keys in a bucket. @@ -130,32 +125,32 @@ liftIO $ sendRequest conn $ Req.listKeys btype bucket let g z = f z . unescape loop z = do - ListKeysResponse{..} <- liftIO $ recvResponse conn - z1 <- F.foldlM g z keys - if fromMaybe False done + response <- liftIO $ (recvResponse conn :: IO Proto.RpbListKeysResp) + z1 <- F.foldlM g z (response ^. Proto.keys) + if response ^. Proto.done then return z1 else loop z1 loop z0 -- | Retrieve the properties of a bucket. -getBucket :: Connection -> Maybe BucketType -> Bucket -> IO BucketProps +getBucket :: Connection -> Maybe BucketType -> Bucket -> IO Proto.RpbBucketProps getBucket conn btype bucket = Resp.getBucket <$> exchange conn (Req.getBucket btype bucket) -- | Store new properties for a bucket. -setBucket :: Connection -> Maybe BucketType -> Bucket -> BucketProps -> IO () +setBucket :: Connection -> Maybe BucketType -> Bucket -> Proto.RpbBucketProps -> IO () setBucket conn btype bucket props = exchange_ conn $ Req.setBucket btype bucket props -- | Gets the bucket properties associated with a bucket type. -getBucketType :: Connection -> T.BucketType -> IO BucketProps +getBucketType :: Connection -> T.BucketType -> IO Proto.RpbBucketProps getBucketType conn btype = Resp.getBucket <$> exchange conn (Req.getBucketType btype) -- | Run a 'MapReduce' job. Its result is consumed via a strict left -- fold. -mapReduce :: Connection -> Job -> (a -> MapReduce -> a) -> a -> IO a +mapReduce :: Connection -> Job -> (a -> Proto.RpbMapRedResp -> a) -> a -> IO a mapReduce conn job f z0 = loop z0 =<< (exchange conn . Req.mapReduce $ job) where loop z mr = do let !z' = f z mr - if fromMaybe False . MapReduce.done $ mr + if mr ^. Proto.done then return z' else loop z' =<< recvResponse conn diff --git a/src/Network/Riak/CRDT/Ops.hs b/src/Network/Riak/CRDT/Ops.hs --- a/src/Network/Riak/CRDT/Ops.hs +++ b/src/Network/Riak/CRDT/Ops.hs @@ -18,29 +18,19 @@ , mapUpdateOp ) where -import Data.ByteString.Lazy (ByteString) -import Data.Semigroup (Semigroup ((<>))) -import qualified Data.Sequence as Seq -import qualified Data.Set as S +import Data.ByteString (ByteString) +import Data.Semigroup (Semigroup((<>))) +import qualified Data.Set as S +import qualified Data.Riak.Proto as Proto import Network.Riak.CRDT.Types -import qualified Network.Riak.Protocol.CounterOp as PB -import qualified Network.Riak.Protocol.DtOp as PB -import qualified Network.Riak.Protocol.MapField as PBMap -import qualified Network.Riak.Protocol.MapField.MapFieldType as PBMap -import qualified Network.Riak.Protocol.MapOp as PBMap -import qualified Network.Riak.Protocol.MapUpdate as PBMap -import qualified Network.Riak.Protocol.MapUpdate.FlagOp as PBFlag -import qualified Network.Riak.Protocol.SetOp as PBSet +import Network.Riak.Lens -counterUpdateOp :: [CounterOp] -> PB.DtOp -counterUpdateOp ops = PB.DtOp { PB.counter_op = Just $ counterOpPB ops, - PB.set_op = Nothing, - PB.map_op = Nothing - } +counterUpdateOp :: [CounterOp] -> Proto.DtOp +counterUpdateOp ops = Proto.defMessage & Proto.counterOp .~ counterOpPB ops -counterOpPB :: [CounterOp] -> PB.CounterOp -counterOpPB ops = PB.CounterOp (Just i) +counterOpPB :: [CounterOp] -> Proto.CounterOp +counterOpPB ops = Proto.defMessage & Proto.increment .~ i where CounterInc i = mconcat ops @@ -61,66 +51,57 @@ -setUpdateOp :: [SetOp] -> PB.DtOp -setUpdateOp ops = PB.DtOp { PB.counter_op = Nothing, - PB.set_op = Just $ setOpPB ops, - PB.map_op = Nothing - } +setUpdateOp :: [SetOp] -> Proto.DtOp +setUpdateOp ops = Proto.defMessage & Proto.setOp .~ setOpPB ops -setOpPB :: [SetOp] -> PBSet.SetOp -setOpPB ops = PBSet.SetOp (toSeq adds) (toSeq rems) +setOpPB :: [SetOp] -> Proto.SetOp +setOpPB ops = Proto.defMessage & Proto.adds .~ S.toList adds + & Proto.removes .~ S.toList rems where SetOpsComb adds rems = mconcat . map toOpsComb $ ops - toSeq = Seq.fromList . S.toList -flagOpPB :: FlagOp -> PBFlag.FlagOp -flagOpPB (FlagSet True) = PBFlag.ENABLE -flagOpPB (FlagSet False) = PBFlag.DISABLE +flagOpPB :: FlagOp -> Proto.MapUpdate'FlagOp +flagOpPB (FlagSet True) = Proto.MapUpdate'ENABLE +flagOpPB (FlagSet False) = Proto.MapUpdate'DISABLE registerOpPB :: RegisterOp -> ByteString registerOpPB (RegisterSet x) = x -mapUpdateOp :: [MapOp] -> PB.DtOp -mapUpdateOp ops = PB.DtOp { PB.counter_op = Nothing, - PB.set_op = Nothing, - PB.map_op = Just $ mapOpPB ops } +mapUpdateOp :: [MapOp] -> Proto.DtOp +mapUpdateOp ops = Proto.defMessage & Proto.mapOp .~ 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 ] +mapOpPB :: [MapOp] -> Proto.MapOp +mapOpPB ops = Proto.defMessage & Proto.removes .~ rems + & Proto.updates .~ updates + where rems = [ toRemove f | MapRemove f <- ops ] + updates = [ toUpdate f u | MapUpdate f u <- ops ] -toRemove :: MapField -> PBMap.MapField +toRemove :: MapField -> Proto.MapField toRemove (MapField t name) = toField name t -toUpdate :: MapPath -> MapValueOp -> PBMap.MapUpdate +toUpdate :: MapPath -> MapValueOp -> Proto.MapUpdate toUpdate (MapPath (e :| [])) op = toUpdate' e (mapEntryTag op) op toUpdate (MapPath (e :| (r:rs))) op = toUpdate' e MapMapTag op' where op' = MapMapOp (MapUpdate (MapPath (r:|rs)) op) -toUpdate' :: ByteString -> MapEntryTag -> MapValueOp -> PBMap.MapUpdate +toUpdate' :: ByteString -> MapEntryTag -> MapValueOp -> Proto.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] } +setSpecificOp :: MapValueOp -> Proto.MapUpdate -> Proto.MapUpdate +setSpecificOp (MapCounterOp cop) = Proto.counterOp .~ counterOpPB [cop] +setSpecificOp (MapSetOp sop) = Proto.setOp .~ setOpPB [sop] +setSpecificOp (MapRegisterOp rop) = Proto.registerOp .~ registerOpPB rop +setSpecificOp (MapFlagOp fop) = Proto.flagOp .~ flagOpPB fop +setSpecificOp (MapMapOp mop) = Proto.mapOp .~ mapOpPB [mop] -updateNothing :: ByteString -> MapEntryTag -> 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 } +updateNothing :: ByteString -> MapEntryTag -> Proto.MapUpdate +updateNothing f t = Proto.defMessage & Proto.field .~ toField f t -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 +toField :: ByteString -> MapEntryTag -> Proto.MapField +toField name t = Proto.defMessage & Proto.name .~ name + & Proto.type' .~ typ t + where typ MapCounterTag = Proto.MapField'COUNTER + typ MapSetTag = Proto.MapField'SET + typ MapRegisterTag = Proto.MapField'REGISTER + typ MapFlagTag = Proto.MapField'FLAG + typ MapMapTag = Proto.MapField'MAP diff --git a/src/Network/Riak/CRDT/Request.hs b/src/Network/Riak/CRDT/Request.hs --- a/src/Network/Riak/CRDT/Request.hs +++ b/src/Network/Riak/CRDT/Request.hs @@ -14,57 +14,37 @@ , mapUpdate ) where -import Data.ByteString.Lazy (ByteString) +import Data.ByteString (ByteString) +import qualified Data.Riak.Proto as Proto import Network.Riak.CRDT.Ops import qualified Network.Riak.CRDT.Types as CRDT -import 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.Lens import Network.Riak.Types counterUpdate :: [CRDT.CounterOp] -> BucketType -> Bucket -> Key - -> DtUpdate.DtUpdateRequest + -> Proto.DtUpdateReq counterUpdate ops = update (counterUpdateOp ops) setUpdate :: [CRDT.SetOp] -> BucketType -> Bucket -> Key - -> DtUpdate.DtUpdateRequest + -> Proto.DtUpdateReq setUpdate ops = update (setUpdateOp ops) mapUpdate :: [CRDT.MapOp] -> BucketType -> Bucket -> Key - -> DtUpdate.DtUpdateRequest + -> Proto.DtUpdateReq 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 - } +update :: Proto.DtOp -> BucketType -> Bucket -> Key -> Proto.DtUpdateReq +update op t b k = Proto.defMessage + & Proto.bucket .~ b + & Proto.key .~ k + & Proto.op .~ op + & Proto.type' .~ t -get :: ByteString -> ByteString -> ByteString -> 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 - } +get :: ByteString -> ByteString -> ByteString -> Proto.DtFetchReq +get t b k = Proto.defMessage + & Proto.bucket .~ b + & Proto.key .~ k + & Proto.type' .~ t diff --git a/src/Network/Riak/CRDT/Response.hs b/src/Network/Riak/CRDT/Response.hs --- a/src/Network/Riak/CRDT/Response.hs +++ b/src/Network/Riak/CRDT/Response.hs @@ -16,48 +16,48 @@ import Control.Applicative ((<$>)) import Data.Traversable #endif -import qualified Data.Sequence as Seq import qualified Data.Map as Map -import Data.Foldable (toList) import Data.Maybe (catMaybes) -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 qualified Data.Riak.Proto as Proto import Network.Riak.CRDT.Types as CRDT +import Network.Riak.Lens -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 +get :: Proto.DtFetchResp -> Maybe CRDT.DataType +get resp = case resp ^. Proto.type' of + Proto.DtFetchResp'COUNTER -> + DTCounter . Counter <$> ((^. Proto.maybe'counterValue) =<< (resp ^. Proto.maybe'value)) + Proto.DtFetchResp'SET -> + DTSet . setFromList . (^. Proto.setValue) <$> (resp ^. Proto.maybe'value) + Proto.DtFetchResp'MAP -> + DTMap . deconstructMap . (^. Proto.mapValue) <$> (resp ^. Proto.maybe'value) -deconstructMap :: Seq.Seq M.MapEntry -> Map -deconstructMap = Map . Map.fromList . catMaybes . map f . toList + -- We don't support hll or gset yet + Proto.DtFetchResp'HLL -> Nothing + Proto.DtFetchResp'GSET -> Nothing -f :: M.MapEntry -> Maybe (MapField, MapEntry) -f (M.MapEntry{..}) = sequenceA (MapField t name, v) - where M.MapField{..} = field +deconstructMap :: [Proto.MapEntry] -> Map +deconstructMap = Map . Map.fromList . catMaybes . map f + +f :: Proto.MapEntry -> Maybe (MapField, MapEntry) +f entry = sequenceA (MapField t (field ^. Proto.name), v) + where field :: Proto.MapField + field = entry ^. Proto.field + type' :: Proto.MapField'MapFieldType + type' = field ^. Proto.type' t :: MapEntryTag t = typeToTag type' v :: Maybe MapEntry v = case type' of - 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 + Proto.MapField'COUNTER -> MapCounter . Counter <$> (entry ^. Proto.maybe'counterValue) + Proto.MapField'SET -> Just . MapSet . setFromList $ (entry ^. Proto.setValue) + Proto.MapField'REGISTER -> MapRegister . Register <$> (entry ^. Proto.maybe'registerValue) + Proto.MapField'FLAG -> MapFlag . Flag <$> (entry ^. Proto.maybe'flagValue) + Proto.MapField'MAP -> Just . MapMap . deconstructMap $ (entry ^. Proto.mapValue) -typeToTag :: M.MapFieldType -> MapEntryTag -typeToTag M.COUNTER = MapCounterTag -typeToTag M.SET = MapSetTag -typeToTag M.REGISTER = MapRegisterTag -typeToTag M.FLAG = MapFlagTag -typeToTag M.MAP = MapMapTag +typeToTag :: Proto.MapField'MapFieldType -> MapEntryTag +typeToTag Proto.MapField'COUNTER = MapCounterTag +typeToTag Proto.MapField'SET = MapSetTag +typeToTag Proto.MapField'REGISTER = MapRegisterTag +typeToTag Proto.MapField'FLAG = MapFlagTag +typeToTag Proto.MapField'MAP = MapMapTag diff --git a/src/Network/Riak/CRDT/Riak.hs b/src/Network/Riak/CRDT/Riak.hs --- a/src/Network/Riak/CRDT/Riak.hs +++ b/src/Network/Riak/CRDT/Riak.hs @@ -28,13 +28,14 @@ import Control.Exception (catchJust) -import qualified Data.ByteString.Lazy as BS +import qualified Data.ByteString as BS +import qualified Data.Riak.Proto as Proto import qualified Network.Riak.CRDT.Request as Req import qualified Network.Riak.CRDT.Response as Resp import qualified Network.Riak.CRDT.Types as CRDT import qualified Network.Riak.Connection as Conn -import qualified Network.Riak.Protocol.ErrorResponse as ER +import Network.Riak.Lens import Network.Riak.Types @@ -66,10 +67,10 @@ -- NotPresent exception here handleEmpty :: IO () -> IO () handleEmpty act = catchJust - (\(e :: ER.ErrorResponse) -> + (\(e :: Proto.RpbErrorResp) -> if | "{precondition,{not_present," - `BS.isPrefixOf` ER.errmsg e -> Just () - | otherwise -> Nothing + `BS.isPrefixOf` (e ^. Proto.errmsg) -> Just () + | otherwise -> Nothing ) act pure diff --git a/src/Network/Riak/CRDT/Types.hs b/src/Network/Riak/CRDT/Types.hs --- a/src/Network/Riak/CRDT/Types.hs +++ b/src/Network/Riak/CRDT/Types.hs @@ -42,18 +42,16 @@ -- *** Modification , FlagOp(..) -- * Misc - , NonEmpty(..), mapEntryTag, setFromSeq, MapEntryTag(..) + , NonEmpty(..), mapEntryTag, setFromList, MapEntryTag(..) ) where import Control.DeepSeq (NFData) -import Data.ByteString.Lazy (ByteString) +import Data.ByteString (ByteString) import Data.Default.Class -import qualified Data.Foldable as F import Data.Int (Int64) import Data.List.NonEmpty import qualified Data.Map.Strict as M import Data.Semigroup -import qualified Data.Sequence as Seq import qualified Data.Set as S import Data.String (IsString(..)) import GHC.Generics (Generic) @@ -257,8 +255,8 @@ -- >>> SetRemove "bar" deriving (Eq,Show) -setFromSeq :: Seq.Seq ByteString -> Set -setFromSeq = Set . S.fromList . F.toList +setFromList :: [ByteString] -> Set +setFromList = Set . S.fromList -- | CRDT Counter hold a integer 'Count' -- 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 @@ -40,23 +40,25 @@ import Control.Concurrent.Async (async, waitBoth) import Control.Exception (Exception, IOException, throwIO, bracketOnError) import Control.Monad (forM_, replicateM) -import Data.Binary.Put (Put, putWord32be, runPut) +import Data.Binary.Get (Get, Decoder(..), getWord32be, runGetIncremental) +import Data.Binary.Put (Put, putWord32be, runPut, putLazyByteString) +import Data.ByteString (ByteString) import Data.IORef (newIORef, readIORef, writeIORef) -import Data.Int (Int64) import Network.Riak.Connection.NoPush (setNoPush) import Network.Riak.Debug as Debug -import Network.Riak.Protocol.ErrorResponse -import Network.Riak.Protocol.SetClientIDRequest +import Network.Riak.Lens import Network.Riak.Tag (getTag, putTag) import Network.Riak.Types.Internal hiding (MessageTag(..)) import Network.Socket as Socket import Numeric (showHex) +import Data.ProtoLens (buildMessage) import System.Random (randomIO) -import Text.ProtocolBuffers (messageGetM, messagePutM, messageSize) -import Text.ProtocolBuffers.Get (Get, Result(..), getWord32be, runGet) import qualified Control.Exception as E import qualified Data.ByteString as B +import qualified Data.ByteString.Char8 as B8 +import qualified Data.ByteString.Builder as BB import qualified Data.ByteString.Lazy.Char8 as L +import qualified Data.Riak.Proto as Proto import qualified Network.Riak.Types.Internal as T import qualified Network.Socket.ByteString as B import qualified Network.Socket.ByteString.Lazy as L @@ -67,26 +69,26 @@ defaultClient = Client { host = "127.0.0.1" , port = "8087" - , clientID = L.empty + , clientID = B.empty } -- | Tell the server our client ID. setClientID :: Connection -> ClientID -> IO () setClientID conn i = do - sendRequest conn $ SetClientIDRequest i + sendRequest conn $ (Proto.defMessage & Proto.clientId .~ i :: Proto.RpbSetClientIdReq) recvResponse_ conn T.SetClientIDResponse -- | Generate a random client ID. makeClientID :: IO ClientID makeClientID = do r <- randomIO :: IO Int - return . L.append "hs_" . L.pack . showHex (abs r) $ "" + return . B.append "hs_" . B8.pack . showHex (abs r) $ "" -- | Add a random 'ClientID' to a 'Client' if the 'Client' doesn't -- already have one. addClientID :: Client -> IO Client addClientID client - | L.null (clientID client) = do + | B.null (clientID client) = do i <- makeClientID return client { clientID = i } | otherwise = return client @@ -100,7 +102,7 @@ , addrSocketType = Stream } debug "connect" $ "server " ++ host ++ ":" ++ port ++ ", client ID " ++ - L.unpack clientID + B8.unpack clientID ais <- getAddrInfo (Just hints) (Just host) (Just port) let ai = case ais of (a:_) -> a @@ -112,7 +114,7 @@ close $ \sock -> do Socket.connect sock (addrAddress ai) - buf <- newIORef L.empty + buf <- newIORef B.empty let conn = Connection sock client buf setClientID conn clientID return conn @@ -121,9 +123,9 @@ disconnect :: Connection -> IO () disconnect Connection{..} = onIOException "disconnect" $ do debug "disconnect" $ "server " ++ host connClient ++ ":" ++ port connClient ++ - ", client ID " ++ L.unpack (clientID connClient) + ", client ID " ++ B8.unpack (clientID connClient) close connSock - writeIORef connBuffer L.empty + writeIORef connBuffer B.empty -- | We use a larger receive buffer than we usually need, and -- generally ask to receive more data than we know we'll need, in the @@ -133,27 +135,27 @@ recvBufferSize = 16384 {-# INLINE recvBufferSize #-} -recvExactly :: Connection -> Int64 -> IO L.ByteString +recvExactly :: Connection -> Int -> IO ByteString recvExactly Connection{..} n0 - | n0 <= 0 = return L.empty + | n0 <= 0 = return B.empty | otherwise = do bs <- readIORef connBuffer - let (h,t) = L.splitAt n0 bs - len = L.length h + let (h,t) = B.splitAt n0 bs + len = B.length h if len == n0 then writeIORef connBuffer t >> return h - else go (reverse (L.toChunks h)) (n0-len) + else go [h] (n0-len) where maxInt = fromIntegral (maxBound :: Int) go (s:acc) n | n < 0 = do let (h,t) = B.splitAt (B.length s + fromIntegral n) s - writeIORef connBuffer $! L.fromChunks [t] - return $ L.fromChunks (reverse (h:acc)) + writeIORef connBuffer $! t + return $ B.concat (reverse (h:acc)) go acc n | n == 0 = do - writeIORef connBuffer L.empty - return $ L.fromChunks (reverse acc) + writeIORef connBuffer B.empty + return $ B.concat (reverse acc) | otherwise = do let n' = max recvBufferSize $ min n maxInt bs <- B.recv connSock (fromIntegral n') @@ -164,51 +166,55 @@ recvGet :: Connection -> Get a -> IO a recvGet Connection{..} get = do - let refill = do - bs <- L.recv connSock recvBufferSize - if L.null bs + let refill :: IO (Maybe ByteString) + refill = do + bs <- B.recv connSock recvBufferSize + if B.null bs then shutdown connSock ShutdownReceive >> return Nothing else return (Just bs) - step (Failed _ err) = moduleError "recvGet" err - step (Finished bs _ r) = writeIORef connBuffer bs >> return r - step (Partial k) = (step . k) =<< refill + -- step :: Decoder a -> IO a + step (Fail _ _ err) = moduleError "recvGet" err + step (Done bs _ r) = writeIORef connBuffer bs >> return r + step (Partial k) = (step . k) =<< refill mbs <- do buf <- readIORef connBuffer - if L.null buf + if B.null buf then refill else return (Just buf) case mbs of - Just bs -> step $ runGet get bs + Just bs -> + case runGetIncremental get of + Fail _ _ err -> moduleError "recvGet" err + Done bs' _ r -> writeIORef connBuffer bs' >> return r + Partial k -> step (k (Just bs)) Nothing -> moduleError "recvGet" "socket closed" -recvGetN :: Connection -> Int64 -> Get a -> IO a -recvGetN conn n get = do +recvGetN :: Proto.Message a => Connection -> Int -> IO a +recvGetN conn n = do bs <- recvExactly conn n - case runGet get bs of - Finished _ _ r -> return r - Partial k -> case k Nothing of - Finished _ _ r -> return r - Failed _ err -> moduleError "recvGetN" err - Partial _ -> moduleError "recvGetN" - "parser wants more input!?" - Failed _ err -> moduleError "recvGetN" err + case Proto.decodeMessage bs of + Left err -> moduleError "recvGetN" err + Right r -> return r putRequest :: (Request req) => req -> Put putRequest req = do - putWord32be (fromIntegral (1 + messageSize req)) + putWord32be (fromIntegral (1 + L.length bytes)) putTag (messageTag req) - messagePutM req + putLazyByteString bytes + where + bytes :: L.ByteString + bytes = BB.toLazyByteString (buildMessage req) -instance Exception ErrorResponse +instance Exception Proto.RpbErrorResp -throwError :: ErrorResponse -> IO a +throwError :: Proto.RpbErrorResp -> IO a throwError = throwIO -getResponse :: Response a => Connection -> Int64 -> a -> T.MessageTag -> IO a +getResponse :: Response a => Connection -> Int -> a -> T.MessageTag -> IO a getResponse conn len _ expected = do tag <- recvGet conn getTag - if | tag == expected -> recvGetN conn (len-1) messageGetM - | tag == T.ErrorResponse -> throwError =<< recvGetN conn (len-1) messageGetM + if | tag == expected -> recvGetN conn (len-1) + | tag == T.ErrorResponse -> throwError =<< recvGetN conn (len-1) | otherwise -> moduleError "getResponse" $ "received unexpected response: expected " ++ show expected ++ ", received " ++ show tag @@ -271,11 +277,11 @@ then recvCorrectTag "recvMaybeResponse" conn tag 1 Nothing else Just `fmap` getResponse conn len dummy tag -recvCorrectTag :: String -> Connection -> T.MessageTag -> Int64 -> a -> IO a +recvCorrectTag :: String -> Connection -> T.MessageTag -> Int -> a -> IO a recvCorrectTag func conn expected len v = do tag <- recvGet conn getTag if | tag == expected -> recvExactly conn (len-1) >> return v - | tag == T.ErrorResponse -> throwError =<< recvGetN conn len messageGetM + | tag == T.ErrorResponse -> throwError =<< recvGetN conn len | otherwise -> moduleError func $ "received unexpected response: expected " ++ show expected ++ ", received " ++ show tag diff --git a/src/Network/Riak/Content.hs b/src/Network/Riak/Content.hs --- a/src/Network/Riak/Content.hs +++ b/src/Network/Riak/Content.hs @@ -13,8 +13,8 @@ module Network.Riak.Content ( -- * Types - Content(..) - , Link.Link(..) + Proto.RpbContent + , Proto.RpbLink -- * Functions , empty , binary @@ -24,40 +24,30 @@ import Data.Aeson (encode) import Data.Aeson.Types (ToJSON) -import Network.Riak.Protocol.Content (Content(..)) +import Data.ByteString (ByteString) +import qualified Data.ByteString.Lazy as LazyByteString +import qualified Data.Riak.Proto as Proto +import Network.Riak.Lens import Network.Riak.Types.Internal (Bucket, Key, Tag) -import qualified Data.ByteString.Lazy.Char8 as L -import qualified Data.Sequence as Seq -import qualified Network.Riak.Protocol.Link as Link -- | Create a link. -link :: Bucket -> Key -> Tag -> Link.Link -link bucket key tag = Link.Link (Just bucket) (Just key) (Just tag) +link :: Bucket -> Key -> Tag -> Proto.RpbLink +link bucket key tag = Proto.defMessage + & Proto.bucket .~ bucket + & Proto.key .~ key + & Proto.tag .~ tag {-# INLINE link #-} -- | An empty piece of content. -empty :: Content -empty = Content { value = L.empty - , content_type = Nothing - , charset = Nothing - , content_encoding = Nothing - , vtag = Nothing - , links = Seq.empty - , last_mod = Nothing - , last_mod_usecs = Nothing - , usermeta = Seq.empty - , indexes = Seq.empty - , deleted = Nothing - } +empty :: Proto.RpbContent +empty = Proto.defMessage -- | Content encoded as @application/octet-stream@. -binary :: L.ByteString -> Content -binary bs = empty { value = bs - , content_type = Just "application/octet-stream" - } +binary :: ByteString -> Proto.RpbContent +binary bs = empty & Proto.value .~ bs + & Proto.contentType .~ "application/octet-stream" -- | Content encoded as @application/json@. -json :: ToJSON a => a -> Content -json j = empty { value = encode j - , content_type = Just "application/json" - } +json :: ToJSON a => a -> Proto.RpbContent +json j = empty & Proto.value .~ LazyByteString.toStrict (encode j) + & Proto.contentType .~ "application/json" diff --git a/src/Network/Riak/Escape.hs b/src/Network/Riak/Escape.hs --- a/src/Network/Riak/Escape.hs +++ b/src/Network/Riak/Escape.hs @@ -28,7 +28,6 @@ import Control.Applicative ((<$>)) #endif import Data.Attoparsec.ByteString as A -import Data.Attoparsec.Lazy as AL import Data.Bits ((.|.), (.&.), shiftL, shiftR) import Data.ByteString (ByteString) #if __GLASGOW_HASKELL__ < 710 @@ -49,45 +48,41 @@ -- unescaped. class Escape e where -- | URL-escape a string. - escape :: e -> L.ByteString + escape :: e -> ByteString -- | URL-unescape a string. - unescape' :: L.ByteString -> Either String e + unescape' :: ByteString -> Either String e -- | URL-unescape a string that is presumed to be properly escaped. -- If the string is invalid, an error will be thrown that cannot be -- caught from pure code. -unescape :: Escape e => L.ByteString -> e +unescape :: Escape e => ByteString -> e unescape bs = case unescape' bs of Left err -> error $ "Network.Riak.Escape.unescape: " ++ err Right v -> v {-# INLINE unescape #-} instance Escape ByteString where - escape = toLazyByteString . B.foldl escapeWord8 mempty + escape = L.toStrict . toLazyByteString . B.foldl escapeWord8 mempty {-# INLINE escape #-} - unescape' = AL.eitherResult . AL.parse (toByteString <$> unescapeBS) + unescape' = A.eitherResult . A.parse (toByteString <$> unescapeBS) {-# INLINE unescape' #-} instance Escape L.ByteString where - escape = toLazyByteString . L.foldl escapeWord8 mempty + escape = L.toStrict . toLazyByteString . L.foldl escapeWord8 mempty {-# INLINE escape #-} - unescape' = AL.eitherResult . AL.parse (toLazyByteString <$> unescapeBS) + unescape' = A.eitherResult . A.parse (toLazyByteString <$> unescapeBS) {-# INLINE unescape' #-} instance Escape Text where escape = escape . T.encodeUtf8 {-# INLINE escape #-} - unescape' lbs = case AL.parse (toByteString <$> unescapeBS) lbs of - AL.Done _ bs -> first show $ T.decodeUtf8' bs - AL.Fail _ _ err -> Left err + unescape' = (>>= first show . T.decodeUtf8') . A.eitherResult . A.parse (toByteString <$> unescapeBS) {-# INLINE unescape' #-} instance Escape TL.Text where escape = escape . TL.encodeUtf8 {-# INLINE escape #-} - unescape' lbs = case AL.parse (toLazyByteString <$> unescapeBS) lbs of - AL.Done _ bs -> first show $ TL.decodeUtf8' bs - AL.Fail _ _ err -> Left err + unescape' = (>>= first show . TL.decodeUtf8') . A.eitherResult . A.parse (toLazyByteString <$> unescapeBS) {-# INLINE unescape' #-} instance Escape [Char] where diff --git a/src/Network/Riak/Lens.hs b/src/Network/Riak/Lens.hs new file mode 100644 --- /dev/null +++ b/src/Network/Riak/Lens.hs @@ -0,0 +1,39 @@ +{-# LANGUAGE RankNTypes #-} + +module Network.Riak.Lens + ( (&) + , (^.) + , (.~) + , (%~) + , mapped + ) where + +import Data.Function ((&)) +import Data.Functor.Const +import Data.Functor.Identity + +type Lens s a + = forall f. Functor f => (a -> f a) -> (s -> f s) + +type Setter s a + = (a -> Identity a) -> (s -> Identity s) + +infixl 8 ^. +(^.) :: s -> Lens s a -> a +s ^. l = + getConst (l Const s) + +infixr 4 .~ +(.~) :: Setter s a -> a -> s -> s +(l .~ a) s = + runIdentity (l (\_ -> Identity a) s) + +infixr 4 %~ +(%~) :: Setter s a -> (a -> a) -> (s -> s) +(l %~ f) s = + runIdentity (l (\a -> Identity (f a)) s) + + +mapped :: Functor f => Setter (f a) a +mapped f s = + Identity (fmap (runIdentity . f) s) 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 @@ -1,4 +1,5 @@ -{-# LANGUAGE OverloadedStrings, CPP #-} +{-# LANGUAGE CPP #-} +{-# LANGUAGE OverloadedStrings #-} -- | -- Module: Network.Riak.Request @@ -11,82 +12,68 @@ -- Smart constructors for Riak types. These functions correctly -- URL-escape bucket, key, and link names. You should thus use them -- in preference to the raw data constructors. - module Network.Riak.Request - ( - -- * Connection management - PingRequest - , ping - , GetClientIDRequest - , getClientID - , GetServerInfoRequest - , getServerInfo + ( -- * Connection management + Proto.RpbPingReq, + ping, + Proto.RpbGetClientIdReq, + getClientID, + Proto.RpbGetServerInfoReq, + getServerInfo, + -- * Data management - , Get.GetRequest - , get - , getByIndex - , Index.IndexRequest - , Put.PutRequest - , put - , Del.DeleteRequest - , delete + Proto.RpbGetReq, + get, + getByIndex, + Proto.RpbIndexReq, + Proto.RpbPutReq, + put, + Proto.RpbDelReq, + delete, + -- * Metadata - , Link.Link - , link - , ListBucketsRequest - , listBuckets - , Keys.ListKeysRequest - , listKeys - , GetBucket.GetBucketRequest - , getBucket - , SetBucket.SetBucketRequest - , setBucket - , getBucketType + Proto.RpbLink, + link, + Proto.RpbListBucketsReq, + listBuckets, + Proto.RpbListKeysReq, + listKeys, + Proto.RpbGetBucketReq, + getBucket, + Proto.RpbSetBucketReq, + setBucket, + getBucketType, + -- * Map/reduce - , MapReduceRequest - , mapReduce + Proto.RpbMapRedReq, + mapReduce, + -- * Search - , search - , getIndex - , putIndex - , deleteIndex - ) where + search, + getIndex, + putIndex, + deleteIndex, + ) +where #if __GLASGOW_HASKELL__ < 710 import Control.Applicative ((<$>)) #endif +import Data.ByteString (ByteString) import qualified Data.ByteString.Char8 as B8 #if __GLASGOW_HASKELL__ < 804 import Data.Monoid #endif -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 qualified Data.Riak.Proto as Proto import Network.Riak.Escape (escape) -import qualified Network.Riak.Protocol.DeleteRequest as Del -import qualified Network.Riak.Protocol.Link as Link -import qualified Network.Riak.Protocol.GetBucketRequest as GetBucket -import qualified Network.Riak.Protocol.GetRequest as Get -import qualified Network.Riak.Protocol.IndexRequest as Index -import qualified Network.Riak.Protocol.IndexRequest.IndexQueryType as IndexQueryType -import qualified Network.Riak.Protocol.ListKeysRequest as Keys -import qualified Network.Riak.Protocol.PutRequest as Put -import qualified Network.Riak.Protocol.SetBucketRequest as SetBucket -import qualified Network.Riak.Protocol.GetBucketTypeRequest as GetBucketType -import qualified Network.Riak.Protocol.SearchQueryRequest as SearchQueryRequest -import qualified Network.Riak.Protocol.YzIndexGetRequest as YzIndex -import qualified Network.Riak.Protocol.YzIndexPutRequest as YzIndex -import qualified Network.Riak.Protocol.YzIndexDeleteRequest as YzIndex +import Network.Riak.Lens +import Network.Riak.Types.Internal hiding (MessageTag (..)) -- | Create a ping request. -ping :: PingRequest -ping = PingRequest +ping :: Proto.RpbPingReq +ping = Proto.defMessage {-# INLINE ping #-} + {- -- | Create a dtUpdate request. dtUpdate :: DtUpdate.DtUpdateRequest @@ -95,171 +82,180 @@ -} -- | Create a client-ID request. -getClientID :: GetClientIDRequest -getClientID = GetClientIDRequest +getClientID :: Proto.RpbGetClientIdReq +getClientID = Proto.defMessage {-# INLINE getClientID #-} -- | Create a server-info request. -getServerInfo :: GetServerInfoRequest -getServerInfo = GetServerInfoRequest +getServerInfo :: Proto.RpbGetServerInfoReq +getServerInfo = Proto.defMessage {-# INLINE getServerInfo #-} -- | Create a get request. The bucket and key names are URL-escaped. -get :: Maybe BucketType -> Bucket -> Key -> R -> Get.GetRequest -get btype bucket key r = Get.GetRequest { - Get.bucket = escape bucket - , Get.key = escape key - , Get.r = fromQuorum r - , Get.pr = Nothing - , Get.basic_quorum = Nothing - , Get.notfound_ok = Nothing - , Get.if_modified = Nothing - , Get.head = Nothing - , Get.deletedvclock = Nothing - , Get.timeout = Nothing - , Get.sloppy_quorum = Nothing - , Get.n_val = Nothing - , Get.type' = btype -- no escaping intentional - -- TODO don't escape anything - } +get :: Maybe BucketType -> Bucket -> Key -> R -> Proto.RpbGetReq +get btype bucket key r = + Proto.defMessage + & Proto.bucket .~ escape bucket + & Proto.key .~ escape key + & Proto.maybe'r .~ fromQuorum r + & Proto.maybe'type' .~ btype -- no escaping intentional + -- TODO don't escape anything {-# INLINE get #-} -- | Create a secondary index request. Bucket, key and index names and -- values are URL-escaped. -getByIndex :: Bucket -> IndexQuery - -> Index.IndexRequest +getByIndex :: + Bucket -> + IndexQuery -> + Proto.RpbIndexReq getByIndex bucket q = - case q of - IndexQueryExactInt index key -> - req (index <> "_int") (showIntKey key) - IndexQueryType.Eq Nothing Nothing - IndexQueryExactBin index key -> - req (index <> "_bin") (showBsKey $ key) - IndexQueryType.Eq Nothing Nothing - IndexQueryRangeInt index from to -> - req (index <> "_int") Nothing - IndexQueryType.Range (showIntKey from) (showIntKey to) - IndexQueryRangeBin index from to -> - req (index <> "_bin") Nothing - IndexQueryType.Range (showBsKey from) (showBsKey to) + case q of + IndexQueryExactInt index key -> + req + (index <> "_int") + (showIntKey key) + Proto.RpbIndexReq'Eq + Nothing + Nothing + IndexQueryExactBin index key -> + req + (index <> "_bin") + (showBsKey $ key) + Proto.RpbIndexReq'Eq + Nothing + Nothing + IndexQueryRangeInt index from to -> + req + (index <> "_int") + Nothing + Proto.RpbIndexReq'Range + (showIntKey from) + (showIntKey to) + IndexQueryRangeBin index from to -> + req + (index <> "_bin") + Nothing + Proto.RpbIndexReq'Range + (showBsKey from) + (showBsKey to) where showIntKey = Just . escape . B8.pack . show showBsKey = Just . escape + req :: + ByteString -> + Maybe ByteString -> + Proto.RpbIndexReq'IndexQueryType -> + Maybe ByteString -> + Maybe ByteString -> + Proto.RpbIndexReq req i k qt rmin rmax = - Index.IndexRequest { Index.bucket = escape bucket - , Index.index = escape i - , Index.qtype = qt - , Index.key = k - , Index.range_min = rmin - , Index.range_max = rmax - , Index.return_terms = Nothing - , Index.stream = Nothing - , Index.max_results = Nothing - , Index.continuation = Nothing - , Index.timeout = Nothing - , Index.type' = Nothing - , Index.term_regex = Nothing - , Index.pagination_sort = Nothing - } + Proto.defMessage + & Proto.bucket .~ escape bucket + & Proto.index .~ escape i + & Proto.qtype .~ qt + & Proto.maybe'key .~ k + & Proto.maybe'rangeMin .~ rmin + & Proto.maybe'rangeMax .~ rmax -- | Create a put request. The bucket and key names are URL-escaped. -- Any 'Link' values inside the 'Content' are assumed to have been -- constructed with the 'link' function, and hence /not/ escaped. -put :: Maybe BucketType -> Bucket -> Key -> Maybe VClock -> Content -> W -> DW -> Bool - -> Put.PutRequest +put :: + Maybe BucketType -> + Bucket -> + Key -> + Maybe VClock -> + Proto.RpbContent -> + W -> + DW -> + Bool -> + Proto.RpbPutReq put btype bucket key mvclock cont mw mdw returnBody = - Put.PutRequest { Put.bucket = escape bucket, - Put.key = Just $ escape key, - Put.vclock = fromVClock <$> mvclock, - Put.content = cont, - Put.w = fromQuorum mw, - Put.dw = fromQuorum mdw, - Put.return_body = Just returnBody, - Put.pw = Nothing, - Put.if_not_modified = Nothing, - Put.if_none_match = Nothing, - Put.return_head = Nothing, - Put.timeout = Nothing, - Put.asis = Nothing, - Put.sloppy_quorum = Nothing, - Put.n_val = Nothing, - Put.type' = btype -- same as get - } + Proto.defMessage + & Proto.bucket .~ escape bucket + & Proto.key .~ escape key + & Proto.maybe'vclock .~ (fromVClock <$> mvclock) + & Proto.content .~ cont + & Proto.maybe'w .~ fromQuorum mw + & Proto.maybe'dw .~ fromQuorum mdw + & Proto.returnBody .~ returnBody + & Proto.maybe'type' .~ btype -- same as get {-# INLINE put #-} -- | Create a link. The bucket and key names are URL-escaped. -link :: Bucket -> Key -> Tag -> Link.Link -link bucket key = Link.Link (Just (escape bucket)) (Just (escape key)) . Just +link :: Bucket -> Key -> Tag -> Proto.RpbLink +link bucket key tag = + Proto.defMessage + & Proto.bucket .~ escape bucket + & Proto.key .~ escape key + & Proto.tag .~ tag {-# INLINE link #-} -- | Create a delete request. The bucket and key names are URL-escaped. -delete :: Maybe BucketType -> Bucket -> Key -> RW -> Del.DeleteRequest -delete btype bucket key rw = Del.DeleteRequest { - Del.bucket = escape bucket, - Del.key = escape key, - Del.rw = fromQuorum rw, - Del.vclock = Nothing, - Del.r = Nothing, - Del.w = Nothing, - Del.pr = Nothing, - Del.pw = Nothing, - Del.dw = Nothing, - Del.timeout = Nothing, - Del.sloppy_quorum = Nothing, - Del.n_val = Nothing, - Del.type' = btype -- same as get - } +delete :: Maybe BucketType -> Bucket -> Key -> RW -> Proto.RpbDelReq +delete btype bucket key rw = + Proto.defMessage + & Proto.bucket .~ escape bucket + & Proto.key .~ escape key + & Proto.maybe'rw .~ fromQuorum rw + & Proto.maybe'type' .~ btype -- same as get {-# INLINE delete #-} -- | Create a list-buckets request. -listBuckets :: Maybe BucketType -> ListBucketsRequest -listBuckets = ListBucketsRequest Nothing Nothing +listBuckets :: Maybe BucketType -> Proto.RpbListBucketsReq +listBuckets btype = Proto.defMessage & Proto.maybe'type' .~ btype {-# INLINE listBuckets #-} -- | Create a list-keys request. The bucket type and name are URL-escaped. -listKeys :: Maybe BucketType -> Bucket -> Keys.ListKeysRequest -listKeys t b = Keys.ListKeysRequest (escape b) Nothing (escape <$> t) +listKeys :: Maybe BucketType -> Bucket -> Proto.RpbListKeysReq +listKeys t b = + Proto.defMessage & Proto.maybe'type' .~ (escape <$> t) + & Proto.bucket .~ escape b {-# INLINE listKeys #-} -- | Create a get-bucket request. The bucket type and name are URL-escaped. -getBucket :: Maybe BucketType -> Bucket -> GetBucket.GetBucketRequest -getBucket t b = GetBucket.GetBucketRequest (escape b) (escape <$> t) +getBucket :: Maybe BucketType -> Bucket -> Proto.RpbGetBucketReq +getBucket t b = + Proto.defMessage & Proto.maybe'type' .~ (escape <$> t) + & Proto.bucket .~ escape b {-# INLINE getBucket #-} -- | Create a set-bucket request. The bucket type and name are URL-escaped. -setBucket :: Maybe BucketType -> Bucket -> BucketProps -> SetBucket.SetBucketRequest -setBucket t b ps = SetBucket.SetBucketRequest (escape b) ps (escape <$> t) +setBucket :: Maybe BucketType -> Bucket -> Proto.RpbBucketProps -> Proto.RpbSetBucketReq +setBucket t b ps = + Proto.defMessage & Proto.bucket .~ escape b + & Proto.maybe'type' .~ (escape <$> t) + & Proto.props .~ ps {-# INLINE setBucket #-} -- | Create a get-bucket-type request. The bucket type is URL-escaped. -getBucketType :: BucketType -> GetBucketType.GetBucketTypeRequest -getBucketType t = GetBucketType.GetBucketTypeRequest (escape t) +getBucketType :: BucketType -> Proto.RpbGetBucketTypeReq +getBucketType t = Proto.defMessage & Proto.type' .~ escape t -- | Create a map-reduce request. -mapReduce :: Job -> MapReduceRequest -mapReduce (JSON bs) = MapReduceRequest bs "application/json" -mapReduce (Erlang bs) = MapReduceRequest bs "application/x-erlang-binary" +mapReduce :: Job -> Proto.RpbMapRedReq +mapReduce (JSON bs) = + Proto.defMessage & Proto.request .~ bs + & Proto.contentType .~ "application/json" +mapReduce (Erlang bs) = + Proto.defMessage & Proto.request .~ bs + & Proto.contentType .~ "application/x-erlang-binary" -- | Create a search request -search :: SearchQuery -> Index -> 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 - } +search :: SearchQuery -> Index -> Proto.RpbSearchQueryReq +search q ix = + Proto.defMessage + & Proto.q .~ q + & Proto.index .~ escape ix -getIndex :: Maybe Index -> YzIndex.YzIndexGetRequest -getIndex = YzIndex.YzIndexGetRequest +getIndex :: Maybe Index -> Proto.RpbYokozunaIndexGetReq +getIndex ix = Proto.defMessage & Proto.maybe'name .~ ix -putIndex :: IndexInfo -> Maybe Timeout -> YzIndex.YzIndexPutRequest -putIndex = YzIndex.YzIndexPutRequest +putIndex :: IndexInfo -> Maybe Timeout -> Proto.RpbYokozunaIndexPutReq +putIndex info timeout = + Proto.defMessage + & Proto.index .~ info + & Proto.maybe'timeout .~ timeout -deleteIndex :: Index -> YzIndex.YzIndexDeleteRequest -deleteIndex = YzIndex.YzIndexDeleteRequest +deleteIndex :: Index -> Proto.RpbYokozunaIndexDeleteReq +deleteIndex ix = Proto.defMessage & Proto.name .~ ix 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 @@ -34,71 +34,64 @@ import Control.Arrow ((&&&)) import Control.Monad (join) #endif +import qualified Data.Riak.Proto as Proto import Network.Riak.Escape (unescape) -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.ListBucketsResponse -import Network.Riak.Protocol.PutResponse -import qualified Network.Riak.Protocol.SearchQueryResponse as Q -import qualified Network.Riak.Protocol.SearchDoc as Q -import qualified Network.Riak.Protocol.YzIndexGetResponse as Yz +import Network.Riak.Lens 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.Sequence as Seq -import Data.Maybe (fromMaybe) +import Data.ByteString (ByteString) import Data.Foldable (toList) -getClientID :: GetClientIDResponse -> ClientID -getClientID = client_id +getClientID :: Proto.RpbGetClientIdResp -> ClientID +getClientID = (^. Proto.clientId) {-# INLINE getClientID #-} -- | Construct a get response. Bucket and key names in links are -- URL-unescaped. -get :: Maybe GetResponse -> Maybe (Seq.Seq Content, VClock) -get (Just (GetResponse content (Just vclock) _)) - = Just (unescapeLinks <$> content, VClock vclock) -get _ = Nothing +get :: Maybe Proto.RpbGetResp -> Maybe ([Proto.RpbContent], VClock) +get mresponse = do + response <- mresponse + let content = response ^. Proto.content + vclock <- response ^. Proto.maybe'vclock + Just (unescapeLinks <$> content, VClock vclock) {-# INLINE get #-} -- | Construct a put response. Bucket and key names in links are -- URL-unescaped. -put :: PutResponse -> (Seq.Seq Content, VClock) -put PutResponse{..} = (unescapeLinks <$> content, - VClock (fromMaybe L.empty vclock)) +put :: Proto.RpbPutResp -> ([Proto.RpbContent], VClock) +put response = + ( unescapeLinks <$> (response ^. Proto.content) + , VClock (response ^. Proto.vclock) + ) {-# INLINE put #-} -- | Construct a list-buckets response. Bucket names are unescaped. -listBuckets :: ListBucketsResponse -> Seq.Seq Bucket -listBuckets = fmap unescape . buckets +listBuckets :: Proto.RpbListBucketsResp -> [Bucket] +listBuckets = fmap unescape . (^. Proto.buckets) {-# INLINE listBuckets #-} -getBucket :: GetBucketResponse -> BucketProps -getBucket = props +getBucket :: Proto.RpbGetBucketResp -> Proto.RpbBucketProps +getBucket = (^. Proto.props) {-# INLINE getBucket #-} -- | URL-unescape the names of keys and buckets in the links of a -- 'Content' value. -unescapeLinks :: Content -> Content -unescapeLinks c = c { links = go <$> links c } - where go l = l { Link.bucket = unescape <$> Link.bucket l - , Link.key = unescape <$> Link.key l } +unescapeLinks :: Proto.RpbContent -> Proto.RpbContent +unescapeLinks = Proto.links . mapped %~ go + where go :: Proto.RpbLink -> Proto.RpbLink + go l = l & Proto.bucket %~ unescape + & Proto.key %~ unescape -search :: Q.SearchQueryResponse -> SearchResult +search :: Proto.RpbSearchQueryResp -> SearchResult search resp = SearchResult - { docs = fmap (fmap unpair . Q.fields) (Q.docs resp) - , maxScore = Q.max_score resp - , numFound = Q.num_found resp + { docs = fmap (fmap unpair . (^. Proto.fields)) (resp ^. Proto.docs) + , maxScore = resp ^. Proto.maybe'maxScore + , numFound = resp ^. Proto.maybe'numFound } where - unpair :: Pair.Pair -> (L.ByteString, Maybe L.ByteString) - unpair Pair.Pair{Pair.key, Pair.value} = (key, value) + unpair :: Proto.RpbPair -> (ByteString, Maybe ByteString) + unpair pair = (pair ^. Proto.key, pair ^. Proto.maybe'value) -getIndex :: Yz.YzIndexGetResponse -> [IndexInfo] -getIndex = toList . Yz.index +getIndex :: Proto.RpbYokozunaIndexGetResp -> [IndexInfo] +getIndex = toList . (^. Proto.index) diff --git a/src/Network/Riak/Search.hs b/src/Network/Riak/Search.hs --- a/src/Network/Riak/Search.hs +++ b/src/Network/Riak/Search.hs @@ -21,10 +21,9 @@ #if __GLASGOW_HASKELL__ <= 708 import Control.Applicative #endif -import Data.Sequence (Seq) +import qualified Data.Riak.Proto as Proto import Network.Riak.Connection.Internal -import Network.Riak.Protocol.Content (Content) -import Network.Riak.Protocol.YzIndex (YzIndex(YzIndex)) +import Network.Riak.Lens import qualified Network.Riak.Request as Req import qualified Network.Riak.Response as Resp import Network.Riak.Types.Internal @@ -37,7 +36,9 @@ -- explicitly set the property. In the default installation of @riak@, this is -- 3 (see https://github.com/basho/riak_core/blob/develop/priv/riak_core.schema). indexInfo :: Index -> Maybe Schema -> Maybe N -> IndexInfo -indexInfo = YzIndex +indexInfo ix schema nval = Proto.defMessage & Proto.name .~ ix + & Proto.maybe'schema .~ schema + & Proto.maybe'nVal .~ nval -- | Get an index info for @Just index@, or get all indexes for @Nothing@. -- @@ -48,7 +49,7 @@ -- | Create a new index or modify an existing index. -- -- https://docs.basho.com/riak/kv/2.1.4/developing/api/protocol-buffers/yz-index-put/ -putIndex :: Connection -> IndexInfo -> Maybe Timeout -> IO (Seq Content, VClock) +putIndex :: Connection -> IndexInfo -> Maybe Timeout -> IO ([Proto.RpbContent], VClock) putIndex conn info timeout = Resp.put <$> exchange conn (Req.putIndex info timeout) -- | Delete an index. 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,312 +18,279 @@ , getTag ) where +import Data.Binary.Get (Get, getWord8) import Data.Binary.Put (Put, putWord8) import Data.Word (Word8) import qualified Data.HashMap.Strict as HM #if __GLASGOW_HASKELL__ <= 708 import Control.Applicative #endif +import Data.Riak.Proto import Data.Tuple (swap) -import Network.Riak.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 -import Network.Riak.Protocol.GetRequest -import Network.Riak.Protocol.GetResponse -import Network.Riak.Protocol.IndexRequest -import Network.Riak.Protocol.IndexResponse -import Network.Riak.Protocol.GetServerInfoRequest -import Network.Riak.Protocol.ListBucketsRequest -import Network.Riak.Protocol.ListBucketsResponse -import Network.Riak.Protocol.ListKeysRequest -import Network.Riak.Protocol.ListKeysResponse -import Network.Riak.Protocol.MapReduce -import Network.Riak.Protocol.MapReduceRequest -import Network.Riak.Protocol.PingRequest -import Network.Riak.Protocol.PutRequest -import Network.Riak.Protocol.PutResponse -import Network.Riak.Protocol.ServerInfo -import Network.Riak.Protocol.SetBucketRequest -import Network.Riak.Protocol.SetClientIDRequest -import 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.Protocol.YzIndexPutRequest -import Network.Riak.Protocol.YzIndexDeleteRequest import Network.Riak.Types.Internal as Types -import Text.ProtocolBuffers.Get (Get, getWord8) -instance Tagged ErrorResponse where +instance Tagged RpbErrorResp where messageTag _ = Types.ErrorResponse {-# INLINE messageTag #-} -instance Response ErrorResponse +instance Response RpbErrorResp -instance Tagged PingRequest where +instance Tagged RpbPingReq where messageTag _ = Types.PingRequest {-# INLINE messageTag #-} -instance Request PingRequest where +instance Request RpbPingReq where expectedResponse _ = Types.PingResponse {-# INLINE expectedResponse #-} -instance Tagged GetClientIDRequest where +instance Tagged RpbGetClientIdReq where messageTag _ = Types.GetClientIDRequest {-# INLINE messageTag #-} -instance Request GetClientIDRequest where +instance Request RpbGetClientIdReq where expectedResponse _ = Types.GetClientIDResponse {-# INLINE expectedResponse #-} -instance Tagged GetClientIDResponse where +instance Tagged RpbGetClientIdResp where messageTag _ = Types.GetClientIDResponse {-# INLINE messageTag #-} -instance Response GetClientIDResponse +instance Response RpbGetClientIdResp -instance Exchange GetClientIDRequest GetClientIDResponse +instance Exchange RpbGetClientIdReq RpbGetClientIdResp -instance Tagged SetClientIDRequest where +instance Tagged RpbSetClientIdReq where messageTag _ = Types.SetClientIDRequest {-# INLINE messageTag #-} -instance Request SetClientIDRequest where +instance Request RpbSetClientIdReq where expectedResponse _ = Types.SetClientIDResponse {-# INLINE expectedResponse #-} -instance Tagged GetServerInfoRequest where +instance Tagged RpbGetServerInfoReq where messageTag _ = Types.GetServerInfoRequest {-# INLINE messageTag #-} -instance Request GetServerInfoRequest where +instance Request RpbGetServerInfoReq where expectedResponse _ = Types.GetServerInfoResponse {-# INLINE expectedResponse #-} -instance Tagged ServerInfo where +instance Tagged RpbGetServerInfoResp where messageTag _ = Types.GetServerInfoResponse {-# INLINE messageTag #-} -instance Response ServerInfo +instance Response RpbGetServerInfoResp -instance Exchange GetServerInfoRequest ServerInfo +instance Exchange RpbGetServerInfoReq RpbGetServerInfoResp -instance Tagged GetRequest where +instance Tagged RpbGetReq where messageTag _ = Types.GetRequest {-# INLINE messageTag #-} -instance Tagged IndexRequest where +instance Tagged RpbIndexReq where messageTag _ = Types.IndexRequest {-# INLINE messageTag #-} -instance Request GetRequest where +instance Request RpbGetReq where expectedResponse _ = Types.GetResponse {-# INLINE expectedResponse #-} -instance Request IndexRequest where +instance Request RpbIndexReq where expectedResponse _ = Types.IndexResponse {-# INLINE expectedResponse #-} -instance Tagged GetResponse where +instance Tagged RpbGetResp where messageTag _ = Types.GetResponse {-# INLINE messageTag #-} -instance Tagged IndexResponse where +instance Tagged RpbIndexResp where messageTag _ = Types.IndexResponse {-# INLINE messageTag #-} -instance Response GetResponse +instance Response RpbGetResp -instance Response IndexResponse +instance Response RpbIndexResp -instance Exchange GetRequest GetResponse +instance Exchange RpbGetReq RpbGetResp -instance Exchange IndexRequest IndexResponse +instance Exchange RpbIndexReq RpbIndexResp -instance Tagged PutRequest where +instance Tagged RpbPutReq where messageTag _ = Types.PutRequest {-# INLINE messageTag #-} -instance Request PutRequest where +instance Request RpbPutReq where expectedResponse _ = Types.PutResponse {-# INLINE expectedResponse #-} -instance Tagged PutResponse where +instance Tagged RpbPutResp where messageTag _ = Types.PutResponse {-# INLINE messageTag #-} -instance Response PutResponse +instance Response RpbPutResp -instance Exchange PutRequest PutResponse +instance Exchange RpbPutReq RpbPutResp -instance Tagged DeleteRequest where +instance Tagged RpbDelReq where messageTag _ = Types.DeleteRequest {-# INLINE messageTag #-} -instance Request DeleteRequest where +instance Request RpbDelReq where expectedResponse _ = Types.DeleteResponse {-# INLINE expectedResponse #-} -instance Tagged ListBucketsRequest where +instance Tagged RpbListBucketsReq where messageTag _ = Types.ListBucketsRequest {-# INLINE messageTag #-} -instance Request ListBucketsRequest where +instance Request RpbListBucketsReq where expectedResponse _ = Types.ListBucketsResponse {-# INLINE expectedResponse #-} -instance Tagged ListBucketsResponse where +instance Tagged RpbListBucketsResp where messageTag _ = Types.ListBucketsResponse {-# INLINE messageTag #-} -instance Response ListBucketsResponse +instance Response RpbListBucketsResp -instance Exchange ListBucketsRequest ListBucketsResponse +instance Exchange RpbListBucketsReq RpbListBucketsResp -instance Tagged ListKeysRequest where +instance Tagged RpbListKeysReq where messageTag _ = Types.ListKeysRequest {-# INLINE messageTag #-} -instance Request ListKeysRequest where +instance Request RpbListKeysReq where expectedResponse _ = Types.ListKeysResponse {-# INLINE expectedResponse #-} -instance Tagged ListKeysResponse where +instance Tagged RpbListKeysResp where messageTag _ = Types.ListKeysResponse {-# INLINE messageTag #-} -instance Response ListKeysResponse +instance Response RpbListKeysResp -instance Tagged GetBucketRequest where +instance Tagged RpbGetBucketReq where messageTag _ = Types.GetBucketRequest {-# INLINE messageTag #-} -instance Request GetBucketRequest where +instance Request RpbGetBucketReq where expectedResponse _ = Types.GetBucketResponse {-# INLINE expectedResponse #-} -instance Tagged GetBucketResponse where +instance Tagged RpbGetBucketResp where messageTag _ = Types.GetBucketResponse {-# INLINE messageTag #-} -instance Response GetBucketResponse +instance Response RpbGetBucketResp -instance Exchange GetBucketRequest GetBucketResponse +instance Exchange RpbGetBucketReq RpbGetBucketResp -instance Tagged SetBucketRequest where +instance Tagged RpbSetBucketReq where messageTag _ = Types.SetBucketRequest {-# INLINE messageTag #-} -instance Request SetBucketRequest where +instance Request RpbSetBucketReq where expectedResponse _ = Types.SetBucketResponse {-# INLINE expectedResponse #-} -instance Request GetBucketTypeRequest where +instance Request RpbGetBucketTypeReq where expectedResponse _ = Types.GetBucketResponse -instance Tagged GetBucketTypeRequest where +instance Tagged RpbGetBucketTypeReq where messageTag _ = Types.GetBucketTypeRequest -instance Exchange GetBucketTypeRequest GetBucketResponse +instance Exchange RpbGetBucketTypeReq RpbGetBucketResp -instance Tagged MapReduceRequest where +instance Tagged RpbMapRedReq where messageTag _ = Types.MapReduceRequest {-# INLINE messageTag #-} -instance Request MapReduceRequest where +instance Request RpbMapRedReq where expectedResponse _ = Types.MapReduceResponse {-# INLINE expectedResponse #-} -instance Tagged MapReduce where +instance Tagged RpbMapRedResp where messageTag _ = Types.MapReduceResponse {-# INLINE messageTag #-} -instance Response MapReduce +instance Response RpbMapRedResp -instance Exchange MapReduceRequest MapReduce +instance Exchange RpbMapRedReq RpbMapRedResp -instance Tagged DtFetchRequest where +instance Tagged DtFetchReq where messageTag _ = Types.DtFetchRequest {-# INLINE messageTag #-} -instance Tagged DtFetchResponse where +instance Tagged DtFetchResp where messageTag _ = Types.DtFetchResponse {-# INLINE messageTag #-} -instance Request DtFetchRequest where +instance Request DtFetchReq where expectedResponse _ = Types.DtFetchResponse {-# INLINE expectedResponse #-} -instance Response DtFetchResponse +instance Response DtFetchResp -instance Exchange DtFetchRequest DtFetchResponse +instance Exchange DtFetchReq DtFetchResp -instance Tagged DtUpdateRequest where +instance Tagged DtUpdateReq where messageTag _ = Types.DtUpdateRequest {-# INLINE messageTag #-} -instance Tagged DtUpdateResponse where +instance Tagged DtUpdateResp where messageTag _ = Types.DtUpdateResponse {-# INLINE messageTag #-} -instance Request DtUpdateRequest where +instance Request DtUpdateReq where expectedResponse _ = Types.DtUpdateResponse {-# INLINE expectedResponse #-} -instance Response DtUpdateResponse +instance Response DtUpdateResp -instance Exchange DtUpdateRequest DtUpdateResponse +instance Exchange DtUpdateReq DtUpdateResp -instance Tagged SearchQueryRequest where +instance Tagged RpbSearchQueryReq where messageTag _ = Types.SearchQueryRequest {-# INLINE messageTag #-} -instance Request SearchQueryRequest where +instance Request RpbSearchQueryReq where expectedResponse _ = Types.SearchQueryResponse {-# INLINE expectedResponse #-} -instance Tagged SearchQueryResponse where +instance Tagged RpbSearchQueryResp where messageTag _ = Types.SearchQueryResponse {-# INLINE messageTag #-} -instance Response SearchQueryResponse +instance Response RpbSearchQueryResp -instance Exchange SearchQueryRequest SearchQueryResponse +instance Exchange RpbSearchQueryReq RpbSearchQueryResp -instance Tagged YzIndexGetRequest where +instance Tagged RpbYokozunaIndexGetReq where messageTag _ = Types.YokozunaIndexGetRequest -instance Request YzIndexGetRequest where +instance Request RpbYokozunaIndexGetReq where expectedResponse _ = Types.YokozunaIndexGetResponse -instance Tagged YzIndexGetResponse where +instance Tagged RpbYokozunaIndexGetResp where messageTag _ = Types.YokozunaIndexGetResponse -instance Response YzIndexGetResponse +instance Response RpbYokozunaIndexGetResp -instance Exchange YzIndexGetRequest YzIndexGetResponse +instance Exchange RpbYokozunaIndexGetReq RpbYokozunaIndexGetResp -instance Request YzIndexPutRequest where +instance Request RpbYokozunaIndexPutReq where expectedResponse _ = Types.YokozunaIndexPutRequest -instance Tagged YzIndexPutRequest where +instance Tagged RpbYokozunaIndexPutReq where messageTag _ = Types.YokozunaIndexPutRequest -instance Exchange YzIndexPutRequest PutResponse +instance Exchange RpbYokozunaIndexPutReq RpbPutResp -instance Tagged YzIndexDeleteRequest where +instance Tagged RpbYokozunaIndexDeleteReq where messageTag _ = Types.YokozunaIndexDeleteRequest -instance Request YzIndexDeleteRequest where +instance Request RpbYokozunaIndexDeleteReq where expectedResponse _ = Types.DeleteResponse putTag :: MessageTag -> Put 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 @@ -57,17 +57,16 @@ ) where import Control.Exception (Exception, throw) -import Data.ByteString.Lazy (ByteString) +import Data.ByteString (ByteString) +import qualified Data.ByteString.Lazy as LazyByteString import Data.Digest.Pure.MD5 (md5) import Data.Hashable (Hashable) import Data.IORef (IORef) -import Data.Sequence (Seq) +import qualified Data.Riak.Proto as Proto 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 @@ -186,7 +185,7 @@ type Score = Double -- | Search index info -type IndexInfo = YzIndex.YzIndex +type IndexInfo = Proto.RpbYokozunaIndex -- | N value -- @@ -198,7 +197,7 @@ -- | Solr search result data SearchResult = SearchResult - { docs :: !(Seq (Seq (ByteString, Maybe ByteString))) + { docs :: ![[(ByteString, Maybe ByteString)]] , maxScore :: !(Maybe Float) , numFound :: !(Maybe Word32) } deriving (Eq, Ord, Show) @@ -255,11 +254,11 @@ {-# INLINE messageTag #-} -- | A message representing a request from client to server. -class (Tagged msg, ReflectDescriptor msg, Show msg, Wire msg) => Request msg +class (Proto.Message msg, Show msg, Tagged msg) => Request msg where expectedResponse :: msg -> MessageTag -- | A message representing a response from server to client. -class (Tagged msg, ReflectDescriptor msg, Show msg, Wire msg) => Response msg +class (Proto.Message msg, Show msg, Tagged msg) => Response msg class (Request req, Response resp) => Exchange req resp | req -> resp @@ -272,7 +271,7 @@ } deriving (Eq, Typeable) instance Show VClock where - show (VClock s) = "VClock " ++ show (md5 s) + show (VClock s) = "VClock " ++ show (md5 (LazyByteString.fromStrict s)) -- | A read/write quorum. The quantity of replicas that must respond -- to a read or write request before it is considered successful. This diff --git a/src/Network/Riak/Value.hs b/src/Network/Riak/Value.hs --- a/src/Network/Riak/Value.hs +++ b/src/Network/Riak/Value.hs @@ -32,36 +32,30 @@ import Control.Applicative import Data.Aeson.Types (Parser, Result(..), parse) -import Data.Foldable (toList) import Data.Monoid ((<>)) import Network.Riak.Connection.Internal -import Network.Riak.Protocol.Content (Content(..)) -import Network.Riak.Protocol.GetResponse (GetResponse(..)) -import Network.Riak.Protocol.IndexResponse (IndexResponse(..)) -import Network.Riak.Protocol.PutResponse (PutResponse(..)) +import Network.Riak.Lens import Network.Riak.Resolvable (ResolvableMonoid(..)) import Network.Riak.Response (unescapeLinks) import Network.Riak.Types.Internal hiding (MessageTag(..)) -import qualified Network.Riak.Protocol.Pair as Pair -import qualified Data.Aeson.Parser as Aeson +import qualified Data.Riak.Proto as Proto +import qualified Data.Aeson as Aeson (eitherDecodeStrict) import qualified Data.Aeson.Types as Aeson -import qualified Data.Attoparsec.Lazy as A -import qualified Data.ByteString.Lazy as L -import qualified Data.ByteString.Lazy.Char8 as CL8 -import qualified Data.Sequence as Seq +import qualified Data.ByteString as L +import qualified Data.ByteString.Char8 as CL8 import qualified Network.Riak.Content as C import qualified Network.Riak.Request as Req -fromContent :: IsContent c => Content -> Maybe c +fromContent :: IsContent c => Proto.RpbContent -> Maybe c fromContent c = case parse parseContent c of Success a -> Just a Error _ -> Nothing class IsContent c where - parseContent :: Content -> Parser c - toContent :: c -> Content + parseContent :: Proto.RpbContent -> Parser c + toContent :: c -> Proto.RpbContent -instance IsContent Content where +instance IsContent Proto.RpbContent where parseContent = return {-# INLINE parseContent #-} @@ -77,10 +71,10 @@ {-# INLINE toContent #-} instance IsContent Aeson.Value where - parseContent c | content_type c == Just "application/json" = - case A.parse Aeson.json (value c) of - A.Done _ a -> return a - A.Fail _ _ err -> fail err + parseContent c | c ^. Proto.maybe'contentType == Just "application/json" = + case Aeson.eitherDecodeStrict (c ^. Proto.value) of + Left err -> fail err + Right a -> return a | otherwise = fail "non-JSON document" toContent = C.json {-# INLINE toContent #-} @@ -88,13 +82,17 @@ deriving instance (IsContent a) => IsContent (ResolvableMonoid a) -- | Add indexes to a content value for a further put request. -addIndexes :: [IndexValue] -> Content -> Content +addIndexes :: [IndexValue] -> Proto.RpbContent -> Proto.RpbContent addIndexes ix c = - c { C.indexes = Seq.fromList . map toPair $ ix } + c & Proto.indexes .~ (map toPair ix) where - toPair (IndexInt k v) = Pair.Pair (k <> "_int") - (Just . CL8.pack . show $ v) - toPair (IndexBin k v) = Pair.Pair (k <> "_bin") (Just v) + toPair :: IndexValue -> Proto.RpbPair + toPair (IndexInt k v) = Proto.defMessage + & Proto.key .~ (k <> "_int") + & Proto.value .~ (CL8.pack . show $ v) + toPair (IndexBin k v) = Proto.defMessage + & Proto.key .~ (k <> "_bin") + & Proto.value .~ v -- | Store a single value. This may return multiple conflicting -- siblings. Choosing among them, and storing a new value, is your @@ -135,12 +133,12 @@ putMany conn bt b puts w dw = mapM putResp =<< pipeline conn (map (\(k,v,c) -> Req.put bt b k v (toContent c) w dw True) puts) -putResp :: (IsContent c) => PutResponse -> IO ([c], VClock) -putResp PutResponse{..} = do - case vclock of +putResp :: (IsContent c) => Proto.RpbPutResp -> IO ([c], VClock) +putResp response = do + case response ^. Proto.maybe'vclock of Nothing -> return ([], VClock L.empty) Just s -> do - c <- convert content + c <- convert (response ^. Proto.content) return (c, VClock s) -- | Store a single value, without the possibility of conflict @@ -187,22 +185,25 @@ getMany conn bt b ks r = mapM getResp =<< pipelineMaybe conn (map (\k -> Req.get bt b k r) ks) -getResp :: (IsContent c) => Maybe GetResponse -> IO (Maybe ([c], VClock)) +getResp :: (IsContent c) => Maybe Proto.RpbGetResp -> IO (Maybe ([c], VClock)) getResp resp = case resp of - Just (GetResponse content (Just s) _) -> do - c <- convert content + Just resp' -> + case resp' ^. Proto.maybe'vclock of + Just s -> do + c <- convert (resp' ^. Proto.content) return $ Just (c, VClock s) - _ -> return Nothing + _ -> return Nothing + _ -> return Nothing -getByIndexResp :: Maybe IndexResponse -> IO [Key] +getByIndexResp :: Maybe Proto.RpbIndexResp -> IO [Key] getByIndexResp resp = case resp of - Just (IndexResponse keys _ _ _) -> return (toList keys) + Just resp' -> return (resp' ^. Proto.keys) Nothing -> return [] -convert :: IsContent v => Seq.Seq Content -> IO [v] -convert = go [] [] . toList +convert :: IsContent v => [Proto.RpbContent] -> IO [v] +convert = go [] [] where go cs vs (x:xs) = case fromContent y of Just v -> go cs (v:vs) xs _ -> go (y:cs) vs xs diff --git a/tests/CRDTProperties.hs b/tests/CRDTProperties.hs --- a/tests/CRDTProperties.hs +++ b/tests/CRDTProperties.hs @@ -20,7 +20,8 @@ import Control.Applicative #endif import Control.Monad.RWS -import Data.ByteString.Lazy (ByteString) +import Control.Monad.Fail +import Data.ByteString (ByteString) import Data.Default.Class import Data.List.NonEmpty import qualified Data.Map as Map @@ -188,7 +189,8 @@ -- | Abstract machine. -- Yields a value on 'Get', modifies state on 'Update'. -machine :: (MonadWriter [Maybe C.DataType] m, +machine :: (MonadFail m, + MonadWriter [Maybe C.DataType] m, MonadState s m, Applicative m, Action t op) @@ -213,7 +215,8 @@ -- | Riak version of the 'machine'. -- State is 'B.Connection', get/update are IO-requests to riak. -riak :: (MonadWriter [Maybe C.DataType] m, +riak :: (MonadFail m, + MonadWriter [Maybe C.DataType] m, MonadState B.Connection m, Applicative m, MonadIO m, Action t op) @@ -231,7 +234,8 @@ -- | Haskell emulation version of the 'machine'. -- State is 'RiakState', get/update try to match riak's behaviour. -pure_ :: (MonadWriter [Maybe C.DataType] m, +pure_ :: (MonadFail m, + MonadWriter [Maybe C.DataType] m, MonadState RiakState m, Applicative m, Action t op) diff --git a/tests/Properties.hs b/tests/Properties.hs --- a/tests/Properties.hs +++ b/tests/Properties.hs @@ -8,7 +8,8 @@ #if __GLASGOW_HASKELL__ < 710 import Control.Applicative ((<$>)) #endif -import qualified Data.ByteString.Lazy as L +import Data.ByteString (ByteString) +import qualified Data.ByteString as BS import Data.Maybe import qualified Network.Riak.Basic as B import Network.Riak.Content (binary) @@ -18,21 +19,21 @@ import Test.Tasty.QuickCheck import Utils -instance Arbitrary L.ByteString where - arbitrary = L.pack `fmap` arbitrary +instance Arbitrary ByteString where + arbitrary = BS.pack `fmap` arbitrary newtype QCBucket = QCBucket Riak.Bucket deriving Show instance Arbitrary QCBucket where - arbitrary = QCBucket <$> arbitrary `suchThat` (not . L.null) + arbitrary = QCBucket <$> arbitrary `suchThat` (not . BS.null) newtype QCKey = QCKey Riak.Key deriving Show instance Arbitrary QCKey where - arbitrary = QCKey <$> arbitrary `suchThat` (not . L.null) + arbitrary = QCKey <$> arbitrary `suchThat` (not . BS.null) -t_put_get :: QCBucket -> QCKey -> L.ByteString -> Property +t_put_get :: QCBucket -> QCKey -> ByteString -> Property t_put_get (QCBucket b) (QCKey k) v = monadicIO $ assert . uncurry (==) =<< run act where @@ -42,7 +43,7 @@ return (p,r) btype = Nothing -put_delete_get :: QCBucket -> QCKey -> L.ByteString -> Property +put_delete_get :: QCBucket -> QCKey -> ByteString -> Property put_delete_get (QCBucket b) (QCKey k) v = monadicIO $ do r <- run act diff --git a/tests/Test.hs b/tests/Test.hs --- a/tests/Test.hs +++ b/tests/Test.hs @@ -11,20 +11,21 @@ import Control.Applicative #endif import qualified Data.Map as M +import qualified Data.Riak.Proto as Proto import qualified Data.Set as S -import qualified Data.Sequence as Seq import Data.List.NonEmpty (NonEmpty(..)) -import Data.Foldable (toList) #if __GLASGOW_HASKELL__ < 804 import Data.Semigroup #endif import Data.Text (Text) +import Data.Function ((&)) import Control.Concurrent (threadDelay) import Control.Exception +import Lens.Micro ((^.), (.~)) import qualified Network.Riak as Riak import Network.Riak.Admin.DSL import qualified Network.Riak.Basic as B -import qualified Network.Riak.Content as B (binary,Content,value) +import qualified Network.Riak.Content as B (binary,RpbContent) import Network.Riak.Connection (exchange) import qualified Network.Riak.CRDT as C import qualified Network.Riak.CRDT.Riak as C @@ -34,8 +35,6 @@ import qualified Network.Riak.JSON as J import Network.Riak.Resolvable (ResolvableMonoid (..)) import Network.Riak.Types -import qualified Network.Riak.Protocol.ErrorResponse as ER -import qualified Network.Riak.Protocol.SearchQueryRequest as S import qualified Properties import qualified CRDTProperties as CRDT import Utils @@ -201,13 +200,13 @@ C.sendModify conn btype buck key [C.SetRemove kw] delay a <- query conn ("set:" <> kw) - assertEqual "should not found non-existing" (S.SearchResult Seq.empty (Just 0.0) (Just 0)) a + assertEqual "should not found non-existing" (S.SearchResult [] (Just 0.0) (Just 0)) a C.sendModify conn btype buck key [C.SetAdd kw] delay b <- query conn ("set:" <> kw) - assertBool "searches specific" $ not (Seq.null (S.docs b)) + assertBool "searches specific" $ not (null (S.docs b)) c <- query conn ("set:*") - assertBool "searches *" $ not (Seq.null (S.docs c)) + assertBool "searches *" $ not (null (S.docs c)) where query conn q = S.searchRaw conn q "set-ix" (btype,buck,key) = ("sets","xxx","yyy") @@ -216,10 +215,10 @@ search2 :: TestTree search2 = testCase "search with fl" $ withGlobalConn $ \conn -> do - let req = (Req.search "set:haskell" "set-ix") { S.fl = Seq.singleton "_yz_rk" } + let req = (Req.search "set:haskell" "set-ix") & Proto.fl .~ ["_yz_rk"] resp <- Resp.search <$> exchange conn req assertEqual "only returns fl" - (Seq.singleton (Seq.singleton ("_yz_rk", Just "yyy"))) + ([[("_yz_rk", Just "yyy")]]) (S.docs resp) @@ -246,9 +245,8 @@ pure () where - f :: ER.ErrorResponse -> Maybe () - f (ER.ErrorResponse "notfound" 0) = Just () - f _ = Nothing + f :: Proto.RpbErrorResp -> Maybe () + f resp = guard (resp ^. Proto.errmsg == "notfound") bucketTypes :: TestTree bucketTypes = testCase "bucketTypes" $ withGlobalConn $ \conn -> do @@ -266,10 +264,10 @@ where (b,k) = ("xxx","0") :: (Bucket,Key) types = [Nothing, Just "untyped-1", Just "untyped-2"] :: [Maybe BucketType] - [o0,o1,o2] = B.binary <$> ["A","B","C"] :: [B.Content] + [o0,o1,o2] = B.binary <$> ["A","B","C"] :: [B.RpbContent] - valok :: Maybe (Seq.Seq B.Content, VClock) -> B.Content -> Bool - valok (Just (rs,_)) o = B.value o `elem` map B.value (toList rs) + valok :: Maybe ([B.RpbContent], VClock) -> B.RpbContent -> Bool + valok (Just (rs,_)) o = (o ^. Proto.value) `elem` map (^. Proto.value) rs valok _ _ = False @@ -287,12 +285,12 @@ put_Err = putSome B.put_ noBtype putSome :: (Connection -> Maybe BucketType -> Bucket -> Key - -> Maybe VClock -> B.Content -> Quorum -> Quorum -> IO a) + -> Maybe VClock -> B.RpbContent -> Quorum -> Quorum -> IO a) -> Maybe BucketType -> Connection -> IO a putSome f bt c = f c bt buck key Nothing val Default Default shouldBeOK act = act >> assertBool "ok" True - shouldThrow act = catch (act >> assertBool "exception" False) (\(_e::ER.ErrorResponse) -> pure ()) + shouldThrow act = catch (act >> assertBool "exception" False) (\(_e::Proto.RpbErrorResp) -> pure ()) btype = Just "untyped-1" noBtype = Just "no such type"