broker

Documentation

Overview

Package broker defines standard interface for a message broker.

Index

• type Broker
• type Encoder
• type Event
• type Handler
• type JSONEncoder
  • func (e JSONEncoder) Decode(b []byte, v *Message) error
  • func (e JSONEncoder) Encode(v *Message) ([]byte, error)
• type Message
  • func (*Message) Descriptor() ([]byte, []int)
  • func (m *Message) GetBody() []byte
  • func (m *Message) GetHeader() map[string]string
  • func (*Message) ProtoMessage()
  • func (m *Message) Reset()
  • func (m *Message) String() string
  • func (m *Message) XXX_DiscardUnknown()
  • func (m *Message) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)
  • func (m *Message) XXX_Merge(src proto.Message)
  • func (m *Message) XXX_Size() int
  • func (m *Message) XXX_Unmarshal(b []byte) error
• type ProtoEncoder
  • func (e ProtoEncoder) Decode(b []byte, v *Message) error
  • func (e ProtoEncoder) Encode(v *Message) ([]byte, error)
• type PublishOption
• type PublishOptions
• type SubscribeOption
  • func DisableAutoAck() SubscribeOption
  • func Queue(name string) SubscribeOption
• type SubscribeOptions
  • func (op *SubscribeOptions) Apply(opts ...SubscribeOption)
• type Subscriber

Types

type Broker

type Broker interface {
	Publish(topic string, m *Message, opts ...PublishOption) error
	Subscribe(topic string, h Handler, opts ...SubscribeOption) (Subscriber, error)
	HealthCheck() health.CheckFunc
}

Broker is an interface used for asynchronous messaging.

type Encoder

type Encoder interface {
	Encode(v *Message) ([]byte, error)
	Decode(b []byte, v *Message) error
}

Encoder is an interface for encoding/decoding used by broker.

type Event

type Event interface {
	Topic() string
	Message() *Message
	Ack() 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 JSONEncoder

type JSONEncoder struct{}

JSONEncoder JSON format encoder.

func (JSONEncoder) Decode

func (e JSONEncoder) Decode(b []byte, v *Message) error

Decode implements Encoder interface.

func (JSONEncoder) Encode

func (e JSONEncoder) Encode(v *Message) ([]byte, error)

Encode implements Encoder interface.

type Message

type Message struct {
	Header               map[string]string `` /* 153-byte string literal not displayed */
	Body                 []byte            `protobuf:"bytes,2,opt,name=body,proto3" json:"body,omitempty"`
	XXX_NoUnkeyedLiteral struct{}          `json:"-"`
	XXX_unrecognized     []byte            `json:"-"`
	XXX_sizecache        int32             `json:"-"`
}

func (*Message) Descriptor

func (*Message) Descriptor() ([]byte, []int)

func (*Message) GetBody

func (m *Message) GetBody() []byte

func (*Message) GetHeader

func (m *Message) GetHeader() map[string]string

func (*Message) ProtoMessage

func (*Message) ProtoMessage()

func (*Message) Reset

func (m *Message) Reset()

func (*Message) String

func (m *Message) String() string

func (*Message) XXX_DiscardUnknown

func (m *Message) XXX_DiscardUnknown()

func (*Message) XXX_Marshal

func (m *Message) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)

func (*Message) XXX_Merge

func (m *Message) XXX_Merge(src proto.Message)

func (*Message) XXX_Size

func (m *Message) XXX_Size() int

func (*Message) XXX_Unmarshal

func (m *Message) XXX_Unmarshal(b []byte) error

type ProtoEncoder

type ProtoEncoder struct{}

ProtoEncoder proto buffer encoder.

func (ProtoEncoder) Decode

func (e ProtoEncoder) Decode(b []byte, v *Message) error

Decode implements Encoder interface.

func (ProtoEncoder) Encode

func (e ProtoEncoder) Encode(v *Message) ([]byte, error)

Encode implements Encoder interface.

type PublishOption

type PublishOption func(*PublishOptions)

PublishOption is a func for config publish options.

type PublishOptions

type PublishOptions struct {
}

PublishOptions is a configuration holder for publish options.

type SubscribeOption

type SubscribeOption func(*SubscribeOptions)

SubscribeOption is a func for config subscription.

func DisableAutoAck

func DisableAutoAck() SubscribeOption

DisableAutoAck will disable auto ack of messages after they have been handled.

func Queue

func Queue(name string) SubscribeOption

Queue sets the name of the queue to share messages on

type SubscribeOptions

type SubscribeOptions struct {
	// AutoAck defaults to true. When a handler returns
	// with a nil error the message is acked.
	AutoAck bool
	// Subscribers with the same queue name
	// will create a shared subscription where each
	// receives a subset of messages.
	Queue string
}

SubscribeOptions is a configuration holder for subscriptions.

func (*SubscribeOptions) Apply

func (op *SubscribeOptions) Apply(opts ...SubscribeOption)

Apply apply the options.

type Subscriber

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

Subscriber is a convenience return type for the Subscribe method