Proposal: Low Level Client vs Managed Client with Subscription Handling

[JaggerJo]

🐛 Bug Report

The client currently exposes a list of Subscriptions.

When subscribing to a new topic, a subscription is added to the list. When unsubscribing the topic is removed.

There are certain issues with the current implementation, and I would argue that subscriptions should not be held in the client at all.

IMHO it would be best to have a low level client that does not try to do anything smart, and then add a ManagedClient that can focus on providing nice user features built on the solid low level client. It would also reduce the API surface of the low level client and simplify thread safety.

@pglombardo what do you think? I'm happy to chat about this is private, feel free to send me an e-mail.

Issues:

• Access to the Subscription list is not synchronised
• Subscriptions are not cleared after a disconnect.

        var client =
            new HiveMQClient(
                new HiveMQClientOptionsBuilder()
                    .WithClientId("ConcurrentSubscribe")
                    .Build()
            );

        var inbox = new ConcurrentBag<string>();

        client.OnMessageReceived += (_, e) =>
        {
            inbox.Add(e.PublishMessage.Topic!);
        };

        var connectResult = await client.ConnectAsync();

        Assert.True(connectResult.ReasonCode == ConnAckReasonCode.Success);
        Assert.True(client.IsConnected());

        var subscribeResult = await client.SubscribeAsync("/test/#", QualityOfService.ExactlyOnceDelivery, false);

        Assert.True(subscribeResult.Subscriptions.Count == 1);
        Assert.True(subscribeResult.Subscriptions[0].SubscribeReasonCode == SubAckReasonCode.GrantedQoS2);

        Assert.Equal(1, client.Subscriptions.Count(s => s.TopicFilter.Topic == "/test/#"));

        await client.PublishAsync("/test/1", "whatever");

        while (!inbox.Contains("/test/1"))
        {
            await Task.Delay(50);
        }

        await client.DisconnectAsync();

        await Task.Delay(1000);

        var reconnectResult = await client.ConnectAsync();

        Assert.True(reconnectResult.ReasonCode == ConnAckReasonCode.Success);
        Assert.True(client.IsConnected());

        await client.PublishAsync("/test/2", "whatever");

        // client still contains subscription.
        Assert.True(client.Subscriptions.Any(s => s.TopicFilter.Topic == "/test/#"));

        // we never receive the message, because the subscription is not re-established
        while (!inbox.Contains("/test/2"))
        {
            await Task.Delay(50);
        }

• Subscribing multiple times results in multiple subscriptions (for the same topic) in the subscription list. This also included the handler.

        var subscribeResultA = await client.SubscribeAsync("/test/#", QualityOfService.ExactlyOnceDelivery, false);

        Assert.True(subscribeResultA.Subscriptions.Count == 1);
        Assert.True(subscribeResultA.Subscriptions[0].SubscribeReasonCode == SubAckReasonCode.GrantedQoS2);

        var subscribeResultB = await client.SubscribeAsync("/test/#", QualityOfService.ExactlyOnceDelivery, false);

        Assert.True(subscribeResultB.Subscriptions.Count == 1);
        Assert.True(subscribeResultB.Subscriptions[0].SubscribeReasonCode == SubAckReasonCode.GrantedQoS2);

        Assert.Equal(2, client.Subscriptions.Count(s => s.TopicFilter.Topic == "/test/#"));

• Subscription level message handlers burn up CPU time, even when not used. All Subscriptions are iterated and checked for a topic match using MatchTopic, which is quite expensive.

hivemq-mqtt-client-dotnet/Source/HiveMQtt/Client/HiveMQClientEvents.cs

Lines 245 to 265 in b3700c3

	foreach (var subscription in this.Subscriptions)
	{
	if (packet.Message.Topic != null && MatchTopic(subscription.TopicFilter.Topic, packet.Message.Topic))
	{
	if (subscription.MessageReceivedHandler != null && subscription.MessageReceivedHandler.GetInvocationList().Length > 0)
	{
	// We have a per-subscription message handler.
	_ = Task.Run(() => subscription.MessageReceivedHandler?.Invoke(this, eventArgs)).ContinueWith(
	t =>
	{
	if (t.IsFaulted)
	{
	Logger.Error($"per-subscription MessageReceivedEventLauncher faulted ({packet.Message.Topic}): " + t.Exception?.Message);
	}
	},
	TaskScheduler.Default);
	messageHandled = true;
	}
	}
	}

🔬 How To Reproduce

Archive.zip

Environment

Current master or latest Nuget

[SaschaBa]

Hi,

We are also affected by this issue and it's currently a blocker for us. We are prepared to invest the time to develop a fix and submit a pull request.

Before we do, it would be very helpful for our planning if you could give us an indication: If we provide a quality PR, is there a realistic chance it could be reviewed and merged in a timely manner?

Our alternative would be to maintain a private fork, but we would strongly prefer to contribute our fix back to the community. We'd appreciate a quick heads-up so we can decide on our next steps.

Thanks for your great work on this project.

[pglombardo]

Hi @SaschaBa - absolutely. A good PR is always welcome.

I would request only to keep it focused on this exact issue (other separate PRs are welcome too).

And although the idea of a separate managed client makes sense eventually, for now we just want to make the existing subscription list thread safe with some sort of synchronisation/locking for access.

If we're on the same page, I'd be more than happy to merge the PR. It would be very much appreciated.

[pglombardo]

The thread issues mentioned in this issue should now be fixed in 0.27.0. I'll leave this issue open as it's a fair request/discussion on whether and where a list of subscriptions should be maintained by the client.

For example, if we wanted to aim for hyper performance, removing the subscription list from a low-level client would be a performance boost although the work would just be shifted to the client consumer. Still desirable in some cases.