[ISSUE #776] Add push consumer for fifo message in C# SDK

csharp/examples/PushConsumerExample.cs

@@ -0,0 +1,75 @@
+/*
+ * 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.
+ */
+using System;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.Extensions.Logging;
+using Org.Apache.Rocketmq;
+
+namespace examples
+{
+    public class PushConsumerExample
+    {
+        private static readonly ILogger Logger = MqLogManager.CreateLogger(typeof(PushConsumerExample).FullName);
+
+        private static readonly string AccessKey = Environment.GetEnvironmentVariable("ROCKETMQ_ACCESS_KEY");
+        private static readonly string SecretKey = Environment.GetEnvironmentVariable("ROCKETMQ_SECRET_KEY");
+        private static readonly string Endpoint = Environment.GetEnvironmentVariable("ROCKETMQ_ENDPOINT");
+
+        internal static async Task QuickStart()
+        {
+            // Enable the switch if you use .NET Core 3.1 and want to disable TLS/SSL.
+            // AppContext.SetSwitch("System.Net.Http.SocketsHttpHandler.Http2UnencryptedSupport", true);
+
+            // Credential provider is optional for client configuration.
+            var credentialsProvider = new StaticSessionCredentialsProvider(AccessKey, SecretKey);
+            var clientConfig = new ClientConfig.Builder()
+                .SetEndpoints(Endpoint)
+                .SetCredentialsProvider(credentialsProvider)
+                .Build();
+
+            // Add your subscriptions.
+            const string consumerGroup = "yourConsumerGroup";
+            const string topic = "yourTopic";
+            var subscription = new Dictionary<string, FilterExpression>
+                { { topic, new FilterExpression("*") } };
+
+            var pushConsumer = await new PushConsumer.Builder()
+                .SetClientConfig(clientConfig)
+                .SetConsumerGroup(consumerGroup)
+                .SetSubscriptionExpression(subscription)
+                .SetMessageListener(new CustomMessageListener())
+                .Build();
+
+            Thread.Sleep(Timeout.Infinite);
+
+            // Close the push consumer if you don't need it anymore.
+            // await pushConsumer.DisposeAsync();
+        }
+
+        private class CustomMessageListener : IMessageListener
+        {
+            public ConsumeResult Consume(MessageView messageView)
+            {
+                // Handle the received message and return consume result.
+                Logger.LogInformation($"Consume message={messageView}");
+                return ConsumeResult.SUCCESS;
+            }
+        }
+    }
+}

csharp/examples/QuickStart.cs

@@ -34,6 +34,7 @@ public static void Main()
             // ProducerFifoMessageExample.QuickStart().Wait();
             // ProducerDelayMessageExample.QuickStart().Wait();
             // ProducerTransactionMessageExample.QuickStart().Wait();
+            // PushConsumerExample.QuickStart().Wait();
             // SimpleConsumerExample.QuickStart().Wait();
             // ProducerBenchmark.QuickStart().Wait();
         }

csharp/rocketmq-client-csharp/Assignment.cs

@@ -0,0 +1,51 @@
+/*
+ * 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.
+ */
+
+using System;
+using System.Collections.Generic;
+
+namespace Org.Apache.Rocketmq
+{
+    public class Assignment
+    {
+        public Assignment(MessageQueue messageQueue)
+        {
+            MessageQueue = messageQueue ?? throw new ArgumentNullException(nameof(messageQueue));
+        }
+
+        public MessageQueue MessageQueue { get; }
+
+        public override bool Equals(object obj)
+        {
+            if (this == obj) return true;
+            if (obj == null || GetType() != obj.GetType()) return false;
+
+            var other = (Assignment)obj;
+            return EqualityComparer<MessageQueue>.Default.Equals(MessageQueue, other.MessageQueue);
+        }
+
+        public override int GetHashCode()
+        {
+            return EqualityComparer<MessageQueue>.Default.GetHashCode(MessageQueue);
+        }
+
+        public override string ToString()
+        {
+            return $"Assignment{{messageQueue={MessageQueue}}}";
+        }
+    }
+}

csharp/rocketmq-client-csharp/Assignments.cs

@@ -0,0 +1,65 @@
+/*
+ * 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.
+ */
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using Apache.Rocketmq.V2;
+
+namespace Org.Apache.Rocketmq
+{
+    public class Assignments
+    {
+        private readonly List<Assignment> _assignmentList;
+
+        public Assignments(List<Assignment> assignmentList)
+        {
+            _assignmentList = assignmentList;
+        }
+
+        public override bool Equals(object obj)
+        {
+            if (this == obj)
+            {
+                return true;
+            }
+
+            if (obj == null || GetType() != obj.GetType())
+            {
+                return false;
+            }
+
+            var other = (Assignments)obj;
+            return _assignmentList.SequenceEqual(other._assignmentList);
+        }
+
+        public override int GetHashCode()
+        {
+            return HashCode.Combine(_assignmentList);
+        }
+
+        public override string ToString()
+        {
+            return $"{nameof(Assignments)} {{ {nameof(_assignmentList)} = {_assignmentList} }}";
+        }
+
+        public List<Assignment> GetAssignmentList()
+        {
+            return _assignmentList;
+        }
+    }
+}

csharp/rocketmq-client-csharp/Client.cs

@@ -113,7 +113,7 @@ protected virtual async Task Shutdown()
             Logger.LogDebug($"Shutdown the rocketmq client successfully, clientId={ClientId}");
         }
 
-        private (bool, Session) GetSession(Endpoints endpoints)
+        private protected (bool, Session) GetSession(Endpoints endpoints)
         {
             _sessionLock.EnterReadLock();
             try
@@ -261,7 +261,7 @@ private void Stats()
                 $"AvailableCompletionPortThreads={availableIo}");
         }
 
-        private void ScheduleWithFixedDelay(Action action, TimeSpan delay, TimeSpan period, CancellationToken token)
+        private protected void ScheduleWithFixedDelay(Action action, TimeSpan delay, TimeSpan period, CancellationToken token)
         {
             Task.Run(async () =>
             {
@@ -432,14 +432,19 @@ internal ClientConfig GetClientConfig()
             return ClientConfig;
         }
 
+        internal IClientManager GetClientManager()
+        {
+            return ClientManager;
+        }
+
         internal virtual void OnRecoverOrphanedTransactionCommand(Endpoints endpoints,
             Proto.RecoverOrphanedTransactionCommand command)
         {
             Logger.LogWarning($"Ignore orphaned transaction recovery command from remote, which is not expected, " +
                               $"clientId={ClientId}, endpoints={endpoints}");
         }
 
-        internal async void OnVerifyMessageCommand(Endpoints endpoints, Proto.VerifyMessageCommand command)
+        internal virtual async void OnVerifyMessageCommand(Endpoints endpoints, Proto.VerifyMessageCommand command)
         {
             // Only push consumer support message consumption verification.
             Logger.LogWarning($"Ignore verify message command from remote, which is not expected, clientId={ClientId}, " +

csharp/rocketmq-client-csharp/ClientManager.cs

@@ -165,6 +165,16 @@ public async Task Shutdown()
             return new RpcInvocation<Proto.ChangeInvisibleDurationRequest, Proto.ChangeInvisibleDurationResponse>(
                 request, response, metadata);
         }
+
+        public async Task<RpcInvocation<Proto.ForwardMessageToDeadLetterQueueRequest, Proto.ForwardMessageToDeadLetterQueueResponse>>
+            ForwardMessageToDeadLetterQueue(Endpoints endpoints,
+                Proto.ForwardMessageToDeadLetterQueueRequest request, TimeSpan timeout)
+        {
+            var metadata = _client.Sign();
+            var response = await GetRpcClient(endpoints).ForwardMessageToDeadLetterQueue(metadata, request, timeout);
+            return new RpcInvocation<Proto.ForwardMessageToDeadLetterQueueRequest, Proto.ForwardMessageToDeadLetterQueueResponse>(
+                    request, response, metadata);
+        }
 
         public async Task<RpcInvocation<Proto.EndTransactionRequest, Proto.EndTransactionResponse>> EndTransaction(
             Endpoints endpoints, Proto.EndTransactionRequest request, TimeSpan timeout)

csharp/rocketmq-client-csharp/ConsumeResult.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
+{
+    /// <summary>
+    /// Designed for push consumer specifically.
+    /// </summary>
+    public enum ConsumeResult
+    {
+        /// <summary>
+        /// Consume message successfully.
+        /// </summary>
+        SUCCESS,
+        /// <summary>
+        /// Failed to consume message.
+        /// </summary>
+        FAILURE
+    }
+}

csharp/rocketmq-client-csharp/ConsumeService.cs

@@ -0,0 +1,82 @@
+/*
+ * 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.
+ */
+
+using System;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.Extensions.Logging;
+
+namespace Org.Apache.Rocketmq
+{
+    public abstract class ConsumeService
+    {
+        private static readonly ILogger Logger = MqLogManager.CreateLogger<ConsumeService>();
+
+        protected readonly string ClientId;
+        private readonly IMessageListener _messageListener;
+        private readonly TaskScheduler _consumptionTaskScheduler;
+        private readonly CancellationToken _consumptionCtsToken;
+
+        public ConsumeService(string clientId, IMessageListener messageListener, TaskScheduler consumptionTaskScheduler,
+            CancellationToken consumptionCtsToken)
+        {
+            ClientId = clientId;
+            _messageListener = messageListener;
+            _consumptionTaskScheduler = consumptionTaskScheduler;
+            _consumptionCtsToken = consumptionCtsToken;
+        }
+
+        public abstract Task Consume(ProcessQueue pq, List<MessageView> messageViews);
+
+        public async Task<ConsumeResult> Consume(MessageView messageView)
+        {
+            return await Consume(messageView, TimeSpan.Zero);
+        }
+
+        public async Task<ConsumeResult> Consume(MessageView messageView, TimeSpan delay)
+        {
+            var task = new ConsumeTask(ClientId, _messageListener, messageView);
+            var delayMilliseconds = (int) delay.TotalMilliseconds;
+
+            if (delayMilliseconds <= 0)
+            {
+                return await Task.Factory.StartNew(() => task.Call(), _consumptionCtsToken, TaskCreationOptions.None,
+                    _consumptionTaskScheduler);
+            }
+
+            var tcs = new TaskCompletionSource<ConsumeResult>();
+
+            await Task.Delay(delay, _consumptionCtsToken).ContinueWith(async _ =>
+            {
+                try
+                {
+                    var result = await Task.Factory.StartNew(() => task.Call(), _consumptionCtsToken,
+                        TaskCreationOptions.None, _consumptionTaskScheduler);
+                    tcs.SetResult(result);
+                }
+                catch (Exception e)
+                {
+                    Logger.LogError(e, $"Error while consuming message, clientId={ClientId}");
+                    tcs.SetException(e);
+                }
+            }, TaskScheduler.Default);
+
+            return await tcs.Task;
+        }
+    }
+}