consumer

package

v2.1.1 Latest Latest Go to latest Published: Feb 14, 2023 License: Apache-2.0 Imports: 30 Imported by: 0

Details

Valid go.mod file
Redistributable license
Tagged version
Stable version
Learn more about best practices

Repository

github.com/A-Zee029/rocketmq-client-go

Links

Open Source Insights

Documentation ¶

Overview ¶

Package consumer is a generated GoMock package.

Index ¶

Constants ¶

View Source

const (
	/**
	 * 
	 * Keywords:
	 * {@code AND, OR, NOT, BETWEEN, IN, TRUE, FALSE, IS, NULL}
	 * 
	 * 
	 * 

	 * Data type:
	 * Boolean, like: TRUE, FALSE
	 * String, like: 'abc'
	 * Decimal, like: 123
	 * Float number, like: 3.1415
	 * 
	 * 
	 * 

	 * Grammar:
	 * {@code AND, OR}
	 * {@code >, >=, <, <=, =}
	 * {@code BETWEEN A AND B}, equals to {@code >=A AND <=B}
	 * {@code NOT BETWEEN A AND B}, equals to {@code >B OR 
	 * 
{@code IN ('a', 'b')}, equals to {@code ='a' OR ='b'}, this operation only support String type.
	 * {@code IS NULL}, {@code IS NOT NULL}, check parameter whether is null, or not.
	 * {@code =TRUE}, {@code =FALSE}, check parameter whether is true, or false.
	 * 
	 * 
	 * 

	 * Example:
	 * (a > 10 AND a < 100) OR (b IS NOT NULL AND b=TRUE)
	 * 
	 */
	SQL92 = ExpressionType("SQL92")

	/**
	 * Only support or operation such as
	 * "tag1 || tag2 || tag3", 

	 * If null or * expression, meaning subscribe all.
	 */
	TAG = ExpressionType("TAG")
)

View Source

const (
	Mb = 1024 * 1024
)

Variables ¶

View Source

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

ErrNoNewMsg returns a "no new message found". Occurs only when no new message found from broker

Functions ¶

func AllocateByAveragely ¶

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

func AllocateByAveragelyCircle ¶

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

func AllocateByMachineNearby ¶

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

TODO

func CreateReplyMessage ¶

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

CreateReplyMessage build reply message from the request message

func GetReplyToClient ¶

func GetReplyToClient(msg *primitive.MessageExt) string

func IsNoNewMsgError ¶

func IsNoNewMsgError(err error) bool

func IsTagType ¶

func IsTagType(exp string) bool

func NewPullConsumer ¶

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

func NewPushConsumer ¶

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

Types ¶

type AllocateStrategy ¶

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

func AllocateByConfig ¶

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

func AllocateByConsistentHash ¶

func AllocateByConsistentHash(virtualNodeCnt int) AllocateStrategy

func AllocateByMachineRoom ¶

func AllocateByMachineRoom(consumeridcs []string) AllocateStrategy

type ConsumeFromWhere ¶

type ConsumeFromWhere int

Consuming point on consumer booting.

There are three consuming points:

CONSUME_FROM_LAST_OFFSET: consumer clients pick up where it stopped previously. If it were a newly booting up consumer client, according aging of the consumer group, there are two cases:
1. if the consumer group is created so recently that the earliest message being subscribed has yet expired, which means the consumer group represents a lately launched business, consuming will start from the very beginning;
2. if the earliest message being subscribed has expired, consuming will start from the latest messages, meaning messages born prior to the booting timestamp would be ignored.
CONSUME_FROM_FIRST_OFFSET: Consumer client will start from earliest messages available.
CONSUME_FROM_TIMESTAMP: Consumer client will start from specified timestamp, which means messages born prior to {@link #consumeTimestamp} will be ignored

const (
	ConsumeFromLastOffset ConsumeFromWhere = iota
	ConsumeFromFirstOffset
	ConsumeFromTimestamp
)

type ConsumeRequest ¶

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

func (*ConsumeRequest) GetMQ ¶

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

func (*ConsumeRequest) GetMsgList ¶

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

func (*ConsumeRequest) GetPQ ¶

func (cr *ConsumeRequest) GetPQ() *processQueue

type ConsumeResult ¶

type ConsumeResult int

const (
	ConsumeSuccess ConsumeResult = iota
	ConsumeRetryLater
	Commit
	Rollback
	SuspendCurrentQueueAMoment
)

type ConsumeResultHolder ¶

type ConsumeResultHolder struct {
	ConsumeResult
}

type ConsumeStatus ¶

type ConsumeStatus struct {
	PullRT            float64
	PullTPS           float64
	ConsumeRT         float64
	ConsumeOKTPS      float64
	ConsumeFailedTPS  float64
	ConsumeFailedMsgs int64
}

type ConsumeType ¶

type ConsumeType string

type ConsumerReturn ¶

type ConsumerReturn int

const (
	SuccessReturn ConsumerReturn = iota
	ExceptionReturn
	NullReturn
	TimeoutReturn
	FailedReturn
)

type ExpressionType ¶

type ExpressionType string

type Limiter ¶

type Limiter func(topic string)

type MessageModel ¶

type MessageModel int

Message model defines the way how messages are delivered to each consumer clients.

RocketMQ supports two message models: clustering and broadcasting. If clustering is set, consumer clients with the same {@link #ConsumerGroup} would only consume shards of the messages subscribed, which achieves load balances; Conversely, if the broadcasting is set, each consumer client will consume all subscribed messages separately.

This field defaults to clustering.

const (
	BroadCasting MessageModel = iota
	Clustering
)

func (MessageModel) String ¶

func (mode MessageModel) String() string

type MessageQueueKey ¶

type MessageQueueKey primitive.MessageQueue

func (MessageQueueKey) MarshalText ¶

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

func (*MessageQueueKey) UnmarshalText ¶

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

type MessageSelector ¶

type MessageSelector struct {
	Type       ExpressionType
	Expression string
}

type MockOffsetStore ¶

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

MockOffsetStore is a mock of OffsetStore interface.

func NewMockOffsetStore ¶

func NewMockOffsetStore(ctrl *gomock.Controller) *MockOffsetStore

NewMockOffsetStore creates a new mock instance.

func (*MockOffsetStore) EXPECT ¶

func (m *MockOffsetStore) EXPECT() *MockOffsetStoreMockRecorder

EXPECT returns an object that allows the caller to indicate expected use.

type MockOffsetStoreMockRecorder ¶

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

MockOffsetStoreMockRecorder is the mock recorder for MockOffsetStore.

type OffsetSerializeWrapper ¶

type OffsetSerializeWrapper struct {
	OffsetTable map[MessageQueueKey]int64 `json:"offsetTable"`
}

type OffsetStore ¶

type OffsetStore interface {
	// contains filtered or unexported methods
}

func NewLocalFileOffsetStore ¶

func NewLocalFileOffsetStore(clientID, group string) OffsetStore

func NewRemoteOffsetStore ¶

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

type Option ¶

type Option func(*consumerOptions)

func WithAutoCommit ¶

func WithAutoCommit(auto bool) Option

func WithConsumeConcurrentlyMaxSpan ¶

func WithConsumeConcurrentlyMaxSpan(consumeConcurrentlyMaxSpan int) Option

func WithConsumeFromWhere ¶

func WithConsumeFromWhere(w ConsumeFromWhere) Option

func WithConsumeGoroutineNums ¶

func WithConsumeGoroutineNums(nums int) Option

func WithConsumeMessageBatchMaxSize ¶

func WithConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize int) Option

func WithConsumeTimeout ¶

func WithConsumeTimeout(timeout time.Duration) Option

func WithConsumeTimestamp ¶

func WithConsumeTimestamp(consumeTimestamp string) Option

func WithConsumerModel ¶

func WithConsumerModel(m MessageModel) Option

func WithConsumerOrder ¶

func WithConsumerOrder(order bool) Option

func WithConsumerPullTimeout ¶

func WithConsumerPullTimeout(consumerPullTimeout time.Duration) Option

func WithCredentials ¶

func WithCredentials(c primitive.Credentials) Option

func WithFilterMessageHook ¶

func WithFilterMessageHook(hooks []hooks.FilterMessageHook) Option

func WithGroupName ¶

func WithGroupName(group string) Option

WithGroupName set group name address

func WithInstance ¶

func WithInstance(name string) Option

func WithInterceptor ¶

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

WithChainConsumerInterceptor returns a ConsumerOption that specifies the chained interceptor for consumer. The first interceptor will be the outer most, while the last interceptor will be the inner most wrapper around the real call.

func WithLimiter ¶

func WithLimiter(limiter Limiter) Option

func WithMaxReconsumeTimes ¶

func WithMaxReconsumeTimes(times int32) Option

WithMaxReconsumeTimes set MaxReconsumeTimes of options, if message reconsume greater than MaxReconsumeTimes, it will be sent to retry or dlq topic. more info reference by examples/consumer/retry.

func WithNameServer ¶

func WithNameServer(nameServers primitive.NamesrvAddr) Option

WithNameServer set NameServer address, only support one NameServer cluster in alpha2

func WithNameServerDomain ¶

func WithNameServerDomain(nameServerUrl string) Option

WithNameServerDomain set NameServer domain

func WithNamespace ¶

func WithNamespace(namespace string) Option

WithNamespace set the namespace of consumer

func WithNsResolver ¶

func WithNsResolver(resolver primitive.NsResolver) Option

WithNsResolver set nameserver resolver to fetch nameserver addr

func WithPullBatchSize ¶

func WithPullBatchSize(batchSize int32) Option

func WithPullInterval ¶

func WithPullInterval(interval time.Duration) Option

func WithPullThresholdForQueue ¶

func WithPullThresholdForQueue(pullThresholdForQueue int64) Option

func WithPullThresholdForTopic ¶

func WithPullThresholdForTopic(pullThresholdForTopic int) Option

func WithPullThresholdSizeForQueue ¶

func WithPullThresholdSizeForQueue(pullThresholdSizeForQueue int) Option

func WithPullThresholdSizeForTopic ¶

func WithPullThresholdSizeForTopic(pullThresholdSizeForTopic int) Option

func WithRebalanceLockInterval ¶

func WithRebalanceLockInterval(interval time.Duration) Option

func WithRetry ¶

func WithRetry(retries int) Option

WithRetry return a Option that specifies the retry times when send failed. TODO: use retry middleware instead

func WithStrategy ¶

func WithStrategy(strategy AllocateStrategy) Option

func WithSuspendCurrentQueueTimeMillis ¶

func WithSuspendCurrentQueueTimeMillis(suspendT time.Duration) Option

func WithTrace ¶

func WithTrace(traceCfg *primitive.TraceConfig) Option

WithTrace support rocketmq trace: https://github.com/apache/rocketmq/wiki/RIP-6-Message-Trace.

func WithVIPChannel ¶

func WithVIPChannel(enable bool) Option

type PullRequest ¶

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

func (*PullRequest) String ¶

func (pr *PullRequest) String() string

type PushConsumerCallback ¶

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

func (PushConsumerCallback) UniqueID ¶

func (callback PushConsumerCallback) UniqueID() string

type QueueLock ¶

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

type StatsManager ¶

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

func NewStatsManager ¶

func NewStatsManager() *StatsManager

func (*StatsManager) GetConsumeStatus ¶

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

func (*StatsManager) ShutDownStat ¶

func (mgr *StatsManager) ShutDownStat()

Source Files ¶

View all Source files

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