anlicheng/iot_cloud
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix service config

Browse Source
This commit is contained in:
anlicheng 2025-08-16 23:50:51 +08:00
parent cd5f6480cd
commit 4495fd0774
7 changed files with 202 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
apps/iot/src/database/micro_service_bo.erl Normal file
View File

@ -0,0 +1,16 @@
%%%-------------------------------------------------------------------
%%% @author aresei
%%% @copyright (C) 2023, <COMPANY>
%%% @doc
%%%
%%% @end
%%% Created : 16. 5月 2023 12:48
%%%-------------------------------------------------------------------
-module(micro_service_bo).
-author("aresei").
-export([get_service_config/1]).
%% API
-spec get_service_config(ServiceId :: binary()) -> {ok, DeviceInfo :: map()} | undefined.
get_service_config(ServiceId) when is_binary(ServiceId) ->
mysql_pool:get_row(mysql_iot, <<"SELECT * FROM micro_service WHERE id = ? LIMIT 1">>, [ServiceId]).

9
apps/iot/src/endpoint/endpoint_mysql.erl
View File

@ -114,11 +114,11 @@ handle_info({timeout, _, create_postman}, State = #state{status = ?DISCONNECTED,
NBuffer = endpoint_buffer:trigger_n(Buffer),
{noreply, State#state{pool_pid = PoolPid, buffer = NBuffer, status = ?CONNECTED}};
ignore ->
erlang:start_timer(?RETRY_INTERVAL, self(), create_postman),
retry_connect(),
{noreply, State};
{error, Reason} ->
lager:warning("[mqtt_postman] start connect pool, get error: ~p", [Reason]),
erlang:start_timer(?RETRY_INTERVAL, self(), create_postman),
retry_connect(),
{noreply, State}
end;
@ -142,7 +142,7 @@ handle_info({next_data, Id, ServiceId, Format, Metric}, State = #state{status =
%% postman进程挂掉时，重新建立新的
handle_info({'EXIT', PoolPid, Reason}, State = #state{endpoint = #endpoint{title = Title}, pool_pid = PoolPid}) ->
lager:warning("[enpoint_mqtt] endpoint: ~p, conn pid exit with reason: ~p", [Title, Reason]),
erlang:start_timer(?RETRY_INTERVAL, self(), create_postman),
retry_connect(),
{noreply, disconnected, State#state{pool_pid = undefined, status = ?DISCONNECTED}};
handle_info(Info, State = #state{status = Status}) ->
@ -173,6 +173,9 @@ code_change(_OldVsn, State = #state{}, _Extra) ->
%%% Internal functions
%%%===================================================================
retry_connect() ->
erlang:start_timer(?RETRY_INTERVAL, self(), create_postman).
-spec insert_sql(Table :: binary(), ServiceId :: binary(), Format :: binary(), FieldsMap :: map(), Metric :: binary()) -> {ok, Sql :: binary(), Values :: list()}.
insert_sql(Table, ServiceId, <<"line">>, FieldsMap, Metric) when is_binary(Table), is_binary(ServiceId), is_binary(Metric) ->
case line_format:parse(Metric) of

30
apps/iot/src/http_handlers/service_handler.erl
View File

@ -15,7 +15,7 @@
%% 下发config.json, 微服务接受后，保存服务配置
handle_request("POST", "/service/push_config", _,
#{<<"uuid">> := UUID, <<"service_id">> := ServiceId, <<"last_edit_user">> := LastEditUser, <<"config_json">> := ConfigJson, <<"timeout">> := Timeout0})
#{<<"uuid">> := UUID, <<"service_id">> := ServiceId, <<"config_json">> := ConfigJson, <<"timeout">> := Timeout0})
when is_binary(UUID), is_binary(ServiceId), is_binary(ConfigJson), is_integer(Timeout0) ->
%% 检查ConfigJson是否是合法的json字符串
@ -30,14 +30,7 @@ handle_request("POST", "/service/push_config", _,
{ok, Ref} ->
case iot_host:await_reply(Ref, Timeout) of
{ok, Result} ->
%% 更新配置信息到数据库
case service_config_model:update(ServiceId, UUID, ConfigJson, LastEditUser) of
ok ->
{ok, 200, iot_util:json_data(Result)};
{error, Reason} ->
lager:debug("[service_handler] set_config service_id: ~p, get error: ~p", [ServiceId, Reason]),
{ok, 200, iot_util:json_error(-1, <<"set service config failed">>)}
end;
{ok, 200, iot_util:json_data(Result)};
{error, Reason} ->
{ok, 200, iot_util:json_error(-1, Reason)}
end;
@ -49,25 +42,6 @@ handle_request("POST", "/service/push_config", _,
{ok, 200, iot_util:json_error(-1, <<"config is invalid json">>)}
end;
%% 获取服务配置信息
handle_request("GET", "/service/get_config", #{<<"service_id">> := ServiceId}, _) when is_binary(ServiceId) ->
case service_config_model:get_config(ServiceId) of
error ->
{ok, 200, iot_util:json_error(-1, <<"service config not found">>)};
{ok, Config} ->
{ok, 200, iot_util:json_data(service_config_model:as_map(Config))}
end;
%% 删除对应的主机信息
handle_request("POST", "/service/delete_config", _, #{<<"service_id">> := ServiceId}) when is_binary(ServiceId) ->
case service_config_model:delete(ServiceId) of
ok ->
{ok, 200, iot_util:json_data(<<"success">>)};
{error, Reason} ->
lager:debug("[service_handler] delete config of service_id: ~p, error: ~p", [ServiceId, Reason]),
{ok, 200, iot_util:json_error(-1, <<"delete service config errror">>)}
end;
%% 部署微服务
handle_request("POST", "/service/deploy", _, #{<<"uuid">> := UUID, <<"task_id">> := TaskId, <<"service_id">> := ServiceId, <<"tar_url">> := TarUrl})
when is_binary(UUID), is_integer(TaskId), is_binary(ServiceId), is_binary(TarUrl) ->

1
apps/iot/src/iot_app.erl
View File

@ -36,7 +36,6 @@ start_mnesia() ->
Tables = mnesia:system_info(tables),
lager:debug("[iot_app] tables: ~p", [Tables]),
%% 创建数据库表
not lists:member(service_config, Tables) andalso service_config_model:create_table(),
ok.
-spec ensure_mnesia_schema() -> any().

4
apps/iot/src/tcp/tcp_channel.erl
View File

@ -140,10 +140,10 @@ handle_info({tcp, Socket, <<?PACKET_REQUEST, PacketId:32, ?METHOD_AUTH:8, AuthRe
%% 请求微服务配置
handle_info({tcp, Socket, <<?PACKET_REQUEST, PacketId:32, ?METHOD_REQUEST_SERVICE_CONFIG:8, ServiceId/binary>>}, State = #state{transport = Transport, socket = Socket}) ->
lager:debug("[ws_channel] service_config request service_id: ~p", [ServiceId]),
case service_config_model:get_config(ServiceId) of
case micro_service_bo:get_service_config(ServiceId) of
error ->
Transport:send(Socket, <<?PACKET_RESPONSE, PacketId:32>>);
{ok, #service_config{config_json = ConfigJson}} when is_binary(ConfigJson) ->
{ok, ConfigJson} when is_binary(ConfigJson) ->
Transport:send(Socket, <<?PACKET_RESPONSE, PacketId:32, ConfigJson/binary>>)
end,
{noreply, State};

0
apps/iot/src/mnesia/service_config_model.erl → backup/service_config_model.erl
View File

176
backup/service_handler.erl Normal file
View File

@ -0,0 +1,176 @@
%%%-------------------------------------------------------------------
%%% @author licheng5
%%% @copyright (C) 2020, <COMPANY>
%%% @doc
%%%
%%% @end
%%% Created : 26. 4月 2020 3:36 下午
%%%-------------------------------------------------------------------
-module(service_handler).
-author("licheng5").
-include("iot.hrl").
%% API
-export([handle_request/4]).
%% 下发config.json, 微服务接受后，保存服务配置
handle_request("POST", "/service/push_config", _,
#{<<"uuid">> := UUID, <<"service_id">> := ServiceId, <<"last_edit_user">> := LastEditUser, <<"config_json">> := ConfigJson, <<"timeout">> := Timeout0})
when is_binary(UUID), is_binary(ServiceId), is_binary(ConfigJson), is_integer(Timeout0) ->
%% 检查ConfigJson是否是合法的json字符串
case iot_util:is_json(ConfigJson) of
true ->
case iot_host:get_pid(UUID) of
undefined ->
{ok, 200, iot_util:json_error(-1, <<"host not found">>)};
Pid when is_pid(Pid) ->
Timeout = Timeout0 * 1000,
case iot_host:async_service_config(Pid, ServiceId, ConfigJson, Timeout) of
{ok, Ref} ->
case iot_host:await_reply(Ref, Timeout) of
{ok, Result} ->
%% 更新配置信息到数据库
case service_config_model:update(ServiceId, UUID, ConfigJson, LastEditUser) of
ok ->
{ok, 200, iot_util:json_data(Result)};
{error, Reason} ->
lager:debug("[service_handler] set_config service_id: ~p, get error: ~p", [ServiceId, Reason]),
{ok, 200, iot_util:json_error(-1, <<"set service config failed">>)}
end;
{error, Reason} ->
{ok, 200, iot_util:json_error(-1, Reason)}
end;
{error, Reason} when is_binary(Reason) ->
{ok, 200, iot_util:json_error(-1, Reason)}
end
end;
false ->
{ok, 200, iot_util:json_error(-1, <<"config is invalid json">>)}
end;
%% 获取服务配置信息
handle_request("GET", "/service/get_config", #{<<"service_id">> := ServiceId}, _) when is_binary(ServiceId) ->
case service_config_model:get_config(ServiceId) of
error ->
{ok, 200, iot_util:json_error(-1, <<"service config not found">>)};
{ok, Config} ->
{ok, 200, iot_util:json_data(service_config_model:as_map(Config))}
end;
%% 删除对应的主机信息
handle_request("POST", "/service/delete_config", _, #{<<"service_id">> := ServiceId}) when is_binary(ServiceId) ->
case service_config_model:delete(ServiceId) of
ok ->
{ok, 200, iot_util:json_data(<<"success">>)};
{error, Reason} ->
lager:debug("[service_handler] delete config of service_id: ~p, error: ~p", [ServiceId, Reason]),
{ok, 200, iot_util:json_error(-1, <<"delete service config errror">>)}
end;
%% 部署微服务
handle_request("POST", "/service/deploy", _, #{<<"uuid">> := UUID, <<"task_id">> := TaskId, <<"service_id">> := ServiceId, <<"tar_url">> := TarUrl})
when is_binary(UUID), is_integer(TaskId), is_binary(ServiceId), is_binary(TarUrl) ->
case iot_host:get_pid(UUID) of
undefined ->
{ok, 200, iot_util:json_error(404, <<"host not found">>)};
Pid when is_pid(Pid) ->
case iot_host:deploy_service(Pid, TaskId, ServiceId, TarUrl) of
{ok, Ref} ->
case iot_host:await_reply(Ref, 5000) of
{ok, Result} ->
{ok, 200, iot_util:json_data(Result)};
{error, Reason} ->
{ok, 200, iot_util:json_error(400, Reason)}
end;
{error, Reason} when is_binary(Reason) ->
{ok, 200, iot_util:json_error(400, Reason)}
end
end;
%% 启动服务
handle_request("POST", "/service/start", _, #{<<"uuid">> := UUID, <<"service_id">> := ServiceId}) when is_binary(UUID), is_binary(ServiceId) ->
case iot_host:get_pid(UUID) of
undefined ->
{ok, 200, iot_util:json_error(404, <<"host not found">>)};
Pid when is_pid(Pid) ->
case iot_host:start_service(Pid, ServiceId) of
{ok, Ref} ->
case iot_host:await_reply(Ref, 5000) of
{ok, Result} ->
{ok, 200, iot_util:json_data(Result)};
{error, Reason} ->
{ok, 200, iot_util:json_error(400, Reason)}
end;
{error, Reason} when is_binary(Reason) ->
{ok, 200, iot_util:json_error(400, Reason)}
end
end;
%% 停止服务
handle_request("POST", "/service/stop", _, #{<<"uuid">> := UUID, <<"service_id">> := ServiceId}) when is_binary(UUID), is_binary(ServiceId) ->
case iot_host:get_pid(UUID) of
undefined ->
{ok, 200, iot_util:json_error(404, <<"host not found">>)};
Pid when is_pid(Pid) ->
case iot_host:stop_service(Pid, ServiceId) of
{ok, Ref} ->
case iot_host:await_reply(Ref, 5000) of
{ok, Result} ->
{ok, 200, iot_util:json_data(Result)};
{error, Reason} ->
{ok, 200, iot_util:json_error(400, Reason)}
end;
{error, Reason} when is_binary(Reason) ->
{ok, 200, iot_util:json_error(400, Reason)}
end
end;
%% 远程调用微服务, 返回值的格式为json
handle_request("POST", "/service/invoke", _, #{<<"uuid">> := UUID, <<"service_id">> := ServiceId, <<"payload">> := Payload, <<"timeout">> := Timeout0})
when is_binary(UUID), is_binary(ServiceId), is_binary(Payload), is_integer(Timeout0) ->
case iot_host:get_pid(UUID) of
undefined ->
{ok, 200, iot_util:json_error(404, <<"host not found">>)};
Pid when is_pid(Pid) ->
Timeout = Timeout0 * 1000,
case iot_host:invoke_service(Pid, ServiceId, Payload, Timeout) of
{ok, Ref} ->
case iot_host:await_reply(Ref, Timeout) of
{ok, Result} ->
{ok, 200, iot_util:json_data(Result)};
{error, Reason} ->
{ok, 200, iot_util:json_error(400, Reason)}
end;
{error, Reason} when is_binary(Reason) ->
{ok, 200, iot_util:json_error(400, Reason)}
end
end;
handle_request("POST", "/service/task_log", _, #{<<"uuid">> := UUID, <<"task_id">> := TaskId}) when is_binary(UUID), is_integer(TaskId) ->
case iot_host:get_pid(UUID) of
undefined ->
{ok, 200, iot_util:json_error(404, <<"host not found">>)};
Pid when is_pid(Pid) ->
case iot_host:task_log(Pid, TaskId) of
{ok, Ref} ->
case iot_host:await_reply(Ref, 5000) of
{ok, Result} ->
{ok, 200, iot_util:json_data(Result)};
{error, Reason} ->
{ok, 200, iot_util:json_error(400, Reason)}
end;
{error, Reason} when is_binary(Reason) ->
{ok, 200, iot_util:json_error(400, Reason)}
end
end;
handle_request(_, Path, _, _) ->
Path1 = list_to_binary(Path),
{ok, 200, iot_util:json_error(-1, <<"url: ", Path1/binary, " not found">>)}.
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%% helper methods
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%