broker

Documentation

Index

• Variables
• type Broker
• type CallBack
• type Event
• type Message
  • func (m *Message) Marshal() string
  • func (m *Message) String() string
  • func (m *Message) Unmarshal(bs []byte)
• type PublishOption
  • func SetPubContext(cxt context.Context) PublishOption
  • func SetPubExchangeAD(ad bool) PublishOption
  • func SetPubExchangeDuration(duration bool) PublishOption
  • func SetPubExchangeName(en string) PublishOption
  • func SetPubExchangeType(et string) PublishOption
  • func SetPubPart(p int) PublishOption
  • func SetPubQOS(qos int) PublishOption
  • func SetPubReplica(r int) PublishOption
  • func SetPubRetained(r bool) PublishOption
• type PublishOptions
  • func (p *PublishOptions) Marshal() string
  • func (p *PublishOptions) String() string
• type SubscribeOption
  • func SetSubAutoAck(autoAck bool) SubscribeOption
  • func SetSubAutoDel(autoDel bool) SubscribeOption
  • func SetSubContext(cxt context.Context) SubscribeOption
  • func SetSubDuration(duration bool) SubscribeOption
  • func SetSubExchangeAD(ad bool) SubscribeOption
  • func SetSubExchangeDuration(duration bool) SubscribeOption
  • func SetSubExchangeName(en string) SubscribeOption
  • func SetSubExchangeType(et string) SubscribeOption
  • func SetSubPart(p int) SubscribeOption
  • func SetSubQOS(qos int) SubscribeOption
  • func SetSubQueue(queue string) SubscribeOption
  • func SetSubReplica(r int) SubscribeOption
• type SubscribeOptions
  • func (s *SubscribeOptions) Marshal() string
  • func (s *SubscribeOptions) String() string
• type Subscriber

Variables

var (
	ErrPublish = errors.New("publish err")
	// ErrSubscribe         = errors.New("subscribe err")
	// ErrConnectParam = errors.New("connect param err")
	ErrConnect           = errors.New("connect err")
	ErrConnectionIsNotOK = errors.New("connection is not ok")
	ErrQOS               = errors.New("qos err")
)

Types

type Broker

type Broker interface {
	fmt.Stringer
	Publish(topic string, msg *Message, opt ...PublishOption) error
	Subscribe(topics []string, callBack CallBack, opt ...SubscribeOption) (Subscriber, error)
	Connect() error
	Disconnect() error
}

Broker

type CallBack

type CallBack func(event Event) error

CallBack

type Event

type Event interface {
	Topic() string
	Ack() error
	Message() *Message
	Context() context.Context
}

Event

type Message

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

Message Publish or Subscribe Message

func (*Message) Marshal

func (m *Message) Marshal() string

func (*Message) String

func (m *Message) String() string

func (*Message) Unmarshal

func (m *Message) Unmarshal(bs []byte)

type PublishOption

type PublishOption func(*PublishOptions)

func SetPubContext

func SetPubContext(cxt context.Context) PublishOption

SetPubContext

func SetPubExchangeAD

func SetPubExchangeAD(ad bool) PublishOption

SetPubExchangeAD

func SetPubExchangeDuration

func SetPubExchangeDuration(duration bool) PublishOption

SetPubExchangeDuration

func SetPubExchangeName

func SetPubExchangeName(en string) PublishOption

SetPubExchangeName

func SetPubExchangeType

func SetPubExchangeType(et string) PublishOption

SetPubExchangeType

func SetPubPart

func SetPubPart(p int) PublishOption

SetPubPart

func SetPubQOS

func SetPubQOS(qos int) PublishOption

Set PublishOption SetPubQOS

func SetPubReplica

func SetPubReplica(r int) PublishOption

SetPubReplica

func SetPubRetained

func SetPubRetained(r bool) PublishOption

Set PublishOption SetPubRetained

type PublishOptions

type PublishOptions struct {
	ExchangeName     string
	ExchangeType     string
	ExchangeAD       bool // exchange auto delete
	ExchangeDuration bool // exchange duration
	Context          context.Context
	QOS              int
	Retained         bool
	// kafka
	Part    int
	Replica int
}

func (*PublishOptions) Marshal

func (p *PublishOptions) Marshal() string

func (*PublishOptions) String

func (p *PublishOptions) String() string

type SubscribeOption

type SubscribeOption func(*SubscribeOptions)

func SetSubAutoAck

func SetSubAutoAck(autoAck bool) SubscribeOption

SetSubAutoAck

func SetSubAutoDel

func SetSubAutoDel(autoDel bool) SubscribeOption

SetSubAutoDel

func SetSubContext

func SetSubContext(cxt context.Context) SubscribeOption

SetSubContext

func SetSubDuration

func SetSubDuration(duration bool) SubscribeOption

SetSubDuration

func SetSubExchangeAD

func SetSubExchangeAD(ad bool) SubscribeOption

SetSubExchangeAD

func SetSubExchangeDuration

func SetSubExchangeDuration(duration bool) SubscribeOption

SetSubExchangeDuration

func SetSubExchangeName

func SetSubExchangeName(en string) SubscribeOption

SetSubExchangeName

func SetSubExchangeType

func SetSubExchangeType(et string) SubscribeOption

SetSubExchangeType

func SetSubPart

func SetSubPart(p int) SubscribeOption

Set SubscribeOption SetSubPart

func SetSubQOS

func SetSubQOS(qos int) SubscribeOption

SetSubQOS

func SetSubQueue

func SetSubQueue(queue string) SubscribeOption

SetSubQueue

func SetSubReplica

func SetSubReplica(r int) SubscribeOption

SetSubReplica

type SubscribeOptions

type SubscribeOptions struct {
	AutoAck          bool
	AutoDel          bool
	Duration         bool
	Queue            string
	ExchangeName     string
	ExchangeType     string
	ExchangeAD       bool // exchange auto delete
	ExchangeDuration bool // exchange duration
	QOS              int
	Context          context.Context
	// kafka
	Part    int
	Replica int
}

func (*SubscribeOptions) Marshal

func (s *SubscribeOptions) Marshal() string

func (*SubscribeOptions) String

func (s *SubscribeOptions) String() string

type Subscriber

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

Subscriber