rpc

func AlwaysRetry ¶

func AlwaysRetry() func(error) bool

AlwaysRetry instructs RetryStreamingServerStream to always retry the errors it encounters when supplied as the errorRetryCallback argument

func ContextValuesMiddleware ¶

func ContextValuesMiddleware(ctx context.Context, handler http.Handler) http.HandlerFunc

ContextValuesMiddleware injects values from a Context into the request Context.

func CustomOtelInterceptor ¶

func CustomOtelInterceptor() connect.Interceptor

func DefaultClientOptions ¶

func DefaultClientOptions(level log.Level) []connect.ClientOption

func DefaultHandlerOptions ¶

func DefaultHandlerOptions() []connect.HandlerOption

func Dial ¶

func Dial[Client Pingable[Req, Resp, RespPtr], Req any, Resp any, RespPtr PingResponse[Resp]](factory ClientFactory[Client, Req, Resp, RespPtr], baseURL string, errorLevel log.Level, opts ...connect.ClientOption) Client

func GetHTTPClient ¶

func GetHTTPClient(url string) *http.Client

GetHTTPClient returns a HTTP client usable for the given URL.

func InitialiseClients ¶

func InitialiseClients(authenticators map[string]string, allowInsecure bool)

InitialiseClients initialises global HTTP clients used by the RPC system.

"authenticators" are authenticator executables to use for each endpoint. The key is the URL of the endpoint, the value is the path to the authenticator executable.

"allowInsecure" skips certificate verification, making TLS susceptible to machine-in-the-middle attacks.

func IsDirectRouted ¶

func IsDirectRouted(ctx context.Context) bool

IsDirectRouted returns true if the incoming request should be directly routed and never redirected.

func IterAsGrpc ¶ added in v0.468.0

func IterAsGrpc[T any](iter iter.Seq[result.Result[*T]], stream *connect.ServerStream[T]) error

IterAsGrpc converts an iterator of results into a gRPC stream. This is a convenience function to make it easier to unit test streaming endpoints.

If the iterator returns an error, it is returned immediately, and the stream is not sent any more messages.

func MetadataInterceptor ¶

func MetadataInterceptor(errorLevel log.Level) connect.Interceptor

MetadataInterceptor propagates FTL metadata through servers and clients.

"errorLevel" is the level at which errors will be logged

func PanicInterceptor ¶

func PanicInterceptor() connect.Interceptor

PanicInterceptor intercepts panics and logs them.

func ParentRequestKeyFromContext ¶

func ParentRequestKeyFromContext(ctx context.Context) (optional.Option[key.Request], error)

func RequestKeyFromContext ¶

func RequestKeyFromContext(ctx context.Context) (optional.Option[key.Request], error)

RequestKeyFromContext returns the request key from the context, if any.

TODO: Return an Option here instead of a bool.

func RetryStreamingClientStream ¶

func RetryStreamingClientStream[Req any, Resp any](
	ctx context.Context,
	retry backoff.Backoff,
	rpc func(context.Context) *connect.ClientStreamForClient[Req, Resp],
	handler func(ctx context.Context, send func(*Req) error) error,
)

RetryStreamingClientStream will repeatedly call handler with the stream returned by "rpc" until handler returns an error or the context is cancelled.

If the stream errors, it will be closed and a new call will be issued.

func RetryStreamingServerStream ¶

func RetryStreamingServerStream[Req, Resp any](
	ctx context.Context,
	name string,
	retry backoff.Backoff,
	req *Req,
	rpc func(context.Context, *connect.Request[Req]) (*connect.ServerStreamForClient[Resp], error),
	handler func(ctx context.Context, resp *Resp) error,
	errorRetryCallback func(err error) bool,
)

RetryStreamingServerStream will repeatedly call handler with responses from the stream returned by "rpc" until either the context is cancelled or the errorRetryCallback returns false.

func Serve ¶

func Serve(ctx context.Context, listen *url.URL, options ...Option) error

Serve starts a HTTP and Connect gRPC server with sane defaults for FTL.

Blocks until the context is cancelled.

func VerbFromContext ¶

func VerbFromContext(ctx context.Context) (*schema.Ref, bool)

VerbFromContext returns the current module.verb of the current request.

func VerbsFromContext ¶

func VerbsFromContext(ctx context.Context) ([]*schema.Ref, bool)

VerbsFromContext returns the module.verb chain of the current request.

func Wait ¶

func Wait[Req any, Resp any, RespPtr PingResponse[Resp]](ctx context.Context, retry backoff.Backoff, deadline time.Duration, client Pingable[Req, Resp, RespPtr]) error

Wait for a client to become available.

This will repeatedly call Ping() according to the retry policy until the client is ready or the deadline is reached.

If "ctx" is cancelled this will return ctx.Err()

Usually rpc errors are logged, but this function will silence ping call errors, and returns the last error if the deadline is reached.

func WithDirectRouting ¶

func WithDirectRouting(ctx context.Context) context.Context

WithDirectRouting ensures any hops in Verb routing do not redirect.

This is used so that eg. calls from Drives do not create recursive loops when calling back to the Agent.

func WithParentRequestKey ¶

func WithParentRequestKey(ctx context.Context, key key.Request) context.Context

func WithRequestKey ¶

func WithRequestKey(ctx context.Context, key key.Request) context.Context

WithRequestKey adds the request key to the context.

func WithVerbs ¶

func WithVerbs(ctx context.Context, verbs []*schema.Ref) context.Context

WithVerbs adds the module.verb chain from the current request to the context.