broker

Documentation

Index

• Variables
• type Broker
• type BrokerOption
  • func WithBrokerAddresses(addrs ...string) BrokerOption
  • func WithBrokerContext(ctx context.Context) BrokerOption
  • func WithBrokerErrorHandler(handler Handler) BrokerOption
  • func WithBrokerTLSConfig(t *tls.Config) BrokerOption
  • func WithLogger(log logger.Logger) BrokerOption
• type BrokerOptions
• type EmptyRequestError
  • func (e *EmptyRequestError) Error() string
• type Event
• type Handler
• type Message
• type PublishOption
  • func WithPublishContext(ctx context.Context) PublishOption
• type PublishOptions
• type Response
  • func SuccessResponse[T any](data T) Response[T]
• type Result
• type SubscribeOption
  • func WithSubscribeAutoAck(autoAck bool) SubscribeOption
  • func WithSubscribeContext(ctx context.Context) SubscribeOption
  • func WithSubscribeGroup(gr string) SubscribeOption
• type SubscribeOptions
• type Subscriber

Variables

var (
	DefaultSuccessResponse = Response[interface{}]{
		Result: Result{
			Code:    "0",
			Message: "Success",
		},
		Data: nil,
	}

	DefaultErrorResponse = Response[interface{}]{
		Result: Result{
			Code:    "1",
			Message: "Internal Server Error",
		},
	}
)

Types

type Broker

type Broker interface {
	Init(...BrokerOption) error
	Options() BrokerOptions
	Address() string
	Connect() error
	Disconnect() error
	Publish(topic string, m *Message, opts ...PublishOption) error
	Subscribe(topic string, h Handler, opts ...SubscribeOption) (Subscriber, error)
	String() string
}

Broker is an interface used for asynchronous messaging.

type BrokerOption

type BrokerOption func(*BrokerOptions)

func WithBrokerAddresses

func WithBrokerAddresses(addrs ...string) BrokerOption

func WithBrokerContext

func WithBrokerContext(ctx context.Context) BrokerOption

func WithBrokerErrorHandler

func WithBrokerErrorHandler(handler Handler) BrokerOption

func WithBrokerTLSConfig

func WithBrokerTLSConfig(t *tls.Config) BrokerOption

func WithLogger

func WithLogger(log logger.Logger) BrokerOption

type BrokerOptions

type BrokerOptions struct {
	Context context.Context

	// underlying logger
	Logger logger.Logger

	// Handler executed when error happens in broker mesage
	// processing
	ErrorHandler Handler

	Addrs []string

	TLSConfig *tls.Config
}

type EmptyRequestError

type EmptyRequestError struct{}

func (*EmptyRequestError) Error

func (e *EmptyRequestError) Error() string

type Event

type Event interface {
	Topic() string
	Message() *Message
	Ack() error
	Error() error
}

Event is given to a subscription handler for processing.

type Handler

type Handler func(Event) error

Handler is used to process messages via a subscription of a topic. The handler is passed a publication interface which contains the message and optional Ack method to acknowledge receipt of the message.

type Message

type Message struct {
	Headers map[string]string
	Body    []byte
}

Message is a message send/received from the broker.

type PublishOption

type PublishOption func(*PublishOptions)

func WithPublishContext

func WithPublishContext(ctx context.Context) PublishOption

type PublishOptions

type PublishOptions struct {
	Context context.Context
}

type Response

type Response[T any] struct {
	Result Result `json:"result"` // Result
	Data   T      `json:"data"`   // Data
}

func SuccessResponse

func SuccessResponse[T any](data T) Response[T]

type Result

type Result struct {
	Status  int         `json:"-"`       // Http status code
	Code    string      `json:"code"`    // Error code
	Message string      `json:"message"` // Message
	Details interface{} `json:"details"` //
}

type SubscribeOption

type SubscribeOption func(*SubscribeOptions)

func WithSubscribeAutoAck

func WithSubscribeAutoAck(autoAck bool) SubscribeOption

func WithSubscribeContext

func WithSubscribeContext(ctx context.Context) SubscribeOption

func WithSubscribeGroup

func WithSubscribeGroup(gr string) SubscribeOption

type SubscribeOptions

type SubscribeOptions struct {
	Context context.Context

	Group string

	// AutoAck defaults to true. When a handler returns
	// with a nil error the message is acked.
	AutoAck bool
}

type Subscriber

type Subscriber interface {
	Options() SubscribeOptions
	Topic() string
	Unsubscribe() error
}

Subscriber is a convenience return type for the Subscribe method.