[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	87.73% `-0.3%`	🍏
Files changed	94.14%	🍏

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	98.9% `-1.1%`	🍏
AbstractTrieNode.java	95.69% `-4.31%`	🍏
InMemorySubscriptionService.java	91.5% `-8.5%`	🍏
MqttBrokerSpringConfig.java	89.74%	🍏
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%	🍏
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

# Conflicts: # core-service/src/main/java/javasabr/mqtt/service/publish/handler/impl/TrackableMqttPublishOutMessageHandler.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?

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

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

[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 5 commits December 8, 2025 07:55

[broker-30] Add more tests

bacefb7

Merge branch 'develop' into feature-broker-30

c2ac2ad

# Conflicts: # core-service/src/main/java/javasabr/mqtt/service/publish/handler/impl/TrackableMqttPublishOutMessageHandler.java

[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


		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

Uh oh!

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!

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!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

crazyrokr commented Nov 19, 2025 •

edited

Loading

github-actions bot commented Dec 1, 2025 •

edited

Loading

crazyrokr Dec 9, 2025 •

edited

Loading