diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,12 @@ +# Changelog + +All notable changes to this project will be documented in this file. + +The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), +and this project adheres to the [Haskell Package Versioning Policy](https://pvp.haskell.org/). + +## [1.0.0] - 2026-07-09 + +### Added + +- Initial release. diff --git a/LICENCE b/LICENCE new file mode 100644 --- /dev/null +++ b/LICENCE @@ -0,0 +1,287 @@ + EUROPEAN UNION PUBLIC LICENCE v. 1.2 + EUPL © the European Union 2007, 2016 + +This European Union Public Licence (the ‘EUPL’) applies to the Work (as defined +below) which is provided under the terms of this Licence. Any use of the Work, +other than as authorised under this Licence is prohibited (to the extent such +use is covered by a right of the copyright holder of the Work). + +The Work is provided under the terms of this Licence when the Licensor (as +defined below) has placed the following notice immediately following the +copyright notice for the Work: + + Licensed under the EUPL + +or has expressed by any other means his willingness to license under the EUPL. + +1. Definitions + +In this Licence, the following terms have the following meaning: + +- ‘The Licence’: this Licence. + +- ‘The Original Work’: the work or software distributed or communicated by the + Licensor under this Licence, available as Source Code and also as Executable + Code as the case may be. + +- ‘Derivative Works’: the works or software that could be created by the + Licensee, based upon the Original Work or modifications thereof. This Licence + does not define the extent of modification or dependence on the Original Work + required in order to classify a work as a Derivative Work; this extent is + determined by copyright law applicable in the country mentioned in Article 15. + +- ‘The Work’: the Original Work or its Derivative Works. + +- ‘The Source Code’: the human-readable form of the Work which is the most + convenient for people to study and modify. + +- ‘The Executable Code’: any code which has generally been compiled and which is + meant to be interpreted by a computer as a program. + +- ‘The Licensor’: the natural or legal person that distributes or communicates + the Work under the Licence. + +- ‘Contributor(s)’: any natural or legal person who modifies the Work under the + Licence, or otherwise contributes to the creation of a Derivative Work. + +- ‘The Licensee’ or ‘You’: any natural or legal person who makes any usage of + the Work under the terms of the Licence. + +- ‘Distribution’ or ‘Communication’: any act of selling, giving, lending, + renting, distributing, communicating, transmitting, or otherwise making + available, online or offline, copies of the Work or providing access to its + essential functionalities at the disposal of any other natural or legal + person. + +2. Scope of the rights granted by the Licence + +The Licensor hereby grants You a worldwide, royalty-free, non-exclusive, +sublicensable licence to do the following, for the duration of copyright vested +in the Original Work: + +- use the Work in any circumstance and for all usage, +- reproduce the Work, +- modify the Work, and make Derivative Works based upon the Work, +- communicate to the public, including the right to make available or display + the Work or copies thereof to the public and perform publicly, as the case may + be, the Work, +- distribute the Work or copies thereof, +- lend and rent the Work or copies thereof, +- sublicense rights in the Work or copies thereof. + +Those rights can be exercised on any media, supports and formats, whether now +known or later invented, as far as the applicable law permits so. + +In the countries where moral rights apply, the Licensor waives his right to +exercise his moral right to the extent allowed by law in order to make effective +the licence of the economic rights here above listed. + +The Licensor grants to the Licensee royalty-free, non-exclusive usage rights to +any patents held by the Licensor, to the extent necessary to make use of the +rights granted on the Work under this Licence. + +3. Communication of the Source Code + +The Licensor may provide the Work either in its Source Code form, or as +Executable Code. If the Work is provided as Executable Code, the Licensor +provides in addition a machine-readable copy of the Source Code of the Work +along with each copy of the Work that the Licensor distributes or indicates, in +a notice following the copyright notice attached to the Work, a repository where +the Source Code is easily and freely accessible for as long as the Licensor +continues to distribute or communicate the Work. + +4. Limitations on copyright + +Nothing in this Licence is intended to deprive the Licensee of the benefits from +any exception or limitation to the exclusive rights of the rights owners in the +Work, of the exhaustion of those rights or of other applicable limitations +thereto. + +5. Obligations of the Licensee + +The grant of the rights mentioned above is subject to some restrictions and +obligations imposed on the Licensee. Those obligations are the following: + +Attribution right: The Licensee shall keep intact all copyright, patent or +trademarks notices and all notices that refer to the Licence and to the +disclaimer of warranties. The Licensee must include a copy of such notices and a +copy of the Licence with every copy of the Work he/she distributes or +communicates. The Licensee must cause any Derivative Work to carry prominent +notices stating that the Work has been modified and the date of modification. + +Copyleft clause: If the Licensee distributes or communicates copies of the +Original Works or Derivative Works, this Distribution or Communication will be +done under the terms of this Licence or of a later version of this Licence +unless the Original Work is expressly distributed only under this version of the +Licence — for example by communicating ‘EUPL v. 1.2 only’. The Licensee +(becoming Licensor) cannot offer or impose any additional terms or conditions on +the Work or Derivative Work that alter or restrict the terms of the Licence. + +Compatibility clause: If the Licensee Distributes or Communicates Derivative +Works or copies thereof based upon both the Work and another work licensed under +a Compatible Licence, this Distribution or Communication can be done under the +terms of this Compatible Licence. For the sake of this clause, ‘Compatible +Licence’ refers to the licences listed in the appendix attached to this Licence. +Should the Licensee's obligations under the Compatible Licence conflict with +his/her obligations under this Licence, the obligations of the Compatible +Licence shall prevail. + +Provision of Source Code: When distributing or communicating copies of the Work, +the Licensee will provide a machine-readable copy of the Source Code or indicate +a repository where this Source will be easily and freely available for as long +as the Licensee continues to distribute or communicate the Work. + +Legal Protection: This Licence does not grant permission to use the trade names, +trademarks, service marks, or names of the Licensor, except as required for +reasonable and customary use in describing the origin of the Work and +reproducing the content of the copyright notice. + +6. Chain of Authorship + +The original Licensor warrants that the copyright in the Original Work granted +hereunder is owned by him/her or licensed to him/her and that he/she has the +power and authority to grant the Licence. + +Each Contributor warrants that the copyright in the modifications he/she brings +to the Work are owned by him/her or licensed to him/her and that he/she has the +power and authority to grant the Licence. + +Each time You accept the Licence, the original Licensor and subsequent +Contributors grant You a licence to their contributions to the Work, under the +terms of this Licence. + +7. Disclaimer of Warranty + +The Work is a work in progress, which is continuously improved by numerous +Contributors. It is not a finished work and may therefore contain defects or +‘bugs’ inherent to this type of development. + +For the above reason, the Work is provided under the Licence on an ‘as is’ basis +and without warranties of any kind concerning the Work, including without +limitation merchantability, fitness for a particular purpose, absence of defects +or errors, accuracy, non-infringement of intellectual property rights other than +copyright as stated in Article 6 of this Licence. + +This disclaimer of warranty is an essential part of the Licence and a condition +for the grant of any rights to the Work. + +8. Disclaimer of Liability + +Except in the cases of wilful misconduct or damages directly caused to natural +persons, the Licensor will in no event be liable for any direct or indirect, +material or moral, damages of any kind, arising out of the Licence or of the use +of the Work, including without limitation, damages for loss of goodwill, work +stoppage, computer failure or malfunction, loss of data or any commercial +damage, even if the Licensor has been advised of the possibility of such damage. +However, the Licensor will be liable under statutory product liability laws as +far such laws apply to the Work. + +9. Additional agreements + +While distributing the Work, You may choose to conclude an additional agreement, +defining obligations or services consistent with this Licence. However, if +accepting obligations, You may act only on your own behalf and on your sole +responsibility, not on behalf of the original Licensor or any other Contributor, +and only if You agree to indemnify, defend, and hold each Contributor harmless +for any liability incurred by, or claims asserted against such Contributor by +the fact You have accepted any warranty or additional liability. + +10. Acceptance of the Licence + +The provisions of this Licence can be accepted by clicking on an icon ‘I agree’ +placed under the bottom of a window displaying the text of this Licence or by +affirming consent in any other similar way, in accordance with the rules of +applicable law. Clicking on that icon indicates your clear and irrevocable +acceptance of this Licence and all of its terms and conditions. + +Similarly, you irrevocably accept this Licence and all of its terms and +conditions by exercising any rights granted to You by Article 2 of this Licence, +such as the use of the Work, the creation by You of a Derivative Work or the +Distribution or Communication by You of the Work or copies thereof. + +11. Information to the public + +In case of any Distribution or Communication of the Work by means of electronic +communication by You (for example, by offering to download the Work from a +remote location) the distribution channel or media (for example, a website) must +at least provide to the public the information requested by the applicable law +regarding the Licensor, the Licence and the way it may be accessible, concluded, +stored and reproduced by the Licensee. + +12. Termination of the Licence + +The Licence and the rights granted hereunder will terminate automatically upon +any breach by the Licensee of the terms of the Licence. + +Such a termination will not terminate the licences of any person who has +received the Work from the Licensee under the Licence, provided such persons +remain in full compliance with the Licence. + +13. Miscellaneous + +Without prejudice of Article 9 above, the Licence represents the complete +agreement between the Parties as to the Work. + +If any provision of the Licence is invalid or unenforceable under applicable +law, this will not affect the validity or enforceability of the Licence as a +whole. Such provision will be construed or reformed so as necessary to make it +valid and enforceable. + +The European Commission may publish other linguistic versions or new versions of +this Licence or updated versions of the Appendix, so far this is required and +reasonable, without reducing the scope of the rights granted by the Licence. New +versions of the Licence will be published with a unique version number. + +All linguistic versions of this Licence, approved by the European Commission, +have identical value. Parties can take advantage of the linguistic version of +their choice. + +14. Jurisdiction + +Without prejudice to specific agreement between parties, + +- any litigation resulting from the interpretation of this License, arising + between the European Union institutions, bodies, offices or agencies, as a + Licensor, and any Licensee, will be subject to the jurisdiction of the Court + of Justice of the European Union, as laid down in article 272 of the Treaty on + the Functioning of the European Union, + +- any litigation arising between other parties and resulting from the + interpretation of this License, will be subject to the exclusive jurisdiction + of the competent court where the Licensor resides or conducts its primary + business. + +15. Applicable Law + +Without prejudice to specific agreement between parties, + +- this Licence shall be governed by the law of the European Union Member State + where the Licensor has his seat, resides or has his registered office, + +- this licence shall be governed by Belgian law if the Licensor has no seat, + residence or registered office inside a European Union Member State. + +Appendix + +‘Compatible Licences’ according to Article 5 EUPL are: + +- GNU General Public License (GPL) v. 2, v. 3 +- GNU Affero General Public License (AGPL) v. 3 +- Open Software License (OSL) v. 2.1, v. 3.0 +- Eclipse Public License (EPL) v. 1.0 +- CeCILL v. 2.0, v. 2.1 +- Mozilla Public Licence (MPL) v. 2 +- GNU Lesser General Public Licence (LGPL) v. 2.1, v. 3 +- Creative Commons Attribution-ShareAlike v. 3.0 Unported (CC BY-SA 3.0) for + works other than software +- European Union Public Licence (EUPL) v. 1.1, v. 1.2 +- Québec Free and Open-Source Licence — Reciprocity (LiLiQ-R) or Strong + Reciprocity (LiLiQ-R+). + +The European Commission may update this Appendix to later versions of the above +licences without producing a new version of the EUPL, as long as they provide +the rights granted in Article 2 of this Licence and protect the covered Source +Code from exclusive appropriation. + +All other changes or additions to this Appendix require the production of a new +EUPL version. diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/http2-client-grpc-effectful.cabal b/http2-client-grpc-effectful.cabal new file mode 100644 --- /dev/null +++ b/http2-client-grpc-effectful.cabal @@ -0,0 +1,61 @@ +cabal-version: 3.0 +name: http2-client-grpc-effectful +version: 1.0.0 +synopsis: Effectful gRPC client built on http2-client-effectful +description: + client for the @@ ecosystem, built on top of @@. + +homepage: https://digital-autonomy.institute +license: EUPL-1.2 +license-file: LICENCE +author: IDA +maintainer: IDA +bug-reports: https://issues.digital-autonomy.institute +category: Web +build-type: Simple +extra-source-files: + CHANGELOG.md + +common common + default-language: Haskell2010 + ghc-options: + -Weverything + -Wno-unsafe + -Wno-missing-safe-haskell-mode + -Wno-missing-export-lists + -Wno-missing-import-lists + -Wno-missing-kind-signatures + -Wno-all-missed-specialisations + -Wno-missing-role-annotations + + default-extensions: + BlockArguments + DataKinds + DerivingStrategies + ExplicitNamespaces + FlexibleContexts + ImportQualifiedPost + LambdaCase + NoImplicitPrelude + OverloadedStrings + RecordWildCards + ScopedTypeVariables + TypeApplications + TypeFamilies + TypeOperators + +library + import: common + hs-source-dirs: src + build-depends: + base >=4.10 && <5, + binary >=0.8 && <0.9, + bytestring >=0.12 && <0.13, + effectful >=2.6 && <2.7, + http-types >=0.12 && <0.13, + http2 >=5.3 && <5.4, + http2-client-effectful >=1.0 && <1.1, + http2-grpc-types >=0.5 && <0.6, + + exposed-modules: + Effectful.GrpcClient diff --git a/src/Effectful/GrpcClient.hs b/src/Effectful/GrpcClient.hs new file mode 100644 --- /dev/null +++ b/src/Effectful/GrpcClient.hs @@ -0,0 +1,667 @@ +{-# LANGUAGE BangPatterns #-} +{-# LANGUAGE Trustworthy #-} + +-- | +-- Module : Effectful.Http2Client +-- Copyright : (c) 2026 Institute for Digital Autonomy +-- License : EUPL-1.2 +-- Maintainer : IDA +-- +-- client for the @@ ecosystem, built on top of the 'Http2Client' effect. +-- +-- This library is inspired by @@. +-- However, as @http2-client-grpc@ is heavily outdated, we do not build on top of it. +-- +-- = Overview +-- +-- An 'RPCCall' describes a gRPC call: the RPC identity together with a +-- handler that drives one HTTP\/2 stream. This module provides four such +-- handlers, one per gRPC call shape: +-- +-- * 'singleRequest' (unary): send one request message, get one reply. +-- * 'streamReply' (server-streaming): send one request message, fold over +-- a stream of replies to produce a single result. +-- * 'streamRequest' (client-streaming): drive a stream of request messages +-- from a state-passing action, then get one reply. +-- * 'steppedBiDiStream' (bidirectional streaming): interleave request and +-- reply messages, driven by an explicit state machine (see 'BiDiStep'). +-- +-- Pass the resulting 'RPCCall' to 'open' to run it against a live 'Http2Client' connection. +-- Every call shape rejects server push promises with 'GrpcUnallowedPushPromise', +-- since gRPC doesn't use them. +-- +-- Calls can fail with either 'ClientError' (from the underlying HTTP\/2 connection) +-- or 'GrpcError' (gRPC-level failures: bad status, malformed messages, protocol violations); +-- handle both in the calling effect stack. +-- +-- = Example Usage +-- +-- The examples below stub out protobuf message encoding and decoding +-- with 'undefined', since they are out of scope for this module. +-- In a real application these instances come from your protobuf codegen or hand-written instances. +-- +-- === Open a connection and run RPCs against it +-- +-- > import Effectful +-- > import Effectful.Error.Static (runErrorNoCallStack) +-- > import Effectful.Http2Client +-- > import Effectful.GrpcClient +-- > +-- > main :: IO () +-- > main = do +-- > result <- runEff . runErrorNoCallStack @ClientError . runErrorNoCallStack @GrpcError $ do +-- > frame <- newHttp2FrameConnection "localhost" 50051 Nothing +-- > runHttp2Client frame 4096 4096 mempty defaultGoAwayHandler ignoreFallbackHandler $ do +-- > -- ... call 'open' here, see the recipes below ... +-- > pure () +-- > print result +-- +-- +-- === Make a unary call +-- +-- Call 'singleRequest' when the RPC sends one message and expects one reply. +-- +-- > import Data.Text (Text) +-- > import Network.GRPC.HTTP2.Proto3Wire (Proto3WireEncoder (..), RPC) +-- > +-- > newtype HelloRequest = HelloRequest { name :: Text } +-- > newtype HelloReply = HelloReply { message :: Text } +-- > +-- > instance Proto3WireEncoder HelloRequest where +-- > proto3WireEncode = undefined -- protobuf decoding & encoding omitted for brevity +-- > proto3WireDecode = undefined +-- > +-- > instance Proto3WireEncoder HelloReply where +-- > proto3WireEncode = undefined +-- > proto3WireDecode = undefined +-- > +-- > sayHelloRpc :: RPC +-- > sayHelloRpc = +-- > RPC +-- > { pkg = "example.hello" +-- > , srv = "Greeter" +-- > , meth = "SayHello" +-- > } +-- > +-- > sayHello +-- > :: (Http2Client :> es, Error ClientError :> es, Error GrpcError :> es) +-- > => HelloRequest +-- > -> Eff es (Either TooMuchConcurrency (GrpcReply HelloReply)) +-- > sayHello req = +-- > open +-- > "localhost:50051" +-- > [] +-- > (Timeout 5) +-- > (Encoding uncompressed) +-- > (Decoding uncompressed) +-- > (singleRequest sayHelloRpc req) +-- +-- +-- == Consume a server-streaming reply +-- +-- Call 'streamReply' to fold over every message the server sends back before it ends the stream. +-- +-- > data Rectangle = Rectangle +-- > data Feature = Feature +-- > +-- > instance Proto3WireEncoder Rectangle where +-- > proto3WireEncode = undefined -- protobuf decoding & encoding omitted for brevity +-- > proto3WireDecode = undefined +-- > +-- > instance Proto3WireEncoder Feature where +-- > proto3WireEncode = undefined +-- > proto3WireDecode = undefined +-- > +-- > listFeaturesRpc :: RPC +-- > listFeaturesRpc = +-- > RPC +-- > { pkg = "routeguide" +-- > , srv = "RouteGuide" +-- > , meth = "ListFeatures" +-- > } +-- > +-- > listFeatures +-- > :: (Http2Client :> es, Error ClientError :> es, Error GrpcError :> es) +-- > => Rectangle +-- > -> Eff es (Either TooMuchConcurrency ([Feature], [Header], [Header])) +-- > listFeatures area = +-- > open +-- > "localhost:50051" +-- > [] +-- > (Timeout 30) +-- > (Encoding uncompressed) +-- > (Decoding uncompressed) +-- > (streamReply listFeaturesRpc area [] (\acc _headers feature -> pure (feature : acc))) +-- +-- +-- == Stream a sequence of requests +-- +-- Call 'streamRequest' with a state-passing action that yields the next +-- message to send, or 'StreamDone' once it has nothing left to send. +-- +-- > data Point = Point +-- > data RouteSummary = RouteSummary +-- > +-- > instance Proto3WireEncoder Point where +-- > proto3WireEncode = undefined -- protobuf decoding & encoding omitted for brevity +-- > proto3WireDecode = undefined +-- > +-- > instance Proto3WireEncoder RouteSummary where +-- > proto3WireEncode = undefined +-- > proto3WireDecode = undefined +-- > +-- > recordRouteRpc :: RPC +-- > recordRouteRpc = +-- > RPC +-- > { pkg = "routeguide" +-- > , srv = "RouteGuide" +-- > , meth = "RecordRoute" +-- > } +-- > +-- > recordRoute +-- > :: (Http2Client :> es, Error ClientError :> es, Error GrpcError :> es) +-- > => [Point] +-- > -> Eff es (Either TooMuchConcurrency ([Point], GrpcReply RouteSummary)) +-- > recordRoute points = +-- > open +-- > "localhost:50051" +-- > [] +-- > (Timeout 30) +-- > (Encoding uncompressed) +-- > (Decoding uncompressed) +-- > (streamRequest recordRouteRpc points nextPoint) +-- > where +-- > nextPoint [] = pure ([], Left StreamDone) +-- > nextPoint (p : ps) = pure (ps, Right (Uncompressed, p)) +-- +-- +-- == Drive a bidirectional stream as a state machine +-- +-- Call 'steppedBiDiStream' when requests and replies interleave. +-- Write the program as a 'RunBiDiStep': given the current state, +-- choose to send a message ('SendInput'), wait for one ('WaitOutput'), or finish ('Abort'). +-- The state carries whatever the program needs to track between steps: +-- Here, the notes still to send, the notes received so far, and whether +-- the server has already sent trailers. +-- +-- > data RouteNote = RouteNote +-- > +-- > instance Proto3WireEncoder RouteNote where +-- > proto3WireEncode = undefined -- protobuf decoding & encoding omitted for brevity +-- > proto3WireDecode = undefined +-- > +-- > routeChatRpc :: RPC +-- > routeChatRpc = +-- > RPC +-- > { pkg = "routeguide" +-- > , srv = "RouteGuide" +-- > , meth = "RouteChat" +-- > } +-- > +-- > data ChatState = ChatState +-- > { pending :: [RouteNote] +-- > , received :: [RouteNote] +-- > , draining :: Bool +-- > } +-- > +-- > routeChat +-- > :: (Http2Client :> es, Error ClientError :> es, Error GrpcError :> es) +-- > => [RouteNote] +-- > -> Eff es (Either TooMuchConcurrency [RouteNote]) +-- > routeChat outbox = +-- > fmap (fmap received) $ +-- > open +-- > "localhost:50051" +-- > [] +-- > (Timeout 60) +-- > (Encoding uncompressed) +-- > (Decoding uncompressed) +-- > (steppedBiDiStream routeChatRpc (ChatState outbox [] False) step) +-- > where +-- > step st +-- > | (n : ns) <- pending st = +-- > pure (st { pending = ns }, SendInput Uncompressed n) +-- > | draining st = +-- > pure (st, Abort) +-- > | otherwise = +-- > pure (st, WaitOutput handleMsg handleTrailers) +-- > +-- > handleMsg _headers st note = pure st { received = note : received st } +-- > handleTrailers _headers st _trailers = pure st { draining = True } +module Effectful.GrpcClient + ( -- * RPC operations + open + , singleRequest + , streamReply + , streamRequest + , steppedBiDiStream + + -- * Types + , RPCCall (..) + , GrpcReply (..) + , GrpcError (..) + , CompressMode (..) + , StreamDone (..) + , BiDiStep (..) + , RunBiDiStep + , HandleMessageStep + , HandleTrailersStep + + -- * Re-exports from http2-grpc-types + , IsRPC (..) + , GRPCInput + , GRPCOutput + , Authority + , Timeout (..) + , Compression + , Encoding (..) + , Decoding (..) + , GRPCStatus (..) + , GRPCStatusCode (..) + , uncompressed + , gzip + ) +where + +import Data.Binary.Get (Decoder (..), pushChunk, pushEndOfInput) +import Data.ByteString (ByteString) +import Data.ByteString qualified as ByteString +import Data.ByteString.Builder qualified as Builder +import Data.ByteString.Char8 qualified as Char8 +import Data.ByteString.Lazy qualified as LazyByteString +import Data.Maybe (fromMaybe) +import Effectful +import Effectful.Error.Static (Error, throwError) +import Effectful.Http2Client +import Network.GRPC.HTTP2.Encoding + ( Compression (..) + , Decoding (..) + , Encoding (..) + , GRPCInput (..) + , GRPCOutput (..) + , fromDecoder + , grpcCompressionHV + , gzip + , uncompressed + ) +--- + +import Network.GRPC.HTTP2.Types + ( Authority + , GRPCStatus (..) + , GRPCStatusCode (..) + , IsRPC (..) + , Timeout (..) + , grpcAcceptEncodingH + , grpcAcceptEncodingHVdefault + , grpcContentTypeHV + , grpcEncodingH + , grpcMessageH + , grpcStatusH + , grpcTimeoutH + , showTimeout + , statusCodeForTrailer + ) +import Network.HTTP.Types (Header) +import Network.HTTP2.Frame (ErrorCode, setEndHeader, setEndStream) +import Prelude + +data RPCCall es r a = RPCCall + { rpcFromCall :: r + , runRPC + :: Http2Stream es + -> IncomingFlowControl es + -> OutgoingFlowControl es + -> Encoding + -> Decoding + -> Eff es a + } + +-- | Main handler to perform gRPC calls to a service. +open + :: ( Http2Client :> es + , Error ClientError :> es + , IsRPC r + ) + => Authority + -- ^ The HTTP2-Authority portion of the URL (e.g., "dicioccio.fr:7777"). + -> [Header] + -- ^ A set of HTTP2 headers (e.g., for adding authentication headers). + -> Timeout + -- ^ Timeout in seconds. + -> Encoding + -- ^ Compression used for encoding. + -> Decoding + -- ^ Compression allowed for decoding + -> RPCCall es r a + -- ^ The actual RPC handler. + -> Eff es (Either TooMuchConcurrency a) +open authority extraheaders timeout encoding decoding call = do + withHttp2Stream + \stream -> do + let + initStream = _headers stream headers setEndHeader + handler isfc osfc = runRPC call stream isfc osfc encoding decoding + in + StreamDefinition initStream handler + where + headers = + [ (":method", "POST") + , (":scheme", "http") + , (":authority", authority) + , (":path", path (rpcFromCall call)) + , (grpcTimeoutH, showTimeout timeout) + , (grpcEncodingH, grpcCompressionHV (_getEncodingCompression encoding)) + , + ( grpcAcceptEncodingH + , mconcat [grpcAcceptEncodingHVdefault, ",", grpcCompressionHV (_getDecodingCompression decoding)] + ) + , ("content-type", grpcContentTypeHV) + , ("te", "trailers") + ] + <> extraheaders + +-- | The result of a successful gRPC call. +data GrpcReply a = GrpcReply + { grpcHeaders :: [Header] + -- ^ Response headers from the server. + , grpcTrailers :: [Header] + -- ^ Response trailers (contain grpc-status). + , grpcResult :: Either String a + -- ^ Decoded response message, or a decode error. + } + deriving stock (Show) + +-- | Errors that can occur during a gRPC call. +data GrpcError + = -- | The HTTP\/2 stream was refused due to concurrency limits. + GrpcTooMuchConcurrency + | -- | A transport-level error on the HTTP\/2 stream. + GrpcStreamError ErrorCode + | -- | The server returned a non-OK gRPC status. + GrpcStatusError GRPCStatus + | -- | The server sent an HTTP\/2 PUSH_PROMISE, which gRPC forbids. + GrpcUnallowedPushPromise + | -- | The RPC reached an invalid point in its state machine (e.g. the + -- server omitted the initial headers). + GrpcInvalidState String + | -- | A response message could not be decoded. + GrpcParseError String + deriving stock (Show) + +-- | A 'PushPromiseHandler' that rejects every server push by raising a +-- 'GrpcUnallowedPushPromise' error. gRPC disables HTTP\/2 push-promises. +throwOnPushPromise :: (Error GrpcError :> es) => PushPromiseHandler es +throwOnPushPromise _ _ _ _ _ = throwError GrpcUnallowedPushPromise + +-- | Whether a single message should be compressed when sent. +data CompressMode = Compressed | Uncompressed + +-- | Marker returned by a client-streaming handler when it is done sending. +data StreamDone = StreamDone + +-- | Handler for a received message in a stepped bidirectional stream. +type HandleMessageStep es o a = [Header] -> a -> o -> Eff es a + +-- | Handler for received trailers in a stepped bidirectional stream. +type HandleTrailersStep es a = [Header] -> a -> [Header] -> Eff es a + +-- | One step of a stepped bidirectional stream. +data BiDiStep es i o a + = -- | Finalise and return the current state. + Abort + | -- | Send a single message. + SendInput !CompressMode !i + | -- | Wait for information from the server; handlers may modify the state. + WaitOutput (HandleMessageStep es o a) (HandleTrailersStep es a) + +-- | State-based program driving a stepped bidirectional stream. +type RunBiDiStep es i o a = a -> Eff es (a, BiDiStep es i o a) + +-- | Decode and validate the unary reply for an RPC. +waitReply + :: forall r o es + . ( GRPCOutput r o + , Http2Client :> es + , Error ClientError :> es + , Error GrpcError :> es + ) + => r + -> Decoding + -> Http2Stream es + -> IncomingFlowControl es + -> Eff es (GrpcReply o) +waitReply rpc decoding stream flowControl = do + response <- fromStreamResult <$> waitStream stream flowControl throwOnPushPromise + case response of + Left errCode -> throwError $ GrpcStreamError errCode + Right (hdrs, dat, mtrls) -> do + let trls = fromMaybe [] mtrls + -- A grpc-message in the headers indicates a Trailers-Only error + -- response and there is no message body to decode. + result = case lookup grpcMessageH hdrs of + Just errMsg -> Left (Char8.unpack errMsg) + Nothing -> + fromDecoder . pushEndOfInput $ + pushChunk (decodeOutput rpc decompress) dat + case statusFromHeaders $ trls <> hdrs of + Just status@(GRPCStatus code _) + | code /= OK -> throwError $ GrpcStatusError status + _ -> + pure + GrpcReply + { grpcHeaders = hdrs + , grpcTrailers = trls + , grpcResult = result + } + where + decompress = _getDecodingCompression decoding + +statusFromHeaders :: [Header] -> Maybe GRPCStatus +statusFromHeaders hs = do + code <- statusCodeForTrailer =<< lookup grpcStatusH hs + let message = fromMaybe "" $ lookup grpcMessageH hs + pure $ GRPCStatus code message + +sendSingleMessage + :: forall r i es + . (GRPCInput r i) + => r + -- ^ RPC to call. + -> i + -- ^ RPC's input. + -> Encoding + -> FlagSetter + -> OutgoingFlowControl es + -> Http2Stream es + -> OutgoingFlowControl es + -> Eff es () +sendSingleMessage rpc msg encoding flagMod connectionFlowControl stream streamFlowControl = do + goUpload + . LazyByteString.toStrict + . Builder.toLazyByteString + . encodeInput rpc (_getEncodingCompression encoding) + $ msg + where + goUpload :: ByteString -> Eff es () + 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 stream flagMod dat + else do + _sendData stream id $ ByteString.take got dat + goUpload $ ByteString.drop got dat + +-- | Perform a unary gRPC call: send one message, then wait for the single reply. +singleRequest + :: ( Http2Client :> es + , Error ClientError :> es + , Error GrpcError :> es + , GRPCInput r i + , GRPCOutput r o + ) + => r + -- ^ RPC to call. + -> i + -- ^ RPC's input. + -> RPCCall es r (GrpcReply o) +singleRequest rpc msg = RPCCall rpc $ \stream isfc osfc encoding decoding -> do + ocfc <- _outgoingFlowControl + sendSingleMessage rpc msg encoding setEndStream ocfc stream osfc + waitReply rpc decoding stream isfc + +-- | Perform a server-streaming gRPC call: send one message, then fold over the +-- stream of responses. Returns the final fold state together with the initial +-- headers and the final trailers. +streamReply + :: ( Http2Client :> es + , Error ClientError :> es + , Error GrpcError :> es + , GRPCInput r i + , GRPCOutput r o + ) + => r + -- ^ RPC to call. + -> i + -- ^ The input. + -> a + -- ^ An initial state. + -> (a -> [Header] -> o -> Eff es a) + -- ^ A state-passing handler called with the initial headers and each + -- message read. + -> RPCCall es r (a, [Header], [Header]) +streamReply rpc req v0 handler = RPCCall rpc $ \stream isfc osfc encoding decoding -> do + ocfc <- _outgoingFlowControl + let + newDecoder = decodeOutput rpc $ _getDecodingCompression decoding + + handleAllChunks v1 hdrs decode dat exitLoop = + case pushChunk decode dat of + Done unusedDat _ (Right val) -> do + v2 <- handler v1 hdrs val + handleAllChunks v2 hdrs newDecoder unusedDat exitLoop + Done _ _ (Left err) -> + throwError (GrpcParseError ("done-error: " <> err)) + Fail _ _ err -> + throwError (GrpcParseError ("fail-error: " <> err)) + partial@(Partial _) -> + exitLoop v1 partial hdrs + + loop v1 decode hdrs = + _waitEvent stream >>= \case + StreamPushPromiseEvent{} -> + throwError GrpcUnallowedPushPromise + StreamHeadersEvent _ trls -> + pure (v1, hdrs, trls) + StreamErrorEvent _ _ -> + throwError (GrpcInvalidState "stream error") + StreamDataEvent _ dat -> do + _addCredit isfc (ByteString.length dat) + _ <- _consumeCredit isfc (ByteString.length dat) + _ <- _updateWindow isfc + handleAllChunks v1 hdrs decode dat loop + sendSingleMessage rpc req encoding setEndStream ocfc stream osfc + _waitEvent stream >>= \case + StreamHeadersEvent _ hdrs -> + loop v0 newDecoder hdrs + _ -> + throwError (GrpcInvalidState "no headers") + +-- | Perform a client-streaming gRPC call: repeatedly pull messages to send from +-- a state-passing action, then read the unary reply. +streamRequest + :: ( Http2Client :> es + , Error ClientError :> es + , Error GrpcError :> es + , GRPCInput r i + , GRPCOutput r o + ) + => r + -- ^ RPC to call. + -> a + -- ^ An initial state. + -> (a -> Eff es (a, Either StreamDone (CompressMode, i))) + -- ^ A state-passing action retrieving the next message to send. + -> RPCCall es r (a, GrpcReply o) +streamRequest rpc v0 handler = RPCCall rpc $ \stream isfc streamFlow encoding decoding -> do + ocfc <- _outgoingFlowControl + let + go v1 = do + (v2, nextEvent) <- handler v1 + case nextEvent of + Right (doCompress, msg) -> do + let compress = case doCompress of + Compressed -> _getEncodingCompression encoding + Uncompressed -> uncompressed + sendSingleMessage rpc msg (Encoding compress) id ocfc stream streamFlow + go v2 + Left StreamDone -> do + _sendData stream setEndStream "" + reply <- waitReply rpc decoding stream isfc + pure (v2, reply) + go v0 + +-- | Perform a stepped bidirectional gRPC call driven by a synchronous state +-- machine. See 'BiDiStep' and 'RunBiDiStep'. +steppedBiDiStream + :: ( Http2Client :> es + , Error ClientError :> es + , Error GrpcError :> es + , GRPCInput r i + , GRPCOutput r o + ) + => r + -- ^ RPC to call. + -> a + -- ^ An initial state. + -> RunBiDiStep es i o a + -- ^ The program. + -> RPCCall es r a +steppedBiDiStream rpc v0 handler = RPCCall rpc $ \stream isfc streamFlow encoding decoding -> do + ocfc <- _outgoingFlowControl + let + newDecoder = decodeOutput rpc $ _getDecodingCompression decoding + + goStep _ _ (v1, Abort) = do + _sendData stream setEndStream "" + pure v1 + goStep mhdrs decode (v1, SendInput doCompress msg) = do + let compress = case doCompress of + Compressed -> _getEncodingCompression encoding + Uncompressed -> uncompressed + sendSingleMessage rpc msg (Encoding compress) id ocfc stream streamFlow + handler v1 >>= goStep mhdrs decode + goStep jhdrs@(Just hdrs) decode unchanged@(v1, WaitOutput handleMsg handleEof) = + _waitEvent stream >>= \case + StreamPushPromiseEvent{} -> + throwError GrpcUnallowedPushPromise + StreamHeadersEvent _ trls -> do + v2 <- handleEof hdrs v1 trls + handler v2 >>= goStep jhdrs newDecoder + StreamErrorEvent _ _ -> + throwError $ GrpcInvalidState "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) -> + throwError (GrpcParseError ("done-err: " <> err)) + Fail _ _ err -> + throwError (GrpcParseError ("done-fail: " <> err)) + partial@(Partial _) -> + goStep jhdrs partial unchanged + goStep Nothing decode unchanged = + _waitEvent stream >>= \case + StreamHeadersEvent _ hdrs -> + goStep (Just hdrs) decode unchanged + _ -> + throwError (GrpcInvalidState "no headers") + handler v0 >>= goStep Nothing newDecoder