You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by ji...@apache.org on 2026/04/28 02:50:01 UTC
(rocketmq-clients) branch master updated: Go client supports Producer with recalling timed/delay messages and Priority Message. (#1231)
This is an automated email from the ASF dual-hosted git repository.
RongtongJin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/rocketmq-clients.git
The following commit(s) were added to refs/heads/master by this push:
new 9cde1945 Go client supports Producer with recalling timed/delay messages and Priority Message. (#1231)
9cde1945 is described below
commit 9cde19453360302111574dba227945005700105d
Author: zhaohai <33...@users.noreply.github.com>
AuthorDate: Tue Apr 28 10:49:55 2026 +0800
Go client supports Producer with recalling timed/delay messages and Priority Message. (#1231)
* Go client supports Producer with recalling timed/delay messages and Priority Message.
* Go client supports Producer with recalling timed/delay messages and Priority Message.
---
README-CN.md | 4 +-
README.md | 4 +-
golang/client_manager.go | 12 ++
golang/client_manager_mock.go | 21 ++++
golang/example/producer/delay_recall/main.go | 128 +++++++++++++++++++++
golang/example/producer/priority/main.go | 163 +++++++++++++++++++++++++++
golang/lite_push_consumer_test.go | 4 +
golang/message.go | 26 +++++
golang/producer.go | 47 ++++++++
golang/publishing_message.go | 16 ++-
golang/rpc_client.go | 10 ++
golang/rpc_client_mock.go | 15 +++
protos | 2 +-
13 files changed, 443 insertions(+), 9 deletions(-)
diff --git a/README-CN.md b/README-CN.md
index 4a4957f9..b5b715a2 100644
--- a/README-CN.md
+++ b/README-CN.md
@@ -23,12 +23,12 @@
| Producer with FIFO messages | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | 🚧 |
| Producer with timed/delay messages | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | 🚧 |
| Producer with transactional messages | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | 🚧 |
-| Producer with recalling timed/delay messages | ✅ | ✅ | 🚧 | 🚧 | 🚧 | ✅ | ✅ | 🚧 |
+| Producer with recalling timed/delay messages | ✅ | ✅ | 🚧 | ✅ | 🚧 | ✅ | ✅ | 🚧 |
| Simple consumer | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | 🚧 |
| Push consumer with concurrent message listener | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | 🚧 |
| Push consumer with FIFO message listener | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | 🚧 |
| Push consumer with FIFO consume accelerator | ✅ | ✅ | 🚧 | ✅ | 🚧 | ✅ | ✅ | 🚧 |
-| Priority Message | ✅ | 🚧 | 🚧 | 🚧 | 🚧 | ✅ | ✅ | 🚧 |
+| Priority Message | ✅ | 🚧 | 🚧 | ✅ | 🚧 | ✅ | ✅ | 🚧 |
## 先决条件和构建
diff --git a/README.md b/README.md
index ce019d2c..0d064685 100644
--- a/README.md
+++ b/README.md
@@ -23,12 +23,12 @@ Provide cloud-native and robust solutions for Java, C++, C#, Golang, Rust and al
| Producer with FIFO messages | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | 🚧 |
| Producer with timed/delay messages | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | 🚧 |
| Producer with transactional messages | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | 🚧 |
-| Producer with recalling timed/delay messages | ✅ | ✅ | 🚧 | 🚧 | 🚧 | ✅ | ✅ | 🚧 |
+| Producer with recalling timed/delay messages | ✅ | ✅ | 🚧 | ✅ | 🚧 | ✅ | ✅ | 🚧 |
| Simple consumer | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | 🚧 |
| Push consumer with concurrent message listener | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | 🚧 |
| Push consumer with FIFO message listener | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | 🚧 |
| Push consumer with FIFO consume accelerator | ✅ | ✅ | 🚧 | ✅ | 🚧 | ✅ | ✅ | 🚧 |
-| Priority Message | ✅ | 🚧 | 🚧 | 🚧 | 🚧 | ✅ | ✅ | 🚧 |
+| Priority Message | ✅ | 🚧 | 🚧 | ✅ | 🚧 | ✅ | ✅ | 🚧 |
## Prerequisite and Build
diff --git a/golang/client_manager.go b/golang/client_manager.go
index 8ba7db92..0c395c4b 100644
--- a/golang/client_manager.go
+++ b/golang/client_manager.go
@@ -45,6 +45,7 @@ type ClientManager interface {
ChangeInvisibleDuration(ctx context.Context, endpoints *v2.Endpoints, request *v2.ChangeInvisibleDurationRequest, duration time.Duration) (*v2.ChangeInvisibleDurationResponse, error)
ForwardMessageToDeadLetterQueue(ctx context.Context, endpoints *v2.Endpoints, request *v2.ForwardMessageToDeadLetterQueueRequest, duration time.Duration) (*v2.ForwardMessageToDeadLetterQueueResponse, error)
SyncLiteSubscription(ctx context.Context, endpoints *v2.Endpoints, request *v2.SyncLiteSubscriptionRequest, duration time.Duration) (*v2.SyncLiteSubscriptionResponse, error)
+ RecallMessage(ctx context.Context, endpoints *v2.Endpoints, request *v2.RecallMessageRequest, duration time.Duration) (*v2.RecallMessageResponse, error)
}
type clientManagerOptions struct {
@@ -361,3 +362,14 @@ func (cm *defaultClientManager) SyncLiteSubscription(ctx context.Context, endpoi
cm.handleGrpcError(rpcClient, err)
return ret, err
}
+
+func (cm *defaultClientManager) RecallMessage(ctx context.Context, endpoints *v2.Endpoints, request *v2.RecallMessageRequest, duration time.Duration) (*v2.RecallMessageResponse, error) {
+ ctx, _ = context.WithTimeout(ctx, duration)
+ rpcClient, err := cm.getRpcClient(endpoints)
+ if err != nil {
+ return nil, err
+ }
+ ret, err := rpcClient.RecallMessage(ctx, request)
+ cm.handleGrpcError(rpcClient, err)
+ return ret, err
+}
diff --git a/golang/client_manager_mock.go b/golang/client_manager_mock.go
index a40f5388..70daf72b 100644
--- a/golang/client_manager_mock.go
+++ b/golang/client_manager_mock.go
@@ -242,6 +242,27 @@ func (m *MockClientManager) SyncLiteSubscription(ctx context.Context, endpoints
return ret0, ret1
}
+// SyncLiteSubscription indicates an expected call of SyncLiteSubscription.
+func (mr *MockClientManagerMockRecorder) SyncLiteSubscription(ctx, endpoints, request, duration interface{}) *gomock.Call {
+ mr.mock.ctrl.T.Helper()
+ return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SyncLiteSubscription", reflect.TypeOf((*MockClientManager)(nil).SyncLiteSubscription), ctx, endpoints, request, duration)
+}
+
+// RecallMessage mocks base method.
+func (m *MockClientManager) RecallMessage(ctx context.Context, endpoints *v2.Endpoints, request *v2.RecallMessageRequest, duration time.Duration) (*v2.RecallMessageResponse, error) {
+ m.ctrl.T.Helper()
+ ret := m.ctrl.Call(m, "RecallMessage", ctx, endpoints, request, duration)
+ ret0, _ := ret[0].(*v2.RecallMessageResponse)
+ ret1, _ := ret[1].(error)
+ return ret0, ret1
+}
+
+// RecallMessage indicates an expected call of RecallMessage.
+func (mr *MockClientManagerMockRecorder) RecallMessage(ctx, endpoints, request, duration interface{}) *gomock.Call {
+ mr.mock.ctrl.T.Helper()
+ return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RecallMessage", reflect.TypeOf((*MockClientManager)(nil).RecallMessage), ctx, endpoints, request, duration)
+}
+
// UnRegisterClient indicates an expected call of UnRegisterClient.
func (mr *MockClientManagerMockRecorder) UnRegisterClient(client interface{}) *gomock.Call {
mr.mock.ctrl.T.Helper()
diff --git a/golang/example/producer/delay_recall/main.go b/golang/example/producer/delay_recall/main.go
new file mode 100644
index 00000000..cb7d5253
--- /dev/null
+++ b/golang/example/producer/delay_recall/main.go
@@ -0,0 +1,128 @@
+/*
+ * 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"
+ "log"
+ "os"
+ "time"
+
+ rmq_client "github.com/apache/rocketmq-clients/golang/v5"
+ "github.com/apache/rocketmq-clients/golang/v5/credentials"
+)
+
+const (
+ Topic = "topic-delay-new"
+ Endpoint = "127.0.0.1:8080"
+ AccessKey = ""
+ SecretKey = ""
+)
+
+func main() {
+ os.Setenv("mq.consoleAppender.enabled", "true")
+ rmq_client.ResetLogger()
+
+ // In most case, you don't need to create many producers, singleton pattern is more recommended.
+ producer, err := rmq_client.NewProducer(&rmq_client.Config{
+ Endpoint: Endpoint,
+ Credentials: &credentials.SessionCredentials{
+ AccessKey: AccessKey,
+ AccessSecret: SecretKey,
+ },
+ },
+ rmq_client.WithTopics(Topic),
+ )
+ if err != nil {
+ log.Fatal(err)
+ }
+
+ // start producer
+ err = producer.Start()
+ if err != nil {
+ log.Fatal(err)
+ }
+ // graceful stop producer
+ defer producer.GracefulStop()
+
+ fmt.Println("========== Delay Message Recall Example ==========")
+ fmt.Printf("Topic: %s\n", Topic)
+ fmt.Printf("Endpoint: %s\n\n", Endpoint)
+
+ // Send delay messages with recall capability
+ fmt.Println("Sending delay messages that can be recalled...")
+ fmt.Println("Messages will be delivered after 10 minutes if not recalled\n")
+
+ var recallHandles []string
+
+ for i := 1; i <= 3; i++ {
+ delayMsg := &rmq_client.Message{
+ Topic: Topic,
+ Body: []byte(fmt.Sprintf("Delay message %d - scheduled for delivery at %s", i, time.Now().Add(10*time.Minute).Format("15:04:05"))),
+ }
+
+ // Set delivery timestamp to 10 minutes later
+ deliveryTime := time.Now().Add(10 * time.Minute)
+ delayMsg.SetDelayTimestamp(deliveryTime)
+
+ receipts, err := producer.Send(context.Background(), delayMsg)
+ if err != nil {
+ log.Printf("Failed to send delay message %d: %v", i, err)
+ } else {
+ fmt.Printf("✓ Delay message %d sent:\n", i)
+ fmt.Printf(" MessageID: %s\n", receipts[0].MessageID)
+ fmt.Printf(" RecallHandle: %s\n", receipts[0].RecallHandle)
+ fmt.Printf(" Delivery Time: %s\n", deliveryTime.Format("15:04:05"))
+
+ if receipts[0].RecallHandle != "" {
+ recallHandles = append(recallHandles, receipts[0].RecallHandle)
+ }
+ }
+
+ time.Sleep(200 * time.Millisecond)
+ }
+
+ fmt.Println("\n========== All delay messages sent ==========")
+
+ // Recall the first delay message
+ if len(recallHandles) > 0 {
+ fmt.Println("\nWaiting 3 seconds before recalling the first message...")
+ time.Sleep(3 * time.Second)
+
+ fmt.Printf("\nAttempting to recall message 1...\n")
+ recalledMessageID, err := producer.Recall(context.Background(), Topic, recallHandles[0])
+ if err != nil {
+ fmt.Printf("✗ Failed to recall message: %v\n", err)
+ fmt.Println("\nPossible reasons:")
+ fmt.Println(" - The message has already been delivered")
+ fmt.Println(" - The recall_handle is invalid")
+ fmt.Println(" - Network issues")
+ } else {
+ fmt.Printf("✓ Message recalled successfully!\n")
+ fmt.Printf(" Recalled MessageID: %s\n", recalledMessageID)
+ }
+ }
+
+ fmt.Println("\n========== Test Complete ==========")
+ fmt.Println("\nNext steps:")
+ fmt.Println("1. Start consumer 'GID-normal-consumer_topic-normal' to receive remaining messages")
+ fmt.Println("2. Messages should be delivered ~10 minutes after sending")
+ fmt.Println("3. Message 1 was recalled and should NOT be delivered")
+ fmt.Println("4. Messages 2 and 3 should be delivered on schedule")
+}
diff --git a/golang/example/producer/priority/main.go b/golang/example/producer/priority/main.go
new file mode 100644
index 00000000..8ce23d13
--- /dev/null
+++ b/golang/example/producer/priority/main.go
@@ -0,0 +1,163 @@
+/*
+ * 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"
+ "log"
+ "os"
+ "time"
+
+ rmq_client "github.com/apache/rocketmq-clients/golang/v5"
+ "github.com/apache/rocketmq-clients/golang/v5/credentials"
+)
+
+const (
+ Topic = "topic-priority"
+ Endpoint = "127.0.0.1:8080"
+ AccessKey = ""
+ SecretKey = ""
+)
+
+func main() {
+ os.Setenv("mq.consoleAppender.enabled", "true")
+ rmq_client.ResetLogger()
+
+ // In most case, you don't need to create many producers, singleton pattern is more recommended.
+ producer, err := rmq_client.NewProducer(&rmq_client.Config{
+ Endpoint: Endpoint,
+ Credentials: &credentials.SessionCredentials{
+ AccessKey: AccessKey,
+ AccessSecret: SecretKey,
+ },
+ },
+ rmq_client.WithTopics(Topic),
+ )
+ if err != nil {
+ log.Fatal(err)
+ }
+
+ // start producer
+ err = producer.Start()
+ if err != nil {
+ log.Fatal(err)
+ }
+ // graceful stop producer
+ defer producer.GracefulStop()
+
+ fmt.Println("========== Priority Message Producer Example ==========")
+ fmt.Printf("Topic: %s\n", Topic)
+ fmt.Printf("Endpoint: %s\n\n", Endpoint)
+
+ // Send priority messages with different levels
+ fmt.Println("Sending priority messages...")
+ fmt.Println("Note: Higher priority value = higher priority\n")
+
+ // Low priority message (priority = 1)
+ lowPriorityMsg := &rmq_client.Message{
+ Topic: Topic,
+ Body: []byte("This is a low priority message"),
+ }
+ err = lowPriorityMsg.SetPriority(1)
+ if err != nil {
+ log.Printf("Failed to set priority: %v", err)
+ } else {
+ receipts, err := producer.Send(context.Background(), lowPriorityMsg)
+ if err != nil {
+ log.Printf("Failed to send low priority message: %v", err)
+ } else {
+ fmt.Printf("✓ Low priority (1) message sent, MessageID: %s\n", receipts[0].MessageID)
+ }
+ }
+
+ // Medium priority message (priority = 5)
+ mediumPriorityMsg := &rmq_client.Message{
+ Topic: Topic,
+ Body: []byte("This is a medium priority message"),
+ }
+ err = mediumPriorityMsg.SetPriority(5)
+ if err != nil {
+ log.Printf("Failed to set priority: %v", err)
+ } else {
+ receipts, err := producer.Send(context.Background(), mediumPriorityMsg)
+ if err != nil {
+ log.Printf("Failed to send medium priority message: %v", err)
+ } else {
+ fmt.Printf("✓ Medium priority (5) message sent, MessageID: %s\n", receipts[0].MessageID)
+ }
+ }
+
+ // High priority message (priority = 10)
+ highPriorityMsg := &rmq_client.Message{
+ Topic: Topic,
+ Body: []byte("This is a high priority message"),
+ }
+ err = highPriorityMsg.SetPriority(10)
+ if err != nil {
+ log.Printf("Failed to set priority: %v", err)
+ } else {
+ receipts, err := producer.Send(context.Background(), highPriorityMsg)
+ if err != nil {
+ log.Printf("Failed to send high priority message: %v", err)
+ } else {
+ fmt.Printf("✓ High priority (10) message sent, MessageID: %s\n", receipts[0].MessageID)
+ }
+ }
+
+ // Demonstrate mutual exclusivity constraints
+ fmt.Println("\nDemonstrating mutual exclusivity constraints...")
+
+ // This will fail: priority + messageGroup
+ fifoMsg := &rmq_client.Message{
+ Topic: Topic,
+ Body: []byte("test"),
+ }
+ fifoMsg.SetMessageGroup("group-1")
+ err = fifoMsg.SetPriority(5)
+ if err != nil {
+ fmt.Printf("✓ Correctly rejected: priority + messageGroup - %v\n", err)
+ }
+
+ // This will fail: priority + deliveryTimestamp
+ delayWithPriorityMsg := &rmq_client.Message{
+ Topic: Topic,
+ Body: []byte("test"),
+ }
+ delayWithPriorityMsg.SetDelayTimestamp(time.Now().Add(1 * time.Minute))
+ err = delayWithPriorityMsg.SetPriority(5)
+ if err != nil {
+ fmt.Printf("✓ Correctly rejected: priority + deliveryTimestamp - %v\n", err)
+ }
+
+ // This will fail: negative priority
+ negativePriorityMsg := &rmq_client.Message{
+ Topic: Topic,
+ Body: []byte("test"),
+ }
+ err = negativePriorityMsg.SetPriority(-1)
+ if err != nil {
+ fmt.Printf("✓ Correctly rejected: negative priority - %v\n", err)
+ }
+
+ fmt.Println("\n========== All validation checks passed! ==========")
+ fmt.Println("\nExpected behavior:")
+ fmt.Println("- Messages with higher priority values are delivered first")
+ fmt.Println("- Priority 10 > Priority 5 > Priority 1")
+ fmt.Println("- Use a consumer to verify the delivery order")
+}
diff --git a/golang/lite_push_consumer_test.go b/golang/lite_push_consumer_test.go
index 953159f0..a084794d 100644
--- a/golang/lite_push_consumer_test.go
+++ b/golang/lite_push_consumer_test.go
@@ -177,6 +177,10 @@ func (m *mockedClientManager) SyncLiteSubscription(ctx context.Context, endpoint
return m.mockRpcClient.SyncLiteSubscription(ctx, request)
}
+func (m *mockedClientManager) RecallMessage(ctx context.Context, endpoints *v2.Endpoints, request *v2.RecallMessageRequest, duration time.Duration) (*v2.RecallMessageResponse, error) {
+ return nil, nil
+}
+
func TestLitePushConsumer_SubscribeLite(t *testing.T) {
setupTest(t)
defer teardownTest()
diff --git a/golang/message.go b/golang/message.go
index fa64136d..74c21e24 100644
--- a/golang/message.go
+++ b/golang/message.go
@@ -21,6 +21,7 @@ import (
"crypto/md5"
"crypto/sha1"
"encoding/hex"
+ "fmt"
"hash/crc32"
"strconv"
"strings"
@@ -79,6 +80,7 @@ type Message struct {
LiteTopic *string
deliveryTimestamp *time.Time
+ priority *int32
parentTraceContext *string
}
@@ -87,6 +89,7 @@ type SendReceipt struct {
TransactionId string
Offset int64
Endpoints *v2.Endpoints
+ RecallHandle string
}
func (msg *Message) SetTag(tag string) {
@@ -136,6 +139,27 @@ func (msg *Message) GetDeliveryTimestamp() *time.Time {
return msg.deliveryTimestamp
}
+func (msg *Message) SetPriority(priority int32) error {
+ if priority < 0 {
+ return fmt.Errorf("priority must be greater than or equal to 0")
+ }
+ if msg.messageGroup != nil {
+ return fmt.Errorf("priority and messageGroup should not be set at same time")
+ }
+ if msg.deliveryTimestamp != nil {
+ return fmt.Errorf("priority and deliveryTimestamp should not be set at same time")
+ }
+ if msg.LiteTopic != nil {
+ return fmt.Errorf("priority and liteTopic should not be set at same time")
+ }
+ msg.priority = &priority
+ return nil
+}
+
+func (msg *Message) GetPriority() *int32 {
+ return msg.priority
+}
+
func (msg *Message) SetMessageGroup(messageGroup string) {
msg.messageGroup = &messageGroup
}
@@ -151,6 +175,7 @@ func (msg *Message) GetMessageCommon() *MessageCommon {
tag: msg.Tag,
messageGroup: msg.messageGroup,
deliveryTimestamp: msg.deliveryTimestamp,
+ priority: msg.priority,
parentTraceContext: msg.parentTraceContext,
keys: msg.keys,
properties: msg.properties,
@@ -166,6 +191,7 @@ type MessageCommon struct {
keys []string
messageGroup *string
deliveryTimestamp *time.Time
+ priority *int32
bornHost *string
parentTraceContext *string
traceContext *string
diff --git a/golang/producer.go b/golang/producer.go
index 74207173..ab3232c2 100644
--- a/golang/producer.go
+++ b/golang/producer.go
@@ -35,6 +35,7 @@ type Producer interface {
Send(context.Context, *Message) ([]*SendReceipt, error)
SendWithTransaction(context.Context, *Message, Transaction) ([]*SendReceipt, error)
SendAsync(context.Context, *Message, func(context.Context, []*SendReceipt, error))
+ Recall(ctx context.Context, topic string, recallHandle string) (string, error)
BeginTransaction() Transaction
Start() error
GracefulStop() error
@@ -257,6 +258,7 @@ func (p *defaultProducer) send1(ctx context.Context, topic string, messageType v
TransactionId: resp.GetEntries()[i].GetTransactionId(),
Offset: resp.GetEntries()[i].GetOffset(),
Endpoints: endpoints,
+ RecallHandle: resp.GetEntries()[i].GetRecallHandle(),
})
}
if attempt > 1 {
@@ -353,6 +355,51 @@ func (p *defaultProducer) SendAsync(ctx context.Context, msg *Message, f func(co
}()
}
+func (p *defaultProducer) Recall(ctx context.Context, topic string, recallHandle string) (string, error) {
+ if !p.isOn() {
+ return "", fmt.Errorf("producer is not running")
+ }
+ if recallHandle == "" {
+ return "", fmt.Errorf("recall handle is invalid")
+ }
+
+ ctx = p.cli.Sign(ctx)
+ request := &v2.RecallMessageRequest{
+ Topic: &v2.Resource{
+ Name: topic,
+ ResourceNamespace: p.cli.config.NameSpace,
+ },
+ RecallHandle: recallHandle,
+ }
+
+ // Get endpoints for the topic
+ endpoints := &v2.Endpoints{}
+ p.pSetting.topics.Range(func(key, value interface{}) bool {
+ if key == topic {
+ route, err := p.cli.getMessageQueues(ctx, topic)
+ if err == nil && len(route) > 0 {
+ endpoints = route[0].GetBroker().GetEndpoints()
+ }
+ return false
+ }
+ return true
+ })
+
+ resp, err := p.cli.clientManager.RecallMessage(ctx, endpoints, request, p.pSetting.GetRequestTimeout())
+ if err != nil {
+ return "", err
+ }
+
+ if resp.GetStatus().GetCode() != v2.Code_OK {
+ return "", &ErrRpcStatus{
+ Code: int32(resp.Status.GetCode()),
+ Message: resp.GetStatus().GetMessage(),
+ }
+ }
+
+ return resp.GetMessageId(), nil
+}
+
func (p *defaultProducer) SendWithTransaction(ctx context.Context, msg *Message, transaction Transaction) ([]*SendReceipt, error) {
if !p.isOn() {
return nil, fmt.Errorf("producer is not running")
diff --git a/golang/publishing_message.go b/golang/publishing_message.go
index 631dba7e..3b601b5b 100644
--- a/golang/publishing_message.go
+++ b/golang/publishing_message.go
@@ -57,7 +57,7 @@ var NewPublishingMessage = func(msg *Message, namespace string, settings *produc
// Generate message id.
pMsg.messageId = GetMessageIdCodecInstance().NextMessageId().String()
// Normal message.
- if msg.GetMessageGroup() == nil && msg.GetDeliveryTimestamp() == nil && !txEnabled && msg.GetLiteTopic() == nil {
+ if msg.GetMessageGroup() == nil && msg.GetDeliveryTimestamp() == nil && msg.GetPriority() == nil && !txEnabled && msg.GetLiteTopic() == nil {
pMsg.messageType = v2.MessageType_NORMAL
return pMsg, nil
}
@@ -71,18 +71,23 @@ var NewPublishingMessage = func(msg *Message, namespace string, settings *produc
pMsg.messageType = v2.MessageType_DELAY
return pMsg, nil
}
+ // Priority message.
+ if msg.GetPriority() != nil && !txEnabled {
+ pMsg.messageType = v2.MessageType_PRIORITY
+ return pMsg, nil
+ }
// Lite message.
if msg.GetLiteTopic() != nil && !txEnabled {
pMsg.messageType = v2.MessageType_LITE
return pMsg, nil
}
// Transaction message.
- if msg.GetMessageGroup() == nil && msg.GetDeliveryTimestamp() == nil && txEnabled {
+ if msg.GetMessageGroup() == nil && msg.GetDeliveryTimestamp() == nil && msg.GetPriority() == nil && txEnabled {
pMsg.messageType = v2.MessageType_TRANSACTION
return pMsg, nil
}
- // Transaction semantics is conflicted with fifo/delay.
- return nil, fmt.Errorf("transactional message should not set messageGroup or deliveryTimestamp")
+ // Transaction semantics is conflicted with fifo/delay/priority/lite.
+ return nil, fmt.Errorf("transactional message should not set messageGroup, deliveryTimestamp, priority or liteTopic")
}
func (pMsg *PublishingMessage) toProtobuf() (*v2.Message, error) {
@@ -121,5 +126,8 @@ func (pMsg *PublishingMessage) toProtobuf() (*v2.Message, error) {
if pMsg.msg.LiteTopic != nil {
msg.SystemProperties.LiteTopic = pMsg.msg.LiteTopic
}
+ if pMsg.msg.priority != nil {
+ msg.SystemProperties.Priority = pMsg.msg.priority
+ }
return msg, nil
}
diff --git a/golang/rpc_client.go b/golang/rpc_client.go
index 8c0f37ec..f4b5a7d2 100644
--- a/golang/rpc_client.go
+++ b/golang/rpc_client.go
@@ -46,6 +46,7 @@ type RpcClient interface {
ChangeInvisibleDuration(ctx context.Context, request *v2.ChangeInvisibleDurationRequest) (*v2.ChangeInvisibleDurationResponse, error)
ForwardMessageToDeadLetterQueue(ctx context.Context, request *v2.ForwardMessageToDeadLetterQueueRequest) (*v2.ForwardMessageToDeadLetterQueueResponse, error)
SyncLiteSubscription(ctx context.Context, request *v2.SyncLiteSubscriptionRequest) (*v2.SyncLiteSubscriptionResponse, error)
+ RecallMessage(ctx context.Context, request *v2.RecallMessageRequest) (*v2.RecallMessageResponse, error)
idleDuration() time.Duration
GetTarget() string
}
@@ -211,3 +212,12 @@ func (rc *rpcClient) SyncLiteSubscription(ctx context.Context, request *v2.SyncL
sugarBaseLogger.Debugf("SyncLiteSubscription request: %v, response: %v, err: %v", request, resp, err)
return resp, err
}
+
+func (rc *rpcClient) RecallMessage(ctx context.Context, request *v2.RecallMessageRequest) (*v2.RecallMessageResponse, error) {
+ rc.mux.Lock()
+ rc.activityNanoTime = time.Now()
+ rc.mux.Unlock()
+ resp, err := rc.msc.RecallMessage(ctx, request)
+ sugarBaseLogger.Debugf("recallMessage request: %v, response: %v, err: %v", request, resp, err)
+ return resp, err
+}
diff --git a/golang/rpc_client_mock.go b/golang/rpc_client_mock.go
index 95177e2f..58bd7734 100644
--- a/golang/rpc_client_mock.go
+++ b/golang/rpc_client_mock.go
@@ -229,6 +229,21 @@ func (mr *MockRpcClientMockRecorder) SyncLiteSubscription(ctx, request interface
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SyncLiteSubscription", reflect.TypeOf((*MockRpcClient)(nil).SyncLiteSubscription), ctx, request)
}
+// RecallMessage mocks base method.
+func (m *MockRpcClient) RecallMessage(ctx context.Context, request *v2.RecallMessageRequest) (*v2.RecallMessageResponse, error) {
+ m.ctrl.T.Helper()
+ ret := m.ctrl.Call(m, "RecallMessage", ctx, request)
+ ret0, _ := ret[0].(*v2.RecallMessageResponse)
+ ret1, _ := ret[1].(error)
+ return ret0, ret1
+}
+
+// RecallMessage indicates an expected call of RecallMessage.
+func (mr *MockRpcClientMockRecorder) RecallMessage(ctx, request interface{}) *gomock.Call {
+ mr.mock.ctrl.T.Helper()
+ return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RecallMessage", reflect.TypeOf((*MockRpcClient)(nil).RecallMessage), ctx, request)
+}
+
// Telemetry mocks base method.
func (m *MockRpcClient) Telemetry(ctx context.Context) (v2.MessagingService_TelemetryClient, error) {
m.ctrl.T.Helper()
diff --git a/protos b/protos
index 68c2cc94..5c9f8419 160000
--- a/protos
+++ b/protos
@@ -1 +1 @@
-Subproject commit 68c2cc9442928f769f8938515a05af6fa05c9993
+Subproject commit 5c9f84199bffa79b2ed73beb37774ca92e749c19