worker

Documentation

Index

• type ConsumeOpts
• type LogFormatter
  • func (w *LogFormatter) Format(entry *logrus.Entry) ([]byte, error)
• type Logger
  • func NewLogger(project, id string, broadcaster bus.Broadcaster) *Logger
  • func (l *Logger) Log(a ...interface{}) (n int, err error)
  • func (l *Logger) Logf(format string, a ...interface{}) (n int, err error)
  • func (l *Logger) Loglnf(format string, a ...interface{}) (n int, err error)
  • func (l *Logger) Write(p []byte) (n int, err error)
• type PublishOpts
• type QueueOpts
• type Worker
  • func NewWorker(amqpURL string) (w *Worker, err error)
  • func (w *Worker) Broadcast(ex, routingKey string, msg []byte) error
  • func (w *Worker) Close()
  • func (w *Worker) Consume(opts ConsumeOpts) (<-chan amqp.Delivery, error)
  • func (w *Worker) EnsureExchanges(exs ...string) (err error)
  • func (w *Worker) EnsureQueues(qs ...string) (err error)
  • func (w *Worker) Publish(qName string, msg []byte) error
  • func (w *Worker) QueueDeclare(opts QueueOpts) (amqp.Queue, error)
  • func (w *Worker) RawPublish(opts PublishOpts) error

Types

type ConsumeOpts

type ConsumeOpts struct {
	Queue     string
	Consumer  string
	AutoAck   bool
	Exclusive bool
	NoLocal   bool
	NoWait    bool
	Args      amqp.Table
}

ConsumeOpts explicit options for amqp.Ch.Consume

type LogFormatter

type LogFormatter struct{}

LogFormatter overrides the default text formatter for logrus. this is so the output is not too cluttered. The only addition to the log that is needed is the project name.

func (*LogFormatter) Format

func (w *LogFormatter) Format(entry *logrus.Entry) ([]byte, error)

Format formats an incomming logrus log.

type Logger

type Logger struct {
	Project     string
	Broadcaster bus.Broadcaster

	ID string
	// contains filtered or unexported fields
}

Logger knows how to send logs to a project-specific channel

func NewLogger

func NewLogger(project, id string, broadcaster bus.Broadcaster) *Logger

NewLogger Logger constructor

func (*Logger) Log

func (l *Logger) Log(a ...interface{}) (n int, err error)

Log like fmt.Print but calls l.Write

func (*Logger) Logf

func (l *Logger) Logf(format string, a ...interface{}) (n int, err error)

Logf like fmt.Printf but instead of printing to stdout, it calls the Logger Write.

func (*Logger) Loglnf

func (l *Logger) Loglnf(format string, a ...interface{}) (n int, err error)

Loglnf like Logf with \n at the end.

func (*Logger) Write

func (l *Logger) Write(p []byte) (n int, err error)

Write implements the io.Writer interface, this way the execute commands can directly link their Stdout & Stderr to project logs.

type PublishOpts

type PublishOpts struct {
	Exchange  string
	Key       string
	Mandatory bool
	Immediate bool
	Msg       amqp.Publishing
}

PublishOpts explicit options for amqp.Ch.Publish

type QueueOpts

type QueueOpts struct {
	Name       string
	Durable    bool
	AutoDelete bool
	Exclusive  bool
	NoWait     bool
	Args       amqp.Table
}

QueueOpts explicit options for amqp.Ch.QueueDeclare

type Worker

type Worker struct {
	Ch   *amqp.Channel
	Conn *amqp.Connection
}

Worker covers rabbitmq's consume essentials: connecting, logging, and acknowledging.

func NewWorker

func NewWorker(amqpURL string) (w *Worker, err error)

NewWorker takes in a queue name and returns a Worker struct

func (*Worker) Broadcast

func (w *Worker) Broadcast(ex, routingKey string, msg []byte) error

Broadcast helper function to publish to an exchange, with default options.

func (*Worker) Close

func (w *Worker) Close()

Close closes the rabbitmq channel and connection.

func (*Worker) Consume

func (w *Worker) Consume(opts ConsumeOpts) (<-chan amqp.Delivery, error)

Consume calls amqp.Channel.Consume with Consume options.

func (*Worker) EnsureExchanges

func (w *Worker) EnsureExchanges(exs ...string) (err error)

EnsureExchanges calles ExchangeDeclare on passed string. Currently used for the event worker. So no need to make a flexible api.

func (*Worker) EnsureQueues

func (w *Worker) EnsureQueues(qs ...string) (err error)

EnsureQueues calles QueueDeclare on every string passed. It uses baghdad default settings. For different settings per queue, use the QueueDeclare directly.

func (*Worker) Publish

func (w *Worker) Publish(qName string, msg []byte) error

Publish satisfies the bus.Publisher interface. Because rabbitmq's publish comes with a lot of configuration. The app almost always sends msgs with the configuration inside this method. Therefore, abstraction this makes a reasonable interface.

func (*Worker) QueueDeclare

func (w *Worker) QueueDeclare(opts QueueOpts) (amqp.Queue, error)

QueueDeclare calls QueueDeclare on the registered channel. Provides better readability for the callers Queue options

func (*Worker) RawPublish

func (w *Worker) RawPublish(opts PublishOpts) error

RawPublish sends a rabbitmq message on the worker's channel.