{-# LANGUAGE BangPatterns #-}
{-# LANGUAGE Trustworthy #-}
-- |
-- Module : Effectful.Http2Client
-- Copyright : (c) 2026 Institute for Digital Autonomy
-- License : EUPL-1.2
-- Maintainer : IDA
--
-- <https://grpc.io gRPC> client for the @<https://hackage.haskell.org/package/effectful Effectful>@ ecosystem, built on top of the 'Http2Client' effect.
--
-- This library is inspired by @<https://hackage.haskell.org/package/http2-client-grpc http2-client-grpc>@.
-- 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