fix(gw): enrich conninfo for coap&lwm2m

This commit is contained in:
JianBo He 2022-03-21 12:01:22 +08:00
parent edb1460b56
commit 5682dcb72e
3 changed files with 48 additions and 14 deletions

View File

@ -119,10 +119,10 @@ stats(_) ->
[]. [].
-spec init(map(), map()) -> channel(). -spec init(map(), map()) -> channel().
init(ConnInfoT = #{peername := {PeerHost, _}, init(ConnInfo = #{peername := {PeerHost, _},
sockname := {_, SockPort}}, sockname := {_, SockPort}},
#{ctx := Ctx} = Config) -> #{ctx := Ctx} = Config) ->
Peercert = maps:get(peercert, ConnInfoT, undefined), Peercert = maps:get(peercert, ConnInfo, undefined),
Mountpoint = maps:get(mountpoint, Config, <<>>), Mountpoint = maps:get(mountpoint, Config, <<>>),
ListenerId = case maps:get(listener, Config, undefined) of ListenerId = case maps:get(listener, Config, undefined) of
undefined -> undefined; undefined -> undefined;
@ -144,10 +144,6 @@ init(ConnInfoT = #{peername := {PeerHost, _},
} }
), ),
%% because it is possible to disconnect after init, and then trigger the
%% $event.disconnected hook and these two fields are required in the hook
ConnInfo = ConnInfoT#{proto_name => <<"CoAP">>, proto_ver => <<"1">>},
Heartbeat = ?GET_IDLE_TIME(Config), Heartbeat = ?GET_IDLE_TIME(Config),
#channel{ ctx = Ctx #channel{ ctx = Ctx
, conninfo = ConnInfo , conninfo = ConnInfo
@ -405,6 +401,25 @@ run_conn_hooks(Input, Channel = #channel{ctx = Ctx,
{ok, Input, Channel} {ok, Input, Channel}
end. end.
enrich_conninfo({Queries, _Msg},
Channel = #channel{
keepalive = KeepAlive,
conninfo = ConnInfo}) ->
case Queries of
#{<<"clientid">> := ClientId} ->
Interval = maps:get(interval, emqx_keepalive:info(KeepAlive)),
NConnInfo = ConnInfo#{ clientid => ClientId
, proto_name => <<"CoAP">>
, proto_ver => <<"1">>
, clean_start => true
, keepalive => Interval
, expiry_interval => 0
},
{ok, Channel#channel{conninfo = NConnInfo}};
_ ->
{error, "invalid queries", Channel}
end.
enrich_clientinfo({Queries, Msg}, enrich_clientinfo({Queries, Msg},
Channel = #channel{clientinfo = ClientInfo0}) -> Channel = #channel{clientinfo = ClientInfo0}) ->
case Queries of case Queries of
@ -583,7 +598,8 @@ process_nothing(_, _, Channel) ->
process_connection({open, Req}, Result, Channel, Iter) -> process_connection({open, Req}, Result, Channel, Iter) ->
Queries = emqx_coap_message:get_option(uri_query, Req), Queries = emqx_coap_message:get_option(uri_query, Req),
case emqx_misc:pipeline( case emqx_misc:pipeline(
[ fun run_conn_hooks/2 [ fun enrich_conninfo/2
, fun run_conn_hooks/2
, fun enrich_clientinfo/2 , fun enrich_clientinfo/2
, fun set_log_meta/2 , fun set_log_meta/2
, fun auth_connect/2 , fun auth_connect/2

View File

@ -108,10 +108,10 @@ info(ctx, #channel{ctx = Ctx}) ->
stats(_) -> stats(_) ->
[]. [].
init(ConnInfoT = #{peername := {PeerHost, _}, init(ConnInfo = #{peername := {PeerHost, _},
sockname := {_, SockPort}}, sockname := {_, SockPort}},
#{ctx := Ctx} = Config) -> #{ctx := Ctx} = Config) ->
Peercert = maps:get(peercert, ConnInfoT, undefined), Peercert = maps:get(peercert, ConnInfo, undefined),
Mountpoint = maps:get(mountpoint, Config, undefined), Mountpoint = maps:get(mountpoint, Config, undefined),
ListenerId = case maps:get(listener, Config, undefined) of ListenerId = case maps:get(listener, Config, undefined) of
undefined -> undefined; undefined -> undefined;
@ -133,8 +133,6 @@ init(ConnInfoT = #{peername := {PeerHost, _},
} }
), ),
ConnInfo = ConnInfoT#{proto_name => <<"LwM2M">>, proto_ver => <<"0.0">>},
#channel{ ctx = Ctx #channel{ ctx = Ctx
, conninfo = ConnInfo , conninfo = ConnInfo
, clientinfo = ClientInfo , clientinfo = ClientInfo
@ -369,6 +367,7 @@ do_takeover(_DesireId, Msg, Channel) ->
do_connect(Req, Result, Channel, Iter) -> do_connect(Req, Result, Channel, Iter) ->
case emqx_misc:pipeline( case emqx_misc:pipeline(
[ fun check_lwm2m_version/2 [ fun check_lwm2m_version/2
, fun enrich_conninfo/2
, fun run_conn_hooks/2 , fun run_conn_hooks/2
, fun enrich_clientinfo/2 , fun enrich_clientinfo/2
, fun set_log_meta/2 , fun set_log_meta/2
@ -427,6 +426,25 @@ run_conn_hooks(Input, Channel = #channel{ctx = Ctx,
{ok, Input, Channel} {ok, Input, Channel}
end. end.
enrich_conninfo(#coap_message{options = Options},
Channel = #channel{
conninfo = ConnInfo}) ->
Query = maps:get(uri_query, Options, #{}),
case Query of
#{<<"ep">> := Epn, <<"lt">> := Lifetime} ->
ClientId = maps:get(<<"device_id">>, Query, Epn),
NConnInfo = ConnInfo#{ clientid => ClientId
, proto_name => <<"LwM2M">>
, proto_ver => <<"1.0.1">>
, clean_start => true
, keepalive => binary_to_integer(Lifetime)
, expiry_interval => 0
},
{ok, Channel#channel{conninfo = NConnInfo}};
_ ->
{error, "invalid queries", Channel}
end.
enrich_clientinfo(#coap_message{options = Options} = Msg, enrich_clientinfo(#coap_message{options = Options} = Msg,
Channel = #channel{clientinfo = ClientInfo0}) -> Channel = #channel{clientinfo = ClientInfo0}) ->
Query = maps:get(uri_query, Options, #{}), Query = maps:get(uri_query, Options, #{}),

View File

@ -185,7 +185,7 @@ eventmsg_connected(_ClientInfo = #{
is_bridge := IsBridge, is_bridge := IsBridge,
mountpoint := Mountpoint mountpoint := Mountpoint
}, },
_ConnInfo = #{ ConnInfo = #{
peername := PeerName, peername := PeerName,
sockname := SockName, sockname := SockName,
clean_start := CleanStart, clean_start := CleanStart,