Any recommendations on a MQTT client for Elixir?

Sebb · July 15, 2021, 3:51pm

I can’t get SSL to work.

I’ve looked at the opts emqtt-cli generates for

./emqtt sub --enable-ssl --CAfile mosquitto.org.crt -h test.mosquitto.org -p 8883 -t "#"

which is

opts = [
  host: 'test.mosquitto.org',
  ssl: true,
  ssl_opts: [cacertfile: "mosquitto.org.crt"]
]

but I just get a crash:

iex> {:ok, pid} = :emqtt.start_link(opts)
{:ok, #PID<0.221.0>}
iex>  :emqtt.connect(pid)
{:error, :closed}
** (EXIT from #PID<0.214.0>) shell process exited with reason: :closed

adding server_name_indication and verify_peer yields the same result.

I can reproduce this in docker elixir:1.11.4 and elixir:1.12.2 for emqtt 1.2 and 1.4.
Just mix new test_proj, then add to deps {:emqtt, github: "emqx/emqtt", tag: "v1.2.0"} and run:

sudo docker run --volume $(pwd):/emqtt -w /emqtt -it elixir:1.11.4 bash

$> iex -S mix

brianmay · July 15, 2021, 9:50pm

I think you might need to set port: 8883 somewhere.

I also set proto_ver: :v5, client_id: <string>, and clean_start: false, although I would like to think these values are optional (not tested though).

According to the docs proto_ver defaults to :v4, not :v5 which might be desirable.

Sebb · July 15, 2021, 10:04pm

that’s default for SSL, see: emqtt_cli.erl#L337.
I’m using the exact same opts the cli uses.

You got SSL running right?
Can you check if it works in docker?

brianmay · July 15, 2021, 10:08pm

Your theory makes sense. But if I don’t specify port: 8883 I get symptoms that look identical to what you observed. Which is somewhat weird and confusing.

… Oh wait, that default is only applicable for the CLI I think.

Sebb · July 15, 2021, 10:11pm

OK thats super stupid.

Its working! Who would have thought, you just have to do it right.

def test_8883 do
  opts = [
    host: 'test.mosquitto.org',
    port: 8883,
    ssl: true,
    ssl_opts: [cacertfile: "mosquitto.org.crt"]
  ]

  {:ok, pid} = :emqtt.start_link(opts)
  {:ok, _} = :emqtt.connect(pid)
  {:ok, _, _} = :emqtt.subscribe(pid, %{}, [{"test/#", []}])
end

And here is code for “8884 : MQTT, encrypted, client certificate required”.
You need to Generate a TLS client certificate for test.mosquitto.org first

def test_8884 do
  opts = [
    host: 'test.mosquitto.org',
    port: 8884,
    ssl: true,
    ssl_opts: [
      cacertfile: "mosquitto.org.crt",
      certfile: "client.crt",
      keyfile: "client.key"
    ]
  ]

  {:ok, pid} = :emqtt.start_link(opts)
  {:ok, _} = :emqtt.connect(pid)
  {:ok, _, _} = :emqtt.subscribe(pid, %{}, [{"test/#", []}])
end

brianmay · July 15, 2021, 11:19pm

OK, I think I understand why I need to override the server_name_indication value. And understand a bit more about the erlang debugger.

Normally a program would use ssl:connect(Host, Port, Options, Timeout) which would create the TCP connection to the host. But emqtt doesn’t do that, it creates the TCP connection and then calls ssl:connect(Sock, SslOpts2, Timeout) on the existing sock connection. As a result the ssl module has no way of knowing what the original hostname was for the original connection.

github.com

emqx/emqtt/blob/master/src/emqtt_sock.erl#L53

    
      
          -spec(connect(inet:ip_address() | inet:hostname(),
                        inet:port_number(), [option()], timeout())
                -> {ok, socket()} | {error, term()}).
          connect(Host, Port, SockOpts, Timeout) ->
              TcpOpts = merge_opts(?DEFAULT_TCP_OPTIONS,
                                   lists:keydelete(ssl_opts, 1, SockOpts)),
              case gen_tcp:connect(Host, Port, TcpOpts, Timeout) of
                  {ok, Sock} ->
                      case lists:keyfind(ssl_opts, 1, SockOpts) of
                          {ssl_opts, SslOpts} ->
                              ssl_upgrade(Sock, SslOpts, Timeout);
                          false -> {ok, Sock}
                      end;
                  {error, Reason} ->
                      {error, Reason}
              end.
          
          
ssl_upgrade(Sock, SslOpts, Timeout) ->
              TlsVersions = proplists:get_value(versions, SslOpts, []),
              Ciphers = proplists:get_value(ciphers, SslOpts, default_ciphers(TlsVersions)),
              SslOpts2 = merge_opts(SslOpts, [{ciphers, Ciphers}]),

github.com

emqx/emqtt/blob/master/src/emqtt_sock.erl#L64

    
      
                          false -> {ok, Sock}
                      end;
                  {error, Reason} ->
                      {error, Reason}
              end.
          
          
ssl_upgrade(Sock, SslOpts, Timeout) ->
              TlsVersions = proplists:get_value(versions, SslOpts, []),
              Ciphers = proplists:get_value(ciphers, SslOpts, default_ciphers(TlsVersions)),
              SslOpts2 = merge_opts(SslOpts, [{ciphers, Ciphers}]),
              case ssl:connect(Sock, SslOpts2, Timeout) of
                  {ok, SslSock} ->
                      ok = ssl:controlling_process(SslSock, self()),
                      {ok, #ssl_socket{tcp = Sock, ssl = SslSock}};
                  {error, Reason} -> {error, Reason}
              end.
          
          
-spec(send(socket(), iodata()) -> ok | {error, einval | closed}).
          send(Sock, Data) when is_port(Sock) ->
              try erlang:port_command(Sock, Data) of
                  true -> ok

brianmay · July 15, 2021, 11:48pm

This behaviour is in the documentation for ssl:connect/2 in fact. Erlang -- ssl

If the option verify is set to verify_peer the option server_name_indication shall also be specified, if it is not no Server Name Indication extension will be sent, and public_key:pkix_verify_hostname/2 will be called with the IP-address of the connection as ReferenceID, which is probably not what you want.

I am starting to think however that emqtt should really supply that value itself, if it wasn’t already supplied.

brianmay · July 16, 2021, 12:17am

Sometimes things in emqtt can be inconsistent. For example, the type spec for the handlers says:

disconnected := fun(({reason_code(), _Properties :: term()}) → any()) | mfas()

Implies the disconnected handler is called as disconnected({reason, properties}). But sometimes it is called as disconnected(error) without the tuple and a completely different type.

github.com

emqx/emqtt/blob/master/src/emqtt.erl#L1052

    
      
                   [StateName, EventContent], State),
              keep_state_and_data;
          
          
handle_event(EventType, EventContent, StateName, State) ->
              ?LOG(error, "State: ~s, Unexpected Event: (~p, ~p)",
                   [StateName, EventType, EventContent], State),
              keep_state_and_data.
          
          
%% Mandatory callback functions
          terminate(Reason, _StateName, State = #state{conn_mod = ConnMod, socket = Socket}) ->
              case Reason of
                  {disconnected, ReasonCode, Properties} ->
                      %% backward compatible
                      ok = eval_msg_handler(State, disconnected, {ReasonCode, Properties});
                  _ ->
                      ok = eval_msg_handler(State, disconnected, Reason)
              end,
              case Socket =:= undefined of
                  true -> ok;
                  _ -> ConnMod:close(Socket)
              end.

I have seen it being called both ways in my test code.

brianmay · July 16, 2021, 1:04am

Not absolutely sure about that now. I am getting myself confused with unrelated stuff.

brianmay · July 16, 2021, 1:29am

Seems like the definition of return_code() isn’t correct. Should be an integer according to the definition. But I get values like:

:shutdown - on certificate validation error.
{:badmatch, :ok} - elixir exception generated in linked process.

Not sure if this will always be the case or not however.

Sebb · July 16, 2021, 8:22am

I can connect to all the ports mosquittos testserver offers. IPV6 and SSL do not work together …!?
I’ve omitted websockets because emqtt does not support TLS for websockets.
I’ve tested with version 1.2.0 because I suddenly can’t build 1.4.0 anymore, there is sth wrong with the quicer submodule.

defmodule MqttHowtoTest do
  use ExUnit.Case

  @hostname 'test.mosquitto.org'
  @host [host: @hostname]
  @credentials [username: "rw", password: "readwrite"]
  @ssl_opts [
    cacertfile: "mosquitto.org.crt",
    server_name_indication: @hostname,
    verify: :verify_peer
  ]
  @client_cert [certfile: "client.crt", keyfile: "client.key"]
  @ipv6 [tcp_opts: [:inet6]]

  @opts_1883 @host ++ [port: 1883]
  @opts_1884 @host ++ @credentials ++ [port: 1884]
  @opts_8883 @host ++ [ssl: true, ssl_opts: @ssl_opts, port: 8883]
  @opts_8884 @host ++ [ssl: true, ssl_opts: @ssl_opts ++ @client_cert, port: 8884]
  @opts_8885 @host ++ @credentials ++ [ssl: true, ssl_opts: @ssl_opts, port: 8885]

  def connect(opts) do
    {:ok, pid} = :emqtt.start_link(opts)
    :emqtt.connect(pid)
  end

  test "1883" do
    assert {:ok, _} = connect(@opts_1883)
  end

  test "1884" do
    assert {:ok, _} = connect(@opts_1884)
  end

  test "8883" do
    assert {:ok, _} = connect(@opts_8883)
  end

  test "8884" do
    assert {:ok, _} = connect(@opts_8884)
  end

  test "8885" do
    assert {:ok, _} = connect(@opts_8885)
  end

  test "1883 ipv6" do
    assert {:ok, _} = connect(@opts_1883 ++ @ipv6)
  end

  test "1884 ipv6" do
    assert {:ok, _} = connect(@opts_1884 ++ @ipv6)
  end

  # these will not work: ** (EXIT from #PID<0.212.0>) :econnrefused
  
  # test "8883 ipv6" do
  #   assert {:ok, _} = connect(@opts_8883 ++ @ipv6)
  # end

  # test "8884 ipv6" do
  #   assert {:ok, _} = connect(@opts_8884 ++ @ipv6)
  # end

  # test "8885 ipv6" do
  #   assert {:ok, _} = connect(@opts_8885 ++ @ipv6)
  # end
end

Sebb · July 16, 2021, 8:24am

yes thats odd, I tried to understand, but I’m not risking any more brain damage looking at too much Erlang code (you saw what happended when I felt clever looking at the cli code).

Over the weekend sometime I’ll write some working sample code. I need:

connect (encrypted, username+password)
static list of subscriptions
publish retained
automatic reconnect (which is not handled by emqtt)

brianmay · July 16, 2021, 10:31am

Seems to work here for me. I think. Assuming it is still using IPv6, which I think it is.

1.4.0 builds fine for me. But quicer won’t build with a recent version of cmake. See the bug report above.

Also (as per diff I posted above) I believe for 1.4 if you set the BUILD_WITHOUT_QUIC environment variable, quic will be completely disabled. No tested. Which might be a good thing, it tends to slow down compilation, and I don’t have any use for it.

At one stage I got warnings that emqtt wasn’t listed as a dependency, which was weird. I deleted the deps and _build directories and downloaded again, and this warning disappeared.

brianmay · July 17, 2021, 4:53am

Publishing an empty string "" to a topic seems to result in my mosquitto server complaining about the topic being invalid and disconnecting the client.

1626496705: Client robotica-canidae sent topic with invalid UTF-8, disconnecting.

Despite the error mentioning the topic, it appears to be the data which triggers this.

I believe sending an empty string should be fully supported (it is the way to delete a retained message), not sure what is going on here.

This only happens with the emqtt client.

From a different version of the server I get:

1626498034: Client test_client disconnected due to malformed packet.

brianmay · July 17, 2021, 6:27am

Never mind, somehow I got my topic and data parameters reversed

Makes sense now, an empty topic string is not allowed.

Hanspagh · July 21, 2021, 1:04pm

I have been using tortoise in production for a while now, without any major issues. The latest release fixed some of the problems we have had. Our use case is very simple, so maybe that is why we haven’t hit any of the problems you described

brianmay · September 2, 2021, 9:49pm

This is the one problem I have with emqtt - no up-to-date hex packages:

Which means in turn I can’t upload anything that depends on it to hex either

alexb · December 6, 2021, 2:45pm

Thank you very much for making the emqtt wrapper. I would like to use it for an elixir / phoenix iot project subscribing to a vernemq broker from an elixir backend. Unfortunately i can’t get it running. I always get the following error message:

** (Mix) Could not compile dependency :quicer, “/home/abe/.mix/rebar3 bare compile --paths /home/abe/Code_Linux/mde/mqtt_test/_build/dev/lib/*/ebin” command failed. You can recompile this dependency with “mix deps.compile quicer”, update it with “mix deps.update quicer” or clean it with “mix deps.clean quicer”

Any suggestions would be highly appreciated.

brianmay · December 6, 2021, 9:40pm

quicer should build… But you probably don’t need it. Try setting the BUILD_WITHOUT_QUIC=true environment variable. Requires a recent version of emqtt, I suspect the latest version on hex doesn’t have this. Personally I wish this was the default, I generally don’t need quic support.

I have found that with all my applications I have needed to subscribe from dynamic processes. e.g. live views, scenic views, etc. And all my projects use the same code (but different versions) to handle dynamic subscriptions from different processes. So I might end up including this, or something like it in MqttPotion also:

github.com

brianmay/scrooge/blob/main/lib/scrooge/mqtt_multiplexer.ex

defmodule Scrooge.MqttMultiplexer do
  @moduledoc """
  Implement per process MQTT subscriptions
  """
  use GenServer
  require Logger

  defmodule State do
    @moduledoc """
    Define current state of Subscriptions.
    """
    @type t :: %__MODULE__{
            subscriptions: %{list(String.t()) => list({String.t(), pid, :json | :raw})},
            last_message: %{list(String.t()) => any()},
            monitor: %{required(pid) => reference()}
          }
    defstruct subscriptions: %{}, last_message: %{}, monitor: %{}
  end

  ## Client API

This file has been truncated. show original

I just noticed some things should be configurable that are currently hard coded (e.g. client_name = PenguinNodes.Mqtt).

brianmay · August 16, 2022, 10:45pm

Am having second thoughts on using emqtt. Randomly the connection keeps going down. Client:

10:28:29.066 [warn]  [MqttPotion] Disconnect received: reason :shutdown, properties: :ssl_closed
10:28:29.067 [info]  Got disconnected :shutdown :ssl_closed.
10:28:29.068 [warn]  [MqttPotion] Got Exit
10:28:29.116 [info]  [MqttPotion] Connected #PID<0.19392.1>
10:28:29.116 [info]  Got connected.

Server:

1660645709: Client robotica-nerves-c775 has exceeded timeout, disconnecting.
1660645709: New connection from 2001:44b8:4112:8a10:ba27:ebff:fe5a:9220:52780 on port 8883.
1660645709: New client connected from 2001:44b8:4112:8a10:ba27:ebff:fe5a:9220:52780 as robotica-nerves-c77
5 (p5, c1, k30, u'robotica').

Still investigating this. But really wish I didn’t have to debug mqtt libraries.

At first I thought maybe emqtt isn’t sending the keepalive packets. So added some debugging, in emqtt, it seems to be working fine. But then it stopped crashing on the test system also.

Also seeing an unrelated process - one that publishes messages using MqttPotion.publish which is just a call to GenServer.cast - hang for long periods of time when the above happens. Which doesn’t make any sense. I used cast precisely to prevent mqtt errors interfering with the process that wants to publish messages.

Or maybe just a coincidence that my process stopped working at the same time. Will add more debug statements to try to work this out. But debugging random intermittent faults is painful.

So far I have noticed that if there is any error in the connection, the emqtt process exits. Which means any packets in transit will get lost, even if they have qos==1 or qos==2. Or maybe I am suppose to i implement these myself. Without good documentation this is somewhat confusing.

Will keep trying but wondering if I should rewrite this in some language that has good mqtt client support.