server

Documentation

Overview

nolint

Index

• Variables
• func ReadResponse(r io.Reader, toCall func(resLen uint32) (int, error)) (int, error)
• func WriteErrorResponse(w io.Writer, respErr error) error
• type DecayingTagSpec
• type Handler
• type Host
• type Opt
  • func WithDecayingTag(tag DecayingTagSpec) Opt
  • func WithHardTimeout(timeout time.Duration) Opt
  • func WithLog(log log.Log) Opt
  • func WithMetrics() Opt
  • func WithQueueSize(size int) Opt
  • func WithRequestSizeLimit(limit int) Opt
  • func WithRequestsPerInterval(n int, interval time.Duration) Opt
  • func WithTimeout(timeout time.Duration) Opt
• type Response
  • func (t *Response) DecodeScale(dec *scale.Decoder) (total int, err error)
  • func (t *Response) EncodeScale(enc *scale.Encoder) (total int, err error)
• type Server
  • func New(h Host, proto string, handler StreamHandler, opts ...Opt) *Server
  • func (s *Server) NumAcceptedRequests() int
  • func (s *Server) Request(ctx context.Context, pid peer.ID, req []byte, extraProtocols ...string) ([]byte, error)
  • func (s *Server) Run(ctx context.Context) error
  • func (s *Server) StreamRequest(ctx context.Context, pid peer.ID, req []byte, callback StreamRequestCallback, ...) error
• type ServerError
  • func NewServerError(msg string) *ServerError
  • func (err *ServerError) Error() string
  • func (*ServerError) Is(target error) bool
• type StreamHandler
  • func WrapHandler(handler Handler) StreamHandler
• type StreamRequestCallback

Variables

var (
	// ErrNotConnected is returned when peer is not connected.
	ErrNotConnected = errors.New("peer is not connected")
	// ErrPeerResponseFailed raised if peer responded with an error.
	ErrPeerResponseFailed = errors.New("peer response failed")
)

Functions

func ReadResponse

func ReadResponse(r io.Reader, toCall func(resLen uint32) (int, error)) (int, error)

func WriteErrorResponse

func WriteErrorResponse(w io.Writer, respErr error) error

Types

type DecayingTagSpec

type DecayingTagSpec struct {
	Interval time.Duration `mapstructure:"interval"`
	Inc      int           `mapstructure:"inc"`
	Dec      int           `mapstructure:"dec"`
	Cap      int           `mapstructure:"cap"`
}

type Handler

type Handler func(context.Context, []byte) ([]byte, error)

Handler is a handler to be defined by the application.

type Host

type Host interface {
	SetStreamHandler(protocol.ID, network.StreamHandler)
	NewStream(context.Context, peer.ID, ...protocol.ID) (network.Stream, error)
	Network() network.Network
	ConnManager() connmgr.ConnManager
}

Host is a subset of libp2p Host interface that needs to be implemented to be usable with server.

type Opt

type Opt func(s *Server)

Opt is a type to configure a server.

func WithDecayingTag

func WithDecayingTag(tag DecayingTagSpec) Opt

func WithHardTimeout

func WithHardTimeout(timeout time.Duration) Opt

WithHardTimeout configures the hard timeout for requests. Requests are terminated if they take longer than the specified duration.

func WithLog

func WithLog(log log.Log) Opt

WithLog configures logger for the server.

func WithMetrics

func WithMetrics() Opt

WithMetrics will enable metrics collection in the server.

func WithQueueSize

func WithQueueSize(size int) Opt

WithQueueSize parametrize number of message that will be kept in queue and eventually processed by server. Otherwise stream is closed immediately.

Size of the queue should be set to account for maximum expected latency, such as if expected latency is 10s and server processes 1000 requests per second size should be 100.

Defaults to 100.

func WithRequestSizeLimit

func WithRequestSizeLimit(limit int) Opt

func WithRequestsPerInterval

func WithRequestsPerInterval(n int, interval time.Duration) Opt

WithRequestsPerInterval parametrizes server rate limit to limit maximum amount of bandwidth that this handler can consume.

Defaults to 100 requests per second.

func WithTimeout

func WithTimeout(timeout time.Duration) Opt

WithTimeout configures stream timeout. The requests are terminated when no data is received or sent for the specified duration.

type Response

type Response struct {
	// keep in line with limit of ResponseMessage.Data in `fetch/wire_types.go`
	Data  []byte `scale:"max=125829120"` // 120 MiB > 3.5 mio ATX * 32 bytes per ID
	Error string `scale:"max=1024"`      // TODO(mafa): make error code instead of string
}

Response is a server response.

func (*Response) DecodeScale

func (t *Response) DecodeScale(dec *scale.Decoder) (total int, err error)

func (*Response) EncodeScale

func (t *Response) EncodeScale(enc *scale.Encoder) (total int, err error)

type Server

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

Server for the Handler.

func New

func New(h Host, proto string, handler StreamHandler, opts ...Opt) *Server

New server for the handler.

func (*Server) NumAcceptedRequests

func (s *Server) NumAcceptedRequests() int

NumAcceptedRequests returns the number of accepted requests for this server. It is used for testing.

func (*Server) Request

func (s *Server) Request(ctx context.Context, pid peer.ID, req []byte, extraProtocols ...string) ([]byte, error)

Request sends a binary request to the peer.

func (*Server) Run

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

func (*Server) StreamRequest

func (s *Server) StreamRequest(
	ctx context.Context,
	pid peer.ID,
	req []byte,
	callback StreamRequestCallback,
	extraProtocols ...string,
) error

StreamRequest sends a binary request to the peer. The response is read from the stream by the specified callback.

type ServerError

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

ServerError is used by the client (Request/StreamRequest) to represent an error returned by the server.

func NewServerError

func NewServerError(msg string) *ServerError

func (*ServerError) Error

func (err *ServerError) Error() string

func (*ServerError) Is

func (*ServerError) Is(target error) bool

type StreamHandler

type StreamHandler func(context.Context, []byte, io.ReadWriter) error

StreamHandler is a handler that writes the response to the stream directly instead of buffering the serialized representation.

func WrapHandler

func WrapHandler(handler Handler) StreamHandler

type StreamRequestCallback

type StreamRequestCallback func(context.Context, io.ReadWriter) error

StreamRequestCallback is a function that executes a streamed request.