beanstalkd

Documentation

Index

• Constants
• Variables
• func NewProducerNode(endpoint, tube string) *producerNode
• func WithAt(at time.Time) queue.CallOptions
• func WithDuration(duration time.Duration) queue.CallOptions
• func WithHandle(handle ConsumeHandle) queue.Consumer
• type Beanstalkd
• type Conf
• type ConsumeHandle
• type ConsumerCluster
  • func NewConsumer(c Conf, handle queue.Consumer, opts ...ConsumerOption) *ConsumerCluster
  • func (c ConsumerCluster) Start()
  • func (c ConsumerCluster) Stop()
• type ConsumerOption
  • func WithMetrics(metrics *stat.Metrics) ConsumerOption
• type DelayPusher
• type ProducerCluster
  • func NewProducer(beanstalk Beanstalkd) *ProducerCluster
  • func (p *ProducerCluster) At(ctx context.Context, body []byte, at time.Time) (string, error)
  • func (p *ProducerCluster) Close() error
  • func (p *ProducerCluster) Delay(ctx context.Context, body []byte, delay time.Duration) (string, error)
  • func (p *ProducerCluster) Name() string
  • func (p *ProducerCluster) Push(ctx context.Context, _, body []byte, opts ...queue.CallOptions) (interface{}, error)
  • func (p *ProducerCluster) Revoke(ctx context.Context, ids string) error

Constants

const (
	PriHigh   = 1
	PriNormal = 2
	PriLow    = 3
)

Variables

var ErrTimeBeforeNow = errors.New("can't schedule task to past time")

Functions

func NewProducerNode

func NewProducerNode(endpoint, tube string) *producerNode

func WithAt

func WithAt(at time.Time) queue.CallOptions

func WithDuration

func WithDuration(duration time.Duration) queue.CallOptions

func WithHandle

func WithHandle(handle ConsumeHandle) queue.Consumer

Types

type Beanstalkd

type Beanstalkd struct {
	Endpoints []string
	Tube      string
}

type Conf

type Conf struct {
	Beanstalkd
	Redis redis.RedisConf
}

type ConsumeHandle

type ConsumeHandle func(ctx context.Context, body []byte)

type ConsumerCluster

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

func NewConsumer

func NewConsumer(c Conf, handle queue.Consumer, opts ...ConsumerOption) *ConsumerCluster

func (ConsumerCluster) Start

func (c ConsumerCluster) Start()

func (ConsumerCluster) Stop

func (c ConsumerCluster) Stop()

type ConsumerOption

type ConsumerOption func(*queueOptions)

func WithMetrics

func WithMetrics(metrics *stat.Metrics) ConsumerOption

type DelayPusher

type DelayPusher interface {
	io.Closer
	At(ctx context.Context, body []byte, at time.Time) (string, error)
	Delay(ctx context.Context, body []byte, delay time.Duration) (string, error)
	Revoke(ctx context.Context, ids string) error
}

type ProducerCluster

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

func NewProducer

func NewProducer(beanstalk Beanstalkd) *ProducerCluster

func (*ProducerCluster) At

func (p *ProducerCluster) At(ctx context.Context, body []byte, at time.Time) (string, error)

func (*ProducerCluster) Close

func (p *ProducerCluster) Close() error

func (*ProducerCluster) Delay

func (p *ProducerCluster) Delay(ctx context.Context, body []byte, delay time.Duration) (
	string, error)

func (*ProducerCluster) Name

func (p *ProducerCluster) Name() string

func (*ProducerCluster) Push

func (p *ProducerCluster) Push(
	ctx context.Context, _, body []byte, opts ...queue.CallOptions) (interface{}, error)

func (*ProducerCluster) Revoke

func (p *ProducerCluster) Revoke(ctx context.Context, ids string) error