Fix broken MQTT client by implementing a reconnect "session"

[ctron]

This is based on the content of PR vert-x3/vertx-mqtt#197

Fixes #1181

[ctron]

Yay, that looks better:

2021-05-27 18:46:35,300 INFO  [io.ver.mqt.imp.MqttClientImpl] (vert.x-eventloop-thread-0) Connection with mqtt-integration-drogue-dev.apps.wonderful.iot-playground.org:443 established successfully
2021-05-27 18:46:52,376 INFO  [io.dro.iot.dem.int.Receiver] (vert.x-eventloop-thread-0) Received message: CloudEvent{id='ffd2b536-a978-4409-b13c-71454c47d548', source=drogue://drogue%2Dpublic%2Dtemperature/A4%3AC1%3A38%3AA2%3A6D%3A42, type='io.drogue.event.v1', datacontenttype='application/json', dataschema=urn:drogue:iot:temperature, subject='status', time=2021-05-27T16:46:52.311145045Z, data=BytesCloudEventData{value=[123, 34, 116, 101, 109, 112, 34, 58, 32, 50, 54, 46, 55, 44, 32, 34, 104, 117, 109, 34, 58, 32, 52, 52, 44, 32, 34, 98, 97, 116, 116, 34, 58, 32, 57, 53, 125]}, extensions={instance=drogue, knativehistory=iot-channel-kn-channel.drogue-dev.svc.cluster.local, partitionkey=drogue%2Dpublic%2Dtemperature/A4%3AC1%3A38%3AA2%3A6D%3A42, application=drogue-public-temperature, device=A4:C1:38:A2:6D:42}}
2021-05-27 18:47:02,363 INFO  [io.dro.iot.dem.int.Receiver] (pool-1-thread-1) Received message: CloudEvent{id='da31b316-dcd7-433b-970f-829fffeb8846', source=drogue://drogue%2Dpublic%2Dtemperature/A4%3AC1%3A38%3AA2%3A6D%3A42, type='io.drogue.event.v1', datacontenttype='application/json', dataschema=urn:drogue:iot:temperature, subject='status', time=2021-05-27T16:47:02.333709259Z, data=BytesCloudEventData{value=[123, 34, 116, 101, 109, 112, 34, 58, 32, 50, 54, 46, 55, 44, 32, 34, 104, 117, 109, 34, 58, 32, 52, 52, 44, 32, 34, 98, 97, 116, 116, 34, 58, 32, 57, 53, 125]}, extensions={instance=drogue, knativehistory=iot-channel-kn-channel.drogue-dev.svc.cluster.local, application=drogue-public-temperature, partitionkey=drogue%2Dpublic%2Dtemperature/A4%3AC1%3A38%3AA2%3A6D%3A42, device=A4:C1:38:A2:6D:42}}
2021-05-27 18:47:04,574 INFO  [io.sma.rea.mes.mqt.ses.imp.MqttClientSessionImpl] (vert.x-eventloop-thread-0) Connection closed: Connection closed
2021-05-27 18:47:15,637 ERROR [io.ver.mqt.imp.MqttClientImpl] (vert.x-eventloop-thread-0) Can't connect to mqtt-integration-drogue-dev.apps.wonderful.iot-playground.org:443: javax.net.ssl.SSLHandshakeException: Failed to create SSL connection
	at io.vertx.core.net.impl.ChannelProvider$1.userEventTriggered(ChannelProvider.java:129)
	at io.netty.channel.AbstractChannelHandlerContext.invokeUserEventTriggered(AbstractChannelHandlerContext.java:346)
	at io.netty.channel.AbstractChannelHandlerContext.invokeUserEventTriggered(AbstractChannelHandlerContext.java:332)
	at io.netty.channel.AbstractChannelHandlerContext.fireUserEventTriggered(AbstractChannelHandlerContext.java:324)
	at io.netty.handler.ssl.SslUtils.handleHandshakeFailure(SslUtils.java:441)
	at io.netty.handler.ssl.SslHandler$7.run(SslHandler.java:2033)
	at io.netty.util.concurrent.PromiseTask.runTask(PromiseTask.java:98)
	at io.netty.util.concurrent.ScheduledFutureTask.run(ScheduledFutureTask.java:170)
	at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:164)
	at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:472)
	at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:500)
	at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989)
	at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	at java.base/java.lang.Thread.run(Thread.java:832)
Caused by: io.netty.handler.ssl.SslHandshakeTimeoutException: handshake timed out after 10000ms
	at io.netty.handler.ssl.SslHandler$7.run(SslHandler.java:2029)
	... 9 more

2021-05-27 18:47:16,779 INFO  [io.ver.mqt.imp.MqttClientImpl] (vert.x-eventloop-thread-0) Connection with mqtt-integration-drogue-dev.apps.wonderful.iot-playground.org:443 established successfully
2021-05-27 18:47:22,360 INFO  [io.dro.iot.dem.int.Receiver] (pool-1-thread-1) Received message: CloudEvent{id='00891b1b-8858-4186-80cd-75fcfb8aa3b6', source=drogue://drogue%2Dpublic%2Dtemperature/A4%3AC1%3A38%3AA2%3A6D%3A42, type='io.drogue.event.v1', datacontenttype='application/json', dataschema=urn:drogue:iot:temperature, subject='status', time=2021-05-27T16:47:22.330814606Z, data=BytesCloudEventData{value=[123, 34, 116, 101, 109, 112, 34, 58, 32, 50, 54, 46, 55, 44, 32, 34, 104, 117, 109, 34, 58, 32, 52, 52, 44, 32, 34, 98, 97, 116, 116, 34, 58, 32, 57, 53, 125]}, extensions={instance=drogue, knativehistory=iot-channel-kn-channel.drogue-dev.svc.cluster.local, application=drogue-public-temperature, partitionkey=drogue%2Dpublic%2Dtemperature/A4%3AC1%3A38%3AA2%3A6D%3A42, device=A4:C1:38:A2:6D:42}}

[ctron]

Ok, I have been thinking a bit about the remarks for start, subscribe, unsubscribe.

Initially my thinking was that this will just record it as a "desired state", and let the state machine try to drive it towards that.

However, the ClientHolder actually provides methods to monitor the state changes, and turn that into Futures/Unis. That could also be done directly be the start/subscribe/unsubscribe methods. Moving that logic there, would that be ok, from your point of view?

[cescoffier]

Yes!

[ctron]

Ok, that took a bit:

[INFO] Results:
[INFO] 
[INFO] Tests run: 37, Failures: 0, Errors: 0, Skipped: 0

I made the changes, and tests look good now.

The PR definitely needs to get squashed 😁

[cescoffier]

Can you squash?

[ctron]

Squashed!

[cescoffier]

Thanks, let's see what the CI says and include it.
It will be part of tomorrow's release.

[cescoffier]

Thaks a lot @ctron !