common

Documentation

Index

• type AMQPConnector
  • func (ac *AMQPConnector) Close(channel *amqp.Channel, conn *amqp.Connection) error
  • func (ac *AMQPConnector) Connect(urls string, urlSeparator string, tlsConfig *tls.Config, ...) (*amqp.Connection, *amqp.Channel, amqp.Queue, <-chan amqp.Confirmation, ...)
  • func (ac *AMQPConnector) DeleteQueue(channel *amqp.Channel, queueName string) error
  • func (*AMQPConnector) InspectQueue(channel *amqp.Channel, queueName string) (*amqp.Queue, error)
  • func (ac *AMQPConnector) Open(url string, tlsConfig *tls.Config) (*amqp.Connection, *amqp.Channel, error)
• type Backend
  • func NewBackend(cnf *config.Config) Backend
  • func (b *Backend) GetConfig() *config.Config
  • func (b *Backend) IsAMQP() bool
• type Broker
  • func NewBroker(cnf *config.Config) Broker
  • func (b *Broker) AdjustRoutingKey(s *tasks.Signature)
  • func (b *Broker) GetConfig() *config.Config
  • func (b *Broker) GetDelayedTasks() ([]*tasks.Signature, error)
  • func (b *Broker) GetPendingTasks(queue string) ([]*tasks.Signature, error)
  • func (b *Broker) GetRegisteredTaskNames() []string
  • func (b *Broker) GetRetry() bool
  • func (b *Broker) GetRetryFunc() func(chan int)
  • func (b *Broker) GetRetryStopChan() chan int
  • func (b *Broker) GetStopChan() chan int
  • 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 iface.TaskProcessor)
  • func (b *Broker) StopConsuming()
• type RedisConnector
  • func (rc *RedisConnector) NewPool(socketPath, host, password string, db int, cnf *config.RedisConfig, ...) *redis.Pool

Types

type AMQPConnector

type AMQPConnector struct{}

AMQPConnector ...

func (*AMQPConnector) Close

func (ac *AMQPConnector) Close(channel *amqp.Channel, conn *amqp.Connection) error

Close connection

func (*AMQPConnector) Connect

func (ac *AMQPConnector) Connect(urls string, urlSeparator string, tlsConfig *tls.Config, exchange, exchangeType, queueName string, queueDurable, queueDelete bool, queueBindingKey string, exchangeDeclareArgs, queueDeclareArgs, queueBindingArgs amqp.Table) (*amqp.Connection, *amqp.Channel, amqp.Queue, <-chan amqp.Confirmation, <-chan *amqp.Error, error)

Connect opens a connection to RabbitMQ, declares an exchange, opens a channel, declares and binds the queue and enables publish notifications

func (*AMQPConnector) DeleteQueue

func (ac *AMQPConnector) DeleteQueue(channel *amqp.Channel, queueName string) error

DeleteQueue deletes a queue by name

func (*AMQPConnector) InspectQueue

func (*AMQPConnector) InspectQueue(channel *amqp.Channel, queueName string) (*amqp.Queue, error)

InspectQueue provides information about a specific queue

func (*AMQPConnector) Open

func (ac *AMQPConnector) Open(url string, tlsConfig *tls.Config) (*amqp.Connection, *amqp.Channel, error)

Open new RabbitMQ connection

type Backend

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

Backend represents a base backend structure

func NewBackend

func NewBackend(cnf *config.Config) Backend

NewBackend creates new Backend instance

func (*Backend) GetConfig

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

GetConfig returns config

func (*Backend) IsAMQP

func (b *Backend) IsAMQP() bool

IsAMQP ...

type Broker

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

Broker represents a base broker structure

func NewBroker

func NewBroker(cnf *config.Config) Broker

NewBroker creates new Broker instance

func (*Broker) AdjustRoutingKey

func (b *Broker) AdjustRoutingKey(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 (*Broker) GetConfig

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

GetConfig returns config

func (*Broker) GetDelayedTasks

func (b *Broker) GetDelayedTasks() ([]*tasks.Signature, error)

GetDelayedTasks returns a slice of task.Signatures that are scheduled, but not yet in the queue

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

GetRegisteredTaskNames returns registered tasks names

func (*Broker) GetRetry

func (b *Broker) GetRetry() bool

GetRetry ...

func (*Broker) GetRetryFunc

func (b *Broker) GetRetryFunc() func(chan int)

GetRetryFunc ...

func (*Broker) GetRetryStopChan

func (b *Broker) GetRetryStopChan() chan int

GetRetryStopChan ...

func (*Broker) GetStopChan

func (b *Broker) GetStopChan() chan int

GetStopChan ...

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 iface.TaskProcessor)

StartConsuming is a common part of StartConsuming method

func (*Broker) StopConsuming

func (b *Broker) StopConsuming()

StopConsuming is a common part of StopConsuming

type RedisConnector

type RedisConnector struct{}

RedisConnector ...

func (*RedisConnector) NewPool

func (rc *RedisConnector) NewPool(socketPath, host, password string, db int, cnf *config.RedisConfig, tlsConfig *tls.Config) *redis.Pool

NewPool returns a new pool of Redis connections