You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by aa...@apache.org on 2022/12/20 02:20:03 UTC
[rocketmq-clients] branch master updated (6010d65e -> 2e64474c)
This is an automated email from the ASF dual-hosted git repository.
aaronai pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/rocketmq-clients.git
from 6010d65e Bump OT to 1.3.1
new 334df3ff Align ClientId generation in dotnet driver with others
new e1bf271f Align dotnet driver with other drivers regarding FIFO message handling
new 422a5709 Add missing periodic heartbeat in dotnet driver
new 2e64474c C# SimpleConsumer should take request timeout into consideration when polling for messages
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
csharp/rocketmq-client-csharp/Client.cs | 11 ++++++++++-
csharp/rocketmq-client-csharp/ClientConfig.cs | 11 +++--------
csharp/rocketmq-client-csharp/ClientManager.cs | 2 +-
csharp/rocketmq-client-csharp/Producer.cs | 14 ++++++++------
csharp/rocketmq-client-csharp/PublishLoadBalancer.cs | 9 ++++++++-
csharp/rocketmq-client-csharp/SimpleConsumer.cs | 9 ++++++---
6 files changed, 36 insertions(+), 20 deletions(-)
[rocketmq-clients] 01/04: Align ClientId generation in dotnet driver with others
Posted by aa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
aaronai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/rocketmq-clients.git
commit 334df3ff99634113cf475c4fa8a66baa95f9773f
Author: colprog <co...@gmail.com>
AuthorDate: Thu Dec 15 03:59:28 2022 +0800
Align ClientId generation in dotnet driver with others
---
csharp/rocketmq-client-csharp/ClientConfig.cs | 11 +++--------
1 file changed, 3 insertions(+), 8 deletions(-)
diff --git a/csharp/rocketmq-client-csharp/ClientConfig.cs b/csharp/rocketmq-client-csharp/ClientConfig.cs
index 86175a2a..bdba0284 100644
--- a/csharp/rocketmq-client-csharp/ClientConfig.cs
+++ b/csharp/rocketmq-client-csharp/ClientConfig.cs
@@ -16,6 +16,7 @@
*/
using System;
using System.Collections.Generic;
+using System.Threading;
using rmq = Apache.Rocketmq.V2;
namespace Org.Apache.Rocketmq
@@ -23,12 +24,13 @@ namespace Org.Apache.Rocketmq
public class ClientConfig : IClientConfig
{
+ private static long instanceSequence = 0;
public ClientConfig()
{
var hostName = System.Net.Dns.GetHostName();
var pid = System.Diagnostics.Process.GetCurrentProcess().Id;
- this.clientId_ = string.Format("{0}@{1}#{2}", hostName, pid, instanceName_);
+ this.clientId_ = string.Format("{0}@{1}#{2}", hostName, pid, Interlocked.Increment(ref instanceSequence));
this._requestTimeout = TimeSpan.FromSeconds(3);
this.longPollingIoTimeout_ = TimeSpan.FromSeconds(30);
this.client_type_ = rmq::ClientType.Unspecified;
@@ -110,11 +112,6 @@ namespace Org.Apache.Rocketmq
set { tracingEnabled_ = value; }
}
- public void setInstanceName(string instanceName)
- {
- this.instanceName_ = instanceName;
- }
-
private string _region = "cn-hangzhou";
private string _serviceName = "ONS";
@@ -132,8 +129,6 @@ namespace Org.Apache.Rocketmq
private bool tracingEnabled_;
- private string instanceName_ = "default";
-
private rmq::ClientType client_type_;
public rmq::ClientType ClientType
{
[rocketmq-clients] 04/04: C# SimpleConsumer should take request timeout into consideration when polling for messages
Posted by aa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
aaronai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/rocketmq-clients.git
commit 2e64474c520bf26b4d0d0f20422b71d608d5e072
Author: colprog <co...@gmail.com>
AuthorDate: Sat Dec 17 18:21:09 2022 +0800
C# SimpleConsumer should take request timeout into consideration when polling for messages
---
csharp/rocketmq-client-csharp/SimpleConsumer.cs | 9 ++++++---
1 file changed, 6 insertions(+), 3 deletions(-)
diff --git a/csharp/rocketmq-client-csharp/SimpleConsumer.cs b/csharp/rocketmq-client-csharp/SimpleConsumer.cs
index 60f0ba94..12e5d8cf 100644
--- a/csharp/rocketmq-client-csharp/SimpleConsumer.cs
+++ b/csharp/rocketmq-client-csharp/SimpleConsumer.cs
@@ -219,9 +219,12 @@ namespace Org.Apache.Rocketmq
var targetUrl = Utilities.TargetUrl(messageQueue);
var metadata = new Metadata();
Signature.Sign(this, metadata);
-
- return await Manager.ReceiveMessage(targetUrl, metadata, request,
- ClientSettings.Subscription.LongPollingTimeout.ToTimeSpan());
+
+ var timeout = ClientSettings.Subscription.LongPollingTimeout
+ .ToTimeSpan()
+ .Add(this.RequestTimeout);
+
+ return await Manager.ReceiveMessage(targetUrl, metadata, request, timeout);
}
[rocketmq-clients] 02/04: Align dotnet driver with other drivers regarding FIFO message handling
Posted by aa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
aaronai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/rocketmq-clients.git
commit e1bf271fcc914706bb7f400674a407da9e1d0672
Author: colprog <co...@gmail.com>
AuthorDate: Sat Dec 17 03:19:26 2022 +0800
Align dotnet driver with other drivers regarding FIFO message handling
---
csharp/rocketmq-client-csharp/Producer.cs | 14 ++++++++------
csharp/rocketmq-client-csharp/PublishLoadBalancer.cs | 9 ++++++++-
2 files changed, 16 insertions(+), 7 deletions(-)
diff --git a/csharp/rocketmq-client-csharp/Producer.cs b/csharp/rocketmq-client-csharp/Producer.cs
index c2c5e5db..77016571 100644
--- a/csharp/rocketmq-client-csharp/Producer.cs
+++ b/csharp/rocketmq-client-csharp/Producer.cs
@@ -107,7 +107,7 @@ namespace Org.Apache.Rocketmq
{
_topicsOfInterest.Add(message.Topic);
- if (!_loadBalancer.ContainsKey(message.Topic))
+ if (!_loadBalancer.TryGetValue(message.Topic, out var publishLb))
{
var topicRouteData = await GetRouteFor(message.Topic, false);
if (null == topicRouteData || null == topicRouteData.MessageQueues || 0 == topicRouteData.MessageQueues.Count)
@@ -116,12 +116,10 @@ namespace Org.Apache.Rocketmq
throw new TopicRouteException(string.Format("No topic route for {0}", message.Topic));
}
- var loadBalancerItem = new PublishLoadBalancer(topicRouteData);
- _loadBalancer.TryAdd(message.Topic, loadBalancerItem);
+ publishLb = new PublishLoadBalancer(topicRouteData);
+ _loadBalancer.TryAdd(message.Topic, publishLb);
}
- var publishLb = _loadBalancer[message.Topic];
-
var request = new rmq::SendMessageRequest();
var entry = new rmq::Message
{
@@ -176,7 +174,7 @@ namespace Org.Apache.Rocketmq
}
List<string> targets = new List<string>();
- List<rmq::MessageQueue> candidates = publishLb.Select(message.MaxAttemptTimes);
+ List<rmq::MessageQueue> candidates = publishLb.Select(message.MessageGroup, message.MaxAttemptTimes);
foreach (var messageQueue in candidates)
{
targets.Add(Utilities.TargetUrl(messageQueue));
@@ -205,6 +203,10 @@ namespace Org.Apache.Rocketmq
return new SendReceipt(messageId);
}
+ else if (response?.Status is not null)
+ {
+ Logger.Warn($"Send failed with code: {response.Status.Code}, error: {response.Status.Message}");
+ }
}
catch (Exception e)
{
diff --git a/csharp/rocketmq-client-csharp/PublishLoadBalancer.cs b/csharp/rocketmq-client-csharp/PublishLoadBalancer.cs
index 7d258b4f..6670beb6 100644
--- a/csharp/rocketmq-client-csharp/PublishLoadBalancer.cs
+++ b/csharp/rocketmq-client-csharp/PublishLoadBalancer.cs
@@ -85,7 +85,7 @@ namespace Org.Apache.Rocketmq
return true;
}
- public List<rmq::MessageQueue> Select(int maxAttemptTimes)
+ public List<rmq::MessageQueue> Select(string messageGroup, int maxAttemptTimes)
{
List<rmq::MessageQueue> result = new List<rmq::MessageQueue>();
@@ -94,6 +94,13 @@ namespace Org.Apache.Rocketmq
{
return result;
}
+
+ if (!string.IsNullOrEmpty(messageGroup))
+ {
+ result.Add(all[messageGroup.GetHashCode() % all.Count]);
+ return result;
+ }
+
int start = ++_roundRobinIndex;
int found = 0;
[rocketmq-clients] 03/04: Add missing periodic heartbeat in dotnet driver
Posted by aa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
aaronai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/rocketmq-clients.git
commit 422a5709ca103c627c98dd4321cea33dc7ededb2
Author: colprog <co...@gmail.com>
AuthorDate: Fri Dec 16 02:35:57 2022 +0800
Add missing periodic heartbeat in dotnet driver
---
csharp/rocketmq-client-csharp/Client.cs | 11 ++++++++++-
csharp/rocketmq-client-csharp/ClientManager.cs | 2 +-
2 files changed, 11 insertions(+), 2 deletions(-)
diff --git a/csharp/rocketmq-client-csharp/Client.cs b/csharp/rocketmq-client-csharp/Client.cs
index ea44abf9..217479d9 100644
--- a/csharp/rocketmq-client-csharp/Client.cs
+++ b/csharp/rocketmq-client-csharp/Client.cs
@@ -70,6 +70,7 @@ namespace Org.Apache.Rocketmq
_topicRouteTable = new ConcurrentDictionary<string, TopicRouteData>();
_updateTopicRouteCts = new CancellationTokenSource();
+ _HeartbeatCts = new CancellationTokenSource();
_telemetryCts = new CancellationTokenSource();
}
@@ -90,6 +91,13 @@ namespace Org.Apache.Rocketmq
CreateSession(accessPointUrl);
await _sessions[accessPointUrl].AwaitSettingNegotiationCompletion();
Logger.Debug($"Session has been created for {accessPointUrl}");
+
+ Schedule(async () =>
+ {
+ Logger.Debug("Sending heartbeat by schedule");
+ await Heartbeat();
+
+ }, 10, _HeartbeatCts.Token);
await Heartbeat();
}
@@ -97,6 +105,7 @@ namespace Org.Apache.Rocketmq
{
Logger.Info($"Shutdown client");
_updateTopicRouteCts.Cancel();
+ _HeartbeatCts.Cancel();
_telemetryCts.Cancel();
await Manager.Shutdown();
}
@@ -518,7 +527,7 @@ namespace Org.Apache.Rocketmq
private readonly ConcurrentDictionary<string, TopicRouteData> _topicRouteTable;
private readonly CancellationTokenSource _updateTopicRouteCts;
-
+ private readonly CancellationTokenSource _HeartbeatCts;
private readonly CancellationTokenSource _telemetryCts;
public CancellationTokenSource TelemetryCts()
diff --git a/csharp/rocketmq-client-csharp/ClientManager.cs b/csharp/rocketmq-client-csharp/ClientManager.cs
index 329f5910..e3bda5d2 100644
--- a/csharp/rocketmq-client-csharp/ClientManager.cs
+++ b/csharp/rocketmq-client-csharp/ClientManager.cs
@@ -198,7 +198,7 @@ namespace Org.Apache.Rocketmq
}
default:
{
- Logger.Warn("Unknown error status");
+ Logger.Warn($"Unknown error status({entry.Status.Code}): {entry.Status.Message}");
break;
}
}