rpc

Documentation

Index

• Constants
• Variables
• func CLIFlags(envPrefix string) []cli.Flag
• func ObtainJWTSecret(logger log.Logger, jwtSecretPath string, generateMissing bool) (eth.Bytes32, error)
• func StreamFallback[E any](fn func(ctx context.Context) (*E, error), frequency time.Duration, ...) (ethereum.Subscription, error)
• func SubscribeRPC[T any](ctx context.Context, logger log.Logger, feed *event.FeedOf[T]) (*gethrpc.Subscription, error)
• func SubscribeStream[E any](ctx context.Context, namespace string, subscriber Subscriber, dest chan *E, ...) (ethereum.Subscription, error)
• func ToGethAdminAPI(api *CommonAdminAPI) rpc.API
• type CLIConfig
  • func DefaultCLIConfig() CLIConfig
  • func ReadCLIConfig(ctx *cli.Context) CLIConfig
  • func (c CLIConfig) Check() error
• type CommonAdminAPI
  • func NewCommonAdminAPI(m metrics.RPCMetricer, log log.Logger) *CommonAdminAPI
  • func (n *CommonAdminAPI) SetLogLevel(ctx context.Context, lvlStr string) error
• type EventEntry
• type HealthzResponse
• type Middleware
• type Server
  • func NewServer(host string, port int, appVersion string, opts ...ServerOption) *Server
  • func (b *Server) AddAPI(api rpc.API)
  • func (b *Server) AddHandler(path string, handler http.Handler)
  • func (b *Server) Endpoint() string
  • func (b *Server) Start() error
  • func (b *Server) Stop() error
• type ServerOption
  • func WithAPIs(apis []rpc.API) ServerOption
  • func WithCORSHosts(hosts []string) ServerOption
  • func WithHTTPRecorder(recorder opmetrics.HTTPRecorder) ServerOption
  • func WithHealthzHandler(hdlr http.Handler) ServerOption
  • func WithHealthzPath(path string) ServerOption
  • func WithJWTSecret(secret []byte) ServerOption
  • func WithLogger(lgr log.Logger) ServerOption
  • func WithMiddleware(middleware func(http.Handler) (hdlr http.Handler)) ServerOption
  • func WithRPCPath(path string) ServerOption
  • func WithTLSConfig(tls *ServerTLSConfig) ServerOption
  • func WithVHosts(hosts []string) ServerOption
  • func WithWebsocketEnabled() ServerOption
• type ServerTLSConfig
• type Service
  • func NewService(log log.Logger, srv *Server) *Service
  • func (s *Service) Start(_ context.Context) error
  • func (s *Service) Stop(_ context.Context) error
  • func (s *Service) Stopped() bool
• type Stream
  • func NewStream[E any](log log.Logger, maxQueueSize int) *Stream[E]
  • func (evs *Stream[E]) Send(ev *E)
  • func (evs *Stream[E]) Serve() (*E, error)
  • func (evs *Stream[E]) Subscribe(ctx context.Context) (*gethrpc.Subscription, error)
• type Subscriber

Constants

const (
	ListenAddrFlagName  = "rpc.addr"
	PortFlagName        = "rpc.port"
	EnableAdminFlagName = "rpc.enable-admin"
)

const OutOfEventsErrCode = -39001

OutOfEventsErrCode is the RPC error-code used to signal that no buffered events are available to dequeue. A polling RPC client should back off in this case.

Variables

var ErrClosedByServer = errors.New("closed by server")

ErrClosedByServer is sent over the subscription error-channel by Subscribe when the server closes the subscription.

var ErrInvalidPort = errors.New("invalid RPC port")

Functions

func CLIFlags

func CLIFlags(envPrefix string) []cli.Flag

func ObtainJWTSecret

func ObtainJWTSecret(logger log.Logger, jwtSecretPath string, generateMissing bool) (eth.Bytes32, error)

ObtainJWTSecret attempts to read a JWT secret, and generates one if necessary. Unlike the geth rpc.ObtainJWTSecret variant, this uses local logging, makes generation optional, and does not blindly overwrite a JWT secret on any read error. Generally it is advised to generate a JWT secret if missing, as a server. Clients should not generate a JWT secret, and use the secret of the server instead.

func StreamFallback

func StreamFallback[E any](fn func(ctx context.Context) (*E, error), frequency time.Duration, dest chan *E) (ethereum.Subscription, error)

StreamFallback polls the given function for data. When the function returns a JSON RPC error with OutOfEventsErrCode error-code, the polling backs off by waiting for the given frequency time duration. When the function returns any other error, the stream is aborted, and the error is forwarded to the subscription-error channel. The dest channel is kept open after stream error, in case re-subscribing is desired.

func SubscribeRPC

func SubscribeRPC[T any](ctx context.Context, logger log.Logger, feed *event.FeedOf[T]) (*gethrpc.Subscription, error)

func SubscribeStream

func SubscribeStream[E any](ctx context.Context, namespace string, subscriber Subscriber, dest chan *E, args ...any) (ethereum.Subscription, error)

SubscribeStream subscribes to a Stream. This may return a gethrpc.ErrNotificationsUnsupported error, if subscriptions over RPC are not supported. The client should then fall back to manual RPC polling, with OutOfEventsErrCode error checks. The returned subscription has an error channel, which may send a ErrClosedByServer when the server closes the subscription intentionally. Or any of the geth RPC errors, when the connection closes or RPC fails. The args work like the Subscriber interface: the subscription identifier needs to be there.

func ToGethAdminAPI

func ToGethAdminAPI(api *CommonAdminAPI) rpc.API

Types

type CLIConfig

type CLIConfig struct {
	ListenAddr  string
	ListenPort  int
	EnableAdmin bool
}

func DefaultCLIConfig

func DefaultCLIConfig() CLIConfig

func ReadCLIConfig

func ReadCLIConfig(ctx *cli.Context) CLIConfig

func (CLIConfig) Check

func (c CLIConfig) Check() error

type CommonAdminAPI

type CommonAdminAPI struct {
	M metrics.RPCMetricer
	// contains filtered or unexported fields
}

func NewCommonAdminAPI

func NewCommonAdminAPI(m metrics.RPCMetricer, log log.Logger) *CommonAdminAPI

func (*CommonAdminAPI) SetLogLevel

func (n *CommonAdminAPI) SetLogLevel(ctx context.Context, lvlStr string) error

type EventEntry

type EventEntry[E any] struct {
	// Data wraps the actual event object. It may be nil if Close is true.
	Data *E `json:"data"`
	// Close is set to true when the server will send no further events over this subscription.
	Close bool `json:"close,omitempty"`
}

EventEntry wraps subscription data, so the server can communicate alternative metadata, such as close instructions.

type HealthzResponse

type HealthzResponse struct {
	Version string `json:"version"`
}

type Middleware

type Middleware func(next http.Handler) http.Handler

type Server

type Server struct {
	// contains filtered or unexported fields
}

func NewServer

func NewServer(host string, port int, appVersion string, opts ...ServerOption) *Server

func (*Server) AddAPI

func (b *Server) AddAPI(api rpc.API)

func (*Server) AddHandler

func (b *Server) AddHandler(path string, handler http.Handler)

AddHandler adds a custom http.Handler to the server, mapped to an absolute path

func (*Server) Endpoint

func (b *Server) Endpoint() string

Endpoint returns the HTTP endpoint without http / ws protocol prefix.

func (*Server) Start

func (b *Server) Start() error

func (*Server) Stop

func (b *Server) Stop() error

type ServerOption

type ServerOption func(b *Server)

func WithAPIs

func WithAPIs(apis []rpc.API) ServerOption

func WithCORSHosts

func WithCORSHosts(hosts []string) ServerOption

func WithHTTPRecorder

func WithHTTPRecorder(recorder opmetrics.HTTPRecorder) ServerOption

func WithHealthzHandler

func WithHealthzHandler(hdlr http.Handler) ServerOption

func WithHealthzPath

func WithHealthzPath(path string) ServerOption

func WithJWTSecret

func WithJWTSecret(secret []byte) ServerOption

WithJWTSecret adds authentication to the RPCs (HTTP, and WS pre-upgrade if enabled). The health endpoint is still available without authentication.

func WithLogger

func WithLogger(lgr log.Logger) ServerOption

func WithMiddleware

func WithMiddleware(middleware func(http.Handler) (hdlr http.Handler)) ServerOption

WithMiddleware adds an http.Handler to the rpc server handler stack The added middleware is invoked directly before the RPC callback

func WithRPCPath

func WithRPCPath(path string) ServerOption

func WithTLSConfig

func WithTLSConfig(tls *ServerTLSConfig) ServerOption

WithTLSConfig configures TLS for the RPC server If this option is passed, the server will use ListenAndServeTLS

func WithVHosts

func WithVHosts(hosts []string) ServerOption

func WithWebsocketEnabled

func WithWebsocketEnabled() ServerOption

WithWebsocketEnabled allows `ws://host:port/`, `ws://host:port/ws` and `ws://host:port/ws/` to be upgraded to a websocket JSON RPC connection.

type ServerTLSConfig

type ServerTLSConfig struct {
	Config    *tls.Config
	CLIConfig *optls.CLIConfig // paths to certificate and key files
}

type Service

type Service struct {
	// contains filtered or unexported fields
}

func NewService

func NewService(log log.Logger, srv *Server) *Service

func (*Service) Start

func (s *Service) Start(_ context.Context) error

func (*Service) Stop

func (s *Service) Stop(_ context.Context) error

func (*Service) Stopped

func (s *Service) Stopped() bool

type Stream

type Stream[E any] struct {
	// contains filtered or unexported fields
}

Stream is a queue of events (wrapped objects) that can be pulled from or subscribed to via RPC. When subscribed, no data is queued, and sent proactively to the client instead (e.g. over websocket). If not subscribed, data can be served one by one manually (e.g. polled over HTTP). At most one concurrent subscription is supported.

func NewStream

func NewStream[E any](log log.Logger, maxQueueSize int) *Stream[E]

NewStream creates a new Stream. With a maxQueueSize, to limit how many events are buffered. The oldest events are dropped first, if overflowing.

func (*Stream[E]) Send

func (evs *Stream[E]) Send(ev *E)

Send will send an event, either by enqueuing it for later retrieval, or by directly sending it to an active subscription.

func (*Stream[E]) Serve

func (evs *Stream[E]) Serve() (*E, error)

Serve serves a single event. It will return a JSON-RPC error with code OutOfEventsErrCode if no events are available to pull at this time. Serve will close any active subscription, as manual event retrieval and event-subscription are mutually exclusive modes.

func (*Stream[E]) Subscribe

func (evs *Stream[E]) Subscribe(ctx context.Context) (*gethrpc.Subscription, error)

Subscribe opens an RPC subscription that will be served with all future events. Previously buffered events will all be dropped.

type Subscriber

type Subscriber interface {
	Subscribe(ctx context.Context, namespace string, channel any, args ...any) (ethereum.Subscription, error)
}

Subscriber implements the subscribe subset of the RPC client interface. The inner geth-native Subscribe interface returns a struct subscription type, this can be interpreted as general ethereum.Subscription but may require a wrapper, like in the op-service client package.