Interacting with emqtt through an Erlang/OTP module #14402

SaveMe0 · 2024-12-12T13:49:32Z

SaveMe0
Dec 12, 2024

Hi EMQX Team

I've created a gen_server module named com.erl to interact with emqtt functions. In the emqtt README.md file is stated that the clientId is the process Pid of the module being used to communicate with emqtt. So in my case com.erl

 -module(com).
 -behaviour(gen_server).
 
 -export([start/0, stop/0, cnct/0, subscribe/0]).
 -export([init/1, handle_call/3, handle_cast/2, terminate/2, handle_info/2]).
 
 -record(state, {clientID}).
 
 %% CLIENT FUNCTIONS
 
 %% start() -> {ok, pid()} | {error, Reason}
 %% Starts the com server. Called by supervisor
 
 start() ->
     gen_server:start_link({local, com}, com, [], []).
 
 %% stop() -> ok.
 %% Stops the com server.
 
 stop() ->
     gen_server:cast(com, stop).
 
 
 cnct() ->
 	gen_server:cast(com, cnct).
 
 subscribe() ->
     gen_server:call(com, subscribe).
 
 init([]) ->
     State = #state{clientID = self()},
     {ok, State}.
 
 %% -spec(open_quic_connection(client()) -> ok | {error, term()}).
 handle_cast(cnct, State) ->
 	io:format("State Pid ~p~n", [State#state.clientID]),
       %emqtt:open_quic_connection(State#state.clientID),
   	emqtt:quic_mqtt_connect(State#state.clientID),
     {noreply, State};
 handle_cast(stop, State) ->
     {stop, normal, State}.
 
 %% -spec(subscribe(client(), topic() | {topic(), qos() | qos_name() | [subopt()]} | [{topic(), qos()}])
       %% -> subscribe_ret()).
 handle_call(subscribe, _From, State) ->
     Properties = #{'Payload-Format-Indicator' => 0,
                   'Message-Expiry-Interval' => 10,
                   'Content-Type' => 1,
                   'Server-Keep-Alive' => 30,
                   %'Maximum-QoS' => 1,
                   'Retain-Available' => 1},
    SubOpts = [{qos, 1}],
    Subscriptions = [{<<"EK9160/EK-98EFB6/Stream1/Json/Tx/Data">>, SubOpts}],
    Reply = emqtt:subscribe(State#state.clientID, Properties, Subscriptions),
    {reply, Reply, State}.

handle_info(_Msg, State) ->
    {noreply, State}.

%% terminate(Reason, LoopData) -> ok.
%% Termination callback. Does nothing, but should instead kill clients.

terminate(_Reason, _LoopData) ->
    ok.

Unfortunately when I compile and run the module I get the following errors

alberto@debian:~/testapp$ rebar3 shell
===> Verifying dependencies...
===> Analyzing applications...
===> Compiling testapp
Erlang/OTP 27 [erts-15.1.2] [source] [64-bit] [smp:1:1] [ds:1:1:10] [async-threads:1] [jit:ns]

===> Booted testapp
Eshell V15.1.2 (press Ctrl+G to abort, type help(). for help)
1> com:start().
{ok,<0.208.0>}
2> com:cnct().
State Pid <0.208.0>
ok
=ERROR REPORT==== 12-Dec-2024::14:29:43.101800 ===
** Generic server com terminating 
** Last message in was {'$gen_cast',cnct}
** When Server state == {state,<0.208.0>}
** Reason for termination ==
** {{calling_self,{gen_statem,call,[<0.208.0>,quic_mqtt_connect,infinity]}},
    [{gen,do_call,4,[{file,"gen.erl"},{line,230}]},
     {gen_statem,call,3,[{file,"gen_statem.erl"},{line,3250}]},
     {com,handle_cast,2,
          [{file,"/home/alberto/testapp/src/com.erl"},{line,38}]},
     {gen_server,try_handle_cast,3,[{file,"gen_server.erl"},{line,2371}]},
     {gen_server,handle_msg,6,[{file,"gen_server.erl"},{line,2433}]},
     {proc_lib,init_p_do_apply,3,[{file,"proc_lib.erl"},{line,329}]}]}

=CRASH REPORT==== 12-Dec-2024::14:29:43.102380 ===
  crasher:
    initial call: com:init/1
    pid: <0.208.0>
    registered_name: com
    exception exit: {calling_self,
                        {gen_statem,call,
                            [<0.208.0>,quic_mqtt_connect,infinity]}}
      in function  gen:do_call/4 (gen.erl, line 230)
      in call from gen_statem:call/3 (gen_statem.erl, line 3250)
      in call from com:handle_cast/2 (/home/alberto/testapp/src/com.erl, line 38)
      in call from gen_server:try_handle_cast/3 (gen_server.erl, line 2371)
      in call from gen_server:handle_msg/6 (gen_server.erl, line 2433)
    ancestors: [<0.206.0>,<0.203.0>,<0.200.0>,<0.66.0>,<0.70.0>,<0.65.0>,
                  kernel_sup,<0.47.0>]
    message_queue_len: 0
    messages: []
    links: [<0.206.0>]
    dictionary: []
    trap_exit: false
    status: running
    heap_size: 4185
    stack_size: 29
    reductions: 7192
  neighbours:
    neighbour:
      pid: <0.206.0>
      registered_name: []
      initial_call: {erlang,apply,2}
      current_function: {shell,eval_loop,4}
      ancestors: [<0.203.0>,<0.200.0>,<0.66.0>,<0.70.0>,<0.65.0>,kernel_sup,
                  <0.47.0>]
      message_queue_len: 0
      links: [<0.203.0>,<0.208.0>]
      trap_exit: false
      status: waiting
      heap_size: 610
      stack_size: 8
      reductions: 2639
      current_stacktrace: [{shell,eval_loop,4,[{file,"shell.erl"},{line,828}]}]
** exception error: {calling_self,
                        {gen_statem,call,
                            [<0.208.0>,quic_mqtt_connect,infinity]}}
3>

Could someone who is currently involved in emqtt kindly run my module and let me know what I'm doing wrong.

ieQu1 · 2024-12-13T04:42:41Z

ieQu1
Dec 13, 2024
Collaborator

This line tells exactly what is wrong:

** {{calling_self,{gen_statem,call,[<0.208.0>,quic_mqtt_connect,infinity]}},

This is because your init function does this:

     State = #state{clientID = self()},

So instead of calling emqtt process, you call self(). gen_server prevents the deadlock by throwing an exception. I recommend reading about the gen_server behavior here, for example: https://learnyousomeerlang.com/what-is-otp#the-basic-server

1 reply

SaveMe0 Dec 14, 2024
Author

Yes, I realized the mistake. I can never learn enough Erlang thank you. I find the README.md lacking coherence. In the example section the following lines of code which ends up to be quite confusing.

Examples

{ok, ConnPid} = emqtt:start_link([{clientid, ClientId}]).
{ok, _Props} = emqtt:connect(ConnPid).

I'm sure I'm not the only person that would ask "Where do I obtain the ClientId". Perhaps if one is using the mqttx client this may make sense as the ClientId is generated by it, but there is not mention in the document about client the example is referring too.

After several reads of the discombobulated documentation, one comes across information quoted below, and it will dawn on you that the Pid returned from the function is to be used as the ClientId for the emqtt client. It may also dawn on you that the {owner, Pid} is an option that should set so that in an "handle_info/2" to get status of the emqtt client.

Connect to the MQTT server over TCP or TLS and send a CONNECT packet with the options specified in start_link/1, 2. Client must be a pid returned from start_link/1, 2 or a name specified in start_link/1, 2.

And if you're still awake you'll realize that there's this export

Exports

emqtt:start_link() -> {ok, Pid} | ignore | {error, Reason}

emqtt:start_link(Options) -> {ok, Pid} | ignore | {error, Reason}

Types

Pid = pid()

Reason = term()

Options = [option()]

Start MQTT client process with specified options. Options will be used in connecting and running.

The following options are available:

{name, Name}

If a name is provided, the gen_statem will be registered with this name. For details see the documentation for the first argument of gen_statem:start_link/4.

{owner, Pid}
client process will send messages like {diconnected, ReasonCode, Properties} to the owner process.

We should all learn to explain things like Fred Herbert the author of "Learn You Some Erlang" in the link you sent me.

My pursuit, is to create proper emqtt clients by referencing functions in the emqtt application. And for this I'm beg your patience and those of other members in the EMQX community.

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Interacting with emqtt through an Erlang/OTP module #14402

{{title}}

Replies: 1 comment 1 reply

{{title}}

{{editor}}'s edit

{{editor}}'s edit

{{title}}

Select a reply

Interacting with emqtt through an Erlang/OTP module #14402

SaveMe0 Dec 12, 2024

Replies: 1 comment · 1 reply

ieQu1 Dec 13, 2024 Collaborator

SaveMe0 Dec 14, 2024 Author

SaveMe0
Dec 12, 2024

Replies: 1 comment 1 reply

ieQu1
Dec 13, 2024
Collaborator

SaveMe0 Dec 14, 2024
Author