diff --git a/Changelog.md b/Changelog.md new file mode 100644 --- /dev/null +++ b/Changelog.md @@ -0,0 +1,7 @@ +# Changelog for http2-client-grpc + +## Unreleased changes + +## 0.5.0.2 + +Add bidirectional and general streams. diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -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) diff --git a/http2-client-grpc.cabal b/http2-client-grpc.cabal --- a/http2-client-grpc.cabal +++ b/http2-client-grpc.cabal @@ -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 diff --git a/src/Network/GRPC/Client.hs b/src/Network/GRPC/Client.hs --- a/src/Network/GRPC/Client.hs +++ b/src/Network/GRPC/Client.hs @@ -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")