From 5b949fda2c451aaf23f7efafda7e185b42440b7b Mon Sep 17 00:00:00 2001
From: zhouzb <zhouzb@emqx.io>
Date: Sat, 7 Dec 2019 13:31:33 +0800
Subject: [PATCH] Add emqx_cm:kick_session/1

---
 src/emqx_cm.erl | 26 ++++++++++++++++++++++++++
 1 file changed, 26 insertions(+)

diff --git a/src/emqx_cm.erl b/src/emqx_cm.erl
index 72472a0a9..1c9e15968 100644
--- a/src/emqx_cm.erl
+++ b/src/emqx_cm.erl
@@ -48,6 +48,7 @@
 -export([ open_session/3
         , discard_session/1
         , takeover_session/1
+        , kick_session/1
         ]).
 
 -export([ lookup_channels/1
@@ -279,6 +280,31 @@ discard_session(ClientId, ChanPid) when node(ChanPid) == node() ->
 discard_session(ClientId, ChanPid) ->
     rpc_call(node(ChanPid), discard_session, [ClientId, ChanPid]).
 
+kick_session(ClientId) ->
+    case lookup_channels(ClientId) of
+        [] -> {error, not_found};
+        [ChanPid] ->
+            kick_session(ClientId, ChanPid);
+        ChanPids ->
+            [ChanPid|StalePids] = lists:reverse(ChanPids),
+            ?LOG(error, "More than one channel found: ~p", [ChanPids]),
+            lists:foreach(fun(StalePid) ->
+                                  catch discard_session(ClientId, StalePid)
+                          end, StalePids),
+            kick_session(ClientId, ChanPid)
+    end.
+
+kick_session(ClientId, ChanPid) when node(ChanPid) == node() ->
+    case get_chan_info(ClientId, ChanPid) of
+        #{conninfo := #{conn_mod := ConnMod}} ->
+            ConnMod:call(ChanPid, kick);
+        undefined ->
+            {error, not_found}
+    end;
+
+kick_session(ClientId, ChanPid) ->
+    rpc_call(node(ChanPid), kick_session, [ClientId, ChanPid]).
+
 %% @doc Is clean start?
 % is_clean_start(#{clean_start := false}) -> false;
 % is_clean_start(_Attrs) -> true.