Skip to content

Commit 4800b86

Browse files
nnhynnhy
committed
CloudProvider统一与依赖升级及多项细节优化
- Producer/Consumer/测试切换为 CloudProvider 属性，支持多云适配 - 升级 NewLife.Core、Microsoft.NET.Test.Sdk 依赖 - HttpClient 实例化与命名空间简化 - gRPC 方法签名与类型前缀优化，便于扩展 - 批量消息解析与 MqCodec 逻辑修正 - 多项测试用例与注释风格一致性调整
1 parent a0afd6c commit 4800b86

17 files changed

Lines changed: 62 additions & 44 deletions

‎NewLife.RocketMQ/AliyunProvider.cs‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,6 @@
1-
namespace NewLife.RocketMQ;
1+
using System.Net.Http;
2+
3+
namespace NewLife.RocketMQ;
24

35
/// <summary>阿里云 RocketMQ 适配器</summary>
46
public class AliyunProvider : ICloudProvider
@@ -53,7 +55,7 @@ public String GetNameServerAddress()
5355
if (String.IsNullOrEmpty(addr) || !addr.StartsWith("http", StringComparison.OrdinalIgnoreCase))
5456
return null;
5557

56-
var http = new System.Net.Http.HttpClient();
58+
var http = new HttpClient();
5759
var html = http.GetStringAsync(addr).ConfigureAwait(false).GetAwaiter().GetResult();
5860

5961
return String.IsNullOrWhiteSpace(html) ? null : html.Trim();

‎NewLife.RocketMQ/Consumer.cs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1276,7 +1276,7 @@ private Dictionary<String, Int64> LoadLocalOffsets()
12761276
var json = File.ReadAllText(file);
12771277
if (String.IsNullOrWhiteSpace(json)) return [];
12781278

1279-
return JsonHelper.Default.Read<Dictionary<String, Int64>>(json) ?? [];
1279+
return JsonHelper.Default.Read<Dictionary<String, Int64>>(json, null) ?? [];
12801280
}
12811281
catch
12821282
{

‎NewLife.RocketMQ/MqBase.cs‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,9 @@
11
using System.Collections.Concurrent;
22
using System.Diagnostics;
3+
using System.Net.Http;
34
using System.Reflection;
45
using System.Security.Authentication;
56
using System.Security.Cryptography.X509Certificates;
6-
using System.Xml.Serialization;
77
using NewLife.Log;
88
using NewLife.Net;
99
using NewLife.RocketMQ.Protocol;
@@ -345,7 +345,7 @@ protected virtual void OnStart()
345345
var server = _aliyunOptions?.Server;
346346
if (!server.IsNullOrEmpty() && server.StartsWithIgnoreCase("http"))
347347
{
348-
var http = new System.Net.Http.HttpClient();
348+
var http = new HttpClient();
349349
var html = http.GetStringAsync(server).ConfigureAwait(false).GetAwaiter().GetResult();
350350

351351
if (!html.IsNullOrWhiteSpace()) NameServerAddress = html.Trim();
@@ -876,7 +876,7 @@ public virtual Int32 RegisterFilterServer(String filterServerAddr)
876876
/// <param name="topic">主题。默认使用当前Topic</param>
877877
/// <param name="cancellationToken">取消通知</param>
878878
/// <returns>路由查询结果</returns>
879-
public async Task<Grpc.QueryRouteResponse> QueryRouteViaGrpcAsync(String topic = null, CancellationToken cancellationToken = default)
879+
public virtual async Task<Grpc.QueryRouteResponse> QueryRouteViaGrpcAsync(String topic = null, CancellationToken cancellationToken = default)
880880
{
881881
if (_GrpcService == null) throw new InvalidOperationException("gRPC service not initialized. Set GrpcProxyAddress first.");
882882

‎NewLife.RocketMQ/NewLife.RocketMQ.csproj‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -62,7 +62,7 @@
6262
</ItemGroup>
6363

6464
<ItemGroup>
65-
<PackageReference Include="NewLife.Core" Version="11.16.2026.601" />
65+
<PackageReference Include="NewLife.Core" Version="11.17.2026.706-beta0929" />
6666
</ItemGroup>
6767

6868
</Project>

‎NewLife.RocketMQ/Producer.cs‎

Lines changed: 12 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@
55
using NewLife.Reflection;
66
using NewLife.RocketMQ.Client;
77
using NewLife.RocketMQ.Common;
8+
using NewLife.RocketMQ.Grpc;
89
using NewLife.RocketMQ.MessageTrace;
910
using NewLife.RocketMQ.Models;
1011
using NewLife.RocketMQ.Protocol;
@@ -152,9 +153,9 @@ public virtual SendResult Publish(Message message, MessageQueue queue, Int32 tim
152153
foreach (var hook in _sendMessageHooks)
153154
{
154155
try { hook.ExecuteHookBefore(context); }
155-
catch (Exception e)
156-
{
157-
if (Log.Enable) Log.Error(e.Message);
156+
catch (Exception e)
157+
{
158+
if (Log.Enable) Log.Error(e.Message);
158159
}
159160
}
160161

@@ -434,7 +435,7 @@ private async Task<SendResult> PublishViaGrpcAsync(Message message, Cancellation
434435
cancellationToken: cancellationToken
435436
).ConfigureAwait(false);
436437

437-
if (rs.Status?.Code != Grpc.GrpcCode.OK)
438+
if (rs.Status?.Code != GrpcCode.OK)
438439
throw new InvalidOperationException($"gRPC SendMessage failed: {rs.Status}");
439440

440441
var entry = rs.Entries.FirstOrDefault();
@@ -1015,7 +1016,7 @@ public virtual async Task<MessageExt> RequestAsync(Message message, Int32 timeou
10151016

10161017
// 等待响应，使用兼容的方式
10171018
var completedTask = await Task.WhenAny(tcs.Task, Task.Delay(timeout, cancellationToken)).ConfigureAwait(false);
1018-
1019+
10191020
if (completedTask == tcs.Task)
10201021
{
10211022
return await tcs.Task.ConfigureAwait(false);
@@ -1120,7 +1121,7 @@ private Command HandleCheckTransaction(Command cmd)
11201121
var bk = Clients?.FirstOrDefault();
11211122
bk?.Invoke(RequestCode.END_TRANSACTION, null, header.GetProperties());
11221123

1123-
WriteLog("事务回查完成，事务ID={0}，状态={1}", transactionId, state);
1124+
WriteLog("事务回查完成，事务ID={0}，状态={1}", transactionId, state);
11241125
}
11251126
catch (Exception ex)
11261127
{
@@ -1163,7 +1164,7 @@ public async Task<SendResult> PublishDelayViaGrpcAsync(
11631164
cancellationToken: cancellationToken
11641165
).ConfigureAwait(false);
11651166

1166-
if (rs.Status?.Code != Grpc.GrpcCode.OK)
1167+
if (rs.Status?.Code != GrpcCode.OK)
11671168
throw new InvalidOperationException($"gRPC SendMessage (delay) failed: {rs.Status}");
11681169

11691170
var entry = rs.Entries.FirstOrDefault();
@@ -1208,7 +1209,7 @@ public async Task<SendResult> PublishTransactionViaGrpcAsync(
12081209
cancellationToken: cancellationToken
12091210
).ConfigureAwait(false);
12101211

1211-
if (rs.Status?.Code != Grpc.GrpcCode.OK)
1212+
if (rs.Status?.Code != GrpcCode.OK)
12121213
throw new InvalidOperationException($"gRPC SendTransactionMessage failed: {rs.Status}");
12131214

12141215
var entry = rs.Entries.FirstOrDefault();
@@ -1233,7 +1234,7 @@ public async Task<SendResult> PublishTransactionViaGrpcAsync(
12331234
/// <param name="commit">是否提交。true提交，false回滚</param>
12341235
/// <param name="cancellationToken">取消通知</param>
12351236
/// <returns></returns>
1236-
public async Task<Grpc.GrpcEndTransactionResponse> EndTransactionViaGrpcAsync(
1237+
public async Task<GrpcEndTransactionResponse> EndTransactionViaGrpcAsync(
12371238
String messageId,
12381239
String transactionId,
12391240
Boolean commit,
@@ -1245,7 +1246,7 @@ public async Task<SendResult> PublishTransactionViaGrpcAsync(
12451246
Topic,
12461247
messageId,
12471248
transactionId,
1248-
commit ? Grpc.GrpcTransactionResolution.COMMIT : Grpc.GrpcTransactionResolution.ROLLBACK,
1249+
commit ? GrpcTransactionResolution.COMMIT : GrpcTransactionResolution.ROLLBACK,
12491250
cancellationToken
12501251
).ConfigureAwait(false);
12511252
}
@@ -1254,7 +1255,7 @@ public async Task<SendResult> PublishTransactionViaGrpcAsync(
12541255
/// <param name="topic">主题名。默认使用当前Topic</param>
12551256
/// <param name="cancellationToken">取消通知</param>
12561257
/// <returns>路由信息</returns>
1257-
public async Task<Grpc.QueryRouteResponse> QueryRouteViaGrpcAsync(String topic = null, CancellationToken cancellationToken = default)
1258+
public override async Task<QueryRouteResponse> QueryRouteViaGrpcAsync(String topic = null, CancellationToken cancellationToken = default)
12581259
{
12591260
if (_GrpcService == null) throw new InvalidOperationException("gRPC service not initialized. Set GrpcProxyAddress first.");
12601261

‎NewLife.RocketMQ/Protocol/MessageExt.cs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -179,7 +179,7 @@ public static IList<MessageExt> ReadAll(IPacket body)
179179
var reader = new SpanReader(buf) { IsLittleEndian = false };
180180

181181
var list = new List<MessageExt>();
182-
while (reader.FreeCapacity > 0)
182+
while (reader.Available > 0)
183183
{
184184
var msg = new MessageExt();
185185
if (!msg.Read(ref reader)) break;
@@ -210,7 +210,7 @@ public static IList<MessageExt> DecodeBatch(MessageExt parent)
210210
var list = new List<MessageExt>();
211211
var reader = new SpanReader(parent.Body) { IsLittleEndian = false };
212212

213-
while (reader.FreeCapacity > 0)
213+
while (reader.Available > 0)
214214
{
215215
try
216216
{

‎NewLife.RocketMQ/Protocol/MqCodec.cs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -39,7 +39,7 @@ protected override IEnumerable<Command> Decode(IHandlerContext context, IPacket
3939
{
4040
var ss = context.Owner as IExtend;
4141
if (ss["Codec"] is not PacketCodec pc)
42-
ss["Codec"] = pc = new PacketCodec { GetLength = p => GetLength(p, 0, -4) };
42+
ss["Codec"] = pc = new PacketCodec { GetLength2 = p => GetLength(p, 0, -4) };
4343

4444
var pks = pc.Parse(pk);
4545
foreach (var e in pks)

‎NewLife.RocketMQ/Protocol/TopicMessageType.cs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
namespace NewLife.RocketMQ.Protocol;
22

3-
/// <summary>Topic 消息类型。用于 <see cref="NewLife.RocketMQ.MqBase.CreateTopic"/> 的 topicMessageType 请求头字段。</summary>
3+
/// <summary>Topic 消息类型。用于 <c>MqBase.CreateTopic</c> 的 topicMessageType 请求头字段。</summary>
44
/// <remarks>
55
/// 对应 RocketMQ Broker 端的 <c>TopicMessageType</c> 枚举。
66
/// LITE 类型需要 RocketMQ 5.5.0+ Broker（RIP-83），旧版 Broker 会忽略该字段并以 Normal 处理。

‎Test/Program.cs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -124,11 +124,11 @@ static void TestAliyun()
124124
Group = "GID_newlife_Group02",
125125
NameServerAddress = "http://MQ_INST_1827694722767531_BXxCwUhm.mq-internet-access.mq-internet.aliyuncs.com:80",
126126

127-
Aliyun = new AliyunOptions
127+
CloudProvider = new AliyunProvider
128128
{
129129
AccessKey = "LTAI5tKTGShu31C61xRARVC4",
130130
SecretKey = "a9oPwph1IcMGanWckzUOwOf3Ork8LO",
131-
//InstanceId = "MQ_INST_1827694722767531_BXxCwUhm",
131+
InstanceId = "MQ_INST_1827694722767531_BXxCwUhm",
132132
},
133133

134134
FromLastOffset = true,

‎XUnitTestRocketMQ/Cloud/AliyunIssuesTests.cs‎

Lines changed: 12 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,12 @@ public void ProducerForAliyun_Test()
3131
var producer = new Producer()
3232
{
3333
Topic = _testTopic,
34-
Aliyun = _aliyunOptions,
34+
CloudProvider = new AliyunProvider
35+
{
36+
AccessKey = _aliyunOptions.AccessKey,
37+
SecretKey = _aliyunOptions.SecretKey,
38+
InstanceId = _aliyunOptions.InstanceId,
39+
},
3540
//NameServerAddress = "http://MQ_INST_xxxxxxxxxx_AXxCwUhm.mq-internet-access.mq-internet.aliyuncs.com:80",
3641
//如果不用上面的默认Server地址，直接将NameServerAddress设为你自己的TCP公网接收点地址也是可以的
3742
};
@@ -56,7 +61,12 @@ public void ConsumerForAliyun_Test()
5661
var consumer = new Consumer()
5762
{
5863
Topic = _testTopic,
59-
Aliyun = _aliyunOptions,
64+
CloudProvider = new AliyunProvider
65+
{
66+
AccessKey = _aliyunOptions.AccessKey,
67+
SecretKey = _aliyunOptions.SecretKey,
68+
InstanceId = _aliyunOptions.InstanceId,
69+
},
6070
Group = _testGroup,
6171
FromLastOffset = true,
6272
BatchSize = 1,

0 commit comments

Comments
 (0)