queue

Documentation

Index

• Variables
• type Config
• type Encoder
  • func (b *Encoder) Decode(data []byte, value interface{}, args ...interface{}) error
  • func (b *Encoder) Encode(value interface{}) (data []byte, err error)
• type Persistence
  • func (q *Persistence) Chan() <-chan *common.Event
  • func (q *Persistence) Close(clean bool) error
  • func (q *Persistence) Pop() (*common.Event, error)
  • func (q *Persistence) Push(e *common.Event) (err error)
• type Queue
  • func NewPersistence(cfg Config, bucket store.Bucket) Queue
  • func NewTemporary(id string, capacity int, dropIfFull bool) Queue
• type Temporary
  • func (q *Temporary) Chan() <-chan *common.Event
  • func (q *Temporary) Close(_ bool) error
  • func (q *Temporary) Pop() (*common.Event, error)
  • func (q *Temporary) Push(e *common.Event) error

Variables

var ErrQueueClosed = errors.New("queue is closed")

ErrQueueClosed queue is closed

Types

type Config

type Config struct {
	Name          string        `yaml:"name" json:"name"`
	BatchSize     int           `yaml:"batchSize" json:"batchSize" default:"10"`
	ExpireTime    time.Duration `yaml:"expireTime" json:"expireTime" default:"168h"`
	CleanInterval time.Duration `yaml:"cleanInterval" json:"cleanInterval" default:"1h"`
	WriteTimeout  time.Duration `yaml:"writeTimeout" json:"writeTimeout" default:"100ms"`
	DeleteTimeout time.Duration `yaml:"deleteTimeout" json:"deleteTimeout" default:"500ms"`
}

Config queue config

type Encoder

type Encoder struct{}

func (*Encoder) Decode

func (b *Encoder) Decode(data []byte, value interface{}, args ...interface{}) error

Decode decode the message from byte array

func (*Encoder) Encode

func (b *Encoder) Encode(value interface{}) (data []byte, err error)

Encode encodes the message to byte array

type Persistence

type Persistence struct {
	utils.Tomb
	// contains filtered or unexported fields
}

Persistence is a persistent queue

func (*Persistence) Chan

func (q *Persistence) Chan() <-chan *common.Event

Chan returns message channel

func (*Persistence) Close

func (q *Persistence) Close(clean bool) error

Close closes this queue and clean queue data when cleanSession is true

func (*Persistence) Pop

func (q *Persistence) Pop() (*common.Event, error)

Pop pops a message from queue

func (*Persistence) Push

func (q *Persistence) Push(e *common.Event) (err error)

Push pushes a message into queue

type Queue

type Queue interface {
	Push(*common.Event) error
	Pop() (*common.Event, error)
	Chan() <-chan *common.Event
	Close(bool) error
}

Queue interfaces

func NewPersistence

func NewPersistence(cfg Config, bucket store.Bucket) Queue

NewPersistence creates a new persistent queue

func NewTemporary

func NewTemporary(id string, capacity int, dropIfFull bool) Queue

NewTemporary creates a new temporary queue

type Temporary

type Temporary struct {
	sync.Once
	// contains filtered or unexported fields
}

Temporary is an temporary queue in memory

func (*Temporary) Chan

func (q *Temporary) Chan() <-chan *common.Event

Chan returns message channel

func (*Temporary) Close

func (q *Temporary) Close(_ bool) error

Close closes this queue

func (*Temporary) Pop

func (q *Temporary) Pop() (*common.Event, error)

Pop pops a message from queue

func (*Temporary) Push

func (q *Temporary) Push(e *common.Event) error

Push pushes a message to queue