find-shared-subs.erl

%% Copy-paste into remote_console to find topics subscribed by multiple clients.
%% Returns {Results, NextCursor} where Results = [{Topic, [ClientId]}].
%% Usage:
%%   {Results, Cursor} = (paste this expression).
%%   %% Next batch:
%%   {More, Cursor2} = FindSharedSubBatch(Cursor, 100).

FindSharedSubBatch = fun FindSharedSubBatch(Cursor0, Limit) ->
    Key0 = case Cursor0 of
        undefined -> ets:first(emqx_suboption);
        _ -> Cursor0
    end,
    PidToClientId = fun(Pid) ->
        case ets:lookup(emqx_channel_conn, Pid) of
            [Conn] -> element(4, Conn);
            _ -> undefined
        end
    end,
    CollectTopic = fun CollectTopic(_Topic, '$end_of_table', Pids) ->
            {lists:reverse(Pids), '$end_of_table'};
        CollectTopic(Topic, {T, SubPid}, Pids) when T =:= Topic ->
            CollectTopic(Topic, ets:next(emqx_suboption, {T, SubPid}), [SubPid | Pids]);
        CollectTopic(_Topic, NextKey, Pids) ->
            {lists:reverse(Pids), NextKey}
    end,
    F = fun F('$end_of_table', _N, Acc) ->
            {lists:reverse(Acc), '$end_of_table'};
        F(_Key, N, Acc) when N =< 0 ->
            {lists:reverse(Acc), _Key};
        F({Topic, _SubPid} = Key, N, Acc) ->
            {Pids, NextKey} = CollectTopic(Topic, Key, []),
            case Pids of
                [_, _ | _] ->
                    ClientIds = lists:filtermap(
                        fun(Pid) ->
                            case PidToClientId(Pid) of
                                undefined -> false;
                                CId -> {true, CId}
                            end
                        end, Pids),
                    case ClientIds of
                        [_, _ | _] -> F(NextKey, N - 1, [{Topic, ClientIds} | Acc]);
                        _ -> F(NextKey, N, Acc)
                    end;
                _ ->
                    F(NextKey, N, Acc)
            end;
        F(Key, N, Acc) ->
            F(ets:next(emqx_suboption, Key), N, Acc)
    end,
    F(Key0, Limit, [])
end,
{Results, Cursor} = FindSharedSubBatch(undefined, 100),
[io:format("~s~n~s~n", [T, [io_lib:format("  ~s~n", [C]) || C <- Cs]]) || {T, Cs} <- Results],
{length(Results), Cursor}.