arpc

Documentation

Index

• type HandlerFunc
• type ReconnectConfig
• type Request
• type Response
• type Router
  • func NewRouter() *Router
  • func (r *Router) Handle(method string, handler HandlerFunc)
  • func (r *Router) ServeStream(stream *smux.Stream)
• type Session
  • func DialWithBackoff(ctx context.Context, dialFunc func() (net.Conn, error), ...) (*Session, error)
  • func HijackUpgradeHTTP(w http.ResponseWriter, r *http.Request, config *smux.Config) (*Session, error)
  • func NewClientSession(conn net.Conn, config *smux.Config) (*Session, error)
  • func NewServerSession(conn net.Conn, config *smux.Config) (*Session, error)
  • func UpgradeHTTPClient(conn net.Conn, requestPath, host string, headers http.Header, ...) (*Session, error)
  • func (s *Session) Call(method string, payload interface{}) (*Response, error)
  • func (s *Session) CallContext(ctx context.Context, method string, payload interface{}) (*Response, error)
  • func (s *Session) CallJSON(ctx context.Context, method string, payload interface{}, v interface{}) error
  • func (s *Session) CallWithHeaders(ctx context.Context, method string, payload interface{}, headers http.Header) (*Response, error)
  • func (s *Session) Close() error
  • func (s *Session) EnableAutoReconnect(rc *ReconnectConfig)
  • func (s *Session) Serve(router *Router) error

Types

type HandlerFunc

type HandlerFunc func(req Request) (Response, error)

HandlerFunc is the type of function that can handle a Request.

type ReconnectConfig

type ReconnectConfig struct {
	// AutoReconnect must be true to enable automatic reconnection.
	AutoReconnect bool

	// DialFunc is a function that establishes a new raw connection.
	DialFunc func() (net.Conn, error)

	// UpgradeFunc upgrades a raw connection (e.g. performing HTTP upgrade)
	// and returns a new Session.
	UpgradeFunc func(net.Conn) (*Session, error)

	// InitialBackoff is the backoff duration for the first reconnect attempt.
	InitialBackoff time.Duration

	// MaxBackoff is the maximum allowed backoff duration.
	MaxBackoff time.Duration

	// ReconnectCtx is the context used during reconnection; if cancelled, reconnection aborts.
	ReconnectCtx context.Context
}

ReconnectConfig holds parameters for automatic reconnection.

type Request

type Request struct {
	Method  string          `json:"method"`
	Payload json.RawMessage `json:"payload"`
	Headers http.Header     `json:"headers,omitempty"`
}

Request defines the JSON request format sent over a stream.

type Response

type Response struct {
	Status  int         `json:"status"`
	Message string      `json:"message,omitempty"`
	Data    interface{} `json:"data,omitempty"`
}

Response defines the JSON response format.

type Router

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

Router holds a mapping of method names to handlers.

func NewRouter

func NewRouter() *Router

NewRouter creates and returns a new Router.

func (*Router) Handle

func (r *Router) Handle(method string, handler HandlerFunc)

Handle registers a new handler for the given method.

func (*Router) ServeStream

func (r *Router) ServeStream(stream *smux.Stream)

ServeStream reads one JSON-encoded Request from the given stream, dispatches it to the appropriate handler, and writes back a JSON response.

type Session

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

Session wraps an underlying smux.Session. It now also holds an optional reconnect configuration and a mutex for protecting the underlying session pointer.

func DialWithBackoff

func DialWithBackoff(
	ctx context.Context,
	dialFunc func() (net.Conn, error),
	upgradeFn func(conn net.Conn) (*Session, error),
	initial, max time.Duration,
) (*Session, error)

DialWithBackoff repeatedly attempts to establish a connection by calling dialFunc and then upgrade it using upgradeFn. It uses exponential backoff between attempts, starting with `initial` duration and capping at `max`. The process respects the provided ctx.

func HijackUpgradeHTTP

func HijackUpgradeHTTP(w http.ResponseWriter, r *http.Request, config *smux.Config) (*Session, error)

HijackUpgradeHTTP is a helper for server-side HTTP hijacking. It attempts to hijack the HTTP connection from the ResponseWriter, writes the 101 Switching Protocols handshake, and then creates and returns a new server-side Session using the underlying connection. The config parameter is passed to smux.Server (or nil for default config).

func NewClientSession

func NewClientSession(conn net.Conn, config *smux.Config) (*Session, error)

NewClientSession creates a new multiplexer session for the client side. The config parameter can be nil to use the default smux configuration.

func NewServerSession

func NewServerSession(conn net.Conn, config *smux.Config) (*Session, error)

NewServerSession creates a new multiplexer session for the server side. The config parameter can be nil to use the default smux configuration.

func UpgradeHTTPClient

func UpgradeHTTPClient(conn net.Conn, requestPath, host string, headers http.Header,
	config *smux.Config) (*Session, error)

UpgradeHTTPClient is a helper for client-side HTTP upgrade. Given an established connection, it writes an HTTP GET request to the specified requestPath and host, adding custom headers from the provided http.Header, along with the necessary Upgrade and Connection headers. It then reads and verifies the 101 Switching Protocols response and drains the remaining headers. Finally, it creates and returns a new client-side Session using the same connection. The config parameter is passed to smux.Client (or nil for default config).

func (*Session) Call

func (s *Session) Call(method string, payload interface{}) (*Response, error)

Call is a helper method for initiating a request/response conversation on a new stream. It marshals the provided payload, sends the request, and waits for the JSON response.

func (*Session) CallContext

func (s *Session) CallContext(ctx context.Context, method string, payload interface{}) (*Response, error)

CallContext is similar to Call but allows passing a context with a deadline or timeout. If the call does not complete before ctx is done, it aborts with the context's error. In case of an error that appears to be from a disconnected session, it will try to reconnect (if enabled) and then retry opening a stream.

func (*Session) CallJSON

func (s *Session) CallJSON(ctx context.Context, method string, payload interface{}, v interface{}) error

CallJSON performs the RPC call and decodes the JSON data into v. It is similar in spirit to http.Get followed by json.NewDecoder(resp.Body).Decode(&v).

func (*Session) CallWithHeaders

func (s *Session) CallWithHeaders(ctx context.Context, method string, payload interface{}, headers http.Header) (*Response, error)

CallWithHeaders is similar to CallContext but allows passing custom http.Header with the request. The headers will be embedded in the Request.Headers field.

func (*Session) Close

func (s *Session) Close() error

Close shuts down the underlying smux session.

func (*Session) EnableAutoReconnect

func (s *Session) EnableAutoReconnect(rc *ReconnectConfig)

EnableAutoReconnect enables automatic reconnection on this session. The supplied ReconnectConfig is used to reconnect if the underlying session disconnects.

func (*Session) Serve

func (s *Session) Serve(router *Router) error

Serve continuously accepts incoming streams on the session. Each incoming stream is dispatched to the provided router. If an error occurs (typically due to disconnection), and auto-reconnect is enabled, it will attempt to re-establish the underlying session and continue.