queue

Mar 18, 2021

Changes in this version

type Packer

+ Marshal func(message interface{}) ([]byte, error)

+ Unmarshal func(data []byte, message interface{}) error

Mar 2, 2021

Changes in this version

+ func Providers() di.Deps

Feb 22, 2021

Changes in this version

+ var ErrEmpty = errors.New("no message available")

+ func UseGauge(gauge metrics.Gauge, interval time.Duration) func(*QueueableDispatcher)

+ func UseLogger(logger log.Logger) func(*QueueableDispatcher)

+ func UsePacker(packer Packer) func(*QueueableDispatcher)

+ func UseParallelism(parallelism int) func(*QueueableDispatcher)

+ type AbortedEvent struct

+ Err error

+ Msg *PersistedEvent

+ type ChannelConfig struct

+ Delayed string

+ Failed string

+ Reserved string

+ Timeout string

+ Waiting string

+ type DeferrablePersistentEvent struct

+ func Persist(event contract.Event, opts ...PersistOption) DeferrablePersistentEvent

+ func (d DeferrablePersistentEvent) Decorate(s *PersistedEvent)

+ func (d DeferrablePersistentEvent) Defer() time.Duration

+ type Dispatcher interface

+ Consume func(ctx context.Context) error

+ type DispatcherFactory struct

+ func (s *DispatcherFactory) Make(name string) (*QueueableDispatcher, error)

+ type DispatcherIn struct

+ AppName contract.AppName

+ Conf contract.ConfigAccessor

+ Dispatcher contract.Dispatcher

+ Env contract.Env

+ Gauge Gauge

+ Logger log.Logger

+ RedisClient redis.UniversalClient

+ type DispatcherMaker interface

+ Make func(string) (*QueueableDispatcher, error)

+ type DispatcherOut struct

+ Dispatcher Dispatcher

+ DispatcherFactory *DispatcherFactory

+ DispatcherMaker DispatcherMaker

+ ExportedConfig []config.ExportedConfig

+ QueueableDispatcher *QueueableDispatcher

+ func Provide(p DispatcherIn) (DispatcherOut, error)

+ func (d DispatcherOut) ProvideRunGroup(group *run.Group)

+ type Driver interface

+ Ack func(ctx context.Context, message *PersistedEvent) error

+ Fail func(ctx context.Context, message *PersistedEvent) error

+ Flush func(ctx context.Context, channel string) error

+ Info func(ctx context.Context) (QueueInfo, error)

+ Pop func(ctx context.Context) (*PersistedEvent, error)

+ Push func(ctx context.Context, message *PersistedEvent, delay time.Duration) error

+ Reload func(ctx context.Context, channel string) (int64, error)

+ Retry func(ctx context.Context, message *PersistedEvent) error

+ type Gauge metrics.Gauge

+ type InProcessDriver struct

+ func NewInProcessDriver() *InProcessDriver

+ func NewInProcessDriverWithPopInterval(duration time.Duration) *InProcessDriver

+ func (i *InProcessDriver) Ack(ctx context.Context, message *PersistedEvent) error

+ func (i *InProcessDriver) Fail(ctx context.Context, message *PersistedEvent) error

+ func (i *InProcessDriver) Flush(ctx context.Context, channel string) error

+ func (i *InProcessDriver) Info(ctx context.Context) (QueueInfo, error)

+ func (i *InProcessDriver) Pop(ctx context.Context) (*PersistedEvent, error)

+ func (i *InProcessDriver) Push(ctx context.Context, message *PersistedEvent, delay time.Duration) error

+ func (i *InProcessDriver) Reload(ctx context.Context, channel string) (int64, error)

+ func (i *InProcessDriver) Retry(ctx context.Context, message *PersistedEvent) error

+ type Module struct

+ Factory *DispatcherFactory

+ func New(factory *DispatcherFactory) Module

+ func (m Module) ProvideCommand(command *cobra.Command)

+ type Packer interface

+ Compress func(message interface{}) ([]byte, error)

+ Decompress func(data []byte, message interface{}) error

+ type PersistOption func(event *DeferrablePersistentEvent)

+ func Defer(duration time.Duration) PersistOption

+ func MaxAttempts(attempts int) PersistOption

+ func ScheduleAt(t time.Time) PersistOption

+ func Timeout(timeout time.Duration) PersistOption

+ func UniqueId(id string) PersistOption

+ type PersistedEvent struct

+ Attempts int

+ Backoff time.Duration

+ HandleTimeout time.Duration

+ Key string

+ MaxAttempts int

+ UniqueId string

+ Value []byte

+ func (s *PersistedEvent) Data() interface{}

+ func (s *PersistedEvent) Type() string

+ type QueueInfo struct

+ Delayed int64

+ Failed int64

+ Timeout int64

+ Waiting int64

+ type QueueableDispatcher struct

+ func WithQueue(baseDispatcher contract.Dispatcher, driver Driver, ...) *QueueableDispatcher

+ func (d *QueueableDispatcher) Consume(ctx context.Context) error

+ func (d *QueueableDispatcher) Dispatch(ctx context.Context, e contract.Event) error

+ func (d *QueueableDispatcher) Driver() Driver

+ func (d *QueueableDispatcher) Subscribe(listener contract.Listener)

+ type RedisDriver struct

+ ChannelConfig ChannelConfig

+ Logger log.Logger

+ Packer Packer

+ PopTimeout time.Duration

+ RedisClient redis.UniversalClient

+ func (r *RedisDriver) Ack(ctx context.Context, message *PersistedEvent) error

+ func (r *RedisDriver) Fail(ctx context.Context, message *PersistedEvent) error

+ func (r *RedisDriver) Flush(ctx context.Context, channel string) error

+ func (r *RedisDriver) Info(ctx context.Context) (QueueInfo, error)

+ func (r *RedisDriver) Pop(ctx context.Context) (*PersistedEvent, error)

+ func (r *RedisDriver) Push(ctx context.Context, message *PersistedEvent, delay time.Duration) error

+ func (r *RedisDriver) Reload(ctx context.Context, channel string) (int64, error)

+ func (r *RedisDriver) Retry(ctx context.Context, message *PersistedEvent) error

+ type RetryingEvent struct

+ Err error

+ Msg *PersistedEvent