consumer

package

Go to main page

Versions in this module

v2

v2.1.2

Aug 3, 2023

Changes in this version

+ var ErrNoNewMsg = errors.New("no new message found")

+ func IsNoNewMsgError(err error) bool

+ type ConsumeRequest struct

+ func (cr *ConsumeRequest) GetMQ() *primitive.MessageQueue

+ func (cr *ConsumeRequest) GetMsgList() []*primitive.MessageExt

+ func (cr *ConsumeRequest) GetPQ() *processQueue

+ type Limiter func(topic string)

type Option

+ func WithConsumeGoroutineNums(nums int) Option

+ func WithConsumeTimeout(timeout time.Duration) Option

+ func WithFilterMessageHook(hooks []hooks.FilterMessageHook) Option

+ func WithLimiter(limiter Limiter) Option

+ func WithPullThresholdForQueue(pullThresholdForQueue int64) Option

+ func WithPullThresholdForTopic(pullThresholdForTopic int) Option

+ func WithPullThresholdSizeForQueue(pullThresholdSizeForQueue int) Option

+ func WithPullThresholdSizeForTopic(pullThresholdSizeForTopic int) Option

+ func WithUnitName(unitName string) Option

v2.1.1

Jul 25, 2022

Changes in this version

+ func CreateReplyMessage(requestMessage *primitive.MessageExt, body []byte) (*primitive.Message, error)

+ func GetReplyToClient(msg *primitive.MessageExt) string

type Option

+ func WithConsumeConcurrentlyMaxSpan(consumeConcurrentlyMaxSpan int) Option

+ func WithConsumeTimestamp(consumeTimestamp string) Option

+ func WithConsumerPullTimeout(consumerPullTimeout time.Duration) Option

+ type StatsManager struct

+ func NewStatsManager() *StatsManager

+ func (mgr *StatsManager) GetConsumeStatus(group, topic string) ConsumeStatus

+ func (mgr *StatsManager) ShutDownStat()

v2.1.1-rc3

Jul 25, 2022

v2.1.1-rc2

Sep 10, 2021

v2.1.1-rc1

Aug 30, 2021

v2.1.0

Mar 16, 2021

Changes in this version

type Option

+ func WithNsResolver(resolver primitive.NsResolver) Option

v2.1.0-rc6

Mar 16, 2021

v2.1.0-rc5

Sep 3, 2020

v2.1.0-rc4

Jun 2, 2020

v2.1.0-rc3

Apr 27, 2020

v2.1.0-rc2

Feb 28, 2020

v2.1.0-rc1

Apr 26, 2020

v2.0.0

Mar 31, 2020

Changes in this version

+ const Mb

+ const SQL92

+ const TAG

+ var ErrBrokerNotFound = errors.New("broker can not found")

+ var ErrCreated = errors.New("consumer group has been created")

+ func AllocateByAveragely(consumerGroup, currentCID string, mqAll []*primitive.MessageQueue, ...) []*primitive.MessageQueue

+ func AllocateByAveragelyCircle(consumerGroup, currentCID string, mqAll []*primitive.MessageQueue, ...) []*primitive.MessageQueue

+ func AllocateByMachineNearby(consumerGroup, currentCID string, mqAll []*primitive.MessageQueue, ...) []*primitive.MessageQueue

+ func IsTagType(exp string) bool

+ func NewPullConsumer(options ...Option) (*defaultPullConsumer, error)

+ func NewPushConsumer(opts ...Option) (*pushConsumer, error)

+ func ShutDownStatis()

+ type AllocateStrategy func(string, string, []*primitive.MessageQueue, []string) []*primitive.MessageQueue

+ func AllocateByConfig(list []*primitive.MessageQueue) AllocateStrategy

+ func AllocateByConsistentHash(virtualNodeCnt int) AllocateStrategy

+ func AllocateByMachineRoom(consumeridcs []string) AllocateStrategy

+ type ConsumeFromWhere int

+ const ConsumeFromFirstOffset

+ const ConsumeFromLastOffset

+ const ConsumeFromTimestamp

+ type ConsumeResult int

+ const Commit

+ const ConsumeRetryLater

+ const ConsumeSuccess

+ const Rollback

+ const SuspendCurrentQueueAMoment

+ type ConsumeResultHolder struct

+ type ConsumeStatus struct

+ ConsumeFailedMsgs int64

+ ConsumeFailedTPS float64

+ ConsumeOKTPS float64

+ ConsumeRT float64

+ PullRT float64

+ PullTPS float64

+ func GetConsumeStatus(group, topic string) ConsumeStatus

+ type ConsumeType string

+ type ConsumerReturn int

+ const ExceptionReturn

+ const FailedReturn

+ const NullReturn

+ const SuccessReturn

+ const TimeoutReturn

+ type ExpressionType string

+ type MessageModel int

+ const BroadCasting

+ const Clustering

+ func (mode MessageModel) String() string

+ type MessageQueueKey primitive.MessageQueue

+ func (mq *MessageQueueKey) UnmarshalText(text []byte) error

+ func (mq MessageQueueKey) MarshalText() (text []byte, err error)

+ type MessageSelector struct

+ Expression string

+ Type ExpressionType

+ type MockOffsetStore struct

+ func NewMockOffsetStore(ctrl *gomock.Controller) *MockOffsetStore

+ func (m *MockOffsetStore) EXPECT() *MockOffsetStoreMockRecorder

+ type MockOffsetStoreMockRecorder struct

+ type OffsetSerializeWrapper struct

+ OffsetTable map[MessageQueueKey]int64

+ type OffsetStore interface

+ func NewLocalFileOffsetStore(clientID, group string) OffsetStore

+ func NewRemoteOffsetStore(group string, client internal.RMQClient, namesrv internal.Namesrvs) OffsetStore

+ type Option func(*consumerOptions)

+ func WithAutoCommit(auto bool) Option

+ func WithConsumeFromWhere(w ConsumeFromWhere) Option

+ func WithConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize int) Option

+ func WithConsumerModel(m MessageModel) Option

+ func WithConsumerOrder(order bool) Option

+ func WithCredentials(c primitive.Credentials) Option

+ func WithGroupName(group string) Option

+ func WithInstance(name string) Option

+ func WithInterceptor(fs ...primitive.Interceptor) Option

+ func WithMaxReconsumeTimes(times int32) Option

+ func WithNameServer(nameServers primitive.NamesrvAddr) Option

+ func WithNameServerDomain(nameServerUrl string) Option

+ func WithNamespace(namespace string) Option

+ func WithPullBatchSize(batchSize int32) Option

+ func WithPullInterval(interval time.Duration) Option

+ func WithRebalanceLockInterval(interval time.Duration) Option

+ func WithRetry(retries int) Option

+ func WithStrategy(strategy AllocateStrategy) Option

+ func WithSuspendCurrentQueueTimeMillis(suspendT time.Duration) Option

+ func WithTrace(traceCfg *primitive.TraceConfig) Option

+ func WithVIPChannel(enable bool) Option

+ type PullConsumer interface

+ CurrentOffset func(queue *primitive.MessageQueue) (int64, error)

+ PersistOffset func(ctx context.Context) error

+ Pull func(ctx context.Context, topic string, selector MessageSelector, numbers int) (*primitive.PullResult, error)

+ PullFrom func(ctx context.Context, queue *primitive.MessageQueue, offset int64, numbers int) (*primitive.PullResult, error)

+ Shutdown func()

+ Start func()

+ UpdateOffset func(queue *primitive.MessageQueue, offset int64) error

+ type PullRequest struct

+ func (pr *PullRequest) String() string

+ type PushConsumerCallback struct

+ func (callback PushConsumerCallback) UniqueID() string

+ type QueueLock struct

v2.0.0-rc2

Feb 28, 2020

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