emqtt

You can configure a 'Will Message' (Last Will and Testament) using the pub or sub commands. This message is sent by the broker if the client disconnects unexpectedly.

Required for Will Messages:

• --will-topic <topic>: The topic for the will message.
• --will-payload <payload>: The message content for the will.

Optional Will Options:

• --will-qos <qos>: QoS for the will message (defaults to 0).
• --will-retain <bool>: Whether the will message is retained.

While the emqtt CLI does not support enhanced authentication, the emqtt library allows you to implement it by providing custom callbacks via the start_link option.

To enable enhanced authentication, pass a map {custom_auth_callbacks, Callbacks} to start_link. The Callbacks map must contain two keys: init and handle_auth.

%% The structure of the Callbacks map
Callbacks = #{ 
  init => InitFunc, 
  handle_auth => HandleAuth 
}.

%% Example start_link usage
{ok, Client} = emqtt:start_link(#{custom_auth_callbacks, Callbacks}).

To use emqtt as a dependency in your Erlang project, add it to your rebar.config file using the following syntax:

{deps, [{emqtt, {git, "https://github.com/emqx/emqtt", {tag, "1.14.4"}}}]}.

Then, compile your project using:

rebar3 compile

To use emqtt as a CLI tool, you must first build it using make. The compiled executable will be located in _build/emqtt/rel/emqtt/bin/emqtt.

If you encounter compilation issues related to QUIC, you can disable QUIC support by setting the BUILD_WITHOUT_QUIC environment variable.

$ make

# To disable QUIC support if compilation fails:
$ BUILD_WITHOUT_QUIC=1 make

When running an emqtt client, you can receive messages via a receive block. Common message patterns include:

• {disconnected, ReasonCode, Properties}: Triggered when the connection is lost.
• {publish, PUBLISH}: Triggered when a PUBLISH packet is received.
• {puback, {PacketId, ReasonCode, Properties}}: Triggered when a PUBACK is received (for QoS 1).

receive
    {disconnected, ReasonCode, Properties} ->
        io:format("Recv a DISCONNECT packet - ReasonCode: ~p, Properties: ~p~n", [ReasonCode, Properties]);
    {publish, PUBLISH} ->
        io:format("Recv a PUBLISH packet: ~p~n", [PUBLISH]);
    {puback, {PacketId, ReasonCode, Properties}} ->
        io:format("Recv a PUBACK packet - PacketId: ~p, ReasonCode: ~p, Properties: ~p~n", [PacketId, ReasonCode, Properties])
end.

Control the client process state using the following functions:

• emqtt:disconnect(Client, [ReasonCode, Properties]): Sends a DISCONNECT packet to the server. ReasonCode defaults to 0 (normal). Returns ok or {error, Reason}.
• emqtt:stop(Client): Stops the client process entirely.
• emqtt:pause(Client): Pauses the client. The process will ignore incoming PUBLISH packets and (if force_ping is false) will not send PINGREQ.
• emqtt:resume(Client): Resumes a paused client process.

ok = emqtt:disconnect(ConnPid).
ok = emqtt:stop(ConnPid).

Once the client process is started, use emqtt:connect(Client) or emqtt:ws_connect(Client) to establish the connection.

• emqtt:connect(Client): Connects over TCP or TLS.
• emqtt:ws_connect(Client): Connects over WebSockets.

Returns:

• {ok, Properties}: Connection established. Properties contains the CONNACK properties from the server.
• {error, timeout}: Connection failed to establish within the timeout.
• {error, inet:posix()}: POSIX error occurred.

{ok, _Props} = emqtt:connect(ConnPid).

Send an UNSUBSCRIBE packet to the server.

Signature: emqtt:unsubscribe(Client, Properties, Topics)

• Properties: MQTT v5 properties (properties()).
• Topics: A list of topic filters ([topic()]).

Returns:

• {ok, Properties, ReasonCodes}: Success.
• {error, Reason}: If the operation fails.

{ok, _Props, _ReasonCode} = emqtt:unsubscribe(ConnPid, #{}, <<"hello">).

Send a PUBLISH packet to the server.

Signature: emqtt:publish(Client, Topic, Properties, Payload, PubOpts)

• Topic: The destination topic (binary()).
• Properties: MQTT v5 properties (properties()).
• Payload: The message content (iodata()).
• PubOpts: Options like {qos, qos()} or {retain, boolean()}. Defaults to [] (equivalent to {qos, 0} and {retain, false}).

Returns:

• ok: For QoS 0 packets.
• {ok, PacketId}: For QoS 1/2 packets (returns the packet identifier).
• {error, Reason}: If the operation fails.

% QoS 0
ok = emqtt:publish(ConnPid, <<"hello">>, #{}, <<"Hello World!">>, [{qos, 0}]).

% QoS 1
{ok, _PktId} = emqtt:publish(ConnPid, <<"hello">>, #{}, <<"Hello World!">>, [{qos, 1}]).

Send a SUBSCRIBE packet to the server.

Signature: emqtt:subscribe(Client, Properties, Subscriptions)

• Properties: MQTT v5 properties (properties()).
• Subscriptions: A list of {Topic, SubOptions} pairs.
  • SubOptions can be [] (defaults to {rh, 0}, {rap, 0}, {nl, 0}, {qos, 0}) or specific options like {qos, qos()}.

Returns:

• {ok, Properties, ReasonCodes}: Success. ReasonCodes is a list corresponding to the subscriptions.
• {error, Reason}: If the operation fails.

SubOpts = [{qos, 1}].
{ok, _Props, _ReasonCodes} = emqtt:subscribe(ConnPid, #{}, [[<"hello">], SubOpts]]).

For practical implementations of enhanced authentication (such as SCRAM or Kerberos), refer to the following test suites in the repository:

• test/emqtt_scram_auth_SUITE.erl (SCRAM mechanism)
• test/emqtt_kerberos_auth_SUITE.erl (Kerberos mechanism)