queue

Documentation

Index

• type Handler
• type Memory
  • func NewMemory() (*Memory, error)
  • func (m *Memory) Close() error
  • func (m *Memory) Publish(topic string, body []byte) error
  • func (m *Memory) Subscribe(topic, channelName string, handler Handler, concurrency int) error
• type MemoryMessage
  • func (m *MemoryMessage) Finish() error
  • func (m *MemoryMessage) GetBody() []byte
  • func (m *MemoryMessage) GetID() []byte
  • func (m *MemoryMessage) GetTimestamp() time.Time
  • func (m *MemoryMessage) Requeue(delay time.Duration) error
  • func (m *MemoryMessage) Touch() error
• type Message
• type NSQ
  • func NewNSQ(cfg config.NSQConfig) (*NSQ, error)
  • func (n *NSQ) Close() error
  • func (n *NSQ) Publish(topic string, body []byte) error
  • func (n *NSQ) Subscribe(topic, channel string, handler Handler, concurrency int) error
• type NSQMessage
  • func (n *NSQMessage) Finish() error
  • func (n *NSQMessage) GetBody() []byte
  • func (n *NSQMessage) GetID() []byte
  • func (n *NSQMessage) GetTimestamp() time.Time
  • func (n *NSQMessage) Requeue(delay time.Duration) error
  • func (n *NSQMessage) Touch() error
• type Queue

Types

type Handler

type Handler func(msg Message) error

Handler is a function used for message handling.

type Memory

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

Memory is an implementation of the Queue that is purely in-memory

func NewMemory

func NewMemory() (*Memory, error)

NewMemory creates a new Memory queue

func (*Memory) Close

func (m *Memory) Close() error

Close closes all queue handlers.

func (*Memory) Publish

func (m *Memory) Publish(topic string, body []byte) error

Publish sends a message to the specified topic.

func (*Memory) Subscribe

func (m *Memory) Subscribe(topic, channelName string, handler Handler, concurrency int) error

Subscribe handles messages by binding to a specific topic and channel.

type MemoryMessage

type MemoryMessage struct {
	Body      []byte
	ID        []byte
	Timestamp time.Time

	Topic   string
	Channel string

	Requeued bool
	Delay    time.Duration
}

MemoryMessage is an impl of Message

func (*MemoryMessage) Finish

func (m *MemoryMessage) Finish() error

Finish is a nop

func (*MemoryMessage) GetBody

func (m *MemoryMessage) GetBody() []byte

GetBody returns body

func (*MemoryMessage) GetID

func (m *MemoryMessage) GetID() []byte

GetID returns id

func (*MemoryMessage) GetTimestamp

func (m *MemoryMessage) GetTimestamp() time.Time

GetTimestamp returns the creation timestamp

func (*MemoryMessage) Requeue

func (m *MemoryMessage) Requeue(delay time.Duration) error

Requeue requeues the message

func (*MemoryMessage) Touch

func (m *MemoryMessage) Touch() error

Touch is a nop

type Message

type Message interface {
	Finish() error
	GetBody() []byte
	GetID() []byte
	GetTimestamp() time.Time
	Requeue(delay time.Duration) error
	Touch() error
}

Message is an abstracted interface for messages from the queue.

type NSQ

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

NSQ is a queue implementation using the nsq client

func NewNSQ

func NewNSQ(cfg config.NSQConfig) (*NSQ, error)

NewNSQ returns a new NSQ-using queue

func (*NSQ) Close

func (n *NSQ) Close() error

Close shuts down all handlers and producers

func (*NSQ) Publish

func (n *NSQ) Publish(topic string, body []byte) error

Publish uses round-robin to publish a msg to one of the NSQ producers.

func (*NSQ) Subscribe

func (n *NSQ) Subscribe(topic, channel string, handler Handler, concurrency int) error

Subscribe creates a new subscriber in the nsq system

type NSQMessage

type NSQMessage struct {
	*nsq.Message
}

NSQMessage wraps nsq.Message

func (*NSQMessage) Finish

func (n *NSQMessage) Finish() error

Finish calls nsq's Finish.

func (*NSQMessage) GetBody

func (n *NSQMessage) GetBody() []byte

GetBody returns body of the nsq message

func (*NSQMessage) GetID

func (n *NSQMessage) GetID() []byte

GetID returns id of the nsq message

func (*NSQMessage) GetTimestamp

func (n *NSQMessage) GetTimestamp() time.Time

GetTimestamp converts the nsq timestamp to time.Time and returns it

func (*NSQMessage) Requeue

func (n *NSQMessage) Requeue(delay time.Duration) error

Requeue calls nsq's Requeue

func (*NSQMessage) Touch

func (n *NSQMessage) Touch() error

Touch calls nsq's Touch

type Queue

type Queue interface {
	Publish(topic string, body []byte) error
	Close() error
	Subscribe(topic, channel string, handler Handler, concurrency int) error
}

Queue is an interface for all queue-like impls.