http2-client-grpc 0.6.0.0 → 0.7.0.0
raw patch · 4 files changed
+89/−81 lines, 4 filesdep +lifted-asyncdep +lifted-basedep ~http2-client
Dependencies added: lifted-async, lifted-base
Dependency ranges changed: http2-client
Files
- Changelog.md +4/−0
- http2-client-grpc.cabal +4/−2
- src/Network/GRPC/Client.hs +63/−62
- src/Network/GRPC/Client/Helpers.hs +18/−17
Changelog.md view
@@ -2,6 +2,10 @@ ## Unreleased changes +## 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.
http2-client-grpc.cabal view
@@ -1,5 +1,5 @@ name: http2-client-grpc-version: 0.6.0.0+version: 0.7.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@@ -24,8 +24,10 @@ , case-insensitive >= 1.2.0 && < 1.3 , data-default-class >= 0.1 && <0.2 , lens >= 4.16 && < 4.18+ , lifted-async >= 0.10 && < 0.11+ , lifted-base >= 0.2 && < 0.3 , http2 >= 1.6 && < 1.7- , http2-client >= 0.8 && < 0.9+ , http2-client >= 0.9 && < 0.10 , http2-grpc-types >= 0.4 && < 0.5 , proto-lens >= 0.5 && < 0.6 , text >= 1.2 && < 1.3
src/Network/GRPC/Client.hs view
@@ -81,8 +81,9 @@ , HeaderList ) where -import Control.Concurrent.Async (concurrently)+import Control.Concurrent.Async.Lifted (concurrently) import Control.Exception (SomeException, Exception(..), throwIO)+import Control.Monad.IO.Class (liftIO) import Data.ByteString.Char8 (unpack) import Data.ByteString.Lazy (toStrict) import Data.Binary.Builder (toLazyByteString)@@ -124,10 +125,10 @@ -- | 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 :: (Service s, HasMethod s meth) => RPC s meth -> Decoding -> Http2Stream -> IncomingFlowControl -> ClientIO (RawReply (MethodOutput s meth)) waitReply rpc decoding stream flowControl = do format . fromStreamResult <$> waitStream stream flowControl throwOnPushPromise where@@ -161,16 +162,16 @@ -- | 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+newtype RPCCall s (meth ::Symbol) a = RPCCall {+ 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 :: RPCCall s meth a -> RPC s meth rpcFromCall _ = RPC -- | Main handler to perform gRPC calls to a service.-open :: (Service s, HasMethod s m)+open :: (Service s, HasMethod s meth) => Http2Client -- ^ A connected HTTP2 client. -> Authority@@ -183,9 +184,9 @@ -- ^ Compression used for encoding. -> Decoding -- ^ Compression allowed for decoding- -> RPCCall s m a+ -> RPCCall s meth 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@@ -209,28 +210,28 @@ -- | gRPC call for Server Streaming. streamReply- :: (Service s, HasMethod s m, MethodStreamingType s m ~ 'ServerStreaming)- => RPC s m+ :: (Service s, HasMethod s meth, MethodStreamingType s meth ~ 'ServerStreaming)+ => RPC s meth -- ^ RPC to call. -> a -- ^ An initial state.- -> MethodInput s m+ -> MethodInput s meth -- ^ The input.- -> (a -> HeaderList -> MethodOutput s m -> IO a)+ -> (a -> HeaderList -> MethodOutput s meth -> ClientIO a) -- ^ A state-passing handler that is called with the message read.- -> RPCCall s m (a, HeaderList, HeaderList)+ -> RPCCall s meth (a, HeaderList, HeaderList) streamReply rpc v0 req handler = RPCCall $ \conn stream isfc osfc encoding decoding -> do let { loop v1 decode hdrs = _waitEvent stream >>= \case (StreamPushPromiseEvent _ _ _) ->- throwIO UnallowedPushPromiseReceived+ lift $ throwIO UnallowedPushPromiseReceived (StreamHeadersEvent _ trls) -> return (v1, hdrs, trls) (StreamErrorEvent _ _) ->- throwIO (InvalidState "stream error")+ lift $ throwIO (InvalidState "stream error") (StreamDataEvent _ dat) -> do- _addCredit isfc (ByteString.length dat)- _ <- _consumeCredit isfc (ByteString.length dat)+ liftIO $ _addCredit isfc (ByteString.length dat)+ _ <- liftIO $ _consumeCredit isfc (ByteString.length dat) _ <- _updateWindow isfc handleAllChunks decoding v1 hdrs decode dat loop } in do@@ -241,7 +242,7 @@ 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@@ -250,9 +251,9 @@ let decompress = _getDecodingCompression decoding handleAllChunks decoding v2 hdrs (decodeOutput rpc decompress) unusedDat exitLoop (Done _ _ (Left err)) -> do- throwIO (StreamReplyDecodingError $ "done-error: " ++ err)+ lift $ throwIO (StreamReplyDecodingError $ "done-error: " ++ err) (Fail _ _ err) -> do- throwIO (StreamReplyDecodingError $ "fail-error: " ++ err)+ lift $ throwIO (StreamReplyDecodingError $ "fail-error: " ++ err) partial@(Partial _) -> exitLoop v1 partial hdrs @@ -262,14 +263,14 @@ -- | gRPC call for Client Streaming. streamRequest- :: (Service s, HasMethod s m, MethodStreamingType s m ~ 'ClientStreaming)- => RPC s m+ :: (Service s, HasMethod s meth, MethodStreamingType s meth ~ 'ClientStreaming)+ => RPC s meth -- ^ RPC to call. -> a -- ^ An initial state.- -> (a -> IO (a, Either StreamDone (CompressMode, MethodInput s m)))+ -> (a -> ClientIO (a, Either StreamDone (CompressMode, MethodInput s meth))) -- ^ A state-passing action to retrieve the next message to send to the server.- -> RPCCall s m (a, RawReply (MethodOutput s m))+ -> RPCCall s meth (a, RawReply (MethodOutput s meth)) streamRequest rpc v0 handler = RPCCall $ \conn stream isfc streamFlowControl encoding decoding -> let ocfc = _outgoingFlowControl conn go v1 = do@@ -289,23 +290,23 @@ -- | Serialize and send a single message. sendSingleMessage- :: (Service s, HasMethod s m)- => RPC s m- -> MethodInput s m+ :: (Service s, HasMethod s meth)+ => RPC s meth+ -> MethodInput s meth -> 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 +317,32 @@ -- | gRPC call for an unary request. singleRequest- :: (Service s, HasMethod s m)- => RPC s m+ :: (Service s, HasMethod s meth)+ => RPC s meth -- ^ RPC to call.- -> MethodInput s m+ -> MethodInput s meth -- ^ RPC's input.- -> RPCCall s m (RawReply (MethodOutput s m))+ -> RPCCall s meth (RawReply (MethodOutput s meth)) singleRequest rpc msg = RPCCall $ \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 s meth a = HeaderList -> a -> MethodOutput s meth -> 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 s meth a = Abort -- ^ Finalize and return the current state.- | SendInput !CompressMode !(MethodInput s m)+ | SendInput !CompressMode !(MethodInput s meth) -- ^ Sends a single message.- | WaitOutput (HandleMessageStep s m a) (HandleTrailersStep a)+ | WaitOutput (HandleMessageStep s meth 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,14 +351,14 @@ -- 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+ :: (Service s, HasMethod s meth, MethodStreamingType s meth ~ 'BiDiStreaming)+ => RPC s meth -- ^ RPC to call. -> a -- ^ An initial state.- -> RunBiDiStep s m a+ -> RunBiDiStep s meth a -- ^ The program.- -> RPCCall s m a+ -> RPCCall s meth a steppedBiDiStream rpc v0 handler = RPCCall $ \conn stream isfc streamFlowControl encoding decoding -> let ocfc = _outgoingFlowControl conn decompress = _getDecodingCompression decoding@@ -375,25 +376,25 @@ goStep jhdrs@(Just hdrs) decode unchanged@(v1, WaitOutput handleMsg handleEof) = do _waitEvent stream >>= \case (StreamPushPromiseEvent _ _ _) ->- throwIO UnallowedPushPromiseReceived+ 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")+ lift $ throwIO (InvalidState "stream error") (StreamDataEvent _ dat) -> do- _addCredit isfc (ByteString.length dat)- _ <- _consumeCredit isfc (ByteString.length dat)+ liftIO $ _addCredit isfc (ByteString.length dat)+ _ <- liftIO $ _consumeCredit isfc (ByteString.length dat) _ <- _updateWindow isfc case pushChunk decode dat of (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+ lift $ throwIO $ InvalidParse $ "done-err: " ++ err (Fail _ _ err) ->- throwIO $ InvalidParse $ "done-fail: " ++ err+ lift $ throwIO $ InvalidParse $ "done-fail: " ++ err partial@(Partial _) -> do goStep jhdrs partial unchanged goStep Nothing decode unchanged = do@@ -401,7 +402,7 @@ (StreamHeadersEvent _ hdrs) -> goStep (Just hdrs) decode unchanged _ ->- throwIO (InvalidState "no headers")+ lift $ throwIO (InvalidState "no headers") in handler v0 >>= goStep Nothing newDecoder @@ -411,10 +412,10 @@ instance Exception InvalidParse where -- | An event for the incoming loop of 'generalHandler'.-data IncomingEvent s m a =+data IncomingEvent s meth a = Headers HeaderList -- ^ The server sent some initial metadata with the headers.- | RecvMessage (MethodOutput s m)+ | RecvMessage (MethodOutput s meth) -- ^ The server send a message. | Trailers HeaderList -- ^ The server send final metadata (the loop stops).@@ -422,10 +423,10 @@ -- ^ Something went wrong (the loop stops). -- | An event for the outgoing loop of 'generalHandler'.-data OutgoingEvent s m b =+data OutgoingEvent s meth b = Finalize -- ^ The client is done with the RPC (the loop stops).- | SendMessage CompressMode (MethodInput s m)+ | SendMessage CompressMode (MethodInput s meth) -- ^ The client sends a message to the server. -- | General RPC handler for decorrelating the handling of received@@ -440,18 +441,18 @@ -- 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+ :: (Service s, HasMethod s meth)+ => RPC s meth -- ^ RPC to call. -> a -- ^ An initial state for the incoming loop.- -> (a -> IncomingEvent s m a -> IO a)+ -> (a -> IncomingEvent s meth 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 s meth b)) -- ^ A state-passing function for the outgoing loop.- -> RPCCall s m (a,b)+ -> RPCCall s meth (a,b) generalHandler rpc v0 handle w0 next = RPCCall $ \conn stream isfc osfc encoding decoding -> go conn stream isfc osfc encoding decoding where@@ -484,8 +485,8 @@ (StreamHeadersEvent _ hdrs) -> handle v1 (Trailers hdrs) (StreamDataEvent _ dat) -> do- _addCredit isfc (ByteString.length dat)- _ <- _consumeCredit isfc (ByteString.length dat)+ liftIO $ _addCredit isfc (ByteString.length dat)+ _ <- liftIO $ _consumeCredit isfc (ByteString.length dat) _ <- _updateWindow isfc case pushChunk decode dat of (Done unusedDat _ (Right val)) ->
src/Network/GRPC/Client/Helpers.hs view
@@ -15,10 +15,11 @@ 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)@@ -27,7 +28,7 @@ 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)@@ -50,9 +51,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 +83,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 @@ -101,7 +102,7 @@ } -setupGrpcClient :: GrpcClientConfig -> IO GrpcClient+setupGrpcClient :: GrpcClientConfig -> ClientIO GrpcClient setupGrpcClient config = do let host = _grpcClientConfigHost config let port = _grpcClientConfigPort config@@ -125,7 +126,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@@ -140,7 +141,7 @@ -- ^ An initialized client. -> MethodInput s m -- ^ The input.- -> IO (Either TooMuchConcurrency (RawReply (MethodOutput s m)))+ -> ClientIO (Either TooMuchConcurrency (RawReply (MethodOutput s m))) 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@@ -167,10 +168,10 @@ -- ^ An initial state. -> MethodInput s m -- ^ The input of the stream request.- -> (a -> HeaderList -> MethodOutput s m -> IO a)+ -> (a -> HeaderList -> MethodOutput s m -> 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@@ -187,9 +188,9 @@ -- ^ An initialized client. -> a -- ^ An initial state.- -> (a -> IO (a, Either StreamDone (CompressMode, MethodInput s m)))+ -> (a -> ClientIO (a, Either StreamDone (CompressMode, MethodInput s m))) -- ^ A state-passing step function to decide the next message.- -> IO (Either TooMuchConcurrency (a, (RawReply (MethodOutput s m))))+ -> ClientIO (Either TooMuchConcurrency (a, (RawReply (MethodOutput s m)))) 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@@ -209,7 +210,7 @@ -- ^ An initial state. -> RunBiDiStep s m 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@@ -226,13 +227,13 @@ -- ^ An initialized client. -> a -- ^ An initial state for the incoming loop.- -> (a -> IncomingEvent s m a -> IO a)+ -> (a -> IncomingEvent s m 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 s m 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