http2-client-grpc (empty) → 0.1.0.0

raw patch · 6 files changed

+459/−0 lines, 6 filesdep +basedep +binarydep +bytestringsetup-changed

Dependencies added: base, binary, bytestring, data-default-class, http2, http2-client, http2-client-grpc, proto-lens, proto-lens-protoc, text, zlib

Files

• LICENSE +30/−0
• README.md +76/−0
• Setup.hs +3/−0
• http2-client-grpc.cabal +42/−0
• src/Network/GRPC.hs +306/−0
• test/Spec.hs +2/−0

@@ -0,0 +1,30 @@+Copyright Author name here (c) 2017++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++    * Redistributions of source code must retain the above copyright+      notice, this list of conditions and the following disclaimer.++    * Redistributions in binary form must reproduce the above+      copyright notice, this list of conditions and the following+      disclaimer in the documentation and/or other materials provided+      with the distribution.++    * Neither the name of Author name here nor the names of other+      contributors may be used to endorse or promote products derived+      from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

@@ -0,0 +1,76 @@+# http2-client-grpc++A native HTTP2 gRPC client library using `proto-lens` and `http2-client`.++## Summary++This project provides a library that leverages the code generated using+proto-lens (in particular, proto-lens-protoc) to implement the client-side of+gRPC services.++## Usage++### Prerequisites++In addition to a working Haskell dev environment, you need to:+- build the `proto-lens-protoc` executable (`proto-lens`)+- install the `protoc` executable++### Adding .proto files to a Haskell package++In order to run gRPC:++- generate the `Proto` stubs in some `gen` directory++A single `protoc` invocation may be enough for both Proto and GRPC outputs:++```bash+protoc  "--plugin=protoc-gen-haskell-protolens=${protolens}" \+    --haskell-protolens_out=./gen \+    -I "${protodir1} \+    -I "${protodir2} \+    ${first.proto} \+    ${second.proto}+```++- add the `gen` sourcedir for the generated to your .cabal/package.yaml file (cf. 'hs-source-dirs').+- add the generated Proto modules to the 'exposed-modules' (or 'other-modules') keys++A reliable way to list the module names is the following bash invocation:++```bash+find gen -name "*.hs" | sed -e 's/gen\///' | sed -e 's/\.hs$//' | tr '/' '.'+```++Unlike `proto-lens`, this project does not yet provide a modified `Setup.hs`.+As a result, we cannot automate these steps from within Cabal/Stack. Hence,+you'll have to automate these steps outside your Haskell toolchain.++## Calling a GRPC service++In short, use `http2-client` with the `Network.GRPC.call` function and let the types guide you.++### Example++You'll find an example leveraging the awesome `grpcb.in` service at https://github.com/lucasdicioccio/http2-client-grpc-example .++### gRPC service mapping++The Protobuf format specifies the notion of Service which can have one or more+RPCs.  The gRPC protocal maps these services onto HTTP2 headers and HTTP2+frames.  In general, gRPC implementation generate one inlined function per RPC.+This implementation differs by decoupling the HTTP2 transport and leveraging+generics to provide generic functions. This design allows the+`Network.GRPC.call` to be type safe and multi-usage. For instance, you can wrap+this function with whatever metrics/benchmark code and have all your RPC calls+be monitored in a uniform way.++## Status++This library is currently an early-stage library. Expect breaking changes.++## TODOs++- pass timeout as argument+- support compression+- provide function to map raw results into commonly-understood GRPC errors

@@ -0,0 +1,3 @@+import Data.ProtoLens.Setup++main = defaultMain "."

@@ -0,0 +1,42 @@+name:                http2-client-grpc+version:             0.1.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+license:             BSD3+license-file:        LICENSE+author:              Lucas DiCioccio+maintainer:          lucas@dicioccio.fr+copyright:           2017 Lucas DiCioccio+category:            Network+build-type:          Simple+extra-source-files:  README.md+cabal-version:       >=1.10++library+  hs-source-dirs:      src+  exposed-modules:     Network.GRPC+  build-depends:       base >= 4.7 && < 5+                     , binary+                     , bytestring+                     , data-default-class+                     , http2+                     , http2-client+                     , proto-lens+                     , proto-lens-protoc+                     , text+                     , zlib+  default-language:    Haskell2010++test-suite http2-client-grpc-test+  type:                exitcode-stdio-1.0+  hs-source-dirs:      test+  main-is:             Spec.hs+  build-depends:       base+                     , http2-client-grpc+  ghc-options:         -threaded -rtsopts -with-rtsopts=-N+  default-language:    Haskell2010++source-repository head+  type:     git+  location: https://github.com/lucasdicioccio/http2-client-grpc

@@ -0,0 +1,306 @@+{-# LANGUAGE BangPatterns        #-}+{-# LANGUAGE DataKinds           #-}+{-# LANGUAGE FlexibleContexts    #-}+{-# LANGUAGE LambdaCase          #-}+{-# LANGUAGE OverloadedStrings   #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeFamilies        #-}++-- | A module adding support for gRPC over HTTP2.+module Network.GRPC (+  -- * Building blocks.+    RPC(..)+  , Authority+  , Timeout(..)+  , open+  , RawReply+  -- * Helpers+  , singleRequest+  , streamReply+  , streamRequest+  , StreamDone(..)+  -- * Errors.+  , InvalidState(..)+  , StreamReplyDecodingError(..)+  , UnallowedPushPromiseReceived(..)+  -- * Compression of individual messages.+  , Compression+  , gzip+  , uncompressed+  ) where++import qualified Codec.Compression.GZip as GZip+import Control.Exception (Exception(..), throwIO)+import Control.Monad (forever)+import Data.Proxy (Proxy(..))+import Data.Monoid ((<>))+import Data.ByteString.Char8 (ByteString, pack, unpack)+import Data.ByteString.Lazy (fromStrict, toStrict)+import Data.Binary.Builder (toLazyByteString, fromByteString, singleton, putWord32be)+import Data.Binary.Get (Decoder(..), getByteString, getInt8, getWord32be, pushChunk, pushEndOfInput, runGetIncremental)+import qualified Data.ByteString.Char8 as ByteString+import Data.ProtoLens.Encoding (encodeMessage, decodeMessage)+import Data.ProtoLens.Message (Message)+import Data.ProtoLens.Service.Types (Service(..), HasMethod, HasMethodImpl(..), StreamingType(..))+import GHC.TypeLits (Symbol, symbolVal)++import Network.HTTP2+import Network.HPACK+import Network.HTTP2.Client+import Network.HTTP2.Client.Helpers++-- | A proxy type for giving static information about RPCs.+data RPC (s :: *) (m :: Symbol) = RPC++-- | Returns the HTTP2 :path for a given RPC.+path :: (Service s, HasMethod s m) => RPC s m -> ByteString+path rpc = "/" <> pkg rpc Proxy <> "." <> srv rpc Proxy <> "/" <> meth rpc Proxy+  where+    pkg :: (Service s) => RPC s m -> Proxy (ServicePackage s) -> ByteString+    pkg _ p = pack $ symbolVal p++    srv :: (Service s) => RPC s m -> Proxy (ServiceName s) -> ByteString+    srv _ p = pack $ symbolVal p++    meth :: (Service s, HasMethod s m) => RPC s m -> Proxy (MethodName s m) -> ByteString+    meth _ p = pack $ symbolVal p++-- | Decodes a method's output.+decodeResult :: (Service s, HasMethod s m) => RPC s m -> Decoder (Either String (MethodOutput s m))+decodeResult _ = runGetIncremental $ do+    isCompressed <- getInt8      -- 1byte+    let compression = if isCompressed == 0 then id else (toStrict . GZip.decompress . fromStrict)+    n <- getWord32be  -- 4bytes+    decodeMessage . compression <$> getByteString (fromIntegral n)++-- | Tries finalizing a Decoder.+fromDecoder :: Decoder (Either String a) -> Either String a+fromDecoder (Fail _ _ msg) = Left msg+fromDecoder (Partial _)    = Left "got only a subet of the message"+fromDecoder (Done _ _ val) = val++-- | A reply.+--+-- This reply object contains a lot of information because a single gRPC call+-- returns a lot of data. A future version of the library will have a proper+-- data structure with properly named-fields on the reply object.+--+-- For now, remember:+-- - 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))++-- | gRPC disables HTTP2 push-promises.+--+-- If a server attempts to send push-promises, this exception will be raised.+data UnallowedPushPromiseReceived = UnallowedPushPromiseReceived deriving Show+instance Exception UnallowedPushPromiseReceived where++-- | http2-client handler for push promise.+throwOnPushPromise :: PushPromiseHandler+throwOnPushPromise _ _ _ _ _ = throwIO UnallowedPushPromiseReceived++-- | Wait for an RPC reply.+waitReply :: (Service s, HasMethod s m) => RPC s m -> Http2Stream -> IncomingFlowControl -> IO (RawReply (MethodOutput s m))+waitReply rpc stream flowControl = do+    format . fromStreamResult <$> waitStream stream flowControl throwOnPushPromise+  where+    format rsp = do+       (hdrs, dat, trls) <- rsp+       let res =+             case lookup "grpc-message" hdrs of+               Nothing     -> fromDecoder $ pushEndOfInput $ flip pushChunk dat $ decodeResult rpc+               Just errMsg -> Left $ unpack errMsg++       return (hdrs, trls, res)++-- | Exception raised when a ServerStreaming RPC results in a decoding+-- error.+data StreamReplyDecodingError = StreamReplyDecodingError String deriving Show+instance Exception StreamReplyDecodingError where++-- | Exception raised when a ServerStreaming RPC results in an invalid+-- state machine.+data InvalidState = InvalidState String deriving Show+instance Exception InvalidState where++-- | The HTTP2-Authority portion of an URL (e.g., "dicioccio.fr:7777").+type Authority = ByteString.ByteString++-- | Newtype helper used to uniformize all type of streaming modes when+-- passing arguments to the 'open' call.+newtype RPCCall a = RPCCall {+    runRPC :: Http2Client -> Http2Stream -> IncomingFlowControl -> OutgoingFlowControl -> IO a+  }++-- | Timeout in seconds.+newtype Timeout = Timeout Int++showTimeout :: Timeout -> ByteString.ByteString+showTimeout (Timeout n) = ByteString.pack $ show n ++ "S"++-- | Main handler to perform gRPC calls to a service.+open :: (Service s, HasMethod s m)+     => RPC s m+     -- ^ A token carrying information specifying the RPC to call.+     -> Http2Client+     -- ^ A connected HTTP2 client.+     -> Authority+     -- ^ The HTTP2-Authority portion of the URL (e.g., "dicioccio.fr:7777").+     -> HeaderList+     -- ^ A set of HTTP2 headers (e.g., for adding authentication headers).+     -> Timeout+     -- ^ Timeout in seconds.+     -> Compression+     -- ^ An indication of the compression that you will be using.  Compression+     -- should be per message, however a bug in gRPC-Go (to be confirmed) seems+     -- to turn message compression mandatory if advertised in the HTTP2+     -- headers, even though the specification states that compression per+     -- message is optional irrespectively of headers.+     -> RPCCall a+     -- ^ The actual RPC handler.+     -> IO (Either TooMuchConcurrency a)+open rpc conn authority extraheaders timeout compression doStuff = do+    let icfc = _incomingFlowControl conn+    let ocfc = _outgoingFlowControl conn+    let request = [ (":method", "POST")+                  , (":scheme", "http")+                  , (":authority", authority)+                  , (":path", path rpc) +                  , ("grpc-timeout", showTimeout timeout)+                  , ("grpc-encoding", _compressionName compression)+                  , ("grpc-accept-encoding", "identity,gzip")+                  , ("content-type", "application/grpc+proto")+                  , ("te", "trailers")+                  ] <> extraheaders+    withHttp2Stream conn $ \stream ->+        let+            initStream = headers stream request (setEndHeader)+            handler isfc osfc = do+                (runRPC doStuff) conn stream isfc osfc+        in StreamDefinition initStream handler++-- | gRPC call for Server Streaming.+streamReply+  :: (Service s, HasMethod s m, MethodStreamingType s m ~ ServerStreaming)+  => RPC s m+  -> Compression+  -> MethodInput s m+  -> (HeaderList -> Either String (MethodOutput s m) -> IO ())+  -> RPCCall (HeaderList, HeaderList)+streamReply rpc compress req handler = RPCCall $ \conn stream isfc osfc -> do+    let {+        loop decoder hdrs = _waitEvent stream >>= \case+            (StreamPushPromiseEvent _ _ _) ->+                throwIO (InvalidState "push promise")+            (StreamHeadersEvent _ trls) ->+                return (hdrs, trls)+            (StreamErrorEvent _ _) ->+                throwIO (InvalidState "stream error")+            (StreamDataEvent _ dat) -> do+                _addCredit isfc (ByteString.length dat)+                _ <- _consumeCredit isfc (ByteString.length dat)+                _ <- _updateWindow isfc+                handleAllChunks hdrs decoder dat loop+    } in do+        let ocfc = _outgoingFlowControl conn+        sendSingleMessage req compress setEndStream conn ocfc stream osfc+        _waitEvent stream >>= \case+            StreamHeadersEvent _ hdrs ->+                loop (decodeResult rpc) hdrs+            _                         ->+                throwIO (InvalidState "no headers")+  where+    handleAllChunks hdrs decoder dat exitLoop =+       case pushChunk decoder dat of+           (Done unusedDat _ val) -> do+               handler hdrs val+               handleAllChunks hdrs (decodeResult rpc) unusedDat exitLoop+           failure@(Fail _ _ err)   -> do+               handler hdrs (fromDecoder failure)+               throwIO (StreamReplyDecodingError err)+           partial@(Partial _)    ->+               exitLoop partial hdrs++data StreamDone = StreamDone++-- | gRPC call for Client Streaming.+streamRequest+  :: (Service s, HasMethod s m, MethodStreamingType s m ~ ClientStreaming)+  => RPC s m+  -> (IO (Either StreamDone (MethodInput s m, Compression)))+  -> RPCCall (RawReply (MethodOutput s m))+streamRequest rpc handler = RPCCall $ \conn stream isfc streamFlowControl ->+    let ocfc = _outgoingFlowControl conn+        go = do+            nextEvent <- handler+            case nextEvent of+                Right (msg, compress) -> do+                    sendSingleMessage msg compress id conn ocfc stream streamFlowControl+                    go+                Left _ -> do+                    sendData conn stream setEndStream ""+                    waitReply rpc stream isfc+    in go+++-- | Serialize and send a single message.+sendSingleMessage :: Message a+                  => a+                  -> Compression+                  -> FlagSetter+                  -> Http2Client+                  -> OutgoingFlowControl+                  -> Http2Stream+                  -> OutgoingFlowControl+                  -> IO ()+sendSingleMessage msg compression flagMod conn connectionFlowControl stream streamFlowControl = do+    let goUpload dat = do+            let !wanted = ByteString.length dat+            gotStream <- _withdrawCredit streamFlowControl wanted+            got       <- _withdrawCredit connectionFlowControl gotStream+            _receiveCredit streamFlowControl (gotStream - got)+            if got == wanted+            then+                sendData conn stream flagMod dat+            else do+                sendData conn stream id (ByteString.take got dat)+                goUpload (ByteString.drop got dat)+    goUpload . toStrict . toLazyByteString . encodePlainMessage $ msg+  where+    compressedByte = if _compressionByteSet compression then 1 else 0+    compressMethod = _compressionFunction compression+    encodePlainMessage plain =+        let bin = encodeMessage plain+        in singleton compressedByte <> putWord32be (fromIntegral $ ByteString.length bin) <> fromByteString (compressMethod $ bin)++-- | gRPC call for an unary request.+singleRequest+  :: (Service s, HasMethod s m)+  => RPC s m+  -> Compression+  -> MethodInput s m+  -> RPCCall (RawReply (MethodOutput s m))+singleRequest rpc compress msg = RPCCall $ \conn stream isfc osfc -> do+    let ocfc = _outgoingFlowControl conn+    sendSingleMessage msg compress setEndStream conn ocfc stream osfc+    waitReply rpc stream isfc++-- | Opaque type for handling compression.+--+-- So far, only "pure" compression algorithms are supported.+-- TODO: suport IO-based compression implementations.+data Compression = Compression {+    _compressionName    :: ByteString.ByteString+  , _compressionByteSet :: Bool+  , _compressionFunction :: (ByteString.ByteString -> ByteString.ByteString)+  }++-- | Use gzip as compression.+gzip :: Compression+gzip = Compression "gzip" True (toStrict . GZip.compress . fromStrict)++-- | Do not compress.+uncompressed :: Compression+uncompressed = Compression "identity" False id

@@ -0,0 +1,2 @@+main :: IO ()+main = putStrLn "Test suite not yet implemented"