testbroker

Documentation

Overview

Package testbroker exports a broker implementation that should be used in package testing.

Index

• type TestBroker
  • func NewTestBroker(b base.Broker) *TestBroker
  • func (tb *TestBroker) AddToGroup(ctx context.Context, msg *base.TaskMessage, gname string) error
  • func (tb *TestBroker) AddToGroupUnique(ctx context.Context, msg *base.TaskMessage, gname string, ttl time.Duration) error
  • func (tb *TestBroker) AggregationCheck(qname, gname string, t time.Time, gracePeriod, maxDelay time.Duration, ...) (aggregationSetID string, err error)
  • func (tb *TestBroker) Archive(ctx context.Context, msg *base.TaskMessage, errMsg string) error
  • func (tb *TestBroker) CancelationPubSub() (*redis.PubSub, error)
  • func (tb *TestBroker) ClearServerState(host string, pid int, serverID string) error
  • func (tb *TestBroker) Close() error
  • func (tb *TestBroker) DeleteAggregationSet(ctx context.Context, qname, gname, aggregationSetID string) error
  • func (tb *TestBroker) DeleteExpiredCompletedTasks(qname string, batchSize int) error
  • func (tb *TestBroker) Dequeue(qnames ...string) (*base.TaskMessage, time.Time, error)
  • func (tb *TestBroker) Done(ctx context.Context, msg *base.TaskMessage) error
  • func (tb *TestBroker) Enqueue(ctx context.Context, msg *base.TaskMessage) error
  • func (tb *TestBroker) EnqueueUnique(ctx context.Context, msg *base.TaskMessage, ttl time.Duration) error
  • func (tb *TestBroker) ExtendLease(qname string, ids ...string) (time.Time, error)
  • func (tb *TestBroker) ForwardIfReady(qnames ...string) error
  • func (tb *TestBroker) ListGroups(qname string) ([]string, error)
  • func (tb *TestBroker) ListLeaseExpired(cutoff time.Time, qnames ...string) ([]*base.TaskMessage, error)
  • func (tb *TestBroker) MarkAsComplete(ctx context.Context, msg *base.TaskMessage) error
  • func (tb *TestBroker) Ping() error
  • func (tb *TestBroker) PublishCancelation(id string) error
  • func (tb *TestBroker) ReadAggregationSet(qname, gname, aggregationSetID string) ([]*base.TaskMessage, time.Time, error)
  • func (tb *TestBroker) ReclaimStaleAggregationSets(qname string) error
  • func (tb *TestBroker) Requeue(ctx context.Context, msg *base.TaskMessage) error
  • func (tb *TestBroker) Retry(ctx context.Context, msg *base.TaskMessage, processAt time.Time, errMsg string, ...) error
  • func (tb *TestBroker) Schedule(ctx context.Context, msg *base.TaskMessage, processAt time.Time) error
  • func (tb *TestBroker) ScheduleUnique(ctx context.Context, msg *base.TaskMessage, processAt time.Time, ...) error
  • func (tb *TestBroker) Sleep()
  • func (tb *TestBroker) Wakeup()
  • func (tb *TestBroker) WriteResult(qname, id string, data []byte) (int, error)
  • func (tb *TestBroker) WriteServerState(info *base.ServerInfo, workers []*base.WorkerInfo, ttl time.Duration) error

Types

type TestBroker

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

TestBroker is a broker implementation which enables to simulate Redis failure in tests.

func NewTestBroker

func NewTestBroker(b base.Broker) *TestBroker

func (*TestBroker) AddToGroup

func (tb *TestBroker) AddToGroup(ctx context.Context, msg *base.TaskMessage, gname string) error

func (*TestBroker) AddToGroupUnique

func (tb *TestBroker) AddToGroupUnique(ctx context.Context, msg *base.TaskMessage, gname string, ttl time.Duration) error

func (*TestBroker) AggregationCheck

func (tb *TestBroker) AggregationCheck(qname, gname string, t time.Time, gracePeriod, maxDelay time.Duration, maxSize int) (aggregationSetID string, err error)

func (*TestBroker) Archive

func (tb *TestBroker) Archive(ctx context.Context, msg *base.TaskMessage, errMsg string) error

func (*TestBroker) CancelationPubSub

func (tb *TestBroker) CancelationPubSub() (*redis.PubSub, error)

func (*TestBroker) ClearServerState

func (tb *TestBroker) ClearServerState(host string, pid int, serverID string) error

func (*TestBroker) Close

func (tb *TestBroker) Close() error

func (*TestBroker) DeleteAggregationSet

func (tb *TestBroker) DeleteAggregationSet(ctx context.Context, qname, gname, aggregationSetID string) error

func (*TestBroker) DeleteExpiredCompletedTasks

func (tb *TestBroker) DeleteExpiredCompletedTasks(qname string, batchSize int) error

func (*TestBroker) Dequeue

func (tb *TestBroker) Dequeue(qnames ...string) (*base.TaskMessage, time.Time, error)

func (*TestBroker) Done

func (tb *TestBroker) Done(ctx context.Context, msg *base.TaskMessage) error

func (*TestBroker) Enqueue

func (tb *TestBroker) Enqueue(ctx context.Context, msg *base.TaskMessage) error

func (*TestBroker) EnqueueUnique

func (tb *TestBroker) EnqueueUnique(ctx context.Context, msg *base.TaskMessage, ttl time.Duration) error

func (*TestBroker) ExtendLease

func (tb *TestBroker) ExtendLease(qname string, ids ...string) (time.Time, error)

func (*TestBroker) ForwardIfReady

func (tb *TestBroker) ForwardIfReady(qnames ...string) error

func (*TestBroker) ListGroups

func (tb *TestBroker) ListGroups(qname string) ([]string, error)

func (*TestBroker) ListLeaseExpired

func (tb *TestBroker) ListLeaseExpired(cutoff time.Time, qnames ...string) ([]*base.TaskMessage, error)

func (*TestBroker) MarkAsComplete

func (tb *TestBroker) MarkAsComplete(ctx context.Context, msg *base.TaskMessage) error

func (*TestBroker) Ping

func (tb *TestBroker) Ping() error

func (*TestBroker) PublishCancelation

func (tb *TestBroker) PublishCancelation(id string) error

func (*TestBroker) ReadAggregationSet

func (tb *TestBroker) ReadAggregationSet(qname, gname, aggregationSetID string) ([]*base.TaskMessage, time.Time, error)

func (*TestBroker) ReclaimStaleAggregationSets

func (tb *TestBroker) ReclaimStaleAggregationSets(qname string) error

func (*TestBroker) Requeue

func (tb *TestBroker) Requeue(ctx context.Context, msg *base.TaskMessage) error

func (*TestBroker) Retry

func (tb *TestBroker) Retry(ctx context.Context, msg *base.TaskMessage, processAt time.Time, errMsg string, isFailure bool) error

func (*TestBroker) Schedule

func (tb *TestBroker) Schedule(ctx context.Context, msg *base.TaskMessage, processAt time.Time) error

func (*TestBroker) ScheduleUnique

func (tb *TestBroker) ScheduleUnique(ctx context.Context, msg *base.TaskMessage, processAt time.Time, ttl time.Duration) error

func (*TestBroker) Sleep

func (tb *TestBroker) Sleep()

func (*TestBroker) Wakeup

func (tb *TestBroker) Wakeup()

func (*TestBroker) WriteResult

func (tb *TestBroker) WriteResult(qname, id string, data []byte) (int, error)

func (*TestBroker) WriteServerState

func (tb *TestBroker) WriteServerState(info *base.ServerInfo, workers []*base.WorkerInfo, ttl time.Duration) error