This is an automated email from the ASF dual-hosted git repository.
RongtongJin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/rocketmq-clients.git
The following commit(s) were added to refs/heads/master by this push:
new 2bbe25b5 [csharp] Add LiteSimpleConsumer with tests and example (#1380)
2bbe25b5 is described below
commit 2bbe25b5682591887bdd9973305774e62792c0d7
Author: zhaohai <[email protected]>
AuthorDate: Wed Sep 16 14:08:37 2026 +0800
[csharp] Add LiteSimpleConsumer with tests and example (#1380)
Implement the lite-topic variant of SimpleConsumer for the C# client,
mirroring the Java/Node.js reference implementation.
Implementation
- LiteSimpleConsumer: consumer bound to a single parent (bind) topic that
dynamically (un)subscribes lite topics via SyncLiteSubscription. Supports
SubscribeLite(liteTopic), SubscribeLite(liteTopic, OffsetOption),
UnsubscribeLite(liteTopic), GetLiteTopicSet(), GetBindTopic() plus the
inherited SimpleConsumer receive/ack/changeInvisibleDuration APIs.
- LiteSimpleSubscriptionSettings: SimpleSubscriptionSettings with
ClientType.LITE_SIMPLE_CONSUMER so the server derives LITE_SELECTIVE mode.
- ClientType: add LiteSimpleConsumer and map it to the protobuf enum.
- LiteSubscriptionManager: generalize from PushConsumer to the Consumer base
(both lite consumer kinds share it) and add Shutdown() to dispose the
periodic sync timer.
- Consumer: add shared
CheckRunning/GetRequestTimeout/Namespace/SyncLiteSubscription
and recognize LiteSimpleSubscriptionSettings; PushConsumer now overrides
them.
- MessageView: expose LiteTopic parsed from system properties.
- SubscriptionLoadBalancer: throw NotFoundException instead of dividing by
zero when the route carries no queue.
Defect fix: missing liteTopic in ack
- WrapAckMessageRequest / WrapChangeInvisibleDuration now set LiteTopic from
MessageView.LiteTopic. Without it the proxy cannot resolve the LMQ receipt
handle and answers 50001 INTERNAL_SERVER_ERROR.
Tests
- tests/LiteSimpleConsumerTest.cs: 20 offline unit tests (builder
validation,
settings/heartbeat client type, subscribe/unsubscribe sync actions, dedup,
quota, notify-unsubscribe command, liteTopic carried in ack and
changeInvisibleDuration, route pruning).
- tests/LiteSimpleConsumerIntegrationTest.cs: 3 real-cluster tests, guarded
by
ROCKETMQ_CSHARP_LITE_ENDPOINTS. Verified against a local RocketMQ 5.5.1
cluster (broker enableLmq/enableMultiDispatch, proxy in CLUSTER mode,
message.type=LITE parent topic, group with +lite.bind.topic): 3/3 passed.
Example
- examples/LiteSimpleConsumerExample.cs and its QuickStart wiring.
---
csharp/examples/LiteSimpleConsumerExample.cs | 135 ++++++++
csharp/examples/QuickStart.cs | 3 +
csharp/rocketmq-client-csharp/ClientType.cs | 4 +-
csharp/rocketmq-client-csharp/Consumer.cs | 38 ++-
.../rocketmq-client-csharp/ILiteSimpleConsumer.cs | 60 ++++
.../rocketmq-client-csharp/LiteSimpleConsumer.cs | 275 +++++++++++++++++
.../LiteSimpleSubscriptionSettings.cs | 46 +++
.../LiteSubscriptionManager.cs | 18 +-
csharp/rocketmq-client-csharp/MessageView.cs | 11 +-
csharp/rocketmq-client-csharp/PushConsumer.cs | 25 +-
csharp/rocketmq-client-csharp/SimpleConsumer.cs | 6 +-
.../SubscriptionLoadBalancer.cs | 8 +
csharp/tests/LiteSimpleConsumerIntegrationTest.cs | 194 ++++++++++++
csharp/tests/LiteSimpleConsumerTest.cs | 338 +++++++++++++++++++++
14 files changed, 1124 insertions(+), 37 deletions(-)
diff --git a/csharp/examples/LiteSimpleConsumerExample.cs
b/csharp/examples/LiteSimpleConsumerExample.cs
new file mode 100644
index 00000000..e4da94d0
--- /dev/null
+++ b/csharp/examples/LiteSimpleConsumerExample.cs
@@ -0,0 +1,135 @@
+/*
+ * 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.Text;
+using System.Threading.Tasks;
+using Microsoft.Extensions.Logging;
+using Org.Apache.Rocketmq;
+
+namespace examples
+{
+ /// <summary>
+ /// Demonstrates how to consume lite topics with LiteSimpleConsumer, which
pulls
+ /// messages explicitly and acks them one by one.
+ ///
+ /// Key Points:
+ /// - The consumer binds to one parent topic, all lite topics belong to it
+ /// - SubscribeLite() / UnsubscribeLite() change the lite topic set at
runtime
+ /// - Receive() returns a batch, Ack() confirms the consumption of each
message
+ /// - The lite topic is reported in ack requests automatically
+ ///
+ /// Prerequisites:
+ /// - Broker: enableLmq=true, enableMultiDispatch=true
+ /// - Parent topic created with message.type=LITE
+ /// - Consumer group created with the attribute
+lite.bind.topic=<parentTopic>
+ /// </summary>
+ internal static class LiteSimpleConsumerExample
+ {
+ private static readonly ILogger Logger =
MqLogManager.CreateLogger(typeof(LiteSimpleConsumerExample).FullName);
+
+ private static readonly string Endpoint =
Environment.GetEnvironmentVariable("ROCKETMQ_ENDPOINT") ?? "127.0.0.1:8081";
+ private const string BindTopic = "topic-lite";
+ private const string ConsumerGroup = "GID-lite-simple-consumer";
+ private static readonly TimeSpan AwaitDuration =
TimeSpan.FromSeconds(5);
+ private static readonly TimeSpan InvisibleDuration =
TimeSpan.FromSeconds(15);
+
+ internal static async Task QuickStart()
+ {
+ var clientConfig = new ClientConfig.Builder()
+ .SetEndpoints(Endpoint)
+ .Build();
+
+ var liteTopic =
$"lite-topic-{DateTimeOffset.UtcNow.ToUnixTimeMilliseconds()}";
+
+ // Send some lite messages first, the producer publishes them to
the parent topic
+ // and the broker indexes them under the lite topic.
+ await SendLiteMessages(clientConfig, BindTopic, liteTopic);
+
+ // Build the lite simple consumer bound to the parent topic.
+ Logger.LogInformation($"Creating LiteSimpleConsumer,
bindTopic={BindTopic}, consumerGroup={ConsumerGroup}");
+ var consumer = await new LiteSimpleConsumer.Builder()
+ .SetClientConfig(clientConfig)
+ .SetConsumerGroup(ConsumerGroup)
+ .SetBindTopic(BindTopic)
+ .SetAwaitDuration(AwaitDuration)
+ .Build();
+
+ try
+ {
+ // Consume from the minimum offset so that messages sent
before the
+ // subscription are delivered as well.
+ await consumer.SubscribeLite(liteTopic,
OffsetOption.MinOffset);
+ Logger.LogInformation($"Subscribed to {liteTopic}, lite topic
set=[{string.Join(", ", consumer.GetLiteTopicSet())}]");
+
+ var received = new List<string>();
+ var deadline = DateTime.UtcNow.AddSeconds(30);
+ while (received.Count < 5 && DateTime.UtcNow < deadline)
+ {
+ var messages = await consumer.Receive(16,
InvisibleDuration);
+ foreach (var message in messages)
+ {
+ Logger.LogInformation($"Received message,
messageId={message.MessageId}, topic={message.Topic}, " +
+ $"liteTopic={message.LiteTopic},
body={Encoding.UTF8.GetString(message.Body)}");
+ await consumer.Ack(message);
+ received.Add(message.MessageId);
+ }
+ }
+
+ Logger.LogInformation($"Received and acked {received.Count}
lite message(s)");
+
+ await consumer.UnsubscribeLite(liteTopic);
+ Logger.LogInformation($"Unsubscribed from {liteTopic}, lite
topic set=[{string.Join(", ", consumer.GetLiteTopicSet())}]");
+ }
+ finally
+ {
+ await consumer.DisposeAsync();
+ Logger.LogInformation("LiteSimpleConsumer closed");
+ }
+ }
+
+ private static async Task SendLiteMessages(ClientConfig clientConfig,
string parentTopic, string liteTopic)
+ {
+ var producer = await new Producer.Builder()
+ .SetTopics(parentTopic)
+ .SetClientConfig(clientConfig)
+ .Build();
+
+ try
+ {
+ for (var i = 0; i < 5; i++)
+ {
+ var message = new Message.Builder()
+ .SetTopic(parentTopic)
+
.SetBody(Encoding.UTF8.GetBytes($"lite-simple-consumer-body-{i}"))
+ .SetTag("LiteTest")
+ .SetKeys($"lite-{i}")
+ .SetLiteTopic(liteTopic)
+ .Build();
+
+ var receipt = await producer.Send(message);
+ Logger.LogInformation($"Sent lite message,
messageId={receipt.MessageId}, liteTopic={liteTopic}");
+ }
+ }
+ finally
+ {
+ await producer.DisposeAsync();
+ }
+ }
+ }
+}
diff --git a/csharp/examples/QuickStart.cs b/csharp/examples/QuickStart.cs
index ec5992d2..b4ae4e8b 100644
--- a/csharp/examples/QuickStart.cs
+++ b/csharp/examples/QuickStart.cs
@@ -37,6 +37,9 @@ namespace examples
// PushConsumerExample.QuickStart().Wait();
// SimpleConsumerExample.QuickStart().Wait();
// ProducerBenchmark.QuickStart().Wait();
+ // ProducerLiteMessageExample.QuickStart().Wait();
+ // LitePushConsumerExample.QuickStart().Wait();
+ // LiteSimpleConsumerExample.QuickStart().Wait();
}
}
}
\ No newline at end of file
diff --git a/csharp/rocketmq-client-csharp/ClientType.cs
b/csharp/rocketmq-client-csharp/ClientType.cs
index a2c7cf78..268c5429 100644
--- a/csharp/rocketmq-client-csharp/ClientType.cs
+++ b/csharp/rocketmq-client-csharp/ClientType.cs
@@ -24,7 +24,8 @@ namespace Org.Apache.Rocketmq
Producer,
SimpleConsumer,
PushConsumer,
- LitePushConsumer
+ LitePushConsumer,
+ LiteSimpleConsumer
}
public static class ClientTypeHelper
@@ -37,6 +38,7 @@ namespace Org.Apache.Rocketmq
ClientType.SimpleConsumer => Proto.ClientType.SimpleConsumer,
ClientType.PushConsumer => Proto.ClientType.PushConsumer,
ClientType.LitePushConsumer =>
Proto.ClientType.LitePushConsumer,
+ ClientType.LiteSimpleConsumer =>
Proto.ClientType.LiteSimpleConsumer,
_ => Proto.ClientType.Unspecified
};
}
diff --git a/csharp/rocketmq-client-csharp/Consumer.cs
b/csharp/rocketmq-client-csharp/Consumer.cs
index ad3e3551..8e62d581 100644
--- a/csharp/rocketmq-client-csharp/Consumer.cs
+++ b/csharp/rocketmq-client-csharp/Consumer.cs
@@ -46,10 +46,42 @@ namespace Org.Apache.Rocketmq
/// <returns>True if this is a lite consumer, false
otherwise.</returns>
public bool IsLiteConsumer()
{
- // For now, we check if GetSettings returns
LitePushSubscriptionSettings
- // This can be extended when LiteSimpleConsumer is implemented
var settings = GetSettings();
- return settings is LitePushSubscriptionSettings;
+ return settings is LitePushSubscriptionSettings || settings is
LiteSimpleSubscriptionSettings;
+ }
+
+ /// <summary>
+ /// Check if the consumer is running.
+ /// </summary>
+ internal virtual void CheckRunning()
+ {
+ if (State != State.Running)
+ {
+ throw new InvalidOperationException("Consumer is not running");
+ }
+ }
+
+ /// <summary>
+ /// Get the request timeout from client config.
+ /// </summary>
+ internal TimeSpan GetRequestTimeout()
+ {
+ return ClientConfig.RequestTimeout;
+ }
+
+ /// <summary>
+ /// Get the namespace from client config.
+ /// </summary>
+ internal string Namespace => ClientConfig.Namespace;
+
+ /// <summary>
+ /// Sync lite subscription with the server, shared by all lite
consumers.
+ /// </summary>
+ internal async Task<Proto.SyncLiteSubscriptionResponse>
SyncLiteSubscription(
+ Proto.SyncLiteSubscriptionRequest request, TimeSpan timeout)
+ {
+ var invocation = await
ClientManager.SyncLiteSubscription(Endpoints, request, timeout);
+ return invocation.Response;
}
/// <summary>
diff --git a/csharp/rocketmq-client-csharp/ILiteSimpleConsumer.cs
b/csharp/rocketmq-client-csharp/ILiteSimpleConsumer.cs
new file mode 100644
index 00000000..f6e5219c
--- /dev/null
+++ b/csharp/rocketmq-client-csharp/ILiteSimpleConsumer.cs
@@ -0,0 +1,60 @@
+/*
+ * 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.Collections.Generic;
+using System.Threading.Tasks;
+
+namespace Org.Apache.Rocketmq
+{
+ /// <summary>
+ /// Lite simple consumer, a simple consumer that is bound to one parent
topic and
+ /// consumes messages from the lite topics it subscribes to.
+ /// </summary>
+ public interface ILiteSimpleConsumer
+ {
+ /// <summary>
+ /// Subscribe to a lite topic.
+ /// </summary>
+ /// <param name="liteTopic">The name of the lite topic to
subscribe</param>
+ Task SubscribeLite(string liteTopic);
+
+ /// <summary>
+ /// Subscribe to a lite topic with an offset option to specify the
consume from offset.
+ /// </summary>
+ /// <param name="liteTopic">The name of the lite topic to
subscribe</param>
+ /// <param name="offsetOption">The consume from offset option</param>
+ Task SubscribeLite(string liteTopic, OffsetOption offsetOption);
+
+ /// <summary>
+ /// Unsubscribe from a lite topic.
+ /// </summary>
+ /// <param name="liteTopic">The name of the lite topic to unsubscribe
from</param>
+ Task UnsubscribeLite(string liteTopic);
+
+ /// <summary>
+ /// Get the lite topic immutable set.
+ /// </summary>
+ /// <returns>Lite topic immutable set</returns>
+ ISet<string> GetLiteTopicSet();
+
+ /// <summary>
+ /// Get the parent topic this consumer binds to.
+ /// </summary>
+ /// <returns>The bound parent topic</returns>
+ string GetBindTopic();
+ }
+}
diff --git a/csharp/rocketmq-client-csharp/LiteSimpleConsumer.cs
b/csharp/rocketmq-client-csharp/LiteSimpleConsumer.cs
new file mode 100644
index 00000000..2ef09981
--- /dev/null
+++ b/csharp/rocketmq-client-csharp/LiteSimpleConsumer.cs
@@ -0,0 +1,275 @@
+/*
+ * 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.Concurrent;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+using Microsoft.Extensions.Logging;
+using Proto = Apache.Rocketmq.V2;
+
+namespace Org.Apache.Rocketmq
+{
+ /// <summary>
+ /// Lite simple consumer extends the standard simple consumer with lite
topic support.
+ /// It binds to one parent topic and (un)subscribes lite topics at
runtime, then pulls
+ /// and acks messages from them explicitly.
+ /// </summary>
+ public class LiteSimpleConsumer : SimpleConsumer, ILiteSimpleConsumer
+ {
+ private static readonly ILogger Logger =
MqLogManager.CreateLogger<LiteSimpleConsumer>();
+
+ private readonly LiteSubscriptionManager _liteSubscriptionManager;
+ private readonly string _bindTopic;
+ private readonly LiteSimpleSubscriptionSettings
_liteSimpleSubscriptionSettings;
+
+ /// <summary>
+ /// Creates a new instance of LiteSimpleConsumer.
+ /// </summary>
+ /// <param name="clientConfig">Client configuration</param>
+ /// <param name="consumerGroup">Consumer group name</param>
+ /// <param name="awaitDuration">Long-polling timeout of receive
requests</param>
+ /// <param name="bindTopic">The parent topic all lite topics belong
to</param>
+ public LiteSimpleConsumer(ClientConfig clientConfig, string
consumerGroup, TimeSpan awaitDuration,
+ string bindTopic)
+ : this(clientConfig, consumerGroup, awaitDuration, bindTopic,
+ new ConcurrentDictionary<string, FilterExpression> {
[bindTopic] = FilterExpression.SubAll })
+ {
+ }
+
+ private LiteSimpleConsumer(ClientConfig clientConfig, string
consumerGroup, TimeSpan awaitDuration,
+ string bindTopic, ConcurrentDictionary<string, FilterExpression>
subscriptionExpressions)
+ : base(clientConfig, consumerGroup, awaitDuration,
subscriptionExpressions.ToDictionary(kv => kv.Key, kv => kv.Value))
+ {
+ if (string.IsNullOrWhiteSpace(bindTopic))
+ {
+ throw new ArgumentException("bindTopic cannot be null or
empty", nameof(bindTopic));
+ }
+
+ _bindTopic = bindTopic;
+ _liteSubscriptionManager = new LiteSubscriptionManager(this,
bindTopic, consumerGroup);
+ _liteSimpleSubscriptionSettings = new
LiteSimpleSubscriptionSettings(clientConfig.Namespace, ClientId,
+ Endpoints, consumerGroup, clientConfig.RequestTimeout,
awaitDuration, subscriptionExpressions);
+ }
+
+ protected override async Task Start()
+ {
+ await base.Start();
+
+ // Fetch the route of the bind topic up-front, the lite
subscription sync and
+ // the receive requests both need it.
+ try
+ {
+ var routeData = await GetRouteData(_bindTopic);
+ Logger.LogInformation($"Fetched route for bind topic:
{_bindTopic}, " +
+
$"messageQueueCount={routeData.MessageQueues.Count}, clientId={ClientId}");
+ }
+ catch (Exception ex)
+ {
+ Logger.LogError(ex, $"Failed to fetch route for bind topic:
{_bindTopic}, clientId={ClientId}. " +
+ "Lite subscription sync may be affected.");
+ }
+
+ _liteSubscriptionManager.Start();
+ Logger.LogInformation($"LiteSimpleConsumer started,
clientId={ClientId}, bindTopic={_bindTopic}, " +
+ $"consumerGroup={ConsumerGroup}");
+ }
+
+ protected override async Task Shutdown()
+ {
+ _liteSubscriptionManager.Shutdown();
+ await base.Shutdown();
+ }
+
+ /// <summary>
+ /// Subscribe to a lite topic.
+ /// </summary>
+ /// <param name="liteTopic">The name of the lite topic to
subscribe</param>
+ public async Task SubscribeLite(string liteTopic)
+ {
+ await _liteSubscriptionManager.SubscribeLite(liteTopic, null);
+ }
+
+ /// <summary>
+ /// Subscribe to a lite topic with an offsetOption to specify the
consume from offset.
+ /// </summary>
+ /// <param name="liteTopic">The name of the lite topic to
subscribe</param>
+ /// <param name="offsetOption">The consume from offset option. If
null, uses the default offset policy.</param>
+ public async Task SubscribeLite(string liteTopic, OffsetOption
offsetOption)
+ {
+ await _liteSubscriptionManager.SubscribeLite(liteTopic,
offsetOption);
+ }
+
+ /// <summary>
+ /// Unsubscribe from a lite topic.
+ /// </summary>
+ /// <param name="liteTopic">The name of the lite topic to unsubscribe
from</param>
+ public async Task UnsubscribeLite(string liteTopic)
+ {
+ await _liteSubscriptionManager.UnsubscribeLite(liteTopic);
+ }
+
+ /// <summary>
+ /// Get the lite topic immutable set.
+ /// </summary>
+ /// <returns>Lite topic immutable set</returns>
+ public ISet<string> GetLiteTopicSet()
+ {
+ return _liteSubscriptionManager.GetLiteTopicSet();
+ }
+
+ /// <summary>
+ /// Get the parent topic this consumer binds to.
+ /// </summary>
+ /// <returns>The bound parent topic</returns>
+ public string GetBindTopic()
+ {
+ return _bindTopic;
+ }
+
+ internal override void OnSettingsCommand(Endpoints endpoints,
Proto.Settings settings)
+ {
+ base.OnSettingsCommand(endpoints, settings);
+ _liteSubscriptionManager.Sync(settings);
+ }
+
+ internal override void OnNotifyUnsubscribeLiteCommand(Endpoints
endpoints, Proto.NotifyUnsubscribeLiteCommand command)
+ {
+ _liteSubscriptionManager.OnNotifyUnsubscribeLiteCommand(command);
+ }
+
+ internal override Proto.HeartbeatRequest WrapHeartbeatRequest()
+ {
+ return new Proto.HeartbeatRequest
+ {
+ ClientType = Proto.ClientType.LiteSimpleConsumer,
+ Group = new Proto.Resource
+ {
+ ResourceNamespace = ClientConfig.Namespace,
+ Name = ConsumerGroup
+ }
+ };
+ }
+
+ internal override Settings GetSettings()
+ {
+ return _liteSimpleSubscriptionSettings;
+ }
+
+ protected override ClientType GetClientType()
+ {
+ return ClientType.LiteSimpleConsumer;
+ }
+
+ /// <summary>
+ /// Lite consumers only need routes to brokers, so keep the first
readable master queue.
+ /// </summary>
+ internal override SubscriptionLoadBalancer
UpdateSubscriptionLoadBalancer(string topic, TopicRouteData topicRouteData)
+ {
+ var firstReadableMasterQueue =
topicRouteData.MessageQueues.FirstOrDefault(mq =>
+ Utilities.MasterBrokerId == mq.Broker.Id && Permission.None !=
mq.Permission &&
+ (Permission.Read == mq.Permission || Permission.ReadWrite ==
mq.Permission));
+
+ var liteTopicRouteData = new TopicRouteData(null ==
firstReadableMasterQueue
+ ? new List<Proto.MessageQueue>()
+ : new List<Proto.MessageQueue> {
firstReadableMasterQueue.ToProtobuf() });
+
+ return base.UpdateSubscriptionLoadBalancer(topic,
liteTopicRouteData);
+ }
+
+ /// <summary>
+ /// Lite consumers must report the lite topic in ack requests,
otherwise the server
+ /// cannot resolve the receipt handle back to the parent topic.
+ /// </summary>
+ internal override Proto.AckMessageRequest
WrapAckMessageRequest(MessageView messageView)
+ {
+ var request = base.WrapAckMessageRequest(messageView);
+ if (!string.IsNullOrEmpty(messageView.LiteTopic) &&
request.Entries.Count > 0)
+ {
+ request.Entries[0].LiteTopic = messageView.LiteTopic;
+ }
+ return request;
+ }
+
+ /// <summary>
+ /// Lite consumers must report the lite topic when changing the
invisible duration.
+ /// </summary>
+ internal override Proto.ChangeInvisibleDurationRequest
WrapChangeInvisibleDuration(MessageView messageView,
+ TimeSpan invisibleDuration)
+ {
+ var request = base.WrapChangeInvisibleDuration(messageView,
invisibleDuration);
+ if (!string.IsNullOrEmpty(messageView.LiteTopic))
+ {
+ request.LiteTopic = messageView.LiteTopic;
+ }
+ return request;
+ }
+
+ /// <summary>
+ /// Builder for creating LiteSimpleConsumer instances.
+ /// </summary>
+ public new class Builder
+ {
+ private ClientConfig _clientConfig;
+ private string _consumerGroup;
+ private string _bindTopic;
+ private TimeSpan _awaitDuration = TimeSpan.FromSeconds(30);
+
+ public Builder SetClientConfig(ClientConfig clientConfig)
+ {
+ Preconditions.CheckArgument(clientConfig != null,
"clientConfig should not be null");
+ _clientConfig = clientConfig;
+ return this;
+ }
+
+ public Builder SetConsumerGroup(string consumerGroup)
+ {
+
Preconditions.CheckArgument(!string.IsNullOrWhiteSpace(consumerGroup),
"consumerGroup should not be null or empty");
+
Preconditions.CheckArgument(ConsumerGroupRegex.Match(consumerGroup).Success,
+ $"consumerGroup does not match the regex
{ConsumerGroupRegex}");
+ _consumerGroup = consumerGroup;
+ return this;
+ }
+
+ public Builder SetBindTopic(string bindTopic)
+ {
+
Preconditions.CheckArgument(!string.IsNullOrWhiteSpace(bindTopic), "bindTopic
should not be null or empty");
+ _bindTopic = bindTopic;
+ return this;
+ }
+
+ public Builder SetAwaitDuration(TimeSpan awaitDuration)
+ {
+ Preconditions.CheckArgument(awaitDuration > TimeSpan.Zero,
"awaitDuration should be positive");
+ _awaitDuration = awaitDuration;
+ return this;
+ }
+
+ public async Task<LiteSimpleConsumer> Build()
+ {
+ Preconditions.CheckArgument(_clientConfig != null,
"clientConfig has not been set yet");
+
Preconditions.CheckArgument(!string.IsNullOrWhiteSpace(_consumerGroup),
"consumerGroup has not been set yet");
+
Preconditions.CheckArgument(!string.IsNullOrWhiteSpace(_bindTopic), "bindTopic
has not been set yet");
+
+ var liteSimpleConsumer = new LiteSimpleConsumer(_clientConfig,
_consumerGroup, _awaitDuration, _bindTopic);
+ await liteSimpleConsumer.Start();
+ return liteSimpleConsumer;
+ }
+ }
+ }
+}
diff --git a/csharp/rocketmq-client-csharp/LiteSimpleSubscriptionSettings.cs
b/csharp/rocketmq-client-csharp/LiteSimpleSubscriptionSettings.cs
new file mode 100644
index 00000000..59700200
--- /dev/null
+++ b/csharp/rocketmq-client-csharp/LiteSimpleSubscriptionSettings.cs
@@ -0,0 +1,46 @@
+/*
+ * 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.Concurrent;
+using Proto = Apache.Rocketmq.V2;
+
+namespace Org.Apache.Rocketmq
+{
+ /// <summary>
+ /// Settings for LiteSimpleConsumer, reports LITE_SIMPLE_CONSUMER as the
client type
+ /// so that the server resolves lite topic subscriptions for this client.
+ /// </summary>
+ public class LiteSimpleSubscriptionSettings : SimpleSubscriptionSettings
+ {
+ public LiteSimpleSubscriptionSettings(string namespaceName, string
clientId, Endpoints endpoints,
+ string consumerGroup, TimeSpan requestTimeout, TimeSpan
longPollingTimeout,
+ ConcurrentDictionary<string, FilterExpression>
subscriptionExpressions)
+ : base(namespaceName, clientId, endpoints, consumerGroup,
requestTimeout, longPollingTimeout,
+ subscriptionExpressions)
+ {
+ // LiteSimpleConsumer uses LITE_SIMPLE_CONSUMER client type
instead of SIMPLE_CONSUMER
+ }
+
+ public override Proto.Settings ToProtobuf()
+ {
+ var settings = base.ToProtobuf();
+ settings.ClientType =
ClientTypeHelper.ToProtobuf(ClientType.LiteSimpleConsumer);
+ return settings;
+ }
+ }
+}
diff --git a/csharp/rocketmq-client-csharp/LiteSubscriptionManager.cs
b/csharp/rocketmq-client-csharp/LiteSubscriptionManager.cs
index 8501c633..8cd0d23e 100644
--- a/csharp/rocketmq-client-csharp/LiteSubscriptionManager.cs
+++ b/csharp/rocketmq-client-csharp/LiteSubscriptionManager.cs
@@ -27,23 +27,24 @@ using Proto = Apache.Rocketmq.V2;
namespace Org.Apache.Rocketmq
{
/// <summary>
- /// Manages lite topic subscriptions for lite push consumer.
+ /// Manages lite topic subscriptions for lite consumers (push and simple).
/// Handles subscription synchronization, quota management, and server
notifications.
/// </summary>
internal class LiteSubscriptionManager
{
private static readonly ILogger Logger =
MqLogManager.CreateLogger<LiteSubscriptionManager>();
- private readonly PushConsumer _consumer;
+ private readonly Consumer _consumer;
private readonly string _bindTopic;
private readonly string _consumerGroup;
private readonly ConcurrentDictionary<string, byte> _liteTopicSet;
+ private System.Threading.Timer _syncTimer;
// Client-side lite subscription quota limit
private volatile int _liteSubscriptionQuota;
private volatile int _maxLiteTopicSize = 64;
- public LiteSubscriptionManager(PushConsumer consumer, string
bindTopic, string consumerGroup)
+ public LiteSubscriptionManager(Consumer consumer, string bindTopic,
string consumerGroup)
{
_consumer = consumer;
_bindTopic = bindTopic;
@@ -68,7 +69,7 @@ namespace Org.Apache.Rocketmq
}
// Schedule periodic sync every 30 seconds using Timer
- var timer = new System.Threading.Timer(
+ _syncTimer = new System.Threading.Timer(
callback: state =>
{
try
@@ -88,6 +89,15 @@ namespace Org.Apache.Rocketmq
Logger.LogInformation($"LiteSubscriptionManager started
successfully, scheduled periodic sync every 30 seconds");
}
+ /// <summary>
+ /// Stop the periodic sync and release the underlying timer.
+ /// </summary>
+ public void Shutdown()
+ {
+ _syncTimer?.Dispose();
+ _syncTimer = null;
+ }
+
public string GetBindTopicName()
{
return _bindTopic;
diff --git a/csharp/rocketmq-client-csharp/MessageView.cs
b/csharp/rocketmq-client-csharp/MessageView.cs
index d20015a8..93c0ce2b 100644
--- a/csharp/rocketmq-client-csharp/MessageView.cs
+++ b/csharp/rocketmq-client-csharp/MessageView.cs
@@ -38,7 +38,7 @@ namespace Org.Apache.Rocketmq
private MessageView(string messageId, string topic, byte[] body,
string tag, string messageGroup,
DateTime? deliveryTimestamp, List<string> keys, Dictionary<string,
string> properties, string bornHost,
DateTime bornTime, int deliveryAttempt, MessageQueue messageQueue,
string receiptHandle, long offset,
- bool corrupted, int? priority = null)
+ bool corrupted, int? priority = null, string liteTopic = null)
{
MessageId = messageId;
Topic = topic;
@@ -56,6 +56,7 @@ namespace Org.Apache.Rocketmq
_offset = offset;
_corrupted = corrupted;
Priority = priority;
+ LiteTopic = liteTopic;
}
public string MessageId { get; }
@@ -85,6 +86,11 @@ namespace Org.Apache.Rocketmq
/// </summary>
public int? Priority { get; }
+ /// <summary>
+ /// Gets the lite topic the message belongs to, which makes sense only
for lite topics.
+ /// </summary>
+ public string LiteTopic { get; }
+
public int IncrementAndGetDeliveryAttempt()
{
return ++DeliveryAttempt;
@@ -199,8 +205,9 @@ namespace Org.Apache.Rocketmq
var receiptHandle = systemProperties.ReceiptHandle;
+ var liteTopic = systemProperties.HasLiteTopic ?
systemProperties.LiteTopic : null;
return new MessageView(messageId, topic, body, tag, messageGroup,
deliveryTime, keys, properties, bornHost,
- bornTime, deliveryAttempt, messageQueue, receiptHandle,
queueOffset, corrupted, priority);
+ bornTime, deliveryAttempt, messageQueue, receiptHandle,
queueOffset, corrupted, priority, liteTopic);
}
public override string ToString()
diff --git a/csharp/rocketmq-client-csharp/PushConsumer.cs
b/csharp/rocketmq-client-csharp/PushConsumer.cs
index a7cf44da..cdb4d35f 100644
--- a/csharp/rocketmq-client-csharp/PushConsumer.cs
+++ b/csharp/rocketmq-client-csharp/PushConsumer.cs
@@ -723,7 +723,7 @@ namespace Org.Apache.Rocketmq
/// <summary>
/// Check if the consumer is running.
/// </summary>
- internal void CheckRunning()
+ internal override void CheckRunning()
{
if (State != State.Running)
{
@@ -739,29 +739,6 @@ namespace Org.Apache.Rocketmq
return State == State.Terminated || State == State.Failed;
}
- /// <summary>
- /// Get the request timeout from client config.
- /// </summary>
- internal TimeSpan GetRequestTimeout()
- {
- return _clientConfig.RequestTimeout;
- }
-
- /// <summary>
- /// Get the namespace from client config.
- /// </summary>
- internal string Namespace => _clientConfig.Namespace;
-
- /// <summary>
- /// Sync lite subscription for lite push consumer.
- /// </summary>
- internal async Task<Proto.SyncLiteSubscriptionResponse>
SyncLiteSubscription(
- Proto.SyncLiteSubscriptionRequest request, TimeSpan timeout)
- {
- var invocation = await
ClientManager.SyncLiteSubscription(Endpoints, request, timeout);
- return invocation.Response;
- }
-
public class Builder
{
private ClientConfig _clientConfig;
diff --git a/csharp/rocketmq-client-csharp/SimpleConsumer.cs
b/csharp/rocketmq-client-csharp/SimpleConsumer.cs
index 1ede7071..77e63913 100644
--- a/csharp/rocketmq-client-csharp/SimpleConsumer.cs
+++ b/csharp/rocketmq-client-csharp/SimpleConsumer.cs
@@ -145,7 +145,7 @@ namespace Org.Apache.Rocketmq
};
}
- private SubscriptionLoadBalancer UpdateSubscriptionLoadBalancer(string
topic, TopicRouteData topicRouteData)
+ internal virtual SubscriptionLoadBalancer
UpdateSubscriptionLoadBalancer(string topic, TopicRouteData topicRouteData)
{
if (_subscriptionRouteDataCache.TryGetValue(topic, out var
subscriptionLoadBalancer))
{
@@ -239,7 +239,7 @@ namespace Org.Apache.Rocketmq
StatusChecker.Check(invocation.Response.Status, request,
invocation.RequestId);
}
- private Proto.AckMessageRequest WrapAckMessageRequest(MessageView
messageView)
+ internal virtual Proto.AckMessageRequest
WrapAckMessageRequest(MessageView messageView)
{
var topicResource = new Proto.Resource
{
@@ -259,7 +259,7 @@ namespace Org.Apache.Rocketmq
};
}
- private Proto.ChangeInvisibleDurationRequest
WrapChangeInvisibleDuration(MessageView messageView,
+ internal virtual Proto.ChangeInvisibleDurationRequest
WrapChangeInvisibleDuration(MessageView messageView,
TimeSpan invisibleDuration)
{
var topicResource = new Proto.Resource
diff --git a/csharp/rocketmq-client-csharp/SubscriptionLoadBalancer.cs
b/csharp/rocketmq-client-csharp/SubscriptionLoadBalancer.cs
index a5a88a6b..6c257910 100644
--- a/csharp/rocketmq-client-csharp/SubscriptionLoadBalancer.cs
+++ b/csharp/rocketmq-client-csharp/SubscriptionLoadBalancer.cs
@@ -19,6 +19,7 @@ using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
+using Org.Apache.Rocketmq.Error;
namespace Org.Apache.Rocketmq
{
@@ -51,6 +52,13 @@ namespace Org.Apache.Rocketmq
public MessageQueue TakeMessageQueue()
{
+ if (0 == _messageQueues.Count)
+ {
+ // Lite consumers prune the route to a single queue, an empty
route is possible
+ // when no readable master queue is served by the broker.
+ throw new NotFoundException("Failed to take message queue, no
readable master queue is available");
+ }
+
var next = Interlocked.Increment(ref _index);
var index = Utilities.GetPositiveMod(next, _messageQueues.Count);
return _messageQueues[index];
diff --git a/csharp/tests/LiteSimpleConsumerIntegrationTest.cs
b/csharp/tests/LiteSimpleConsumerIntegrationTest.cs
new file mode 100644
index 00000000..6c596978
--- /dev/null
+++ b/csharp/tests/LiteSimpleConsumerIntegrationTest.cs
@@ -0,0 +1,194 @@
+/*
+ * 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 System.Text;
+using System.Threading.Tasks;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using Org.Apache.Rocketmq;
+
+namespace tests
+{
+ /// <summary>
+ /// Integration tests of LiteSimpleConsumer against a real RocketMQ
cluster.
+ ///
+ /// They are skipped unless ROCKETMQ_CSHARP_LITE_ENDPOINTS is set, so that
the
+ /// default test run stays offline.
+ ///
+ /// Server prerequisites:
+ /// - broker.conf: enableLmq=true, enableMultiDispatch=true
+ /// - parent topic created with message.type=LITE
+ /// - consumer group created with the attribute
+lite.bind.topic=<parentTopic>
+ /// </summary>
+ [TestClass]
+ public class LiteSimpleConsumerIntegrationTest
+ {
+ private static readonly string Endpoint =
Environment.GetEnvironmentVariable("ROCKETMQ_CSHARP_LITE_ENDPOINTS");
+ private static readonly string BindTopic =
+
Environment.GetEnvironmentVariable("ROCKETMQ_CSHARP_LITE_PARENT_TOPIC") ??
"lite-parent-topic";
+ private static readonly string ConsumerGroup =
+ Environment.GetEnvironmentVariable("ROCKETMQ_CSHARP_LITE_GROUP")
?? "csharp-lite-unittest-group";
+
+ private static readonly TimeSpan AwaitDuration =
TimeSpan.FromSeconds(5);
+ private static readonly TimeSpan InvisibleDuration =
TimeSpan.FromSeconds(15);
+
+ private ClientConfig _clientConfig;
+ private LiteSimpleConsumer _consumer;
+ private Producer _producer;
+
+ [TestInitialize]
+ public void SetUp()
+ {
+ if (string.IsNullOrEmpty(Endpoint))
+ {
+ Assert.Inconclusive("ROCKETMQ_CSHARP_LITE_ENDPOINTS is not
set, skip the lite integration test");
+ }
+
+ _clientConfig = new ClientConfig.Builder()
+ .SetEndpoints(Endpoint)
+ .SetRequestTimeout(TimeSpan.FromSeconds(10))
+ .Build();
+ }
+
+ [TestCleanup]
+ public async Task TearDown()
+ {
+ if (null != _consumer)
+ {
+ await _consumer.DisposeAsync();
+ _consumer = null;
+ }
+ if (null != _producer)
+ {
+ await _producer.DisposeAsync();
+ _producer = null;
+ }
+ }
+
+ [TestMethod]
+ public async Task TestReceiveAndAckLiteMessages()
+ {
+ var liteTopic =
$"lite-topic-it-{DateTimeOffset.UtcNow.ToUnixTimeMilliseconds()}";
+ await StartClientsAsync();
+
+ var bodies = await SendLiteMessagesAsync(liteTopic, 5);
+
+ await _consumer.SubscribeLite(liteTopic, OffsetOption.MinOffset);
+ Assert.IsTrue(_consumer.GetLiteTopicSet().Contains(liteTopic));
+
+ var received = await ReceiveAndAckAsync(5,
TimeSpan.FromSeconds(30));
+
+ CollectionAssert.AreEquivalent(bodies.OrderBy(b => b).ToList(),
received.OrderBy(b => b).ToList());
+
+ await _consumer.UnsubscribeLite(liteTopic);
+ Assert.AreEqual(0, _consumer.GetLiteTopicSet().Count);
+ }
+
+ [TestMethod]
+ public async Task TestDeliverMessagesSentAfterSubscribeLite()
+ {
+ var liteTopic =
$"lite-topic-live-{DateTimeOffset.UtcNow.ToUnixTimeMilliseconds()}";
+ await StartClientsAsync();
+
+ await _consumer.SubscribeLite(liteTopic);
+ var bodies = await SendLiteMessagesAsync(liteTopic, 3);
+
+ var received = await ReceiveAndAckAsync(3,
TimeSpan.FromSeconds(30));
+
+ CollectionAssert.AreEquivalent(bodies.OrderBy(b => b).ToList(),
received.OrderBy(b => b).ToList());
+ }
+
+ [TestMethod]
+ public async Task TestStopDeliveringAfterUnsubscribeLite()
+ {
+ var liteTopic =
$"lite-topic-off-{DateTimeOffset.UtcNow.ToUnixTimeMilliseconds()}";
+ await StartClientsAsync();
+
+ await _consumer.SubscribeLite(liteTopic, OffsetOption.MinOffset);
+ await _consumer.UnsubscribeLite(liteTopic);
+ Assert.AreEqual(0, _consumer.GetLiteTopicSet().Count);
+
+ await SendLiteMessagesAsync(liteTopic, 3);
+
+ var deadline = DateTime.UtcNow.AddSeconds(10);
+ while (DateTime.UtcNow < deadline)
+ {
+ var messages = await _consumer.Receive(16, InvisibleDuration);
+ Assert.AreEqual(0, messages.Count, "No message is expected
after unsubscribeLite");
+ }
+ }
+
+ private async Task StartClientsAsync()
+ {
+ _producer = await new Producer.Builder()
+ .SetTopics(BindTopic)
+ .SetClientConfig(_clientConfig)
+ .Build();
+
+ _consumer = await new LiteSimpleConsumer.Builder()
+ .SetClientConfig(_clientConfig)
+ .SetConsumerGroup(ConsumerGroup)
+ .SetBindTopic(BindTopic)
+ .SetAwaitDuration(AwaitDuration)
+ .Build();
+
+ Assert.AreEqual(BindTopic, _consumer.GetBindTopic());
+ Assert.AreEqual(0, _consumer.GetLiteTopicSet().Count);
+ }
+
+ private async Task<List<string>> SendLiteMessagesAsync(string
liteTopic, int count)
+ {
+ var bodies = new List<string>();
+ for (var i = 0; i < count; i++)
+ {
+ var body = $"lite-body-{liteTopic}-{i}";
+ var message = new Message.Builder()
+ .SetTopic(BindTopic)
+ .SetBody(Encoding.UTF8.GetBytes(body))
+ .SetTag("LiteTest")
+ .SetKeys($"lite-{i}")
+ .SetLiteTopic(liteTopic)
+ .Build();
+
+ var receipt = await _producer.Send(message);
+ Assert.IsNotNull(receipt.MessageId);
+ bodies.Add(body);
+ }
+ return bodies;
+ }
+
+ private async Task<List<string>> ReceiveAndAckAsync(int expected,
TimeSpan timeout)
+ {
+ var received = new List<string>();
+ var deadline = DateTime.UtcNow.Add(timeout);
+ while (received.Count < expected && DateTime.UtcNow < deadline)
+ {
+ var messages = await _consumer.Receive(16, InvisibleDuration);
+ foreach (var message in messages)
+ {
+ Assert.IsFalse(string.IsNullOrEmpty(message.LiteTopic),
+ "Received lite message must carry its lite topic");
+ await _consumer.Ack(message);
+ received.Add(Encoding.UTF8.GetString(message.Body));
+ }
+ }
+ return received;
+ }
+ }
+}
diff --git a/csharp/tests/LiteSimpleConsumerTest.cs
b/csharp/tests/LiteSimpleConsumerTest.cs
new file mode 100644
index 00000000..5a72e526
--- /dev/null
+++ b/csharp/tests/LiteSimpleConsumerTest.cs
@@ -0,0 +1,338 @@
+/*
+ * 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 System.Text;
+using System.Threading.Tasks;
+using Google.Protobuf;
+using Google.Protobuf.WellKnownTypes;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using Moq;
+using Org.Apache.Rocketmq;
+using Org.Apache.Rocketmq.Error;
+using Proto = Apache.Rocketmq.V2;
+
+namespace tests
+{
+ [TestClass]
+ public class LiteSimpleConsumerTest
+ {
+ private const string BindTopic = "lite-parent-topic";
+ private const string ConsumerGroup = "lite-simple-consumer-group";
+
+ [TestMethod]
+ public void TestBuildWithoutClientConfig()
+ {
+ var builder = new
LiteSimpleConsumer.Builder().SetConsumerGroup(ConsumerGroup).SetBindTopic(BindTopic);
+ Assert.ThrowsExactly<ArgumentException>(() =>
builder.SetClientConfig(null));
+ }
+
+ [TestMethod]
+ public void TestBuildWithInvalidConsumerGroup()
+ {
+ Assert.ThrowsExactly<ArgumentException>(() =>
+ new LiteSimpleConsumer.Builder().SetConsumerGroup("invalid
group!"));
+ Assert.ThrowsExactly<ArgumentException>(() =>
+ new LiteSimpleConsumer.Builder().SetConsumerGroup(null));
+ }
+
+ [TestMethod]
+ public void TestBuildWithBlankBindTopic()
+ {
+ Assert.ThrowsExactly<ArgumentException>(() => new
LiteSimpleConsumer.Builder().SetBindTopic(""));
+ }
+
+ [TestMethod]
+ public void TestSetNonPositiveAwaitDuration()
+ {
+ Assert.ThrowsExactly<ArgumentException>(() =>
+ new
LiteSimpleConsumer.Builder().SetAwaitDuration(TimeSpan.Zero));
+ }
+
+ [TestMethod]
+ public void TestBuildWithoutRequiredOptions()
+ {
+ var builder = new LiteSimpleConsumer.Builder();
+ Assert.ThrowsExactly<ArgumentException>(() =>
builder.Build().GetAwaiter().GetResult());
+ }
+
+ [TestMethod]
+ public void TestConstructorRejectsBlankBindTopic()
+ {
+ var clientConfig = new
ClientConfig.Builder().SetEndpoints("127.0.0.1:8081").Build();
+ Assert.ThrowsExactly<ArgumentException>(() =>
+ new LiteSimpleConsumer(clientConfig, ConsumerGroup,
TimeSpan.FromSeconds(5), " "));
+ }
+
+ [TestMethod]
+ public void TestSettingsUseLiteSimpleConsumerClientType()
+ {
+ var consumer = CreateConsumer(out _);
+ var settings = consumer.GetSettings().ToProtobuf();
+ Assert.AreEqual(Proto.ClientType.LiteSimpleConsumer,
settings.ClientType);
+ Assert.IsTrue(consumer.IsLiteConsumer());
+ }
+
+ [TestMethod]
+ public void TestHeartbeatUsesLiteSimpleConsumerClientType()
+ {
+ var consumer = CreateConsumer(out _);
+ Assert.AreEqual(Proto.ClientType.LiteSimpleConsumer,
consumer.WrapHeartbeatRequest().ClientType);
+ }
+
+ [TestMethod]
+ public async Task TestSubscribeLiteBeforeStartup()
+ {
+ var clientConfig = new
ClientConfig.Builder().SetEndpoints("127.0.0.1:8081").Build();
+ var consumer = new LiteSimpleConsumer(clientConfig, ConsumerGroup,
TimeSpan.FromSeconds(5), BindTopic);
+ await Assert.ThrowsExactlyAsync<InvalidOperationException>(async
() =>
+ await consumer.SubscribeLite("lite-topic-1"));
+ }
+
+ [TestMethod]
+ public async Task TestSubscribeLiteSyncsPartialAdd()
+ {
+ var consumer = CreateConsumer(out var mockClientManager);
+ var invocations = SetupSyncLiteSubscription(mockClientManager);
+
+ await consumer.SubscribeLite("lite-topic-1",
OffsetOption.MinOffset);
+
+ Assert.AreEqual(1, invocations.Count);
+ var request = invocations[0];
+ Assert.AreEqual(Proto.LiteSubscriptionAction.PartialAdd,
request.Action);
+ CollectionAssert.Contains(request.LiteTopicSet.ToList(),
"lite-topic-1");
+ Assert.AreEqual(BindTopic, request.Topic.Name);
+ Assert.AreEqual(ConsumerGroup, request.Group.Name);
+ Assert.AreEqual(Proto.OffsetOption.Types.Policy.Min,
request.OffsetOption.Policy);
+ Assert.IsTrue(consumer.GetLiteTopicSet().Contains("lite-topic-1"));
+ }
+
+ [TestMethod]
+ public async Task TestDuplicateSubscribeLiteSkipsRpc()
+ {
+ var consumer = CreateConsumer(out var mockClientManager);
+ var invocations = SetupSyncLiteSubscription(mockClientManager);
+
+ await consumer.SubscribeLite("lite-topic-1");
+ await consumer.SubscribeLite("lite-topic-1");
+
+ Assert.AreEqual(1, invocations.Count);
+ }
+
+ [TestMethod]
+ public async Task TestUnsubscribeLiteSyncsPartialRemove()
+ {
+ var consumer = CreateConsumer(out var mockClientManager);
+ var invocations = SetupSyncLiteSubscription(mockClientManager);
+
+ await consumer.SubscribeLite("lite-topic-1");
+ await consumer.UnsubscribeLite("lite-topic-1");
+
+ Assert.AreEqual(2, invocations.Count);
+ Assert.AreEqual(Proto.LiteSubscriptionAction.PartialRemove,
invocations[1].Action);
+ Assert.AreEqual(0, consumer.GetLiteTopicSet().Count);
+ }
+
+ [TestMethod]
+ public async Task TestUnsubscribeUnknownLiteTopicIsNoop()
+ {
+ var consumer = CreateConsumer(out var mockClientManager);
+ var invocations = SetupSyncLiteSubscription(mockClientManager);
+
+ await consumer.UnsubscribeLite("unknown-lite-topic");
+
+ Assert.AreEqual(0, invocations.Count);
+ }
+
+ [TestMethod]
+ public async Task TestSubscribeLiteWithBlankTopic()
+ {
+ var consumer = CreateConsumer(out var mockClientManager);
+ SetupSyncLiteSubscription(mockClientManager);
+
+ await Assert.ThrowsExactlyAsync<ArgumentException>(async () =>
await consumer.SubscribeLite(" "));
+ }
+
+ [TestMethod]
+ public async Task TestOnNotifyUnsubscribeLiteCommand()
+ {
+ var consumer = CreateConsumer(out var mockClientManager);
+ SetupSyncLiteSubscription(mockClientManager);
+ await consumer.SubscribeLite("lite-topic-1");
+
+ consumer.OnNotifyUnsubscribeLiteCommand(null, new
Proto.NotifyUnsubscribeLiteCommand { LiteTopic = "lite-topic-1" });
+ Assert.AreEqual(0, consumer.GetLiteTopicSet().Count);
+
+ await consumer.SubscribeLite("lite-topic-2");
+ consumer.OnNotifyUnsubscribeLiteCommand(null, new
Proto.NotifyUnsubscribeLiteCommand { LiteTopic = "" });
+ Assert.IsTrue(consumer.GetLiteTopicSet().Contains("lite-topic-2"));
+ }
+
+ [TestMethod]
+ public void TestAckRequestCarriesLiteTopic()
+ {
+ var consumer = CreateConsumer(out _);
+ var messageView = CreateMessageView("lite-topic-1");
+
+ var request = consumer.WrapAckMessageRequest(messageView);
+
+ Assert.AreEqual(1, request.Entries.Count);
+ Assert.AreEqual("lite-topic-1", request.Entries[0].LiteTopic);
+ Assert.AreEqual(BindTopic, request.Topic.Name);
+ }
+
+ [TestMethod]
+ public void TestAckRequestWithoutLiteTopic()
+ {
+ var consumer = CreateConsumer(out _);
+ var messageView = CreateMessageView(null);
+
+ var request = consumer.WrapAckMessageRequest(messageView);
+
+ Assert.IsFalse(request.Entries[0].HasLiteTopic);
+ }
+
+ [TestMethod]
+ public void TestChangeInvisibleDurationCarriesLiteTopic()
+ {
+ var consumer = CreateConsumer(out _);
+ var messageView = CreateMessageView("lite-topic-1");
+
+ var request = consumer.WrapChangeInvisibleDuration(messageView,
TimeSpan.FromSeconds(10));
+
+ Assert.AreEqual("lite-topic-1", request.LiteTopic);
+ }
+
+ [TestMethod]
+ public void
TestUpdateSubscriptionLoadBalancerKeepsOnlyFirstReadableMasterQueue()
+ {
+ var consumer = CreateConsumer(out _);
+ var routeData = new TopicRouteData(new List<Proto.MessageQueue>
+ {
+ CreateMessageQueue(0, 0, Proto.Permission.ReadWrite),
+ CreateMessageQueue(1, 0, Proto.Permission.ReadWrite),
+ });
+
+ var loadBalancer =
consumer.UpdateSubscriptionLoadBalancer(BindTopic, routeData);
+
+ Assert.IsNotNull(loadBalancer);
+ var mq = loadBalancer.TakeMessageQueue();
+ Assert.AreEqual(0, mq.QueueId);
+ }
+
+ [TestMethod]
+ public void
TestUpdateSubscriptionLoadBalancerWithoutReadableMasterQueue()
+ {
+ var consumer = CreateConsumer(out _);
+ var routeData = new TopicRouteData(new List<Proto.MessageQueue>
+ {
+ CreateMessageQueue(0, 1, Proto.Permission.ReadWrite),
+ });
+
+ var loadBalancer =
consumer.UpdateSubscriptionLoadBalancer(BindTopic, routeData);
+
+ Assert.ThrowsExactly<NotFoundException>(() =>
loadBalancer.TakeMessageQueue());
+ }
+
+ private static LiteSimpleConsumer CreateConsumer(out
Mock<IClientManager> mockClientManager)
+ {
+ var clientConfig = new
ClientConfig.Builder().SetEndpoints("127.0.0.1:8081").Build();
+ var consumer = new LiteSimpleConsumer(clientConfig, ConsumerGroup,
TimeSpan.FromSeconds(5), BindTopic);
+ mockClientManager = new Mock<IClientManager>();
+ consumer.SetClientManager(mockClientManager.Object);
+ consumer.State = State.Running;
+ // The lite subscription quota is pushed by the server through the
settings command.
+ consumer.OnSettingsCommand(null, new Proto.Settings
+ {
+ Subscription = new Proto.Subscription
+ {
+ LiteSubscriptionQuota = 64,
+ MaxLiteTopicSize = 64
+ }
+ });
+ return consumer;
+ }
+
+ private static List<Proto.SyncLiteSubscriptionRequest>
SetupSyncLiteSubscription(Mock<IClientManager> mockClientManager)
+ {
+ var invocations = new List<Proto.SyncLiteSubscriptionRequest>();
+ mockClientManager.Setup(cm =>
cm.SyncLiteSubscription(It.IsAny<Endpoints>(),
+ It.IsAny<Proto.SyncLiteSubscriptionRequest>(),
It.IsAny<TimeSpan>()))
+ .Returns((Endpoints _, Proto.SyncLiteSubscriptionRequest
request, TimeSpan _) =>
+ {
+ invocations.Add(request.Clone());
+ var response = new Proto.SyncLiteSubscriptionResponse
+ {
+ Status = new Proto.Status { Code = Proto.Code.Ok }
+ };
+ return Task.FromResult(
+ new RpcInvocation<Proto.SyncLiteSubscriptionRequest,
Proto.SyncLiteSubscriptionResponse>(
+ request, response, null));
+ });
+ return invocations;
+ }
+
+ private static MessageView CreateMessageView(string liteTopic)
+ {
+ var systemProperties = new Proto.SystemProperties
+ {
+ MessageType = Proto.MessageType.Normal,
+ MessageId = MessageIdGenerator.GetInstance().Next(),
+ BodyDigest = new Proto.Digest { Type = Proto.DigestType.Crc32,
Checksum = "9EF61F95" },
+ BodyEncoding = Proto.Encoding.Identity,
+ BornHost = "127.0.0.1",
+ BornTimestamp = new Timestamp(),
+ ReceiptHandle = "fake-receipt-handle",
+ };
+ if (null != liteTopic)
+ {
+ systemProperties.LiteTopic = liteTopic;
+ }
+
+ var message = new Proto.Message
+ {
+ SystemProperties = systemProperties,
+ Topic = new Proto.Resource { Name = BindTopic },
+ Body = ByteString.CopyFrom("foobar", Encoding.UTF8)
+ };
+ return MessageView.FromProtobuf(message);
+ }
+
+ private static Proto.MessageQueue CreateMessageQueue(int queueId, int
brokerId, Proto.Permission permission)
+ {
+ return new Proto.MessageQueue
+ {
+ Id = queueId,
+ Permission = permission,
+ Broker = new Proto.Broker
+ {
+ Id = brokerId,
+ Name = "broker0",
+ Endpoints = new Proto.Endpoints
+ {
+ Scheme = Proto.AddressScheme.Ipv4,
+ Addresses = { new Proto.Address { Host = "127.0.0.1",
Port = 8081 } }
+ }
+ },
+ Topic = new Proto.Resource { Name = BindTopic },
+ AcceptMessageTypes = { Proto.MessageType.Lite }
+ };
+ }
+ }
+}
