yuanbiao
/
emqx
1
0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge pull request #12134 from lafirest/fix/gbt 

fix(gbt): Add unimplemented command call
This commit is contained in:
JianBo He 2023-12-11 13:46:07 +08:00 committed by GitHub
parent 107efda78f 1a8b59045b
commit b34e66bdf7
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 18 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
apps/emqx_gateway_gbt32960/src/emqx_gbt32960_channel.erl
View File

@ -51,6 +51,8 @@
inflight :: emqx_inflight:inflight(), inflight :: emqx_inflight:inflight(),
%% Message Queue %% Message Queue
mqueue :: queue:queue(), mqueue :: queue:queue(),
%% Subscriptions
subscriptions :: map(),
retx_interval, retx_interval,
retx_max_times, retx_max_times,
max_mqueue_len max_mqueue_len
@ -115,7 +117,7 @@ stats(#channel{inflight = Inflight, mqueue = Queue}) ->
%% XXX: A fake stats for managed by emqx_management %% XXX: A fake stats for managed by emqx_management
[ [
{subscriptions_cnt, 1}, {subscriptions_cnt, 1},
{subscriptions_max, 0}, {subscriptions_max, 1},
{inflight_cnt, emqx_inflight:size(Inflight)}, {inflight_cnt, emqx_inflight:size(Inflight)},
{inflight_max, emqx_inflight:max_size(Inflight)}, {inflight_max, emqx_inflight:max_size(Inflight)},
{mqueue_len, queue:len(Queue)}, {mqueue_len, queue:len(Queue)},
@ -181,6 +183,7 @@ init(
clientinfo = ClientInfo, clientinfo = ClientInfo,
inflight = emqx_inflight:new(1), inflight = emqx_inflight:new(1),
mqueue = queue:new(), mqueue = queue:new(),
subscriptions = #{},
timers = #{}, timers = #{},
conn_state = idle, conn_state = idle,
retx_interval = RetxInterv, retx_interval = RetxInterv,
@ -370,9 +373,15 @@ handle_call(kick, _From, Channel) ->
disconnect_and_shutdown(kicked, ok, Channel1); disconnect_and_shutdown(kicked, ok, Channel1);
handle_call(discard, _From, Channel) -> handle_call(discard, _From, Channel) ->
disconnect_and_shutdown(discarded, ok, Channel); disconnect_and_shutdown(discarded, ok, Channel);
handle_call({subscribe, _Topic, _SubOpts}, _From, Channel) ->
reply({error, not_support}, Channel);
handle_call({unsubscribe, _Topic}, _From, Channel) ->
reply({error, not_found}, Channel);
handle_call(subscriptions, _From, Channel = #channel{subscriptions = Subscriptions}) ->
reply({ok, maps:to_list(Subscriptions)}, Channel);
handle_call(Req, _From, Channel) -> handle_call(Req, _From, Channel) ->
log(error, #{msg => "unexpected_call", call => Req}, Channel), log(error, #{msg => "unexpected_call", call => Req}, Channel),
reply(ignored, Channel). reply({error, unexpected_call}, Channel).
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
%% Handle cast %% Handle cast
@ -606,8 +615,8 @@ process_connect(
) )
of of
{ok, #{session := Session}} -> {ok, #{session := Session}} ->
NChannel = Channel#channel{session = Session}, NChannel0 = Channel#channel{session = Session},
subscribe_downlink(?DEFAULT_DOWNLINK_TOPIC, Channel), NChannel = subscribe_downlink(?DEFAULT_DOWNLINK_TOPIC, NChannel0),
_ = upstreaming(Frame, NChannel), _ = upstreaming(Frame, NChannel),
%% XXX: connection_accepted is not defined by stomp protocol %% XXX: connection_accepted is not defined by stomp protocol
_ = run_hooks(Ctx, 'client.connack', [ConnInfo, connection_accepted, #{}]), _ = run_hooks(Ctx, 'client.connack', [ConnInfo, connection_accepted, #{}]),
@ -854,11 +863,13 @@ subscribe_downlink(
#{ #{
clientid := ClientId, clientid := ClientId,
mountpoint := Mountpoint mountpoint := Mountpoint
} },
} subscriptions = Subscriptions
} = Channel
) -> ) ->
{ParsedTopic, SubOpts0} = emqx_topic:parse(Topic), {ParsedTopic, SubOpts0} = emqx_topic:parse(Topic),
SubOpts = maps:merge(emqx_gateway_utils:default_subopts(), SubOpts0), SubOpts = maps:merge(emqx_gateway_utils:default_subopts(), SubOpts0),
MountedTopic = emqx_mountpoint:mount(Mountpoint, ParsedTopic), MountedTopic = emqx_mountpoint:mount(Mountpoint, ParsedTopic),
_ = emqx_broker:subscribe(MountedTopic, ClientId, SubOpts), _ = emqx_broker:subscribe(MountedTopic, ClientId, SubOpts),
run_hooks(Ctx, 'session.subscribed', [ClientInfo, MountedTopic, SubOpts]). run_hooks(Ctx, 'session.subscribed', [ClientInfo, MountedTopic, SubOpts]),
Channel#channel{subscriptions = Subscriptions#{MountedTopic => SubOpts}}.