rpc

Documentation

Index

• Constants
• func ClientFromContext[Client Pingable](ctx context.Context) Client
• func ContextValuesMiddleware(ctx context.Context, handler http.Handler) http.HandlerFunc
• func ContextWithClient[Client Pingable](ctx context.Context, client Client) context.Context
• func DefaultClientOptions(level log.Level) []connect.ClientOption
• func DefaultHandlerOptions() []connect.HandlerOption
• func Dial[Client Pingable](factory ClientFactory[Client], baseURL string, errorLevel log.Level, ...) Client
• func GetHTTPClient(url string) *http.Client
• func InitialiseClients(authenticators map[string]string)
• func IsClientAvailableInContext[Client Pingable](ctx context.Context) bool
• func IsDirectRouted(ctx context.Context) bool
• func MetadataInterceptor(errorLevel log.Level) connect.Interceptor
• func PanicInterceptor() connect.Interceptor
• func RequestKeyFromContext(ctx context.Context) (optional.Option[model.RequestKey], error)
• func RetryStreamingClientStream[Req, Resp any](ctx context.Context, retry backoff.Backoff, ...)
• func RetryStreamingServerStream[Req, Resp any](ctx context.Context, retry backoff.Backoff, req *Req, ...)
• func Serve(ctx context.Context, listen *url.URL, options ...Option) error
• func VerbFromContext(ctx context.Context) (*schema.Ref, bool)
• func VerbsFromContext(ctx context.Context) ([]*schema.Ref, bool)
• func Wait(ctx context.Context, retry backoff.Backoff, client Pingable) error
• func WithDirectRouting(ctx context.Context) context.Context
• func WithRequestName(ctx context.Context, key model.RequestKey) context.Context
• func WithVerbs(ctx context.Context, verbs []*schema.Ref) context.Context
• type ClientFactory
• type GRPCServerConstructor
• type Option
  • func GRPC[Iface, Impl Pingable](constructor GRPCServerConstructor[Iface], impl Impl, ...) Option
  • func HTTP(prefix string, handler http.Handler) Option
  • func RawGRPC[Iface, Impl any](constructor RawGRPCServerConstructor[Iface], impl Impl, ...) Option
• type Pingable
• type RawGRPCServerConstructor
• type Server
  • func NewServer(ctx context.Context, listen *url.URL, options ...Option) (*Server, error)
  • func (s *Server) Serve(ctx context.Context) error

Constants

const ShutdownGracePeriod = time.Second * 5

Functions

func ClientFromContext

func ClientFromContext[Client Pingable](ctx context.Context) Client

ClientFromContext returns the given RPC client from the context, or panics.

func ContextValuesMiddleware

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

ContextValuesMiddleware injects values from a Context into the request Context.

func ContextWithClient

func ContextWithClient[Client Pingable](ctx context.Context, client Client) context.Context

ContextWithClient returns a context with an RPC client attached.

func DefaultClientOptions

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

func DefaultHandlerOptions

func DefaultHandlerOptions() []connect.HandlerOption

func Dial

func Dial[Client Pingable](factory ClientFactory[Client], 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)

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.

func IsClientAvailableInContext

func IsClientAvailableInContext[Client Pingable](ctx context.Context) bool

func IsDirectRouted

func IsDirectRouted(ctx context.Context) bool

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

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 RequestKeyFromContext

func RequestKeyFromContext(ctx context.Context) (optional.Option[model.RequestKey], 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, 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,
	retry backoff.Backoff,
	req *Req,
	rpc func(context.Context, *connect.Request[Req]) (*connect.ServerStreamForClient[Resp], error),
	handler func(ctx context.Context, resp *Resp) error,
)

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

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(ctx context.Context, retry backoff.Backoff, client Pingable) error

Wait for a client to become available.

This will repeatedly call Ping() every 100ms until the service becomes ready. TODO: This will probably need to be smarter at some point.

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

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 WithRequestName

func WithRequestName(ctx context.Context, key model.RequestKey) context.Context

WithRequestName 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.

Types

type ClientFactory

type ClientFactory[Client Pingable] func(httpClient connect.HTTPClient, baseURL string, opts ...connect.ClientOption) Client

ClientFactory is a function that creates a new client and is typically one of the New*Client functions generated by protoc-gen-connect-go.

type GRPCServerConstructor

type GRPCServerConstructor[Iface Pingable] func(svc Iface, opts ...connect.HandlerOption) (string, http.Handler)

type Option

type Option func(*serverOptions)

func GRPC

func GRPC[Iface, Impl Pingable](constructor GRPCServerConstructor[Iface], impl Impl, options ...connect.HandlerOption) Option

GRPC is a convenience function for registering a GRPC server with default options. TODO(aat): Do we need pingable here?

func HTTP

func HTTP(prefix string, handler http.Handler) Option

HTTP adds a HTTP route to the server.

func RawGRPC

func RawGRPC[Iface, Impl any](constructor RawGRPCServerConstructor[Iface], impl Impl, options ...connect.HandlerOption) Option

RawGRPC is a convenience function for registering a GRPC server with default options without Pingable.

type Pingable

type Pingable interface {
	Ping(ctx context.Context, req *connect.Request[ftlv1.PingRequest]) (*connect.Response[ftlv1.PingResponse], error)
}

type RawGRPCServerConstructor

type RawGRPCServerConstructor[Iface any] func(svc Iface, opts ...connect.HandlerOption) (string, http.Handler)

type Server

type Server struct {
	Bind   *pubsub.Topic[*url.URL] // Will be updated with the actual bind address.
	Server *http.Server
	// contains filtered or unexported fields
}

func NewServer

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

func (*Server) Serve

func (s *Server) Serve(ctx context.Context) error

Serve runs the server, updating .Bind with the actual bind address.