warp-grpc 0.1.0.3 → 0.2.0.0
raw patch · 5 files changed
+120/−98 lines, 5 filesdep +http2dep −proto-lensdep ~http2-grpc-types
Dependencies added: http2
Dependencies removed: proto-lens
Dependency ranges changed: http2-grpc-types
Files
- ChangeLog.md +4/−0
- README.md +2/−0
- src/Network/GRPC/Server/Handlers.hs +87/−83
- src/Network/GRPC/Server/Helpers.hs +12/−2
- warp-grpc.cabal +15/−13
ChangeLog.md view
@@ -2,6 +2,10 @@ ## Unreleased changes +## 0.2.0.0++Support for different Protocol Buffers serialization libraries.+ ## 0.1.0.3 Add bidirectional and general stream handlers.
README.md view
@@ -27,6 +27,8 @@ Please refer to https://github.com/lucasdicioccio/warp-grpc-example for an example. +TODO(ldicioccio): migrate this example to the new monorepo.+ ## Next steps * Helper to set metadatas (a.k.a., headers and trailers).
src/Network/GRPC/Server/Handlers.hs view
@@ -6,21 +6,19 @@ module Network.GRPC.Server.Handlers where import Control.Concurrent.Async (concurrently)-import Control.Monad (void)+import Control.Monad (void, (>=>)) import Data.Binary.Get (pushChunk, Decoder(..)) import qualified Data.ByteString.Char8 as ByteString import Data.ByteString.Char8 (ByteString) import Data.ByteString.Lazy (toStrict)-import Data.ProtoLens.Message (Message)-import Data.ProtoLens.Service.Types (Service(..), HasMethod, HasMethodImpl(..), StreamingType(..))-import Network.GRPC.HTTP2.Encoding (decodeInput, encodeOutput, Encoding(..), Decoding(..))-import Network.GRPC.HTTP2.Types (RPC(..), GRPCStatus(..), GRPCStatusCode(..), path)-import Network.Wai (Request, requestBody, strictRequestBody)+import Network.GRPC.HTTP2.Encoding+import Network.GRPC.HTTP2.Types (path, GRPCStatus(..), GRPCStatusCode(..))+import Network.Wai (Request, getRequestBodyChunk, strictRequestBody) import Network.GRPC.Server.Wai (WaiHandler, ServiceHandler(..), closeEarly) -- | Handy type to refer to Handler for 'unary' RPCs handler.-type UnaryHandler s m = Request -> MethodInput s m -> IO (MethodOutput s m)+type UnaryHandler i o = Request -> i -> IO o -- | Handy type for 'server-streaming' RPCs. --@@ -28,10 +26,10 @@ -- - read the input request -- - return an initial state and an state-passing action that the server code will call to fetch the output to send to the client (or close an a Nothing) -- See 'ServerStream' for the type which embodies these requirements.-type ServerStreamHandler s m a = Request -> MethodInput s m -> IO (a, ServerStream s m a)+type ServerStreamHandler i o a = Request -> i -> IO (a, ServerStream o a) -newtype ServerStream s m a = ServerStream {- serverStreamNext :: a -> IO (Maybe (a, MethodOutput s m))+newtype ServerStream o a = ServerStream {+ serverStreamNext :: a -> IO (Maybe (a, o)) } -- | Handy type for 'client-streaming' RPCs.@@ -41,11 +39,11 @@ -- - a state-passing handler for new client message -- - a state-aware handler for answering the client when it is ending its stream -- See 'ClientStream' for the type which embodies these requirements.-type ClientStreamHandler s m a = Request -> IO (a, ClientStream s m a)+type ClientStreamHandler i o a = Request -> IO (a, ClientStream i o a) -data ClientStream s m a = ClientStream {- clientStreamHandler :: a -> MethodInput s m -> IO a- , clientStreamFinalizer :: a -> IO (MethodOutput s m)+data ClientStream i o a = ClientStream {+ clientStreamHandler :: a -> i -> IO a+ , clientStreamFinalizer :: a -> IO o } -- | Handy type for 'bidirectional-streaming' RPCs.@@ -60,87 +58,89 @@ -- -- 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)+type BiDiStreamHandler i o a = Request -> IO (a, BiDiStream i o a) -data BiDiStep s m a+data BiDiStep i o a = Abort- | WaitInput !(a -> MethodInput s m -> IO a) !(a -> IO a)- | WriteOutput !a (MethodOutput s m)+ | WaitInput !(a -> i -> IO a) !(a -> IO a)+ | WriteOutput !a o -data BiDiStream s m a = BiDiStream {- bidirNextStep :: a -> IO (BiDiStep s m a)+newtype BiDiStream i o a = BiDiStream {+ bidirNextStep :: a -> IO (BiDiStep i o a) } -- | Construct a handler for handling a unary RPC. unary- :: (Service s, HasMethod s m)- => RPC s m- -> UnaryHandler s m+ :: (GRPCInput r i, GRPCOutput r o)+ => r+ -> UnaryHandler i o -> ServiceHandler unary rpc handler = ServiceHandler (path rpc) (handleUnary rpc handler) -- | Construct a handler for handling a server-streaming RPC. serverStream- :: (Service s, HasMethod s m, MethodStreamingType s m ~ 'ServerStreaming)- => RPC s m- -> ServerStreamHandler s m a+ :: (GRPCInput r i, GRPCOutput r o)+ => r+ -> ServerStreamHandler i o a -> ServiceHandler serverStream rpc handler = ServiceHandler (path rpc) (handleServerStream rpc handler) -- | Construct a handler for handling a client-streaming RPC. clientStream- :: (Service s, HasMethod s m, MethodStreamingType s m ~ 'ClientStreaming)- => RPC s m- -> ClientStreamHandler s m a+ :: (GRPCInput r i, GRPCOutput r o)+ => r+ -> ClientStreamHandler i o 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+ :: (GRPCInput r i, GRPCOutput r o)+ => r+ -> BiDiStreamHandler i o 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+ :: (GRPCInput r i, GRPCOutput r o)+ => r+ -> GeneralStreamHandler i o a b -> ServiceHandler generalStream rpc handler = ServiceHandler (path rpc) (handleGeneralStream rpc handler) -- | Handle unary RPCs.-handleUnary ::- (Service s, HasMethod s m)- => RPC s m- -> UnaryHandler s m+handleUnary+ :: (GRPCInput r i, GRPCOutput r o)+ => r+ -> UnaryHandler i o -> WaiHandler-handleUnary rpc handler decoding encoding req write flush = do- handleRequestChunksLoop (decodeInput rpc $ _getDecodingCompression decoding) handleMsg handleEof nextChunk+handleUnary rpc handler decoding encoding req write flush =+ handleRequestChunksLoop (decodeInput rpc $ _getDecodingCompression decoding)+ handleMsg handleEof nextChunk where nextChunk = toStrict <$> strictRequestBody req- handleMsg = errorOnLeftOver (\i -> handler req i >>= reply)+ handleMsg = errorOnLeftOver (handler req >=> reply) handleEof = closeEarly (GRPCStatus INVALID_ARGUMENT "early end of request body") reply msg = write (encodeOutput rpc (_getEncodingCompression encoding) msg) >> flush -- | Handle Server-Streaming RPCs.-handleServerStream ::- (Service s, HasMethod s m)- => RPC s m- -> ServerStreamHandler s m a+handleServerStream+ :: (GRPCInput r i, GRPCOutput r o)+ => r+ -> ServerStreamHandler i o a -> WaiHandler-handleServerStream rpc handler decoding encoding req write flush = do- handleRequestChunksLoop (decodeInput rpc $ _getDecodingCompression decoding) handleMsg handleEof nextChunk+handleServerStream rpc handler decoding encoding req write flush =+ handleRequestChunksLoop (decodeInput rpc $ _getDecodingCompression decoding)+ handleMsg handleEof nextChunk where nextChunk = toStrict <$> strictRequestBody req- handleMsg = errorOnLeftOver (\i -> handler req i >>= replyN)+ handleMsg = errorOnLeftOver (handler req >=> replyN) handleEof = closeEarly (GRPCStatus INVALID_ARGUMENT "early end of request body") replyN (v, sStream) = do let go v1 = serverStreamNext sStream v1 >>= \case@@ -151,39 +151,44 @@ go v -- | Handle Client-Streaming RPCs.-handleClientStream ::- (Service s, HasMethod s m)- => RPC s m- -> ClientStreamHandler s m a+handleClientStream+ :: (GRPCInput r i, GRPCOutput r o)+ => r+ -> ClientStreamHandler i o a -> WaiHandler-handleClientStream rpc handler0 decoding encoding req write flush = do+handleClientStream rpc handler0 decoding encoding req write flush = handler0 req >>= go where- go (v, cStream) = handleRequestChunksLoop (decodeInput rpc $ _getDecodingCompression decoding) (handleMsg v) (handleEof v) nextChunk+ go (v, cStream) = handleRequestChunksLoop (decodeInput rpc $ _getDecodingCompression decoding)+ (handleMsg v) (handleEof v) nextChunk where- nextChunk = requestBody req+ nextChunk = getRequestBodyChunk req handleMsg v0 dat msg = clientStreamHandler cStream v0 msg >>= \v1 -> loop dat v1 handleEof v0 = clientStreamFinalizer cStream v0 >>= reply 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+ 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+handleBiDiStream+ :: (GRPCInput r i, GRPCOutput r o)+ => r+ -> BiDiStreamHandler i o a -> WaiHandler-handleBiDiStream rpc handler0 decoding encoding req write flush = do+handleBiDiStream rpc handler0 decoding encoding req write flush = handler0 req >>= go "" where- nextChunk = requestBody req+ nextChunk = getRequestBodyChunk 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+ step <- bidirNextStep bStream v0 case step of- WaitInput handleMsg handleEof -> do- handleRequestChunksLoop (flip pushChunk chunk $ decodeInput rpc $ _getDecodingCompression decoding)+ WaitInput handleMsg handleEof ->+ handleRequestChunksLoop (flip pushChunk chunk+ $ decodeInput rpc+ $ _getDecodingCompression decoding) (\dat msg -> handleMsg v0 msg >>= cont dat) (handleEof v0 >>= cont "") nextChunk@@ -193,40 +198,40 @@ 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)+type GeneralStreamHandler i o a b =+ Request -> IO (a, IncomingStream i a, b, OutgoingStream o b) -- | Pair of handlers for reacting to incoming messages.-data IncomingStream s m a = IncomingStream {- incomingStreamHandler :: a -> MethodInput s m -> IO a+data IncomingStream i a = IncomingStream {+ incomingStreamHandler :: a -> i -> 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))+newtype OutgoingStream o a = OutgoingStream {+ outgoingStreamNext :: a -> IO (Maybe (a, o)) } -- | 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+handleGeneralStream+ :: (GRPCInput r i, GRPCOutput r o)+ => r+ -> GeneralStreamHandler i o a b -> WaiHandler-handleGeneralStream rpc handler0 decoding encoding req write flush = void $ do+handleGeneralStream rpc handler0 decoding encoding req write flush = void $ handler0 req >>= go where newDecoder = decodeInput rpc $ _getDecodingCompression decoding- nextChunk = requestBody req+ nextChunk = getRequestBodyChunk 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+ replyLoop v0 sstream@(OutgoingStream next) = next v0 >>= \case Nothing -> return v0 (Just (v1, msg)) -> reply msg >> replyLoop v1 sstream@@ -241,8 +246,7 @@ -- | Helpers to consume input in chunks. handleRequestChunksLoop- :: (Message a)- => Decoder (Either String a)+ :: Decoder (Either String a) -- ^ Message decoder. -> (ByteString -> a -> IO b) -- ^ Handler for a single message.@@ -255,9 +259,9 @@ {-# INLINEABLE handleRequestChunksLoop #-} handleRequestChunksLoop decoder handleMsg handleEof nextChunk = case decoder of- (Done unusedDat _ (Right val)) -> do+ (Done unusedDat _ (Right val)) -> handleMsg unusedDat val- (Done _ _ (Left err)) -> do+ (Done _ _ (Left err)) -> closeEarly (GRPCStatus INVALID_ARGUMENT (ByteString.pack $ "done-error: " ++ err)) (Fail _ _ err) -> closeEarly (GRPCStatus INVALID_ARGUMENT (ByteString.pack $ "fail-error: " ++ err))
src/Network/GRPC/Server/Helpers.hs view
@@ -1,10 +1,12 @@--+{-# LANGUAGE CPP #-} module Network.GRPC.Server.Helpers where import qualified Data.ByteString.Char8 as ByteString import Data.Maybe (fromMaybe) import Network.GRPC.HTTP2.Types (GRPCStatus(..), trailerForStatusCode, grpcStatusH, grpcMessageH)+#if MIN_VERSION_warp(3,3,0)+import Network.HTTP2.Server (NextTrailersMaker(..))+#endif import Network.Wai (Request) import Network.Wai.Handler.Warp (http2dataTrailers, defaultHTTP2Data, modifyHTTP2Data, HTTP2Data) @@ -14,8 +16,16 @@ makeGRPCTrailers :: GRPCStatus -> (Maybe HTTP2Data -> Maybe HTTP2Data) makeGRPCTrailers (GRPCStatus s msg) h2data =+#if MIN_VERSION_warp(3,3,0)+ Just $! (fromMaybe defaultHTTP2Data h2data) { http2dataTrailers = trailersMaker }+#else Just $! (fromMaybe defaultHTTP2Data h2data) { http2dataTrailers = trailers }+#endif where+#if MIN_VERSION_warp(3,3,0)+ trailersMaker (Just _) = return $ NextTrailersMaker trailersMaker+ trailersMaker Nothing = return $ Trailers trailers+#endif trailers = if ByteString.null msg then [status] else [status, message] status = (grpcStatusH, trailerForStatusCode s) message = (grpcMessageH, msg)
warp-grpc.cabal view
@@ -1,30 +1,31 @@--- This file has been generated from package.yaml by hpack version 0.28.2.+cabal-version: 1.12++-- This file has been generated from package.yaml by hpack version 0.31.2. -- -- see: https://github.com/sol/hpack ----- hash: 1c4d10b1b0541ea7c74d968b365ce9a913989a6cd9fd9c7b0296510dea917627+-- hash: 5cc20b886f598a4b752f762493d70b24c9979d6893ffd1e239e260d1b36bd47b name: warp-grpc-version: 0.1.0.3+version: 0.2.0.0 synopsis: A minimal gRPC server on top of Warp.-description: Please see the README on Github at <https://github.com/lucasdicioccio/warp-grpc#readme>+description: Please see the README on Github at <https://github.com/haskell-grpc-native/http2-grpc-haskell/blob/master/warp-grpc/README.md> category: Networking-homepage: https://github.com/lucasdicioccio/warp-grpc#readme-bug-reports: https://github.com/lucasdicioccio/warp-grpc/issues-author: Lucas DiCioccio+homepage: https://github.com/haskell-grpc-native/http2-grpc-haskell#readme+bug-reports: https://github.com/haskell-grpc-native/http2-grpc-haskell/issues+author: Lucas DiCioccio, Alejandro Serrano maintainer: lucas@dicioccio.fr-copyright: 2017 Lucas DiCioccio+copyright: 2017 - 2019 Lucas DiCioccio, Alejandro Serrano license: BSD3 license-file: LICENSE build-type: Simple-cabal-version: >= 1.10 extra-source-files:- ChangeLog.md README.md+ ChangeLog.md source-repository head type: git- location: https://github.com/lucasdicioccio/warp-grpc+ location: https://github.com/haskell-grpc-native/http2-grpc-haskell library exposed-modules:@@ -36,6 +37,7 @@ Paths_warp_grpc hs-source-dirs: src+ ghc-options: -Wall build-depends: async >=2.2.1 && <3 , base >=4.11 && <5@@ -43,8 +45,8 @@ , 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+ , http2 >=1.6 && <3+ , http2-grpc-types >=0.5 && <0.6 , wai >=3.2 && <3.3 , warp >=3.2.24 && <3.3 , warp-tls >=3.2 && <3.3