http2-client-grpc 0.5.0.4 → 0.8.0.0
raw patch · 7 files changed
Files
- Changelog.md +12/−0
- LICENSE +1/−1
- README.md +1/−73
- Setup.hs +2/−3
- http2-client-grpc.cabal +13/−13
- src/Network/GRPC/Client.hs +114/−114
- src/Network/GRPC/Client/Helpers.hs +37/−45
Changelog.md view
@@ -2,6 +2,18 @@ ## Unreleased changes +## 0.8.0.0++Fork `http2-client-grpc` to drop dependency on `proto-lens`.++## 0.6.0.0++Update to http2-client-0.9.0.0++## 0.5.0.4++Use proto-lens-0.4.0.0, this change is package-list incompatible but requires no code change.+ ## 0.5.0.3 Add helpers for new bidirectional and general streams.
LICENSE view
@@ -1,4 +1,4 @@-Copyright Author name here (c) 2017+Copyright Lucas Di Cioccio (c) 2017 All rights reserved.
README.md view
@@ -1,75 +1,3 @@ # http2-client-grpc -A native HTTP2 gRPC client library using `proto-lens` and `http2-client`.--## Summary--This project provides a library that leverages the code generated using-proto-lens (in particular, proto-lens-protoc) to implement the client-side of-gRPC services.--## Usage--### Prerequisites--In addition to a working Haskell dev environment, you need to:-- build the `proto-lens-protoc` executable (`proto-lens`)-- install the `protoc` executable--### Adding .proto files to a Haskell package--In order to run gRPC:--- generate the `Proto` stubs in some `gen` directory--A single `protoc` invocation may be enough for both Proto and GRPC outputs:--```bash-protoc "--plugin=protoc-gen-haskell-protolens=${protolens}" \- --haskell-protolens_out=./gen \- -I "${protodir1} \- -I "${protodir2} \- ${first.proto} \- ${second.proto}-```--- add the `gen` sourcedir for the generated to your .cabal/package.yaml file (cf. 'hs-source-dirs').-- add the generated Proto modules to the 'exposed-modules' (or 'other-modules') keys--A reliable way to list the module names is the following bash invocation:--```bash-find gen -name "*.hs" | sed -e 's/gen\///' | sed -e 's/\.hs$//' | tr '/' '.'-```--Unlike `proto-lens`, this project does not yet provide a modified `Setup.hs`.-As a result, we cannot automate these steps from within Cabal/Stack. Hence,-you'll have to automate these steps outside your Haskell toolchain.--## Calling a GRPC service--In short, use `http2-client` with the `Network.GRPC.call` function and let the types guide you.--### Example--You'll find an example leveraging the awesome `grpcb.in` service at https://github.com/lucasdicioccio/http2-client-grpc-example .--### gRPC service mapping--The Protobuf format specifies the notion of Service which can have one or more-RPCs. The gRPC protocal maps these services onto HTTP2 headers and HTTP2-frames. In general, gRPC implementation generate one inlined function per RPC.-This implementation differs by decoupling the HTTP2 transport and leveraging-generics to provide generic functions. This design allows the-`Network.GRPC.call` to be type safe and multi-usage. For instance, you can wrap-this function with whatever metrics/benchmark code and have all your RPC calls-be monitored in a uniform way.--## Status--This library is currently an early-stage library. Expect breaking changes.--## Next steps--- provide function to map raw results into commonly-understood GRPC errors-- (nice-to-have) support effectful compression (Zlib pure compression is supported)+A native HTTP2 gRPC client library using `http2-client`.
Setup.hs view
@@ -1,3 +1,2 @@-import Data.ProtoLens.Setup--main = defaultMain "."+import Distribution.Simple+main = defaultMain
http2-client-grpc.cabal view
@@ -1,13 +1,13 @@ name: http2-client-grpc-version: 0.5.0.4+version: 0.8.0.0 synopsis: Implement gRPC-over-HTTP2 clients.-description: A gRPC over http2-client using proto-lens to generate client code.-homepage: https://github.com/lucasdicioccio/http2-client-grpc#readme+description: A gRPC over http2-client.+homepage: https://github.com/haskell-grpc-native/http2-grpc-haskell/blob/master/http2-client-grpc/README.md license: BSD3 license-file: LICENSE-author: Lucas DiCioccio+author: Lucas DiCioccio, Alejandro Serrano maintainer: lucas@dicioccio.fr-copyright: 2017 Lucas DiCioccio+copyright: 2017 - 2019 Lucas DiCioccio, Alejandro Serrano category: Network build-type: Simple extra-source-files: README.md Changelog.md@@ -20,16 +20,16 @@ build-depends: base >= 4.11 && < 5 , async >= 2.2 && < 2.3 , binary >= 0.8 && < 0.9- , bytestring >= 0.10.8 && < 0.10.9+ , bytestring >= 0.10.8 && < 0.11 , case-insensitive >= 1.2.0 && < 1.3 , data-default-class >= 0.1 && <0.2- , lens >= 4.16 && < 4.17- , http2 >= 1.6 && < 1.7- , http2-client >= 0.8 && < 0.9- , http2-grpc-types >= 0.3 && < 0.4- , proto-lens >= 0.4 && < 0.5+ , lifted-async >= 0.10 && < 0.11+ , lifted-base >= 0.2 && < 0.3+ , http2 >= 1.6 && < 2.1+ , http2-client >= 0.9 && < 0.10+ , http2-grpc-types >= 0.5 && < 0.6 , text >= 1.2 && < 1.3- , tls >= 1.4 && < 1.5+ , tls >= 1.4 && < 1.6 default-language: Haskell2010 test-suite http2-client-grpc-test@@ -43,4 +43,4 @@ source-repository head type: git- location: https://github.com/lucasdicioccio/http2-client-grpc+ location: https://github.com/haskell-grpc-native/http2-grpc-haskell
src/Network/GRPC/Client.hs view
@@ -47,8 +47,7 @@ -- inspiration from the existing RPC functions and learn how to write one. module Network.GRPC.Client ( -- * Building blocks.- RPC(..)- , RPCCall(..)+ RPCCall(..) , CIHeaderList , Authority , Timeout(..)@@ -81,9 +80,9 @@ , HeaderList ) where -import Control.Concurrent.Async (concurrently)+import Control.Concurrent.Async.Lifted (concurrently) import Control.Exception (SomeException, Exception(..), throwIO)-import Data.ByteString.Char8 (unpack)+import Control.Monad.IO.Class (liftIO) import Data.ByteString.Lazy (toStrict) import Data.Binary.Builder (toLazyByteString) import Data.Binary.Get (Decoder(..), pushChunk, pushEndOfInput)@@ -92,8 +91,6 @@ import Data.CaseInsensitive (CI) import qualified Data.CaseInsensitive as CI import Data.Monoid ((<>))-import Data.ProtoLens.Service.Types (Service(..), HasMethod, HasMethodImpl(..), StreamingType(..))-import GHC.TypeLits (Symbol) import Network.GRPC.HTTP2.Types import Network.GRPC.HTTP2.Encoding@@ -114,7 +111,7 @@ -- - 1st item: initial HTTP2 response -- - 2nd item: second (trailers) HTTP2 response -- - 3rd item: proper gRPC answer-type RawReply a = Either ErrorCode (CIHeaderList, Maybe CIHeaderList, (Either String a))+type RawReply a = Either ErrorCode (CIHeaderList, Maybe CIHeaderList, Either String a) -- | gRPC disables HTTP2 push-promises. --@@ -124,11 +121,14 @@ -- | http2-client handler for push promise. throwOnPushPromise :: PushPromiseHandler-throwOnPushPromise _ _ _ _ _ = throwIO UnallowedPushPromiseReceived+throwOnPushPromise _ _ _ _ _ = lift $ throwIO UnallowedPushPromiseReceived -- | Wait for an RPC reply.-waitReply :: (Service s, HasMethod s m) => RPC s m -> Decoding -> Http2Stream -> IncomingFlowControl -> IO (RawReply (MethodOutput s m))-waitReply rpc decoding stream flowControl = do+waitReply+ :: (GRPCOutput r o)+ => r -> Decoding -> Http2Stream -> IncomingFlowControl+ -> ClientIO (RawReply o)+waitReply rpc decoding stream flowControl = format . fromStreamResult <$> waitStream stream flowControl throwOnPushPromise where decompress = _getDecodingCompression decoding@@ -140,8 +140,10 @@ -- presence of a message indicate an error -- TODO: double check this is true in general case lookup grpcMessageH hdrs2 of- Nothing -> fromDecoder $ pushEndOfInput $ flip pushChunk dat $ decodeOutput rpc decompress- Just errMsg -> Left $ unpack errMsg+ Nothing -> fromDecoder $ pushEndOfInput+ $ flip pushChunk dat+ $ decodeOutput rpc decompress+ Just errMsg -> Left $ ByteString.unpack errMsg return (hdrs2, trls2, res) @@ -150,27 +152,25 @@ -- | Exception raised when a ServerStreaming RPC results in a decoding -- error.-data StreamReplyDecodingError = StreamReplyDecodingError String deriving Show+newtype StreamReplyDecodingError = StreamReplyDecodingError String deriving Show instance Exception StreamReplyDecodingError where -- | Exception raised when a ServerStreaming RPC results in an invalid -- state machine.-data InvalidState = InvalidState String- deriving Show+newtype InvalidState = InvalidState String deriving Show instance Exception InvalidState where -- | Newtype helper used to uniformize all type of streaming modes when -- passing arguments to the 'open' call.-newtype RPCCall s (m ::Symbol) a = RPCCall {- runRPC :: Http2Client -> Http2Stream -> IncomingFlowControl -> OutgoingFlowControl -> Encoding -> Decoding -> IO a+data RPCCall r a = RPCCall {+ rpcFromCall :: r+ , runRPC :: Http2Client -> Http2Stream+ -> IncomingFlowControl -> OutgoingFlowControl+ -> Encoding -> Decoding -> ClientIO a } --- | Helper to get the proxy object from an RPCCall.-rpcFromCall :: RPCCall s m a -> RPC s m-rpcFromCall _ = RPC- -- | Main handler to perform gRPC calls to a service.-open :: (Service s, HasMethod s m)+open :: (IsRPC r) => Http2Client -- ^ A connected HTTP2 client. -> Authority@@ -183,9 +183,9 @@ -- ^ Compression used for encoding. -> Decoding -- ^ Compression allowed for decoding- -> RPCCall s m a+ -> RPCCall r a -- ^ The actual RPC handler.- -> IO (Either TooMuchConcurrency a)+ -> ClientIO (Either TooMuchConcurrency a) open conn authority extraheaders timeout encoding decoding call = do let rpc = rpcFromCall call let compress = _getEncodingCompression encoding@@ -202,35 +202,34 @@ ] <> extraheaders withHttp2Stream conn $ \stream -> let- initStream = headers stream request (setEndHeader)- handler isfc osfc = do- (runRPC call) conn stream isfc osfc encoding decoding+ initStream = headers stream request setEndHeader+ handler isfc osfc = runRPC call conn stream isfc osfc encoding decoding in StreamDefinition initStream handler -- | gRPC call for Server Streaming. streamReply- :: (Service s, HasMethod s m, MethodStreamingType s m ~ 'ServerStreaming)- => RPC s m+ :: (GRPCInput r i, GRPCOutput r o)+ => r -- ^ RPC to call. -> a -- ^ An initial state.- -> MethodInput s m+ -> i -- ^ The input.- -> (a -> HeaderList -> MethodOutput s m -> IO a)+ -> (a -> HeaderList -> o -> ClientIO a) -- ^ A state-passing handler that is called with the message read.- -> RPCCall s m (a, HeaderList, HeaderList)-streamReply rpc v0 req handler = RPCCall $ \conn stream isfc osfc encoding decoding -> do+ -> RPCCall r (a, HeaderList, HeaderList)+streamReply rpc v0 req handler = RPCCall rpc $ \conn stream isfc osfc encoding decoding -> let { loop v1 decode hdrs = _waitEvent stream >>= \case- (StreamPushPromiseEvent _ _ _) ->- throwIO UnallowedPushPromiseReceived- (StreamHeadersEvent _ trls) ->+ StreamPushPromiseEvent {} ->+ lift $ throwIO UnallowedPushPromiseReceived+ StreamHeadersEvent _ trls -> return (v1, hdrs, trls)- (StreamErrorEvent _ _) ->- throwIO (InvalidState "stream error")- (StreamDataEvent _ dat) -> do- _addCredit isfc (ByteString.length dat)- _ <- _consumeCredit isfc (ByteString.length dat)+ StreamErrorEvent _ _ ->+ lift $ throwIO (InvalidState "stream error")+ StreamDataEvent _ dat -> do+ liftIO $ _addCredit isfc (ByteString.length dat)+ _ <- liftIO $ _consumeCredit isfc (ByteString.length dat) _ <- _updateWindow isfc handleAllChunks decoding v1 hdrs decode dat loop } in do@@ -241,18 +240,18 @@ StreamHeadersEvent _ hdrs -> loop v0 (decodeOutput rpc decompress) hdrs _ ->- throwIO (InvalidState "no headers")+ lift $ throwIO (InvalidState "no headers") where handleAllChunks decoding v1 hdrs decode dat exitLoop = case pushChunk decode dat of- (Done unusedDat _ (Right val)) -> do+ Done unusedDat _ (Right val) -> do v2 <- handler v1 hdrs val let decompress = _getDecodingCompression decoding handleAllChunks decoding v2 hdrs (decodeOutput rpc decompress) unusedDat exitLoop- (Done _ _ (Left err)) -> do- throwIO (StreamReplyDecodingError $ "done-error: " ++ err)- (Fail _ _ err) -> do- throwIO (StreamReplyDecodingError $ "fail-error: " ++ err)+ Done _ _ (Left err) ->+ lift $ throwIO (StreamReplyDecodingError $ "done-error: " ++ err)+ Fail _ _ err ->+ lift $ throwIO (StreamReplyDecodingError $ "fail-error: " ++ err) partial@(Partial _) -> exitLoop v1 partial hdrs @@ -262,15 +261,15 @@ -- | gRPC call for Client Streaming. streamRequest- :: (Service s, HasMethod s m, MethodStreamingType s m ~ 'ClientStreaming)- => RPC s m+ :: (GRPCInput r i, GRPCOutput r o)+ => r -- ^ RPC to call. -> a -- ^ An initial state.- -> (a -> IO (a, Either StreamDone (CompressMode, MethodInput s m)))+ -> (a -> ClientIO (a, Either StreamDone (CompressMode, i))) -- ^ A state-passing action to retrieve the next message to send to the server.- -> RPCCall s m (a, RawReply (MethodOutput s m))-streamRequest rpc v0 handler = RPCCall $ \conn stream isfc streamFlowControl encoding decoding ->+ -> RPCCall r (a, RawReply o)+streamRequest rpc v0 handler = RPCCall rpc$ \conn stream isfc streamFlowControl encoding decoding -> let ocfc = _outgoingFlowControl conn go v1 = do (v2, nextEvent) <- handler v1@@ -279,7 +278,8 @@ let compress = case doCompress of Compressed -> _getEncodingCompression encoding Uncompressed -> uncompressed- sendSingleMessage rpc msg (Encoding compress) id conn ocfc stream streamFlowControl+ sendSingleMessage rpc msg (Encoding compress) id+ conn ocfc stream streamFlowControl go v2 Left _ -> do sendData conn stream setEndStream ""@@ -289,23 +289,23 @@ -- | Serialize and send a single message. sendSingleMessage- :: (Service s, HasMethod s m)- => RPC s m- -> MethodInput s m+ :: (GRPCInput r i)+ => r+ -> i -> Encoding -> FlagSetter -> Http2Client -> OutgoingFlowControl -> Http2Stream -> OutgoingFlowControl- -> IO ()+ -> ClientIO () sendSingleMessage rpc msg encoding flagMod conn connectionFlowControl stream streamFlowControl = do let compress = _getEncodingCompression encoding let goUpload dat = do let !wanted = ByteString.length dat gotStream <- _withdrawCredit streamFlowControl wanted got <- _withdrawCredit connectionFlowControl gotStream- _receiveCredit streamFlowControl (gotStream - got)+ liftIO $ _receiveCredit streamFlowControl (gotStream - got) if got == wanted then sendData conn stream flagMod dat@@ -316,32 +316,32 @@ -- | gRPC call for an unary request. singleRequest- :: (Service s, HasMethod s m)- => RPC s m+ :: (GRPCInput r i, GRPCOutput r o)+ => r -- ^ RPC to call.- -> MethodInput s m+ -> i -- ^ RPC's input.- -> RPCCall s m (RawReply (MethodOutput s m))-singleRequest rpc msg = RPCCall $ \conn stream isfc osfc encoding decoding -> do+ -> RPCCall r (RawReply o)+singleRequest rpc msg = RPCCall rpc $ \conn stream isfc osfc encoding decoding -> do let ocfc = _outgoingFlowControl conn sendSingleMessage rpc msg encoding setEndStream conn ocfc stream osfc waitReply rpc decoding stream isfc -- | Handler for received message.-type HandleMessageStep s m a = HeaderList -> a -> MethodOutput s m -> IO a+type HandleMessageStep i o a = HeaderList -> a -> o -> ClientIO a -- | Handler for received trailers.-type HandleTrailersStep a = HeaderList -> a -> HeaderList -> IO a+type HandleTrailersStep a = HeaderList -> a -> HeaderList -> ClientIO a -data BiDiStep s m a =+data BiDiStep i o a = Abort -- ^ Finalize and return the current state.- | SendInput !CompressMode !(MethodInput s m)+ | SendInput !CompressMode !i -- ^ Sends a single message.- | WaitOutput (HandleMessageStep s m a) (HandleTrailersStep a)+ | WaitOutput (HandleMessageStep i o a) (HandleTrailersStep a) -- ^ Wait for information from the server, handlers can modify the state. -- | State-based function.-type RunBiDiStep s m a = a -> IO (a, BiDiStep s m a)+type RunBiDiStep s meth a = a -> ClientIO (a, BiDiStep s meth a) -- | gRPC call for a stepped bidirectional stream. --@@ -350,15 +350,15 @@ -- See 'BiDiStep' and 'RunBiDiStep' to understand the type of programs one can -- write with this function. steppedBiDiStream- :: (Service s, HasMethod s m, MethodStreamingType s m ~ 'BiDiStreaming)- => RPC s m+ :: (GRPCInput r i, GRPCOutput r o)+ => r -- ^ RPC to call. -> a -- ^ An initial state.- -> RunBiDiStep s m a+ -> RunBiDiStep i o a -- ^ The program.- -> RPCCall s m a-steppedBiDiStream rpc v0 handler = RPCCall $ \conn stream isfc streamFlowControl encoding decoding ->+ -> RPCCall r a+steppedBiDiStream rpc v0 handler = RPCCall rpc $ \conn stream isfc streamFlowControl encoding decoding -> let ocfc = _outgoingFlowControl conn decompress = _getDecodingCompression decoding newDecoder = decodeOutput rpc decompress@@ -372,49 +372,49 @@ Uncompressed -> uncompressed sendSingleMessage rpc msg (Encoding compress) id conn ocfc stream streamFlowControl handler v1 >>= goStep hdrs decode- goStep jhdrs@(Just hdrs) decode unchanged@(v1, WaitOutput handleMsg handleEof) = do+ goStep jhdrs@(Just hdrs) decode unchanged@(v1, WaitOutput handleMsg handleEof) = _waitEvent stream >>= \case- (StreamPushPromiseEvent _ _ _) ->- throwIO UnallowedPushPromiseReceived- (StreamHeadersEvent _ trls) -> do+ StreamPushPromiseEvent {} ->+ lift $ throwIO UnallowedPushPromiseReceived+ StreamHeadersEvent _ trls -> do v2 <- handleEof hdrs v1 trls -- TODO: consider failing the decoder here handler v2 >>= goStep jhdrs newDecoder- (StreamErrorEvent _ _) ->- throwIO (InvalidState "stream error")- (StreamDataEvent _ dat) -> do- _addCredit isfc (ByteString.length dat)- _ <- _consumeCredit isfc (ByteString.length dat)+ StreamErrorEvent _ _ ->+ lift $ throwIO (InvalidState "stream error")+ StreamDataEvent _ dat -> do+ liftIO $ _addCredit isfc (ByteString.length dat)+ _ <- liftIO $ _consumeCredit isfc (ByteString.length dat) _ <- _updateWindow isfc case pushChunk decode dat of- (Done unusedDat _ (Right val)) -> do+ Done unusedDat _ (Right val) -> do v2 <- handleMsg hdrs v1 val handler v2 >>= goStep jhdrs (pushChunk newDecoder unusedDat)- (Done _ _ (Left err)) -> do- throwIO $ InvalidParse $ "done-err: " ++ err- (Fail _ _ err) ->- throwIO $ InvalidParse $ "done-fail: " ++ err- partial@(Partial _) -> do+ Done _ _ (Left err) ->+ lift $ throwIO $ InvalidParse $ "done-err: " ++ err+ Fail _ _ err ->+ lift $ throwIO $ InvalidParse $ "done-fail: " ++ err+ partial@(Partial _) -> goStep jhdrs partial unchanged- goStep Nothing decode unchanged = do+ goStep Nothing decode unchanged = _waitEvent stream >>= \case- (StreamHeadersEvent _ hdrs) ->+ StreamHeadersEvent _ hdrs -> goStep (Just hdrs) decode unchanged _ ->- throwIO (InvalidState "no headers")+ lift $ throwIO (InvalidState "no headers") in handler v0 >>= goStep Nothing newDecoder -- | Exception raised when a BiDiStreaming RPC results in an invalid -- parse.-data InvalidParse = InvalidParse String deriving Show+newtype InvalidParse = InvalidParse String deriving Show instance Exception InvalidParse where -- | An event for the incoming loop of 'generalHandler'.-data IncomingEvent s m a =+data IncomingEvent o a = Headers HeaderList -- ^ The server sent some initial metadata with the headers.- | RecvMessage (MethodOutput s m)+ | RecvMessage o -- ^ The server send a message. | Trailers HeaderList -- ^ The server send final metadata (the loop stops).@@ -422,10 +422,10 @@ -- ^ Something went wrong (the loop stops). -- | An event for the outgoing loop of 'generalHandler'.-data OutgoingEvent s m b =+data OutgoingEvent i b = Finalize -- ^ The client is done with the RPC (the loop stops).- | SendMessage CompressMode (MethodInput s m)+ | SendMessage CompressMode i -- ^ The client sends a message to the server. -- | General RPC handler for decorrelating the handling of received@@ -440,19 +440,19 @@ -- and stops on Trailers or Invalid. The other loop waits for messages to send to -- the server or finalize and returns. generalHandler - :: (Service s, HasMethod s m)- => RPC s m+ :: (GRPCInput r i, GRPCOutput r o)+ => r -- ^ RPC to call. -> a -- ^ An initial state for the incoming loop.- -> (a -> IncomingEvent s m a -> IO a)+ -> (a -> IncomingEvent o a -> ClientIO a) -- ^ A state-passing function for the incoming loop. -> b -- ^ An initial state for the outgoing loop.- -> (b -> IO (b, OutgoingEvent s m b))+ -> (b -> ClientIO (b, OutgoingEvent i b)) -- ^ A state-passing function for the outgoing loop.- -> RPCCall s m (a,b)-generalHandler rpc v0 handle w0 next = RPCCall $ \conn stream isfc osfc encoding decoding ->+ -> RPCCall r (a,b)+generalHandler rpc v0 handle w0 next = RPCCall rpc $ \conn stream isfc osfc encoding decoding -> go conn stream isfc osfc encoding decoding where go conn stream isfc osfc encoding decoding =@@ -475,28 +475,28 @@ outGoingLoop v2 incomingLoop Nothing decode v1 = _waitEvent stream >>= \case- (StreamHeadersEvent _ hdrs) ->+ StreamHeadersEvent _ hdrs -> handle v1 (Headers hdrs) >>= incomingLoop (Just hdrs) decode _ -> handle v1 (Invalid $ toException $ InvalidState "no headers") incomingLoop jhdrs decode v1 = _waitEvent stream >>= \case- (StreamHeadersEvent _ hdrs) ->+ StreamHeadersEvent _ hdrs -> handle v1 (Trailers hdrs)- (StreamDataEvent _ dat) -> do- _addCredit isfc (ByteString.length dat)- _ <- _consumeCredit isfc (ByteString.length dat)+ StreamDataEvent _ dat -> do+ liftIO $ _addCredit isfc (ByteString.length dat)+ _ <- liftIO $ _consumeCredit isfc (ByteString.length dat) _ <- _updateWindow isfc case pushChunk decode dat of- (Done unusedDat _ (Right val)) ->+ Done unusedDat _ (Right val) -> handle v1 (RecvMessage val) >>= incomingLoop jhdrs (pushChunk newDecoder unusedDat)- partial@(Partial _) -> do+ partial@(Partial _) -> incomingLoop jhdrs partial v1- (Done _ _ (Left err)) -> do+ Done _ _ (Left err) -> handle v1 (Invalid $ toException $ InvalidParse $ "invalid-done-parse: " ++ err)- (Fail _ _ err) ->+ Fail _ _ err -> handle v1 (Invalid $ toException $ InvalidParse $ "invalid-parse: " ++ err)- (StreamPushPromiseEvent _ _ _) ->+ StreamPushPromiseEvent {} -> handle v1 (Invalid $ toException UnallowedPushPromiseReceived)- (StreamErrorEvent _ _) ->+ StreamErrorEvent {} -> handle v1 (Invalid $ toException $ InvalidState "stream error")
src/Network/GRPC/Client/Helpers.hs view
@@ -2,6 +2,7 @@ {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE GADTs #-} {-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE CPP #-} -- | Set of helpers helping with writing gRPC clients with not much exposure of -- the http2-client complexity.@@ -14,23 +15,22 @@ -- are not planned in this library and should be added at higher-levels. module Network.GRPC.Client.Helpers where -import Control.Lens-import Control.Concurrent.Async (Async, async, cancel)-import Control.Concurrent (threadDelay)+import Control.Concurrent.Async.Lifted (Async, async, cancel)+import Control.Concurrent.Lifted (threadDelay) import Control.Exception (throwIO) import Control.Monad (forever)+import Control.Monad.IO.Class (liftIO) import qualified Data.ByteString.Char8 as ByteString import Data.ByteString.Char8 (ByteString) import Data.Default.Class (def)-import Data.ProtoLens.Service.Types (Service(..), HasMethod, HasMethodImpl(..), StreamingType(..)) import qualified Network.TLS as TLS import qualified Network.TLS.Extra.Cipher as TLS import Network.HPACK (HeaderList) -import Network.HTTP2.Client (newHttp2FrameConnection, newHttp2Client, Http2Client(..), IncomingFlowControl(..), GoAwayHandler, FallBackFrameHandler, ignoreFallbackHandler, HostName, PortNumber, TooMuchConcurrency)+import Network.HTTP2.Client (ClientIO, ClientError, newHttp2FrameConnection, newHttp2Client, Http2Client(..), IncomingFlowControl(..), GoAwayHandler, FallBackFrameHandler, ignoreFallbackHandler, HostName, PortNumber, TooMuchConcurrency) import Network.HTTP2.Client.Helpers (ping)-import Network.GRPC.Client (RPC, open, singleRequest, streamReply, streamRequest, steppedBiDiStream, generalHandler, Authority, Timeout(..), StreamDone, CompressMode, RawReply, RunBiDiStep, IncomingEvent(..), OutgoingEvent(..))-import Network.GRPC.HTTP2.Encoding (Compression, Encoding(..), Decoding(..), gzip)+import Network.GRPC.Client+import Network.GRPC.HTTP2.Encoding -- | A simplified gRPC Client connected via an HTTP2Client to a given server. -- Each call from one client will share similar headers, timeout, compression.@@ -50,9 +50,9 @@ } data BackgroundTasks = BackgroundTasks {- backgroundWindowUpdate :: Async ()+ backgroundWindowUpdate :: Async (Either ClientError ()) -- ^ Periodically give the server credit to use the connection.- , backgroundPing :: Async ()+ , backgroundPing :: Async (Either ClientError ()) -- ^ Periodically ping the server. } @@ -82,7 +82,7 @@ grpcClientConfigSimple :: HostName -> PortNumber -> UseTlsOrNot -> GrpcClientConfig grpcClientConfigSimple host port tls =- GrpcClientConfig host port [] (Timeout 3000) gzip (tlsSettings tls host port) throwIO ignoreFallbackHandler 5000000 1000000+ GrpcClientConfig host port [] (Timeout 3000) gzip (tlsSettings tls host port) (liftIO . throwIO) ignoreFallbackHandler 5000000 1000000 type UseTlsOrNot = Bool @@ -98,10 +98,13 @@ } , TLS.clientSupported = def { TLS.supportedCiphers = TLS.ciphersuite_default } , TLS.clientDebug = def+#if MIN_VERSION_tls(1,5,0)+ , TLS.clientEarlyData = Nothing+#endif } -setupGrpcClient :: GrpcClientConfig -> IO GrpcClient+setupGrpcClient :: GrpcClientConfig -> ClientIO GrpcClient setupGrpcClient config = do let host = _grpcClientConfigHost config let port = _grpcClientConfigPort config@@ -125,7 +128,7 @@ return $ GrpcClient cli authority headers timeout compression tasks -- | Cancels background tasks and closes the underlying HTTP2 client.-close :: GrpcClient -> IO ()+close :: GrpcClient -> ClientIO () close grpc = do cancel $ backgroundPing $ _grpcClientBackground grpc cancel $ backgroundWindowUpdate $ _grpcClientBackground grpc@@ -133,44 +136,33 @@ -- | Run an unary query. rawUnary- :: (Service s, HasMethod s m)- => RPC s m+ :: (GRPCInput r i, GRPCOutput r o)+ => r -- ^ The RPC to call. -> GrpcClient -- ^ An initialized client.- -> MethodInput s m+ -> i -- ^ The input.- -> IO (Either TooMuchConcurrency (RawReply (MethodOutput s m)))+ -> ClientIO (Either TooMuchConcurrency (RawReply o)) rawUnary rpc (GrpcClient client authority headers timeout compression _) input = let call = singleRequest rpc input in open client authority headers timeout (Encoding compression) (Decoding compression) call --- | Prism helper to unpack an unary gRPC call output.------ @ out <- rawUnary rpc grpc method--- print $ out ^? unaryOutput . somefield--- @-unaryOutput- :: (Applicative f, Field3 a1 b1 (Either c1 a2) (Either c1 b2)) =>- (a2 -> f b2)- -> Either c2 (Either c3 a1) -> f (Either c2 (Either c3 b1))-unaryOutput = _Right . _Right . _3 . _Right- -- | Calls for a server stream of requests. rawStreamServer- :: (Service s, HasMethod s m, MethodStreamingType s m ~ 'ServerStreaming)- => RPC s m+ :: (GRPCInput r i, GRPCOutput r o)+ => r -- ^ The RPC to call. -> GrpcClient -- ^ An initialized client. -> a -- ^ An initial state.- -> MethodInput s m+ -> i -- ^ The input of the stream request.- -> (a -> HeaderList -> MethodOutput s m -> IO a)+ -> (a -> HeaderList -> o -> ClientIO a) -- ^ A state-passing handler called for each server-sent output. -- Headers are repeated for convenience but are the same for every iteration.- -> IO (Either TooMuchConcurrency (a, HeaderList, HeaderList))+ -> ClientIO (Either TooMuchConcurrency (a, HeaderList, HeaderList)) rawStreamServer rpc (GrpcClient client authority headers timeout compression _) v0 input handler = let call = streamReply rpc v0 input handler in open client authority headers timeout (Encoding compression) (Decoding compression) call@@ -180,16 +172,16 @@ -- Messages are submitted to the HTTP2 underlying client and hence this -- function can block until the HTTP2 client has some network credit. rawStreamClient- :: (Service s, HasMethod s m, MethodStreamingType s m ~ 'ClientStreaming)- => RPC s m+ :: (GRPCInput r i, GRPCOutput r o)+ => r -- ^ The RPC to call. -> GrpcClient -- ^ An initialized client. -> a -- ^ An initial state.- -> (a -> IO (a, Either StreamDone (CompressMode, MethodInput s m)))+ -> (a -> ClientIO (a, Either StreamDone (CompressMode, i))) -- ^ A state-passing step function to decide the next message.- -> IO (Either TooMuchConcurrency (a, (RawReply (MethodOutput s m))))+ -> ClientIO (Either TooMuchConcurrency (a, RawReply o)) rawStreamClient rpc (GrpcClient client authority headers timeout compression _) v0 getNext = let call = streamRequest rpc v0 getNext in open client authority headers timeout (Encoding compression) (Decoding compression) call@@ -200,16 +192,16 @@ -- protocols, that is, when after sending a message a client can know how many -- messages to wait for before sending the next message. rawSteppedBidirectional- :: (Service s, HasMethod s m, MethodStreamingType s m ~ 'BiDiStreaming)- => RPC s m+ :: (GRPCInput r i, GRPCOutput r o)+ => r -- ^ The RPC to call. -> GrpcClient -- ^ An initialized client. -> a -- ^ An initial state.- -> RunBiDiStep s m a+ -> RunBiDiStep i o a -- ^ The sequential program to iterate between sending and receiving messages.- -> IO (Either TooMuchConcurrency a)+ -> ClientIO (Either TooMuchConcurrency a) rawSteppedBidirectional rpc (GrpcClient client authority headers timeout compression _) v0 handler = let call = steppedBiDiStream rpc v0 handler in open client authority headers timeout (Encoding compression) (Decoding compression) call@@ -219,20 +211,20 @@ -- This handler allows to concurrently write messages and wait for incoming -- messages. rawGeneralStream- :: (Service s, HasMethod s m)- => RPC s m+ :: (GRPCInput r i, GRPCOutput r o)+ => r -- ^ The RPC to call. -> GrpcClient -- ^ An initialized client. -> a -- ^ An initial state for the incoming loop.- -> (a -> IncomingEvent s m a -> IO a)+ -> (a -> IncomingEvent o a -> ClientIO a) -- ^ A state-passing function for the incoming loop. -> b -- ^ An initial state for the outgoing loop.- -> (b -> IO (b, OutgoingEvent s m b))+ -> (b -> ClientIO (b, OutgoingEvent i b)) -- ^ A state-passing function for the ougoing loop.- -> IO (Either TooMuchConcurrency (a,b))+ -> ClientIO (Either TooMuchConcurrency (a,b)) rawGeneralStream rpc (GrpcClient client authority headers timeout compression _) v0 handler w0 next = let call = generalHandler rpc v0 handler w0 next in open client authority headers timeout (Encoding compression) (Decoding compression) call