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 2023/03/06 04:54:41 UTC
[rocketmq-clients] branch master updated (a905eba6 -> 00492cc7)
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 a905eba6 Add request-id support
new a376ca5a Allow to adjust log configuration
new 4c135bb2 Override message ToString method
new 0a02f18a Allow user to disable TLS
new c8af7a4c Adjust settings sync frequency
new 00492cc7 Bugfix: wrong message type judgement
The 5 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/README.md | 2 +-
csharp/examples/ProducerBenchmark.cs | 2 +-
csharp/examples/ProducerDelayMessageExample.cs | 1 +
csharp/examples/ProducerNormalMessageExample.cs | 2 +
.../examples/ProducerTransactionMessageExample.cs | 4 +-
csharp/examples/QuickStart.cs | 6 +--
csharp/rocketmq-client-csharp/Client.cs | 49 +++++++++++----------
csharp/rocketmq-client-csharp/ClientConfig.cs | 15 ++++++-
csharp/rocketmq-client-csharp/ClientManager.cs | 2 +-
.../rocketmq-client-csharp/ClientMeterManager.cs | 2 +-
csharp/rocketmq-client-csharp/Endpoints.cs | 23 +++++-----
csharp/rocketmq-client-csharp/Message.cs | 15 ++++++-
csharp/rocketmq-client-csharp/MessageView.cs | 9 ++--
csharp/rocketmq-client-csharp/MqLogManager.cs | 31 +++++++++++--
csharp/rocketmq-client-csharp/PublishingMessage.cs | 2 +-
csharp/rocketmq-client-csharp/RpcClient.cs | 4 +-
csharp/rocketmq-client-csharp/Session.cs | 13 ++----
csharp/rocketmq-client-csharp/SimpleConsumer.cs | 3 +-
csharp/rocketmq-client-csharp/logo.png | Bin 0 -> 85598 bytes
.../rocketmq-client-csharp.csproj | 7 ++-
20 files changed, 124 insertions(+), 68 deletions(-)
create mode 100644 csharp/rocketmq-client-csharp/logo.png
[rocketmq-clients] 01/05: Allow to adjust log configuration
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 a376ca5ac4d353381084dad353e3e3f828baddc6
Author: Aaron Ai <ya...@gmail.com>
AuthorDate: Thu Mar 2 17:43:43 2023 +0800
Allow to adjust log configuration
---
csharp/rocketmq-client-csharp/Client.cs | 42 ++++++++++++++++-----------
csharp/rocketmq-client-csharp/MqLogManager.cs | 31 +++++++++++++++++---
2 files changed, 52 insertions(+), 21 deletions(-)
diff --git a/csharp/rocketmq-client-csharp/Client.cs b/csharp/rocketmq-client-csharp/Client.cs
index a9bdb705..e78a8651 100644
--- a/csharp/rocketmq-client-csharp/Client.cs
+++ b/csharp/rocketmq-client-csharp/Client.cs
@@ -313,28 +313,36 @@ namespace Org.Apache.Rocketmq
private async Task<TopicRouteData> FetchTopicRoute0(string topic)
{
- var request = new Proto::QueryRouteRequest
+ try
{
- Topic = new Proto::Resource
+ var request = new Proto::QueryRouteRequest
{
- Name = topic
- },
- Endpoints = Endpoints.ToProtobuf()
- };
+ Topic = new Proto::Resource
+ {
+ Name = topic
+ },
+ Endpoints = Endpoints.ToProtobuf()
+ };
+
+ var invocation =
+ await ClientManager.QueryRoute(Endpoints, request, ClientConfig.RequestTimeout);
+ var code = invocation.Response.Status.Code;
+ if (!Proto.Code.Ok.Equals(code))
+ {
+ Logger.Error($"Failed to fetch topic route, clientId={ClientId}, topic={topic}, code={code}, " +
+ $"statusMessage={invocation.Response.Status.Message}");
+ }
- var invocation =
- await ClientManager.QueryRoute(Endpoints, request, ClientConfig.RequestTimeout);
- var code = invocation.Response.Status.Code;
- if (!Proto.Code.Ok.Equals(code))
+ StatusChecker.Check(invocation.Response.Status, request, invocation.RequestId);
+
+ var messageQueues = invocation.Response.MessageQueues.ToList();
+ return new TopicRouteData(messageQueues);
+ }
+ catch (Exception e)
{
- Logger.Error($"Failed to fetch topic route, clientId={ClientId}, topic={topic}, code={code}, " +
- $"statusMessage={invocation.Response.Status.Message}");
+ Logger.Error(e, $"Failed to fetch topic route, clientId={ClientId}, topic={topic}");
+ throw;
}
-
- StatusChecker.Check(invocation.Response.Status, request, invocation.RequestId);
-
- var messageQueues = invocation.Response.MessageQueues.ToList();
- return new TopicRouteData(messageQueues);
}
private async void Heartbeat()
diff --git a/csharp/rocketmq-client-csharp/MqLogManager.cs b/csharp/rocketmq-client-csharp/MqLogManager.cs
index 41ebf5c1..1e67bb56 100644
--- a/csharp/rocketmq-client-csharp/MqLogManager.cs
+++ b/csharp/rocketmq-client-csharp/MqLogManager.cs
@@ -35,13 +35,36 @@ namespace Org.Apache.Rocketmq
private static readonly Lazy<LogFactory> LazyInstance = new(BuildLogFactory);
+ private const string FileLogLevelKey = "rocketmq.log.level";
+ private const string FileLogLevel = "Info";
+
+ private const string ConsoleAppenderEnabledKey = "mq.consoleAppender.enabled";
+ private const string ConsoleAppenderEnabled = "false";
+ private const string ConsoleAppenderLogLevel = "Off";
+
+
+ private const string FileLogRootKey = "rocketmq.log.root";
+
+ private const string FileMaxIndexKey = "rocketmq.log.file.maxIndex";
+ private const string FileMaxIndex = "10";
+
private static LogFactory BuildLogFactory()
{
+ var fileLogLevel = Environment.GetEnvironmentVariable(FileLogLevelKey) ?? FileLogLevel;
+ var consoleAppenderEnabled =
+ Environment.GetEnvironmentVariable(ConsoleAppenderEnabledKey) ?? ConsoleAppenderEnabled;
+ var consoleLogLevel = bool.Parse(consoleAppenderEnabled) ? fileLogLevel : ConsoleAppenderLogLevel;
+ var fileLogRoot = Environment.GetEnvironmentVariable(FileLogRootKey) ??
+ Environment.GetFolderPath(Environment.SpecialFolder.UserProfile);
+ var fileMaxIndexStr = Environment.GetEnvironmentVariable(FileMaxIndexKey) ?? FileMaxIndex;
+ var fileMaxIndex = int.Parse(fileMaxIndexStr);
+
+
var config = new LoggingConfiguration();
var fileTarget = new FileTarget();
fileTarget.Name = "log_file";
fileTarget.FileName =
- new SimpleLayout("${specialfolder:folder=UserProfile}/logs/rocketmq/rocketmq-client.log");
+ new SimpleLayout($"{fileLogRoot}/logs/rocketmq/rocketmq-client.log");
fileTarget.Layout =
new SimpleLayout(
"${longdate} ${level:uppercase=true:padding=-5} [${processid}] [${threadid}] [${callsite}:${callsite-linenumber}] ${message} ${onexception:${exception:format=ToString,Data}}");
@@ -49,7 +72,7 @@ namespace Org.Apache.Rocketmq
new SimpleLayout("${specialfolder:folder=UserProfile}/logs/rocketmq/rocketmq-client.{######}.log");
fileTarget.ArchiveAboveSize = 67108864;
fileTarget.ArchiveNumbering = ArchiveNumberingMode.DateAndSequence;
- fileTarget.MaxArchiveFiles = 10;
+ fileTarget.MaxArchiveFiles = fileMaxIndex;
fileTarget.ConcurrentWrites = true;
fileTarget.KeepFileOpen = false;
@@ -66,10 +89,10 @@ namespace Org.Apache.Rocketmq
config.AddTarget(consoleTarget);
- var asyncFileRule = new LoggingRule("*", LogLevel.FromString("Debug"), asyncTargetWrapper);
+ var asyncFileRule = new LoggingRule("*", LogLevel.FromString(fileLogLevel), asyncTargetWrapper);
config.LoggingRules.Add(asyncFileRule);
- var consoleRule = new LoggingRule("*", LogLevel.FromString("Debug"), consoleTarget);
+ var consoleRule = new LoggingRule("*", LogLevel.FromString(consoleLogLevel), consoleTarget);
config.LoggingRules.Add(consoleRule);
var logFactory = new LogFactory();
[rocketmq-clients] 02/05: Override message ToString method
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 4c135bb2e6b1617c17b580c7efdff3e22019674e
Author: Aaron Ai <ya...@gmail.com>
AuthorDate: Thu Mar 2 18:56:39 2023 +0800
Override message ToString method
---
csharp/rocketmq-client-csharp/Message.cs | 15 +++++++++++++--
csharp/rocketmq-client-csharp/MessageView.cs | 9 ++++-----
2 files changed, 17 insertions(+), 7 deletions(-)
diff --git a/csharp/rocketmq-client-csharp/Message.cs b/csharp/rocketmq-client-csharp/Message.cs
index 2430bbbd..32c4352c 100644
--- a/csharp/rocketmq-client-csharp/Message.cs
+++ b/csharp/rocketmq-client-csharp/Message.cs
@@ -17,13 +17,15 @@
using System;
using System.Collections.Generic;
+using System.Linq;
using System.Text.RegularExpressions;
namespace Org.Apache.Rocketmq
{
public class Message
{
- internal static readonly Regex TopicRegex = new("^[%a-zA-Z0-9_-]+$");
+ internal static readonly Regex TopicRegex = new("^[%a-zA-Z0-9_-]+$");
+
private Message(string topic, byte[] body, string tag, List<string> keys,
Dictionary<string, string> properties, DateTime? deliveryTimestamp, string messageGroup)
{
@@ -60,6 +62,14 @@ namespace Org.Apache.Rocketmq
public string MessageGroup { get; }
+ public override string ToString()
+ {
+ return
+ $"{nameof(Topic)}: {Topic}, {nameof(Tag)}: {Tag}, {nameof(Keys)}: {string.Join(", ", Keys)}, {nameof(Properties)}: " +
+ $"{string.Join(", ", Properties.Select(kvp => kvp.ToString()))}, {nameof(DeliveryTimestamp)}: {DeliveryTimestamp}, {nameof(MessageGroup)}: " +
+ $"{MessageGroup}";
+ }
+
public class Builder
{
private string _topic;
@@ -73,7 +83,8 @@ namespace Org.Apache.Rocketmq
public Builder SetTopic(string topic)
{
Preconditions.CheckArgument(null != topic, "topic should not be null");
- Preconditions.CheckArgument(topic != null && TopicRegex.Match(topic).Success, $"topic does not match the regex {TopicRegex}");
+ Preconditions.CheckArgument(topic != null && TopicRegex.Match(topic).Success,
+ $"topic does not match the regex {TopicRegex}");
_topic = topic;
return this;
}
diff --git a/csharp/rocketmq-client-csharp/MessageView.cs b/csharp/rocketmq-client-csharp/MessageView.cs
index 2b5529ea..57a10619 100644
--- a/csharp/rocketmq-client-csharp/MessageView.cs
+++ b/csharp/rocketmq-client-csharp/MessageView.cs
@@ -179,11 +179,10 @@ namespace Org.Apache.Rocketmq
public override string ToString()
{
- return
- $"{nameof(MessageId)}: {MessageId}, {nameof(Topic)}: {Topic}, {nameof(Tag)}: {Tag}," +
- $" {nameof(MessageGroup)}: {MessageGroup}, {nameof(DeliveryTimestamp)}: {DeliveryTimestamp}," +
- $" {nameof(Keys)}: {Keys}, {nameof(Properties)}: {Properties}, {nameof(BornHost)}: {BornHost}, " +
- $"{nameof(BornTime)}: {BornTime}, {nameof(DeliveryAttempt)}: {DeliveryAttempt}";
+ return $"{nameof(MessageId)}: {MessageId}, {nameof(Topic)}: {Topic}, {nameof(Tag)}: {Tag}," +
+ $" {nameof(MessageGroup)}: {MessageGroup}, {nameof(DeliveryTimestamp)}: {DeliveryTimestamp}," +
+ $" {nameof(Keys)}: {string.Join(", ", Keys)}, {nameof(Properties)}: {string.Join(", ", Properties.Select(kvp => kvp.ToString()))}, {nameof(BornHost)}: {BornHost}, " +
+ $"{nameof(BornTime)}: {BornTime}, {nameof(DeliveryAttempt)}: {DeliveryAttempt}";
}
}
}
\ No newline at end of file
[rocketmq-clients] 05/05: Bugfix: wrong message type judgement
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 00492cc71d69ba03668a4a0e51cfebfc32a89cc9
Author: Aaron Ai <ya...@gmail.com>
AuthorDate: Mon Mar 6 12:02:13 2023 +0800
Bugfix: wrong message type judgement
---
csharp/examples/ProducerDelayMessageExample.cs | 1 +
csharp/examples/ProducerNormalMessageExample.cs | 8 ++------
.../examples/ProducerTransactionMessageExample.cs | 4 +++-
csharp/rocketmq-client-csharp/ClientManager.cs | 2 +-
csharp/rocketmq-client-csharp/PublishingMessage.cs | 2 +-
csharp/rocketmq-client-csharp/SimpleConsumer.cs | 3 ++-
csharp/rocketmq-client-csharp/logo.png | Bin 83556 -> 85598 bytes
7 files changed, 10 insertions(+), 10 deletions(-)
diff --git a/csharp/examples/ProducerDelayMessageExample.cs b/csharp/examples/ProducerDelayMessageExample.cs
index e2e362c3..f1119245 100644
--- a/csharp/examples/ProducerDelayMessageExample.cs
+++ b/csharp/examples/ProducerDelayMessageExample.cs
@@ -31,6 +31,7 @@ namespace examples
{
const string accessKey = "yourAccessKey";
const string secretKey = "yourSecretKey";
+
// Credential provider is optional for client configuration.
var credentialsProvider = new StaticSessionCredentialsProvider(accessKey, secretKey);
const string endpoints = "foobar.com:8080";
diff --git a/csharp/examples/ProducerNormalMessageExample.cs b/csharp/examples/ProducerNormalMessageExample.cs
index 7cd3fabb..d1d6fd3d 100644
--- a/csharp/examples/ProducerNormalMessageExample.cs
+++ b/csharp/examples/ProducerNormalMessageExample.cs
@@ -61,12 +61,8 @@ namespace examples
.SetKeys("yourMessageKey-7044358f98fc")
.Build();
- for (int i = 0; i < 99999999; i++)
- {
- var sendReceipt = await producer.Send(message);
- Logger.Info($"Send message successfully, sendReceipt={sendReceipt}");
- await Task.Delay(TimeSpan.FromSeconds(1));
- }
+ var sendReceipt = await producer.Send(message);
+ Logger.Info($"Send message successfully, sendReceipt={sendReceipt}");
// Or you could close the producer manually.
// await producer.DisposeAsync();
diff --git a/csharp/examples/ProducerTransactionMessageExample.cs b/csharp/examples/ProducerTransactionMessageExample.cs
index d5c05e5b..f9c34ffa 100644
--- a/csharp/examples/ProducerTransactionMessageExample.cs
+++ b/csharp/examples/ProducerTransactionMessageExample.cs
@@ -30,6 +30,7 @@ namespace examples
{
public TransactionResolution Check(MessageView messageView)
{
+ Logger.Info("Receive transaction check, messageId={}", messageView.MessageId);
return TransactionResolution.COMMIT;
}
}
@@ -38,6 +39,7 @@ namespace examples
{
const string accessKey = "yourAccessKey";
const string secretKey = "yourSecretKey";
+
// Credential provider is optional for client configuration.
var credentialsProvider = new StaticSessionCredentialsProvider(accessKey, secretKey);
const string endpoints = "foobar.com:8080";
@@ -74,7 +76,7 @@ namespace examples
// Commit the transaction.
transaction.Commit();
// Or rollback the transaction.
- // transaction.rollback();
+ // transaction.Rollback();
// Or you could close the producer manually.
// await producer.DisposeAsync();
}
diff --git a/csharp/rocketmq-client-csharp/ClientManager.cs b/csharp/rocketmq-client-csharp/ClientManager.cs
index 498beef8..9d0b92a5 100644
--- a/csharp/rocketmq-client-csharp/ClientManager.cs
+++ b/csharp/rocketmq-client-csharp/ClientManager.cs
@@ -20,7 +20,7 @@ using System;
using System.Threading;
using System.Threading.Tasks;
using grpc = Grpc.Core;
-using System.Collections.Generic;
+using System.Collections.Generic;
namespace Org.Apache.Rocketmq
{
diff --git a/csharp/rocketmq-client-csharp/PublishingMessage.cs b/csharp/rocketmq-client-csharp/PublishingMessage.cs
index 369b7270..b5c92f51 100644
--- a/csharp/rocketmq-client-csharp/PublishingMessage.cs
+++ b/csharp/rocketmq-client-csharp/PublishingMessage.cs
@@ -67,7 +67,7 @@ namespace Org.Apache.Rocketmq
}
// For TRANSACTION message.
- if (string.IsNullOrEmpty(message.MessageGroup) || message.DeliveryTimestamp.HasValue || !txEnabled)
+ if (!string.IsNullOrEmpty(message.MessageGroup) || message.DeliveryTimestamp.HasValue || !txEnabled)
throw new InternalErrorException(
"Transactional message should not set messageGroup or deliveryTimestamp");
MessageType = MessageType.Transaction;
diff --git a/csharp/rocketmq-client-csharp/SimpleConsumer.cs b/csharp/rocketmq-client-csharp/SimpleConsumer.cs
index a9c4713b..017e7665 100644
--- a/csharp/rocketmq-client-csharp/SimpleConsumer.cs
+++ b/csharp/rocketmq-client-csharp/SimpleConsumer.cs
@@ -137,7 +137,8 @@ namespace Org.Apache.Rocketmq
{
return new Proto::HeartbeatRequest
{
- ClientType = Proto.ClientType.SimpleConsumer
+ ClientType = Proto.ClientType.SimpleConsumer,
+ Group = GetProtobufGroup()
};
}
diff --git a/csharp/rocketmq-client-csharp/logo.png b/csharp/rocketmq-client-csharp/logo.png
index 9ba4581d..88761dc1 100644
Binary files a/csharp/rocketmq-client-csharp/logo.png and b/csharp/rocketmq-client-csharp/logo.png differ
[rocketmq-clients] 03/05: Allow user to disable TLS
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 0a02f18a7d7da2f27c8a0ca3e8ad131fade5dfb1
Author: Aaron Ai <ya...@gmail.com>
AuthorDate: Mon Mar 6 10:53:18 2023 +0800
Allow user to disable TLS
---
csharp/README.md | 2 +-
csharp/examples/ProducerBenchmark.cs | 2 +-
csharp/examples/ProducerNormalMessageExample.cs | 10 +++++++--
csharp/examples/QuickStart.cs | 6 +++---
csharp/rocketmq-client-csharp/ClientConfig.cs | 15 ++++++++++++--
csharp/rocketmq-client-csharp/ClientManager.cs | 4 ++--
.../rocketmq-client-csharp/ClientMeterManager.cs | 2 +-
csharp/rocketmq-client-csharp/Endpoints.cs | 23 ++++++++++++---------
csharp/rocketmq-client-csharp/RpcClient.cs | 4 ++--
csharp/rocketmq-client-csharp/logo.png | Bin 0 -> 83556 bytes
.../rocketmq-client-csharp.csproj | 7 ++++++-
11 files changed, 50 insertions(+), 25 deletions(-)
diff --git a/csharp/README.md b/csharp/README.md
index 3de6bc65..d536cfb8 100644
--- a/csharp/README.md
+++ b/csharp/README.md
@@ -18,7 +18,7 @@ The client would be developed using the protocols outlined in [rocketmq-apis](ht
dotnet add package RocketMQ.Client
```
-You can obtain the latest version of `RocketMQ.Client` from [NuGet Gallery](https://www.nuget.org/packages/RocketMQ.Client). To assist with getting started quickly and working with various message types and clients, we offer examples [here](./examples) here.
+You can obtain the latest version of `RocketMQ.Client` from [NuGet Gallery](https://www.nuget.org/packages/RocketMQ.Client). To assist with getting started quickly and working with various message types and clients, we offer examples [here](./examples).
## Build
diff --git a/csharp/examples/ProducerBenchmark.cs b/csharp/examples/ProducerBenchmark.cs
index 50559e31..fc8d3b98 100644
--- a/csharp/examples/ProducerBenchmark.cs
+++ b/csharp/examples/ProducerBenchmark.cs
@@ -91,7 +91,7 @@ namespace examples
var tasks = new List<Task>();
while (true)
{
- Semaphore.Wait();
+ await Semaphore.WaitAsync();
Interlocked.Increment(ref _counter);
var task = producer.Send(message);
tasks.Add(task);
diff --git a/csharp/examples/ProducerNormalMessageExample.cs b/csharp/examples/ProducerNormalMessageExample.cs
index aad9fc56..7cd3fabb 100644
--- a/csharp/examples/ProducerNormalMessageExample.cs
+++ b/csharp/examples/ProducerNormalMessageExample.cs
@@ -15,6 +15,7 @@
* limitations under the License.
*/
+using System;
using System.Text;
using System.Threading.Tasks;
using NLog;
@@ -60,8 +61,13 @@ namespace examples
.SetKeys("yourMessageKey-7044358f98fc")
.Build();
- var sendReceipt = await producer.Send(message);
- Logger.Info($"Send message successfully, sendReceipt={sendReceipt}");
+ for (int i = 0; i < 99999999; i++)
+ {
+ var sendReceipt = await producer.Send(message);
+ Logger.Info($"Send message successfully, sendReceipt={sendReceipt}");
+ await Task.Delay(TimeSpan.FromSeconds(1));
+ }
+
// Or you could close the producer manually.
// await producer.DisposeAsync();
}
diff --git a/csharp/examples/QuickStart.cs b/csharp/examples/QuickStart.cs
index 174fccd2..d9730cff 100644
--- a/csharp/examples/QuickStart.cs
+++ b/csharp/examples/QuickStart.cs
@@ -22,9 +22,9 @@ namespace examples
public static void Main()
{
ProducerNormalMessageExample.QuickStart().Wait();
- // await ProducerFifoMessageExample.QuickStart();
- // await ProducerDelayMessageExample.QuickStart();
- // await SimpleConsumerExample.QuickStart();
+ // ProducerFifoMessageExample.QuickStart().Wait();
+ // ProducerDelayMessageExample.QuickStart().Wait();
+ // SimpleConsumerExample.QuickStart().Wait();
// ProducerBenchmark.QuickStart().Wait();
}
}
diff --git a/csharp/rocketmq-client-csharp/ClientConfig.cs b/csharp/rocketmq-client-csharp/ClientConfig.cs
index dc73b2da..609ad1d7 100644
--- a/csharp/rocketmq-client-csharp/ClientConfig.cs
+++ b/csharp/rocketmq-client-csharp/ClientConfig.cs
@@ -21,11 +21,13 @@ namespace Org.Apache.Rocketmq
{
public class ClientConfig : IClientConfig
{
- private ClientConfig(ISessionCredentialsProvider sessionCredentialsProvider, TimeSpan requestTimeout, string endpoints)
+ private ClientConfig(ISessionCredentialsProvider sessionCredentialsProvider, TimeSpan requestTimeout,
+ string endpoints, bool sslEnabled)
{
SessionCredentialsProvider = sessionCredentialsProvider;
RequestTimeout = requestTimeout;
Endpoints = endpoints;
+ SslEnabled = sslEnabled;
}
public ISessionCredentialsProvider SessionCredentialsProvider { get; }
@@ -34,11 +36,14 @@ namespace Org.Apache.Rocketmq
public string Endpoints { get; }
+ public bool SslEnabled { get; }
+
public class Builder
{
private ISessionCredentialsProvider _sessionCredentialsProvider;
private TimeSpan _requestTimeout = TimeSpan.FromSeconds(3);
private string _endpoints;
+ private bool _sslEnabled;
public Builder SetCredentialsProvider(ISessionCredentialsProvider sessionCredentialsProvider)
{
@@ -58,9 +63,15 @@ namespace Org.Apache.Rocketmq
return this;
}
+ public Builder EnableSsl(bool sslEnabled)
+ {
+ _sslEnabled = sslEnabled;
+ return this;
+ }
+
public ClientConfig Build()
{
- return new ClientConfig(_sessionCredentialsProvider, _requestTimeout, _endpoints);
+ return new ClientConfig(_sessionCredentialsProvider, _requestTimeout, _endpoints, _sslEnabled);
}
}
}
diff --git a/csharp/rocketmq-client-csharp/ClientManager.cs b/csharp/rocketmq-client-csharp/ClientManager.cs
index da46af30..498beef8 100644
--- a/csharp/rocketmq-client-csharp/ClientManager.cs
+++ b/csharp/rocketmq-client-csharp/ClientManager.cs
@@ -20,7 +20,7 @@ using System;
using System.Threading;
using System.Threading.Tasks;
using grpc = Grpc.Core;
-using System.Collections.Generic;
+using System.Collections.Generic;
namespace Org.Apache.Rocketmq
{
@@ -63,7 +63,7 @@ namespace Org.Apache.Rocketmq
}
// client does not exist, generate a new one
- var client = new RpcClient(endpoints);
+ var client = new RpcClient(endpoints, _client.GetClientConfig().SslEnabled);
_rpcClients.Add(endpoints, client);
return client;
}
diff --git a/csharp/rocketmq-client-csharp/ClientMeterManager.cs b/csharp/rocketmq-client-csharp/ClientMeterManager.cs
index 67801335..e8eed8d8 100644
--- a/csharp/rocketmq-client-csharp/ClientMeterManager.cs
+++ b/csharp/rocketmq-client-csharp/ClientMeterManager.cs
@@ -80,7 +80,7 @@ namespace Org.Apache.Rocketmq
.AddOtlpExporter(delegate(OtlpExporterOptions options, MetricReaderOptions readerOptions)
{
options.Protocol = OtlpExportProtocol.Grpc;
- options.Endpoint = new Uri(metric.Endpoints.GrpcTarget);
+ options.Endpoint = new Uri(metric.Endpoints.GrpcTarget(_client.GetClientConfig().SslEnabled));
options.TimeoutMilliseconds = (int)_client.GetClientConfig().RequestTimeout.TotalMilliseconds;
options.HttpClientFactory = () => _httpClient;
readerOptions.PeriodicExportingMetricReaderOptions.ExportIntervalMilliseconds =
diff --git a/csharp/rocketmq-client-csharp/Endpoints.cs b/csharp/rocketmq-client-csharp/Endpoints.cs
index 27130a33..8d560494 100644
--- a/csharp/rocketmq-client-csharp/Endpoints.cs
+++ b/csharp/rocketmq-client-csharp/Endpoints.cs
@@ -118,21 +118,24 @@ namespace Org.Apache.Rocketmq
public override string ToString()
{
- return GrpcTarget;
+ foreach (var address in Addresses)
+ {
+ return address.Host + EndpointSeparator + address.Port;
+ }
+
+ throw new ArgumentException("No available address");
}
- // TODO: Support non-TLS and multiple addresses.
- public string GrpcTarget
+ // TODO: Support multiple addresses.
+ public string GrpcTarget(bool sslEnabled)
{
- get
+ var prefix = sslEnabled ? HttpsPrefix : HttpPrefix;
+ foreach (var address in Addresses)
{
- foreach (var address in Addresses)
- {
- return HttpsPrefix + address.Host + EndpointSeparator + address.Port;
- }
-
- return "";
+ return prefix + address.Host + EndpointSeparator + address.Port;
}
+
+ throw new ArgumentException("No available address");
}
public bool Equals(Endpoints other)
diff --git a/csharp/rocketmq-client-csharp/RpcClient.cs b/csharp/rocketmq-client-csharp/RpcClient.cs
index 47db6830..e4c230b2 100644
--- a/csharp/rocketmq-client-csharp/RpcClient.cs
+++ b/csharp/rocketmq-client-csharp/RpcClient.cs
@@ -36,9 +36,9 @@ namespace Org.Apache.Rocketmq
private readonly GrpcChannel _channel;
private readonly string _target;
- public RpcClient(Endpoints endpoints)
+ public RpcClient(Endpoints endpoints, bool sslEnabled)
{
- _target = endpoints.GrpcTarget;
+ _target = endpoints.GrpcTarget(sslEnabled);
_channel = GrpcChannel.ForAddress(_target, new GrpcChannelOptions
{
HttpHandler = CreateHttpHandler()
diff --git a/csharp/rocketmq-client-csharp/logo.png b/csharp/rocketmq-client-csharp/logo.png
new file mode 100644
index 00000000..9ba4581d
Binary files /dev/null and b/csharp/rocketmq-client-csharp/logo.png differ
diff --git a/csharp/rocketmq-client-csharp/rocketmq-client-csharp.csproj b/csharp/rocketmq-client-csharp/rocketmq-client-csharp.csproj
index ab50f685..94c2d718 100644
--- a/csharp/rocketmq-client-csharp/rocketmq-client-csharp.csproj
+++ b/csharp/rocketmq-client-csharp/rocketmq-client-csharp.csproj
@@ -14,7 +14,8 @@
<Description>.NET Client for Apache RocketMQ</Description>
<PackageProjectUrl>https://github.com/apache/rocketmq-clients</PackageProjectUrl>
<RepositoryUrl>https://github.com/apache/rocketmq-clients</RepositoryUrl>
- <PackageVersion>0.0.3-SNAPSHOT</PackageVersion>
+ <PackageVersion>0.0.6-SNAPSHOT</PackageVersion>
+ <PackageIcon>logo.png</PackageIcon>
</PropertyGroup>
<ItemGroup>
@@ -40,6 +41,10 @@
<Link>Protos\google\rpc\status.proto</Link>
<Link>Protos\google\rpc\error_details.proto</Link>
</Protobuf>
+ <None Update="logo.png">
+ <Pack>True</Pack>
+ <PackagePath></PackagePath>
+ </None>
</ItemGroup>
<ItemGroup>
[rocketmq-clients] 04/05: Adjust settings sync frequency
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 c8af7a4c41fd000a129b5bf4311f8cd297677c18
Author: Aaron Ai <ya...@gmail.com>
AuthorDate: Mon Mar 6 11:30:14 2023 +0800
Adjust settings sync frequency
---
csharp/rocketmq-client-csharp/Client.cs | 7 +------
csharp/rocketmq-client-csharp/Session.cs | 13 ++++---------
2 files changed, 5 insertions(+), 15 deletions(-)
diff --git a/csharp/rocketmq-client-csharp/Client.cs b/csharp/rocketmq-client-csharp/Client.cs
index e78a8651..b3a38dc6 100644
--- a/csharp/rocketmq-client-csharp/Client.cs
+++ b/csharp/rocketmq-client-csharp/Client.cs
@@ -40,7 +40,7 @@ namespace Org.Apache.Rocketmq
private readonly CancellationTokenSource _topicRouteUpdateCts;
private static readonly TimeSpan SettingsSyncScheduleDelay = TimeSpan.FromSeconds(1);
- private static readonly TimeSpan SettingsSyncSchedulePeriod = TimeSpan.FromSeconds(1);
+ private static readonly TimeSpan SettingsSyncSchedulePeriod = TimeSpan.FromMinutes(5);
private readonly CancellationTokenSource _settingsSyncCts;
private static readonly TimeSpan StatsScheduleDelay = TimeSpan.FromSeconds(60);
@@ -227,11 +227,6 @@ namespace Org.Apache.Rocketmq
Logger.Error(e, $"Failed to update topic route cache, topic={item}");
}
}
-
- foreach (var topic in GetTopics())
- {
- await FetchTopicRoute(topic);
- }
}
catch (Exception e)
{
diff --git a/csharp/rocketmq-client-csharp/Session.cs b/csharp/rocketmq-client-csharp/Session.cs
index c6e98418..86316b9d 100644
--- a/csharp/rocketmq-client-csharp/Session.cs
+++ b/csharp/rocketmq-client-csharp/Session.cs
@@ -17,7 +17,6 @@
using System;
using System.Threading;
-using System.Threading.Channels;
using System.Threading.Tasks;
using Grpc.Core;
using grpc = Grpc.Core;
@@ -32,12 +31,12 @@ namespace Org.Apache.Rocketmq
private static readonly Logger Logger = MqLogManager.Instance.GetCurrentClassLogger();
private static readonly TimeSpan SettingsInitializationTimeout = TimeSpan.FromSeconds(3);
+ private readonly ManualResetEventSlim _event = new(false);
private readonly AsyncDuplexStreamingCall<Proto::TelemetryCommand, Proto::TelemetryCommand>
_streamingCall;
private readonly Client _client;
- private readonly Channel<bool> _channel;
private readonly Endpoints _endpoints;
private readonly SemaphoreSlim _semaphore;
@@ -49,10 +48,6 @@ namespace Org.Apache.Rocketmq
_semaphore = new SemaphoreSlim(1);
_streamingCall = streamingCall;
_client = client;
- _channel = Channel.CreateBounded<bool>(new BoundedChannelOptions(1)
- {
- FullMode = BoundedChannelFullMode.DropOldest
- });
Loop();
}
@@ -66,7 +61,7 @@ namespace Org.Apache.Rocketmq
public async Task SyncSettings(bool awaitResp)
{
// Add more buffer time.
- await _semaphore.WaitAsync(_client.GetClientConfig().RequestTimeout.Add(SettingsInitializationTimeout));
+ await _semaphore.WaitAsync();
try
{
var settings = _client.GetSettings();
@@ -78,7 +73,7 @@ namespace Org.Apache.Rocketmq
// await writer.CompleteAsync();
if (awaitResp)
{
- await _channel.Reader.ReadAsync();
+ _event.Wait(_client.GetClientConfig().RequestTimeout.Add(SettingsInitializationTimeout));
}
}
finally
@@ -100,7 +95,7 @@ namespace Org.Apache.Rocketmq
Logger.Info(
$"Receive setting from remote, endpoints={_endpoints}, clientId={_client.GetClientId()}");
_client.OnSettingsCommand(_endpoints, response.Settings);
- await _channel.Writer.WriteAsync(true);
+ _event.Set();
break;
}
case Proto.TelemetryCommand.CommandOneofCase.RecoverOrphanedTransactionCommand: