stream

Documentation

Index

• Constants
• Variables
• func Middleware(next peanats.Handler) peanats.Handler
• type Client
  • func NewClient(conn peanats.Connection, opts ...ClientOption) Client
• type ClientOption
  • func WithReplySubjecter(rs ReplySubjecter) ClientOption
• type Receiver
• type ReplySubjecter
  • func ReplySubjectCryptoRand() ReplySubjecter
  • func ReplySubjectInbox() ReplySubjecter
  • func ReplySubjectNUID() ReplySubjecter
  • func ReplySubjectRand() ReplySubjecter
• type ReplySubjecterFunc
  • func (f ReplySubjecterFunc) ReplySubject() string
• type TypedClient
  • func NewTypedClient[A, R any](nc peanats.Connection, opts ...TypedClientOption[A, R]) TypedClient[A, R]
• type TypedClientOption
  • func WithCodec[A, R any](codec peanats.Codec) TypedClientOption[A, R]
  • func WithTypedReplySubjecter[A, R any](rs ReplySubjecter) TypedClientOption[A, R]
• type TypedReceiver

Constants

const (
	HeaderUID      = "Stream-UID"
	HeaderSequence = "Stream-Sequence"
	HeaderControl  = "Stream-Control"

	HeaderControlAck     = "ack"
	HeaderControlProceed = "proceed"
	HeaderControlDone    = "done"
)

Variables

var (
	ErrClient            = errors.New("stream client error")
	ErrProtocolViolation = fmt.Errorf("%w: %s", ErrClient, "protocol violation")
)

Functions

func Middleware

func Middleware(next peanats.Handler) peanats.Handler

Middleware is a middleware that forms handler publication into a stream of messages.

1. As a first step, it sends an message to the original reply subject containing the subject of the stream. 2. All subsequent messages published from inner handler are sent to the stream subject. 3. When the handler returns, middleware sends an empty message to the stream subject marking the end of the stream.

Types

type Client

type Client interface {
	Start(ctx context.Context, subj string, data []byte) (Receiver, error)
}

func NewClient

func NewClient(conn peanats.Connection, opts ...ClientOption) Client

type ClientOption

type ClientOption = func(*clientImpl) *clientImpl

func WithReplySubjecter

func WithReplySubjecter(rs ReplySubjecter) ClientOption

type Receiver

type Receiver interface {
	UID() string
	Sequence() int
	Receive(context.Context) (*nats.Msg, error)
	ReceiveAll(context.Context) ([]*nats.Msg, error)
}

type ReplySubjecter

type ReplySubjecter interface {
	ReplySubject() string
}

func ReplySubjectCryptoRand

func ReplySubjectCryptoRand() ReplySubjecter

ReplySubjectCryptoRand returns reply subject generator based on cryptographic-grade randomness

func ReplySubjectInbox

func ReplySubjectInbox() ReplySubjecter

ReplySubjectInbox returns reply subject generator based on nats.NewInbox

func ReplySubjectNUID

func ReplySubjectNUID() ReplySubjecter

ReplySubjectNUID returns fast stream reply subject generator

func ReplySubjectRand

func ReplySubjectRand() ReplySubjecter

ReplySubjectRand returns reply subject generator based on pseudo-randomness

type ReplySubjecterFunc

type ReplySubjecterFunc func() string

func (ReplySubjecterFunc) ReplySubject

func (f ReplySubjecterFunc) ReplySubject() string

type TypedClient

type TypedClient[A any, R any] interface {
	Start(ctx context.Context, subj string, arg *A) (TypedReceiver[R], error)
}

func NewTypedClient

func NewTypedClient[A, R any](nc peanats.Connection, opts ...TypedClientOption[A, R]) TypedClient[A, R]

type TypedClientOption

type TypedClientOption[A, R any] func(*typedClientImpl[A, R]) *typedClientImpl[A, R]

func WithCodec

func WithCodec[A, R any](codec peanats.Codec) TypedClientOption[A, R]

func WithTypedReplySubjecter

func WithTypedReplySubjecter[A, R any](rs ReplySubjecter) TypedClientOption[A, R]

type TypedReceiver

type TypedReceiver[T any] interface {
	UID() string
	Sequence() int
	Receive(context.Context) (*T, error)
	ReceiveAll(context.Context) ([]*T, error)
}