gmq

package

Go to main page

Versions in this module

v0

v0.0.15

Mar 16, 2024

v0.0.14

May 30, 2023

Changes in this version

+ func TestBroker_AutoDeduplicateCompletedMsg(t *testing.T, broker Broker)

+ func TestBroker_AutoDeduplicateFailedMsg(t *testing.T, broker Broker)

+ func TestBroker_AutoDeduplicateMsgByDefault(t *testing.T, broker Broker)

+ func TestBroker_Complete(t *testing.T, broker Broker)

+ func TestBroker_DeleteAgo(t *testing.T, broker Broker)

+ func TestBroker_DeleteMsg(t *testing.T, broker Broker)

+ func TestBroker_DeleteQueue(t *testing.T, broker Broker)

+ func TestBroker_Dequeue(t *testing.T, broker Broker)

+ func TestBroker_Enqueue(t *testing.T, broker Broker)

+ func TestBroker_Fail(t *testing.T, broker Broker)

+ func TestBroker_GetMsg(t *testing.T, broker Broker)

+ func TestBroker_GetStats(t *testing.T, broker Broker)

+ func TestBroker_GetStatsByDate(t *testing.T, broker Broker)

+ func TestBroker_ListFailed(t *testing.T, broker Broker)

+ func TestBroker_ListFailedMaxItems(t *testing.T, broker Broker)

+ func TestBroker_ListMsg(t *testing.T, broker Broker)

+ func TestBroker_ListQueue(t *testing.T, broker Broker)

type Client

+ Broker Broker

+ type MockBroker struct

+ func NewMockBroker(ctrl *gomock.Controller) *MockBroker

+ func (m *MockBroker) Close() error

+ func (m *MockBroker) Complete(ctx context.Context, msg IMsg) error

+ func (m *MockBroker) DeleteAgo(ctx context.Context, queueName string, duration time.Duration) error

+ func (m *MockBroker) DeleteMsg(ctx context.Context, queueName, id string) error

+ func (m *MockBroker) DeleteQueue(ctx context.Context, queueName string) error

+ func (m *MockBroker) Dequeue(ctx context.Context, queueName string) (*Msg, error)

+ func (m *MockBroker) EXPECT() *MockBrokerMockRecorder

+ func (m *MockBroker) Enqueue(ctx context.Context, msg IMsg, opts ...OptionClient) (*Msg, error)

+ func (m *MockBroker) Fail(ctx context.Context, msg IMsg, errFail error) error

+ func (m *MockBroker) GetMsg(ctx context.Context, queueName, id string) (*Msg, error)

+ func (m *MockBroker) GetStats(ctx context.Context) ([]*QueueStat, error)

+ func (m *MockBroker) GetStatsByDate(ctx context.Context, listQueueNames []string, YYYYMMDD string) (*QueueDailyStat, error)

+ func (m *MockBroker) GetStatsWeekly(ctx context.Context, listQueueNames []string) ([]*QueueDailyStat, error)

+ func (m *MockBroker) Init(ctx context.Context, queueName string) error

+ func (m *MockBroker) ListFailed(ctx context.Context, queueName, msgId string, limit, offset int64) ([]*Msg, error)

+ func (m *MockBroker) ListMsg(ctx context.Context, queueName, state string, limit, offset int64) ([]string, error)

+ func (m *MockBroker) ListQueue(ctx context.Context) ([]string, error)

+ func (m *MockBroker) Pause(ctx context.Context, queueName string) error

+ func (m *MockBroker) Ping(ctx context.Context) error

+ func (m *MockBroker) Resume(ctx context.Context, queueName string) error

+ func (m *MockBroker) SetClock(c Clock)

+ func (m *MockBroker) UTC(flag bool)

+ type MockBrokerMockRecorder struct

+ func (mr *MockBrokerMockRecorder) Close() *gomock.Call

+ func (mr *MockBrokerMockRecorder) Complete(ctx, msg interface{}) *gomock.Call

+ func (mr *MockBrokerMockRecorder) DeleteAgo(ctx, queueName, duration interface{}) *gomock.Call

+ func (mr *MockBrokerMockRecorder) DeleteMsg(ctx, queueName, id interface{}) *gomock.Call

+ func (mr *MockBrokerMockRecorder) DeleteQueue(ctx, queueName interface{}) *gomock.Call

+ func (mr *MockBrokerMockRecorder) Dequeue(ctx, queueName interface{}) *gomock.Call

+ func (mr *MockBrokerMockRecorder) Enqueue(ctx, msg interface{}, opts ...interface{}) *gomock.Call

+ func (mr *MockBrokerMockRecorder) Fail(ctx, msg, errFail interface{}) *gomock.Call

+ func (mr *MockBrokerMockRecorder) GetMsg(ctx, queueName, id interface{}) *gomock.Call

+ func (mr *MockBrokerMockRecorder) GetStats(ctx interface{}) *gomock.Call

+ func (mr *MockBrokerMockRecorder) GetStatsByDate(ctx, listQueueNames, YYYYMMDD interface{}) *gomock.Call

+ func (mr *MockBrokerMockRecorder) GetStatsWeekly(ctx, listQueueNames interface{}) *gomock.Call

+ func (mr *MockBrokerMockRecorder) Init(ctx, queueName interface{}) *gomock.Call

+ func (mr *MockBrokerMockRecorder) ListFailed(ctx, queueName, msgId, limit, offset interface{}) *gomock.Call

+ func (mr *MockBrokerMockRecorder) ListMsg(ctx, queueName, state, limit, offset interface{}) *gomock.Call

+ func (mr *MockBrokerMockRecorder) ListQueue(ctx interface{}) *gomock.Call

+ func (mr *MockBrokerMockRecorder) Pause(ctx, queueName interface{}) *gomock.Call

+ func (mr *MockBrokerMockRecorder) Ping(ctx interface{}) *gomock.Call

+ func (mr *MockBrokerMockRecorder) Resume(ctx, queueName interface{}) *gomock.Call

+ func (mr *MockBrokerMockRecorder) SetClock(c interface{}) *gomock.Call

+ func (mr *MockBrokerMockRecorder) UTC(flag interface{}) *gomock.Call

type Msg

+ func GenerateNewMsg() *Msg

type OptTypeClient

+ const OptTypeIgnoreUnique

type OptionClient

+ func OptIgnoreUnique(s bool) OptionClient

v0.0.13

Feb 8, 2023

Changes in this version

+ const DefaultDurationRestIfNoMsg

+ const DefaultMaxBytes

+ const DefaultMaxItemsLimit

+ const DefaultTTLMsgUniq

+ const MsgStateCompleted

+ var ErrNotImplemented = errors.New("method not implemented")

+ func NewKeyDailyStatCompleted(ns, queueName, YYYYMMDD string) string

+ func NewKeyQueueCompleted(ns, queueName string) string

+ func NewKeyQueueFailedHistory(ns, queueName, msgId string) string

type Broker

+ func NewBrokerInMemory(opts *BrokerInMemoryOpts) (rs Broker, err error)

+ ListFailed func(ctx context.Context, queueName, msgId string, limit, offset int64) ([]*Msg, error)

+ ListQueue func(ctx context.Context) ([]string, error)

+ UTC func(flag bool)

+ type BrokerInMemory struct

+ func (it *BrokerInMemory) Close() error

+ func (it *BrokerInMemory) Complete(ctx context.Context, msg IMsg) error

+ func (it *BrokerInMemory) DeleteAgo(ctx context.Context, queueName string, duration time.Duration) error

+ func (it *BrokerInMemory) DeleteMsg(ctx context.Context, queueName string, msgId string) error

+ func (it *BrokerInMemory) DeleteQueue(ctx context.Context, queueName string) error

+ func (it *BrokerInMemory) Dequeue(ctx context.Context, queueName string) (*Msg, error)

+ func (it *BrokerInMemory) Enqueue(ctx context.Context, msg IMsg, opts ...OptionClient) (*Msg, error)

+ func (it *BrokerInMemory) Fail(ctx context.Context, msg IMsg, errFail error) error

+ func (it *BrokerInMemory) GetMsg(ctx context.Context, queueName string, msgId string) (*Msg, error)

+ func (it *BrokerInMemory) GetStats(ctx context.Context) ([]*QueueStat, error)

+ func (it *BrokerInMemory) GetStatsByDate(ctx context.Context, YYYYMMDD string) (*QueueDailyStat, error)

+ func (it *BrokerInMemory) GetStatsWeekly(ctx context.Context) ([]*QueueDailyStat, error)

+ func (it *BrokerInMemory) Init(ctx context.Context, queueName string) error

+ func (it *BrokerInMemory) ListFailed(ctx context.Context, queueName string, msgId string, limit int64, offset int64) (rs []*Msg, err error)

+ func (it *BrokerInMemory) ListMsg(ctx context.Context, queueName string, state string, limit int64, offset int64) ([]string, error)

+ func (it *BrokerInMemory) ListQueue(ctx context.Context) (rs []string, err error)

+ func (it *BrokerInMemory) Pause(ctx context.Context, queueName string) error

+ func (it *BrokerInMemory) Ping(ctx context.Context) error

+ func (it *BrokerInMemory) Resume(ctx context.Context, queueName string) error

+ func (it *BrokerInMemory) SetClock(c Clock)

+ func (it *BrokerInMemory) UTC(flag bool)

+ type BrokerInMemoryOpts struct

+ MaxBytes int64

type BrokerRedis

+ func (it *BrokerRedis) ListFailed(ctx context.Context, queueName string, msgId string, limit int64, offset int64) (rs []*Msg, err error)

+ func (it *BrokerRedis) ListQueue(ctx context.Context) (rs []string, err error)

+ func (it *BrokerRedis) UTC(flag bool)

+ type BrokerUnimplemented struct

+ func (*BrokerUnimplemented) Close() error

+ func (*BrokerUnimplemented) Complete(ctx context.Context, msg IMsg) error

+ func (*BrokerUnimplemented) DeleteAgo(ctx context.Context, queueName string, seconds int64) error

+ func (*BrokerUnimplemented) DeleteMsg(ctx context.Context, queueName string, msgId string) error

+ func (*BrokerUnimplemented) DeleteQueue(ctx context.Context, queueName string) error

+ func (*BrokerUnimplemented) Dequeue(ctx context.Context, queueName string) (*Msg, error)

+ func (*BrokerUnimplemented) Enqueue(ctx context.Context, msg IMsg, opts ...OptionClient) (*Msg, error)

+ func (*BrokerUnimplemented) Fail(ctx context.Context, msg IMsg, errFail error) error

+ func (*BrokerUnimplemented) GetMsg(ctx context.Context, queueName string, msgId string) (*Msg, error)

+ func (*BrokerUnimplemented) GetStats(ctx context.Context) ([]*QueueStat, error)

+ func (*BrokerUnimplemented) GetStatsByDate(ctx context.Context, YYYYMMDD string) (*QueueDailyStat, error)

+ func (*BrokerUnimplemented) GetStatsWeekly(ctx context.Context) (*[]QueueDailyStat, *QueueDailyStat, error)

+ func (*BrokerUnimplemented) Init(ctx context.Context, queueName string) error

+ func (*BrokerUnimplemented) ListMsg(ctx context.Context, queueName string, state string, limit int64, offset int64) ([]string, error)

+ func (*BrokerUnimplemented) Pause(ctx context.Context, Queuename string) error

+ func (*BrokerUnimplemented) Ping(ctx context.Context) error

+ func (*BrokerUnimplemented) Resume(ctx context.Context, Queuename string) error

+ func (*BrokerUnimplemented) SetClock(c Clock)

type CleanerParams

+ DeleteAgo time.Duration

+ MsgMaxTTL time.Duration

type Client

+ func NewClientRedis(dsn string) (rs *Client, err error)

type Config

+ MsgMaxTTL time.Duration

+ RestIfNoMsg time.Duration

type Msg

+ Expireat int64

+ Updated int64

+ type MsgHistory map[string]*list.List

type ProcessorParams

+ RestIfNoMsg time.Duration

type QueueDailyStat

+ Completed int64

+ Total int64

type Server

+ func (it *Server) Cfg() *Config

v0.0.12

Oct 27, 2022

v0.0.11

Sep 15, 2022

Changes in this version

+ func NewKeyQueuePattern(ns, queueName string) string

+ func NewKeyQueueState(ns, queueName string, state string) string

type Broker

+ DeleteMsg func(ctx context.Context, queueName, msgId string) error

+ DeleteQueue func(ctx context.Context, queueName string) error

+ GetMsg func(ctx context.Context, queueName, msgId string) (*Msg, error)

+ ListMsg func(ctx context.Context, queueName, state string, limit, offset int64) ([]string, error)

type BrokerRedis

+ func (it *BrokerRedis) DeleteMsg(ctx context.Context, queueName, msgId string) (err error)

+ func (it *BrokerRedis) DeleteQueue(ctx context.Context, queueName string) (err error)

+ func (it *BrokerRedis) GetMsg(ctx context.Context, queueName, msgId string) (msg *Msg, err error)

+ func (it *BrokerRedis) ListMsg(ctx context.Context, queueName, state string, offset, limit int64) (values []string, err error)

v0.0.10

Aug 8, 2022

Changes in this version

+ var ErrInvalidQueue = errors.New("invalid queue name")

+ var ErrWaitTimeOut = errors.New("operation times out")

type Broker

+ GetStatsWeekly func(ctx context.Context) (*[]QueueDailyStat, *QueueDailyStat, error)

+ Pause func(ctx context.Context, Queuename string) error

+ Resume func(ctx context.Context, Queuename string) error

type BrokerRedis

+ func (it *BrokerRedis) GetStatsWeekly(ctx context.Context) (*[]QueueDailyStat, *QueueDailyStat, error)

+ func (it *BrokerRedis) Pause(ctx context.Context, qname string) error

+ func (it *BrokerRedis) Resume(ctx context.Context, qname string) error

+ type MonitorInfo struct

+ Period int

+ Total int

+ TotalFailed int

+ TotalProcessed int

type Msg

+ Dieat int64

+ Err string

type Server

+ func (it *Server) Pause(qname string) error

+ func (it *Server) Resume(qname string) error

v0.0.9

Jul 6, 2022

Changes in this version

+ type CronJob struct

+ func (it *CronJob) Run()

+ type FuncEnqueueErrorHandler func(msg IMsg, opts []OptionClient, err error)

type Scheduler

+ func (it *Scheduler) Shutdown()

type SchedulerParams

+ EnqueueErrorHandler FuncEnqueueErrorHandler

v0.0.8

Jun 30, 2022

v0.0.7

Jun 29, 2022

v0.0.6

Jun 29, 2022

v0.0.5

Jun 28, 2022

Changes in this version

+ const DefaultInternalName

+ const DefaultQueueName

+ const LuaReturnCodeError

+ const LuaReturnCodeSucc

+ const MsgStateFailed

+ const MsgStatePending

+ const MsgStateProcessing

+ const MsgStateWaiting

+ const Namespace

+ const QueueNameDailyStatFailed

+ const QueueNameDailyStatProcessed

+ const QueueNameList

+ const QueueNamePaused

+ const TTLDeadMsg

+ const TTLMsg

+ var ErrIncompatibleVer = errors.New("incompatible version")

+ var ErrInternal = errors.New("internal error")

+ var ErrMsgIdConflict = errors.New("msg id conflict")

+ var ErrNoMsg = errors.New("no msg")

+ func MakeRedisUniversalClient(opts *redis.Options) (rs interface{})

+ func NewKeyDailyStatFailed(ns, queueName, YYYYMMDD string) string

+ func NewKeyDailyStatProcessed(ns, queueName, YYYYMMDD string) string

+ func NewKeyMsgDetail(ns, queueName, msgId string) string

+ func NewKeyMsgUnique(ns, queueName, msgId string) string

+ func NewKeyQueueFailed(ns, queueName string) string

+ func NewKeyQueueList() string

+ func NewKeyQueuePaused(ns, queueName string) string

+ func NewKeyQueuePending(ns, queueName string) string

+ func NewKeyQueueProcessing(ns, queueName string) string

+ func NewKeyQueueWaiting(ns, queueName string) string

+ func NotFound(ctx context.Context, msg IMsg) error

+ type Broker interface

+ Close func() error

+ Complete func(ctx context.Context, msg IMsg) error

+ Delete func(ctx context.Context, queueName, msgId string) error

+ DeleteAgo func(ctx context.Context, queueName string, seconds int64) error

+ Dequeue func(ctx context.Context, queueName string) (*Msg, error)

+ Enqueue func(ctx context.Context, msg IMsg, opts ...OptionClient) (*Msg, error)

+ Fail func(ctx context.Context, msg IMsg, errFail error) error

+ Get func(ctx context.Context, queueName, msgId string) (*Msg, error)

+ GetStats func(ctx context.Context) ([]*QueueStat, error)

+ GetStatsByDate func(ctx context.Context, YYYYMMDD string) (*QueueDailyStat, error)

+ Init func(ctx context.Context, queueName string) error

+ Ping func(ctx context.Context) error

+ SetClock func(c Clock)

+ func NewBrokerFromRedisClient(cli redis.UniversalClient) (rs Broker, err error)

+ func NewBrokerRedis(dsn string) (rs Broker, err error)

+ type BrokerRedis struct

+ func (it *BrokerRedis) Close() error

+ func (it *BrokerRedis) Complete(ctx context.Context, msg IMsg) (err error)

+ func (it *BrokerRedis) Delete(ctx context.Context, queueName, msgId string) (err error)

+ func (it *BrokerRedis) DeleteAgo(ctx context.Context, queueName string, seconds int64) (err error)

+ func (it *BrokerRedis) Dequeue(ctx context.Context, queueName string) (msg *Msg, err error)

+ func (it *BrokerRedis) Enqueue(ctx context.Context, msg IMsg, opts ...OptionClient) (rs *Msg, err error)

+ func (it *BrokerRedis) Fail(ctx context.Context, msg IMsg, errFail error) (err error)

+ func (it *BrokerRedis) Get(ctx context.Context, queueName, msgId string) (msg *Msg, err error)

+ func (it *BrokerRedis) GetStats(ctx context.Context) (rs []*QueueStat, err error)

+ func (it *BrokerRedis) GetStatsByDate(ctx context.Context, date string) (rs *QueueDailyStat, err error)

+ func (it *BrokerRedis) Init(ctx context.Context, queueName string) (err error)

+ func (it *BrokerRedis) Ping(ctx context.Context) error

+ func (it *BrokerRedis) SetClock(c Clock)

+ type Cleaner struct

+ func NewCleaner(params CleanerParams) *Cleaner

+ type CleanerParams struct

+ Broker Broker

+ Ctx context.Context

+ Logger Logger

+ QueueNames map[string]struct{}

+ type Client struct

+ func NewClient(dsn string) (rs *Client, err error)

+ func NewClientFromBroker(broker Broker) (rs *Client, err error)

+ func (it *Client) Close() error

+ func (it *Client) Enqueue(ctx context.Context, msg IMsg, opts ...OptionClient) (rs *Msg, err error)

+ type Clock interface

+ Now func() time.Time

+ func NewWallClock() Clock

+ type Config struct

+ Logger Logger

+ QueueCfgs map[string]*QueueCfg

+ type FuncWorkInterval func() time.Duration

+ type Handler interface

+ ProcessMsg func(ctx context.Context, msg IMsg) error

+ func LoggingElapsed(h Handler) Handler

+ func NotFoundHandler() Handler

+ type HandlerFunc func(ctx context.Context, msg IMsg) error

+ func (fn HandlerFunc) ProcessMsg(ctx context.Context, msg IMsg) error

+ type IMsg interface

+ GetId func() string

+ GetPayload func() []byte

+ GetQueue func() string

+ String func() string

+ type Logger interface

+ Debug func(args ...interface{})

+ Debugf func(template string, args ...interface{})

+ Error func(args ...interface{})

+ Errorf func(template string, args ...interface{})

+ Fatal func(args ...interface{})

+ Fatalf func(template string, args ...interface{})

+ Info func(args ...interface{})

+ Infof func(template string, args ...interface{})

+ Warn func(args ...interface{})

+ Warnf func(template string, args ...interface{})

+ type MiddlewareFunc func(Handler) Handler

+ type Msg struct

+ Created int64

+ Id string

+ Payload []byte

+ Processedat int64

+ Queue string

+ State string

+ func (it *Msg) GetId() string

+ func (it *Msg) GetPayload() []byte

+ func (it *Msg) GetQueue() string

+ func (it *Msg) String() string

+ type Mux struct

+ func NewMux() *Mux

+ func (it *Mux) GetPatterns() map[string]MuxEntry

+ func (it *Mux) Handle(pattern string, handler Handler)

+ func (it *Mux) Handler(msg IMsg) (h Handler, pattern string)

+ func (it *Mux) ProcessMsg(ctx context.Context, msg IMsg) error

+ func (it *Mux) Use(mws ...MiddlewareFunc)

+ type MuxEntry struct

+ type OptTypeClient int

+ const OptTypeQueueName

+ const OptTypeUniqueIn

+ type OptTypeServer int

+ const OptTypeServerWorkerNum

+ const OptTypeServerWorkerWorkIntervalFunc

+ type OptionClient interface

+ String func() string

+ Type func() OptTypeClient

+ Value func() interface{}

+ func OptQueueName(s string) OptionClient

+ func OptUniqueIn(a time.Duration) OptionClient

+ type OptionServer interface

+ String func() string

+ Type func() OptTypeServer

+ Value func() interface{}

+ func OptQueueWorkerNum(n uint16) OptionServer

+ func OptWorkerWorkInterval(f FuncWorkInterval) OptionServer

+ type Processor struct

+ func NewProcessor(params ProcessorParams) *Processor

+ type ProcessorParams struct

+ Broker Broker

+ Ctx context.Context

+ Handler Handler

+ Logger Logger

+ QueueName string

+ WorkerNum uint16

+ WorkerWorkIntervalFunc FuncWorkInterval

+ type QueueCfg struct

+ func NewQueueCfg(opts ...OptionServer) *QueueCfg

+ type QueueDailyStat struct

+ Date string

+ Failed int64

+ Processed int64

+ type QueueStat struct

+ Failed int64

+ Name string

+ Pending int64

+ Processing int64

+ Total int64

+ Waiting int64

+ type Scheduler struct

+ func NewScheduler(parmas SchedulerParams) *Scheduler

+ func (it *Scheduler) Register(cronspec string, msg IMsg, opts ...OptionServer) (entryId string, err error)

+ func (s *Scheduler) Run() (err error)

+ func (s *Scheduler) Start() (err error)

+ func (s *Scheduler) Unregister(r string) (err error)

+ type SchedulerParams struct

+ Broker Broker

+ Ctx context.Context

+ Location *time.Location

+ Logger Logger

+ type Server struct

+ func NewServer(ctx context.Context, b Broker, cfg *Config) *Server

+ func (it *Server) Run(mux *Mux) (err error)

+ func (it *Server) Shutdown()

+ type SimulatedClock struct

+ func NewSimulatedClock(t time.Time) *SimulatedClock

+ func (it *SimulatedClock) AdvanceTime(d time.Duration)

+ func (it *SimulatedClock) Now() time.Time

+ func (it *SimulatedClock) SetTime(t time.Time)

+ type WallClock struct

+ func (it *WallClock) Now() time.Time

v0.0.4

Jun 27, 2022

v0.0.3

Jun 26, 2022

v0.0.2

Jun 26, 2022

v0.0.1

Jun 26, 2022

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