requester

Documentation

Index

• Constants
• Variables
• type Proceeder
• type RequestOption
  • func RequestContentEncoding(e codec.ContentEncoding) RequestOption
  • func RequestContentType(c codec.ContentType) RequestOption
  • func RequestHeader(header peanats.Header) RequestOption
• type Requester
  • func New[RQ, RS any](nc transport.Conn) Requester[RQ, RS]
• type Response
• type ResponseReceiver
• type ResponseReceiverOption
  • func ResponseReceiverBuffer(size uint) ResponseReceiverOption
  • func ResponseReceiverProceeder(p Proceeder) ResponseReceiverOption
  • func ResponseReceiverRequestOptions(opts ...RequestOption) ResponseReceiverOption
  • func ResponseReceiverSkipper(s Skipper) ResponseReceiverOption
• type Skipper

Constants

const DefaultBuffer = 0

Variables

var (
	// DefaultSkipper skips empty messages.
	DefaultSkipper Skipper = &skipperImpl{}

	// DefaultProceeder proceeds only on non-empty messages.
	DefaultProceeder Proceeder = &proceederImpl{}
)

var (
	ErrSkip = errors.New("message skipped by the skipper")
	ErrOver = fmt.Errorf("%w: sequence is over", io.EOF)
)

Types

type Proceeder

type Proceeder interface {
	Proceed(context.Context, peanats.Msg) (bool, error)
}

Proceeder makes decision whether or not proceed with the response sequence. Decision is made late, after the message handler is invoked or skipped.

type RequestOption

type RequestOption func(*requestParams)

func RequestContentEncoding

func RequestContentEncoding(e codec.ContentEncoding) RequestOption

RequestContentEncoding sets the compression algorithm for the request message.

func RequestContentType

func RequestContentType(c codec.ContentType) RequestOption

func RequestHeader

func RequestHeader(header peanats.Header) RequestOption

RequestHeader merges the provided headers with existing headers for the message. If the same header key exists, the new values are appended to the existing ones.

type Requester

type Requester[RQ, RS any] interface {
	Request(context.Context, string, *RQ, ...RequestOption) (Response[RS], error)
	ResponseReceiver(context.Context, string, *RQ, ...ResponseReceiverOption) (ResponseReceiver[RS], error)
}

func New

func New[RQ, RS any](nc transport.Conn) Requester[RQ, RS]

type Response

type Response[T any] interface {
	Header() peanats.Header
	Value() *T
}

type ResponseReceiver

type ResponseReceiver[T any] interface {
	Next(context.Context) (Response[T], error)
	Stop() error
}

type ResponseReceiverOption

type ResponseReceiverOption func(*responseReceiverParams)

func ResponseReceiverBuffer

func ResponseReceiverBuffer(size uint) ResponseReceiverOption

ResponseReceiverBuffer sets the buffer size for the response receiver.

func ResponseReceiverProceeder

func ResponseReceiverProceeder(p Proceeder) ResponseReceiverOption

ResponseReceiverProceeder sets the proceeder for the response sequence.

func ResponseReceiverRequestOptions

func ResponseReceiverRequestOptions(opts ...RequestOption) ResponseReceiverOption

ResponseReceiverRequestOptions appends the set of request options for the request produced by the response receiver.

func ResponseReceiverSkipper

func ResponseReceiverSkipper(s Skipper) ResponseReceiverOption

ResponseReceiverSkipper sets the skipper for the response sequence.

type Skipper

type Skipper interface {
	Skip(context.Context, peanats.Msg) (bool, error)
}

Skipper makes decision whether or not skip the message handler without interrupting the response sequence.