[broker-30] Implement delivering retained messages #54

crazyrokr · 2025-11-19T13:31:38Z

Initial implementation of retain message feature

3 MQTT Control Packets
└ 3.3 PUBLISH – Publish message
└ 3.3.1 PUBLISH Fixed Header
└ 3.3.1.3 RETAIN

If a client publishes a message with RETAIN flag, then the message is retained within the corresponding topic name
If a client subscribes to a topic filter, then all retained messages within topic names matching the topic filter are automatically released to the client
If during the releasing of retained messages a target subscription has retainAsPublished=0 then RETAIN flag is set to 0 in the forwarder publish message

model/src/main/java/javasabr/mqtt/model/topic/tree/ConcurrentTopicTree.java

model/src/main/java/javasabr/mqtt/model/topic/TopicFilter.java

model/src/main/java/javasabr/mqtt/model/topic/tree/TopicMessageNode.java

# Conflicts: # model/src/main/java/javasabr/mqtt/model/subscriber/tree/ConcurrentRetainedMessageTree.java # model/src/main/java/javasabr/mqtt/model/subscriber/tree/ConcurrentSubscriberTree.java # model/src/main/java/javasabr/mqtt/model/subscriber/tree/RetainedMessageNode.java # model/src/main/java/javasabr/mqtt/model/subscriber/tree/SubscriberNode.java # model/src/main/java/javasabr/mqtt/model/subscriber/tree/SubscriberTreeBase.java # model/src/main/java/javasabr/mqtt/model/subscribtion/tree/ConcurrentSubscriptionTree.java # model/src/main/java/javasabr/mqtt/model/subscribtion/tree/TopicFilterNode.java # model/src/main/java/javasabr/mqtt/model/subscribtion/tree/TopicFilterTreeBase.java # model/src/main/java/javasabr/mqtt/model/topic/tree/ConcurrentTopicTree.java # model/src/main/java/javasabr/mqtt/model/topic/tree/TopicNode.java # model/src/main/java/javasabr/mqtt/model/topic/tree/TopicTreeBase.java # model/src/test/groovy/javasabr/mqtt/model/topic/tree/SubscriberTreeTest.groovy # service/src/main/java/javasabr/mqtt/service/impl/InMemorySubscriptionService.java

# Conflicts: # application/src/main/java/javasabr/mqtt/broker/application/config/MqttBrokerSpringConfig.java # core-service/src/main/java/javasabr/mqtt/service/impl/InMemorySubscriptionService.java # core-service/src/test/groovy/javasabr/mqtt/service/message/handler/impl/SubscribeMqttInMessageHandlerTest.groovy

github-actions · 2025-12-01T19:42:34Z

Test Coverage Report

Overall Project	86.53% `-0.33%`	🍏
Files changed	92.99%	🍏

File	Coverage
Subscription.java	100%	🍏
QoS.java	100%	🍏
ConcurrentRetainedMessageTree.java	100%	🍏
DefaultRetainMessageService.java	100%	🍏
SubscriptionService.java	100%	🍏
ConcurrentSubscriberTree.java	100%	🍏
SubscriberTreeBase.java	100%	🍏
SubscriberNode.java	100%	🍏
Qos0MqttPublishOutMessageHandler.java	100%	🍏
TrackableMqttPublishInMessageHandler.java	100%	🍏
Publish.java	98.95%	🍏
RetainedMessageNode.java	95.83% `-4.17%`	🍏
AbstractTrieNode.java	95.69% `-4.31%`	🍏
InMemorySubscriptionService.java	91.72% `-8.28%`	🍏
MqttBrokerSpringConfig.java	90.63%	🍏
AbstractMqttPublishOutMessageHandler.java	82.64%	🍏
Qos1MqttPublishInMessageHandler.java	80.6%	🍏
Qos1MqttPublishOutMessageHandler.java	78.61%	🍏
Qos2MqttPublishOutMessageHandler.java	75.95%	🍏
TrackableMqttPublishOutMessageHandler.java	72.61%	🍏
Qos2MqttPublishInMessageHandler.java	65.43% `-5.86%`	🍏
AbstractMqttPublishInMessageHandler.java	64.43%	🍏
Qos0MqttPublishInMessageHandler.java	52.5%	🍏

# Conflicts: # core-service/src/main/java/javasabr/mqtt/service/SubscriptionService.java # core-service/src/main/java/javasabr/mqtt/service/impl/InMemorySubscriptionService.java # core-service/src/main/java/javasabr/mqtt/service/publish/handler/impl/AbstractMqttPublishOutMessageHandler.java # core-service/src/main/java/javasabr/mqtt/service/publish/handler/impl/PersistedMqttPublishOutMessageHandler.java # core-service/src/main/java/javasabr/mqtt/service/publish/handler/impl/Qos0MqttPublishOutMessageHandler.java # core-service/src/main/java/javasabr/mqtt/service/publish/handler/impl/Qos1MqttPublishOutMessageHandler.java # core-service/src/main/java/javasabr/mqtt/service/publish/handler/impl/Qos2MqttPublishOutMessageHandler.java

JavaSaBr · 2025-12-08T16:42:40Z

application/src/main/java/javasabr/mqtt/broker/application/config/MqttBrokerSpringConfig.java

-        publishReceivingService,
-        messageOutFactoryService,
-        topicService);
+    return new PublishMqttInMessageHandler(publishReceivingService, messageOutFactoryService, topicService);


wrong formating

It's after code reformatting with the latest code style from master https://github.com/JavaSaBr/settings/blob/master/intellij-idea/javasabr-intellij-codestyle.xml
Have you configured something else in code style?

JavaSaBr · 2025-12-08T16:42:52Z

application/src/main/java/javasabr/mqtt/broker/application/config/MqttBrokerSpringConfig.java

-        subscriptionService,
-        messageOutFactoryService,
-        topicService);
+    return new UnsubscribeMqttInMessageHandler(subscriptionService, messageOutFactoryService, topicService);


wrong formatting

Could you clarify what exactly is wrong?

JavaSaBr · 2025-12-08T17:01:22Z

application/src/main/java/javasabr/mqtt/broker/application/config/MqttBrokerSpringConfig.java

+        env.getProperty("mqtt.external.connection.max.string.length", int.class, MqttProperties.MAXIMUM_STRING_LENGTH),
+        env.getProperty("mqtt.external.connection.max.binary.size", int.class, MqttProperties.MAXIMUM_BINARY_SIZE),
+        env.getProperty("mqtt.external.connection.max.topic.levels", int.class, MqttProperties.MAXIMUM_TOPIC_LEVELS),
+        env.getProperty("mqtt.external.connection.min.keep.alive", int.class, MqttProperties.SERVER_KEEP_ALIVE_DEFAULT),


bad formatting

Could you clarify what exactly is wrong?

JavaSaBr · 2025-12-08T17:03:19Z

core-service/src/main/java/javasabr/mqtt/service/PublishDeliveringService.java


  PublishHandlingResult startDelivering(Publish publish, SingleSubscriber subscriber);
+
+  Array<PublishHandlingResult> deliverRetainedMessages(SingleSubscriber subscriber);


PublishDeliveringService is responsible only for delivering a publish to a user. That is it.
I think you need to have a separated service like "RetainMessagesService" which will have a topic tree + with retain messages (instead of subscribers) and will use 'PublishDeliveringService' to deliver retain messages

JavaSaBr · 2025-12-08T17:04:12Z

model/src/main/java/javasabr/mqtt/model/AbstractTrieNode.java

+import javasabr.rlib.collections.dictionary.LockableRefToRefDictionary;
+import org.jspecify.annotations.Nullable;
+
+public abstract class AbstractTrieNode<T> {


looks like typo -> AbstractTreeNode

It's actually the name of a prefix tree: https://drstearns.github.io/tutorials/trie/

JavaSaBr · 2025-12-08T17:06:52Z

model/src/main/java/javasabr/mqtt/model/subscriber/tree/SubscriberNode.java

      return;
    }
-    if (level == lastLevel) {
+    if (level == lastLevel || TopicFilter.MULTI_LEVEL_WILDCARD.equals(segment)) {


why do we need to have it here?

This condition allows us to merge the exactlyTopicMatch method with the following one

mqtt-broker/model/src/main/java/javasabr/mqtt/model/subscriber/tree/SubscriberNode.java

Lines 97 to 102 in 294a6c6

private void multiWildcardTopicMatch(MutableArray<SingleSubscriber> result) {

SubscriberNode subscriberNode = childNode(TopicFilter.MULTI_LEVEL_WILDCARD);

if (subscriberNode != null) {

appendSubscribersTo(result, subscriberNode);

}

}

I think in this case one method is better than the three, don't you think so?

@crazyrokr
I isolated 3 cases separately because:
a. Avoid one extra check for each segment (you did it 3 times per node, for deep trees it produces 100+ extra checks)
b. To have the clear separation of looking for subscribers because the exactly match and the single wildcard can be tuned in different ways

The TopcTree is generaly one of the high loaded data structures in our project.

As I said, I believe your approach reduces readability and increases cognitive load. However, the performance-improvement argument may be valid, and if you compare the performance of the two approaches, I'd be happy to see the results.

JavaSaBr · 2025-12-08T17:07:44Z

model/src/main/java/javasabr/mqtt/model/subscriber/tree/SubscriberNode.java

-    } else if (level < lastLevel) {
-      subscriberNode.matchesTo(level + 1, topicName, lastLevel, result);
-    }
+    collectMatchingSubscribers(topicName.segment(level), level, topicName, lastLevel, container);


so you returned the old way which I specially removed :D

It's a reduction of duplication, and it seems more readable to me

@crazyrokr I had clearly separated 3 matching methods, it cannot be less readable... more code? yes, but not less readable

A developer has to keep in mind 3 different cases with your approach, despite the fact that these three cases differ only slightly. I regard this as duplication of trivial code.

JavaSaBr · 2025-12-08T17:09:35Z

model/src/main/java/javasabr/mqtt/model/subscriber/tree/SubscriberTreeBase.java

-
-    if (existeQos.ordinal() < candidateQos.ordinal()) {
+    QoS existedQos = result
+        .get(found)


wrong formattimg

Then it seems like the following option has to be set to 1
<option name="METHOD_CALL_CHAIN_WRAP" value="2" />
https://github.com/JavaSaBr/settings/blob/28bafa79d458c1cbeb76d6c8e24980ce81a0c491/intellij-idea/javasabr-intellij-codestyle.xml#L80-L80

The same formatting in other places

mqtt-broker/core-service/src/main/java/javasabr/mqtt/service/impl/InMemorySubscriptionService.java

Lines 114 to 116 in 294a6c6

Array<Subscription> subscriptions = session

.activeSubscriptions()

.subscriptions();

https://github.com/JavaSaBr/mqtt-broker/blame/294a6c6ed5b19d350980900dcff40a12fcd62ce8/core-service/src/main/java/javasabr/mqtt/service/impl/InMemorySubscriptionService.java#L103-L105

JavaSaBr · 2025-12-08T17:10:48Z

model/src/main/java/javasabr/mqtt/model/topic/tree/RetainedMessageNode.java

+    return ArrayFactory.mutableArray(RetainedMessageNode.class);
+  }
+
+  final AtomicReference<@Nullable Publish> retainedMessage = new AtomicReference<>();


we can have only 1 message per topic? I think here should be a list

yeah, just a message per topic name, if I haven't messed anything up
please check yourself https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901104

# Conflicts: # application/src/main/java/javasabr/mqtt/broker/application/config/MqttBrokerSpringConfig.java # core-service/src/test/groovy/javasabr/mqtt/service/IntegrationServiceSpecification.groovy

JavaSaBr · 2025-12-10T17:22:02Z

core-service/src/main/java/javasabr/mqtt/service/impl/DefaultRetainMessageService.java

+@FieldDefaults(level = AccessLevel.PRIVATE, makeFinal = true)
+public class DefaultRetainMessageService implements RetainMessageService {
+
+  PublishDeliveringService defaultPublishDeliveringService;


defaultPublishDeliveringService -> publishDeliveringService

JavaSaBr · 2025-12-10T17:22:44Z

core-service/src/main/java/javasabr/mqtt/service/impl/DefaultRetainMessageService.java

+  }
+
+  @Override
+  public void retainMessage(Publish publish) {


I think we don't need to check it inside this method, the caller shuold check it before calling

JavaSaBr · 2025-12-10T17:25:10Z

core-service/src/main/java/javasabr/mqtt/service/impl/InMemorySubscriptionService.java

+    if (previousSubscriber != null) {
+      activeSubscriptions.remove(previousSubscriber.subscription());
+    }
+    QoS subscriptionQoS = subscription.qos();


at least I think it should be extracted to a method

JavaSaBr · 2025-12-10T17:27:26Z

model/src/main/java/javasabr/mqtt/model/subscriber/tree/ConcurrentSubscriberTree.java

  @Nullable
-  public SingleSubscriber subscribe(MqttUser user, Subscription subscription) {
-    return rootNode.subscribe(0, user, subscription, subscription.topicFilter());
+  public SingleSubscriber subscribe(SingleSubscriber subscriber) {


you break the abstraction

I believe SingleSubscriber has to be created by SubscriptionService, not by SubscriberTree

JavaSaBr · 2025-12-10T17:31:16Z

model/src/main/java/javasabr/mqtt/model/topic/tree/RetainedMessageNode.java

+    var child = getOrCreateChildNode(topicName.segment(level));
+    boolean isLastLevel = (level + 1 == topicName.levelsCount());
+    if (isLastLevel) {
+      child.retainedMessage.set(message.payload().length == 0 ? null : message);


Updating the message should be done by a separated method, without direct external access

[broker-30] Implement delivering retained messages

26129f5

JavaSaBr reviewed Nov 19, 2025

View reviewed changes

JavaSaBr assigned crazyrokr Nov 19, 2025

crazyrokr added 3 commits November 19, 2025 19:05

[broker-30] Rewrite retained messages collecting

c6ada4d

[broker-30] Cleanup code after merge

01bac5b

crazyrokr force-pushed the feature-broker-30 branch from 24664c4 to 01bac5b Compare November 19, 2025 20:50

crazyrokr and others added 5 commits November 20, 2025 08:15

[broker-30] Fix corner cases in retained messages

c903510

[broker-30] Handle SubscribeRetainHandling

e1f5c2a

[broker-30] Fix build

a1b87dd

Merge branch 'develop' into feature-broker-30

fd848eb

crazyrokr added 3 commits December 2, 2025 23:56

[broker-30] SubscriberNode refactoring

be8070b

[broker-30] Revert wrong access level modifier

0c219dd

[broker-30] Avoid tree branch locking

62900ca

crazyrokr linked an issue Dec 5, 2025 that may be closed by this pull request

Implement Retained Messages #30

Open

crazyrokr added 3 commits December 6, 2025 15:13

[broker-30] Introduce AbstractTrieNode

7314ed1

[broker-30] Remove redundant code from SubscriberNode

901e865

crazyrokr marked this pull request as ready for review December 7, 2025 14:18

crazyrokr added 2 commits December 7, 2025 16:10

[broker-30] Revert redundant changes

cc91db2

[broker-30] Improve AbstractTrieNode

42af014

crazyrokr requested a review from JavaSaBr December 7, 2025 16:04

crazyrokr added enhancement New feature or request core labels Dec 7, 2025

crazyrokr added 4 commits December 7, 2025 18:29

[broker-30] Implement retainAsPublished support

b1aa456

[broker-30] Update debug message

f2f2f67

[broker-30] Small refactoring

bf837eb

[broker-30] Avoid double message retaining

cc08339

crazyrokr added 3 commits December 8, 2025 08:07

[broker-30] Fix tests

1d37242

[broker-30] Fix retainAsPublished logic

c3ab5c5

[broker-30] Add tests

b829090

JavaSaBr reviewed Dec 8, 2025

View reviewed changes

crazyrokr added 2 commits December 9, 2025 00:27

[broker-30] Introduce RetainMessageService

26f45dc

[broker-30] Rework retainAsPublished handling

6c16962

Repository owner deleted a comment from github-actions bot Dec 9, 2025

crazyrokr added 6 commits December 9, 2025 01:59

Merge branch 'develop' into feature-broker-30

12f8069

[broker-30] Fix formatting

31ff664

[broker-30] Improve RetainedMessageTreeTest

8191bc8

[broker-30] Improve code readability

c7c72d3

[broker-30] Extract creation of SingleSubscriber to SubscriptionService

8a23e97

[broker-30] Revert unnecessary changes

57eeb77

crazyrokr force-pushed the feature-broker-30 branch from 4d8e497 to 57eeb77 Compare December 9, 2025 14:05

[broker-30] Fix QoS comparison

28e73a5

crazyrokr requested a review from JavaSaBr December 9, 2025 14:36

crazyrokr added 2 commits December 10, 2025 10:21

Merge branch 'develop' into feature-broker-30

2fb16a5

# Conflicts: # application/src/main/java/javasabr/mqtt/broker/application/config/MqttBrokerSpringConfig.java # core-service/src/test/groovy/javasabr/mqtt/service/IntegrationServiceSpecification.groovy

[broker-30] Update tests

f4a6727

JavaSaBr reviewed Dec 10, 2025

View reviewed changes

crazyrokr added 5 commits December 10, 2025 21:55

[broker-30] Introduce setRetainedMessage and clearRetainedMessage

751624f

[broker-30] Move publish.retained() check to caller

649f57e

[broker-30] Extract isRetainHandlingSatisfied() method

b4e1232

[broker-30] Revert formatting

1f3d50a

[broker-30] Revert SubscriberNode

588aae4


		PublishHandlingResult startDelivering(Publish publish, SingleSubscriber subscriber);

		Array<PublishHandlingResult> deliverRetainedMessages(SingleSubscriber subscriber);

	private void multiWildcardTopicMatch(MutableArray<SingleSubscriber> result) {
	SubscriberNode subscriberNode = childNode(TopicFilter.MULTI_LEVEL_WILDCARD);
	if (subscriberNode != null) {
	appendSubscribersTo(result, subscriberNode);
	}
	}

	Array<Subscription> subscriptions = session
	.activeSubscriptions()
	.subscriptions();

[broker-30] Implement delivering retained messages #54

Are you sure you want to change the base?

[broker-30] Implement delivering retained messages #54

Conversation

crazyrokr commented Nov 19, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Initial implementation of retain message feature

Uh oh!

Uh oh!

Uh oh!

Uh oh!

github-actions bot commented Dec 1, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Test Coverage Report

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

crazyrokr Dec 9, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

crazyrokr commented Nov 19, 2025 •

edited

Loading

github-actions bot commented Dec 1, 2025 •

edited

Loading

crazyrokr Dec 9, 2025 •

edited

Loading

crazyrokr Dec 10, 2025 •

edited

Loading