broker

Documentation

Index

• type Broker
  • func NewBroker(c *config.Config, pf *persist.Factory, report Report) (b *Broker, err error)
  • func (b *Broker) Close()
  • func (b *Broker) Config() *config.Config
  • func (b *Broker) FetchQ1(offset uint64, batchSize int) ([]*common.Message, error)
  • func (b *Broker) Flow(msg *common.Message)
  • func (b *Broker) InitOffset(id string, persistent bool) (uint64, error)
  • func (b *Broker) MsgQ0Chan() <-chan *common.Message
  • func (b *Broker) OffsetChanLen() int
  • func (b *Broker) OffsetPersisted(id string) (*uint64, error)
  • func (b *Broker) PersistOffset(id string, offset uint64) error
  • func (b *Broker) WaitOffsetPersisted()
• type Offset
• type Report

Types

type Broker

type Broker struct {
	// contains filtered or unexported fields
}

Broker a mqtt broker

func NewBroker

func NewBroker(c *config.Config, pf *persist.Factory, report Report) (b *Broker, err error)

NewBroker NewBroker

func (*Broker) Close

func (b *Broker) Close()

Close closes broker

func (*Broker) Config

func (b *Broker) Config() *config.Config

Config returns config

func (*Broker) FetchQ1

func (b *Broker) FetchQ1(offset uint64, batchSize int) ([]*common.Message, error)

FetchQ1 fetches messages with qos=1

func (*Broker) Flow

func (b *Broker) Flow(msg *common.Message)

Flow flows message to broker

func (*Broker) InitOffset

func (b *Broker) InitOffset(id string, persistent bool) (uint64, error)

InitOffset init offset

func (*Broker) MsgQ0Chan

func (b *Broker) MsgQ0Chan() <-chan *common.Message

MsgQ0Chan returns config

func (*Broker) OffsetChanLen

func (b *Broker) OffsetChanLen() int

OffsetChanLen returns the length of offset channel

func (*Broker) OffsetPersisted

func (b *Broker) OffsetPersisted(id string) (*uint64, error)

OffsetPersisted gets sink's offset from database

func (*Broker) PersistOffset

func (b *Broker) PersistOffset(id string, offset uint64) error

PersistOffset puts sink's offset to offset channal

func (*Broker) WaitOffsetPersisted

func (b *Broker) WaitOffsetPersisted()

WaitOffsetPersisted waits all offsets in channel to be persisted

type Offset

type Offset struct {
	// contains filtered or unexported fields
}

Offset sink's offset to persist

type Report

type Report func(map[string]interface{}) error

Report reports stats