queue

Documentation

Index

• Variables
• type BalancedPusher
  • func (pusher *BalancedPusher) Name() string
  • func (pusher *BalancedPusher) Push(message string) error
• type Consumer
• type ConsumerFactory
• type Listener
• type MessageQueue
• type MultiPusher
  • func (pusher *MultiPusher) Name() string
  • func (pusher *MultiPusher) Push(message string) error
• type Poller
• type ProduceListener
• type Producer
• type ProducerFactory
• type Pusher
  • func NewBalancedPusher(pushers []Pusher) Pusher
  • func NewMultiPusher(pushers []Pusher) Pusher
• type Queue
  • func NewQueue(producerFactory ProducerFactory, consumerFactory ConsumerFactory) *Queue
  • func (q *Queue) AddListener(listener Listener)
  • func (q *Queue) Broadcast(message any)
  • func (q *Queue) SetName(name string)
  • func (q *Queue) SetNumConsumer(count int)
  • func (q *Queue) SetNumProducer(count int)
  • func (q *Queue) Start()
  • func (q *Queue) Stop()

Variables

var ErrNoAvailablePusher = errors.New("no available pusher")

ErrNoAvailablePusher indicates no pusher available.

Types

type BalancedPusher

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

A BalancedPusher is used to push messages to multiple pusher with round robin algorithm.

func (*BalancedPusher) Name

func (pusher *BalancedPusher) Name() string

Name returns the name of pusher.

func (*BalancedPusher) Push

func (pusher *BalancedPusher) Push(message string) error

Push pushes message to one of the underlying pushers.

type Consumer

type Consumer interface {
	Consume(string) error
	OnEvent(event any)
}

A Consumer interface represents a consumer that can consume string messages.

type ConsumerFactory

type ConsumerFactory func() (Consumer, error)

ConsumerFactory defines the factory to generate consumers.

type Listener

type Listener interface {
	OnPause()
	OnResume()
}

A Listener interface represents a listener that can be notified with queue events.

type MessageQueue

type MessageQueue interface {
	Start()
	Stop()
}

A MessageQueue interface represents a message queue.

type MultiPusher

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

A MultiPusher is a pusher that can push messages to multiple underlying pushers.

func (*MultiPusher) Name

func (pusher *MultiPusher) Name() string

Name returns the name of pusher.

func (*MultiPusher) Push

func (pusher *MultiPusher) Push(message string) error

Push pushes a message into the underlying queue.

type Poller

type Poller interface {
	Name() string
	Poll() string
}

A Poller interface wraps the method Poll.

type ProduceListener

type ProduceListener interface {
	OnProducerPause()
	OnProducerResume()
}

A ProduceListener interface represents a produce listener.

type Producer

type Producer interface {
	AddListener(listener ProduceListener)
	Produce() (string, bool)
}

A Producer interface represents a producer that produces messages.

type ProducerFactory

type ProducerFactory func() (Producer, error)

ProducerFactory defines the method to generate a Producer.

type Pusher

type Pusher interface {
	Name() string
	Push(string) error
}

A Pusher interface wraps the method Push.

func NewBalancedPusher

func NewBalancedPusher(pushers []Pusher) Pusher

NewBalancedPusher returns a BalancedPusher.

func NewMultiPusher

func NewMultiPusher(pushers []Pusher) Pusher

NewMultiPusher returns a MultiPusher.

type Queue

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

A Queue is a message queue.

func NewQueue

func NewQueue(producerFactory ProducerFactory, consumerFactory ConsumerFactory) *Queue

NewQueue returns a Queue.

func (*Queue) AddListener

func (q *Queue) AddListener(listener Listener)

AddListener adds a listener to q.

func (*Queue) Broadcast

func (q *Queue) Broadcast(message any)

Broadcast broadcasts message to all event channels.

func (*Queue) SetName

func (q *Queue) SetName(name string)

SetName sets the name of q.

func (*Queue) SetNumConsumer

func (q *Queue) SetNumConsumer(count int)

SetNumConsumer sets the number of consumers.

func (*Queue) SetNumProducer

func (q *Queue) SetNumProducer(count int)

SetNumProducer sets the number of producers.

func (*Queue) Start

func (q *Queue) Start()

Start starts q.

func (*Queue) Stop

func (q *Queue) Stop()

Stop stops q.