Need help using emqtt with quic over mqtt on Erlang/OTP 27 #14415

SaveMe0 · 2024-12-14T14:29:51Z

SaveMe0
Dec 14, 2024

Hi emqtt team,

I'm having problems trying to determine which of the following functions I should use properly connect to emqtt. By trial and error I have found that this is the only function that does not return an error is the uncommented one in the code snippet below.

    %emqtt:connect(State#state.clientID),
    %emqtt:quic_connect(State#state.clientID),
    emqtt:open_quic_connection(State#state.clientID),
    %emqtt:quic_mqtt_connect(State#state.clientID),

Assuming that the function I used is correct, when I run the following code to carry out a subscription, I get a timeout error event though I've extended it

handle_call(subscribe, _From, State) ->
    io:format("State in subscribe ~p~n", [State#state.clientID]),
    Properties = #{},
    SubOpts = [{qos, 1}],

   Subscriptions = [{<<"EK9160/EK-98EFB6/Stream1/Json/Tx/Data">>, SubOpts}],
   Reply = emqtt:subscribe(State#state.clientID, Properties, Subscriptions),
   {reply, Reply, State, ?TIMEOUT}.

Here is a copy of the module I've created
-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]).


-define(TIMEOUT, 8000).

-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(_Args) ->
     Properties = #{'Session-Expiry-Interval' => 300,
                    'Server-Keep-Alive' => 30,
                    'Retain-Available' => 1
                },

    Options = [ {host,'192.168.178.33'},
                {port, 14567},
                {proto_ver, v5},
                {reconnect_timeout, 1000},
                %{quic_opts, {quicer:conn_opts(), quicer:stream_opts()}},
                {properties, Properties}
            ],

    {ok, ConnPid} = emqtt:start_link(Options),
    io:format("ConnPid ~p~n", [ConnPid]),
    State = #state{clientID = ConnPid},
    {ok, State}.


%% -spec(open_quic_connection(client()) -> ok | {error, term()}).
handle_cast(cnct, State) ->
io:format("ConnPid ~p~n", [State#state.clientID]),
    %emqtt:connect(State#state.clientID),
    %emqtt:quic_connect(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) ->
    io:format("State in subscribe ~p~n", [State#state.clientID]),
    Properties = #{},
    SubOpts = [{qos, 1}],

   Subscriptions = [{<<"EK9160/EK-98EFB6/Stream1/Json/Tx/Data">>, SubOpts}],
   Reply = emqtt:subscribe(State#state.clientID, Properties, Subscriptions),
   {reply, Reply, State, ?TIMEOUT}.


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

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

terminate(_Reason, _LoopData) ->
    ok.

When I run the code I get this error report

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> 
User switch command (type h for help)
 --> qalberto@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().
ConnPid <0.216.0>
{ok,<0.215.0>}
2> com:cnct().
ConnPid <0.216.0>
ok
3> com:subscribe().
State in subscribe <0.216.0>
=ERROR REPORT==== 14-Dec-2024::15:00:48.950822 ===
    state: initialized
    msg: unexpected_event
    clientid: <<>>
    event: {subscribe,#{},
                      [{<<"EK9160/EK-98EFB6/Stream1/Json/Tx/Data">>,
                        #{nl => 0,qos => 1,rap => 0,rh => 0}}]}
    event_type: {call,{<0.215.0>,#Ref<0.142594499.1627652097.24636>}}
** exception exit: {timeout,{gen_server,call,[com,subscribe]}}
     in function  gen_server:call/2 (gen_server.erl, line 1142)
=ERROR REPORT==== 14-Dec-2024::15:00:53.988575 ===
** State machine <0.216.0> terminating
** When server state  = {initialized,
                         {state,undefined,<0.215.0>,undefined,
                          '192.168.178.33',14567,[],emqtt_quic,
                          {quic,#Ref<0.142594499.1627783169.24626>,undefined},
                          [{handle,#Ref<0.142594499.1627783169.24626>}],
                          60000,false,<<>>,true,undefined,undefined,5,
                          <<"MQTT">>,60,undefined,false,false,undefined,
                          #{'Session-Expiry-Interval' => 300,
                            'Server-Keep-Alive' => 30,'Retain-Available' => 1},
                          [],#{},
                          #{seq => 1,sent => #{},max_inflight => infinity},
                          #{},true,30000,undefined,30000,undefined,undefined,
                          1,false,
                          {none,
                           #{version => 5,max_size => 268435455,
                             strict_mode => false}},
                          0,1000000,false,undefined,
                          {[],[]},
                          #{clientid => <<>>,reconnect => false,
                            control_stream_sock => undefined,
                            conn_parse_state =>
                             {none,
                              #{version => 5,max_size => 268435455,
                                strict_mode => false}},
                            data_stream_socks => [],logic_stream_map => #{},
                            state_name => init,is_local => true,
                            is_unidir => false,
                            quic_conn_cb => emqtt_quic_connection,
                            quic_stream_cb => emqtt_quic_stream}}}
** Reason for termination = exit:{{timeout,{gen_server,call,[com,subscribe]}},
                                  [{gen_server,call,2,
                                       [{file,"gen_server.erl"},{line,1142}]},
                                   {erl_eval,do_apply,7,
                                       [{file,"erl_eval.erl"},{line,904}]},
                                   {shell,exprs,7,
                                       [{file,"shell.erl"},{line,893}]},
                                   {shell,eval_exprs,7,
                                       [{file,"shell.erl"},{line,849}]},
                                   {shell,eval_loop,4,
                                       [{file,"shell.erl"},{line,834}]}]}
** Callback modules = [emqtt]
** Callback mode = state_functions
** Stacktrace =
**  [{gen_statem,loop_receive,3,[{file,"gen_statem.erl"},{line,3625}]},
     {proc_lib,init_p_do_apply,3,[{file,"proc_lib.erl"},{line,329}]}]

=CRASH REPORT==== 14-Dec-2024::15:00:53.989086 ===
  crasher:
    initial call: emqtt:init/1
    pid: <0.216.0>
    registered_name: []
    exception exit: {{timeout,{gen_server,call,[com,subscribe]}},
                     [{gen_server,call,2,
                                  [{file,"gen_server.erl"},{line,1142}]},
                      {erl_eval,do_apply,7,[{file,"erl_eval.erl"},{line,904}]},
                      {shell,exprs,7,[{file,"shell.erl"},{line,893}]},
                      {shell,eval_exprs,7,[{file,"shell.erl"},{line,849}]},
                      {shell,eval_loop,4,[{file,"shell.erl"},{line,834}]}]}
      in function  gen_statem:loop_receive/3 (gen_statem.erl, line 3625)
    ancestors: [com,<0.213.0>,<0.210.0>,<0.207.0>,<0.66.0>,<0.70.0>,
                  <0.65.0>,kernel_sup,<0.47.0>]
    message_queue_len: 0
    messages: []
    links: []
    dictionary: []
    trap_exit: true
    status: running
    heap_size: 987
    stack_size: 29
    reductions: 22642
  neighbours:

4>

I been trying to find a solution to this problem for over a week now, with no success. So I really need your help - please.

ieQu1 · 2024-12-17T08:11:18Z

ieQu1
Dec 17, 2024
Collaborator

You need to use emqtt:quick_connect function.

2 replies

SaveMe0 Dec 18, 2024
Author

when I use the > emqtt:quick_connect(ClientId), function I get the error

ClientId <0.209.0>
=ERROR REPORT==== 18-Dec-2024::09:04:30.564203 ===
** State machine com2 terminating
** Last event = {cast,cnct}
** When server state  = {cnctState,{connPid,<0.209.0>}}
** Reason for termination = error:{'function not exported',
                                      {emqtt,quick_connect,1}}

The ClientId is the Pid return by the > {ok, ConnPid} = emqtt:start_link(Options),

ieQu1 Dec 19, 2024
Collaborator

Sorry, I made a typo, should be emqtt:quic_connect.

qzhuyan · 2024-12-17T10:00:50Z

qzhuyan
Dec 17, 2024
Collaborator

emqtt:open_quic_connection is not exposed as it is for testing EMQX intrenally.

0 replies

SaveMe0 · 2024-12-18T09:38:16Z

SaveMe0
Dec 18, 2024
Author

After studying the emqtt.erl module in the emqtt application I realized that the problems with the emqtt module result in an error because the "clientid = ClientId," in the options that make up the loop data for the generic process in the code named "State" is not initialized. In fact it can never be initialized because when the emqtt: start_link/0 or start_link/1is called to create a client (in my case through my com2.erl module), the clientId can only be returned no matter which init/0 or init/1 function is invoked base on options passed to it, the ClientId is made available only when the init/0 or init/1 function returns to from start_link/0 or start_link/1. The ClientId is the Pid value in the {ok, Pid} tuple returned from start_link/0 or start_link/1 is the ClientId, e.g. (ClientId = Pid). Although, the ClientId is passed to functions in the calls in the emqtt module, e.g.

{ok, ConnPid} = emqtt:start_link(Options),
emqtt:open_quic_connection(ClientId),
emqtt:subscribe(ClientId, Properties, Subscriptions),

other functions that are called internally in the emqtt module are expecting the ClientId in the State loop variable which is not initialized.

init([Options]) ->
process_flag(trap_exit, true),
ClientId = maybe_rand_id(proplists:get_value(proto_ver, Options, v4),
proplists:get_value(clientid, Options, ?NO_CLIENT_ID)),
State = check_options(
init(Options,
#state{...
clientid = ClientId,
....
})),
{ok, initialized, init_parse_state(State)}.

From my perspective this is a design failure in the emqtt module. The problem shows up when a user if the emqtt application goes about trying to create his/her own emqtt OTP client.

10 replies

id Dec 18, 2024
Maintainer

In fact, emqtt will generate random client id automatically if it's not supplied in the options. You would've noticed it if you cared enough to read documentation and the code.

SaveMe0 Dec 18, 2024
Author

Sorry. I'm just learning, so please have some patience with me.
Can you confirm if what I'm doing here is correct? I've put the ClientId in the Options list.
-record(myemqtt, {connPid}).

init(_Args) ->
process_flag(trap_exit, true),
Properties = #{'Session-Expiry-Interval' => 300,
'Server-Keep-Alive' => 30,
'Retain-Available' => 1
},

Options = [ {host,'192.168.178.33'},
            {port, 14567},
            **{clientid, <<"alb65McbrideBCCanada">>}**,
            {proto_ver, v5},
            {reconnect_timeout, 1000},
            %{quic_opts, {quicer:conn_opts(), quicer:stream_opts()}},
            {properties, Properties}
        ],

{ok, ConnPid} = emqtt:start_link(Options),
State = cnctState, 
Data = #myemqtt{connPid = ConnPid},
{ok, State, Data}.

SaveMe0 Dec 18, 2024
Author

Could someone please tell me if the way I subscribing is correct. I get a successful emqtt:start(opitons) and a successful emqtt:cnct(). Yet when I subscribe to a published message I get errors. I have trying to do this for a very long time, and all I'm asking is that someone can tell what I doing wrong.

handle_event({call,From}, sub, Data) ->
    io:format("ConnPid ~p~n", [Data#connPid.clientID]),

    ConnPid = Data#connPid.clientID,
    Properties = #{},
    SubOpts = [{qos, 1}],
    Subscriptions = [{<<"EK9160/EK-98EFB6/Stream1/Json/Tx/Data">>, SubOpts}],
    EmqttMsg = emqtt:subscribe(ConnPid, Properties, Subscriptions),
    io:format("EmqttMsg~p~n", [EmqttMsg]),
    {keep_state,Data,[{reply,From,Data}]};
handle_event(_, _, Data) ->
    %% Ignore all other events
    {keep_state, Data}.

Here is the error report

3> com2:sub().
ConnPid <0.216.0>
=ERROR REPORT==== 18-Dec-2024::15:58:23.008606 ===
** State machine com2 terminating
** Last event = {{call,{<0.213.0>,#Ref<0.705875627.1455685633.68715>}},sub}
** When server state  = {cnctState,{connPid,<0.216.0>,undefined}}
** Reason for termination = exit:{noproc,
                                  {gen_statem,call,
                                   [<0.216.0>,
                                    {subscribe,#{},
                                     [{<<"EK9160/EK-98EFB6/Stream1/Json/Tx/Data">>,
                                       #{nl => 0,qos => 1,rap => 0,rh => 0}}]},
                                    infinity]}}
** Callback modules = [com2]
** Callback mode = [state_functions,state_enter]
** Stacktrace =
**  [{gen,do_call,4,[{file,"gen.erl"},{line,247}]},
     {gen_statem,call,3,[{file,"gen_statem.erl"},{line,3250}]},
     {com2,handle_event,3,
           [{file,"/home/alberto/testapp/src/com2.erl"},{line,74}]},
     {gen_statem,loop_state_callback,11,[{file,"gen_statem.erl"},{line,3735}]},
     {proc_lib,init_p_do_apply,3,[{file,"proc_lib.erl"},{line,329}]}]
** Client <0.213.0> stacktrace
** [{gen,do_call,4,[{file,"gen.erl"},{line,241}]},
    {gen_statem,call,3,[{file,"gen_statem.erl"},{line,3250}]},
    {erl_eval,do_apply,7,[{file,"erl_eval.erl"},{line,904}]},
    {shell,exprs,7,[{file,"shell.erl"},{line,893}]},
    {shell,eval_exprs,7,[{file,"shell.erl"},{line,849}]},
    {shell,eval_loop,4,[{file,"shell.erl"},{line,834}]}]

=CRASH REPORT==== 18-Dec-2024::15:58:23.013811 ===
  crasher:
    initial call: com2:init/1
    pid: <0.215.0>
    registered_name: com2
    exception exit: {noproc,
                        {gen_statem,call,
                            [<0.216.0>,
                             {subscribe,#{},
                                 [{<<"EK9160/EK-98EFB6/Stream1/Json/Tx/Data">>,
                                   #{nl => 0,qos => 1,rap => 0,rh => 0}}]},
                             infinity]}}
      in function  gen:do_call/4 (gen.erl, line 247)
      in call from gen_statem:call/3 (gen_statem.erl, line 3250)
      in call from com2:handle_event/3 (/home/alberto/testapp/src/com2.erl, line 74)
      in call from gen_statem:loop_state_callback/11 (gen_statem.erl, line 3735)
    ancestors: [<0.213.0>,<0.210.0>,<0.207.0>,<0.66.0>,<0.70.0>,<0.65.0>,
                  kernel_sup,<0.47.0>]
    message_queue_len: 0
    messages: []
    links: [<0.213.0>]
    dictionary: []
    trap_exit: true
    status: running
    heap_size: 610
    stack_size: 29
    reductions: 13879
  neighbours:
    neighbour:
      pid: <0.213.0>
      registered_name: []
      initial_call: {erlang,apply,2}
      current_function: {gen,do_call,4}
      ancestors: [<0.210.0>,<0.207.0>,<0.66.0>,<0.70.0>,<0.65.0>,kernel_sup,
                  <0.47.0>]
      message_queue_len: 0
      links: [<0.210.0>,<0.215.0>]
      trap_exit: false
      status: waiting
      heap_size: 610
      stack_size: 34
      reductions: 3691
      current_stacktrace: [{gen,do_call,4,[{file,"gen.erl"},{line,241}]},
                  {gen_statem,call,3,[{file,"gen_statem.erl"},{line,3250}]},
                  {erl_eval,do_apply,7,[{file,"erl_eval.erl"},{line,904}]},
                  {shell,exprs,7,[{file,"shell.erl"},{line,893}]},
                  {shell,eval_exprs,7,[{file,"shell.erl"},{line,849}]},
                  {shell,eval_loop,4,[{file,"shell.erl"},{line,834}]}]
** exception exit: {noproc,
                       {gen_statem,call,
                           [<0.216.0>,
                            {subscribe,#{},
                                [{<<"EK9160/EK-98EFB6/Stream1/Json/Tx/Data">>,
                                  #{nl => 0,qos => 1,rap => 0,rh => 0}}]},
                            infinity]}}

SaveMe0 Dec 18, 2024
Author

Dear Mr. Maintainer, a comment like these not fair.

Please stop acting so aggressive and entitled @SaveMe0. You just started learning this library, Erlang and MQTT, and already giving out advices and judgement. This behavior is not something we appreciate.

In fact, emqtt will generate random client id automatically if it's not supplied in the options. You would've noticed it if you cared enough to read documentation and the code.

It is truly not my intention to judge programmes who write application that users find themselves successfully using to the extent at which the application was developed to be used. And Secondly, before I ask any questions, I go to great lengths to do all the necessary reading to answer my questions first.

I have used the mqttx application to get familiar with how mqtt works under emqx and to know that ClientId is generated automatically. When I started with emqtt the following functions worked, which convinced that emqtt also generates random ClientIds automatically as well.

emqtt:start_link(Options),
emqtt:quic_connect(ConnPid),

My problem is not being successful to subscribing to a topic now for a very very long time, led me I read the emqtt README.md documentation several times where the following lines of code distorted my understanding. I stated to believe that that even though ClientIds are generated, in order to subscribe to a topic, ClientIds had to be created manually. I'm sure I'm not the only person getting confused this way.

Examples

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

Please, I'm not asking anyone to write code for me, I'm asking you and other in the emqtt programmers in the group what I'm doing wrong with in this small snippets of code. I have place in a gen_server and in a gen_statem with the hope that perhaps that would solve the problem. In the gen_statem I handheld calls and casts in appropriate states and finding that that did not work modified the subscription state to use a handle_event/3 callback instead show below. The change unfortunately gave the same error. Could you be so kind to point what is wrong with the following snippet of code. In the case that has to be amended with additional code could you provide me with a hint of what that may be.

handle_event({call,From}, sub, Data) ->
    ConnPid = Data#connPid.clientID,
    Properties = #{},
    SubOpts = [{qos, 1}],
    Subscriptions = [{<<"EK9160/EK-98EFB6/Stream1/Json/Tx/Data">>, SubOpts}],
    EmqttMsg = emqtt:subscribe(ConnPid, Properties, Subscriptions),
    io:format("EmqttMsg~p~n", [EmqttMsg]),
    {keep_state,Data,[{reply,From,Data}]};
handle_event(_, _, Data) ->
    %% Ignore all other events
    {keep_state, Data}.

Here is the Error report that is generated for it.

3> com2:sub().
=ERROR REPORT==== 18-Dec-2024::18:11:31.996673 ===
** State machine com2 terminating
** Last event = {{call,{<0.213.0>,#Ref<0.3051605496.3355705345.22542>}},sub}
** When server state  = {cnctState,{connPid,<0.216.0>,undefined}}
** Reason for termination = exit:{noproc,
                                  {gen_statem,call,
                                   [<0.216.0>,
                                    {subscribe,#{},
                                     [{<<"EK9160/EK-98EFB6/Stream1/Json/Tx/Data">>,
                                       #{nl => 0,qos => 1,rap => 0,rh => 0}}]},
                                    infinity]}}
** Callback modules = [com2]
** Callback mode = [state_functions,state_enter]
** Stacktrace =
**  [{gen,do_call,4,[{file,"gen.erl"},{line,247}]},
     {gen_statem,call,3,[{file,"gen_statem.erl"},{line,3250}]},
     {com2,handle_event,3,
           [{file,"/home/alberto/testapp/src/com2.erl"},{line,72}]},
     {gen_statem,loop_state_callback,11,[{file,"gen_statem.erl"},{line,3735}]},
     {proc_lib,init_p_do_apply,3,[{file,"proc_lib.erl"},{line,329}]}]
** Client <0.213.0> stacktrace
** [{gen,do_call,4,[{file,"gen.erl"},{line,241}]},
    {gen_statem,call,3,[{file,"gen_statem.erl"},{line,3250}]},
    {erl_eval,do_apply,7,[{file,"erl_eval.erl"},{line,904}]},
    {shell,exprs,7,[{file,"shell.erl"},{line,893}]},
    {shell,eval_exprs,7,[{file,"shell.erl"},{line,849}]},
    {shell,eval_loop,4,[{file,"shell.erl"},{line,834}]}]

=CRASH REPORT==== 18-Dec-2024::18:11:31.999263 ===
  crasher:
    initial call: com2:init/1
    pid: <0.215.0>
    registered_name: com2
    exception exit: {noproc,
                        {gen_statem,call,
                            [<0.216.0>,
                             {subscribe,#{},
                                 [{<<"EK9160/EK-98EFB6/Stream1/Json/Tx/Data">>,
                                   #{nl => 0,qos => 1,rap => 0,rh => 0}}]},
                             infinity]}}
      in function  gen:do_call/4 (gen.erl, line 247)
      in call from gen_statem:call/3 (gen_statem.erl, line 3250)
      in call from com2:handle_event/3 (/home/alberto/testapp/src/com2.erl, line 72)
      in call from gen_statem:loop_state_callback/11 (gen_statem.erl, line 3735)
    ancestors: [<0.213.0>,<0.210.0>,<0.207.0>,<0.66.0>,<0.70.0>,<0.65.0>,
                  kernel_sup,<0.47.0>]
    message_queue_len: 0
    messages: []
    links: [<0.213.0>]
    dictionary: []
    trap_exit: true
    status: running
    heap_size: 610
    stack_size: 29
    reductions: 13864
  neighbours:
    neighbour:
      pid: <0.213.0>
      registered_name: []
      initial_call: {erlang,apply,2}
      current_function: {gen,do_call,4}
      ancestors: [<0.210.0>,<0.207.0>,<0.66.0>,<0.70.0>,<0.65.0>,kernel_sup,
                  <0.47.0>]
      message_queue_len: 0
      links: [<0.210.0>,<0.215.0>]
      trap_exit: false
      status: waiting
      heap_size: 610
      stack_size: 34
      reductions: 3691
      current_stacktrace: [{gen,do_call,4,[{file,"gen.erl"},{line,241}]},
                  {gen_statem,call,3,[{file,"gen_statem.erl"},{line,3250}]},
                  {erl_eval,do_apply,7,[{file,"erl_eval.erl"},{line,904}]},
                  {shell,exprs,7,[{file,"shell.erl"},{line,893}]},
                  {shell,eval_exprs,7,[{file,"shell.erl"},{line,849}]},
                  {shell,eval_loop,4,[{file,"shell.erl"},{line,834}]}]
** exception exit: {noproc,
                       {gen_statem,call,
                           [<0.216.0>,
                            {subscribe,#{},
                                [{<<"EK9160/EK-98EFB6/Stream1/Json/Tx/Data">>,
                                  #{nl => 0,qos => 1,rap => 0,rh => 0}}]},
                            infinity]}}

Thank you. Have a great Thursday.

SaveMe0 Dec 18, 2024
Author

To rule out the possibility that broker is not receiving data on my machine I have run the subscription in Linux shell successfully.

alberto@debian:~/emqtt/_build/emqtt/rel/emqtt/bin$ ./emqtt sub -h '192.168.178.33' -p '14567' --enable-quic=true -q '1' -k '300' -t 'EK9160/EK-98EFB6/Stream1/Json/Tx/Data'
2024-12-18T18:57:33.500+01:00 Generated clientid: emqtt-debian-5b45c5e5220caf88477e
2024-12-18T18:57:33.703+01:00 Connected:
#{'Maximum-Packet-Size' => 1048576,'Receive-Maximum' => 32,
  'Retain-Available' => 1,'Shared-Subscription-Available' => 1,
  'Subscription-Identifier-Available' => 1,'Topic-Alias-Maximum' => 65535,
  'Wildcard-Subscription-Available' => 1}
2024-12-18T18:57:33.711+01:00 Subscribed to: EK9160/EK-98EFB6/Stream1/Json/Tx/Data
2024-12-18T18:57:33.776+01:00 {"Timestamp":"2024-12-18T17:57:33.340Z","Values":{"Slave 1 (EL1008).Channel 1.Input":true,"Slave 1 (EL1008).Channel 2.Input":false},"MetaData":{}}
2024-12-18T18:57:34.779+01:00 {"Timestamp":"2024-12-18T17:57:34.340Z","Values":{"Slave 1 (EL1008).Channel 1.Input":true,"Slave 1 (EL1008).Channel 2.Input":false},"MetaData":{}}
2024-12-18T18:57:35.776+01:00 {"Timestamp":"2024-12-18T17:57:35.340Z","Values":{"Slave 1 (EL1008).Channel 1.Input":true,"Slave 1 (EL1008).Channel 2.Input":false},"MetaData":{}}

The data is automatically published to the broker received from a slave configured on the EK9160 Bus Coupler.

SaveMe0 · 2024-12-19T10:17:58Z

SaveMe0
Dec 19, 2024
Author

Hi emqtt team,

when call the following functions from my Erlang/OTP com2.erl module I get no errors

{ok, ConnPid} = emqtt:start_link(Options),
emqtt:quic_connect(ConnPid),

However when I call these functions

emqtt:ping(ConnPid),
emqtt:subscribe(ConnPid, Properties, Subscriptions),

my com2.erl process dies.

exception exit: {noproc,{gen_statem,call,[<0.216.0>,ping,infinity]}}

This happens regardless of whether I make a "gen_statem:call/2" or "gen_statem:cast/2".

Here is a snapshot of my module where the aforementioned calls are made

subState({call, From}, sub, Data) ->
    io:format("CommPid ~p~n", [Data#comm.connPid]),
    ConnPid = Data#comm.connPid,
    %Properties = #{},
    emqtt:ping(ConnPid),
    Properties = #{ 'Maximum-Packet-Size' => 1048576,
                    'Receive-Maximum' => 32,
                    'Retain-Available' => 1,
                    'Shared-Subscription-Available' => 1,
                    'Subscription-Identifier-Available' => 1,
                    'Topic-Alias-Maximum' => 65535,
                    'Wildcard-Subscription-Available' => 1},
    SubOpts = [{qos, 1}],
    Subscriptions = [{<<"EK9160/EK-98EFB6/Stream1/Json/Tx/Data">>, SubOpts}],
    {ok, _Props, ReasonCodes} = emqtt:subscribe(ConnPid, Properties, Subscriptions),
    io:format("ReasonCodes~p~n", [ReasonCodes]),
    {keep_state,Data, [{reply, From, Data}]};
subState(EventType, EventContent, Data) ->
    handle_event(EventType, EventContent, Data).

Is there a special way that I need to call these functions from my process so that my com2.erl process does not die? Without the function calls my process does not die.

0 replies

qzhuyan · 2024-12-20T09:57:54Z

qzhuyan
Dec 20, 2024
Collaborator

this is emqtt related, here is emqx, pls create issue in emqx/emqtt

0 replies

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Need help using emqtt with quic over mqtt on Erlang/OTP 27 #14415

{{title}}

Replies: 5 comments 12 replies

{{title}}

{{title}}

{{title}}

{{title}}

{{title}}

{{title}}

{{editor}}'s edit

{{editor}}'s edit

{{title}}

{{title}}

{{title}}

{{title}}

{{title}}

{{title}}

Select a reply

Need help using emqtt with quic over mqtt on Erlang/OTP 27 #14415

SaveMe0 Dec 14, 2024

Replies: 5 comments · 12 replies

ieQu1 Dec 17, 2024 Collaborator

SaveMe0 Dec 18, 2024 Author

ieQu1 Dec 19, 2024 Collaborator

qzhuyan Dec 17, 2024 Collaborator

SaveMe0 Dec 18, 2024 Author

id Dec 18, 2024 Maintainer

SaveMe0 Dec 18, 2024 Author

SaveMe0 Dec 18, 2024 Author

SaveMe0 Dec 18, 2024 Author

SaveMe0 Dec 18, 2024 Author

SaveMe0 Dec 19, 2024 Author

qzhuyan Dec 20, 2024 Collaborator

SaveMe0
Dec 14, 2024

Replies: 5 comments 12 replies

ieQu1
Dec 17, 2024
Collaborator

SaveMe0 Dec 18, 2024
Author

ieQu1 Dec 19, 2024
Collaborator

qzhuyan
Dec 17, 2024
Collaborator

SaveMe0
Dec 18, 2024
Author

id Dec 18, 2024
Maintainer

SaveMe0 Dec 18, 2024
Author

SaveMe0 Dec 18, 2024
Author

SaveMe0 Dec 18, 2024
Author

SaveMe0 Dec 18, 2024
Author

SaveMe0
Dec 19, 2024
Author

qzhuyan
Dec 20, 2024
Collaborator