[ISSUE #776] Add Reentrant Push Consumer Message Receiving Support for 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 () =>
  {
@@ -313,6 +313,7 @@ private async Task<TopicRouteData> FetchTopicRoute0(string topic)
  {
  Topic = new Proto::Resource
  {
+ ResourceNamespace = ClientConfig.Namespace,
  Name = topic
  },
  Endpoints = Endpoints.ToProtobuf()
@@ -432,14 +433,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/ClientConfig.cs

@@ -22,12 +22,13 @@ namespace Org.Apache.Rocketmq
  public class ClientConfig
  {
  private ClientConfig(ISessionCredentialsProvider sessionCredentialsProvider, TimeSpan requestTimeout,
- string endpoints, bool sslEnabled)
+ string endpoints, bool sslEnabled, string namespaceName)
  {
  SessionCredentialsProvider = sessionCredentialsProvider;
  RequestTimeout = requestTimeout;
  Endpoints = endpoints;
  SslEnabled = sslEnabled;
+ Namespace = namespaceName;
  }
 
  public ISessionCredentialsProvider SessionCredentialsProvider { get; }
@@ -37,13 +38,16 @@ private ClientConfig(ISessionCredentialsProvider sessionCredentialsProvider, Tim
  public string Endpoints { get; }
 
  public bool SslEnabled { get; }
+
+ public string Namespace { get; }
 
  public class Builder
  {
  private ISessionCredentialsProvider _sessionCredentialsProvider;
  private TimeSpan _requestTimeout = TimeSpan.FromSeconds(3);
  private string _endpoints;
  private bool _sslEnabled = true;
+ private string _namespace = "";
 
  public Builder SetCredentialsProvider(ISessionCredentialsProvider sessionCredentialsProvider)
  {
@@ -68,10 +72,16 @@ public Builder EnableSsl(bool sslEnabled)
  _sslEnabled = sslEnabled;
  return this;
  }
+
+ public Builder SetNamespace(string namespaceName)
+ {
+ _namespace = namespaceName;
+ return this;
+ }
 
  public ClientConfig Build()
  {
- return new ClientConfig(_sessionCredentialsProvider, _requestTimeout, _endpoints, _sslEnabled);
+ return new ClientConfig(_sessionCredentialsProvider, _requestTimeout, _endpoints, _sslEnabled, _namespace);
  }
  }
  }

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
+ }
+}