brokers

Documentation

Index

• func AdjustRoutingKey(b Interface, s *tasks.Signature)
• func IsAMQP(b Interface) bool
• type AMQPBroker
  • func (b *AMQPBroker) Publish(signature *tasks.Signature) error
  • func (b *AMQPBroker) StartConsuming(consumerTag string, concurrency int, taskProcessor TaskProcessor) (bool, error)
  • func (b *AMQPBroker) StopConsuming()
• type AWSSQSBroker
  • func (b *AWSSQSBroker) GetPendingTasks(queue string) ([]*tasks.Signature, error)
  • func (b *AWSSQSBroker) Publish(signature *tasks.Signature) error
  • func (b *AWSSQSBroker) StartConsuming(consumerTag string, concurrency int, taskProcessor TaskProcessor) (bool, error)
  • func (b *AWSSQSBroker) StopConsuming()
• type Broker
  • func New(cnf *config.Config) Broker
  • func (b *Broker) GetConfig() *config.Config
  • func (b *Broker) GetPendingTasks(queue string) ([]*tasks.Signature, error)
  • func (b *Broker) GetRegisteredTaskNames() []string
  • func (b *Broker) IsTaskRegistered(name string) bool
  • func (b *Broker) Publish(signature *tasks.Signature) error
  • func (b *Broker) SetRegisteredTaskNames(names []string)
  • func (b *Broker) StartConsuming(consumerTag string, concurrency int, taskProcessor TaskProcessor) (bool, error)
  • func (b *Broker) StopConsuming()
• type EagerBroker
  • func (eagerBroker *EagerBroker) AssignWorker(w TaskProcessor)
  • func (eagerBroker *EagerBroker) GetPendingTasks(queue string) ([]*tasks.Signature, error)
  • func (eagerBroker *EagerBroker) Publish(task *tasks.Signature) error
  • func (eagerBroker *EagerBroker) StartConsuming(consumerTag string, concurrency int, p TaskProcessor) (bool, error)
  • func (eagerBroker *EagerBroker) StopConsuming()
• type EagerMode
• type ErrCouldNotUnmarshaTaskSignature
  • func NewErrCouldNotUnmarshaTaskSignature(msg []byte, err error) ErrCouldNotUnmarshaTaskSignature
  • func (e ErrCouldNotUnmarshaTaskSignature) Error() string
• type Interface
  • func NewAMQPBroker(cnf *config.Config) Interface
  • func NewAWSSQSBroker(cnf *config.Config) Interface
  • func NewEagerBroker() Interface
  • func NewRedisBroker(cnf *config.Config, host, password, socketPath string, db int) Interface
• type RedisBroker
  • func (b *RedisBroker) GetPendingTasks(queue string) ([]*tasks.Signature, error)
  • func (b *RedisBroker) Publish(signature *tasks.Signature) error
  • func (b *RedisBroker) StartConsuming(consumerTag string, concurrency int, taskProcessor TaskProcessor) (bool, error)
  • func (b *RedisBroker) StopConsuming()
• type TaskProcessor

Functions

func AdjustRoutingKey

func AdjustRoutingKey(b Interface, s *tasks.Signature)

AdjustRoutingKey makes sure the routing key is correct. If the routing key is an empty string: a) set it to binding key for direct exchange type b) set it to default queue name

func IsAMQP

func IsAMQP(b Interface) bool

IsAMQP returns true if the broker is AMQP

Types

type AMQPBroker

type AMQPBroker struct {
	Broker
	common.AMQPConnector
	// contains filtered or unexported fields
}

AMQPBroker represents an AMQP broker

func (*AMQPBroker) Publish

func (b *AMQPBroker) Publish(signature *tasks.Signature) error

Publish places a new message on the default queue

func (*AMQPBroker) StartConsuming

func (b *AMQPBroker) StartConsuming(consumerTag string, concurrency int, taskProcessor TaskProcessor) (bool, error)

StartConsuming enters a loop and waits for incoming messages

func (*AMQPBroker) StopConsuming

func (b *AMQPBroker) StopConsuming()

StopConsuming quits the loop

type AWSSQSBroker

type AWSSQSBroker struct {
	Broker
	// contains filtered or unexported fields
}

AWSSQSBroker represents a AWS SQS broker There are examples on: https://docs.aws.amazon.com/sdk-for-go/v1/developer-guide/sqs-example-create-queue.html

func (*AWSSQSBroker) GetPendingTasks

func (b *AWSSQSBroker) GetPendingTasks(queue string) ([]*tasks.Signature, error)

GetPendingTasks returns a slice of task.Signatures waiting in the queue

func (*AWSSQSBroker) Publish

func (b *AWSSQSBroker) Publish(signature *tasks.Signature) error

Publish places a new message on the default queue

func (*AWSSQSBroker) StartConsuming

func (b *AWSSQSBroker) StartConsuming(consumerTag string, concurrency int, taskProcessor TaskProcessor) (bool, error)

StartConsuming enters a loop and waits for incoming messages

func (*AWSSQSBroker) StopConsuming

func (b *AWSSQSBroker) StopConsuming()

StopConsuming quits the loop

type Broker

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

Broker represents a base broker structure

func New

func New(cnf *config.Config) Broker

New creates new Broker instance

func (*Broker) GetConfig

func (b *Broker) GetConfig() *config.Config

GetConfig returns config

func (*Broker) GetPendingTasks

func (b *Broker) GetPendingTasks(queue string) ([]*tasks.Signature, error)

GetPendingTasks returns a slice of task.Signatures waiting in the queue

func (*Broker) GetRegisteredTaskNames

func (b *Broker) GetRegisteredTaskNames() []string

func (*Broker) IsTaskRegistered

func (b *Broker) IsTaskRegistered(name string) bool

IsTaskRegistered returns true if the task is registered with this broker

func (*Broker) Publish

func (b *Broker) Publish(signature *tasks.Signature) error

Publish places a new message on the default queue

func (*Broker) SetRegisteredTaskNames

func (b *Broker) SetRegisteredTaskNames(names []string)

SetRegisteredTaskNames sets registered task names

func (*Broker) StartConsuming

func (b *Broker) StartConsuming(consumerTag string, concurrency int, taskProcessor TaskProcessor) (bool, error)

StartConsuming enters a loop and waits for incoming messages

func (*Broker) StopConsuming

func (b *Broker) StopConsuming()

StopConsuming quits the loop

type EagerBroker

type EagerBroker struct {
	Broker
	// contains filtered or unexported fields
}

EagerBroker represents an "eager" in-memory broker

func (*EagerBroker) AssignWorker

func (eagerBroker *EagerBroker) AssignWorker(w TaskProcessor)

AssignWorker assigns a worker to the eager broker

func (*EagerBroker) GetPendingTasks

func (eagerBroker *EagerBroker) GetPendingTasks(queue string) ([]*tasks.Signature, error)

GetPendingTasks returns a slice of task.Signatures waiting in the queue

func (*EagerBroker) Publish

func (eagerBroker *EagerBroker) Publish(task *tasks.Signature) error

Publish places a new message on the default queue

func (*EagerBroker) StartConsuming

func (eagerBroker *EagerBroker) StartConsuming(consumerTag string, concurrency int, p TaskProcessor) (bool, error)

StartConsuming enters a loop and waits for incoming messages

func (*EagerBroker) StopConsuming

func (eagerBroker *EagerBroker) StopConsuming()

StopConsuming quits the loop

type EagerMode

type EagerMode interface {
	AssignWorker(p TaskProcessor)
}

EagerMode interface with methods specific for this broker

type ErrCouldNotUnmarshaTaskSignature

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

ErrCouldNotUnmarshaTaskSignature ...

func NewErrCouldNotUnmarshaTaskSignature

func NewErrCouldNotUnmarshaTaskSignature(msg []byte, err error) ErrCouldNotUnmarshaTaskSignature

NewErrCouldNotUnmarshaTaskSignature returns new NewErrCouldNotUnmarshaTaskSignature instance

func (ErrCouldNotUnmarshaTaskSignature) Error

func (e ErrCouldNotUnmarshaTaskSignature) Error() string

Error implements the error interface

type Interface

type Interface interface {
	GetConfig() *config.Config
	SetRegisteredTaskNames(names []string)
	IsTaskRegistered(name string) bool
	StartConsuming(consumerTag string, concurrency int, p TaskProcessor) (bool, error)
	StopConsuming()
	Publish(task *tasks.Signature) error
	GetPendingTasks(queue string) ([]*tasks.Signature, error)
}

Interface - a common interface for all brokers

func NewAMQPBroker

func NewAMQPBroker(cnf *config.Config) Interface

NewAMQPBroker creates new AMQPBroker instance

func NewAWSSQSBroker

func NewAWSSQSBroker(cnf *config.Config) Interface

NewAWSSQSBroker creates new Broker instance

func NewEagerBroker

func NewEagerBroker() Interface

NewEagerBroker creates new EagerBroker instance

func NewRedisBroker

func NewRedisBroker(cnf *config.Config, host, password, socketPath string, db int) Interface

NewRedisBroker creates new RedisBroker instance

type RedisBroker

type RedisBroker struct {
	Broker
	common.RedisConnector
	// contains filtered or unexported fields
}

RedisBroker represents a Redis broker

func (*RedisBroker) GetPendingTasks

func (b *RedisBroker) GetPendingTasks(queue string) ([]*tasks.Signature, error)

GetPendingTasks returns a slice of task signatures waiting in the queue

func (*RedisBroker) Publish

func (b *RedisBroker) Publish(signature *tasks.Signature) error

Publish places a new message on the default queue

func (*RedisBroker) StartConsuming

func (b *RedisBroker) StartConsuming(consumerTag string, concurrency int, taskProcessor TaskProcessor) (bool, error)

StartConsuming enters a loop and waits for incoming messages

func (*RedisBroker) StopConsuming

func (b *RedisBroker) StopConsuming()

StopConsuming quits the loop

type TaskProcessor

type TaskProcessor interface {
	Process(signature *tasks.Signature) error
}

TaskProcessor - can process a delivered task This will probably always be a worker instance