feat: implement MQTT wildcard subscription matching

[wy471x]

Make sure that:

• You have read the contribution guidelines.
• You submit test cases (unit or integration tests) that back your changes.
• Your local test passed ./mvnw clean install -Dmaven.javadoc.skip=true.

Summary:

Problem

Subscriptions using + (single-level wildcard) or # (multi-level wildcard) were silently broken. Publish.send() performed an exact-key lookup (ConcurrentHashMap.getOrDefault), so a publish to sensor/room1/temperature would never match a subscription filter like sensor/+/temperature.

Changes

1. New: TopicMatcher.java — Utility class implementing MQTT topic filter matching per the MQTT-4.7 spec:
   - + matches exactly one topic level
   - # matches any number of levels (must appear at the end of the filter)
   - Wildcards at the first level do not match $-prefixed topics
2. Modified: SubscribeRepository.java — Added getChannelsByTopic(String topic) method that iterates over all stored subscription filters and returns channels whose filter matches
   the published topic using TopicMatcher.matches().
3. Modified: Publish.java:116 — Changed send() from get(topic) (exact-key lookup) to getChannelsByTopic(topic) (wildcard-aware matching).
4. New: TopicMatcherTest.java — 6 unit tests covering: exact match, + single-level, # multi-level, mixed wildcards, $ topic protection, and null inputs.

close #6851

[Aias00]

Review: #6906 — feat: implement MQTT wildcard subscription matching

Verdict: ✅ Approve (with two follow-up suggestions, one a real correctness edge case)

This fills a genuinely broken feature — wildcard subscriptions silently never matched because Publish.send did an exact Map.get. The fix is well-structured.

What's correct

• TopicMatcher is spec-accurate. I traced the algorithm across the test matrix and beyond:
  • + matches exactly one level (sport/+/player1 ❌ sport/tennis/stadium/player1, ✓ sport/football/player1);
  • # matches any number of levels incl. the parent (sport/# ✓ sport);
  • # only matches when it's its own level (sport# / sport/tennis# correctly rejected);
  • $-prefixed topics are not matched by a leading wildcard (#/+ → false) but are matched by explicit $SYS/# / $SYS/+ — exactly MQTT-4.7.2-1;
  • null inputs return false (no NPE).
• The get() exact-lookup method is retained and still used by add()/remove() internally, so this isn't introducing dead code — good call keeping it.
• getChannelsByTopic is a clean O(N) scan that delegates entirely to TopicMatcher; no logic duplicated.
• TopicMatcherTest is thorough — exact, single-level, multi-level, mixed, $-topic, and null cases all covered.

Suggestions (non-blocking)

1. Duplicate delivery on overlapping subscriptions (real, please track as a follow-up). getChannelsByTopic does result.addAll(entry.getValue()) over every matching filter. If one client holds two overlapping subscriptions (e.g. sport/# and #), it appears under both keys, so a publish to sport/x adds the same Channel twice → the client receives the message twice. MQTT requires at most one delivery per publish per client. Collect into a Set<Channel> (or LinkedHashSet if you care about order/stability) before returning to avoid this.
2. Performance fast-path. Every publish now scans all subscriptions. For the very common case where the topic has an exact (non-wildcard) subscriber, you could result.addAll(get(topic)) first (O(1) exact hit) and then only scan filters containing +/#. Not necessary for correctness, just a scale consideration.
3. Minor: invalid filters (e.g. # not as its own level, or trailing text after #) silently return false here. Optionally reject malformed filters at subscription time in Subscribe.add so bad subscriptions fail fast instead of silently never matching.

Verdict

Approving. The core matching logic is correct and well-tested, and get() is correctly preserved. Suggestion #1 (dedupe) is worth a quick follow-up PR before wildcard support sees production traffic with multi-subscription clients.

[wy471x]

Review: #6906 — feat: implement MQTT wildcard subscription matching

Verdict: ✅ Approve (with two follow-up suggestions, one a real correctness edge case)

This fills a genuinely broken feature — wildcard subscriptions silently never matched because Publish.send did an exact Map.get. The fix is well-structured.

What's correct

• TopicMatcher is spec-accurate. I traced the algorithm across the test matrix and beyond:

  • + matches exactly one level (sport/+/player1 ❌ sport/tennis/stadium/player1, ✓ sport/football/player1);
  • # matches any number of levels incl. the parent (sport/# ✓ sport);
  • # only matches when it's its own level (sport# / sport/tennis# correctly rejected);
  • $-prefixed topics are not matched by a leading wildcard (#/+ → false) but are matched by explicit $SYS/# / $SYS/+ — exactly MQTT-4.7.2-1;
  • null inputs return false (no NPE).

• The get() exact-lookup method is retained and still used by add()/remove() internally, so this isn't introducing dead code — good call keeping it.

• getChannelsByTopic is a clean O(N) scan that delegates entirely to TopicMatcher; no logic duplicated.

• TopicMatcherTest is thorough — exact, single-level, multi-level, mixed, $-topic, and null cases all covered.

Suggestions (non-blocking)

1. Duplicate delivery on overlapping subscriptions (real, please track as a follow-up). getChannelsByTopic does result.addAll(entry.getValue()) over every matching filter. If one client holds two overlapping subscriptions (e.g. sport/# and #), it appears under both keys, so a publish to sport/x adds the same Channel twice → the client receives the message twice. MQTT requires at most one delivery per publish per client. Collect into a Set<Channel> (or LinkedHashSet if you care about order/stability) before returning to avoid this.
2. Performance fast-path. Every publish now scans all subscriptions. For the very common case where the topic has an exact (non-wildcard) subscriber, you could result.addAll(get(topic)) first (O(1) exact hit) and then only scan filters containing +/#. Not necessary for correctness, just a scale consideration.
3. Minor: invalid filters (e.g. # not as its own level, or trailing text after #) silently return false here. Optionally reject malformed filters at subscription time in Subscribe.add so bad subscriptions fail fast instead of silently never matching.

Verdict

Approving. The core matching logic is correct and well-tested, and get() is correctly preserved. Suggestion #1 (dedupe) is worth a quick follow-up PR before wildcard support sees production traffic with multi-subscription clients.

Thank you for the code review on this PR.

Fixes for the three review comments:

1. Duplicate delivery on overlapping subscriptions — SubscribeRepository.getChannelsByTopic now collects channels into a LinkedHashSet before returning, so a client holding
   overlapping filters (e.g. sport/# and #) receives at most one delivery per publish, per MQTT spec.
2. Performance fast-path — exact topic subscribers are added via an O(1) map lookup first; the wildcard scan then skips filters without +/#.
3. Malformed filters fail fast — added TopicMatcher.isValidFilter (MQTT-4.7.1 rules); Subscribe registers only valid filters and sends SUBACK return code 0x80 (FAILURE) for
   invalid ones.

Tests added: TopicMatcherTest validation cases, new SubscribeRepositoryTest (dedup/fast-path), new SubscribeTest (rejection + SUBACK codes).

[Aias00]

Summary (re-review after update)

Re-reviewed after the branch was updated (prior review was dismissed by a rebase). The current version already addresses the duplicate-delivery concern I raised before: getChannelsByTopic now collects results into a LinkedHashSet, so a client with overlapping subscriptions (sport/# and #) receives the message at most once — exactly as MQTT requires.

What is correct:

• New TopicMatcher implements matches(filter, topic) and isValidFilter(filter) per MQTT-4.7, including the $-topic rule (leading +/# must not match $SYS-style topics) and the rule that wildcards occupy an entire level with # last.
• Subscribe.subscribe filters out invalid filters and rejects them in the SUBACK with QoS FAILURE (0x80) while accepting valid ones; only valid subscriptions are stored.
• SubscribeRepository.getChannelsByTopic scans stored filters, dedupes channels via LinkedHashSet (overlapping wildcard filters deliver once per client), and uses the matcher for wildcard resolution.
• Publish.send now calls getChannelsByTopic so publishes reach wildcard subscribers.
• Tests are strong: TopicMatcherTest (exact/single/multi/mixed wildcards, $ topics, null, valid/invalid filters), SubscribeTest (invalid filter rejected with 0x80), and SubscribeRepositoryTest (exact/wildcard/dedup/multiple subscribers).

Conclusion

Approved. The matching logic is correct against the MQTT spec, the dedupe I previously flagged is now present, and the tests back the behavior.

Non-blocking suggestions

• Performance: getChannelsByTopic iterates the entire TOPIC_CHANNEL_FACTORY map on every publish (O(number of distinct subscription filters) per publish). For shenyu's expected scale this is acceptable; if subscription counts grow, an index of wildcard filters would help. Not a blocker.
• Operator precedence nit: filter.equals(topic) || filter.indexOf('+') < 0 && filter.indexOf('#') < 0 relies on && binding tighter than ||. Correct today, but parentheses would make intent clearer and prevent future mis-edits.
• Merge ordering: this PR and #6902 add the same getChannelsByTopic to SubscribeRepository, while #6913 rewrites SubscribeRepository to Map<Channel, MqttQoS>. Merge #6902/#6906 first and rebase #6913 (or merge #6913 last) to avoid a conflict. Coordinate the three MQTT PRs.

Thanks for adding the dedupe.

[dengliming]

This test looks flaky / incorrect because it calls new Subscribe().subscribe(ctx, msg) on a plain EmbeddedChannel, but Subscribe.subscribe(...) returns early when isConnected(channel) is false. In that case repository.add(...) is never reached, so awaitUntil(() -> repository.get("sport/#").contains(channel)) can only time out.

Could we update the test setup to put the channel into the same connected/authenticated state that MessageType.isConnected(...) expects before invoking subscribe(...)? Otherwise this test is asserting a post-condition that the method never has a chance to produce.

[Aias00]

REQUEST_CHANGES (PMC Aias00): the main build job is failing — this PR does not compile/pass tests, so it cannot be merged. Please fix the compilation/test failures and re-run CI, then request a re-review.