(feat): add filter hooks (#891)

Co-authored-by: wumu.hx <[email protected]>

consumer/consumer.go

@@ -20,6 +20,7 @@ package consumer
 import (
 	"context"
 	"fmt"
+	"github.com/apache/rocketmq-client-go/v2/hooks"
 	"sort"
 	"strconv"
 	"strings"
@@ -892,6 +893,18 @@ func (dc *defaultConsumer) processPullResult(mq *primitive.MessageQueue, result
 			}
 		}
 
+		if dc.option.filterMessageHooks != nil {
+			for _, hook := range dc.option.filterMessageHooks {
+				ctx := &hooks.FilterMessageContext{
+					ConsumerGroup: dc.consumerGroup,
+					Msg:           msgListFilterAgain,
+					MQ:            mq,
+					UnitMode:      dc.unitMode,
+				}
+				msgListFilterAgain, _ = hook(ctx)
+			}
+		}
+
 		// TODO: add filter message hook
 		for _, msg := range msgListFilterAgain {
 			msg.Queue = mq

consumer/option.go

@@ -18,6 +18,7 @@ limitations under the License.
 package consumer
 
 import (
+	"github.com/apache/rocketmq-client-go/v2/hooks"
 	"time"
 
 	"github.com/apache/rocketmq-client-go/v2/internal"
@@ -108,6 +109,8 @@ type consumerOptions struct {
 	Resolver primitive.NsResolver
 
 	ConsumeGoroutineNums int
+
+	filterMessageHooks []hooks.FilterMessageHook
 }
 
 func defaultPushConsumerOptions() consumerOptions {
@@ -335,3 +338,9 @@ func WithConsumeGoroutineNums(nums int) Option {
 		opts.ConsumeGoroutineNums = nums
 	}
 }
+
+func WithFilterMessageHook(hooks []hooks.FilterMessageHook) Option {
+	return func(opts *consumerOptions) {
+		opts.filterMessageHooks = hooks
+	}
+}

go.mod

@@ -20,4 +20,4 @@ require (
 	gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c // indirect
 	gopkg.in/natefinch/lumberjack.v2 v2.0.0
 	stathat.com/c/consistent v1.0.0
-)
+)

hooks/filter_message_hook.go

@@ -0,0 +1,30 @@
+/*
+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 hooks
+
+import "github.com/apache/rocketmq-client-go/v2/primitive"
+
+type FilterMessageContext struct {
+	ConsumerGroup string
+	Msg           []*primitive.MessageExt
+	MQ            *primitive.MessageQueue
+	Arg           interface{}
+	UnitMode      bool
+}
+
+type FilterMessageHook func(ctx *FilterMessageContext) ([]*primitive.MessageExt, error)