eventing

Documentation

Index

• Variables
• type Client
  • func NewRedisClient(ctx context.Context, logger logger.Logger, rdb *redis.Client) (Client, error)
• type Headers
  • func (h Headers) Get(key string) string
  • func (h Headers) Keys() []string
  • func (h Headers) Set(key string, value string)
• type Message
• type MessageCallback
• type MessageReplier
• type MessageSet
• type PublishOption
  • func WithHeader(key, value string) PublishOption
  • func WithTrim(trim int64) PublishOption
• type Subscriber

Variables

var ErrNotReplyable = errors.New("message is not replyable")

var ErrRequestTimeout = errors.New("request timed out")

var ReplyHeader = "reply-to"

var SubjectHeader = "subject"

Types

type Client

type Client interface {
	// Publish publishes a message to a subject
	Publish(ctx context.Context, subject string, data []byte, opts ...PublishOption) error
	// QueuePublish publishes a message to a subject in a consumer group named queue
	QueuePublish(ctx context.Context, subject string, data []byte, opts ...PublishOption) error
	// Request requests a message from a subject, and synchronously waits for a reply. All subscribers will receive the message and try to reply. You probably want to use QueueRequest instead.
	Request(ctx context.Context, subject string, data []byte, timeout time.Duration, opts ...PublishOption) (Message, error)
	// QueueRequest requests a message from a subject in a consumer, and synchronously waits for a reply
	QueueRequest(ctx context.Context, subject string, data []byte, timeout time.Duration, opts ...PublishOption) (Message, error)
	// Subscribe subscribes to a subject
	Subscribe(ctx context.Context, subject string, cb MessageCallback) (Subscriber, error)
	// QueueSubscribe subscribes to a subject in a consumer group named queue
	QueueSubscribe(ctx context.Context, subject, queue string, cb MessageCallback) (Subscriber, error)
	// QueueFetchMessages fetches messages from a subject in a consumer group named queue, the messages must be acknowledged by calling the Ack method on the MessageSet
	QueueFetchMessages(ctx context.Context, subject, queue string, count int64) (MessageSet, error)
	// Close closes the client
	Close() error
}

Client defines the interface for event clients

func NewRedisClient

func NewRedisClient(ctx context.Context, logger logger.Logger, rdb *redis.Client) (Client, error)

type Headers

type Headers map[string]string

Headers represents message headers that can be used for both map operations and propagation

func (Headers) Get

func (h Headers) Get(key string) string

func (Headers) Keys

func (h Headers) Keys() []string

func (Headers) Set

func (h Headers) Set(key string, value string)

type Message

type Message interface {
	Data() []byte
	Headers() Headers
	Reply(ctx context.Context, data []byte, opts ...PublishOption) error
	Subject() string
}

Message represents a message received from the event system

type MessageCallback

type MessageCallback func(ctx context.Context, msg Message)

type MessageReplier

type MessageReplier func(ctx context.Context, data []byte, opts ...PublishOption) error

type MessageSet

type MessageSet interface {
	Messages() []Message
	Ack(ctx context.Context) error
}

type PublishOption

type PublishOption func(*publishOptions)

func WithHeader

func WithHeader(key, value string) PublishOption

func WithTrim

func WithTrim(trim int64) PublishOption

sets the trim publish option, which is the number of old messages to trim from the stream after the message is published

type Subscriber

type Subscriber interface {
	// Close stops the subscriber
	Close() error
	// IsValid returns true if the subscriber is running, false if it has been closed, or is nil or is otherwise not in a running state
	IsValid() bool
	// CloseWithCallback closes the subscriber and calls the callback with the error if there is one
	CloseWithCallback(ctx context.Context, cb func(err error))
}