mbus

Documentation

Index

• Constants
• func Register(vendor string, service Service)
• func RegisterNotifier(vendor string, notifier Notifier)
• type Acknowledgement
  • func NewAcknowledgement() *Acknowledgement
  • func (c *Acknowledgement) Ack() error
  • func (c *Acknowledgement) IsAck() bool
  • func (c *Acknowledgement) IsNack() bool
  • func (c *Acknowledgement) Nack() error
• type Confirmation
  • func (c *Confirmation) String() string
• type EncodedResource
  • func (e EncodedResource) Decode() (*Resource, error)
• type Message
  • func (m *Message) AddAttribute(name string, value interface{})
  • func (m *Message) Payload() ([]byte, error)
• type Messenger
• type Notifier
  • func LookupNotifier(vendor string) Notifier
• type NotifierOption
  • func WithMaxMessages(value int) NotifierOption
  • func WithResource(value *Resource) NotifierOption
• type NotifierOptions
  • func NewNotifierOptions(options ...NotifierOption) *NotifierOptions
• type Resource
  • func (r *Resource) Init() error
• type Service
  • func Lookup(vendor string) Service

Constants

const (
	ResourceTypeTopic        = "topic"
	ResourceTypeSubscription = "subscription"
	ResourceTypeQueue        = "queue"
)

Functions

func Register

func Register(vendor string, service Service)

Register register vendor service

func RegisterNotifier

func RegisterNotifier(vendor string, notifier Notifier)

RegisterNotifier register vendor notifier

Types

type Acknowledgement

type Acknowledgement struct {
	Error error
	// contains filtered or unexported fields
}

Acknowledgement represents message acknowledgement

func NewAcknowledgement

func NewAcknowledgement() *Acknowledgement

func (*Acknowledgement) Ack

func (c *Acknowledgement) Ack() error

func (*Acknowledgement) IsAck

func (c *Acknowledgement) IsAck() bool

func (*Acknowledgement) IsNack

func (c *Acknowledgement) IsNack() bool

func (*Acknowledgement) Nack

func (c *Acknowledgement) Nack() error

type Confirmation

type Confirmation struct {
	MessageID string
}

Confirmation represents message confirmation

func (*Confirmation) String

func (c *Confirmation) String() string

type EncodedResource

type EncodedResource string

EncodedResource represents encoded resource

func (EncodedResource) Decode

func (e EncodedResource) Decode() (*Resource, error)

type Message

type Message struct {
	ID         string
	Resource   *Resource
	TraceID    string
	Attributes map[string]interface{}
	Subject    string
	Data       interface{}
}

func (*Message) AddAttribute

func (m *Message) AddAttribute(name string, value interface{})

func (*Message) Payload

func (m *Message) Payload() ([]byte, error)

type Messenger

type Messenger interface {
	//OnMessage handles message
	OnMessage(ctx context.Context, message *Message, ack *Acknowledgement) error
}

Messenger represents message listener

type Notifier

type Notifier interface {
	Notify(ctx context.Context, messenger Messenger, options ...NotifierOption) error
}

Notifier represents message Notifier

func LookupNotifier

func LookupNotifier(vendor string) Notifier

LookupNotifier lookup notifier

type NotifierOption

type NotifierOption func(Notifier *NotifierOptions)

func WithMaxMessages

func WithMaxMessages(value int) NotifierOption

WithMaxMessages sets max messages to process

func WithResource

func WithResource(value *Resource) NotifierOption

WithResource sets resource

type NotifierOptions

type NotifierOptions struct {
	//MaxPending max messages to process
	MaxPending int
	Resource   *Resource
}

func NewNotifierOptions

func NewNotifierOptions(options ...NotifierOption) *NotifierOptions

type Resource

type Resource struct {
	ID          string        `yaml:"ID" json:",omitempty"`
	Name        string        `yaml:"Name"  json:",omitempty"`
	Region      string        `yaml:"Region"  json:",omitempty"`
	Vendor      string        `yaml:"Vendor"  json:",omitempty"`
	URL         string        `yaml:"URL"  json:",omitempty"`
	Credentials *scy.Resource `yaml:"Resource"  json:",omitempty"`
	Type        string        `description:"resource type: topic, subscription" yaml:"Type"  json:",omitempty"`
	Client      interface{}   `description:"client" yaml:"-"`
	sync.Mutex  `yaml:"-" json:"-"`
}

func (*Resource) Init

func (r *Resource) Init() error

type Service

type Service interface {
	Push(ctx context.Context, dest *Resource, message *Message) (*Confirmation, error)
}

func Lookup

func Lookup(vendor string) Service

Lookup lookup service