http2-client-grpc 0.5.0.1 → 0.5.0.2

raw patch · 4 files changed

+257/−25 lines, 4 filesdep −zlibdep ~asyncdep ~basedep ~binaryPVP: minor bump suggested

API additions: PVP suggests at least a minor version bump

Dependencies removed: zlib

Dependency ranges changed: async, base, binary, bytestring, case-insensitive, data-default-class, http2, http2-client, http2-grpc-types, lens, proto-lens, proto-lens-protoc, text, tls

API changes (from Hackage documentation)

Files

• Changelog.md +7/−0
• README.md +2/−2
• http2-client-grpc.cabal +17/−18
• src/Network/GRPC/Client.hs +231/−5

@@ -0,0 +1,7 @@+# Changelog for http2-client-grpc++## Unreleased changes++## 0.5.0.2++Add bidirectional and general streams.

@@ -69,8 +69,8 @@  This library is currently an early-stage library. Expect breaking changes. -## TODOs+## Next steps  - pass timeout as argument-- support effectful compression (Zlib pure compression is supported) - provide function to map raw results into commonly-understood GRPC errors+- (nice-to-have) support effectful compression (Zlib pure compression is supported)

@@ -1,7 +1,7 @@ name:                http2-client-grpc-version:             0.5.0.1+version:             0.5.0.2 synopsis:            Implement gRPC-over-HTTP2 clients.-description:         Uses http2-client and proto-lens to generate client code.+description:         A gRPC over http2-client using proto-lens to generate client code. homepage:            https://github.com/lucasdicioccio/http2-client-grpc#readme license:             BSD3 license-file:        LICENSE@@ -10,28 +10,27 @@ copyright:           2017 Lucas DiCioccio category:            Network build-type:          Simple-extra-source-files:  README.md+extra-source-files:  README.md Changelog.md cabal-version:       >=1.10  library   hs-source-dirs:      src   exposed-modules:     Network.GRPC.Client                      , Network.GRPC.Client.Helpers-  build-depends:       base >= 4.7 && < 5-                     , async-                     , binary-                     , bytestring-                     , case-insensitive-                     , data-default-class-                     , lens-                     , http2-                     , http2-client-                     , http2-grpc-types >= 0.3-                     , proto-lens-                     , proto-lens-protoc-                     , text-                     , tls-                     , zlib+  build-depends:       base >= 4.11 && < 5+                     , async >= 2.2 && < 2.3+                     , binary >= 0.8 && < 0.9+                     , bytestring >= 0.10.8 && < 0.10.9+                     , 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.3 && < 0.4+                     , proto-lens-protoc >= 0.3 && < 0.4+                     , text >= 1.2 && < 1.3+                     , tls >= 1.4 && < 1.5   default-language:    Haskell2010  test-suite http2-client-grpc-test

@@ -7,6 +7,44 @@ {-# LANGUAGE TypeFamilies        #-}  -- | A module adding support for gRPC over HTTP2.+--+-- This module provides helpers to encode gRPC queries on top of HTTP2 client.+--+-- The helpers we provide for streaming RPCs ('streamReply', 'streamRequest',+-- 'steppedBiDiStream') are a subset of what gRPC allows: the gRPC definition+-- of streaming RPCs offers a large amount of valid behaviors regarding timing+-- of headers, trailers, and end of streams.+--+-- The limitations of these functions should be clear from the type signatures.+-- But in general, the design is to only allow synchronous state machines. Such+-- state-machines cannot immediately react to server-sent messages but must+-- wait for the client code to poll for some server-sent information.  In+-- short, these handlers prevents programs from observing intermediary steps+-- which may be valid applications by gRPC standards. Simply put, it is not+-- possibly to simultaneously wait for some information from the server and+-- send some information.+-- For instance, in a client-streaming RPC, the server is allowed to send+-- trailers at any time, even before receiving any input message. The+-- 'streamRequest' functions disallows reading trailers until the client code+-- is done sending requests.+-- A result from this design choice is to offer a simple programming surface+-- for the most common use cases. Further, these simple state-machines require+-- little runtime overhead.+--+-- A more general handler 'generalHandler' is provided which runs two thread in+-- parallel. This handler allows to send an receive messages concurrently using+-- one loop each, which allows to circumvent the limitations of the above+-- handlers (but at a cost: complexity and threading overhead). It also means+-- that a sending action may be stuck indefinitely on flow-control and cannot+-- be cancelled without killing the 'RPC' thread. You see where we are going:+-- the more elaborate the semantics, the more a programmer has to think.+--+-- Though, all hope of expressing wacky application semantics is not lost: it+-- is always possible to write its own 'RPC' function.  Writing one's own 'RPC'+-- function allows to leverage the specific semantics of the RPC call to save+-- some overhead (much like the three above streaming helpers assume a simple+-- behavior from the server). Hence, it is generally a good idea to take+-- inspiration from the existing RPC functions and learn how to write one. module Network.GRPC.Client (   -- * Building blocks.     RPC(..)@@ -18,24 +56,37 @@   , singleRequest   , streamReply   , streamRequest+  , steppedBiDiStream+  , generalHandler   , CompressMode(..)   , StreamDone(..)+  , BiDiStep(..)+  , RunBiDiStep+  , HandleMessageStep+  , HandleTrailersStep+  , IncomingEvent(..)+  , OutgoingEvent(..)   -- * Errors.   , InvalidState(..)   , StreamReplyDecodingError(..)   , UnallowedPushPromiseReceived(..)+  , InvalidParse(..)   -- * Compression of individual messages.   , Compression   , gzip   , uncompressed+  -- * Re-exports.+  , HeaderList   ) where -import Control.Exception (Exception(..), throwIO)+import Control.Concurrent.Async (concurrently)+import Control.Exception (SomeException, Exception(..), throwIO) 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.ByteString.Char8 (ByteString) import Data.CaseInsensitive (CI) import qualified Data.CaseInsensitive as CI import Data.Monoid ((<>))@@ -46,10 +97,10 @@ import Network.GRPC.HTTP2.Encoding import Network.HTTP2 import Network.HPACK-import Network.HTTP2.Client+import Network.HTTP2.Client hiding (next) import Network.HTTP2.Client.Helpers -type CIHeaderList = [(CI ByteString.ByteString, ByteString.ByteString)]+type CIHeaderList = [(CI ByteString, ByteString)]  -- | A reply. --@@ -102,7 +153,8 @@  -- | Exception raised when a ServerStreaming RPC results in an invalid -- state machine.-data InvalidState = InvalidState String deriving Show+data InvalidState = InvalidState String+  deriving Show instance Exception InvalidState where  -- | Newtype helper used to uniformize all type of streaming modes when@@ -169,7 +221,7 @@     let {         loop v1 decode hdrs = _waitEvent stream >>= \case             (StreamPushPromiseEvent _ _ _) ->-                throwIO (InvalidState "push promise")+                throwIO UnallowedPushPromiseReceived             (StreamHeadersEvent _ trls) ->                 return (v1, hdrs, trls)             (StreamErrorEvent _ _) ->@@ -272,3 +324,177 @@     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+-- | Handler for received trailers.+type HandleTrailersStep a = HeaderList -> a -> HeaderList -> IO a++data BiDiStep s m a =+    Abort+  -- ^ Finalize and return the current state.+  | SendInput !CompressMode !(MethodInput s m)+  -- ^ Sends a single message.+  | WaitOutput (HandleMessageStep s m 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)++-- | gRPC call for a stepped bidirectional stream.+--+-- This helper limited.+--+-- 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+  -- ^ RPC to call.+  -> a+  -- ^ An initial state.+  -> RunBiDiStep s m a+  -- ^ The program.+  -> RPCCall s m a+steppedBiDiStream rpc v0 handler = RPCCall $ \conn stream isfc streamFlowControl encoding decoding ->+    let ocfc = _outgoingFlowControl conn+        decompress = _getDecodingCompression decoding+        newDecoder = decodeOutput rpc decompress++        goStep _ _ (v1, Abort) = do+            sendData conn stream setEndStream ""+            pure v1+        goStep hdrs decode (v1, SendInput doCompress msg) = do+            let compress = case doCompress of+                    Compressed -> _getEncodingCompression encoding+                    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+            _waitEvent stream >>= \case+                (StreamPushPromiseEvent _ _ _) ->+                    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)+                    _ <- _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+                        (Fail _ _ err)         ->+                            throwIO $ InvalidParse $ "done-fail: " ++ err+                        partial@(Partial _)    -> do+                            goStep jhdrs partial unchanged+        goStep Nothing decode unchanged = do+            _waitEvent stream >>= \case+                (StreamHeadersEvent _ hdrs) ->+                   goStep (Just hdrs) decode unchanged+                _ ->+                   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+instance Exception InvalidParse where++-- | An event for the incoming loop of 'generalHandler'.+data IncomingEvent s m a =+    Headers HeaderList+  -- ^ The server sent some initial metadata with the headers.+  | RecvMessage (MethodOutput s m)+  -- ^ The server send a message.+  | Trailers HeaderList+  -- ^ The server send final metadata (the loop stops).+  | Invalid SomeException+  -- ^ Something went wrong (the loop stops).++-- | An event for the outgoing loop of 'generalHandler'.+data OutgoingEvent s m b =+    Finalize+  -- ^ The client is done with the RPC (the loop stops).+  | SendMessage CompressMode (MethodInput s m)+  -- ^ The client sends a message to the server.++-- | General RPC handler for decorrelating the handling of received+-- headers/trailers from the sending of messages.+--+-- There is no constraints on the stream-arity of the RPC. It requires a bit of+-- viligence to avoid breaking the gRPC semantics but this one is easy to pay+-- attention to.+--+-- This handler runs two loops 'concurrently':+-- One loop accepts and chunks messages from the HTTP2 stream, then return events+-- 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+  -- ^ RPC to call.+  -> a+  -- ^ An initial state for the incoming loop.+  -> (a -> IncomingEvent s m a -> IO a)+  -- ^ A state-passing function for the incoming loop.+  -> b+  -- ^ An initial state for the outgoing loop.+  -> (b -> IO (b, OutgoingEvent s m 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 ->+    go conn stream isfc osfc encoding decoding+  where+    go conn stream isfc osfc encoding decoding =+        concurrently (incomingLoop Nothing newDecoder v0) (outGoingLoop w0)+      where+        ocfc = _outgoingFlowControl conn+        newDecoder = decodeOutput rpc decompress+        decompress = _getDecodingCompression decoding+        outGoingLoop v1 = do+             (v2, event) <- next v1+             case event of+                 Finalize -> do+                    sendData conn stream setEndStream ""+                    return v2+                 SendMessage doCompress msg -> do+                    let compress = case doCompress of+                            Compressed -> _getEncodingCompression encoding+                            Uncompressed -> uncompressed+                    sendSingleMessage rpc msg (Encoding compress) id conn ocfc stream osfc+                    outGoingLoop v2+        incomingLoop Nothing decode v1 =+                _waitEvent stream >>= \case+                    (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) ->+                    handle v1 (Trailers hdrs)+                (StreamDataEvent _ dat) -> do+                    _addCredit isfc (ByteString.length dat)+                    _ <- _consumeCredit isfc (ByteString.length dat)+                    _ <- _updateWindow isfc+                    case pushChunk decode dat of+                        (Done unusedDat _ (Right val)) ->+                            handle v1 (RecvMessage val) >>= incomingLoop jhdrs (pushChunk newDecoder unusedDat)+                        partial@(Partial _)    -> do+                            incomingLoop jhdrs partial v1+                        (Done _ _ (Left err)) -> do+                            handle v1 (Invalid $ toException $ InvalidParse $ "invalid-done-parse: " ++ err)+                        (Fail _ _ err)         ->+                            handle v1 (Invalid $ toException $ InvalidParse $ "invalid-parse: " ++ err)+                (StreamPushPromiseEvent _ _ _) ->+                    handle v1 (Invalid $ toException UnallowedPushPromiseReceived)+                (StreamErrorEvent _ _) ->+                    handle v1 (Invalid $ toException $ InvalidState "stream error")