warp-grpc 0.1.0.2 → 0.1.0.3

raw patch · 5 files changed

+159/−19 lines, 5 filesdep +asyncdep ~basedep ~binarydep ~bytestring

Dependencies added: async

Dependency ranges changed: base, binary, bytestring, case-insensitive, http-types, http2-grpc-types, proto-lens, wai, warp-tls

Files

• ChangeLog.md +4/−0
• README.md +13/−1
• src/Network/GRPC/Server.hs +9/−1
• src/Network/GRPC/Server/Handlers.hs +120/−5
• warp-grpc.cabal +13/−12

@@ -1,3 +1,7 @@ # Changelog for warp-grpc  ## Unreleased changes++## 0.1.0.3++Add bidirectional and general stream handlers.

@@ -23,10 +23,22 @@ handlers for the `service` stanzas defined in the `.proto` files (see Haddocks). Finally, serve `warp` over TLS`. +## Example++Please refer to https://github.com/lucasdicioccio/warp-grpc-example for an example.+ ## Next steps -* Handler type for bidirectional streams.+* Helper to set metadatas (a.k.a., headers and trailers).+  - (API-breaking) some or all handlers will get an IO-step to return extra metadata+* Helper to map request headerlists into client metadatas (probably in `http2-grpc-types`)  ## Limitations  * Only supports "h2" with TLS (I'd argue it's a feature, not a bug. Don't @-me)+* Some valid gRPC applications may not be expressible directly on top of warp+  because sending HTTP2 trailers (i.e., signalling the server's desire to stop+  sending messages) is correlated with closing the HTTP2 stream (i.e., stop+  accepting client messages). Hence it's not feasible to create a bidirectional+  stream that terminates on the server end while continuing to ingest client+  messages. This use case, however, seems like a corner case.

@@ -11,11 +11,19 @@     , ServerStream(..)     , ClientStreamHandler     , ClientStream(..)+    , BiDiStreamHandler+    , BiDiStream(..)+    , BiDiStep(..)+    , GeneralStreamHandler+    , IncomingStream(..)+    , OutgoingStream(..)     -- * registration     , ServiceHandler     , unary     , serverStream     , clientStream+    , bidiStream+    , generalStream     -- * registration     , GRPCStatus (..)     , throwIO@@ -32,7 +40,7 @@ import           Network.Wai.Handler.WarpTLS (TLSSettings, runTLS) import           Network.Wai.Handler.Warp (Settings) -import           Network.GRPC.Server.Handlers (UnaryHandler, unary, ServerStreamHandler, ServerStream(..), serverStream, ClientStreamHandler, ClientStream(..), clientStream)+import           Network.GRPC.Server.Handlers (UnaryHandler, unary, ServerStreamHandler, ServerStream(..), serverStream, ClientStreamHandler, ClientStream(..), clientStream, BiDiStreamHandler, BiDiStream(..), BiDiStep(..), bidiStream, GeneralStreamHandler, IncomingStream(..), OutgoingStream(..), generalStream) import           Network.GRPC.Server.Wai (ServiceHandler(..), grpcApp, grpcService)  -- | Helper to constructs and serve a gRPC over HTTP2 application.

@@ -5,6 +5,8 @@ {-# LANGUAGE TypeFamilies #-} module Network.GRPC.Server.Handlers where +import           Control.Concurrent.Async (concurrently)+import           Control.Monad (void) import           Data.Binary.Get (pushChunk, Decoder(..)) import qualified Data.ByteString.Char8 as ByteString import           Data.ByteString.Char8 (ByteString)@@ -46,6 +48,29 @@   , clientStreamFinalizer :: a -> IO (MethodOutput s m)   } +-- | Handy type for 'bidirectional-streaming' RPCs.+--+-- We expect an implementation to:+-- - acknowlege a new bidirection stream by returning an initial state and one functions:+-- - a state-passing function that returns a single action step+-- The action may be to+-- - stop immediately+-- - wait and handle some input with a callback and a finalizer (if the client closes the stream on its side) that may change the state+-- - return a value and a new state+--+-- There is no way to stop locally (that would mean sending HTTP2 trailers) and+-- keep receiving messages from the client.+type BiDiStreamHandler s m a = Request -> IO (a, BiDiStream s m a)++data BiDiStep s m a+  = Abort+  | WaitInput !(a -> MethodInput s m -> IO a) !(a -> IO a)+  | WriteOutput !a (MethodOutput s m)++data BiDiStream s m a = BiDiStream {+    bidirNextStep :: a -> IO (BiDiStep s m a)+  }+ -- | Construct a handler for handling a unary RPC. unary   :: (Service s, HasMethod s m)@@ -57,7 +82,7 @@  -- | Construct a handler for handling a server-streaming RPC. serverStream-  :: (Service s, HasMethod s m,  MethodStreamingType s m ~ ServerStreaming)+  :: (Service s, HasMethod s m,  MethodStreamingType s m ~ 'ServerStreaming)   => RPC s m   -> ServerStreamHandler s m a   -> ServiceHandler@@ -66,13 +91,31 @@  -- | Construct a handler for handling a client-streaming RPC. clientStream-  :: (Service s, HasMethod s m,  MethodStreamingType s m ~ ClientStreaming)+  :: (Service s, HasMethod s m,  MethodStreamingType s m ~ 'ClientStreaming)   => RPC s m   -> ClientStreamHandler s m a   -> ServiceHandler clientStream rpc handler =     ServiceHandler (path rpc) (handleClientStream rpc handler) +-- | Construct a handler for handling a bidirectional-streaming RPC.+bidiStream+  :: (Service s, HasMethod s m,  MethodStreamingType s m ~ 'BiDiStreaming)+  => RPC s m+  -> BiDiStreamHandler s m a+  -> ServiceHandler+bidiStream rpc handler =+    ServiceHandler (path rpc) (handleBiDiStream rpc handler)++-- | Construct a handler for handling a bidirectional-streaming RPC.+generalStream+  :: (Service s, HasMethod s m)+  => RPC s m+  -> GeneralStreamHandler s m a b+  -> ServiceHandler+generalStream rpc handler =+    ServiceHandler (path rpc) (handleGeneralStream rpc handler)+ -- | Handle unary RPCs. handleUnary ::      (Service s, HasMethod s m)@@ -124,19 +167,91 @@         reply msg = write (encodeOutput rpc (_getEncodingCompression encoding) msg) >> flush         loop chunk v1 = handleRequestChunksLoop (flip pushChunk chunk $ decodeInput rpc (_getDecodingCompression decoding)) (handleMsg v1) (handleEof v1) nextChunk +-- | Handle Bidirectional-Streaming RPCs.+handleBiDiStream ::+    (Service s, HasMethod s m)+  => RPC s m+  -> BiDiStreamHandler s m a+  -> WaiHandler+handleBiDiStream rpc handler0 decoding encoding req write flush = do+    handler0 req >>= go ""+  where+    nextChunk = requestBody req+    reply msg = write (encodeOutput rpc (_getEncodingCompression encoding) msg) >> flush+    go chunk (v0, bStream) = do+        let cont dat v1 = go dat (v1, bStream)+        step <- (bidirNextStep bStream) v0+        case step of+            WaitInput handleMsg handleEof -> do+                handleRequestChunksLoop (flip pushChunk chunk $ decodeInput rpc $ _getDecodingCompression decoding)+                                        (\dat msg -> handleMsg v0 msg >>= cont dat)+                                        (handleEof v0 >>= cont "")+                                        nextChunk+            WriteOutput v1 msg -> do+                reply msg+                cont "" v1+            Abort -> return ()++-- | A GeneralStreamHandler combining server and client asynchronous streams.+type GeneralStreamHandler s m a b =+    Request -> IO (a, IncomingStream s m a, b, OutgoingStream s m b)++-- | Pair of handlers for reacting to incoming messages.+data IncomingStream s m a = IncomingStream {+    incomingStreamHandler   :: a -> MethodInput s m -> IO a+  , incomingStreamFinalizer :: a -> IO ()+  }++-- | Handler to decide on the next message (if any) to return.+data OutgoingStream s m a = OutgoingStream {+    outgoingStreamNext  :: a -> IO (Maybe (a, MethodOutput s m))+  }++-- | Handler for the somewhat general case where two threads behave concurrently:+-- - one reads messages from the client+-- - one returns messages to the client+handleGeneralStream ::+    (Service s, HasMethod s m)+  => RPC s m+  -> GeneralStreamHandler s m a b+  -> WaiHandler+handleGeneralStream rpc handler0 decoding encoding req write flush = void $ do+    handler0 req >>= go+  where+    newDecoder = decodeInput rpc $ _getDecodingCompression decoding+    nextChunk = requestBody req+    reply msg = write (encodeOutput rpc (_getEncodingCompression encoding) msg) >> flush++    go (in0, instream, out0, outstream) = concurrently+        (incomingLoop newDecoder in0 instream)+        (replyLoop out0 outstream)++    replyLoop v0 sstream@(OutgoingStream next) = do+        next v0 >>= \case+            Nothing          -> return v0+            (Just (v1, msg)) -> reply msg >> replyLoop v1 sstream++    incomingLoop decode v0 cstream = do+        let handleMsg dat msg = do+                v1 <- incomingStreamHandler cstream v0 msg+                incomingLoop (pushChunk newDecoder dat) v1 cstream+        let handleEof = incomingStreamFinalizer cstream v0 >> pure v0+        handleRequestChunksLoop decode handleMsg handleEof nextChunk++ -- | Helpers to consume input in chunks. handleRequestChunksLoop   :: (Message a)   => Decoder (Either String a)   -- ^ Message decoder.-  -> (ByteString -> a -> IO ())+  -> (ByteString -> a -> IO b)   -- ^ Handler for a single message.   -- The ByteString corresponds to leftover data.-  -> IO ()+  -> IO b   -- ^ Handler for handling end-of-streams.   -> IO ByteString   -- ^ Action to retrieve the next chunk.-  -> IO ()+  -> IO b {-# INLINEABLE handleRequestChunksLoop #-} handleRequestChunksLoop decoder handleMsg handleEof nextChunk =     case decoder of

@@ -2,12 +2,12 @@ -- -- see: https://github.com/sol/hpack ----- hash: cbdbea712bc540cca84b17d953a1acbd8a15b3d7dbc97332893fd84b75b9a574+-- hash: 1c4d10b1b0541ea7c74d968b365ce9a913989a6cd9fd9c7b0296510dea917627  name:           warp-grpc-version:        0.1.0.2+version:        0.1.0.3 synopsis:       A minimal gRPC server on top of Warp.-description:    Please see the README on Github at <https://github.com/githubuser/warp-grpc#readme>+description:    Please see the README on Github at <https://github.com/lucasdicioccio/warp-grpc#readme> category:       Networking homepage:       https://github.com/lucasdicioccio/warp-grpc#readme bug-reports:    https://github.com/lucasdicioccio/warp-grpc/issues@@ -37,14 +37,15 @@   hs-source-dirs:       src   build-depends:-      base >=4.7 && <5-    , binary-    , bytestring-    , case-insensitive-    , http-types-    , http2-grpc-types >=0.2-    , proto-lens-    , wai+      async >=2.2.1 && <3+    , base >=4.11 && <5+    , binary >=0.8.5 && <0.9+    , bytestring >=0.10.8 && <0.11+    , case-insensitive >=1.2.0 && <1.3+    , http-types >=0.12 && <0.13+    , http2-grpc-types >=0.3 && <0.4+    , proto-lens >=0.3 && <0.4+    , wai >=3.2 && <3.3     , warp >=3.2.24 && <3.3-    , warp-tls+    , warp-tls >=3.2 && <3.3   default-language: Haskell2010