aaron-ai commented on code in PR #463: URL: https://github.com/apache/rocketmq-clients/pull/463#discussion_r1161372742
########## golang/push_consumer.go: ########## @@ -0,0 +1,516 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package golang + +import ( + "context" + "fmt" + "io" + "sync" + "sync/atomic" + "time" + + "github.com/apache/rocketmq-clients/golang/v5/pkg/utils" + v2 "github.com/apache/rocketmq-clients/golang/v5/protocol/v2" + "github.com/patrickmn/go-cache" + "google.golang.org/protobuf/types/known/durationpb" + "google.golang.org/protobuf/types/known/timestamppb" +) + +type PushConsumerCallback struct { + callBackFunc func(context.Context, *MessageView) (ConsumeResult, error) +} + +type PushConsumer interface { + Consumer + + Start() error + GracefulStop() error + + Subscribe(topic string, filterExpression *FilterExpression, callBackFunc func(context.Context, *MessageView) (ConsumeResult, error)) error + Unsubscribe(topic string) error + Ack(ctx context.Context, messageView *MessageView) error + Receive(ctx context.Context, invisibleDuration time.Duration) error + ChangeInvisibleDuration(messageView *MessageView, invisibleDuration time.Duration) error + ChangeInvisibleDurationAsync(messageView *MessageView, invisibleDuration time.Duration) +} + +var _ = PushConsumer(&defaultPushConsumer{}) + +type defaultPushConsumer struct { + cli *defaultClient + + once sync.Once + groupName string + topicIndex int32 + pcOpts pushConsumerOptions + scSettings *pushConsumerSettings + awaitDuration time.Duration + invisibleDuration time.Duration + subscriptionExpressionsLock sync.RWMutex + subscriptionExpressions map[string]*FilterExpression + subTopicRouteDataResultCache sync.Map + messageViewCacheSize int + messageViewCache *cache.Cache + consumeFuncs map[string]*PushConsumerCallback +} + +var NewPushConsumer = func(config *Config, opts ...PushConsumerOption) (PushConsumer, error) { + copyOpt := defaultPushConsumerOptions + pcOpts := ©Opt + cOpts := pcOpts.consumerOptions + for _, opt := range opts { + if opt.(*FuncPushConsumerOption).FuncConsumerOption != nil { + opt.apply(cOpts) + } + if opt.(*FuncPushConsumerOption).f1 != nil { + opt.apply0(pcOpts) + } + } + if len(config.ConsumerGroup) == 0 { + return nil, fmt.Errorf("consumerGroup could not be nil") + } + cli, err := cOpts.clientFunc(config) + if err != nil { + return nil, err + } + pc := &defaultPushConsumer{ + pcOpts: *pcOpts, + cli: cli.(*defaultClient), + groupName: config.ConsumerGroup, + + awaitDuration: cOpts.awaitDuration, + invisibleDuration: pcOpts.invisibleDuration, + subscriptionExpressions: cOpts.subscriptionExpressions, + messageViewCacheSize: pcOpts.messageViewCacheSize, + // This cache will never expire + messageViewCache: cache.New(0, 0), + } + if pc.subscriptionExpressions == nil { + pc.subscriptionExpressions = make(map[string]*FilterExpression) + } + pc.cli.initTopics = make([]string, 0) + for topic, _ := range pcOpts.consumerOptions.subscriptionExpressions { + pc.cli.initTopics = append(pc.cli.initTopics, topic) + } + endpoints, err := utils.ParseTarget(config.Endpoint) + if err != nil { + return nil, err + } + pc.scSettings = &pushConsumerSettings{ + clientId: pc.cli.GetClientID(), + endpoints: endpoints, + clientType: v2.ClientType_SIMPLE_CONSUMER, + requestTimeout: pc.cli.opts.timeout, + + groupName: &v2.Resource{ + Name: pc.groupName, + }, + longPollingTimeout: cOpts.awaitDuration, + subscriptionExpressions: cOpts.subscriptionExpressions, + } + pc.cli.settings = pc.scSettings + pc.cli.clientImpl = pc + return pc, nil +} + +func (pc *defaultPushConsumer) GetGroupName() string { + return pc.groupName +} + +func (pc *defaultPushConsumer) wrapReceiveMessageRequest(batchSize int, messageQueue *v2.MessageQueue, filterExpression *FilterExpression, invisibleDuration time.Duration) *v2.ReceiveMessageRequest { + return &v2.ReceiveMessageRequest{ + Group: &v2.Resource{ + Name: pc.groupName, + }, + MessageQueue: messageQueue, + FilterExpression: &v2.FilterExpression{ + Expression: filterExpression.expression, + }, + BatchSize: int32(batchSize), + InvisibleDuration: durationpb.New(invisibleDuration), + AutoRenew: false, + } +} + +func (pc *defaultPushConsumer) GracefulStop() error { + return pc.cli.GracefulStop() +} + +func (pc *defaultPushConsumer) Subscribe(topic string, filterExpression *FilterExpression, + callBackFunc func(context.Context, *MessageView) (ConsumeResult, error)) error { + _, err := pc.cli.getMessageQueues(context.Background(), topic) + if err != nil { + pc.cli.log.Errorf("subscribe error=%v with topic %s for simpleConsumer", err, topic) + return err + } + pc.subscriptionExpressionsLock.Lock() + defer pc.subscriptionExpressionsLock.Unlock() + + pc.subscriptionExpressions[topic] = filterExpression + pc.consumeFuncs[topic] = &PushConsumerCallback{callBackFunc} + return nil +} + +func (pc *defaultPushConsumer) Unsubscribe(topic string) error { + pc.subscriptionExpressionsLock.Lock() + defer pc.subscriptionExpressionsLock.Unlock() + + delete(pc.subscriptionExpressions, topic) + return nil +} + +func (pc *defaultPushConsumer) Ack(ctx context.Context, messageView *MessageView) error { + if !pc.isOn() { + return fmt.Errorf("simple consumer is not running") + } + endpoints := messageView.endpoints + watchTime := time.Now() + messageCommons := []*MessageCommon{messageView.GetMessageCommon()} + pc.cli.doBefore(MessageHookPoints_ACK, messageCommons) + request := pc.wrapAckMessageRequest(messageView) + ctx = pc.cli.Sign(ctx) + resp, err := pc.cli.clientManager.AckMessage(ctx, endpoints, request, pc.cli.opts.timeout) + messageHookPointsStatus := MessageHookPointsStatus_ERROR + duration := time.Since(watchTime) + if err != nil { + pc.cli.doAfter(MessageHookPoints_ACK, messageCommons, duration, messageHookPointsStatus) + return err + } + if resp.GetStatus().GetCode() != v2.Code_OK { + messageHookPointsStatus = MessageHookPointsStatus_OK + } + pc.cli.doAfter(MessageHookPoints_ACK, messageCommons, duration, messageHookPointsStatus) + return nil +} + +func (pc *defaultPushConsumer) isOn() bool { + return pc.cli.on.Load() +} + +func (pc *defaultPushConsumer) wrapAckMessageRequest(messageView *MessageView) *v2.AckMessageRequest { + return &v2.AckMessageRequest{ + Group: pc.scSettings.groupName, + Topic: &v2.Resource{ + Name: messageView.GetTopic(), + }, + Entries: []*v2.AckMessageEntry{ + { + MessageId: messageView.GetMessageId(), + ReceiptHandle: messageView.GetReceiptHandle(), + }, + }, + } +} + +func (pc *defaultPushConsumer) ChangeInvisibleDuration(messageView *MessageView, invisibleDuration time.Duration) error { + if !pc.isOn() { + return fmt.Errorf("simple consumer is not running") Review Comment: Simple consumer? ########## golang/push_consumer.go: ########## @@ -0,0 +1,516 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package golang Review Comment: More unit tests are needed. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
