Mqtt toolbox in Matlab unable to process all mqtt messages

Hi all,

I have here an application which relies on mqtt traffic being read into matlab. Unfortunately, it seems that the mqtt toolbox is unable to process all messages in a topic: when i suscribe to a certain topic from within mosquitto_sub, I collect +- 40000messages in a 15s timeframe, whilst in MAtlab I'm only able to gather like 4000 messages.

I'm sure the broker can handle this amount of messages, but somehow, 90% of the packets are not detected in the variable mqttData15s.

Has anyone a solution for this inconveniece ,(R2018b)

Regards

Jeroen

try
    try
        mysub=subscribe(myMQTT,'node/#','QoS',2);
    catch
    end
    disp('Gathering live data 16s....');
    pause(15);
    mqttData15s=mysub.readall;
    mysub.unsubscribe;
        myMQTT.disconnect;
        clear myMQTT
        clear mysub
    disp('Done.');
catch
    disp('MQTT read error');
    mysub.unsubscribe;
    myMQTT.disconnect;
    clear myMQTT
    clear mysub
end

Answer 1

[Johan]

Sorry that I dont have a solution for you but just wanted to add to this topic.

I also have a situation where the broker can send messages at high-throughput (hundreds of hz). While the broker has no issue of dealing with this (other clients get the data just fine) there seems to be an issue with the matlab callback system that adds a ~15 ms delay for the callback to even intiate.

You can see that for instance in the simple example code below (also tested this internally with more precise measures)

Interestingly the toolbox can send messages at highspeeds (sub-millisecond) but the callback is relatively slow. Would really like to see a solution for this as it really diminsihes the useability of this toolbox for device communication.

function testMqtt()
    myMQTT = mqtt('tcp://127.0.0.1');
    mySub = subscribe(myMQTT,'myTopic');
    mySub.Callback = @msgReceived;
        for (i=1:10)
            tic
            publish(myMQTT, 'myTopic', 'Hello World!');
            pause(1);
        end
end

function msgReceived(topic, msg)
    toc
end

Answer 2

[Sylvain]

I suspected the Quality Of Service set to 2, but it is not influencing the number of readings

Answer 3

[Sylvain]

I also confirm that the function readall is not performing as expected, causing a downsampling.

I am generating a sinus wave 2Hz and publish the signal. Using the timestamp of the broker is good enough.

clear all;close all;clc

format long

topic = 'TOPIC/'
MQTTBrokerIPAdress = 'tcp://XXX.XXX.XXX.XXX'
myMQTT=mqtt(MQTTBrokerIPAdress) %raspberry pi address

%% Animated Lines
hfig = figure;clf
hax = axes(hfig);
anim_lineX = animatedline('Parent',hax,'Color','blue','MaximumNumPoints',500);

DisplayXlim = 5; %seonds
%% Get the first timestamp to set the t0
    SubX = subscribe(myMQTT,topic,'QoS',0);    
    NmessagesToRead = SubX.MessageCount;
    while  NmessagesToRead < 1
        pause(0.001)
        NmessagesToRead = SubX.MessageCount;
    end
    SubX.unsubscribe
 [~,message_time]=SubX.read; %get the time
  t0 = datenum(message_time);
 
 
 
%% Start the reading
 for ii=1:10 
    SubX = subscribe(myMQTT,topic,'QoS',0);    
    NmessagesToRead = SubX.MessageCount;
    while  NmessagesToRead < 500 %
        pause(0.001)
        NmessagesToRead = SubX.MessageCount;
    end
    
    %% Extract the data
    SubX.MessageCount
    messageX = SubX.readall;
    SubX.unsubscribe
    x = str2double(messageX.Data);
    tx = datenum(messageX.Time);
    time = (tx-t0)*1e5;
    
        %% Manage Axis
    if time(end) <= DisplayXlim
        hax.XLim = [0 DisplayXlim];
    else
        hax.XLim = [(time(end) - DisplayXlim)  time(end)];
    end
    
    %2D plot
    addpoints(anim_lineX,time,x) 
    drawnow limitrate
 end  
 

In comparison, here is what you get if you loop the Sub.read(). Still some distortion though.