mq

Documentation

Index

• Constants
• func IsCoordinatorQueue(qname string) bool
• func IsWorkerQueue(qname string) bool
• type Broker
• type InMemoryBroker
  • func NewInMemoryBroker() *InMemoryBroker
  • func (b *InMemoryBroker) PublishHeartbeat(_ context.Context, n node.Node) error
  • func (b *InMemoryBroker) PublishJob(ctx context.Context, j *job.Job) error
  • func (b *InMemoryBroker) PublishTask(ctx context.Context, qname string, t *task.Task) error
  • func (b *InMemoryBroker) Queues(ctx context.Context) ([]QueueInfo, error)
  • func (b *InMemoryBroker) Shutdown(ctx context.Context) error
  • func (b *InMemoryBroker) SubscribeForHeartbeats(handler func(n node.Node) error) error
  • func (b *InMemoryBroker) SubscribeForJobs(handler func(j *job.Job) error) error
  • func (b *InMemoryBroker) SubscribeForTasks(qname string, handler func(t *task.Task) error) error
• type QueueInfo
• type RabbitMQBroker
  • func NewRabbitMQBroker(url string) (*RabbitMQBroker, error)
  • func (b *RabbitMQBroker) PublishHeartbeat(ctx context.Context, n node.Node) error
  • func (b *RabbitMQBroker) PublishJob(ctx context.Context, j *job.Job) error
  • func (b *RabbitMQBroker) PublishTask(ctx context.Context, qname string, t *task.Task) error
  • func (b *RabbitMQBroker) Queues(ctx context.Context) ([]QueueInfo, error)
  • func (b *RabbitMQBroker) Shutdown(ctx context.Context) error
  • func (b *RabbitMQBroker) SubscribeForHeartbeats(handler func(n node.Node) error) error
  • func (b *RabbitMQBroker) SubscribeForJobs(handler func(j *job.Job) error) error
  • func (b *RabbitMQBroker) SubscribeForTasks(qname string, handler func(t *task.Task) error) error

Constants

const (
	BROKER_INMEMORY = "inmemory"
	BROKER_RABBITMQ = "rabbitmq"
)

const (
	// The queue used by the API to insert new tasks into
	QUEUE_PENDING = "pending"
	// The queue used by workers to notify the coordinator
	// that a task has began processing
	QUEUE_STARTED = "started"
	// The queue used by workers to send tasks to when
	// a task completes successfully
	QUEUE_COMPLETED = "completed"
	// The queue used by workers to send tasks to when an error
	// occurs in processing
	QUEUE_ERROR = "error"
	// The default queue for tasks
	QUEUE_DEFAULT = "default"
	// The queue used by workers to periodically
	// notify the coordinator about their aliveness
	QUEUE_HEARBEAT = "hearbeat"
	// The queue used by for job creation
	// and job-related state changes (e.g. cancellation)
	QUEUE_JOBS = "jobs"
	// The prefix used for queues that
	// are exclusive
	QUEUE_EXCLUSIVE_PREFIX = "x-"
)

Functions

func IsCoordinatorQueue

func IsCoordinatorQueue(qname string) bool

func IsWorkerQueue

func IsWorkerQueue(qname string) bool

Types

type Broker

type Broker interface {
	Queues(ctx context.Context) ([]QueueInfo, error)
	PublishTask(ctx context.Context, qname string, t *task.Task) error
	SubscribeForTasks(qname string, handler func(t *task.Task) error) error
	PublishHeartbeat(ctx context.Context, n node.Node) error
	SubscribeForHeartbeats(handler func(n node.Node) error) error
	PublishJob(ctx context.Context, j *job.Job) error
	SubscribeForJobs(handler func(j *job.Job) error) error
	Shutdown(ctx context.Context) error
}

Broker is the message-queue, pub/sub mechanism used for delivering tasks.

type InMemoryBroker

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

InMemoryBroker a very simple implementation of the Broker interface which uses in-memory channels to exchange messages. Meant for local development, tests etc.

func NewInMemoryBroker

func NewInMemoryBroker() *InMemoryBroker

func (*InMemoryBroker) PublishHeartbeat

func (b *InMemoryBroker) PublishHeartbeat(_ context.Context, n node.Node) error

func (*InMemoryBroker) PublishJob

func (b *InMemoryBroker) PublishJob(ctx context.Context, j *job.Job) error

func (*InMemoryBroker) PublishTask

func (b *InMemoryBroker) PublishTask(ctx context.Context, qname string, t *task.Task) error

func (*InMemoryBroker) Queues

func (b *InMemoryBroker) Queues(ctx context.Context) ([]QueueInfo, error)

func (*InMemoryBroker) Shutdown

func (b *InMemoryBroker) Shutdown(ctx context.Context) error

func (*InMemoryBroker) SubscribeForHeartbeats

func (b *InMemoryBroker) SubscribeForHeartbeats(handler func(n node.Node) error) error

func (*InMemoryBroker) SubscribeForJobs

func (b *InMemoryBroker) SubscribeForJobs(handler func(j *job.Job) error) error

func (*InMemoryBroker) SubscribeForTasks

func (b *InMemoryBroker) SubscribeForTasks(qname string, handler func(t *task.Task) error) error

type QueueInfo

type QueueInfo struct {
	Name        string `json:"name"`
	Size        int    `json:"size"`
	Subscribers int    `json:"subscribers"`
	Unacked     int    `json:"unacked"`
}

type RabbitMQBroker

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

func NewRabbitMQBroker

func NewRabbitMQBroker(url string) (*RabbitMQBroker, error)

func (*RabbitMQBroker) PublishHeartbeat

func (b *RabbitMQBroker) PublishHeartbeat(ctx context.Context, n node.Node) error

func (*RabbitMQBroker) PublishJob

func (b *RabbitMQBroker) PublishJob(ctx context.Context, j *job.Job) error

func (*RabbitMQBroker) PublishTask

func (b *RabbitMQBroker) PublishTask(ctx context.Context, qname string, t *task.Task) error

func (*RabbitMQBroker) Queues

func (b *RabbitMQBroker) Queues(ctx context.Context) ([]QueueInfo, error)

func (*RabbitMQBroker) Shutdown

func (b *RabbitMQBroker) Shutdown(ctx context.Context) error

func (*RabbitMQBroker) SubscribeForHeartbeats

func (b *RabbitMQBroker) SubscribeForHeartbeats(handler func(n node.Node) error) error

func (*RabbitMQBroker) SubscribeForJobs

func (b *RabbitMQBroker) SubscribeForJobs(handler func(j *job.Job) error) error

func (*RabbitMQBroker) SubscribeForTasks

func (b *RabbitMQBroker) SubscribeForTasks(qname string, handler func(t *task.Task) error) error