run

Documentation

Index

• Variables
• func DerefOrZero[T any](x *T) T
• func GatherPublishers[W AbstractEnvelopeWriter, PUB AbstractPublisher[W], B any, ...](ctx context.Context, chName ParamString, channelBindings *B, producers []PRD) ([]PUB, error)
• func GatherSubscribers[R AbstractEnvelopeReader, SUB AbstractSubscriber[R], B any, ...](ctx context.Context, chName ParamString, channelBindings *B, consumers []C) ([]SUB, error)
• func ToPtr[T any](x T) *T
• type AbstractConsumer
• type AbstractEnvelopeReader
• type AbstractEnvelopeWriter
• type AbstractProducer
• type AbstractPublisher
• type AbstractSubscriber
• type ErrorPool
  • func NewErrorPool() *ErrorPool
  • func (p *ErrorPool) Go(cb func() error)
  • func (p *ErrorPool) Wait() (err error)
• type FanOut
  • func NewFanOut[MessageT any]() *FanOut[MessageT]
  • func (cm *FanOut[MessageT]) Add(cb func(msg MessageT)) *list.Element
  • func (cm *FanOut[MessageT]) Put(newItem func() MessageT)
  • func (cm *FanOut[MessageT]) Remove(el *list.Element)
• type Headers
  • func (h Headers) ToByteValues() map[string][]byte
• type ParamString
  • func (c ParamString) Expand() (string, error)
  • func (c ParamString) String() string
• type Parameter
• type PublisherFanOut
  • func (p PublisherFanOut[W, P]) Close() (err error)
  • func (p PublisherFanOut[W, P]) Send(ctx context.Context, envelopes ...W) error
• type SubscriberFanIn
  • func (s SubscriberFanIn[R, S]) Close() (err error)
  • func (s SubscriberFanIn[R, S]) Receive(ctx context.Context, cb func(envelope R)) error

Variables

var ErrEmptyServers = errors.New("empty servers list")

Functions

func DerefOrZero

func DerefOrZero[T any](x *T) T

func GatherPublishers

func GatherPublishers[W AbstractEnvelopeWriter, PUB AbstractPublisher[W], B any, PRD AbstractProducer[B, W, PUB]](
	ctx context.Context,
	chName ParamString,
	channelBindings *B,
	producers []PRD,
) ([]PUB, error)

func GatherSubscribers

func GatherSubscribers[R AbstractEnvelopeReader, SUB AbstractSubscriber[R], B any, C AbstractConsumer[B, R, SUB]](
	ctx context.Context,
	chName ParamString,
	channelBindings *B,
	consumers []C,
) ([]SUB, error)

func ToPtr

func ToPtr[T any](x T) *T

Types

type AbstractConsumer

type AbstractConsumer[B any, R AbstractEnvelopeReader, S AbstractSubscriber[R]] interface {
	Subscriber(ctx context.Context, channelName string, bindings *B) (S, error)
}

type AbstractEnvelopeReader

type AbstractEnvelopeReader interface {
	io.Reader
	Headers() Headers
}

type AbstractEnvelopeWriter

type AbstractEnvelopeWriter interface {
	io.Writer
	ResetPayload()
	SetHeaders(headers Headers)
	// SetContentType is here because it may be set in message definition in AsyncAPI. Also, some protocols may have
	// content type property of data, AMQP, for instance.
	SetContentType(contentType string)
}

type AbstractProducer

type AbstractProducer[B any, W AbstractEnvelopeWriter, P AbstractPublisher[W]] interface {
	Publisher(ctx context.Context, channelName string, bindings *B) (P, error)
}

type AbstractPublisher

type AbstractPublisher[W AbstractEnvelopeWriter] interface {
	Send(ctx context.Context, envelopes ...W) error
	Close() error
}

type AbstractSubscriber

type AbstractSubscriber[R AbstractEnvelopeReader] interface {
	Receive(ctx context.Context, cb func(envelope R)) error
	Close() error
}

type ErrorPool

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

func NewErrorPool

func NewErrorPool() *ErrorPool

func (*ErrorPool) Go

func (p *ErrorPool) Go(cb func() error)

func (*ErrorPool) Wait

func (p *ErrorPool) Wait() (err error)

type FanOut

type FanOut[MessageT any] struct {
	// contains filtered or unexported fields
}

func NewFanOut

func NewFanOut[MessageT any]() *FanOut[MessageT]

func (*FanOut[MessageT]) Add

func (cm *FanOut[MessageT]) Add(cb func(msg MessageT)) *list.Element

func (*FanOut[MessageT]) Put

func (cm *FanOut[MessageT]) Put(newItem func() MessageT)

func (*FanOut[MessageT]) Remove

func (cm *FanOut[MessageT]) Remove(el *list.Element)

type Headers

type Headers map[string]any

func (Headers) ToByteValues

func (h Headers) ToByteValues() map[string][]byte

type ParamString

type ParamString struct {
	Expr       string
	Parameters map[string]string
}

func (ParamString) Expand

func (c ParamString) Expand() (string, error)

func (ParamString) String

func (c ParamString) String() string

type Parameter

type Parameter interface {
	Name() string
	String() string
}

type PublisherFanOut

type PublisherFanOut[W AbstractEnvelopeWriter, P AbstractPublisher[W]] struct {
	Publishers []P
}

func (PublisherFanOut[W, P]) Close

func (p PublisherFanOut[W, P]) Close() (err error)

func (PublisherFanOut[W, P]) Send

func (p PublisherFanOut[W, P]) Send(ctx context.Context, envelopes ...W) error

type SubscriberFanIn

type SubscriberFanIn[R AbstractEnvelopeReader, S AbstractSubscriber[R]] struct {
	Subscribers []S
}

func (SubscriberFanIn[R, S]) Close

func (s SubscriberFanIn[R, S]) Close() (err error)

func (SubscriberFanIn[R, S]) Receive

func (s SubscriberFanIn[R, S]) Receive(ctx context.Context, cb func(envelope R)) error