[ISSUE #891] Implement message recalling API in C# SDK

Author: tsunghanjacktsai

csharp/rocketmq-client-csharp/ClientManager.cs

@@ -120,6 +120,15 @@ public async Task Shutdown()
                 request, response, metadata);
         }
 
+        public async Task<RpcInvocation<Proto.RecallMessageRequest, Proto.RecallMessageResponse>>
+            RecallMessage(Endpoints endpoints, Proto.RecallMessageRequest request, TimeSpan timeout)
+        {
+            var metadata = _client.Sign();
+            var response = await GetRpcClient(endpoints).RecallMessage(metadata, request, timeout);
+            return new RpcInvocation<Proto.RecallMessageRequest, Proto.RecallMessageResponse>(
+                request, response, metadata);
+        }
+
         public async Task<RpcInvocation<Proto.SendMessageRequest, Proto.SendMessageResponse>> SendMessage(
             Endpoints endpoints, Proto::SendMessageRequest request, TimeSpan timeout)
         {

csharp/rocketmq-client-csharp/IClientManager.cs

@@ -62,6 +62,16 @@ Task<RpcInvocation<HeartbeatRequest, HeartbeatResponse>> Heartbeat(Endpoints end
         Task<RpcInvocation<NotifyClientTerminationRequest, NotifyClientTerminationResponse>> NotifyClientTermination(
             Endpoints endpoints, NotifyClientTerminationRequest request, TimeSpan timeout);
 
+        /// <summary>
+        /// Recall messages.
+        /// </summary>
+        /// <param name="endpoints">The target endpoints.</param>
+        /// <param name="request">gRPC request of recalling messages.</param>
+        /// <param name="timeout">Request max duration.</param>
+        /// <returns>Task of response.</returns>
+        Task<RpcInvocation<RecallMessageRequest, RecallMessageResponse>> RecallMessage(
+            Endpoints endpoints, RecallMessageRequest request, TimeSpan timeout);
+
         /// <summary>
         /// Send message to remote endpoints.
         /// </summary>

csharp/rocketmq-client-csharp/IRecallReceipt.cs

@@ -0,0 +1,24 @@
+/*
+ * 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.
+ */
+
+namespace Org.Apache.Rocketmq
+{
+    public interface IRecallReceipt
+    {
+        string MessageId { get; }
+    }
+}

csharp/rocketmq-client-csharp/IRpcClient.cs

@@ -52,6 +52,8 @@ Task<ForwardMessageToDeadLetterQueueResponse> ForwardMessageToDeadLetterQueue(Me
         Task<NotifyClientTerminationResponse> NotifyClientTermination(Metadata metadata,
             NotifyClientTerminationRequest request, TimeSpan timeout);
 
+        Task<RecallMessageResponse> RecallMessage(Metadata metadata, RecallMessageRequest request, TimeSpan timeout);
+
         Task Shutdown();
     }
 }

csharp/rocketmq-client-csharp/ISendReceipt.cs

@@ -20,5 +20,6 @@ namespace Org.Apache.Rocketmq
     public interface ISendReceipt
     {
         string MessageId { get; }
+        string RecallHandle { get; }
     }
 }

csharp/rocketmq-client-csharp/Producer.cs

@@ -342,6 +342,33 @@ internal async Task EndTransaction(Endpoints endpoints, string topic, string mes
             StatusChecker.Check(invocation.Response.Status, request, invocation.RequestId);
         }
 
+        public async Task<IRecallReceipt> RecallMessage(string topic, string recallhandle)
+        {
+            var recallReceipt = await RecallMessage0(topic, recallhandle);
+            return recallReceipt;
+        }
+
+        private async Task<RecallReceipt> RecallMessage0(string topic, string recallhandle)
+        {
+            if (State.Running != State)
+            {
+                throw new InvalidOperationException("Producer is not running");
+            }
+            if (recallhandle == null)
+            {
+                throw new InvalidOperationException("Recall handle is invalid");
+            }
+            var request = new Proto.RecallMessageRequest
+            {
+                Topic = new Proto.Resource { ResourceNamespace = ClientConfig.Namespace, Name = topic },
+                RecallHandle = recallhandle
+            };
+            var invocation =
+                await ClientManager.RecallMessage(new Endpoints(ClientConfig.Endpoints), request, ClientConfig.RequestTimeout);
+            StatusChecker.Check(invocation.Response.Status, request, invocation.RequestId);
+            return new RecallReceipt(invocation.Response.MessageId);
+        }
+
         public class Builder
         {
             private ClientConfig _clientConfig;

csharp/rocketmq-client-csharp/RecallReceipt.cs

@@ -0,0 +1,34 @@
+/*
+ * 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.
+ */
+
+namespace Org.Apache.Rocketmq
+{
+    public sealed class RecallReceipt : IRecallReceipt
+    {
+        public RecallReceipt(string messageId)
+        {
+            MessageId = messageId;
+        }
+
+        public string MessageId { get; }
+
+        public override string ToString()
+        {
+            return $"{nameof(MessageId)}: {MessageId}";
+        }
+    }
+}

csharp/rocketmq-client-csharp/RpcClient.cs

@@ -189,5 +189,14 @@ internal static HttpMessageHandler CreateHttpHandler()
             var call = _stub.NotifyClientTerminationAsync(request, callOptions);
             return await call.ResponseAsync;
         }
+
+        public async Task<Proto::RecallMessageResponse> RecallMessage(Metadata metadata, Proto.RecallMessageRequest request, TimeSpan timeout)
+        {
+            var deadline = DateTime.UtcNow.Add(timeout);
+            var callOptions = new CallOptions(metadata, deadline);
+
+            var call = _stub.RecallMessageAsync(request, callOptions);
+            return await call.ResponseAsync;
+        }
     }
 }

csharp/rocketmq-client-csharp/SendReceipt.cs

@@ -23,15 +23,21 @@ namespace Org.Apache.Rocketmq
 {
     public sealed class SendReceipt : ISendReceipt
     {
-        private SendReceipt(string messageId, string transactionId, MessageQueue messageQueue)
+        private SendReceipt(string messageId, string transactionId, MessageQueue messageQueue, long offset, string recallHandle)
         {
             MessageId = messageId;
             TransactionId = transactionId;
             MessageQueue = messageQueue;
+            Offset = offset;
+            RecallHandle = recallHandle;
         }
 
         public string MessageId { get; }
 
+        public string RecallHandle { get; }
+
+        public long Offset { get; }
+
         public string TransactionId { get; }
 
         private MessageQueue MessageQueue { get; }
@@ -40,7 +46,7 @@ private SendReceipt(string messageId, string transactionId, MessageQueue message
 
         public override string ToString()
         {
-            return $"{nameof(MessageId)}: {MessageId}";
+            return $"{nameof(MessageId)}: {MessageId}, {nameof(RecallHandle)}: {RecallHandle}";
         }
 
         public static IEnumerable<SendReceipt> ProcessSendMessageResponse(MessageQueue mq,
@@ -58,7 +64,7 @@ public static IEnumerable<SendReceipt> ProcessSendMessageResponse(MessageQueue m
 
             // May throw exception.
             StatusChecker.Check(status, invocation.Request, invocation.RequestId);
-            return invocation.Response.Entries.Select(entry => new SendReceipt(entry.MessageId, entry.TransactionId, mq)).ToList();
+            return invocation.Response.Entries.Select(entry => new SendReceipt(entry.MessageId, entry.TransactionId, mq, entry.Offset, entry.RecallHandle)).ToList();
         }
     }
 }

csharp/tests/ClientManagerTest.cs

@@ -121,6 +121,15 @@ public void TestNotifyClientTermination()
             // Expect no exception thrown.
         }
 
+        [TestMethod]
+        public void TestRecallMessage()
+        {
+            var request = new RecallMessageRequest();
+            _clientManager.RecallMessage(FakeEndpoints, request, TimeSpan.FromSeconds(1));
+            _clientManager.RecallMessage(null, request, TimeSpan.FromSeconds(1));
+            // Expect no exception thrown.
+        }
+
         private Client CreateTestClient()
         {
             return new Producer(_clientConfig, new ConcurrentDictionary<string, bool>(), 1, null);