From 845c5eddc1c4069adfb04eeeca8ded9622edfcf2 Mon Sep 17 00:00:00 2001 From: Feng Lee Date: Tue, 21 Nov 2017 22:41:02 +0800 Subject: [PATCH] Merge with emqx --- src/emqx_rest_api.erl.bk | 551 --------------------------------------- 1 file changed, 551 deletions(-) delete mode 100644 src/emqx_rest_api.erl.bk diff --git a/src/emqx_rest_api.erl.bk b/src/emqx_rest_api.erl.bk deleted file mode 100644 index f04a7c538..000000000 --- a/src/emqx_rest_api.erl.bk +++ /dev/null @@ -1,551 +0,0 @@ -%%-------------------------------------------------------------------- -%% Copyright (c) 2013-2017 EMQ Enterprise, Inc. (http://emqtt.io) -%% -%% Licensed under the Apache License, Version 2.0 (the "License"); -%% you may not use this file except in compliance with the License. -%% You may obtain a copy of the License at -%% -%% http://www.apache.org/licenses/LICENSE-2.0 -%% -%% Unless required by applicable law or agreed to in writing, software -%% distributed under the License is distributed on an "AS IS" BASIS, -%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -%% See the License for the specific language governing permissions and -%% limitations under the License. -%%-------------------------------------------------------------------- - --module (emqx_rest_api). - --include("emqx.hrl"). - --include("emqx_rest.hrl"). - --http_api({"^nodes/(.+?)/alarms/?$", 'GET', alarm_list, []}). - --http_api({"^nodes/(.+?)/clients/?$", 'GET', client_list, []}). --http_api({"^nodes/(.+?)/clients/(.+?)/?$", 'GET',client_list, []}). --http_api({"^clients/(.+?)/?$", 'GET', client, []}). --http_api({"^clients/(.+?)/?$", 'DELETE', kick_client, []}). --http_api({"^clients/(.+?)/clean_acl_cache?$", 'PUT', clean_acl_cache, [{<<"topic">>, binary}]}). - --http_api({"^routes?$", 'GET', route_list, []}). --http_api({"^routes/(.+?)/?$", 'GET', route, []}). - --http_api({"^nodes/(.+?)/sessions/?$", 'GET', session_list, []}). --http_api({"^nodes/(.+?)/sessions/(.+?)/?$", 'GET', session_list, []}). --http_api({"^sessions/(.+?)/?$", 'GET', session, []}). - --http_api({"^nodes/(.+?)/subscriptions/?$", 'GET', subscription_list, []}). --http_api({"^nodes/(.+?)/subscriptions/(.+?)/?$", 'GET', subscription_list, []}). --http_api({"^subscriptions/(.+?)/?$", 'GET', subscription, []}). - --http_api({"^mqtt/publish?$", 'POST', publish, [{<<"topic">>, binary}]}). --http_api({"^mqtt/subscribe?$", 'POST', subscribe, [{<<"client_id">>, binary},{<<"topic">>, binary}]}). --http_api({"^mqtt/unsubscribe?$", 'POST', unsubscribe, [{<<"client_id">>, binary},{<<"topic">>, binary}]}). - --http_api({"^management/nodes/?$", 'GET', brokers, []}). --http_api({"^management/nodes/(.+?)/?$", 'GET', broker, []}). --http_api({"^monitoring/nodes/?$", 'GET', nodes, []}). --http_api({"^monitoring/nodes/(.+?)/?$", 'GET', node, []}). --http_api({"^monitoring/listeners/?$", 'GET', listeners, []}). --http_api({"^monitoring/listeners/(.+?)/?$", 'GET', listener, []}). --http_api({"^monitoring/metrics/?$", 'GET', metrics, []}). --http_api({"^monitoring/metrics/(.+?)/?$", 'GET', metric, []}). --http_api({"^monitoring/stats/?$", 'GET', stats, []}). --http_api({"^monitoring/stats/(.+?)/?$", 'GET', stat, []}). - --http_api({"^nodes/(.+?)/plugins/?$", 'GET', plugin_list, []}). --http_api({"^nodes/(.+?)/plugins/(.+?)/?$", 'PUT', enabled, [{<<"active">>, bool}]}). - --http_api({"^configs/(.+?)/?$", 'PUT', modify_config, [{<<"key">>, binary}, {<<"value">>, binary}]}). --http_api({"^configs/?$", 'GET', config_list, []}). --http_api({"^nodes/(.+?)/configs/(.+?)/?$", 'PUT', modify_config, [{<<"key">>, binary}, {<<"value">>, binary}]}). --http_api({"^nodes/(.+?)/configs/?$", 'GET', config_list, []}). --http_api({"^nodes/(.+?)/plugin_configs/(.+?)/?$", 'GET', plugin_config_list, []}). --http_api({"^nodes/(.+?)/plugin_configs/(.+?)/?$", 'PUT', modify_plugin_config, []}). - --http_api({"^users/?$", 'GET', users, []}). --http_api({"^users/?$", 'POST', users, [{<<"username">>, binary}, - {<<"password">>, binary}, - {<<"tags">>, binary}]}). --http_api({"^users/(.+?)/?$", 'GET', users, []}). --http_api({"^users/(.+?)/?$", 'PUT', users, [{<<"tags">>, binary}]}). --http_api({"^users/(.+?)/?$", 'DELETE', users, []}). - --http_api({"^auth/?$", 'POST', auth, [{<<"username">>, binary}, {<<"password">>, binary}]}). --http_api({"^change_pwd/(.+?)/?$", 'PUT', change_pwd, [{<<"old_pwd">>, binary}, - {<<"new_pwd">>, binary}]}). - --import(proplists, [get_value/2, get_value/3]). - --export([alarm_list/3]). --export([client/3, client_list/3, client_list/4, kick_client/3, clean_acl_cache/3]). --export([route/3, route_list/2]). --export([session/3, session_list/3, session_list/4]). --export([subscription/3, subscription_list/3, subscription_list/4]). --export([nodes/2, node/3, brokers/2, broker/3, listeners/2, listener/3, metrics/2, metric/3, stats/2, stat/3]). --export([publish/2, subscribe/2, unsubscribe/2]). --export([plugin_list/3, enabled/4]). --export([modify_config/3, modify_config/4, config_list/2, config_list/3, - plugin_config_list/4, modify_plugin_config/4]). - --export([users/2,users/3, auth/2, change_pwd/3]). - -%%-------------------------------------------------------------------------- -%% alarm -%%-------------------------------------------------------------------------- -alarm_list('GET', _Req, _Node) -> - Alarms = emqx_mgmt:alarm_list(), - {ok, lists:map(fun alarm_row/1, Alarms)}. - -alarm_row(#mqtt_alarm{id = AlarmId, - severity = Severity, - title = Title, - summary = Summary, - timestamp = Timestamp}) -> - [{id, AlarmId}, - {severity, Severity}, - {title, l2b(Title)}, - {summary, l2b(Summary)}, - {occurred_at, l2b(strftime(Timestamp))}]. - -%%-------------------------------------------------------------------------- -%% client -%%-------------------------------------------------------------------------- -client('GET', _Params, Key) -> - Data = emqx_mgmt:client(l2b(Key)), - {ok, [{objects, [client_row(Row) || Row <- Data]}]}. - -client_list('GET', Params, Node) -> - {PageNo, PageSize} = page_params(Params), - Data = emqx_mgmt:client_list(l2a(Node), undefined, PageNo, PageSize), - Rows = get_value(result, Data), - TotalPage = get_value(totalPage, Data), - TotalNum = get_value(totalNum, Data), - {ok, [{current_page, PageNo}, - {page_size, PageSize}, - {total_num, TotalNum}, - {total_page, TotalPage}, - {objects, [client_row(Row) || Row <- Rows]}]}. - -client_list('GET', Params, Node, Key) -> - {PageNo, PageSize} = page_params(Params), - Data = emqx_mgmt:client_list(l2a(Node), l2b(Key), PageNo, PageSize), - {ok, [{objects, [client_row(Row) || Row <- Data]}]}. - -kick_client('DELETE', _Params, Key) -> - case emqx_mgmt:kick_client(l2b(Key)) of - true -> {ok, []}; - false -> {error, [{code, ?ERROR12}]} - end. - -clean_acl_cache('PUT', Params, Key0) -> - Topic = get_value(<<"topic">>, Params), - [Key | _] = string:tokens(Key0, "/"), - case emqx_mgmt:clean_acl_cache(l2b(Key), Topic) of - true -> {ok, []}; - false -> {error, [{code, ?ERROR12}]} - end. - -client_row(#mqtt_client{client_id = ClientId, - peername = {IpAddr, Port}, - username = Username, - clean_sess = CleanSess, - proto_ver = ProtoVer, - keepalive = KeepAlvie, - connected_at = ConnectedAt}) -> - [{client_id, ClientId}, - {username, Username}, - {ipaddress, l2b(ntoa(IpAddr))}, - {port, Port}, - {clean_sess, CleanSess}, - {proto_ver, ProtoVer}, - {keepalive, KeepAlvie}, - {connected_at, l2b(strftime(ConnectedAt))}]. - -%%-------------------------------------------------------------------------- -%% route -%%-------------------------------------------------------------------------- -route('GET', _Params, Key) -> - Data = emqx_mgmt:route(l2b(Key)), - {ok, [{objects, [route_row(Row) || Row <- Data]}]}. - -route_list('GET', Params) -> - {PageNo, PageSize} = page_params(Params), - Data = emqx_mgmt:route_list(undefined, PageNo, PageSize), - Rows = get_value(result, Data), - TotalPage = get_value(totalPage, Data), - TotalNum = get_value(totalNum, Data), - {ok, [{current_page, PageNo}, - {page_size, PageSize}, - {total_num, TotalNum}, - {total_page, TotalPage}, - {objects, [route_row(Row) || Row <- Rows]}]}. - -route_row(Route) when is_record(Route, mqtt_route) -> - [{topic, Route#mqtt_route.topic}, {node, Route#mqtt_route.node}]; - -route_row({Topic, Node}) -> - [{topic, Topic}, {node, Node}]. - -%%-------------------------------------------------------------------------- -%% session -%%-------------------------------------------------------------------------- -session('GET', _Params, Key) -> - Data = emqx_mgmt:session(l2b(Key)), - {ok, [{objects, [session_row(Row) || Row <- Data]}]}. - -session_list('GET', Params, Node) -> - {PageNo, PageSize} = page_params(Params), - Data = emqx_mgmt:session_list(l2a(Node), undefined, PageNo, PageSize), - Rows = get_value(result, Data), - TotalPage = get_value(totalPage, Data), - TotalNum = get_value(totalNum, Data), - {ok, [{current_page, PageNo}, - {page_size, PageSize}, - {total_num, TotalNum}, - {total_page, TotalPage}, - {objects, [session_row(Row) || Row <- Rows]}]}. - -session_list('GET', Params, Node, ClientId) -> - {PageNo, PageSize} = page_params(Params), - Data = emqx_mgmt:session_list(l2a(Node), l2b(ClientId), PageNo, PageSize), - {ok, [{objects, [session_row(Row) || Row <- Data]}]}. - -session_row({ClientId, _Pid, _Persistent, Session}) -> - Data = lists:append(Session, emqttd_stats:get_session_stats(ClientId)), - InfoKeys = [clean_sess, subscriptions, max_inflight, inflight_len, mqueue_len, - mqueue_dropped, awaiting_rel_len, deliver_msg,enqueue_msg, created_at], - [{client_id, ClientId} | [{Key, format(Key, get_value(Key, Data))} || Key <- InfoKeys]]. - -%%-------------------------------------------------------------------------- -%% subscription -%%-------------------------------------------------------------------------- -subscription('GET', _Params, Key) -> - Data = emqx_mgmt:subscription(l2b(Key)), - {ok, [{objects, [subscription_row(Row) || Row <- Data]}]}. - -subscription_list('GET', Params, Node) -> - {PageNo, PageSize} = page_params(Params), - Data = emqx_mgmt:subscription_list(l2a(Node), undefined, PageNo, PageSize), - Rows = get_value(result, Data), - TotalPage = get_value(totalPage, Data), - TotalNum = get_value(totalNum, Data), - {ok, [{current_page, PageNo}, - {page_size, PageSize}, - {total_num, TotalNum}, - {total_page, TotalPage}, - {objects, [subscription_row(Row) || Row <- Rows]}]}. - -subscription_list('GET', Params, Node, Key) -> - {PageNo, PageSize} = page_params(Params), - Data = emqx_mgmt:subscription_list(l2a(Node), l2b(Key), PageNo, PageSize), - {ok, [{objects, [subscription_row(Row) || Row <- Data]}]}. - -subscription_row({{Topic, SubPid}, Options}) when is_pid(SubPid) -> - subscription_row({{Topic, {undefined, SubPid}}, Options}); -subscription_row({{Topic, {SubId, SubPid}}, Options}) -> - Qos = proplists:get_value(qos, Options), - ClientId = case SubId of - undefined -> list_to_binary(pid_to_list(SubPid)); - SubId -> SubId - end, - [{client_id, ClientId}, {topic, Topic}, {qos, Qos}]. - -%%-------------------------------------------------------------------------- -%% management/monitoring -%%-------------------------------------------------------------------------- -nodes('GET', _Params) -> - Data = emqx_mgmt:nodes_info(), - {ok, Data}. - -node('GET', _Params, Node) -> - Data = emqx_mgmt:node_info(l2a(Node)), - {ok, Data}. - -brokers('GET', _Params) -> - Data = emqx_mgmt:brokers(), - {ok, [format_broker(Node, Broker) || {Node, Broker} <- Data]}. - -broker('GET', _Params, Node) -> - Data = emqx_mgmt:broker(l2a(Node)), - {ok, format_broker(Data)}. - -listeners('GET', _Params) -> - Data = emqx_mgmt:listeners(), - {ok, [{Node, format_listeners(Listeners, [])} || {Node, Listeners} <- Data]}. - -listener('GET', _Params, Node) -> - Data = emqx_mgmt:listener(l2a(Node)), - {ok, [format_listener(Listeners) || Listeners <- Data]}. - -metrics('GET', _Params) -> - Data = emqx_mgmt:metrics(), - {ok, Data}. - -metric('GET', _Params, Node) -> - Data = emqx_mgmt:metrics(l2a(Node)), - {ok, Data}. - -stats('GET', _Params) -> - Data = emqx_mgmt:stats(), - {ok, [Data]}. - -stat('GET', _Params, Node) -> - Data = emqx_mgmt:stats(l2a(Node)), - {ok, Data}. - -format_broker(Node, Broker) -> - OtpRel = "R" ++ erlang:system_info(otp_release) ++ "/" ++ erlang:system_info(version), - [{name, Node}, - {version, bin(get_value(version, Broker))}, - {sysdescr, bin(get_value(sysdescr, Broker))}, - {uptime, bin(get_value(uptime, Broker))}, - {datetime, bin(get_value(datetime, Broker))}, - {otp_release, l2b(OtpRel)}, - {node_status, 'Running'}]. - -format_broker(Broker) -> - OtpRel = "R" ++ erlang:system_info(otp_release) ++ "/" ++ erlang:system_info(version), - [{version, bin(get_value(version, Broker))}, - {sysdescr, bin(get_value(sysdescr, Broker))}, - {uptime, bin(get_value(uptime, Broker))}, - {datetime, bin(get_value(datetime, Broker))}, - {otp_release, l2b(OtpRel)}, - {node_status, 'Running'}]. - -format_listeners([], Acc) -> - Acc; -format_listeners([{Protocol, ListenOn, Info}| Listeners], Acc) -> - format_listeners(Listeners, [format_listener({Protocol, ListenOn, Info}) | Acc]). - -format_listener({Protocol, ListenOn, Info}) -> - Listen = l2b(esockd:to_string(ListenOn)), - lists:append([{protocol, Protocol}, {listen, Listen}], Info). - -%%-------------------------------------------------------------------------- -%% mqtt -%%-------------------------------------------------------------------------- -publish('POST', Params) -> - Topic = get_value(<<"topic">>, Params), - ClientId = get_value(<<"client_id">>, Params, http), - Payload = get_value(<<"payload">>, Params, <<>>), - Qos = get_value(<<"qos">>, Params, 0), - Retain = get_value(<<"retain">>, Params, false), - case emqx_mgmt:publish({ClientId, Topic, Payload, Qos, Retain}) of - ok -> - {ok, []}; - {error, Error} -> - {error, [{code, ?ERROR2}, {message, Error}]} - end. - -subscribe('POST', Params) -> - ClientId = get_value(<<"client_id">>, Params), - Topic = get_value(<<"topic">>, Params), - Qos = get_value(<<"qos">>, Params, 0), - case emqx_mgmt:subscribe({ClientId, Topic, Qos}) of - ok -> - {ok, []}; - {error, Error} -> - {error, [{code, ?ERROR2}, {message, Error}]} - end. - -unsubscribe('POST', Params) -> - ClientId = get_value(<<"client_id">>, Params), - Topic = get_value(<<"topic">>, Params), - case emqx_mgmt:unsubscribe({ClientId, Topic})of - ok -> - {ok, []}; - {error, Error} -> - {error, [{code, ?ERROR2}, {message, Error}]} - end. - -%%-------------------------------------------------------------------------- -%% plugins -%%-------------------------------------------------------------------------- -plugin_list('GET', _Params, Node) -> - Plugins = lists:map(fun plugin/1, emqx_mgmt:plugin_list(l2a(Node))), - {ok, Plugins}. - -enabled('PUT', Params, Node, PluginName) -> - Active = get_value(<<"active">>, Params), - case Active of - true -> - return(emqx_mgmt:plugin_load(l2a(Node), l2a(PluginName))); - false -> - return(emqx_mgmt:plugin_unload(l2a(Node), l2a(PluginName))) - end. - -return(Result) -> - case Result of - ok -> - {ok, []}; - {ok, _} -> - {ok, []}; - {error, already_started} -> - {error, [{code, ?ERROR10}, {message, <<"already_started">>}]}; - {error, not_started} -> - {error, [{code, ?ERROR11}, {message, <<"not_started">>}]}; - Error -> - lager:error("error:~p", [Error]), - {error, [{code, ?ERROR2}, {message, <<"unknown">>}]} - end. -plugin(#mqtt_plugin{name = Name, version = Ver, descr = Descr, - active = Active}) -> - [{name, Name}, - {version, iolist_to_binary(Ver)}, - {description, iolist_to_binary(Descr)}, - {active, Active}]. - -%%-------------------------------------------------------------------------- -%% modify config -%%-------------------------------------------------------------------------- -modify_config('PUT', Params, App) -> - Key = get_value(<<"key">>, Params, <<"">>), - Value = get_value(<<"value">>, Params, <<"">>), - case emqx_mgmt:modify_config(l2a(App), b2l(Key), b2l(Value)) of - true -> {ok, []}; - false -> {error, [{code, ?ERROR2}]} - end. - -modify_config('PUT', Params, Node, App) -> - Key = get_value(<<"key">>, Params, <<"">>), - Value = get_value(<<"value">>, Params, <<"">>), - case emqx_mgmt:modify_config(l2a(Node), l2a(App), b2l(Key), b2l(Value)) of - ok -> {ok, []}; - _ -> {error, [{code, ?ERROR2}]} - end. - -config_list('GET', _Params) -> - Data = emqx_mgmt:get_configs(), - {ok, [{Node, format_config(Config, [])} || {Node, Config} <- Data]}. - -config_list('GET', _Params, Node) -> - Data = emqx_mgmt:get_config(l2a(Node)), - {ok, [format_config(Config) || Config <- lists:reverse(Data)]}. - -plugin_config_list('GET', _Params, Node, App) -> - {ok, Data} = emqx_mgmt:get_plugin_config(l2a(Node), l2a(App)), - {ok, [format_plugin_config(Config) || Config <- lists:reverse(Data)]}. - -modify_plugin_config('PUT', Params, Node, App) -> - PluginName = l2a(App), - case emqx_mgmt:modify_plugin_config(l2a(Node), PluginName, Params) of - ok -> - Plugins = emqttd_plugins:list(), - {_, _, _, _, Status} = lists:keyfind(PluginName, 2, Plugins), - case Status of - true -> - emqttd_plugins:unload(PluginName), - timer:sleep(500), - emqttd_plugins:load(PluginName), - {ok, []}; - false -> - {ok, []} - end; - _ -> - {error, [{code, ?ERROR2}]} - end. - - -format_config([], Acc) -> - Acc; -format_config([{Key, Value, Datatpye, App}| Configs], Acc) -> - format_config(Configs, [format_config({Key, Value, Datatpye, App}) | Acc]). - -format_config({Key, Value, Datatpye, App}) -> - [{<<"key">>, l2b(Key)}, - {<<"value">>, l2b(Value)}, - {<<"datatpye">>, l2b(Datatpye)}, - {<<"app">>, App}]. - -format_plugin_config({Key, Value, Desc, Required}) -> - [{<<"key">>, l2b(Key)}, - {<<"value">>, l2b(Value)}, - {<<"desc">>, l2b(Desc)}, - {<<"required">>, Required}]. - -%%-------------------------------------------------------------------------- -%% Admin -%%-------------------------------------------------------------------------- -auth('POST', Params) -> - Username = get_value(<<"username">>, Params), - Password = get_value(<<"password">>, Params), - case emqx_mgmt:check_user(Username, Password) of - ok -> - {ok, []}; - {error, Reason} -> - {error, [{code, ?ERROR3}, {message, list_to_binary(Reason)}]} - end. - -users('POST', Params) -> - Username = get_value(<<"username">>, Params), - Password = get_value(<<"password">>, Params), - Tag = get_value(<<"tags">>, Params), - code(emqx_mgmt:add_user(Username, Password, Tag)); - -users('GET', _Params) -> - {ok, [Admin || Admin <- emqx_mgmt:user_list()]}. - -users('GET', _Params, Username) -> - {ok, emqx_mgmt:lookup_user(list_to_binary(Username))}; - -users('PUT', Params, Username) -> - code(emqx_mgmt:update_user(list_to_binary(Username), Params)); - -users('DELETE', _Params, "admin") -> - {error, [{code, ?ERROR6}, {message, <<"admin cannot be deleted">>}]}; -users('DELETE', _Params, Username) -> - code(emqx_mgmt:remove_user(list_to_binary(Username))). - -change_pwd('PUT', Params, Username) -> - OldPwd = get_value(<<"old_pwd">>, Params), - NewPwd = get_value(<<"new_pwd">>, Params), - code(emqx_mgmt:change_password(list_to_binary(Username), OldPwd, NewPwd)). - -code(ok) -> {ok, []}; -code(error) -> {error, [{code, ?ERROR2}]}; -code({error, Error}) -> {error, Error}. -%%-------------------------------------------------------------------------- -%% Inner function -%%-------------------------------------------------------------------------- -format(created_at, Val) -> - l2b(strftime(Val)); -format(_, Val) -> - Val. - -ntoa({0,0,0,0,0,16#ffff,AB,CD}) -> - inet_parse:ntoa({AB bsr 8, AB rem 256, CD bsr 8, CD rem 256}); -ntoa(IP) -> - inet_parse:ntoa(IP). - -%%-------------------------------------------------------------------- -%% Strftime -%%-------------------------------------------------------------------- -strftime({MegaSecs, Secs, _MicroSecs}) -> - strftime(datetime(MegaSecs * 1000000 + Secs)); - -strftime({{Y,M,D}, {H,MM,S}}) -> - lists:flatten( - io_lib:format( - "~4..0w-~2..0w-~2..0w ~2..0w:~2..0w:~2..0w", [Y, M, D, H, MM, S])). - -datetime(Timestamp) when is_integer(Timestamp) -> - Universal = calendar:gregorian_seconds_to_datetime(Timestamp + - calendar:datetime_to_gregorian_seconds({{1970,1,1}, {0,0,0}})), - calendar:universal_time_to_local_time(Universal). - -bin(S) when is_list(S) -> l2b(S); -bin(A) when is_atom(A) -> bin(atom_to_list(A)); -bin(B) when is_binary(B) -> B; -bin(undefined) -> <<>>. -int(L) -> list_to_integer(L). -l2a(L) -> list_to_atom(L). -l2b(L) -> list_to_binary(L). -b2l(B) -> binary_to_list(B). - - -page_params(Params) -> - PageNo = int(get_value("curr_page", Params, "1")), - PageSize = int(get_value("page_size", Params, "20")), - {PageNo, PageSize}.