http2-client-grpc 0.4.0.0 → 0.5.0.0

raw patch · 3 files changed

+65/−26 lines, 3 filesdep +asyncdep +case-insensitivedep ~http2-grpc-typesPVP ok

version bump matches the API change (PVP)

Dependencies added: async, case-insensitive

Dependency ranges changed: http2-grpc-types

API changes (from Hackage documentation)

Files

• http2-client-grpc.cabal +4/−2
• src/Network/GRPC/Client.hs +20/−10
• src/Network/GRPC/Client/Helpers.hs +41/−14

@@ -1,5 +1,5 @@ name:                http2-client-grpc-version:             0.4.0.0+version:             0.5.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@@ -18,12 +18,14 @@   exposed-modules:     Network.GRPC.Client                      , Network.GRPC.Client.Helpers   build-depends:       base >= 4.7 && < 5+                     , async                      , binary                      , bytestring+                     , case-insensitive                      , data-default-class                      , http2                      , http2-client-                     , http2-grpc-types >= 0.2+                     , http2-grpc-types >= 0.3                      , proto-lens                      , proto-lens-protoc                      , text

@@ -31,12 +31,14 @@   ) where  import Control.Exception (Exception(..), throwIO)-import Data.Monoid ((<>)) import Data.ByteString.Char8 (unpack) import Data.ByteString.Lazy (toStrict) import Data.Binary.Builder (toLazyByteString) import Data.Binary.Get (Decoder(..), pushChunk, pushEndOfInput) import qualified Data.ByteString.Char8 as ByteString+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) @@ -47,6 +49,8 @@ import Network.HTTP2.Client import Network.HTTP2.Client.Helpers +type CIHeaderList = [(CI ByteString.ByteString, ByteString.ByteString)]+ -- | A reply. -- -- This reply object contains a lot of information because a single gRPC call@@ -57,7 +61,7 @@ -- - 1st item: initial HTTP2 response -- - 2nd item: second (trailers) HTTP2 response -- - 3rd item: proper gRPC answer-type RawReply a = Either ErrorCode (HeaderList, Maybe HeaderList, (Either String a))+type RawReply a = Either ErrorCode (CIHeaderList, Maybe CIHeaderList, (Either String a))  -- | gRPC disables HTTP2 push-promises. --@@ -77,13 +81,20 @@     decompress = _getDecodingCompression decoding     format rsp = do        (hdrs, dat, trls) <- rsp+       let hdrs2 = headerstoCIHeaders hdrs+       let trls2 = fmap headerstoCIHeaders trls        let res =-             case lookup grpcMessageH hdrs of+             -- 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 -       return (hdrs, trls, res)+       return (hdrs2, trls2, res) +headerstoCIHeaders :: HeaderList -> CIHeaderList+headerstoCIHeaders hdrs = [(CI.mk k, v) | (k,v) <- hdrs]+ -- | Exception raised when a ServerStreaming RPC results in a decoding -- error. data StreamReplyDecodingError = StreamReplyDecodingError String deriving Show@@ -129,9 +140,9 @@                   , (":scheme", "http")                   , (":authority", authority)                   , (":path", path rpc) -                  , (grpcTimeoutH, showTimeout timeout)-                  , (grpcEncodingH, grpcCompressionHV compress)-                  , (grpcAcceptEncodingH, mconcat [grpcAcceptEncodingHVdefault, ",", grpcCompressionHV decompress])+                  , (CI.original grpcTimeoutH, showTimeout timeout)+                  , (CI.original grpcEncodingH, grpcCompressionHV compress)+                  , (CI.original grpcAcceptEncodingH, mconcat [grpcAcceptEncodingHVdefault, ",", grpcCompressionHV decompress])                   , ("content-type", grpcContentTypeHV)                   , ("te", "trailers")                   ] <> extraheaders@@ -184,7 +195,7 @@                v2 <- handler v1 hdrs val                let decompress = _getDecodingCompression decoding                handleAllChunks decoding v2 hdrs (decodeOutput rpc decompress) unusedDat exitLoop-           (Done unusedDat _ (Left err)) -> do+           (Done _ _ (Left err)) -> do                throwIO (StreamReplyDecodingError $ "done-error: " ++ err)            (Fail _ _ err)                 -> do                throwIO (StreamReplyDecodingError $ "fail-error: " ++ err)@@ -206,7 +217,6 @@   -- ^ 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 ->-    let decompress = _getDecodingCompression decoding in     let ocfc = _outgoingFlowControl conn         go v1 = do             (v2, nextEvent) <- handler v1@@ -215,7 +225,7 @@                     let compress = case doCompress of                             Compressed -> _getEncodingCompression encoding                             Uncompressed -> uncompressed-                    sendSingleMessage rpc msg encoding id conn ocfc stream streamFlowControl+                    sendSingleMessage rpc msg (Encoding compress) id conn ocfc stream streamFlowControl                     go v2                 Left _ -> do                     sendData conn stream setEndStream ""

@@ -5,10 +5,18 @@ -- | Set of helpers helping with writing gRPC clients with not much exposure of -- the http2-client complexity. ----- Th+-- The GrpcClient handles automatic background connection-level window updates+-- to prevent the connection from starving and pings to force a connection+-- alive.+--+-- There is no automatic reconnection, retry, or healthchecking. These features+-- are not planned in this library and should be added at higher-levels. module Network.GRPC.Client.Helpers where +import Control.Concurrent.Async (Async, async)+import Control.Concurrent (threadDelay) import Control.Exception (throwIO)+import Control.Monad (forever) import qualified Data.ByteString.Char8 as ByteString import Data.ByteString.Char8 (ByteString) import Data.Default.Class (def)@@ -17,11 +25,10 @@ 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+import Network.HTTP2.Client (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, Authority, Timeout(..), StreamDone, CompressMode, RawReply)+import Network.GRPC.HTTP2.Encoding (Compression, Encoding(..), Decoding(..), gzip)  -- | A simplified gRPC Client connected via an HTTP2Client to a given server. -- Each call from one client will share similar headers, timeout, compression.@@ -36,8 +43,17 @@   -- ^ Timeout for RPCs.   , _grpcClientCompression :: Compression   -- ^ Compression shared for every call and expected for every answer.+  , _grpcClientBackground  :: BackgroundTasks+  -- ^ Running background tasks.   } +data BackgroundTasks = BackgroundTasks {+    backgroundWindowUpdate :: Async ()+  -- ^ Periodically give the server credit to use the connection.+  , backgroundPing         :: Async ()+  -- ^ Periodically ping the server.+  }+ -- | Configuration to setup a GrpcClient. data GrpcClientConfig = GrpcClientConfig {     _grpcClientConfigHost            :: !HostName@@ -50,21 +66,25 @@   -- ^ Timeout for RPCs.   , _grpcClientConfigCompression     :: !Compression   -- ^ Compression shared for every call and expected for every answer.-  , _grpcClientConfigTLS             :: !(Maybe ClientParams)+  , _grpcClientConfigTLS             :: !(Maybe TLS.ClientParams)   -- ^ TLS parameters for the session.   , _grpcClientConfigGoAwayHandler   :: GoAwayHandler   -- ^ HTTP2 handler for GoAways.   , _grpcClientConfigFallbackHandler :: FallBackFrameHandler   -- ^ HTTP2 handler for unhandled frames.+  , _grpcClientConfigWindowUpdateDelay :: Int+  -- ^ Delay in microsecond between to window updates.+  , _grpcClientConfigPingDelay         :: Int+  -- ^ Delay in microsecond between to pings.   }  grpcClientConfigSimple :: HostName -> PortNumber -> UseTlsOrNot -> GrpcClientConfig grpcClientConfigSimple host port tls =-    GrpcClientConfig host port [] (Timeout 3000) gzip (tlsSettings tls host port) throwIO ignoreFallbackHandler+    GrpcClientConfig host port [] (Timeout 3000) gzip (tlsSettings tls host port) throwIO ignoreFallbackHandler 5000000 1000000  type UseTlsOrNot = Bool -tlsSettings :: UseTlsOrNot -> HostName -> PortNumber -> Maybe ClientParams+tlsSettings :: UseTlsOrNot -> HostName -> PortNumber -> Maybe TLS.ClientParams tlsSettings False _ _ = Nothing tlsSettings True host port = Just $ TLS.ClientParams {           TLS.clientWantSessionResume    = Nothing@@ -76,7 +96,7 @@                                                }         , TLS.clientSupported            = def { TLS.supportedCiphers = TLS.ciphersuite_default }         , TLS.clientDebug                = def-         }+        }   setupGrpcClient :: GrpcClientConfig -> IO GrpcClient@@ -93,7 +113,14 @@    conn <- newHttp2FrameConnection host port tls   cli <- newHttp2Client conn 8192 8192 [] onGoAway onFallback-  return $ GrpcClient cli authority headers timeout compression +  wuAsync <- async $ forever $ do+      threadDelay $ _grpcClientConfigWindowUpdateDelay config+      _updateWindow $ _incomingFlowControl cli+  pingAsync <- async $ forever $ do+      threadDelay $ _grpcClientConfigPingDelay config+      ping cli 3000000 "grpc.hs"+  let tasks = BackgroundTasks wuAsync pingAsync+  return $ GrpcClient cli authority headers timeout compression tasks  rawUnary   :: (Service s, HasMethod s m)@@ -101,7 +128,7 @@   -> GrpcClient   -> MethodInput s m   -> IO (Either TooMuchConcurrency (RawReply (MethodOutput s m)))-rawUnary rpc (GrpcClient client authority headers timeout compression) input =+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 @@ -113,7 +140,7 @@   -> 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 =+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 @@ -124,6 +151,6 @@   -> 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 =+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