rocketmq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From huzongt...@apache.org
Subject [rocketmq-client-go] branch native updated: [ISSUE #86] Add Interceptor for producer and consumer. (#85)
Date Sat, 06 Jul 2019 13:30:41 GMT
This is an automated email from the ASF dual-hosted git repository.

huzongtang pushed a commit to branch native
in repository https://gitbox.apache.org/repos/asf/rocketmq-client-go.git


The following commit(s) were added to refs/heads/native by this push:
     new 3a3f93b  [ISSUE #86] Add Interceptor for producer and consumer. (#85)
3a3f93b is described below

commit 3a3f93bf5d18d8e680146c50e76af730e72595fc
Author: xujianhai666 <52450794+xujianhai666@users.noreply.github.com>
AuthorDate: Sat Jul 6 21:30:36 2019 +0800

    [ISSUE #86] Add Interceptor for producer and consumer. (#85)
    
    * add interceptor
    
    * add log
    
    * add producer example
    
    * refactor code according to new version
    
    * refactor code according to new version
    
    * add example
    
    * fix nil bug
    
    * delete extra code
    
    * delete test code
    
    * add comment. resolves #86
    
    * rename
    
    * rename
    
    * stash
    
    * stash
    
    * fix  bug
    
    * stash
    
    * refactor consumer interceptor
    
    * add example
    
    * add example
    
    * 重构interceptor
    
    * fix typo
    
    * add ctx key
    
    * remove extra code
    
    * add ctx to conusme
    
    * refactor consumer interceptor
    
    * refactor consumer interceptor
    
    * refactor consumer interceptor
    
    * lower case chainInterceptor
    
    * rename println
---
 examples/consumer/interceptor/main.go       |  76 +++++++++++++++++
 examples/consumer/{ => simple}/main.go      |  10 +--
 examples/producer/{ => interceptor}/main.go |  32 +++++--
 examples/producer/{ => simple}/main.go      |   8 +-
 internal/consumer/consumer.go               |   2 +-
 internal/consumer/pull_consumer.go          |   4 +-
 internal/consumer/push_consumer.go          | 124 +++++++++++++++++++---------
 internal/kernel/client.go                   |  56 +++++++------
 internal/producer/producer.go               | 101 ++++++++++++++++++----
 primitive/consume.go                        |  16 +++-
 primitive/ctx.go                            |  60 ++++++++++++++
 primitive/interceptor.go                    |  57 +++++++++++++
 primitive/options.go                        |  99 +++++++++++++++++++++-
 primitive/result.go                         |  18 +++-
 14 files changed, 556 insertions(+), 107 deletions(-)

diff --git a/examples/consumer/interceptor/main.go b/examples/consumer/interceptor/main.go
new file mode 100644
index 0000000..01fedc3
--- /dev/null
+++ b/examples/consumer/interceptor/main.go
@@ -0,0 +1,76 @@
+/*
+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 main
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"time"
+
+	"github.com/apache/rocketmq-client-go/internal/consumer"
+	"github.com/apache/rocketmq-client-go/primitive"
+)
+
+func main() {
+	c, _ := consumer.NewPushConsumer("testGroup", "127.0.0.1:9876",
+		primitive.WithConsumerModel(primitive.Clustering),
+		primitive.WithConsumeFromWhere(primitive.ConsumeFromFirstOffset),
+		primitive.WithChainConsumerInterceptor(UserFistInterceptor(), UserSecondInterceptor()))
+	err := c.Subscribe("TopicTest", primitive.MessageSelector{}, func(ctx *primitive.ConsumeMessageContext,
+		msgs []*primitive.MessageExt) (primitive.ConsumeResult, error) {
+		fmt.Println("subscribe callback: %v", msgs)
+		return primitive.ConsumeSuccess, nil
+	})
+	if err != nil {
+		fmt.Println(err.Error())
+	}
+	// Note: start after subscribe
+	err = c.Start()
+	if err != nil {
+		fmt.Println(err.Error())
+		os.Exit(-1)
+	}
+	time.Sleep(time.Hour)
+}
+
+func UserFistInterceptor() primitive.CInterceptor {
+	return func(ctx context.Context, req, reply interface{}, next primitive.CInvoker) error
{
+		msgCtx, _ := primitive.GetConsumerCtx(ctx)
+		fmt.Printf("msgCtx: %v, mehtod: %s", msgCtx, primitive.GetMethod(ctx))
+
+		msgs := req.([]*primitive.MessageExt)
+		fmt.Printf("user first interceptor before invoke: %v\n", msgs)
+		e := next(ctx, msgs, reply)
+
+		holder := reply.(*primitive.ConsumeResultHolder)
+		fmt.Printf("user first interceptor after invoke: %v, result: %v\n", msgs, holder)
+		return e
+	}
+}
+
+func UserSecondInterceptor() primitive.CInterceptor {
+	return func(ctx context.Context, req, reply interface{}, next primitive.CInvoker)  error
{
+		msgs := req.([]*primitive.MessageExt)
+		fmt.Printf("user second interceptor before invoke: %v\n", msgs)
+		e := next(ctx, msgs, reply)
+		holder := reply.(*primitive.ConsumeResultHolder)
+		fmt.Printf("user second interceptor after invoke: %v, result: %v\n", msgs, holder)
+		return e
+	}
+}
diff --git a/examples/consumer/main.go b/examples/consumer/simple/main.go
similarity index 79%
rename from examples/consumer/main.go
rename to examples/consumer/simple/main.go
index f433f73..70bbbd4 100644
--- a/examples/consumer/main.go
+++ b/examples/consumer/simple/main.go
@@ -27,14 +27,10 @@ import (
 )
 
 func main() {
-	c, _ := consumer.NewPushConsumer("testGroup", primitive.ConsumerOption{
-		NameServerAddr: "127.0.0.1:9876",
-		ConsumerModel:  primitive.Clustering,
-		FromWhere:      primitive.ConsumeFromFirstOffset,
-	})
-	err := c.Subscribe("test", primitive.MessageSelector{}, func(ctx *consumer.ConsumeMessageContext,
+	c, _ := consumer.NewPushConsumer("testGroup", "127.0.0.1:9876")
+	err := c.Subscribe("TopicTest", primitive.MessageSelector{}, func(ctx *primitive.ConsumeMessageContext,
 		msgs []*primitive.MessageExt) (primitive.ConsumeResult, error) {
-		fmt.Println(msgs)
+		fmt.Println("subscribe callback: %v", msgs)
 		return primitive.ConsumeSuccess, nil
 	})
 	if err != nil {
diff --git a/examples/producer/main.go b/examples/producer/interceptor/main.go
similarity index 56%
copy from examples/producer/main.go
copy to examples/producer/interceptor/main.go
index b52b266..f7bcf7a 100644
--- a/examples/producer/main.go
+++ b/examples/producer/interceptor/main.go
@@ -15,6 +15,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
+// Package main implements a producer with user custom interceptor.
 package main
 
 import (
@@ -27,19 +28,18 @@ import (
 )
 
 func main() {
-	opt := primitive.ProducerOptions{
-		NameServerAddr:           "127.0.0.1:9876",
-		RetryTimesWhenSendFailed: 2,
-	}
-	p, _ := producer.NewProducer(opt)
+	nameServerAddr := "127.0.0.1:9876"
+	p, _ := producer.NewProducer(nameServerAddr, primitive.WithRetry(2),
+		primitive.WithChainProducerInterceptor(UserFirstInterceptor(), UserSecondInterceptor()))
 	err := p.Start()
 	if err != nil {
 		fmt.Printf("start producer error: %s", err.Error())
 		os.Exit(1)
 	}
-	for i := 0; i < 1000; i++ {
+	for i := 0; i < 10; i++ {
 		res, err := p.SendSync(context.Background(), &primitive.Message{
-			Topic: "test",
+			//Topic: "test",
+			Topic: "TopicTest",
 			Body:  []byte("Hello RocketMQ Go Client!"),
 		})
 
@@ -54,3 +54,21 @@ func main() {
 		fmt.Printf("shundown producer error: %s", err.Error())
 	}
 }
+
+func UserFirstInterceptor() primitive.PInterceptor {
+	return func(ctx context.Context, req, reply interface{}, next primitive.PInvoker) error
{
+		fmt.Printf("user first interceptor before invoke: req:%v, reply: %v\n", req, reply)
+		err := next(ctx, req, reply)
+		fmt.Printf("user first interceptor after invoke: req: %v, reply: %v \n", req, reply)
+		return err
+	}
+}
+
+func UserSecondInterceptor() primitive.PInterceptor {
+	return func(ctx context.Context, req, reply interface{}, next primitive.PInvoker) error
{
+		fmt.Printf("user second interceptor before invoke: req: %v, reply: %v\n", req, reply)
+		err := next(ctx, req, reply)
+		fmt.Printf("user second interceptor after invoke: req: %v, reply: %v \n", req, reply)
+		return err
+	}
+}
diff --git a/examples/producer/main.go b/examples/producer/simple/main.go
similarity index 89%
rename from examples/producer/main.go
rename to examples/producer/simple/main.go
index b52b266..39c885d 100644
--- a/examples/producer/main.go
+++ b/examples/producer/simple/main.go
@@ -26,12 +26,10 @@ import (
 	"github.com/apache/rocketmq-client-go/primitive"
 )
 
+// Package main implements a simple producer to send message.
 func main() {
-	opt := primitive.ProducerOptions{
-		NameServerAddr:           "127.0.0.1:9876",
-		RetryTimesWhenSendFailed: 2,
-	}
-	p, _ := producer.NewProducer(opt)
+	nameServerAddr := "127.0.0.1:9876"
+	p, _ := producer.NewProducer(nameServerAddr, primitive.WithRetry(2))
 	err := p.Start()
 	if err != nil {
 		fmt.Printf("start producer error: %s", err.Error())
diff --git a/internal/consumer/consumer.go b/internal/consumer/consumer.go
index e9aa1b0..ecbbbb4 100644
--- a/internal/consumer/consumer.go
+++ b/internal/consumer/consumer.go
@@ -99,7 +99,7 @@ type defaultConsumer struct {
 	state     kernel.ServiceState
 	pause     bool
 	once      sync.Once
-	option    primitive.ConsumerOption
+	option    primitive.ConsumerOptions
 	// key: int, hash(*primitive.MessageQueue)
 	// value: *processQueue
 	processQueueTable sync.Map
diff --git a/internal/consumer/pull_consumer.go b/internal/consumer/pull_consumer.go
index ead633f..2338199 100644
--- a/internal/consumer/pull_consumer.go
+++ b/internal/consumer/pull_consumer.go
@@ -38,7 +38,7 @@ var (
 	queueCounterTable sync.Map
 )
 
-func NewConsumer(config primitive.ConsumerOption) *defaultPullConsumer {
+func NewConsumer(config primitive.ConsumerOptions) *defaultPullConsumer {
 	return &defaultPullConsumer{
 		option: config,
 	}
@@ -46,7 +46,7 @@ func NewConsumer(config primitive.ConsumerOption) *defaultPullConsumer {
 
 type defaultPullConsumer struct {
 	state     kernel.ServiceState
-	option    primitive.ConsumerOption
+	option    primitive.ConsumerOptions
 	client    *kernel.RMQClient
 	GroupName string
 	Model     primitive.MessageModel
diff --git a/internal/consumer/push_consumer.go b/internal/consumer/push_consumer.go
index be839bc..8acbc4c 100644
--- a/internal/consumer/push_consumer.go
+++ b/internal/consumer/push_consumer.go
@@ -49,46 +49,51 @@ type PushConsumer interface {
 	Start() error
 	Shutdown()
 	Subscribe(topic string, selector primitive.MessageSelector,
-		f func(*ConsumeMessageContext, []*primitive.MessageExt) (primitive.ConsumeResult, error))
error
+		f func(*primitive.ConsumeMessageContext, []*primitive.MessageExt) (primitive.ConsumeResult,
error)) error
 }
 
 type pushConsumer struct {
 	*defaultConsumer
 	queueFlowControlTimes        int
 	queueMaxSpanFlowControlTimes int
-	consume                      func(*ConsumeMessageContext, []*primitive.MessageExt) (primitive.ConsumeResult,
error)
+	consume                      func(*primitive.ConsumeMessageContext, []*primitive.MessageExt)
(primitive.ConsumeResult, error)
 	submitToConsume              func(*processQueue, *primitive.MessageQueue)
 	subscribedTopic              map[string]string
+
+	interceptor primitive.CInterceptor
 }
 
-func NewPushConsumer(consumerGroup string, opt primitive.ConsumerOption) (PushConsumer, error)
{
-	if err := utils.VerifyIP(opt.NameServerAddr); err != nil {
+func NewPushConsumer(consumerGroup string, nameServerAddr string, opts ...*primitive.ConsumerOption)
(PushConsumer, error) {
+	if err := utils.VerifyIP(nameServerAddr); err != nil {
 		return nil, err
 	}
-	opt.InstanceName = "DEFAULT"
-	opt.ClientIP = utils.LocalIP()
-	if opt.NameServerAddr == "" {
-		rlog.Fatal("opt.NameServerAddr can't be empty")
+	if nameServerAddr == "" {
+		rlog.Fatal("opts.NameServerAddr can't be empty")
 	}
-	err := os.Setenv(kernel.EnvNameServerAddr, opt.NameServerAddr)
+	err := os.Setenv(kernel.EnvNameServerAddr, nameServerAddr)
 	if err != nil {
 		rlog.Fatal("set env=EnvNameServerAddr error: %s ", err.Error())
 	}
+
+	pushOpts := primitive.DefaultPushConsumerOptions()
+	for _, op := range opts {
+		op.Apply(&pushOpts)
+	}
+
+	pushOpts.NameServerAddr = nameServerAddr
+
 	dc := &defaultConsumer{
 		consumerGroup:  consumerGroup,
 		cType:          _PushConsume,
 		state:          kernel.StateCreateJust,
 		prCh:           make(chan PullRequest, 4),
-		model:          opt.ConsumerModel,
-		consumeOrderly: opt.ConsumeOrderly,
-		fromWhere:      opt.FromWhere,
-		option:         opt,
+		model:          pushOpts.ConsumerModel,
+		consumeOrderly: pushOpts.ConsumeOrderly,
+		fromWhere:      pushOpts.FromWhere,
+		allocate:       pushOpts.Strategy,
+		option:         pushOpts,
 	}
 
-	if opt.Strategy == nil {
-		opt.Strategy = primitive.AllocateByAveragely
-	}
-	dc.allocate = opt.Strategy
 	p := &pushConsumer{
 		defaultConsumer: dc,
 		subscribedTopic: make(map[string]string, 0),
@@ -99,9 +104,37 @@ func NewPushConsumer(consumerGroup string, opt primitive.ConsumerOption)
(PushCo
 	} else {
 		p.submitToConsume = p.consumeMessageCurrently
 	}
+
+	chainInterceptor(p)
+
 	return p, nil
 }
 
+// chainInterceptor chain list of interceptor as one interceptor
+func chainInterceptor(p *pushConsumer) {
+	interceptors := p.option.Interceptors
+	switch len(interceptors) {
+	case 0:
+		p.interceptor = nil
+	case 1:
+		p.interceptor = interceptors[0]
+	default:
+		p.interceptor = func(ctx context.Context, req, reply interface{}, invoker primitive.CInvoker)
error {
+			return interceptors[0](ctx, req, reply, getChainedInterceptor(interceptors, 0, invoker))
+		}
+	}
+}
+
+// getChainedInterceptor recursively generate the chained invoker.
+func getChainedInterceptor(interceptors []primitive.CInterceptor, cur int, finalInvoker primitive.CInvoker)
primitive.CInvoker {
+	if cur == len(interceptors)-1 {
+		return finalInvoker
+	}
+	return func(ctx context.Context, req, reply interface{}) error {
+		return interceptors[cur+1](ctx, req, reply, getChainedInterceptor(interceptors, cur+1,
finalInvoker))
+	}
+}
+
 func (pc *pushConsumer) Start() error {
 	var err error
 	pc.once.Do(func() {
@@ -164,7 +197,7 @@ func (pc *pushConsumer) Start() error {
 func (pc *pushConsumer) Shutdown() {}
 
 func (pc *pushConsumer) Subscribe(topic string, selector primitive.MessageSelector,
-	f func(*ConsumeMessageContext, []*primitive.MessageExt) (primitive.ConsumeResult, error))
error {
+	f func(*primitive.ConsumeMessageContext, []*primitive.MessageExt) (primitive.ConsumeResult,
error)) error {
 	if pc.state != kernel.StateCreateJust {
 		return errors.New("subscribe topic only started before")
 	}
@@ -424,6 +457,7 @@ func (pc *pushConsumer) pullMessage(request *PullRequest) {
 			sleepTime = _PullDelayTimeWhenError
 			goto NEXT
 		}
+
 		result, err := pc.client.PullMessage(context.Background(), brokerResult.BrokerAddr, pullRequest)
 		if err != nil {
 			rlog.Warnf("pull message from %s error: %s", brokerResult.BrokerAddr, err.Error())
@@ -447,6 +481,8 @@ func (pc *pushConsumer) pullMessage(request *PullRequest) {
 			rt := time.Now().Sub(beginTime)
 			increasePullRT(pc.consumerGroup, request.mq.Topic, rt)
 
+			result.SetMessageExts(primitive.DecodeMessage(result.GetBody()))
+
 			msgFounded := result.GetMessageExts()
 			firstMsgOffset := int64(math.MaxInt64)
 			if msgFounded != nil && len(msgFounded) != 0 {
@@ -485,7 +521,7 @@ func (pc *pushConsumer) correctTagsOffset(pr *PullRequest) {
 	// TODO
 }
 
-func (pc *pushConsumer) sendMessageBack(ctx *ConsumeMessageContext, msg *primitive.MessageExt)
bool {
+func (pc *pushConsumer) sendMessageBack(ctx *primitive.ConsumeMessageContext, msg *primitive.MessageExt)
bool {
 	return true
 }
 
@@ -570,16 +606,6 @@ func (pc *pushConsumer) removeUnnecessaryMessageQueue(mq *primitive.MessageQueue
 	return true
 }
 
-type ConsumeMessageContext struct {
-	consumerGroup string
-	msgs          []*primitive.MessageExt
-	mq            *primitive.MessageQueue
-	success       bool
-	status        string
-	// mqTractContext
-	properties map[string]string
-}
-
 func (pc *pushConsumer) consumeMessageCurrently(pq *processQueue, mq *primitive.MessageQueue)
{
 	msgs := pq.getMessages()
 	if msgs == nil {
@@ -603,8 +629,8 @@ func (pc *pushConsumer) consumeMessageCurrently(pq *processQueue, mq *primitive.
 				return
 			}
 
-			ctx := &ConsumeMessageContext{
-				properties: make(map[string]string),
+			msgCtx := &primitive.ConsumeMessageContext{
+				Properties: make(map[string]string),
 			}
 			// TODO hook
 			beginTime := time.Now()
@@ -620,16 +646,40 @@ func (pc *pushConsumer) consumeMessageCurrently(pq *processQueue, mq
*primitive.
 						beginTime.UnixNano()/int64(time.Millisecond), 10)
 				}
 			}
-			result, err := pc.consume(ctx, subMsgs)
+			var result primitive.ConsumeResult
+
+			var err error
+			if pc.interceptor == nil {
+				result, err = pc.consume(msgCtx, subMsgs)
+			} else {
+				var container primitive.ConsumeResultHolder
+
+				ctx := context.Background()
+				ctx = primitive.WithConsumerCtx(ctx, msgCtx)
+				ctx = primitive.WithMehod(ctx, primitive.ConsumerPush)
+
+				err = pc.interceptor(ctx, subMsgs, &container, func(ctx context.Context, req, reply
interface{}) error {
+					consumerCtx, _ := primitive.GetConsumerCtx(ctx)
+
+					msgs := req.([]*primitive.MessageExt)
+					r, e := pc.consume(consumerCtx, msgs)
+
+					realReply := reply.(*primitive.ConsumeResultHolder)
+					realReply.ConsumeResult = r
+					return e
+				})
+				result = container.ConsumeResult
+			}
+
 			consumeRT := time.Now().Sub(beginTime)
 			if err != nil {
-				ctx.properties["ConsumeContextType"] = "EXCEPTION"
+				msgCtx.Properties["ConsumeContextType"] = "EXCEPTION"
 			} else if consumeRT >= pc.option.ConsumeTimeout {
-				ctx.properties["ConsumeContextType"] = "TIMEOUT"
+				msgCtx.Properties["ConsumeContextType"] = "TIMEOUT"
 			} else if result == primitive.ConsumeSuccess {
-				ctx.properties["ConsumeContextType"] = "SUCCESS"
+				msgCtx.Properties["ConsumeContextType"] = "SUCCESS"
 			} else {
-				ctx.properties["ConsumeContextType"] = "RECONSUME_LATER"
+				msgCtx.Properties["ConsumeContextType"] = "RECONSUME_LATER"
 			}
 
 			// TODO hook
@@ -648,7 +698,7 @@ func (pc *pushConsumer) consumeMessageCurrently(pq *processQueue, mq *primitive.
 					} else {
 						for i := 0; i < len(msgs); i++ {
 							msg := msgs[i]
-							if !pc.sendMessageBack(ctx, msg) {
+							if !pc.sendMessageBack(msgCtx, msg) {
 								msg.ReconsumeTimes += 1
 								msgBackFailed = append(msgBackFailed, msg)
 							}
diff --git a/internal/kernel/client.go b/internal/kernel/client.go
index 4687989..94f45c9 100644
--- a/internal/kernel/client.go
+++ b/internal/kernel/client.go
@@ -292,7 +292,7 @@ func (c *RMQClient) SendMessageOneWay(ctx context.Context, brokerAddrs
string, r
 	return nil, err
 }
 
-func (c *RMQClient) ProcessSendResponse(brokerName string, cmd *remote.RemotingCommand, msgs
...*primitive.Message) *primitive.SendResult {
+func (c *RMQClient) ProcessSendResponse(brokerName string, cmd *remote.RemotingCommand, resp
*primitive.SendResult, msgs ...*primitive.Message) {
 	var status primitive.SendStatus
 	switch cmd.Code {
 	case ResFlushDiskTimeout:
@@ -321,20 +321,20 @@ func (c *RMQClient) ProcessSendResponse(brokerName string, cmd *remote.RemotingC
 
 	qId, _ := strconv.Atoi(cmd.ExtFields["queueId"])
 	off, _ := strconv.ParseInt(cmd.ExtFields["queueOffset"], 10, 64)
-	return &primitive.SendResult{
-		Status:      status,
-		MsgID:       cmd.ExtFields["msgId"],
-		OffsetMsgID: cmd.ExtFields["msgId"],
-		MessageQueue: &primitive.MessageQueue{
-			Topic:      msgs[0].Topic,
-			BrokerName: brokerName,
-			QueueId:    qId,
-		},
-		QueueOffset: off,
-		//TransactionID: sendResponse.TransactionId,
-		RegionID: regionId,
-		TraceOn:  trace != "" && trace != _TranceOff,
+
+	resp.Status = status
+	resp.MsgID = cmd.ExtFields["msgId"]
+	resp.OffsetMsgID = cmd.ExtFields["msgId"]
+	resp.MessageQueue = &primitive.MessageQueue{
+		Topic:      msgs[0].Topic,
+		BrokerName: brokerName,
+		QueueId:    qId,
 	}
+	resp.QueueOffset = off
+	//TransactionID: sendResponse.TransactionId,
+	resp.RegionID = regionId
+	resp.TraceOn = trace != "" && trace != _TranceOff
+
 }
 
 // PullMessage with sync
@@ -349,6 +349,7 @@ func (c *RMQClient) PullMessage(ctx context.Context, brokerAddrs string,
request
 }
 
 func (c *RMQClient) processPullResponse(response *remote.RemotingCommand) (*primitive.PullResult,
error) {
+
 	pullResult := &primitive.PullResult{}
 	switch response.Code {
 	case ResSuccess:
@@ -363,29 +364,32 @@ func (c *RMQClient) processPullResponse(response *remote.RemotingCommand)
(*prim
 		return nil, fmt.Errorf("unknown Response Code: %d, remark: %s", response.Code, response.Remark)
 	}
 
-	v, exist := response.ExtFields["maxOffset"]
+	c.decodeCommandCustomHeader(pullResult, response)
+	pullResult.SetBody(response.Body)
+
+	return pullResult, nil
+}
+
+func (c *RMQClient) decodeCommandCustomHeader(pr *primitive.PullResult, cmd *remote.RemotingCommand)
{
+	v, exist := cmd.ExtFields["maxOffset"]
 	if exist {
-		pullResult.MaxOffset, _ = strconv.ParseInt(v, 10, 64)
+		pr.MaxOffset, _ = strconv.ParseInt(v, 10, 64)
 	}
 
-	v, exist = response.ExtFields["minOffset"]
+	v, exist = cmd.ExtFields["minOffset"]
 	if exist {
-		pullResult.MinOffset, _ = strconv.ParseInt(v, 10, 64)
+		pr.MinOffset, _ = strconv.ParseInt(v, 10, 64)
 	}
 
-	v, exist = response.ExtFields["nextBeginOffset"]
+	v, exist = cmd.ExtFields["nextBeginOffset"]
 	if exist {
-		pullResult.NextBeginOffset, _ = strconv.ParseInt(v, 10, 64)
+		pr.NextBeginOffset, _ = strconv.ParseInt(v, 10, 64)
 	}
 
-	v, exist = response.ExtFields["suggestWhichBrokerId"]
+	v, exist = cmd.ExtFields["suggestWhichBrokerId"]
 	if exist {
-		pullResult.SuggestWhichBrokerId, _ = strconv.ParseInt(v, 10, 64)
+		pr.SuggestWhichBrokerId, _ = strconv.ParseInt(v, 10, 64)
 	}
-
-	//pullResult.messageExts = decodeMessage(response.Body) TODO parse in top
-
-	return pullResult, nil
 }
 
 // PullMessageAsync pull message async
diff --git a/internal/producer/producer.go b/internal/producer/producer.go
index 91bd389..fbcf5ee 100644
--- a/internal/producer/producer.go
+++ b/internal/producer/producer.go
@@ -40,25 +40,59 @@ type Producer interface {
 	SendOneWay(context.Context, *primitive.Message) error
 }
 
-func NewProducer(opt primitive.ProducerOptions) (Producer, error) {
-	if err := utils.VerifyIP(opt.NameServerAddr); err != nil {
+func NewProducer(nameServerAddr string, opts ...*primitive.ProducerOption) (Producer, error)
{
+	if err := utils.VerifyIP(nameServerAddr); err != nil {
 		return nil, err
 	}
-	if opt.RetryTimesWhenSendFailed == 0 {
-		opt.RetryTimesWhenSendFailed = 2
-	}
-	if opt.NameServerAddr == "" {
-		rlog.Fatal("opt.NameServerAddr can't be empty")
+
+	if nameServerAddr == "" {
+		rlog.Fatal("nameServerAddr can't be empty")
 	}
-	err := os.Setenv(kernel.EnvNameServerAddr, opt.NameServerAddr)
+	err := os.Setenv(kernel.EnvNameServerAddr, nameServerAddr)
 	if err != nil {
 		rlog.Fatal("set env=EnvNameServerAddr error: %s ", err.Error())
 	}
-	return &defaultProducer{
+
+	popts := primitive.DefaultProducerOptions()
+	for _, opt := range opts {
+		opt.Apply(&popts)
+	}
+	popts.NameServerAddr = nameServerAddr
+
+	producer := &defaultProducer{
 		group:   "default",
-		client:  kernel.GetOrNewRocketMQClient(opt.ClientOption),
-		options: opt,
-	}, nil
+		client:  kernel.GetOrNewRocketMQClient(popts.ClientOption),
+		options: popts,
+	}
+
+	chainInterceptor(producer)
+
+	return producer, nil
+}
+
+// chainInterceptor chain list of interceptor as one interceptor
+func chainInterceptor(p *defaultProducer) {
+	interceptors := p.options.Interceptors
+	switch len(interceptors) {
+	case 0:
+		p.interceptor = nil
+	case 1:
+		p.interceptor = interceptors[0]
+	default:
+		p.interceptor = func(ctx context.Context, req, reply interface{}, invoker primitive.PInvoker)
error {
+			return interceptors[0](ctx, req, reply, getChainedInterceptor(interceptors, 0, invoker))
+		}
+	}
+}
+
+// getChainedInterceptor recursively generate the chained invoker.
+func getChainedInterceptor(interceptors []primitive.PInterceptor, cur int, finalInvoker primitive.PInvoker)
primitive.PInvoker {
+	if cur == len(interceptors)-1 {
+		return finalInvoker
+	}
+	return func(ctx context.Context, req, reply interface{}) error {
+		return interceptors[cur+1](ctx, req, reply, getChainedInterceptor(interceptors, cur+1,
finalInvoker))
+	}
 }
 
 type defaultProducer struct {
@@ -67,6 +101,8 @@ type defaultProducer struct {
 	state       kernel.ServiceState
 	options     primitive.ProducerOptions
 	publishInfo sync.Map
+
+	interceptor primitive.PInterceptor
 }
 
 func (p *defaultProducer) Start() error {
@@ -89,11 +125,31 @@ func (p *defaultProducer) SendSync(ctx context.Context, msg *primitive.Message)
 		return nil, errors.New("topic is nil")
 	}
 
+	resp := new(primitive.SendResult)
+	if p.interceptor != nil {
+		primitive.WithMehod(ctx, primitive.SendSync)
+		err := p.interceptor(ctx, msg, resp, func(ctx context.Context, req, reply interface{})
error {
+			var err error
+			realReq := req.(*primitive.Message)
+			realReply := reply.(*primitive.SendResult)
+			err = p.sendSync(ctx, realReq, realReply)
+			return err
+		})
+		return resp, err
+	}
+
+	p.sendSync(ctx, msg, resp)
+	return resp, nil
+}
+
+func (p *defaultProducer) sendSync(ctx context.Context, msg *primitive.Message, resp *primitive.SendResult)
error {
+
 	retryTime := 1 + p.options.RetryTimesWhenSendFailed
 
 	var (
 		err error
 	)
+
 	for retryCount := 0; retryCount < retryTime; retryCount++ {
 		mq := p.selectMessageQueue(msg.Topic)
 		if mq == nil {
@@ -103,7 +159,7 @@ func (p *defaultProducer) SendSync(ctx context.Context, msg *primitive.Message)
 
 		addr := kernel.FindBrokerAddrByName(mq.BrokerName)
 		if addr == "" {
-			return nil, fmt.Errorf("topic=%s route info not found", mq.Topic)
+			return fmt.Errorf("topic=%s route info not found", mq.Topic)
 		}
 
 		res, _err := p.client.InvokeSync(addr, p.buildSendRequest(mq, msg), 3*time.Second)
@@ -111,9 +167,10 @@ func (p *defaultProducer) SendSync(ctx context.Context, msg *primitive.Message)
 			err = _err
 			continue
 		}
-		return p.client.ProcessSendResponse(mq.BrokerName, res, msg), nil
+		p.client.ProcessSendResponse(mq.BrokerName, res, resp, msg)
+		return nil
 	}
-	return nil, err
+	return err
 }
 
 func (p *defaultProducer) SendOneWay(ctx context.Context, msg *primitive.Message) error {
@@ -125,6 +182,17 @@ func (p *defaultProducer) SendOneWay(ctx context.Context, msg *primitive.Message
 		return errors.New("topic is nil")
 	}
 
+	if p.interceptor != nil {
+		primitive.WithMehod(ctx, primitive.SendOneway)
+		return p.interceptor(ctx, msg, nil, func(ctx context.Context, req, reply interface{}) error
{
+			return p.SendOneWay(ctx, msg)
+		})
+	}
+
+	return p.sendOneWay(ctx, msg)
+}
+
+func (p *defaultProducer) sendOneWay(ctx context.Context, msg *primitive.Message) error {
 	retryTime := 1 + p.options.RetryTimesWhenSendFailed
 
 	var (
@@ -151,7 +219,8 @@ func (p *defaultProducer) SendOneWay(ctx context.Context, msg *primitive.Message
 	return err
 }
 
-func (p *defaultProducer) buildSendRequest(mq *primitive.MessageQueue, msg *primitive.Message)
*remote.RemotingCommand {
+func (p *defaultProducer) buildSendRequest(mq *primitive.MessageQueue,
+	msg *primitive.Message) *remote.RemotingCommand {
 	req := &kernel.SendMessageRequest{
 		ProducerGroup:  p.group,
 		Topic:          mq.Topic,
diff --git a/primitive/consume.go b/primitive/consume.go
index 40e89b9..5edd461 100644
--- a/primitive/consume.go
+++ b/primitive/consume.go
@@ -4,7 +4,7 @@ package primitive
 // </p>
 //
 // 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
+// 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.
 // </p>
@@ -126,3 +126,17 @@ const (
 	ConsumeSuccess ConsumeResult = iota
 	ConsumeRetryLater
 )
+
+type ConsumeMessageContext struct {
+	ConsumerGroup string
+	Msgs          []*MessageExt
+	MQ            *MessageQueue
+	Success       bool
+	Status        string
+	// mqTractContext
+	Properties map[string]string
+}
+
+type ConsumeResultHolder struct {
+	ConsumeResult
+}
\ No newline at end of file
diff --git a/primitive/ctx.go b/primitive/ctx.go
new file mode 100644
index 0000000..e74f91a
--- /dev/null
+++ b/primitive/ctx.go
@@ -0,0 +1,60 @@
+/*
+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.
+*/
+
+/*
+ * Define the ctx key and value type.
+ */
+package primitive
+
+import "context"
+
+type CtxKey int
+
+const (
+	method CtxKey = iota
+	msgCtx
+
+
+	// method name in  producer
+	SendSync = "SendSync"
+	SendOneway = "SendOneway"
+	// method name in consumer
+	ConsumerPush = "ConsumerPush"
+	ConsumerPull = "ConsumerPull"
+)
+
+// WithMehod set call method name
+func WithMehod(ctx context.Context, m string) context.Context {
+	return context.WithValue(ctx, method, m)
+}
+
+// GetMethod get call method name
+func GetMethod(ctx context.Context) string {
+	return ctx.Value(method).(string)
+}
+
+// WithConsumerCtx set ConsumeMessageContext in PushConsumer
+func WithConsumerCtx(ctx context.Context, c *ConsumeMessageContext) context.Context {
+	return context.WithValue(ctx, msgCtx, c)
+}
+
+// GetConsumerCtx get ConsumeMessageContext, only legal in PushConsumer. so should add bool
return param indicate
+// whether exist.
+func GetConsumerCtx(ctx context.Context) (*ConsumeMessageContext, bool) {
+	c, exist := ctx.Value(msgCtx).(*ConsumeMessageContext)
+	return c, exist
+}
diff --git a/primitive/interceptor.go b/primitive/interceptor.go
new file mode 100644
index 0000000..cb86d19
--- /dev/null
+++ b/primitive/interceptor.go
@@ -0,0 +1,57 @@
+/*
+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 primitive
+
+import (
+	"context"
+)
+
+// PInvoker finish a send invoke on producer.
+type PInvoker func(ctx context.Context, req, reply interface{}) error
+
+// PInterceptor intercepts the execution of a send invoke on producer.
+type PInterceptor func(ctx context.Context, req, reply interface{}, next PInvoker) error
+
+// RetryInterceptor retry when send failed.
+func RetryPInterceptor() PInterceptor {
+	return func(ctx context.Context, req, reply interface{}, next PInvoker) error {
+		return nil
+	}
+}
+
+// TimeoutInterceptor add a timeout listener in case of operation timeout.
+func TimeoutPInterceptor() PInterceptor {
+	return func(ctx context.Context, req, reply interface{}, next PInvoker) error {
+		return nil
+	}
+}
+
+// LogInterceptor log a send invoke.
+func LogPInterceptor() PInterceptor {
+	return func(ctx context.Context, req, reply interface{}, next PInvoker) error {
+		return nil
+	}
+}
+
+// CInvoker finish a message invoke on consumer. In PushConsumer call, the req is []*MessageExt
type and the reply is *ConsumeResultHolder,
+// use type assert to get real type.
+type CInvoker func(ctx context.Context, req , reply interface{}) error
+
+// CInterceptor intercepts the invoke of a consume on messages. In PushConsumer call, the
req is []*MessageExt type and the reply is *ConsumeResultHolder,
+// use type assert to get real type.
+type CInterceptor func(ctx context.Context, req, reply interface{}, next CInvoker) error
diff --git a/primitive/options.go b/primitive/options.go
index b657ac0..0e40415 100644
--- a/primitive/options.go
+++ b/primitive/options.go
@@ -22,9 +22,13 @@ import (
 	"os"
 	"strconv"
 	"time"
+
+	"github.com/apache/rocketmq-client-go/utils"
 )
 
 type ProducerOptions struct {
+	Interceptors []PInterceptor
+
 	ClientOption
 	NameServerAddr           string
 	GroupName                string
@@ -32,7 +36,48 @@ type ProducerOptions struct {
 	UnitMode                 bool
 }
 
-type ConsumerOption struct {
+func DefaultProducerOptions() ProducerOptions {
+	return ProducerOptions{
+		RetryTimesWhenSendFailed:  2,
+	}
+}
+
+// ProducerOption configures how we create the producer by set ProducerOptions value.
+type ProducerOption struct {
+	Apply func(*ProducerOptions)
+}
+
+func NewProducerOption(f func(options *ProducerOptions)) *ProducerOption {
+	return &ProducerOption{
+		Apply: f,
+	}
+}
+
+// WithProducerInterceptor returns a ProducerOption that specifies the interceptor for producer.
+func WithProducerInterceptor(f PInterceptor) *ProducerOption {
+	return NewProducerOption(func(options *ProducerOptions) {
+		options.Interceptors = append(options.Interceptors, f)
+	})
+}
+
+// WithChainProducerInterceptor returns a ProducerOption that specifies the chained interceptor
for producer.
+// The first interceptor will be the outer most, while the last interceptor will be the inner
most wrapper
+// around the real call.
+func WithChainProducerInterceptor(fs ...PInterceptor) *ProducerOption {
+	return NewProducerOption(func(options *ProducerOptions) {
+		options.Interceptors = append(options.Interceptors, fs...)
+	})
+}
+
+// WithRetry return a ProducerOption that specifies the retry times when send failed.
+// TODO: use retryMiddleeware instead.
+func WithRetry(retries int) *ProducerOption {
+	return  NewProducerOption(func(options *ProducerOptions) {
+		options.RetryTimesWhenSendFailed = retries
+	})
+}
+
+type ConsumerOptions struct {
 	ClientOption
 	NameServerAddr string
 
@@ -92,7 +137,7 @@ type ConsumerOption struct {
 
 	// Max re-consume times. -1 means 16 times.
 	//
-	// If messages are re-consumed more than {@link #maxReconsumeTimes} before success, it's
be directed to a deletion
+	// If messages are re-consumed more than {@link #maxReconsumeTimes} before Success, it's
be directed to a deletion
 	// queue waiting.
 	MaxReconsumeTimes int
 
@@ -107,6 +152,56 @@ type ConsumerOption struct {
 	ConsumeOrderly bool
 	FromWhere      ConsumeFromWhere
 	// TODO traceDispatcher
+
+	Interceptors []CInterceptor
+}
+
+func DefaultPushConsumerOptions() ConsumerOptions{
+	return ConsumerOptions{
+		ClientOption: ClientOption{
+			InstanceName: "DEFAULT",
+			ClientIP: utils.LocalIP(),
+		},
+		Strategy: AllocateByAveragely,
+	}
+}
+
+type ConsumerOption struct {
+	Apply func(*ConsumerOptions)
+}
+
+func NewConsumerOption(f func(*ConsumerOptions)) *ConsumerOption {
+	return &ConsumerOption{
+		Apply: f,
+	}
+}
+
+func WithConsumerModel(m MessageModel) *ConsumerOption {
+	return NewConsumerOption(func(options *ConsumerOptions) {
+		options.ConsumerModel = m
+	})
+}
+
+func WithConsumeFromWhere(w ConsumeFromWhere) *ConsumerOption{
+	return NewConsumerOption(func(options *ConsumerOptions) {
+		options.FromWhere = w
+	})
+}
+
+// WithConsumerInterceptor returns a ConsumerOption that specifies the interceptor for consumer.
+func WithConsumerInterceptor(f CInterceptor) *ConsumerOption {
+	return NewConsumerOption(func(options *ConsumerOptions) {
+		options.Interceptors = append(options.Interceptors, f)
+	})
+}
+
+// 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 WithChainConsumerInterceptor(fs ...CInterceptor) *ConsumerOption {
+	return NewConsumerOption(func(options *ConsumerOptions) {
+		options.Interceptors = append(options.Interceptors, fs...)
+	})
 }
 
 func (opt *ClientOption) ChangeInstanceNameToPID() {
diff --git a/primitive/result.go b/primitive/result.go
index c37358c..628f243 100644
--- a/primitive/result.go
+++ b/primitive/result.go
@@ -56,10 +56,10 @@ func (result *SendResult) String() string {
 		result.Status, result.MsgID, result.OffsetMsgID, result.QueueOffset, result.MessageQueue.String())
 }
 
-// PullStatus pull status
+// PullStatus pull Status
 type PullStatus int
 
-// predefined pull status
+// predefined pull Status
 const (
 	PullFound PullStatus = iota
 	PullNoNewMsg
@@ -75,7 +75,11 @@ type PullResult struct {
 	MaxOffset            int64
 	Status               PullStatus
 	SuggestWhichBrokerId int64
+
+	// messageExts message info
 	messageExts          []*MessageExt
+	//
+	body []byte
 }
 
 func (result *PullResult) GetMessageExts() []*MessageExt {
@@ -93,11 +97,19 @@ func (result *PullResult) GetMessages() []*Message {
 	return toMessages(result.messageExts)
 }
 
+func (result *PullResult) SetBody(data []byte) {
+	result.body = data
+}
+
+func (result *PullResult) GetBody() []byte {
+	return result.body
+}
+
 func (result *PullResult) String() string {
 	return ""
 }
 
-func decodeMessage(data []byte) []*MessageExt {
+func DecodeMessage(data []byte) []*MessageExt {
 	msgs := make([]*MessageExt, 0)
 	buf := bytes.NewBuffer(data)
 	count := 0


Mime
View raw message