rmq

package module

Go to main page

Versions in this module

v1

v1.0.10

Apr 15, 2024

Changes in this version

type Queue

+ RemoveMessage func(list string, message string, count int64) (int64, error)

type TestQueue

+ func (queue *TestQueue) RemoveMessage(list string, message string, count int64) (int64, error)

v1.0.9

Apr 15, 2024

Changes in this version

type TestDelivery

+ func (delivery *TestDelivery) Return() error

+ func (delivery *TestDelivery) SetPayload(payload string) error

+ func (delivery *TestDelivery) SetPayloadAndReject(payload string) error

type TestQueue

+ func (queue *TestQueue) GetReadyMessages() ([]string, error)

+ func (queue *TestQueue) GetRejectedMessages() ([]string, error)

+ func (queue *TestQueue) GetUnackedMessages() ([]string, error)

type TestRedisClient

+ func (client *TestRedisClient) LRange(key string, start, stop int64) ([]string, error)

v1.0.8

Apr 15, 2024

Changes in this version

type Queue

+ GetReadyMessages func() ([]string, error)

+ GetRejectedMessages func() ([]string, error)

+ GetUnackedMessages func() ([]string, error)

type RedisClient

+ LRange func(key string, start, stop int64) ([]string, error)

type RedisWrapper

+ func (wrapper RedisWrapper) LRange(key string, start, stop int64) ([]string, error)

v1.0.7

Apr 5, 2024

Changes in this version

+ type BatchConsumerPro interface

+ Consume func(ctx context.Context, sigChan chan Signal, batch Deliveries)

+ type BatchConsumerProFunc func(context.Context, chan Signal, Deliveries)

+ func (batchConsumerProFunc BatchConsumerProFunc) Consume(ctx context.Context, sigChan chan Signal, batch Deliveries)

type Deliveries

+ func (deliveries Deliveries) Return() (errMap map[int]error)

type Delivery

+ Return func() error

+ type SigChannel struct

+ type SigChannels struct

+ func NewSigChannels() *SigChannels

+ func (sigChannels *SigChannels) Add(name string, sigChanel chan Signal)

+ func (sigChannels *SigChannels) Get(name string) chan Signal

+ func (sigChannels *SigChannels) Remove(name string)

+ type Signal int32

v1.0.6

Aug 7, 2023

Changes in this version

type Delivery

+ SetPayloadAndReject func(payload string) error

v1.0.5

Jul 17, 2023

v1.0.4

Jul 17, 2023

v1.0.3

Jul 4, 2023

Changes in this version

type Delivery

+ SetPayload func(payload string) error

v1.0.2

Jul 3, 2023

Changes in this version

+ const HeartbeatErrorLimit

+ var ErrorAlreadyConsuming = errors.New("must not call StartConsuming() multiple times")

+ var ErrorConsumingStopped = errors.New("consuming stopped")

+ var ErrorNotConsuming = errors.New("must call StartConsuming() before adding consumers")

+ var ErrorNotFound = errors.New("entity not found")

+ func ActiveSign(active bool) string

+ func ExtractHeaderAndPayload(payload string) (http.Header, string, error)

+ func PayloadBytesWithHeader(payload []byte, header http.Header) []byte

+ func PayloadWithHeader(payload string, header http.Header) string

+ func RandomString(n int) string

+ type BatchConsumer interface

+ Consume func(batch Deliveries)

+ type BatchConsumerFunc func(Deliveries)

+ func (batchConsumerFunc BatchConsumerFunc) Consume(batch Deliveries)

+ type Cleaner struct

+ func NewCleaner(connection Connection) *Cleaner

+ func (cleaner *Cleaner) Clean() (returned int64, err error)

+ type Connection interface

+ CollectStats func(queueList []string) (Stats, error)

+ GetOpenQueues func() ([]string, error)

+ OpenQueue func(name string) (Queue, error)

+ StopAllConsuming func() <-chan struct{}

+ func OpenConnection(tag string, network string, address string, db int, errChan chan<- error) (Connection, error)

+ func OpenConnectionWithRedisClient(tag string, redisClient redis.Cmdable, errChan chan<- error) (Connection, error)

+ func OpenConnectionWithRmqRedisClient(tag string, redisClient RedisClient, errChan chan<- error) (Connection, error)

+ func OpenConnectionWithTestRedisClient(tag string, errChan chan<- error) (Connection, error)

+ type ConnectionStat struct

+ func (stat ConnectionStat) String() string

+ type ConnectionStats map[string]ConnectionStat

+ type ConsumeError struct

+ Count int

+ RedisErr error

+ func (e *ConsumeError) Error() string

+ func (e *ConsumeError) Unwrap() error

+ type Consumer interface

+ Consume func(delivery Delivery)

+ type ConsumerFunc func(Delivery)

+ func (consumerFunc ConsumerFunc) Consume(delivery Delivery)

+ type Deliveries []Delivery

+ func (deliveries Deliveries) Ack() (errMap map[int]error)

+ func (deliveries Deliveries) Extract() []Delivery

+ func (deliveries Deliveries) Payloads() []string

+ func (deliveries Deliveries) Push() (errMap map[int]error)

+ func (deliveries Deliveries) Reject() (errMap map[int]error)

+ type Delivery interface

+ Ack func() error

+ Payload func() string

+ Push func() error

+ Reject func() error

+ type DeliveryError struct

+ Count int

+ Delivery Delivery

+ RedisErr error

+ func (e *DeliveryError) Error() string

+ func (e *DeliveryError) Unwrap() error

+ type HeartbeatError struct

+ Count int

+ RedisErr error

+ func (e *HeartbeatError) Error() string

+ func (e *HeartbeatError) Unwrap() error

+ type Queue interface

+ AddBatchConsumer func(tag string, batchSize int64, timeout time.Duration, consumer BatchConsumer) (string, error)

+ AddBatchConsumerFunc func(tag string, batchSize int64, timeout time.Duration, ...) (string, error)

+ AddConsumer func(tag string, consumer Consumer) (string, error)

+ AddConsumerFunc func(tag string, consumerFunc ConsumerFunc) (string, error)

+ Destroy func() (readyCount, rejectedCount int64, err error)

+ Drain func(count int64) ([]string, error)

+ Publish func(payload ...string) error

+ PublishBytes func(payload ...[]byte) error

+ PurgeReady func() (int64, error)

+ PurgeRejected func() (int64, error)

+ ReturnRejected func(max int64) (int64, error)

+ ReturnUnacked func(max int64) (int64, error)

+ SetPushQueue func(pushQueue Queue)

+ StartConsuming func(prefetchLimit int64, pollDuration time.Duration) error

+ StopConsuming func() <-chan struct{}

+ type QueueStat struct

+ ReadyCount int64

+ RejectedCount int64

+ func NewQueueStat(readyCount, rejectedCount int64) QueueStat

+ func (stat QueueStat) ConnectionCount() int64

+ func (stat QueueStat) ConsumerCount() int64

+ func (stat QueueStat) String() string

+ func (stat QueueStat) UnackedCount() int64

+ type QueueStats map[string]QueueStat

+ type RedisClient interface

+ Del func(key string) (affected int64, err error)

+ FlushDb func() error

+ LLen func(key string) (affected int64, err error)

+ LPush func(key string, value ...string) (total int64, err error)

+ LRem func(key string, count int64, value string) (affected int64, err error)

+ LTrim func(key string, start, stop int64) error

+ RPop func(key string) (value string, err error)

+ RPopLPush func(source, destination string) (value string, err error)

+ SAdd func(key, value string) (total int64, err error)

+ SMembers func(key string) (members []string, err error)

+ SRem func(key, value string) (affected int64, err error)

+ Set func(key string, value string, expiration time.Duration) error

+ TTL func(key string) (ttl time.Duration, err error)

+ type RedisWrapper struct

+ func (wrapper RedisWrapper) Del(key string) (affected int64, err error)

+ func (wrapper RedisWrapper) FlushDb() error

+ func (wrapper RedisWrapper) LLen(key string) (affected int64, err error)

+ func (wrapper RedisWrapper) LPush(key string, value ...string) (total int64, err error)

+ func (wrapper RedisWrapper) LRem(key string, count int64, value string) (affected int64, err error)

+ func (wrapper RedisWrapper) LTrim(key string, start, stop int64) error

+ func (wrapper RedisWrapper) RPop(key string) (value string, err error)

+ func (wrapper RedisWrapper) RPopLPush(source, destination string) (value string, err error)

+ func (wrapper RedisWrapper) SAdd(key, value string) (total int64, err error)

+ func (wrapper RedisWrapper) SMembers(key string) (members []string, err error)

+ func (wrapper RedisWrapper) SRem(key, value string) (affected int64, err error)

+ func (wrapper RedisWrapper) Set(key string, value string, expiration time.Duration) error

+ func (wrapper RedisWrapper) TTL(key string) (ttl time.Duration, err error)

+ type State int

+ const Acked

+ const Pushed

+ const Rejected

+ const Unacked

+ func (i State) String() string

+ type Stats struct

+ QueueStats QueueStats

+ func CollectStats(queueList []string, mainConnection Connection) (Stats, error)

+ func NewStats() Stats

+ func (stats Stats) GetHtml(layout, refresh string) string

+ func (stats Stats) String() string

+ type TestBatchConsumer struct

+ AutoFinish bool

+ ConsumedCount int64

+ LastBatch Deliveries

+ func NewTestBatchConsumer() *TestBatchConsumer

+ func (consumer *TestBatchConsumer) Consume(batch Deliveries)

+ func (consumer *TestBatchConsumer) Consumed() int64

+ func (consumer *TestBatchConsumer) Finish()

+ func (consumer *TestBatchConsumer) Last() Deliveries

+ type TestConnection struct

+ func NewTestConnection() TestConnection

+ func (TestConnection) CollectStats([]string) (Stats, error)

+ func (TestConnection) GetOpenQueues() ([]string, error)

+ func (TestConnection) StopAllConsuming() <-chan struct{}

+ func (connection TestConnection) GetDeliveries(queueName string) []string

+ func (connection TestConnection) GetDelivery(queueName string, index int) string

+ func (connection TestConnection) OpenQueue(name string) (Queue, error)

+ func (connection TestConnection) Reset()

+ type TestConsumer struct

+ AutoAck bool

+ AutoFinish bool

+ LastDeliveries []Delivery

+ LastDelivery Delivery

+ SleepDuration time.Duration

+ func NewTestConsumer(name string) *TestConsumer

+ func (consumer *TestConsumer) Consume(delivery Delivery)

+ func (consumer *TestConsumer) Deliveries() []Delivery

+ func (consumer *TestConsumer) Finish()

+ func (consumer *TestConsumer) FinishAll()

+ func (consumer *TestConsumer) Last() Delivery

+ func (consumer *TestConsumer) String() string

+ type TestDelivery struct

+ State State

+ func NewTestDelivery(content interface{}) *TestDelivery

+ func NewTestDeliveryString(payload string) *TestDelivery

+ func (delivery *TestDelivery) Ack() error

+ func (delivery *TestDelivery) Payload() string

+ func (delivery *TestDelivery) Push() error

+ func (delivery *TestDelivery) Reject() error

+ type TestQueue struct

+ LastDeliveries []string

+ func NewTestQueue(name string) *TestQueue

+ func (*TestQueue) AddBatchConsumer(string, int64, time.Duration, BatchConsumer) (string, error)

+ func (*TestQueue) AddBatchConsumerFunc(string, int64, time.Duration, BatchConsumerFunc) (string, error)

+ func (*TestQueue) AddConsumer(string, Consumer) (string, error)

+ func (*TestQueue) AddConsumerFunc(string, ConsumerFunc) (string, error)

+ func (*TestQueue) Destroy() (int64, int64, error)

+ func (*TestQueue) Drain(count int64) ([]string, error)

+ func (*TestQueue) PurgeReady() (int64, error)

+ func (*TestQueue) PurgeRejected() (int64, error)

+ func (*TestQueue) ReturnRejected(int64) (int64, error)

+ func (*TestQueue) ReturnUnacked(int64) (int64, error)

+ func (*TestQueue) SetPushQueue(Queue)

+ func (*TestQueue) StartConsuming(int64, time.Duration) error

+ func (*TestQueue) StopConsuming() <-chan struct{}

+ func (queue *TestQueue) Publish(payload ...string) error

+ func (queue *TestQueue) PublishBytes(payload ...[]byte) error

+ func (queue *TestQueue) Reset()

+ func (queue *TestQueue) String() string

+ type TestRedisClient struct

+ func NewTestRedisClient() *TestRedisClient

+ func (client *TestRedisClient) Del(key string) (affected int64, err error)

+ func (client *TestRedisClient) FlushDb() error

+ func (client *TestRedisClient) Get(key string) (string, error)

+ func (client *TestRedisClient) LLen(key string) (affected int64, err error)

+ func (client *TestRedisClient) LPush(key string, values ...string) (total int64, err error)

+ func (client *TestRedisClient) LRem(key string, count int64, value string) (affected int64, err error)

+ func (client *TestRedisClient) LTrim(key string, start, stop int64) error

+ func (client *TestRedisClient) RPop(key string) (value string, err error)

+ func (client *TestRedisClient) RPopLPush(source, destination string) (value string, err error)

+ func (client *TestRedisClient) SAdd(key, value string) (total int64, err error)

+ func (client *TestRedisClient) SMembers(key string) (members []string, err error)

+ func (client *TestRedisClient) SRem(key, value string) (affected int64, err error)

+ func (client *TestRedisClient) Set(key string, value string, expiration time.Duration) error

+ func (client *TestRedisClient) TTL(key string) (ttl time.Duration, err error)

+ type WithHeader interface

+ Header func() http.Header

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL