Feature/automatic resubscribe #398

SgtSilvio · 2020-04-06T07:35:12Z

Motivation
Resolves #297 and resolves #288
This is a prerequisite for proper implementation of manual acknowledgement and message persistence.

Changes

Enable subscribe(Stream) and unsubscribe independent of connect (similar to publishes)
Automatic resubscribe when automatically reconnecting and no session is present
Flows are not expired when automatic reconnect is enabled (instead subscriptions are automatically restored)
Improved concurrency of MqttPublishFlowableAckLink, FlowableWithSingleObserveOn and FluxWithSinglePublishOn

...ain/java/com/hivemq/client/internal/mqtt/handler/publish/outgoing/MqttAckSingleFlowable.java

… to publishes) Subscription and global flows are not expired when reconnect is enabled

…connecting

Removed Flowable/FluxWithSingleCombine which had concurrency issues

…e is pending

…alled serially

Restored Flowable/FluxWithSingleCombine and fixed concurrency issues

…nHandler

…andwidth when restoring subscriptions)

…s more consistent: MqttIncomingPublishFlow(s) and MqttSubscribedPublishFlow(s))

src/main/java/com/hivemq/client/internal/mqtt/handler/subscribe/MqttSubscriptionHandler.java

sauroter · 2020-04-17T13:33:58Z

src/main/java/com/hivemq/client/internal/mqtt/handler/subscribe/MqttSubscriptionHandler.java

+        subscriptionIdentifiersAvailable = connectionConfig.areSubscriptionIdentifiersAvailable();
+
+        if (!hasSession) {
+            incomingPublishFlows.getSubscriptions().forEach((subscriptionIdentifier, subscriptions) -> {


I think this bahavior should be configurable somehow.

What should be configurable?
If you use automatic reconnect and the server lost the session between the connection loss and a successful reconnect, why would you not want to resubscribe?

This is actually feedback from users who are confused by SessionExpiredExceptions when they use automatic reconnect.

Use case: You are in extremely constrained networks and want to be very precise on which data you send and when you send it.

The same is true for resending publishes.
Isn't automatic reconnect the wrong choice if you want to have control over everything?

The same is true for resending publishes.

I did not know that. But I would argue that the same logic applies.

I feel like the connect is something else. As it is the basis for everything afterwards. And in the connect there are possibilities to influence the behavior.

So would a config to disable automatic resubscribe be enough?

Yeah for a start I think so.

Added configuration option: Reconnector.resubscribeIfSessionExpired (default true)
Also added a similar option for republishing: Reconnector.republishIfSessionExpired (default false)

sauroter

The code looks good. The question about the configurability of the resubscribe must be resolved.

…ibedPublishFlowTree, reasons: the first is tightly coupled to the MqttOutgoingQosHandler the second is similar to a collection which is instantiated

…xpired

cla-bot bot added the cla-signed label Apr 6, 2020

SgtSilvio self-assigned this Apr 6, 2020

SgtSilvio force-pushed the feature/automatic-resubscribe branch from e3d20ec to 09efec5 Compare April 6, 2020 10:17

SgtSilvio requested review from gitseti, a user, YannickWeber and RobinAtherton and removed request for YannickWeber April 8, 2020 09:15

SgtSilvio mentioned this pull request Apr 15, 2020

Feature/manual ack #403

Merged

sauroter reviewed Apr 16, 2020

View reviewed changes

...ain/java/com/hivemq/client/internal/mqtt/handler/publish/outgoing/MqttAckSingleFlowable.java Show resolved Hide resolved

SgtSilvio removed the request for review from RobinAtherton April 16, 2020 14:21

SgtSilvio added 19 commits April 17, 2020 10:58

Enable subscribe(Stream), unsubscribe independent of connect (similar…

c223a56

… to publishes) Subscription and global flows are not expired when reconnect is enabled

Refactored global publish flow subscribe

f04dfa2

Fixed MqttSubscriptionFlowsTest

1a2cb9a

Removed unnecessary parameter from MqttSubscriptionFlows.unsubscribe

bd16bf6

Automatic resubscribe when no session is present and automatically re…

3240415

…connecting

Refactor MqttIncomingPublishFlows method order

f8b3269

Added comments to assert statements

d98399f

Improved MqttSubscriptionFlowsTest and MqttSubscriptionFlowTreeTest

44d0093

Improved FlowableWithSingleObserveOn/FluxWithSinglePublishOn

6feca4c

Removed Flowable/FluxWithSingleCombine which had concurrency issues

Improved FlowableWithSingleObserveOn: delay complete/error when singl…

09d9db0

…e is pending

Improved MqttPublishFlowableAckLink: ensured request and cancel are c…

3334672

…alled serially

Refactored MqttAckSingleFlowable and MqttAckSingle

bf8899e

Small refactorings

056e4ac

Improved FlowableWithSingleObserveOn/FluxWithSinglePublishOn

c1a22f9

Restored Flowable/FluxWithSingleCombine and fixed concurrency issues

Improved Flowable/FluxWithSingleCombine

993b9ca

Improved MqttSubscriptionHandler for automatic resubscribe

4308b56

Refactored MqttOutgoing/IncomingQosHandler similar to MqttSubscriptio…

0341918

…nHandler

Unify MqttSubscriptionHandler and MqttOutgoingQosHandler

d1b72ab

Reduce flush calls for many subscribe packets (reduces used network b…

a057665

…andwidth when restoring subscriptions)

Renamed MqttSubscriptionFlows -> MqttSubscribedPublishFlows as this i…

828edad

…s more consistent: MqttIncomingPublishFlow(s) and MqttSubscribedPublishFlow(s))

SgtSilvio force-pushed the feature/automatic-resubscribe branch from 7b7a86e to 828edad Compare April 17, 2020 09:21

sauroter self-requested a review April 17, 2020 09:21

sauroter reviewed Apr 17, 2020

View reviewed changes

src/main/java/com/hivemq/client/internal/mqtt/handler/subscribe/MqttSubscriptionHandler.java Show resolved Hide resolved

sauroter reviewed Apr 17, 2020

View reviewed changes

src/main/java/com/hivemq/client/internal/mqtt/handler/subscribe/MqttSubscriptionHandler.java Show resolved Hide resolved

sauroter reviewed Apr 17, 2020

View reviewed changes

sauroter approved these changes Apr 17, 2020

View reviewed changes

SgtSilvio added 2 commits April 17, 2020 23:20

Removed dependency injection from MqttPublishFlowables and MqttSubscr…

1bd97f4

…ibedPublishFlowTree, reasons: the first is tightly coupled to the MqttOutgoingQosHandler the second is similar to a collection which is instantiated

Added Reconnector.resubscribeIfSessionExpired and republishIfSessionE…

0d53c11

…xpired

SgtSilvio merged commit d7b38e9 into develop Apr 18, 2020

SgtSilvio deleted the feature/automatic-resubscribe branch April 18, 2020 21:17

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Feature/automatic resubscribe #398

Feature/automatic resubscribe #398

SgtSilvio commented Apr 6, 2020 •

edited

Loading

sauroter Apr 17, 2020

SgtSilvio Apr 17, 2020

SgtSilvio Apr 17, 2020

sauroter Apr 17, 2020

SgtSilvio Apr 17, 2020

sauroter Apr 17, 2020 •

edited

Loading

SgtSilvio Apr 17, 2020

sauroter Apr 17, 2020

SgtSilvio Apr 18, 2020

sauroter left a comment •

edited

Loading

Feature/automatic resubscribe #398

Feature/automatic resubscribe #398

Conversation

SgtSilvio commented Apr 6, 2020 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sauroter Apr 17, 2020 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sauroter left a comment • edited Loading

Choose a reason for hiding this comment

SgtSilvio commented Apr 6, 2020 •

edited

Loading

sauroter Apr 17, 2020 •

edited

Loading

sauroter left a comment •

edited

Loading