Skip to content

Stop / Resume reading from Broker - #3378

Open
dometec wants to merge 5 commits into
smallrye:mainfrom
dometec:stop_resume_reading
Open

Stop / Resume reading from Broker#3378
dometec wants to merge 5 commits into
smallrye:mainfrom
dometec:stop_resume_reading

Conversation

@dometec

@dometec dometec commented Apr 20, 2026

Copy link
Copy Markdown
Contributor

This is a Draft PR since use vertx-mqtt compiled with vert-x3/vertx-mqtt#267 and vert-x3/vertx-mqtt#268 not released yet.

The first commit add a test to show the error where the application message buffer is full.

In the second commit I've added the logic to stop and restart the socket read when the buffer is going to be almost full.
Added also some documentation in receiving-mqtt-messages.md, I hope that clear the problem and the parameters usage.

The attirbutes "receive-buffer-size", "recv-bytebuf-allocator-size" have direction = INCOMING_AND_OUTGOING becouse are used by the mqtt client that can be common between incoming and outgoing channel.

@dometec

dometec commented Apr 28, 2026

Copy link
Copy Markdown
Contributor Author

I tested this feature with some private application module and it works well. I'll switch this PR in ready on the first release of verts-mqtt.

@codecov

codecov Bot commented Jul 6, 2026

Copy link
Copy Markdown

Codecov Report

❌ Patch coverage is 95.00000% with 2 lines in your changes missing coverage. Please review.
✅ Project coverage is 77.86%. Comparing base (a96442f) to head (c955cdc).
⚠️ Report is 1216 commits behind head on main.

Files with missing lines Patch % Lines
...a/io/smallrye/reactive/messaging/mqtt/Clients.java 89.47% 0 Missing and 2 partials ⚠️
Additional details and impacted files

Impacted file tree graph

@@             Coverage Diff              @@
##               main    #3378      +/-   ##
============================================
+ Coverage     77.47%   77.86%   +0.38%     
- Complexity     3778     5607    +1829     
============================================
  Files           306      484     +178     
  Lines         12673    18768    +6095     
  Branches       1648     2312     +664     
============================================
+ Hits           9818    14613    +4795     
- Misses         2116     3014     +898     
- Partials        739     1141     +402     
Files with missing lines Coverage Δ
...mallrye/reactive/messaging/mqtt/MqttConnector.java 100.00% <ø> (ø)
...o/smallrye/reactive/messaging/mqtt/MqttSource.java 81.42% <100.00%> (+2.06%) ⬆️
.../reactive/messaging/mqtt/internal/MqttHelpers.java 50.99% <100.00%> (-0.74%) ⬇️
...tive/messaging/mqtt/session/MqttClientSession.java 66.66% <ø> (ø)
...ssaging/mqtt/session/MqttClientSessionOptions.java 38.69% <100.00%> (+2.27%) ⬆️
...aging/mqtt/session/impl/MqttClientSessionImpl.java 55.48% <100.00%> (+1.90%) ⬆️
...a/io/smallrye/reactive/messaging/mqtt/Clients.java 95.23% <89.47%> (-0.92%) ⬇️

... and 279 files with indirect coverage changes

🚀 New features to boost your workflow:
  • ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

@dometec
dometec marked this pull request as ready for review July 6, 2026 16:03
dometec added 4 commits July 14, 2026 18:47
Guard the buffer-driven pause() and resume() calls in ClientHolder with
isConnected(). Acting on a disconnected session is a no-op, and the paused
state could otherwise leak across a reconnect, leaving consumption stuck.
@dometec
dometec force-pushed the stop_resume_reading branch from 7926481 to f7d214e Compare July 14, 2026 16:48
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants