http2-client-grpc 0.3.0.0 → 0.4.0.0

raw patch · 3 files changed

+190/−47 lines, 3 filesdep +tlsPVP ok

version bump matches the API change (PVP)

Dependencies added: tls

API changes (from Hackage documentation)

Files

• http2-client-grpc.cabal +3/−1
• src/Network/GRPC/Client.hs +58/−46
• src/Network/GRPC/Client/Helpers.hs +129/−0

@@ -1,5 +1,5 @@ name:                http2-client-grpc-version:             0.3.0.0+version:             0.4.0.0 synopsis:            Implement gRPC-over-HTTP2 clients. description:         Uses http2-client and proto-lens to generate client code. homepage:            https://github.com/lucasdicioccio/http2-client-grpc#readme@@ -16,6 +16,7 @@ library   hs-source-dirs:      src   exposed-modules:     Network.GRPC.Client+                     , Network.GRPC.Client.Helpers   build-depends:       base >= 4.7 && < 5                      , binary                      , bytestring@@ -26,6 +27,7 @@                      , proto-lens                      , proto-lens-protoc                      , text+                     , tls                      , zlib   default-language:    Haskell2010 

@@ -18,6 +18,7 @@   , singleRequest   , streamReply   , streamRequest+  , CompressMode(..)   , StreamDone(..)   -- * Errors.   , InvalidState(..)@@ -37,6 +38,7 @@ import Data.Binary.Get (Decoder(..), pushChunk, pushEndOfInput) import qualified Data.ByteString.Char8 as ByteString import Data.ProtoLens.Service.Types (Service(..), HasMethod, HasMethodImpl(..), StreamingType(..))+import GHC.TypeLits (Symbol)  import Network.GRPC.HTTP2.Types import Network.GRPC.HTTP2.Encoding@@ -68,15 +70,16 @@ throwOnPushPromise _ _ _ _ _ = throwIO UnallowedPushPromiseReceived  -- | Wait for an RPC reply.-waitReply :: (Service s, HasMethod s m) => RPC s m -> Compression -> Http2Stream -> IncomingFlowControl -> IO (RawReply (MethodOutput s m))-waitReply rpc compress stream flowControl = do+waitReply :: (Service s, HasMethod s m) => RPC s m -> Decoding -> Http2Stream -> IncomingFlowControl -> IO (RawReply (MethodOutput s m))+waitReply rpc decoding stream flowControl = do     format . fromStreamResult <$> waitStream stream flowControl throwOnPushPromise   where+    decompress = _getDecodingCompression decoding     format rsp = do        (hdrs, dat, trls) <- rsp        let res =              case lookup grpcMessageH hdrs of-               Nothing     -> fromDecoder $ pushEndOfInput $ flip pushChunk dat $ decodeOutput rpc compress+               Nothing     -> fromDecoder $ pushEndOfInput $ flip pushChunk dat $ decodeOutput rpc decompress                Just errMsg -> Left $ unpack errMsg         return (hdrs, trls, res)@@ -93,15 +96,17 @@  -- | Newtype helper used to uniformize all type of streaming modes when -- passing arguments to the 'open' call.-newtype RPCCall a = RPCCall {-    runRPC :: Http2Client -> Http2Stream -> IncomingFlowControl -> OutgoingFlowControl -> IO a+newtype RPCCall s (m ::Symbol) a = RPCCall {+    runRPC :: Http2Client -> Http2Stream -> IncomingFlowControl -> OutgoingFlowControl -> Encoding -> Decoding -> IO 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)-     => RPC s m-     -- ^ A token carrying information specifying the RPC to call.-     -> Http2Client+     => Http2Client      -- ^ A connected HTTP2 client.      -> Authority      -- ^ The HTTP2-Authority portion of the URL (e.g., "dicioccio.fr:7777").@@ -109,23 +114,24 @@      -- ^ A set of HTTP2 headers (e.g., for adding authentication headers).      -> Timeout      -- ^ Timeout in seconds.-     -> Compression-     -- ^ An indication of the compression that you will be using and accepting.  Compression-     -- should be per message, however a bug in gRPC-Go (to be confirmed) seems-     -- to turn message compression mandatory if advertised in the HTTP2-     -- headers, even though the specification states that compression per-     -- message is optional irrespectively of headers.-     -> RPCCall a+     -> Encoding+     -- ^ Compression used for encoding.+     -> Decoding+     -- ^ Compression allowed for decoding+     -> RPCCall s m a      -- ^ The actual RPC handler.      -> IO (Either TooMuchConcurrency a)-open rpc conn authority extraheaders timeout compression doStuff = do+open conn authority extraheaders timeout encoding decoding call = do+    let rpc = rpcFromCall call+    let compress = _getEncodingCompression encoding+    let decompress = _getDecodingCompression decoding     let request = [ (":method", "POST")                   , (":scheme", "http")                   , (":authority", authority)                   , (":path", path rpc)                    , (grpcTimeoutH, showTimeout timeout)-                  , (grpcEncodingH, grpcCompressionHV compression)-                  , (grpcAcceptEncodingH, mconcat [grpcAcceptEncodingHVdefault, ",", grpcCompressionHV compression])+                  , (grpcEncodingH, grpcCompressionHV compress)+                  , (grpcAcceptEncodingH, mconcat [grpcAcceptEncodingHVdefault, ",", grpcCompressionHV decompress])                   , ("content-type", grpcContentTypeHV)                   , ("te", "trailers")                   ] <> extraheaders@@ -133,25 +139,22 @@         let             initStream = headers stream request (setEndHeader)             handler isfc osfc = do-                (runRPC doStuff) conn stream 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-  -- ^ The RPC to call.-  -> Compression-  -- ^ The compression to use.-  -- TODO: revisit compressions.+  -- ^ RPC to call.   -> a   -- ^ An initial state.   -> MethodInput s m   -- ^ The input.   -> (a -> HeaderList -> MethodOutput s m -> IO a)   -- ^ A state-passing handler that is called with the message read.-  -> RPCCall (a, HeaderList, HeaderList)-streamReply rpc compress v0 req handler = RPCCall $ \conn stream isfc osfc -> do+  -> RPCCall s m (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 _ _ _) ->@@ -164,21 +167,23 @@                 _addCredit isfc (ByteString.length dat)                 _ <- _consumeCredit isfc (ByteString.length dat)                 _ <- _updateWindow isfc-                handleAllChunks v1 hdrs decode dat loop+                handleAllChunks decoding v1 hdrs decode dat loop     } in do         let ocfc = _outgoingFlowControl conn-        sendSingleMessage rpc req compress setEndStream conn ocfc stream osfc+        let decompress = _getDecodingCompression decoding+        sendSingleMessage rpc req encoding setEndStream conn ocfc stream osfc         _waitEvent stream >>= \case             StreamHeadersEvent _ hdrs ->-                loop v0 (decodeOutput rpc compress) hdrs+                loop v0 (decodeOutput rpc decompress) hdrs             _                         ->                 throwIO (InvalidState "no headers")   where-    handleAllChunks v1 hdrs decode dat exitLoop =+    handleAllChunks decoding v1 hdrs decode dat exitLoop =        case pushChunk decode dat of            (Done unusedDat _ (Right val)) -> do                v2 <- handler v1 hdrs val-               handleAllChunks v2 hdrs (decodeOutput rpc compress) unusedDat exitLoop+               let decompress = _getDecodingCompression decoding+               handleAllChunks decoding v2 hdrs (decodeOutput rpc decompress) unusedDat exitLoop            (Done unusedDat _ (Left err)) -> do                throwIO (StreamReplyDecodingError $ "done-error: " ++ err)            (Fail _ _ err)                 -> do@@ -188,6 +193,8 @@  data StreamDone = StreamDone +data CompressMode = Compressed | Uncompressed+ -- | gRPC call for Client Streaming. streamRequest   :: (Service s, HasMethod s m, MethodStreamingType s m ~ 'ClientStreaming)@@ -195,37 +202,41 @@   -- ^ RPC to call.   -> a   -- ^ An initial state.-  -> (a -> IO (Compression, a, Either StreamDone (MethodInput s m)))+  -> (a -> IO (a, Either StreamDone (CompressMode, MethodInput s m)))   -- ^ A state-passing action to retrieve the next message to send to the server.-  -> RPCCall (a, RawReply (MethodOutput s m))-streamRequest rpc v0 handler = RPCCall $ \conn stream isfc streamFlowControl ->+  -> RPCCall s m (a, RawReply (MethodOutput s m))+streamRequest rpc v0 handler = RPCCall $ \conn stream isfc streamFlowControl encoding decoding ->+    let decompress = _getDecodingCompression decoding in     let ocfc = _outgoingFlowControl conn         go v1 = do-            (compress, v2, nextEvent) <- handler v1+            (v2, nextEvent) <- handler v1             case nextEvent of-                Right msg -> do-                    sendSingleMessage rpc msg compress id conn ocfc stream streamFlowControl+                Right (doCompress, msg) -> do+                    let compress = case doCompress of+                            Compressed -> _getEncodingCompression encoding+                            Uncompressed -> uncompressed+                    sendSingleMessage rpc msg encoding id conn ocfc stream streamFlowControl                     go v2                 Left _ -> do                     sendData conn stream setEndStream ""-                    reply <- waitReply rpc compress stream isfc+                    reply <- waitReply rpc decoding stream isfc                     pure (v2, reply)     in go v0 - -- | Serialize and send a single message. sendSingleMessage   :: (Service s, HasMethod s m)   => RPC s m   -> MethodInput s m-  -> Compression+  -> Encoding   -> FlagSetter   -> Http2Client   -> OutgoingFlowControl   -> Http2Stream   -> OutgoingFlowControl   -> IO ()-sendSingleMessage rpc msg compression flagMod conn connectionFlowControl stream streamFlowControl = do+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@@ -237,16 +248,17 @@             else do                 sendData conn stream id (ByteString.take got dat)                 goUpload (ByteString.drop got dat)-    goUpload . toStrict . toLazyByteString . encodeInput rpc compression $ msg+    goUpload . toStrict . toLazyByteString . encodeInput rpc compress $ msg  -- | gRPC call for an unary request. singleRequest   :: (Service s, HasMethod s m)   => RPC s m-  -> Compression+  -- ^ RPC to call.   -> MethodInput s m-  -> RPCCall (RawReply (MethodOutput s m))-singleRequest rpc compress msg = RPCCall $ \conn stream isfc osfc -> do+  -- ^ RPC's input.+  -> RPCCall s m (RawReply (MethodOutput s m))+singleRequest rpc msg = RPCCall $ \conn stream isfc osfc encoding decoding -> do     let ocfc = _outgoingFlowControl conn-    sendSingleMessage rpc msg compress setEndStream conn ocfc stream osfc-    waitReply rpc compress stream isfc+    sendSingleMessage rpc msg encoding setEndStream conn ocfc stream osfc+    waitReply rpc decoding stream isfc

@@ -0,0 +1,129 @@+{-# LANGUAGE DataKinds         #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE GADTs             #-}++-- | Set of helpers helping with writing gRPC clients with not much exposure of+-- the http2-client complexity.+--+-- Th+module Network.GRPC.Client.Helpers where++import Control.Exception (throwIO)+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+import Network.HTTP2.Client+import Network.GRPC.Client+import Network.GRPC.HTTP2.Types+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.+data GrpcClient = GrpcClient {+    _grpcClientHttp2Client :: Http2Client+  -- ^ Underlying HTTP2 client.+  , _grpcClientAuthority   :: Authority+  -- ^ Authority header of the server the client is connected to.+  , _grpcClientHeaders     :: [(ByteString, ByteString)]+  -- ^ Extra HTTP2 headers to pass to every call (e.g., authentication tokens).+  , _grpcClientTimeout     :: Timeout+  -- ^ Timeout for RPCs.+  , _grpcClientCompression :: Compression+  -- ^ Compression shared for every call and expected for every answer.+  }++-- | Configuration to setup a GrpcClient.+data GrpcClientConfig = GrpcClientConfig {+    _grpcClientConfigHost            :: !HostName+  -- ^ Hostname of the server.+  , _grpcClientConfigPort            :: !PortNumber+  -- ^ Port of the server.+  , _grpcClientConfigHeaders         :: ![(ByteString, ByteString)]+  -- ^ Extra HTTP2 headers to pass to every call (e.g., authentication tokens).+  , _grpcClientConfigTimeout         :: !Timeout+  -- ^ Timeout for RPCs.+  , _grpcClientConfigCompression     :: !Compression+  -- ^ Compression shared for every call and expected for every answer.+  , _grpcClientConfigTLS             :: !(Maybe ClientParams)+  -- ^ TLS parameters for the session.+  , _grpcClientConfigGoAwayHandler   :: GoAwayHandler+  -- ^ HTTP2 handler for GoAways.+  , _grpcClientConfigFallbackHandler :: FallBackFrameHandler+  -- ^ HTTP2 handler for unhandled frames.+  }++grpcClientConfigSimple :: HostName -> PortNumber -> UseTlsOrNot -> GrpcClientConfig+grpcClientConfigSimple host port tls =+    GrpcClientConfig host port [] (Timeout 3000) gzip (tlsSettings tls host port) throwIO ignoreFallbackHandler++type UseTlsOrNot = Bool++tlsSettings :: UseTlsOrNot -> HostName -> PortNumber -> Maybe ClientParams+tlsSettings False _ _ = Nothing+tlsSettings True host port = Just $ TLS.ClientParams {+          TLS.clientWantSessionResume    = Nothing+        , TLS.clientUseMaxFragmentLength = Nothing+        , TLS.clientServerIdentification = (host, ByteString.pack $ show port)+        , TLS.clientUseServerNameIndication = True+        , TLS.clientShared               = def+        , TLS.clientHooks                = def { TLS.onServerCertificate = \_ _ _ _ -> return []+                                               }+        , TLS.clientSupported            = def { TLS.supportedCiphers = TLS.ciphersuite_default }+        , TLS.clientDebug                = def+         }+++setupGrpcClient :: GrpcClientConfig -> IO GrpcClient+setupGrpcClient config = do+  let host = _grpcClientConfigHost config+  let port = _grpcClientConfigPort config+  let tls = _grpcClientConfigTLS config+  let compression = _grpcClientConfigCompression config+  let onGoAway = _grpcClientConfigGoAwayHandler config+  let onFallback = _grpcClientConfigFallbackHandler config+  let timeout = _grpcClientConfigTimeout config+  let headers = _grpcClientConfigHeaders config+  let authority = ByteString.pack $ host <> ":" <> show port++  conn <- newHttp2FrameConnection host port tls+  cli <- newHttp2Client conn 8192 8192 [] onGoAway onFallback+  return $ GrpcClient cli authority headers timeout compression ++rawUnary+  :: (Service s, HasMethod s m)+  => RPC s m+  -> GrpcClient+  -> MethodInput s m+  -> IO (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++rawStreamServer +  :: (Service s, HasMethod s m, MethodStreamingType s m ~ 'ServerStreaming)+  => RPC s m+  -> GrpcClient+  -> a+  -> MethodInput s m+  -> (a -> HeaderList -> MethodOutput s m -> IO a)+  -> IO (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++rawStreamClient+  :: (Service s, HasMethod s m, MethodStreamingType s m ~ 'ClientStreaming)+  => RPC s m+  -> GrpcClient+  -> a+  -> (a -> IO (a, Either StreamDone (CompressMode, MethodInput s m)))+  -> IO (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