Disconnect on Publish when using WebSocket

[stefxx]

Describe the bug

After successfully connecting to AWS using WebSockets, and being able to receive messages, the connection is immediatly disconnected on a PublishAsync except when sending a empty payload.

Which component is your bug related to?

• Client

To Reproduce

Connect to AWS using WebSockets
Publish anything (except empty payload)
This is the code I use to Publish:

    Function MqttPublish(Topic As String, ByVal Payload As String) As Task(Of MqttClientPublishResult)
        Return mqttClient.PublishAsync(New MqttApplicationMessage With {
        .Topic = Topic,
        .Payload = Encoding.Unicode.GetBytes(Payload)
    })
    End Function

Note that the MqttClientDisconnectedEventArgs.Exception is nothing. Also, the PublishAsync.Result is "Success" and does not trigger an exception.

Expected behavior

That the Payload is being send without disconnecting from the broker

Thanks!

[stefxx]

Maybe this is related, it sounds like a similar issue (but should be resolved in the latest build that I am using, 4.1.4.563):

#1499

[stefxx]

Some trace logging, maybe it helps:

>> [12-3-2023 17:01:35] [15] [MqttClient] [Verbose]: Trying to connect with server 'iot.eu-west-1.xx-redacted-xx.com'.
>> [12-3-2023 17:01:35] [11] [MqttClient] [Verbose]: Connection with server established.
>> [12-3-2023 17:01:35] [13] [MqttClient] [Verbose]: Start receiving packets.
>> [12-3-2023 17:01:35] [11] [MqttChannelAdapter] [Verbose]: TX (69 bytes) >>> Connect: [ClientId=xx-redacted-xx] [Username=] [Password=] [KeepAlivePeriod=15] [CleanSession=True]
>> [12-3-2023 17:01:35] [11] [MqttChannelAdapter] [Verbose]: RX (4 bytes) <<< ConnAck: [ReturnCode=ConnectionAccepted] [ReasonCode=Success] [IsSessionPresent=False]
>> [12-3-2023 17:01:35] [10] [MqttClient] [Verbose]: Authenticated MQTT connection with server established.
>> [12-3-2023 17:01:35] [10] [MqttClient] [Info]: Connected.
>> [12-3-2023 17:01:35] [9] [MqttClient] [Verbose]: Start sending keep alive packets.
>> [12-3-2023 17:01:36] [15] [MqttChannelAdapter] [Verbose]: TX (36 bytes) >>> Subscribe: [PacketIdentifier=1] [TopicFilters=xx-redacted-xx/xx-redacted-xx/commandOut@AtMostOnce]
>> [12-3-2023 17:01:36] [15] [MqttChannelAdapter] [Verbose]: TX (34 bytes) >>> Publish: [Topic=xx-redacted-xx/xx-redacted-xx/commandIn] [Payload.Length=2] [QoSLevel=AtMostOnce] [Dup=False] [Retain=False] [PacketIdentifier=0]
>> [12-3-2023 17:01:36] [11] [MqttClient] [Verbose]: Stopped sending keep alive packets.
>> [12-3-2023 17:01:36] [11] [MqttClient] [Verbose]: Disconnecting [Timeout=00:01:40]
>> [12-3-2023 17:01:36] [11] [MqttClient] [Verbose]: Disconnected from adapter.
>> [12-3-2023 17:01:36] [11] [MqttClient] [Info]: Disconnected.
>> [12-3-2023 17:01:36] [11] [MqttClient] [Verbose]: Stopped receiving packets.

[stefxx]

I have tested a Publish to test.mosquitto.org:8080, also using WebSockets and it works just fine. So there is something different when connecting to my AWS endpoint.

I also did a test using MQTTnet 3.1.2 and my AWS endpoint which also works fine. I tested a few MQTTnet 4.x versions and they all fail.

What could be different since 4.x when Publishing to AWS?

[logicaloud]

That's odd. If it works with 3.x then I would expect it to work with 4.x also.

The trace shows a "Subscribe" just before a "Publish"; does it disconnect on "Publish" without any "Subscribe" also?

BTW, have you tried #1691 with version 3.x also?

[stefxx]

The trace shows a "Subscribe" just before a "Publish"; does it disconnect on "Publish" without any "Subscribe" also?

Without the subscribe, the connection still closes. I also tried adding a few random delays (5 to 30 seconds) to make sure it is the publish that causes this. The only publish that doesn't cause the connection to close is a publish with an empty payload.

BTW, have you tried #1691 with version 3.x also?

Yes, same result. So at least that is consistent (and expected since the root cause is a .Net Framework bug).

[logicaloud]

I'll try to have a look during the week. Maybe this is related to #1689.

[logicaloud]

It looks like there is some incompatibility between the AWS WebSocket server and the .NET ClientWebSocket that is used in MQTTnet. I'm not sure where this incompatibility lies but I could publish continuously without disconnection by using the WebSocket4Net extension:

            var mqttFactory = new MqttFactory();
            mqttFactory.UseWebSocket4Net();

Does that work for you?

[stefxx]

Yes, that does work indeed. Sounds like a reasonable workaround!

[logicaloud]

The problem seems to be that the AWS WebSocket does not handle message fragments. I think compatibility here can be supported in MQTTnet by providing a WebSocket option .WithUnfragmentedFrames. Maybe leave this issue open for a now, I may add a pull request for this next week.

[stefxx]

Thanks @logicaloud , appreciate it. I'll leave it open for now.

[logicaloud]

Here is the implementation of a client web socket option to use unfragmented messages:

https://github.com/logicaloud/MQTTnet/tree/websocket-unfragmented-messages-option

To use this option, you would need to set the option to true before creating the WebSocket client:

webSocketOptions.PreferUnfragmentedMessages = true;

The effect is that messages are preassembled into a single buffer before hand-over to the underlying WebSocket.

@stefxx, does that work for you?

@chkr1011, I have considered implementing a fragment buffer in the MqttWebSocketChannel, but that would make the object stateful and may lead to problems accessing it from multiple threads. Instead, the IMqttChannel interface has the PreferUnfragmentedMessages flag added, which is then evaluated in the MqttChannelAdapter. Although a few files are touched because of the modification of the interface, effective changes are restricted to the MqttChannelAdapter and the web socket channels. Other interface implementers simply return the default false value for PreferUnfragmentedMessages. - Is it worthwhile creating a pull request from this?

[stefxx]

Yes, excellent! This does fix the problem. Thank you very much!

[chkr1011]

@logicaloud Thanks for pointing to the right direction with solving this issue with AWS.

I added a branch to this PR and implemented a new option for the client. It will only change the internal memory management so that it will also apply for TCP connections (if ever needed there) and is not stateful.

Please check the code and let me know what you think.

[logicaloud]

Looks good to me (without paying too much attention to other refactoring). Having the client adapter factory handle the new option is better than extending the channel interface, and I don't see any harm in moving the AvoidPacketFragmentation to the general MqttClientOptions.

One suggestion would be to avoid Amazon specific naming in the MqttClientOptionsBuilder API (WithAmazonWebServicesCompatibility) because the new option may well be used for other WebSocket servers. If there are other options required for Amazon compatibility, then these can also be added separately for greater flexibility. Amazon specific examples can then simply configure those options into a working client.

[chkr1011]

@logicaloud That is a good hint. I will update accordingly and remove the Amazon naming stuff.

Could you please test the new commit as well? I have no access to an AWS broker...

[logicaloud]

There seems to be a bug in the WithoutPacketFragmentation method (assigns true instead of false). After changing it to false it works with AWS:

        public MqttClientOptionsBuilder WithoutPacketFragmentation()
        {
            _options.AllowPacketFragmentation = false;
            return this;
        }

[chkr1011]

Oh that was a stupid one 😄

[stefxx]

Thanks again. I've just tested .WithoutPacketFragmentation with the 4.2 release and works perfectly!