Subscription handling cleanup by hylkevds · Pull Request #931 · moquette-io/moquette

hylkevds · 2026-01-31T11:48:16Z

While working on a subscription related feature I had to carefully review the way subscriptions are passed through Moquette, and did some massive cleanup and simplification.

Subscription information was being converted back and forth between several classes,
resulting in much unneeded object creation. But in the end it would always end up as Subscription.

This removes all the intermediate Objects, simplifying the code.
This includes the removal of the SharedSubscription class, since in the end, those are also turned into normal Subscriptions, making the extra class meaningless.
The logic contained many superfluous if/else checks for subscriptionIdentifiers caused by the wrapping and unwrapping of Subscription data.
All methods that would take several parameters with subscription information now simply accept a subscription.
Subscriptions are (already) registered on the Session, so when an unsubscribe is received, the subscription can be fetched from the Session. This allows the subscription-remove functions to be greatly simplified.
To simplify and speed up fetching the subscription from the Session, it is now stored in a Map instead of a Set.

These changes will greatly improve the maintainability of the code, and probably increase efficiency.

Checklist

My code follows the style guidelines of this project
I have commented my code, particularly in hard-to-understand areas
~~I have made corresponding changes to the documentation~~
~~I have made corresponding change to the default configuration files (and/or docker env variables)~~
I have added tests that prove my fix is effective or that my feature works
I have updated the Changelog if it's a feature or a fix that has to be reported

hylkevds · 2026-01-31T11:50:34Z

Note that this PR is on top of PR #919, since I discovered that problem at the start of this work.

andsel · 2026-02-06T12:25:30Z

Hi @hylkevds I'll check it, sorry but I'm pretty loaded and this PR is not so trivial.

hylkevds · 2026-02-07T20:17:37Z

No rush, I know how life is. I had wanted to get this done half a year ago, but I also just didn't find the time for it.
You should also prioritize 919, not this one.
This PR is far from trivial, and it doesn't actually fix or add any functionality. It just makes the logic a lot easier to follow, and probably improves performance though I've not measured that.

Subscription information was being converted back and forth between several classes, resulting in much unneeded object creation. But in the end it would always end up as Subscription. - This removes all the intermediate Objects, simplifying the code. - This includes the removal of the SharedSubscription class, since in the end, those are also turned into normal Subscriptions, making the extra class meaningless. - The logic contained many superfluous if/else checks for subscriptionIdentifiers caused by the wrapping and unwrapping of Subscription data. - All methods that would take several parameters with subscription information now simply accept a subscription.

Subscriptions are registered on the Session, so when an unsubscribe is received, the subscription can be fetched from the Session. This allows the subscription-remove functions to be greatly simplified.

Copilot

Pull request overview

This PR refactors Moquette’s subscription handling by removing intermediate subscription wrapper types (including SharedSubscription) and standardizing APIs to pass Subscription objects end-to-end, with corresponding updates across persistence, session tracking, and subscription matching.

Changes:

Removed SharedSubscription and migrated shared-subscription storage/matching to use Subscription directly.
Simplified subscribe/unsubscribe flows by storing subscriptions on Session in a map keyed by topic string and reusing the stored Subscription for removals.
Updated repositories (H2, in-memory), trie structures, and tests to the new Subscription-centric APIs.

Reviewed changes

Copilot reviewed 23 out of 23 changed files in this pull request and generated 5 comments.

Show a summary per file

File	Description
broker/src/main/java/io/moquette/broker/PostOffice.java	Builds shared/non-shared `Subscription` objects directly; unsubscribe now looks up stored subscriptions from `Session`.
broker/src/main/java/io/moquette/broker/Session.java	Stores subscriptions in a `Map` keyed by topic string; adds lookup helpers for unsubscribe.
broker/src/main/java/io/moquette/broker/SessionRegistry.java	Adjusts subscription removal calls to new signatures (but still needs shared vs non-shared handling fixes).
broker/src/main/java/io/moquette/broker/ISubscriptionsRepository.java	Repository API now removes/adds using `Subscription` objects; shared listing now returns `Subscription`.
broker/src/main/java/io/moquette/persistence/H2SubscriptionsRepository.java	Persists shared subscriptions as `Subscription`; updates MVStore (de)serialization logic accordingly.
broker/src/main/java/io/moquette/persistence/MemorySubscriptionsRepository.java	In-memory shared-subscription storage now uses `Subscription`; removal now takes `Subscription`.
broker/src/main/java/io/moquette/broker/subscriptions/CTrieSubscriptionDirectory.java	Directory API now accepts `Subscription`; shared ops now use `Subscription` directly.
broker/src/main/java/io/moquette/broker/subscriptions/CTrie.java	Removes `SubscriptionRequest`/`UnsubscribeRequest` wrappers; trie now mutates using `Subscription`.
broker/src/main/java/io/moquette/broker/subscriptions/CNode.java	Stores shared subscriptions as `Subscription` and returns them directly during match selection.
broker/src/main/java/io/moquette/broker/subscriptions/Subscription.java	Share name is now `ShareName`; adds helper to reconstruct `$share/...` topic string; renames option accessor.
broker/src/main/java/io/moquette/broker/subscriptions/ShareName.java	Adds `EMPTY_SHARENAME`, `isEmpty()`, and `Comparable`.
broker/src/main/java/io/moquette/broker/subscriptions/TNode.java	Updates signatures to accept `Subscription`.
broker/src/main/java/io/moquette/broker/subscriptions/ISubscriptionsDirectory.java	Directory interface now accepts/removes `Subscription` objects directly.
broker/src/main/java/io/moquette/broker/subscriptions/DumpTreeVisitor.java	Updates debug dump to use new option accessor.
broker/src/main/java/io/moquette/interception/messages/InterceptSubscribeMessage.java	Uses `Subscription.getOption()` accessor.
broker/src/main/java/io/moquette/broker/subscriptions/SharedSubscription.java	Deleted (type removed).
broker/src/test/java/io/moquette/persistence/H2SubscriptionsRepositorySharedSubscriptionsTest.java	Updates tests to use `Subscription` for shared subscriptions.
broker/src/test/java/io/moquette/broker/subscriptions/SubscriptionTestUtils.java	Updates helpers to construct shared subscriptions with `ShareName`.
broker/src/test/java/io/moquette/broker/subscriptions/CTrieTest.java	Updates tests to call trie methods with `Subscription` directly.
broker/src/test/java/io/moquette/broker/subscriptions/CTrieSubscriptionDirectoryMatchingTest.java	Updates tests to new directory signatures (`add/remove` take `Subscription`).
broker/src/test/java/io/moquette/broker/subscriptions/CTrieSharedSubscriptionDirectoryMatchingTest.java	Updates shared-subscription matching tests to use `Subscription`.
broker/src/test/java/io/moquette/broker/subscriptions/CTrieSpeedTest.java	Updates perf test to use `Subscription` directly.
ChangeLog.txt	Adds changelog entry for the cleanup.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

andsel · 2026-03-22T10:24:10Z

-    public List<Subscription> getSubscriptions() {
-        return new ArrayList<>(subscriptions);
+    public Collection<Subscription> getSubscriptions() {
+        return subscriptions.values();


@hylkevds I think it's right

Indeed. The performance cost of making the copy are mostly irrelevant, since the method is only used when re-initialising a server after a restart, or when cleaning up a session.
I'll also change the method header back to returning a List.

hylkevds · 2026-03-22T12:54:20Z

+            Optional<Subscription> invalidSharedSubscription = sharedSubscriptions.stream()
+                .filter(subData -> !SharedSubscriptionUtils.validateShareName(subData.getShareName().toString()))
                .findFirst();


I think the actual correct fix would be to make sure an empty share name can never happen. En empty share name is bound to cause other issues too.
We should track this as a separate issue, since it already exists in the current version and is not specific to the PR.

On second look, ShareName#toString is indeed the wrong method. Should be ShareName#getShareName. That is a bit confusing indeed.
Fixed now.

andsel · 2026-03-22T10:15:53Z

+        for (Subscription shared : subscriptionsRepository.listAllSharedSubscription()) {
            LOG.debug("Re-subscribing shared {}", shared);
-            ctrie.addToTree(SubscriptionRequest.buildShared(shared.getShareName(), shared.topicFilter(),
-                shared.clientId(), MqttSubscriptionOption.onlyFromQos(shared.requestedQoS())));
+            // TODO: This must call all registered InterceptHandler.onSubscribe
+            ctrie.addToTree(shared);
        }


@hylkevds I think this observation is right. I think that we can address in a follow-up PR, because requires also unit test to verify the error, and this PR is already big and maybe this change if out of context with the PR.

andsel · 2026-03-22T10:27:02Z

    private void unsubscribe(Session session) {
        for (Subscription existingSub : session.getSubscriptions()) {
-            subscriptionsDirectory.removeSubscription(existingSub.getTopicFilter(), session.getClientID());
+            subscriptionsDirectory.removeSubscription(existingSub);
        }


@hylkevds Also here I think it's right. I recall that I created pretty big unit tests, but maybe this is more an integration one that's missed. WDYT ?

This unsubscribe(Session) method is only used when purging a session. So we'd need an integration tests that

Creates a shared subscription with a persistent session.

Reconnects with the same clientId, but requesting a clean session.

That test currently doesn't exist... And I'm pretty sure the problem already existed before as a result. I'll add the hasShareName check.

andsel

Hi @hylkevds thank's a TON for your monumental work here, the spirit of the original code was to make think smoother but at the end created a little bit of mess that you correctly fixed.

Now that we don't have anymore a specialized SharedSubscription but only the Subscription which can be shared or not, I think that in every method that is something like "sharedSubscription(Subscription)" we have to put a defensive check that effectively the provided Subscription parameter is of shared kind. This to avoid of potential leak of non shared Subscription in the shared Subscription paths. WDYT?

I left suggestion comments and also spotted that Copilot spotted maybe real problems, please could you check those?

I've create #934 as follow-up work on this, to avoid making the context of this PR too big. Feel free to add even more tasks to that issue, maybe all the other Copilot suggestions if you think that fixing in this PR is out-of-context or make it too big.

Great work 👏

andsel · 2026-03-22T08:25:45Z

        for (Subscription subscription : this.subscriptionsRepository.listAllSubscriptions()) {
            LOG.debug("Re-subscribing {}", subscription);
-            ctrie.addToTree(SubscriptionRequest.buildNonShared(subscription));
+            // TODO: This must call all registered InterceptHandler.onSubscribe


@hylkevds why should call the onSubscribe? onSubscribe callback is intended to be invoked when a client effectively send a SUBSCRIBE to a series of topic filter. This init method, just reconstructs the directory from the data provided by the subscription's repository (the storage) so aren't real SUBSCRIBE commands.

Because an embedded server should also be brought back up to the state of all subscriptions. For instance, FROST-Server only sends messages on topics that actually have subscriptions. But to know a subscription exists, it registers an InterceptHandler so it knows about subscribe and unsubscribe events.

You could argue that the embedding server should save and restore its own state, but that just leads to the double storing of data, with the significant chance of these two states diverging due to race-conditions during shutdown. It's much more reliable to have Moquette holding the master storage, and the embedding server following that.

An alternative would be to have a separate method on the InterceptHandler for subscriptions restored from storage.

Maybe Moquette could expose an API to query the subscription storage, so that FROST-Server queries Moquette's subscription storage. In that case I should ask your help to shape the possible queries, the most obvious is "list all subscriptions", but maybe you have other use cases. Then let the asynch listener mechanism to receive the updates when happens, instead of polling this API to aligns continuously its internal state.
WDYT?

Yes, some dedicated methods to inspect the current topics would be even better. It would also allow for occasionally checking if the state of the embedding server still matches that of the broker.
I'll add it to our subscription TODO issue (#934) and create a wishlist for discussion.

andsel · 2026-03-22T08:34:12Z

@@ -1,4 +1,5 @@
 Version 0.19-SNAPSHOT
+   [Cleanup] Cleaned up subscription handling.


Suggested change

[Cleanup] Cleaned up subscription handling.

[cleanup] Cleaned up subscription handling. (#931)

andsel · 2026-03-22T08:41:17Z

+     *
+     * @return the original topic string.
+     */
+    public String getOriginalTopicWithSharename() {


Suggested change

public String getOriginalTopicWithSharename() {

public String getOriginalTopicFilterWithSharename() {

andsel · 2026-03-22T08:52:18Z

+
+    @Override
+    public int compareTo(ShareName o) {
+        return shareName.compareTo(o.shareName);


If shareName is null this is a NPE.
If we know that it can't ever be null we need to use a null check assertion in the constructor, or alternatively if null is same as EMPTY_SHARENAME replace null shareName with EMPTY_SHARENAME in the constructor and remove null checks around.

I just noticed validateShareName already does the null and empty check. But I'll add a non-null check regardless, to ensure we can't accidentally add a null sharename in the future.

andsel · 2026-03-22T09:07:39Z

    }

-    private void addSharedSubscriptionRequest(SubscriptionRequest shareSubRequest) {
+    private void addSharedSubscriptionRequest(Subscription shareSubRequest) {


addSharedSubscriptionRequest -> addShared

andsel · 2026-03-22T10:15:53Z

+        for (Subscription shared : subscriptionsRepository.listAllSharedSubscription()) {
            LOG.debug("Re-subscribing shared {}", shared);
-            ctrie.addToTree(SubscriptionRequest.buildShared(shared.getShareName(), shared.topicFilter(),
-                shared.clientId(), MqttSubscriptionOption.onlyFromQos(shared.requestedQoS())));
+            // TODO: This must call all registered InterceptHandler.onSubscribe
+            ctrie.addToTree(shared);
        }


@hylkevds I think this observation is right. I think that we can address in a follow-up PR, because requires also unit test to verify the error, and this PR is already big and maybe this change if out of context with the PR.

andsel · 2026-03-22T10:18:52Z

-            Optional<SharedSubscriptionData> invalidSharedSubscription = sharedSubscriptions.stream()
-                .filter(subData -> !SharedSubscriptionUtils.validateShareName(subData.name.toString()))
+            Optional<Subscription> invalidSharedSubscription = sharedSubscriptions.stream()
+                .filter(subData -> !SharedSubscriptionUtils.validateShareName(subData.getShareName().toString()))


subData now can became simply sub

andsel · 2026-03-22T10:24:10Z

-    public List<Subscription> getSubscriptions() {
-        return new ArrayList<>(subscriptions);
+    public Collection<Subscription> getSubscriptions() {
+        return subscriptions.values();


@hylkevds I think it's right

andsel · 2026-03-22T10:27:02Z

    private void unsubscribe(Session session) {
        for (Subscription existingSub : session.getSubscriptions()) {
-            subscriptionsDirectory.removeSubscription(existingSub.getTopicFilter(), session.getClientID());
+            subscriptionsDirectory.removeSubscription(existingSub);
        }


@hylkevds Also here I think it's right. I recall that I created pretty big unit tests, but maybe this is more an integration one that's missed. WDYT ?

andsel · 2026-04-07T12:27:13Z

@hylkevds let me know when this PR is ready for another review :-)

hylkevds · 2026-04-13T08:23:09Z

I think I've addressed all comments, except for removing the TODO comment on lime 57 of CTrieSubscriptionDirectory.java

hylkevds · 2026-04-17T09:24:36Z

The TODO comments are now also gone

andsel

@hylkevds maybe a nitpick, but please consider to add helper checker methods on subscription type, like assertion methods.

andsel · 2026-05-13T11:47:11Z

-        boolean notExistingSubscription = ctrie.addToTree(subRequest);
-        subscriptionsRepository.addNewSubscription(subRequest.subscription());
+    public boolean add(Subscription sub) {
+        if (sub.hasShareName()) {


Would you mind to add checker methods like in andsel#2 ?

Hmm, personally I dislike hiding the throwing of exceptions away like that, especially if the method name doesn't make it very obvious that its only goal is to throw an exception sometimes... It doesn't make the code more readable nor easier to follow.

Of course it's your project, so if you insist I'll change it :)

Dinner and some Yoga later...
I see the point that those if ... then ... throw new x are very ugly, but a method should either encapsulate significant complexity, or be re-useable. How about this further development of the pattern:

public class Exceptions { public static final void illegalArgumentIf(boolean predicate, String message) { if (predicate) { throw new IllegalArgumentException(message); } } }

That is re-usable, and calling

Exceptions.illegalArgumentIf(sub.hasShareName(), "Adding a shared subscription using the non-shared method.");

makes it very clear what the intention is.
It also open the option of making a version that accepts a lambda instead of a boolean, for more complex checks.

We can then also do the same for other exceptions. What do you think?

My initial consideration came from https://docs.oracle.com/javase/8/docs/api/java/util/Objects.html#requireNonNull-T- where in some methods, to assure a parameter is not null we use:

public void aMethod(Object some) { Object.requireNonNull("some parameter can't be null"); ... }

Your idea is not bad, is a good brick to build other "check-style" method. I think that if the method is more constrained provides implicitly more context.

What do you say of renaming checkIsShared with requireShared(subscription, message), it's very domain specific and carries all the meaning it needs.

Yes, requireShared is better.
On that note, shouldn't it just be a method on the subscription itself? Then we don't need to pass the subscription to the method: subscription.requireShared(message).

That's a nice suggestion, however the invariant is part of the client code, in this code CTrieSubscriptionDirectory and it's not proper business for the subscription itself. Just to limit the public API surface of the Subscription and limiting the scope.

Do you want to update your PR to this PR, or shall I add the changes as a separate commit to this PR, or do we do this completely separate?

You can add the changes as a separate commit to this PR 🙏

andsel

LGTM

Thank you so so so much @hylkevds for your work on this! 👏

hylkevds requested a review from andsel January 31, 2026 14:01

hylkevds force-pushed the subscription_rework branch from 657050e to da9b7da Compare March 9, 2026 09:27

hylkevds and others added 2 commits March 9, 2026 16:26

Cleaned up removing subscriptions

ee0cf5a

Subscriptions are registered on the Session, so when an unsubscribe is received, the subscription can be fetched from the Session. This allows the subscription-remove functions to be greatly simplified.

hylkevds force-pushed the subscription_rework branch from da9b7da to ee0cf5a Compare March 9, 2026 15:29

hylkevds mentioned this pull request Mar 14, 2026

Feature/topic rewriting #932

Draft

5 tasks

andsel requested a review from Copilot March 15, 2026 15:22

Copilot started reviewing on behalf of andsel March 15, 2026 15:23 View session

Copilot AI reviewed Mar 15, 2026

View reviewed changes

andsel requested changes Mar 22, 2026

View reviewed changes

hylkevds added 8 commits March 22, 2026 13:49

Reverted signature of Session#getSubscriptions

8b5fca6

Make distinction between shared/non-shared when reactivating a session

c22f77c

Updated ChangeLog

9f452ff

Added null-check on shareName

216c1e9

Inlined superfluous methods and added sanity checks on shared state

7068b28

Refactor lambda into static method

902fde0

Fixed minor formatting and naming issues

3d0e318

Fixed incorrect use of ShareName.toString instead of getShareName

4081b67

Removed TODO comments, they are now tracked in moquette-io#934

295bb85

andsel self-requested a review May 13, 2026 11:37

andsel requested changes May 13, 2026

View reviewed changes

Refactored check-then-throw blocks into require* methods

d8676fc

andsel self-requested a review May 15, 2026 07:41

andsel approved these changes May 15, 2026

View reviewed changes

andsel merged commit e7ac96e into moquette-io:main May 15, 2026
4 checks passed

		@@ -1,4 +1,5 @@
		Version 0.19-SNAPSHOT
		[Cleanup] Cleaned up subscription handling.

	public String getOriginalTopicWithSharename() {
	public String getOriginalTopicFilterWithSharename() {

Conversation

hylkevds commented Jan 31, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Checklist

Uh oh!

hylkevds commented Jan 31, 2026

Uh oh!

andsel commented Feb 6, 2026

Uh oh!

hylkevds commented Feb 7, 2026

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

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!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

andsel left a comment

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!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

andsel commented Apr 7, 2026

Uh oh!

hylkevds commented Apr 13, 2026

Uh oh!

hylkevds commented Apr 17, 2026

Uh oh!

andsel left a comment

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!

hylkevds commented Jan 31, 2026 •

edited

Loading