emqttc bug

[emqplus]

Hope you are doing well. We are using emqttc in our application and are facing errors in receive block. Some of the messages are completely jumbled and dirty payload buffer. Also the message with error keeps repeating (emqttc keeps publishing the same message).

We are using it in Elixir Genserver. The relevant blocks are as follows

Genserver handle_info to receive publish events

def handle_info({:publish, channel,message}, %{client: client, chan: chan}) do
spawn fn -> mqtt_receive(channel,message,chan) end
{:noreply, %{client: client, chan: chan}}
end

defp mqtt_receive(channel,payload,chan) do
[_|values] = String.split(channel,"/")
[service|[iris_id|[request|[]]]] = values
try do
case JSON.decode(payload) do
{:ok,message} ->
payload = %{message: message}
case JSON.encode(payload) do
{:ok,jsonpayload} ->
####
:ok
{:error,msg} ->
IO.puts "Encode Error: #{msg}"
end
{:error,err} ->
IO.puts "Error: #{err}"
end
rescue
exception ->
IO.puts "Error while decoding json from mosca"
:ok
end
end

[witeman]

we met the similar issue in our project as well. When we change to erlmqtt, it is doing well. If I have some time, I would help to investigate this issue. It may be relevant to the parse or serialize part of your code.

[emqplus]

@witeman, duplicated message? could you reoccure the issue? Could send me your code by email: [email protected]

[witeman]

You can send any payload(publish) of size more than 2.5k, then the second coming payload would become different between publisher's sending and subscriber's receiving.

[emqplus]

okay, I will try it now

[emqplus]

@witeman I use emqttd_benchmak to send/receive:

Send 14336 bytes
Received 14336 bytes
Received 14336 bytes
Send 14336 bytes
Received 14336 bytes

It works well

[witeman]

the size is the same but content differs.

I guess that problem lies at emqttc_parser:parse

BR
Witeman

在 2015年5月26日，下午11:00，Feng Lee [email protected] 写道：

@witeman I use emqttd_benchmak to send/receive:

Send 14336 bytes
Received 14336 bytes
Received 14336 bytes
Send 14336 bytes
Received 14336 bytes
It works well

—
Reply to this email directly or view it on GitHub.

[witeman]

Reproduce steps:

1. open two emqttc client(clientA, clientB), connect to the rabbitmq-mqtt plugin or emqttd.
2. clientA: emqttc:subscribe(C, <<"TopicA">>, qos0)
3. clientB: emqttc:publish(C, <<"TopicA">>, term_to_binary(lists:seq(1,2000)))
4. clientA: receive normally
5. clientB: emqttc:publish(C, <<"TopicA">>, <<"Payload...">>)
6. clientA: receive the result of OTP 18.0 compatible #3 again (problematic!)
7. clientB: emqttc:publish(C, <<"TopicA">>, <<"Payload...">>)
8. clientA: receive nothing.

So I guess the problem is caused by the emqttc_parser:parse/2.

[emqplus]

@witeman what's the fuck stupid bug! Forgot to reinit the parser after a frame is parsed successfully!

Thx for your help, I fix this issue now.

src/emqttc_socket.erl:

 parse_received_bytes(ClientPid, Data, ParseState) ->
     case emqttc_parser:parse(Data, ParseState) of
-    {more, ParseState1} ->
-        ParseState1;
+    {more, ParseState1} ->
+        {ok, ParseState1};
     {ok, Packet, Rest} ->
         gen_fsm:send_event(ClientPid, Packet),
-        parse_received_bytes(ClientPid, Rest, ParseState)
+        parse_received_bytes(ClientPid, Rest, emqttc_parser:new());