useSubscription hook does not catch all messages

[anzeanzic]

I'm using mqtt-react-hooks with Webpack 5 and React 18 (manually added polyfill dependencies to work) to connect via MQTT over WebSocket to the Aedes broker. On the other side, there's an IoT device that every time publishes 5 test messages on a received command.

I have a problem that the built-in useSubscription hook, which returns the received, doesn't catch all published messages that it should. For comparison, Paho MQTT and the MQTTX app catch all 5 messages, while this library catches only 3 or 4, very rarely all 5. All messages are binary because we're using Google Protocol buffers, but I guess that shouldn't be a problem. Because of that, I'm using the parser method parserMethod={msg => msg} on the Connector.

The source code to listen to the messages is as follows:

const { message } = useSubscription(deviceToCloudTopic);

useEffect(() => {
  if (message) {
	  onMessageArrivedHandler(message);
	  console.log(message);
  }
}, [message]);

Is this a known issue or am I doing something wrong?

[anzeanzic]

I played a bit with the code today as I thought that it will work if change the client from the mqtt.js library to PAHO mqtt library (it worked with a standalone PAHO client).

Everything is working except all messages are also not caught as with mqtt.js. I figured out that the problem probably is in the useSubscription custom hook. It seems that if messages are received within the same millisecond (or two), the message state is overwritten with the new message before the component that is showing messages can detect the change.

[chacalgbi]

I have the same problem. Did you get any solution?

[anzeanzic]

As I mentioned in my previous comment, I rewrote the client part from mqtt.js to PAHO MQTT library as I thought mqtt.js has a problem. It was far more obvious when supporting PAHO MQTT where the real problem is.

What I have done is that instead of only one message I can also return an array of messages. const [message, setMessage] = useState<IMessage | IMessage[] | undefined>(undefined);

if I return one message or an array it depends on what time difference they are received. When I receive the message I store the timestamp I received the message on and if another message comes in within a few milliseconds apart from the first message, I start storing them into an array.

const callback = useCallback((receivedTopic: string, receivedMessage: any) => {
		const receivedTopicArr = receivedTopic.split('/');
		const isStateTopic = receivedTopicArr[receivedTopicArr.length - 1] === 'state';

		if (![topic].flat().some(rTopic => MqttPattern.matches(rTopic, receivedTopic))) return;

		const now = new Date().getTime();
		let newMessage = null;

		setMessage((prevMessage) => {
			console.log(now - previousMessageUxts.current, now, previousMessageUxts.current);

			if (prevMessage !== undefined && previousMessageUxts.current !== 0 && now - previousMessageUxts.current < 2) {
				if (Array.isArray(prevMessage)) {
					newMessage = [
						...prevMessage,
						createANewMessage(receivedTopic, receivedMessage, isStateTopic)
					] as IMessage[];
				}
				else {
					newMessage = [
						prevMessage,
						createANewMessage(receivedTopic, receivedMessage, isStateTopic)
					] as IMessage[];
				}
			}
			else {
				newMessage = createANewMessage(receivedTopic, receivedMessage, isStateTopic);
			}

			return newMessage;
		});

		previousMessageUxts.current = now;
	}, [topic]);

I know this is far from an ideal solution and it could be done way better but I just hadn't had the time to do it. I use this code for a concept project. I was also hopeful Victor would "chime in" a bit.