Alternative for ClientMessageQueueInterceptor in V4.x.x?

[RossHNPC]

How to replicate functionality when the ClientMessageQueueInterceptor has been removed.

We're working on upgrading an existing codebase that currently uses v2.x.x release of MqttNet.

Mostly going well, but I'm unable to determine a replacement for the following code snippet.

// This is related to this Server sending messages to the Clients
// Needs further testing as the library possibly has newer methods/patterns that we can use
_mqttServer.Options.ClientMessageQueueInterceptor = c =>
{
   // Messages from this server to the clients have a blank SendClientId.
   if (string.IsNullOrWhiteSpace(c.SenderClientId))
   {
       var topic = c.ApplicationMessage.Topic;
       if (_specificClients.ContainsKey(topic))
       {
            c.AcceptEnqueue = false;
            var allowedClients = _specificClients[topic];
            c.AcceptEnqueue = allowedClients.Contains(c.ReceiverClientId);
       }
   }
};

Investigating this; the messages that we send from the server should go to specific clients; we determine this be examining the messages for a blank ClientId then comparing the RecieverClientId to a screened list.

Searching the current code base, SenderClientId, ReceiverClientsId and AcceptEnqueue all appear to be historical.

Is there a similar to intercept the ApplicationMessages where the Sender and Recevier ClientIds are available as they are being published in the current version of MqttNet?

Which project is your question related to?

• Server

[Jeanot-Zubler]

Intercepting publish

You can see how to intercept publish events here.

For checking whether the message was sent by the server you use args.ClientId and compare that with the ClientId you defined when calling server.InjectApplicationMessage (the new way to publish messages from the server).

To get the topic you can still use args.ApplicationMessage.Topic.

To stop the message from being published, you can use args.ProcessPublish = false. However i don't think you can drop messages by Client here.

Intercepting packets

Use server.InterceptingOutboundPacketAsync to intercept all outbound Packets, syntax is the same as above. Use something like the following to get the topic, and Id of the client:

if (args.Packet is MqttPublishPacket publish)
{
    string topic = publish.Topic;
    string client = args.ClientId;
    ...
    args.ProcessPacket = true / false;
}

However, if you really need to check, whether the sender is the Server, I think you are out of luck with this method.

[chkr1011]

I added a similar event for version 4.0. Please either check the code from the branch attached to this ticket or use the build 651 from the myget feed (see link in readme). If that new event works for you I will merge this feature.

[nivek90]

I need a similar feature for some projects, I checked the code for new InterceptingClientEnqueueEvent and it looks good.

[Crichen]

@chkr1011 Yes that would be perfect, as long as it doesn't go against existing plans for the Mqtt library. Sorry this is an alternate login to @RossHNPC

[RossHNPC]

@chkr1011 I'll check out the branch this morning and run it with our code.

[RossHNPC]

@chkr1011 currently I'm using this code snippet adapted from @Jeanot-Zubler suggestion above which mostly works.

_mqttServer.InterceptingOutboundPacketAsync += o =>
{
  if (o.Packet is not MqttPublishPacket packet) return Task.CompletedTask;
  
  var client = o.ClientId;    // receiving clientId?
  var topic = packet.Topic;

  // restricted topic?
  if (_restrictedTopics.ContainsKey(topic))
  {
    o.ProcessPacket = false;
    
    var allowedClients = _restrictedTopics[topic];
    o.ProcessPacket = allowedClients.Contains(client);
  }

  return Task.CompletedTask;
};

Unfortunately the clients are registered with a "request/+" which is leading a duplication of requests amongst the listening clients, which I'll need to investigate further.

If the former functionality through the ClientMessageQueueInterceptor could be restored it would be much appreciated.

[chkr1011]

@RossHNPC Please let me know if the code from the branch works then I will merge the feature.

[RossHNPC]

@chkr1011 yes code working and that's restored the old functionality. Thank you for adding this back in.

[RossHNPC]

@chkr1011 Do you have an ETA for the next official Release to NuGet? We can use a build a of the 1648 branch for testing, but would rather be using an official NuGet package in production.

[chkr1011]

I assume it will be released within the next two weeks.

[RossHNPC]

@chkr1011 I've discovered a problem with this. If we have 3 Clients subscribed to 'request/+' as a topic. When the MqttClientSessionsManager is iteratting through the subscriberSessions ~line 184 it works perfectly, until you reach the InterceptingClientEnqueueEvent.

If we are processing subscriptions for Clients A,B & C: if A is processed by the EnqueueEventHandler and returns with AcceptEnqueue = false then the loop exits without Client B&C being checked. Similarly if A has AcceptEnqueue = true but B has AcceptEnqueue = false then Client C will not be tested.

I'll try to make a fix for this; but not fully knowing the code base I can't promise it won't break anything :-(

[RossHNPC]

@chkr1011 fairly certain it should just be a continue statement and that would allow the code to progress through the subscriptions without exiting early.