GH-3685: Share MQTT connection across components #3857

oxcafedead · 2022-07-21T13:31:50Z

Introduce some initial design.
Add a new interface MqttClientManager which will manage clients and
connections.
Use this manager in v3 topic adapter and message handler.

oxcafedead · 2022-07-21T13:33:28Z

Hi @artembilan
Can you please review - this is not a complete implementation but rather some approach. I have replaced v3 classes and adjusted tests to prove it's working (and added a couple of new tests as well).
I'd like to hear your feedback. Is this something which you think may be a solution?

...egration-mqtt/src/main/java/org/springframework/integration/mqtt/core/MqttClientManager.java

...n/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java

artembilan

You know doesn't look so bad.
But still some concerns to clean up.
I wish we have started from a central management for the client from day first...

...-integration-mqtt/src/main/java/org/springframework/integration/mqtt/core/ClientManager.java

...ration-mqtt/src/main/java/org/springframework/integration/mqtt/core/Mqttv3ClientManager.java

...n/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java

...java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java

oxcafedead · 2022-07-27T13:27:21Z

I've added some changes to use client manager in adapters and handlers, also added support for both versions of clients.
TODO: unit tests covering shared client usage, possible bugfixes, XSD schemas and parsers update, documentation. (also squash commits)

artembilan

TODO: unit tests covering shared client usage, possible bugfixes, XSD schemas and parsers update, documentation. (also squash commits)

Unit tests are great to have.
The rest of work can be done in the separate PRs.
No, please, not squashing: we do that on merge.
It is better to preserve commits history in the PR as is: must easier to review.

...-mqtt/src/main/java/org/springframework/integration/mqtt/core/AbstractMqttClientManager.java

...ration-mqtt/src/main/java/org/springframework/integration/mqtt/core/Mqttv3ClientManager.java

...java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java

...ration-mqtt/src/main/java/org/springframework/integration/mqtt/core/Mqttv5ClientManager.java

...on-mqtt/src/test/java/org/springframework/integration/mqtt/ClientManagerBackToBackTests.java

artembilan

@oxcafedead ,

your contribution is great and very valuable.
No need to worry about me: this is my project and it is my responsibility to review your changes as long as it is needed.

That's my apologies that this task takes more time, than it looked originally: you see I also change my mind from time to time. Plus indeed I feel like I learn something from you as well 😄

So, big thanks to you again and keep her steady!

Let me know when you are going to give up, so I'll take from there my self: we still have to implement this feature since it looks like very reasonable from target use-case 😉

...ration-mqtt/src/main/java/org/springframework/integration/mqtt/core/Mqttv3ClientManager.java

...ration-mqtt/src/main/java/org/springframework/integration/mqtt/core/Mqttv5ClientManager.java

...va/org/springframework/integration/mqtt/inbound/AbstractMqttMessageDrivenChannelAdapter.java

...on-mqtt/src/test/java/org/springframework/integration/mqtt/ClientManagerBackToBackTests.java

artembilan

Please, consider to rebase your branch to the latest main: we finally switched to the CompletableFuture whenever ListenableFuture is deprecated.
This way your local build won't fail for some unexpected reasons 😄

oxcafedead · 2022-07-29T12:46:04Z

@artembilan - I will push latest changes with rebase in an hour.

Fixes spring-projects#3685 Introduce some initial design. Add a new interface `ClientManager` which will manage clients and connections. Use this manager in v3 topic adapter and message handler.

Fixes spring-projects#3685 Add a new interface `ClientManager` which will manage clients and connections. Add different implementations for v3 and v5 MQTT clients. Use this manager in v3/v5 topic adapters and message handlers.

Fixes spring-projects#3685 Add a couple of unit/integration tests to cover client manager usage. Several small code improvements after the code review: * Improve client manager usage via providing several mutual exclusive constructors, whether the users provides `url` or `connectionOptions` or `clientFactory` for v3. * Move the logger to `AbstractMqttClientManager` * Do not inject TaskScheduler in constructor for v3 client manager but use lazy init via `BeanFactory` and `IntegrationContextUtils` * Other smaller code readability improvements

Fixes spring-projects#3685 Add new tests with reconnect cases. Other code improvements after the code review: * Adjust javadocs according to standards * Remove `setClientManager` and use exclusive ctors * Make automatic reconnects using the v3 client instead of manually using task scheduler

artembilan

I didn't review v3 variant because I was not involved in its implementation back in days, but I believe with our discussion to realign both protocols in their ClientManagers my v5 review will apply there as well.

...-mqtt/src/main/java/org/springframework/integration/mqtt/core/AbstractMqttClientManager.java

...ration-mqtt/src/main/java/org/springframework/integration/mqtt/core/Mqttv5ClientManager.java

...java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java

artembilan · 2022-07-29T17:08:16Z

...java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java

+				MessageListener[] listeners = IntStream.range(0, topics.length)
+						.mapToObj(t -> new MessageListener())
+						.toArray(MessageListener[]::new);
+				getClientManager().getClient().subscribe(subscriptions, null, null, listeners, null)


I wonder what is wrong with this method:

public IMqttToken subscribe(MqttSubscription[] subscriptions, IMqttMessageListener messageListener) throws MqttException;

and remember we talked about using a MessageListener even of an internally controlled client?

That's why I'm asking about that this.mqttClient = getClientManager().getClient()

Initially I have used this method you mentioned - but it gives a runtime exception. Yeah, I found it when ran my back-to-back integration tests with mosquitto. And it seems like the bug in the client lib.
org/eclipse/paho/mqttv5/client/MqttAsyncClient.java:1220 - passes an empty new MqttProperties()
and then org/eclipse/paho/mqttv5/client/MqttAsyncClient.java:1276 throws no such element

and remember we talked about using a MessageListener even of an internally controlled client?

Probably some other misunderstanding... sorry, I don't get it. MessageListener is the internal class of adapter which is a bridge: it implements the interface IMqttMessageListener of the library and reuses the existing logic of messageArrived method of the adapter class.

The messageArrived() is not this class method. It is dictated by that MqttCallback we implement on this class.
With my finding it looks like this messageArrived() and IMqttMessageListener are exactly for the same purpose.
With only a difference that MqttCallback is per client, but IMqttMessageListener is per request. Which is exactly what we need for our "shared client" use-case.
At the same time I want to see if the same IMqttMessageListener can be used for our case where we create client internally.

I also wonder if it really makes sense to new MessageListener() for every request, but not have it as a single one per channel adapter and just reuse for all the requests.
Kinda some minor, but still optimization.

org/eclipse/paho/mqttv5/client/MqttAsyncClient.java:1220 - passes an empty new MqttProperties()
and then org/eclipse/paho/mqttv5/client/MqttAsyncClient.java:1276 throws no such element

That is good catch! And I can confirm the bug is there.
Here is a bug report related: eclipse-paho/paho.mqtt.java#826

.../src/main/java/org/springframework/integration/mqtt/outbound/AbstractMqttMessageHandler.java

...tt/src/main/java/org/springframework/integration/mqtt/outbound/Mqttv5PahoMessageHandler.java

oxcafedead · 2022-07-31T12:46:19Z

@artembilan I will push all the changes regarding you last review which I understand 🙂 and marked as resolved.
To my regret, since next week I will have less time as my vacation ends 😄 so apologies in advance if some changes will take more time to appear...

Fixes spring-projects#3685 Some fixes and improvements after another code review iteration: * Rearrange the code according to the code style guides * Move client instance to `AbstractClientManager` with `isRunning` method * Fix abstract adapter/handler fields visibility and `final`ize them where we can * Send application event if automatic reconnect is not enabled for the client manager

artembilan

I would say we are very close to the final solution.

Please, take a look what you can do with this my latest review.

...-mqtt/src/main/java/org/springframework/integration/mqtt/core/AbstractMqttClientManager.java

...java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java

.../src/main/java/org/springframework/integration/mqtt/outbound/AbstractMqttMessageHandler.java

...tt/src/main/java/org/springframework/integration/mqtt/outbound/Mqttv5PahoMessageHandler.java

artembilan · 2022-08-02T17:48:15Z

...on-mqtt/src/test/java/org/springframework/integration/mqtt/ClientManagerBackToBackTests.java

+ * @author Artem Vozhdayenko
+ * @since 6.0
+ */
+@LongRunningTest


How slow is this so it has driven you to for such a marker?

It took ~18 seconds.

Oops, just found - after implementing event-based subscription and count down latches in tests - now tests pass in 4 seconds (last run result).
Maybe it makes sense to remove this @LongRunningTest anno

I think so. Thanks. I believe we need to take into account the start for the Mosquito container anyway

Are you going to remove this annotation?
How slow is the test right now?

Last run time it was ~6.5 secs. I suppose we can remove it now, since we have other Mosquitto tests without it, plus the container creation time is 'shared' among these tests, as the container should be reused if I am not mistaken.
I will remove it.

oxcafedead · 2022-08-04T08:55:30Z

@artembilan - I just ran all tests again and now I got the issue with client not connected in doStart again (after I have removed reconnect() invocation). For some reason, the default timeout is not enough. Sometimes. It sporadically passes. Probably the tests order matters? If container has not managed to fully start yet. However, sometimes I have 3 out of 4 failed - it is 20 * 3 = a whole minute of mqtt server unavailability. Strange.

2022-08-04 01:18:18,399 ERROR [Test worker] [org.springframework.integration.mqtt.core.Mqttv5ClientManager] - could not start client manager, client_id=client-manager-client-id-v5
org.eclipse.paho.mqttv5.common.MqttException: Timed out waiting for a response from the server

As the result, it does not reconnect immediately and test is failed with null message after the timeout, because the adapter has not subscribed to topics (which happens only on doStart method). There is a convenient callback connectComplete(isReconnect) to determine if the client has been connected the first time to subscribe. We could improve the logic by passing this callback through the client manager to adapter class. I will try to make it and see if tests will be stable and green.

oxcafedead · 2022-08-04T12:50:59Z

The approach with connectComplete broke v3 adapter because it contradicts with manual scheduleReconnects.
This becomes more messy, I will get back to it when I have time...

artembilan · 2022-08-04T16:55:09Z

The approach with connectComplete broke v3 adapter

That's correct. That's why at some point I asked if we can re-write v3 channel adapter logic to avoid that manual reconnetion.
Kinda implement it the same way we have now in v5 version of the channel adapter.

The Mqttv5PahoMessageDrivenChannelAdapter has that connectComplete(boolean reconnect, String serverURI) impl from the MqttCallback to make a subscription when we really have connected to the server.
This one works for an internal client.
Since we cannot provide this MqttCallback to a shared client, we have to implement something like addConnectCompleteCallback(BiConsumer<boolean, String> connectCompleteCallback) in the ClientManager.
Store all the provided into a List and delegate to them from the ClientManager 's MqttCallback.connectComplete() impl.
That add is going to be called from the onInit() (or doStart()) of the message-driver channel adapter.
Of course, it sounds like removeConnectCompleteCallback() has to be there as well to be called from doStop() or destroy().

artembilan · 2022-08-04T16:57:02Z

...ration-mqtt/src/main/java/org/springframework/integration/mqtt/core/Mqttv3ClientManager.java

@@ -96,6 +90,9 @@ public synchronized void start() {
 					logger.error("MQTT client failed to re-connect.", ex);
 				}
 			}
+			else if (getApplicationEventPublisher() != null) {
+				getApplicationEventPublisher().publishEvent(new MqttConnectionFailedEvent(this, e));


Better to extract to the local variable and avoid extra method call.
I don't think the compiler is smart enough to do such an optimization on our behalf...

artembilan · 2022-08-04T16:57:18Z

...ration-mqtt/src/main/java/org/springframework/integration/mqtt/core/Mqttv5ClientManager.java

@@ -95,6 +89,9 @@ public synchronized void start() {
 					logger.error("MQTT client failed to re-connect.", ex);
 				}
 			}
+			else if (getApplicationEventPublisher() != null) {
+				getApplicationEventPublisher().publishEvent(new MqttConnectionFailedEvent(this, e));


artembilan · 2022-08-04T16:59:28Z

...va/org/springframework/integration/mqtt/inbound/AbstractMqttMessageDrivenChannelAdapter.java

@@ -74,7 +74,7 @@

 	private MqttMessageConverter converter;

-	protected ClientManager<T> clientManager;
+	private final ClientManager<T> clientManager;


Just nit-pick: final props must be before the others.

artembilan · 2022-08-04T17:00:52Z

...n/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java

-	public MqttPahoMessageDrivenChannelAdapter(String url, String clientId, String... topic) {
-		this(url, clientId, new DefaultMqttPahoClientFactory(), topic);
+	public MqttPahoMessageDrivenChannelAdapter(ClientManager<IMqttAsyncClient> clientManager, String... topic) {
+		this(clientManager, new MqttConnectOptions(), topic);


I thought that we have discussed: no contradicting options. Better to take them from the provided client.

I still haven't pushed any changes yet 🙂
Sorry, maybe the fact I resolve our conversations confuses you. I do it for myself to check that locally I have resolved this. Probably better to leave them unless the code it on 'your' side..

I see . Sorry, yes, makes sense to have it resolved that way. Really confused me: have a lot to review in parallel , so I lose context from time to time 😅

...java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java

artembilan · 2022-08-04T17:07:46Z

...java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java

@@ -162,6 +204,11 @@ protected void onInit() {

 	@Override
 	protected void doStart() {
+		if (getClientManager() != null) {
+			subscribeToAll();


Oh! I see now: you have broken the logic.
We must not try to subscribe in the doStart().
We have to try to do that exactly from the connectComplete() callback.
The problem is present just because mqttClient.reconnect() is an async scheduled operation.
So, even if client manager is started, it doesn't mean that it has been connected.
Therefore we really need to wait for the connectComplete() before trying to subscribe.
Please, see my other comment where I suggest to implement addConnectCompleteCallback()

Since we cannot provide this MqttCallback to a shared client, we have to implement something like addConnectCompleteCallback(BiConsumer<boolean, String> connectCompleteCallback) in the ClientManager.

I already did smth similar but I have created an inner Callback interface in ClientManager with semantically clear method connectComplete(boolean isReconnect, String serverURI) - I guess it should be easier for users to understand the purpose of the callback when they see parameter and method names then just plain BiConsumer with types. If you don't have objections I would like to use this interface.

Sure, we can go that way, but let’s name it a ConnectCallback to be more specific . Also I don’t see a reason in the url param: we definitely know to what manager we provide a callback.

artembilan · 2022-08-04T17:17:24Z

...java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java

+				MessageListener[] listeners = IntStream.range(0, topics.length)
+						.mapToObj(t -> new MessageListener())
+						.toArray(MessageListener[]::new);
+				getClientManager().getClient().subscribe(subscriptions, null, null, listeners, null)


org/eclipse/paho/mqttv5/client/MqttAsyncClient.java:1220 - passes an empty new MqttProperties()
and then org/eclipse/paho/mqttv5/client/MqttAsyncClient.java:1276 throws no such element

That is good catch! And I can confirm the bug is there.
Here is a bug report related: eclipse-paho/paho.mqtt.java#826

oxcafedead · 2022-08-06T15:22:10Z

That's correct. That's why at some point I asked if we can re-write v3 channel adapter logic to avoid that manual reconnetion.
Kinda implement it the same way we have now in v5 version of the channel adapter.

I think this will be much better and cleaner, to get rid of this task scheduler and scheduled reconnects.
However, this will be not backwards compatible. Now, if automatic reconnect is turned off for some reason, the client will still be automatically recreated in scheduleReconnect method. I know this is an obvious bug - but may not be so obvious for some users 🙂
If you don't see here any issues - fine by me.

artembilan · 2022-08-06T15:27:36Z

If you don't see here any issues - fine by me.

thanks for confirmation. No, I don’t see an issue . This is a major release, so it is fully fine to some breaking changes. More over I believe most solution are based on just an url for MQTT broker in which case we add an auto-recconnect option internally . Either way we are going to document this properly .

oxcafedead · 2022-08-06T15:35:42Z

If you don't see here any issues - fine by me.

thanks for confirmation. No, I don’t see an issue . This is a major release, so it is fully fine to some breaking changes. More over I believe most solution are based on just an url for MQTT broker in which case we add an auto-recconnect option internally . Either way we are going to document this properly .

OK, and regarding recoveryInterval - seems like this property becomes useless. Can we just remove it as well?

artembilan · 2022-08-06T16:14:06Z

OK, and regarding recoveryInterval - seems like this property becomes useless. Can we just remove it as well?

yes, this makes sense . Let’s remove it then! Thank you for the pointer!

Honestly, my idea was to implement the feature for v5. Confirm that it is OK and then rework v3 accordingly . Now it’s up to you to go ahead with your convenient way.

oxcafedead · 2022-08-06T16:18:16Z

yes, this makes sense . Let’s remove it then! Thank you for the pointer!

Honestly, my idea was to implement the feature for v5. Confirm that it is OK and then rework v3 accordingly . Now it’s up to you to go ahead with your convenient way.

As far as I understand, this functionality can be covered by MqttConnectionOptions.automaticReconnectMinDelay, MqttConnectionOptions.automaticReconnectMaxDelay so users should have this flexibility - however not configurable via the XML approach conveniently.

artembilan · 2022-08-06T16:31:02Z

Right , we just remove that option from the XSD as well. Let’s talk about an XML support when we are done with this. No reason to cover everything in one PR: the bigger it is - the slower review and merge process.

Fixes spring-projects#3685 Other fixes and improvements after code review: * Changes around fields, methods, ctors visibility * Removed contradictory ctors * Reduce amount of unnecessary `getClientManager() != null` checks in logic and make it as similar as possible for client manager and the old approach * Use auto-reconnect where possible * Remove manual reconnect trigger and rely on events instead to know where to subscribe * Do not close the connection in adapter to be able to use reconnect logic without lose of subscriptions * Make `ClientManager` extend `MqttComponent` so that it knows about connection options as part of its contract * Remove not relevant auto test cases (relying on connection close or manual reconnect) * Other code style smaller changes

oxcafedead · 2022-08-06T18:24:12Z

Phew, finally, another push... a lot of changes. Tests pass locally, finally

artembilan

The review for current state of code.
Again: I didn't review v3 changes, just because I didn't find v5 implemented correctly yet.

Thanks and sorry that this task became so complex.

...-integration-mqtt/src/main/java/org/springframework/integration/mqtt/core/ClientManager.java

...-mqtt/src/main/java/org/springframework/integration/mqtt/core/AbstractMqttClientManager.java

artembilan · 2022-08-08T21:16:09Z

...-mqtt/src/main/java/org/springframework/integration/mqtt/core/AbstractMqttClientManager.java

+	}
+
+	protected Set<ConnectCallback> getCallbacks() {
+		return Collections.unmodifiableSet(this.connectCallbacks);


The method is protected, so unlikely going to be used for some outside harm.
Therefore I don't see a reason in extra wrapping to prevent modification.

Just the defense if someone decides to implement own client manager incorrectly, but I will remove it, no problem

to implement and to use is not the same and there is likely less harm when it is an internal logic than some end-user API.

artembilan · 2022-08-08T21:20:25Z

...ration-mqtt/src/main/java/org/springframework/integration/mqtt/core/Mqttv5ClientManager.java

+		catch (MqttException e) {
+			logger.error("could not start client manager, client_id=" + getClientId(), e);
+
+			var applicationEventPublisher = getApplicationEventPublisher();


Why no reconnect logic as we discussed?
Exactly the same what we have right now in the Mqttv5PahoMessageDrivenChannelAdapter.doStart().
That exactly was a purpose of our discussion around that setAutomaticReconnect(true) option.

Sorry, I don't see any reason to call reconnect manually because if automatic reconnect is enabled, the client is reconnected automatically anyway (tests show it works). This reconnect just zeroes out the timer.
Also if automatic reconnect is disabled, I thought we have discussed that the user gets info message about it and have to subscribe to MqttConnectionFailedEvent

Well, see this issue for more info: #3822.

In two words: if client is not connected via first connect(), the reconnect() is not initiated.
That's why we have that logic in the Mqttv5PahoMessageDrivenChannelAdapter.doStart().

I wonder if you have modified the Mqttv5BackToBackAutomaticReconnectTests to adjust its behavior to the logic you have changed.
However it should be like: "why this test is failing when I have removed reconnect()"?

WDYT?

Ok I see now, I was really sure that there was no point for reconnect but this makes sense now. Indeed I should have made git blame, sometimes I think that smth might be implemented for a reason but not this time, haha.

artembilan · 2022-08-08T21:25:55Z

...va/org/springframework/integration/mqtt/inbound/AbstractMqttMessageDrivenChannelAdapter.java

+		return this.clientManagerConnectCallback;
+	}
+
+	protected void setClientManagerCallback(ClientManager.ConnectCallback clientManagerConnectCallback) {


That's not.
The AbstractMqttMessageDrivenChannelAdapter creates such a ConnectCallback internally and register it with the provided ClientManager.
See, for example, an AbstractRSocketConnector and its addEndpoint() call from the RSocketInboundGateway.
No reason to have some extra API around: we just call getClientManager().addCallback() from target channel adapter impl.
I don't see, though, how you connect two of them here...

This setter and field is just to be able to save the listener instance to remove it later: see in destroy method,

if (clientManager != null) { clientManager.removeCallback(getClientManagerCallback()); }

Ah! I see. Yep, you are right!
How about something like clientManager.addCallback(this) and clientManager.removeCallback(this)?
And have that ClientManager.ConnectCallback impl directly on the AbstractMqttMessageDrivenChannelAdapter.
This is an impl for just one more method which is going to delegate to another.
And there won't be those protected settter and getter.

Or is this going to be too confusing from code perspective?
For end-user nothing changed though: they even not going to use this API in most cases.

I think both approaches are not ideal 🙂 - but I guess having protected getters/setters is kinda an overkill for it.
I like your approach with (this) better. I will implement it.

.../src/main/java/org/springframework/integration/mqtt/outbound/AbstractMqttMessageHandler.java

...tt/src/main/java/org/springframework/integration/mqtt/outbound/Mqttv5PahoMessageHandler.java

artembilan · 2022-08-08T21:42:40Z

...tt/src/main/java/org/springframework/integration/mqtt/outbound/Mqttv5PahoMessageHandler.java

+
+	private static MqttConnectionOptions buildDefaultConnectionOptions(@Nullable String url) {
+		final MqttConnectionOptions connectionOptions;
+		connectionOptions = new MqttConnectionOptions();


Why divide declaration and initialization to two separate lines?

...ng-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java

artembilan · 2022-08-08T21:49:21Z

.../test/java/org/springframework/integration/mqtt/Mqttv5BackToBackAutomaticReconnectTests.java

-				.withCauseExactlyInstanceOf(MqttException.class)
-				.withRootCauseExactlyInstanceOf(UnknownHostException.class);
-
-		connectionOptions.setServerURIs(new String[]{ MosquittoContainerTest.mqttUrl() });


Didn't you remove this as a collateral damage just because you have removed a reconnect() from the channel adapter when client is created internally?

artembilan · 2022-08-09T13:47:01Z

...ration-mqtt/src/main/java/org/springframework/integration/mqtt/core/Mqttv5ClientManager.java

+		catch (MqttException e) {
+			logger.error("could not start client manager, client_id=" + getClientId(), e);
+
+			var applicationEventPublisher = getApplicationEventPublisher();


Well, see this issue for more info: #3822.

In two words: if client is not connected via first connect(), the reconnect() is not initiated.
That's why we have that logic in the Mqttv5PahoMessageDrivenChannelAdapter.doStart().

I wonder if you have modified the Mqttv5BackToBackAutomaticReconnectTests to adjust its behavior to the logic you have changed.
However it should be like: "why this test is failing when I have removed reconnect()"?

WDYT?

artembilan · 2022-08-09T13:51:34Z

...va/org/springframework/integration/mqtt/inbound/AbstractMqttMessageDrivenChannelAdapter.java

+		return this.clientManagerConnectCallback;
+	}
+
+	protected void setClientManagerCallback(ClientManager.ConnectCallback clientManagerConnectCallback) {


Ah! I see. Yep, you are right!
How about something like clientManager.addCallback(this) and clientManager.removeCallback(this)?
And have that ClientManager.ConnectCallback impl directly on the AbstractMqttMessageDrivenChannelAdapter.
This is an impl for just one more method which is going to delegate to another.
And there won't be those protected settter and getter.

Or is this going to be too confusing from code perspective?
For end-user nothing changed though: they even not going to use this API in most cases.

...tt/src/main/java/org/springframework/integration/mqtt/outbound/Mqttv5PahoMessageHandler.java

.../src/main/java/org/springframework/integration/mqtt/outbound/AbstractMqttMessageHandler.java

artembilan · 2022-08-09T14:06:52Z

.../test/java/org/springframework/integration/mqtt/Mqttv5BackToBackAutomaticReconnectTests.java

@@ -104,7 +92,7 @@ public static class Config {
 		@Bean
 		public MqttConnectionOptions mqttConnectOptions() {
 			return new MqttConnectionOptionsBuilder()
-					.serverURI("wss://badMqttUrl")
+					.serverURI(MosquittoContainerTest.mqttUrl())


There was really the point to have this test exactly this way.
Please, see my comment about remove reconnect() an related issue when that feature and this test have need introduced.
And consider to reinstate that logic and this test.

Thank you!

Fixes spring-projects#3685 Other fixes and improvements after code review: * Get manual `reconnect` invocation back for v3/v5 adapters and client managers (see bug spring-projectsGH-3822 for a reasoning) * Remove unnecessary getters/setter for a listener and use adapter class as listener instead * Optimize MessageListener: remove redundant inner class and use a single method reference instead of N instances per each subscribe * Javadocs improvements

artembilan · 2022-08-09T15:20:04Z

GH-3685: Share MQTT connection across components

No need to add a header to every single commit.
In the end, on merge, I'm going to squash them all to a single one.

I prefer to add extra commits just with the messages about changes in this commit.
Now I see what you meant about adding them, but seeing a header on all your commits it confuses me since the rest of comments are hidden. 😄

This probably was just my bad habit to believe that people does not add comments into their commits.
Sorry I will try to be more careful net time

artembilan

Good progress!

Thank you for your patience!

...-mqtt/src/main/java/org/springframework/integration/mqtt/core/AbstractMqttClientManager.java

artembilan · 2022-08-09T15:52:15Z

...-mqtt/src/main/java/org/springframework/integration/mqtt/core/AbstractMqttClientManager.java

+	AbstractMqttClientManager(String clientId) {
+		Assert.notNull(clientId, "'clientId' is required");
+		this.clientId = clientId;
+		this.connectCallbacks = Collections.synchronizedSet(new HashSet<>());


I thought that I have commented before that this initialization could be done directly on the property declaration.

artembilan · 2022-08-09T15:55:49Z

...ration-mqtt/src/main/java/org/springframework/integration/mqtt/core/Mqttv5ClientManager.java

+					.waitForCompletion(this.connectionOptions.getConnectionTimeout());
+		}
+		catch (MqttException e) {
+			logger.error("could not start client manager, client_id=" + getClientId(), e);


I don't think we need this noise in logs since we are going to initiate a reconnect action.
Otherwise we are going to publish that MqttConnectionFailedEvent.
We may log in the third (very rare) case when there is no applicationEventPublisher provided.

artembilan · 2022-08-09T15:57:48Z

...va/org/springframework/integration/mqtt/inbound/AbstractMqttMessageDrivenChannelAdapter.java

+		this.clientId = null;
+	}
+
+	private Set<Topic> initTopics(String[] topic) {


artembilan · 2022-08-09T16:06:23Z

...java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java

+		this.connectionOptions = clientManager.getConnectionInfo();
+	}
+
+	private static MqttConnectionOptions buildDefaultConnectionOptions(@Nullable String url) {


I think we don't need this method any more.
Also I think we just need to add Assert.hasText(url, "'url' cannot be null or empty"); into that url-based ctor.
Just because we cannot an empty url without external options: or url, or options with server uris.

artembilan · 2022-08-09T16:07:57Z

...java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java

@@ -158,30 +185,39 @@ protected void onInit() {
 					.getBean(IntegrationContextUtils.ARGUMENT_RESOLVER_MESSAGE_CONVERTER_BEAN_NAME,
 							SmartMessageConverter.class));
 		}
+		if (clientManager != null) {
+			clientManager.addCallback(this);


Cannot we do this in a single place in the super.onInit() ?

artembilan · 2022-08-09T16:08:24Z

...java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java

 				this.mqttClient.close(true);
 			}
 		}
 		catch (MqttException ex) {
 			logger.error(ex, "Failed to close 'MqttAsyncClient'");
 		}
+		if (clientManager != null) {
+			clientManager.removeCallback(this);


Same here about moving this logic into a common super.destroy()

artembilan · 2022-08-09T16:12:50Z

...tt/src/main/java/org/springframework/integration/mqtt/outbound/Mqttv5PahoMessageHandler.java

+		this.connectionOptions = clientManager.getConnectionInfo();
+	}
+
+	private static MqttConnectionOptions buildDefaultConnectionOptions(@Nullable String url) {


I still think that we don't need this method and url must not be null for that url-based ctor.
Exactly the same comment I did for the message-driven channel adapter.

Sure. Makes sense.

artembilan · 2022-08-09T16:17:47Z

...tt/src/main/java/org/springframework/integration/mqtt/outbound/Mqttv5PahoMessageHandler.java

+			}
+			else {
+				if (!this.mqttClient.isConnected()) {
+					this.mqttClient.connect(this.connectionOptions).waitForCompletion(completionTimeout);


I think we have to try to reconnect over here independently of the client source.
Even if client is provided by the manager, it is still legit to call this connect() in case of lost connection in the client.

I guess you are right. Won't be harmful - except that if client is being connected in background, connect method will throw an MqttException. But this is probably a rare case.

artembilan · 2022-08-09T16:18:45Z

...on-mqtt/src/test/java/org/springframework/integration/mqtt/ClientManagerBackToBackTests.java

+ * @author Artem Vozhdayenko
+ * @since 6.0
+ */
+@LongRunningTest


Are you going to remove this annotation?
How slow is the test right now?

* Extract common callback add/rm logic to abstract adapter class * Small code cleanups/fixes related to code style & simplicity, ctor inits and unnecessary methods; eliminate unnecessary logs noise * Remove `@LongRunningTest` for `ClientManagerBackToBackTests` as test run time is ~6-7 secs

artembilan

Some further review.

artembilan · 2022-08-10T15:28:10Z

...-mqtt/src/main/java/org/springframework/integration/mqtt/core/AbstractMqttClientManager.java

+
+	private String url;
+
+	private volatile T client;


Just another nit-pick: volatile props go after all others 😄

artembilan · 2022-08-10T15:30:49Z

...-mqtt/src/main/java/org/springframework/integration/mqtt/core/AbstractMqttClientManager.java

+	}
+
+	/**
+	 * Sets the phase of component autostart in {@link SmartLifecycle}.


The method Javadoc must be imperative, not infinitive: https://github.com/spring-projects/spring-framework/wiki/Code-Style#javadoc-formatting

...ration-mqtt/src/main/java/org/springframework/integration/mqtt/core/Mqttv5ClientManager.java

artembilan · 2022-08-10T15:37:27Z

...java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java

@@ -221,7 +250,8 @@ public void addTopic(String topic, int qos) {
 		try {
 			super.addTopic(topic, qos);
 			if (this.mqttClient != null && this.mqttClient.isConnected()) {
-				this.mqttClient.subscribe(topic, qos).waitForCompletion(getCompletionTimeout());
+				this.mqttClient.subscribe(new MqttSubscription(topic, qos), this::messageArrived)


Don't we need to subscribe to newly added topics even if the client comes from the manager?

We init mqttClient field from the client manager in the start method, just as for v3 adapter 🙂
So this code is almost not changed, at least no additional ifs

artembilan · 2022-08-10T15:42:00Z

...tt/src/main/java/org/springframework/integration/mqtt/outbound/Mqttv5PahoMessageHandler.java

 		}
 	}

 	@Override
 	protected void doStop() {
 		try {
-			this.mqttClient.disconnect().waitForCompletion(getDisconnectCompletionTimeout());
+			if (this.mqttClient != null) {


Since you do now:

else if (clientManager != null) { this.mqttClient = clientManager.getClient(); }

in the doStart(), we have to check here for the getClientManager() != null before performing disconnect().
As we agreed: we cannot not close the client we don't owe.
Same applies for the close() later on in the destroy()

You are right - changed it at the last moment and missed this part

I wonder if the mqttClient != null check is redundant in doStart/doStop/destroy methods. It cannot be null from onInit method - if client cannot be created we throw BeanCreationException

artembilan

That's it.
Reviewed all of them.

Thank you!

artembilan · 2022-08-10T19:51:21Z

...ration-mqtt/src/main/java/org/springframework/integration/mqtt/core/Mqttv3ClientManager.java

+		this(buildDefaultClientFactory(url), clientId);
+	}
+
+	public Mqttv3ClientManager(MqttPahoClientFactory clientFactory, String clientId) {


I wonder if we still need to rely on this factory for this our new ClientManager abstraction.
How about just to make this v3 version similar to v5 - based on the MqttConnectionOptions.
And perhaps we need to add one more MqttClientPersistence option to these managers impls.
A setter should be OK.

Ok, initially if I remember correctly you asked to implement client manager using existing client factory abstraction 😄

...ration-mqtt/src/main/java/org/springframework/integration/mqtt/core/Mqttv3ClientManager.java

...n/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java

artembilan · 2022-08-10T19:56:44Z

...n/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java

-		if (this.client instanceof MqttClient) {
-			((MqttClient) this.client).setTimeToWait(getCompletionTimeout());
+
+		long completionTimeout = getCompletionTimeout();


No reason to extract this into a variable: used only in one place.

artembilan · 2022-08-10T19:58:06Z

...n/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java

+	private void subscribe() {
+		var clientManager = getClientManager();
+		if (clientManager != null && this.client == null) {
+			this.client = clientManager.getClient();


I wonder if we still need this logic in the subscribe(). Looks like you are extracting this.client in the connect(), so should be OK here in the subscribe() to not worry about that property value any more.
Do I miss anything?

I guess you're right, no sense to have it duplicated

artembilan · 2022-08-10T20:00:44Z

...n/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java

+				IMqttToken subscribeToken = this.client.subscribe(topics, requestedQos, listeners);
+				subscribeToken.waitForCompletion(getCompletionTimeout());
+				int[] grantedQos = subscribeToken.getGrantedQos();
+				if (grantedQos.length == 1 && grantedQos[0] == 0x80) {


Magic number? // NOSONAR ?

This is an interesting case 🙂
Initially v3 adapter had a blocking client impl and I have changed it to use async client.
One test case started failing after that, it was MqttAdapterTests#testSubscribeFailure.
You can see it mocks this magic number 0x80:
given(token.getGrantedQos()).willReturn(new int[]{ 0x80 });

I have looked through the code of a blocking client and it is just a wrapper around async one - but it really has this strange difference in subscribe method, /org/eclipse/paho/client/mqttv3/MqttClient.java:465

if (grantedQos.length == 1 && qos[0] == 0x80) { throw new MqttException(MqttException.REASON_CODE_SUBSCRIBE_FAILED); }

So I decided to keep the behavior, especially given we had a unit test for it.

I will add //NOSONAR for now..

Sure! Makes sense. Let's live with this for now only in v3 impl!

artembilan · 2022-08-10T20:03:48Z

...java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java

+					.mapToObj(t -> listener)
+					.toArray(IMqttMessageListener[]::new);
+			this.mqttClient.subscribe(subscriptions, null, null, listeners, null)
+					.waitForCompletion(getCompletionTimeout());


Don't we need to do the same subscribeToken.getGrantedQos() logic over here as well, like you have introduced in v3 impl?

For v5 we didn't have it initially. So I am not sure. Probably not?

Hm, I don't find this 0x80 in the blocking client in v5 as well. Probably it was something related to v3 only.

plain connection properties and client persistence instead * Add missed javadocs * Other code style & cleanup improvements

artembilan

I think that's it.
I'll pull this locally today for final review and possible merging.

Thank you very much for a great work!

artembilan · 2022-08-11T15:47:08Z

...-mqtt/src/main/java/org/springframework/integration/mqtt/core/AbstractMqttClientManager.java

+		this.clientId = clientId;
+	}
+
+	protected void setManualAcks(boolean manualAcks) {


I believe that this one has to be public

artembilan · 2022-08-11T18:51:32Z

Merged as 5f12729 after some minor code clean up and refactoring.

@oxcafedead ,
this was huge; excellent collaboration!
Now, if you wish, you can go ahead and add docs for this new feature. See mqtt.adoc. I believe a new === Shared MQTT Client Support paragraph in the end would be great.
Thank you very much!

artembilan requested changes Jul 21, 2022

View reviewed changes

...egration-mqtt/src/main/java/org/springframework/integration/mqtt/core/MqttClientManager.java Outdated Show resolved Hide resolved

oxcafedead force-pushed the GH-3685 branch from 4ae3a1f to 180eda3 Compare July 26, 2022 13:56

oxcafedead commented Jul 26, 2022

View reviewed changes

...n/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java Outdated Show resolved Hide resolved

artembilan requested changes Jul 26, 2022

View reviewed changes

artembilan requested changes Jul 27, 2022

View reviewed changes

oxcafedead commented Jul 28, 2022

View reviewed changes

...on-mqtt/src/test/java/org/springframework/integration/mqtt/ClientManagerBackToBackTests.java Show resolved Hide resolved

oxcafedead commented Jul 28, 2022

View reviewed changes

...on-mqtt/src/test/java/org/springframework/integration/mqtt/ClientManagerBackToBackTests.java Outdated Show resolved Hide resolved

oxcafedead marked this pull request as ready for review July 28, 2022 14:02

artembilan requested changes Jul 28, 2022

View reviewed changes

oxcafedead added 4 commits July 29, 2022 16:14

spring-projectsGH-3685: Share MQTT connection across components

3251fd2

Fixes spring-projects#3685 Introduce some initial design. Add a new interface `ClientManager` which will manage clients and connections. Use this manager in v3 topic adapter and message handler.

oxcafedead force-pushed the GH-3685 branch from 2b111be to d338577 Compare July 29, 2022 13:14

artembilan requested changes Jul 29, 2022

View reviewed changes

artembilan requested changes Aug 2, 2022

View reviewed changes

artembilan requested changes Aug 4, 2022

View reviewed changes

artembilan requested changes Aug 8, 2022

View reviewed changes

artembilan requested changes Aug 9, 2022

View reviewed changes

artembilan requested changes Aug 10, 2022

View reviewed changes

* Remove client factory as dependency for v3 client manager and use

318fd82

plain connection properties and client persistence instead * Add missed javadocs * Other code style & cleanup improvements

artembilan requested changes Aug 11, 2022

View reviewed changes

artembilan closed this Aug 11, 2022

garyrussell mentioned this pull request Aug 29, 2022

Why getClientInstance is used for inbound and getAsyncClientInstance is used for outbound to get synchronous and asynchronous clients. spring-projects/spring-integration-samples#331

Closed

GH-3685: Share MQTT connection across components #3857

GH-3685: Share MQTT connection across components #3857

Conversation

oxcafedead commented Jul 21, 2022

oxcafedead commented Jul 21, 2022

artembilan left a comment

Choose a reason for hiding this comment

oxcafedead commented Jul 27, 2022 • edited Loading

artembilan left a comment

Choose a reason for hiding this comment

artembilan left a comment

Choose a reason for hiding this comment

artembilan left a comment

Choose a reason for hiding this comment

oxcafedead commented Jul 29, 2022

artembilan left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

oxcafedead Jul 31, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

oxcafedead commented Jul 31, 2022

artembilan left a 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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

oxcafedead commented Aug 4, 2022

oxcafedead commented Aug 4, 2022

artembilan commented Aug 4, 2022

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

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

oxcafedead commented Aug 6, 2022

artembilan commented Aug 6, 2022

oxcafedead commented Aug 6, 2022

artembilan commented Aug 6, 2022

oxcafedead commented Aug 6, 2022

artembilan commented Aug 6, 2022

oxcafedead commented Aug 6, 2022

artembilan left a 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

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

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

artembilan commented Aug 9, 2022

artembilan left a 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

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

oxcafedead commented Jul 27, 2022 •

edited

Loading

oxcafedead Jul 31, 2022 •

edited

Loading