From ecfbdae8f4319f73b9f67500f4c3781694430bda Mon Sep 17 00:00:00 2001 From: anlicheng <244108715@qq.com> Date: Sun, 19 Apr 2026 18:15:48 +0800 Subject: [PATCH] =?UTF-8?q?erlang=E5=86=85=E7=BD=AE=E6=94=AF=E6=8C=81ssl?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- config/sys-dev.config | 2 +- config/sys-prod.config | 7 +++ src/host/iot_host.erl | 18 +++--- src/iot_app.erl | 23 +++++--- .../tcp/{tcp_channel.erl => ssl_channel.erl} | 57 ++++++++++--------- todo_docs/mysql.md | 37 ------------ 6 files changed, 62 insertions(+), 82 deletions(-) rename src/transport/tcp/{tcp_channel.erl => ssl_channel.erl} (92%) delete mode 100644 todo_docs/mysql.md diff --git a/config/sys-dev.config b/config/sys-dev.config index 062f386..6e41661 100644 --- a/config/sys-dev.config +++ b/config/sys-dev.config @@ -7,7 +7,7 @@ {backlog, 10240} ]}, - {tcp_server, [ + {ssl_server, [ {port, 18092}, {acceptors, 500}, {max_connections, 10240}, diff --git a/config/sys-prod.config b/config/sys-prod.config index ca862f4..37c5485 100644 --- a/config/sys-prod.config +++ b/config/sys-prod.config @@ -7,6 +7,13 @@ {backlog, 10240} ]}, + {ssl_server, [ + {port, 18092}, + {acceptors, 500}, + {max_connections, 10240}, + {backlog, 10240} + ]}, + {redis_server, [ {port, 16379}, {acceptors, 500}, diff --git a/src/host/iot_host.erl b/src/host/iot_host.erl index 1c9e5fd..d5ed1b2 100644 --- a/src/host/iot_host.erl +++ b/src/host/iot_host.erl @@ -219,7 +219,7 @@ handle_event({call, From}, {rpc_call, ReceiverPid, RpcCall}, _, State = #state{u case HasSession andalso is_pid(ChannelPid) of true -> %% 通过websocket发送请求 - Ref = tcp_channel:rpc_call(ChannelPid, ReceiverPid, RpcCall), + Ref = ssl_channel:rpc_call(ChannelPid, ReceiverPid, RpcCall), {keep_state, State, [{reply, From, {ok, Ref}}]}; false -> logger:debug("[iot_host] uuid: ~p, invalid state: ~p", [UUID, state_map(State)]), @@ -229,7 +229,7 @@ handle_event({call, From}, {rpc_call, ReceiverPid, RpcCall}, _, State = #state{u handle_event({call, From}, {container_call, ReceiverPid, Request}, _, State = #state{uuid = UUID, channel_pid = ChannelPid, has_session = HasSession}) -> case HasSession andalso is_pid(ChannelPid) of true -> - Ref = tcp_channel:container_call(ChannelPid, ReceiverPid, Request), + Ref = ssl_channel:container_call(ChannelPid, ReceiverPid, Request), {keep_state, State, [{reply, From, {ok, Ref}}]}; false -> logger:debug("[iot_host] uuid: ~p, invalid state: ~p", [UUID, state_map(State)]), @@ -239,7 +239,7 @@ handle_event({call, From}, {container_call, ReceiverPid, Request}, _, State = #s handle_event({call, From}, {cancel_rpc_call, Ref}, _, State = #state{channel_pid = ChannelPid}) -> case is_pid(ChannelPid) of true -> - ok = tcp_channel:cancel_rpc_call(ChannelPid, Ref), + ok = ssl_channel:cancel_rpc_call(ChannelPid, Ref), {keep_state, State, [{reply, From, ok}]}; false -> {keep_state, State, [{reply, From, ok}]} @@ -251,7 +251,7 @@ handle_event({call, From}, {pub, Topic, Qos, Content}, ?STATE_ACTIVATED, State = true -> logger:debug("[iot_host] host: ~p, publish to topic: ~p, content: ~p", [UUID, Topic, Content]), %% 通过websocket发送请求 - tcp_channel:pub(ChannelPid, Topic, Qos, Content), + ssl_channel:pub(ChannelPid, Topic, Qos, Content), {keep_state, State, [{reply, From, ok}]}; false -> @@ -265,7 +265,7 @@ handle_event({call, From}, {command, CommandType, Command}, ?STATE_ACTIVATED, St true -> logger:debug("[iot_host] host: ~p, command_type: ~p, command: ~p", [UUID, CommandType, Command]), %% 通过websocket发送请求 - tcp_channel:command(ChannelPid, CommandType, Command), + ssl_channel:command(ChannelPid, CommandType, Command), {keep_state, State, [{reply, From, ok}]}; false -> logger:debug("[iot_host] host: ~p, command_type: ~p, command: ~p, invalid state: ~p", [UUID, CommandType, Command, state_map(State)]), @@ -277,7 +277,7 @@ handle_event({call, From}, {activate, true}, _, State = #state{uuid = UUID, chan case is_pid(ChannelPid) of true -> logger:debug("[iot_host] uuid: ~p, activate: true", [UUID]), - tcp_channel:command(ChannelPid, ?COMMAND_AUTH, <<1:8>>); + ssl_channel:command(ChannelPid, ?COMMAND_AUTH, <<1:8>>); false -> logger:debug("[iot_host] uuid: ~p, activate: true, no channel", [UUID]) end, @@ -287,9 +287,9 @@ handle_event({call, From}, {activate, true}, _, State = #state{uuid = UUID, chan handle_event({call, From}, {activate, false}, _, State = #state{uuid = UUID, channel_pid = ChannelPid}) -> case is_pid(ChannelPid) of true -> - tcp_channel:command(ChannelPid, ?COMMAND_AUTH, <<0:8>>), + ssl_channel:command(ChannelPid, ?COMMAND_AUTH, <<0:8>>), logger:debug("[iot_host] uuid: ~p, activate: false", [UUID]), - tcp_channel:stop(ChannelPid, closed); + ssl_channel:stop(ChannelPid, closed); false -> logger:debug("[iot_host] uuid: ~p, activate: false, no channel", [UUID]) end, @@ -347,7 +347,7 @@ handle_event(info, {timeout, _, heartbeat_ticker}, _, State = #state{uuid = UUID end, %% 关闭channel，主机需要重新连接，才能保存状态的一致 - is_pid(ChannelPid) andalso tcp_channel:stop(ChannelPid, closed), + is_pid(ChannelPid) andalso ssl_channel:stop(ChannelPid, closed), erlang:start_timer(?HEARTBEAT_INTERVAL, self(), heartbeat_ticker), {keep_state, State#state{channel_pid = undefined, has_session = false, heartbeat_counter = 0}}; diff --git a/src/iot_app.erl b/src/iot_app.erl index 7d67918..6d341f5 100644 --- a/src/iot_app.erl +++ b/src/iot_app.erl @@ -19,8 +19,8 @@ start(_StartType, _StartArgs) -> %% 启动http服务 start_http_server(), - %% 启动tcp服务 - start_tcp_server(), + %% 启动ssl服务 + start_ssl_server(), iot_sup:start_link(). @@ -68,26 +68,31 @@ start_http_server() -> logger:debug("[http_server] the http server start at: ~p, pid is: ~p", [Port, Pid]). -%% 启动tcp服务 -start_tcp_server() -> - {ok, Props} = application:get_env(iot, tcp_server), +%% 启动ssl服务 +start_ssl_server() -> + {ok, Props} = application:get_env(iot, ssl_server), Acceptors = proplists:get_value(acceptors, Props, 50), MaxConnections = proplists:get_value(max_connections, Props, 10240), Backlog = proplists:get_value(backlog, Props, 1024), Port = proplists:get_value(port, Props), + PrivDir = code:priv_dir(iot), + CertFile = filename:join([PrivDir, "ssl", "server.crt"]), + KeyFile = filename:join([PrivDir, "ssl", "server.key"]), TransOpts = #{ max_connections => MaxConnections, num_acceptors => Acceptors, shutdown => brutal_kill, socket_opts => [ - {nodelay, false}, + {nodelay, true}, {backlog, Backlog}, - {port, Port} + {port, Port}, + {certfile, CertFile}, + {keyfile, KeyFile} ] }, - {ok, _} = ranch:start_listener(tcp_server, ranch_tcp, TransOpts, tcp_channel, []), - logger:debug("[iot_app] the tcp server start at: ~p", [Port]). + {ok, _} = ranch:start_listener(ssl_server, ranch_ssl, TransOpts, ssl_channel, []), + logger:debug("[iot_app] the ssl server start at: ~p", [Port]). -spec ensure_mnesia_schema() -> any(). ensure_mnesia_schema() -> diff --git a/src/transport/tcp/tcp_channel.erl b/src/transport/tcp/ssl_channel.erl similarity index 92% rename from src/transport/tcp/tcp_channel.erl rename to src/transport/tcp/ssl_channel.erl index be00d67..7ae7195 100644 --- a/src/transport/tcp/tcp_channel.erl +++ b/src/transport/tcp/ssl_channel.erl @@ -6,7 +6,7 @@ %%% @end %%% Created : 11. 1月 2021 上午12:17 %%%------------------------------------------------------------------- --module(tcp_channel). +-module(ssl_channel). -author("licheng5"). -include("protocol.hrl"). -include("message_pb.hrl"). @@ -134,20 +134,6 @@ handle_cast({request_call, ReceiverPid, Ref, Body}, State = #state{transport = T inflight = maps:put(PacketId, RequestInfo, Inflight) }}. -%% auth验证 -handle_info({tcp, Socket, <>}, State = #state{transport = Transport, socket = Socket}) -> - RequestFrame = message_pb:decode_msg(FrameBin, 'RequestFrame'), - handle_request_frame(RequestFrame, Transport, Socket, State); - -handle_info({tcp, Socket, <>}, State = #state{socket = Socket, host_pid = HostPid}) -> - CastFrame = message_pb:decode_msg(FrameBin, 'CastFrame'), - handle_cast_frame(CastFrame, HostPid, State); - -%% 主机端的消息响应 -handle_info({tcp, Socket, <>}, State = #state{socket = Socket, inflight = Inflight}) -> - ResponseFrame = message_pb:decode_msg(FrameBin, 'ResponseFrame'), - handle_response_frame(ResponseFrame, Inflight, State); - handle_info({timeout, TimerRef, {request_timeout, PacketId}}, State = #state{inflight = Inflight}) -> case maps:get(PacketId, Inflight, undefined) of #inflight_request{ref = Ref, timer_ref = TimerRef} -> @@ -157,14 +143,6 @@ handle_info({timeout, TimerRef, {request_timeout, PacketId}}, State = #state{inf {noreply, State} end; -handle_info({tcp_error, Sock, Reason}, State = #state{socket = Sock}) -> - logger:notice("[sdlan_channel] tcp_error: ~p", [Reason]), - {stop, normal, State}; - -handle_info({tcp_closed, Sock}, State = #state{socket = Sock}) -> - logger:notice("[sdlan_channel] tcp_closed"), - {stop, normal, State}; - %% 关闭当前通道 handle_info({stop, Reason}, State) -> {stop, Reason, State}; @@ -174,8 +152,35 @@ handle_info({'DOWN', _, process, HostPid, Reason}, State = #state{uuid = UUID, h logger:debug("[ws_channel] uuid: ~p, channel will close because host exited with reason: ~p", [UUID, Reason]), {stop, State}; +handle_info({ssl, Socket, <>}, + State = #state{transport = Transport, socket = Socket}) -> + RequestFrame = message_pb:decode_msg(FrameBin, 'RequestFrame'), + handle_request_frame(RequestFrame, Transport, Socket, State); + +handle_info({ssl, Socket, <>}, + State = #state{socket = Socket, host_pid = HostPid}) -> + CastFrame = message_pb:decode_msg(FrameBin, 'CastFrame'), + handle_cast_frame(CastFrame, HostPid, State); + +handle_info({ssl, Socket, <>}, + State = #state{socket = Socket, inflight = Inflight}) -> + ResponseFrame = message_pb:decode_msg(FrameBin, 'ResponseFrame'), + handle_response_frame(ResponseFrame, Inflight, State); + +handle_info({ssl_closed, Socket}, State = #state{socket = Socket}) -> + logger:notice("[sdlan_channel] ssl socket closed"), + {stop, normal, State}; + +handle_info({ssl_error, Socket, Reason}, State = #state{socket = Socket}) -> + logger:notice("[sdlan_channel] ssl socket error: ~p", [Reason]), + {stop, normal, State}; + +handle_info({ssl_passive, Socket}, State = #state{transport = Transport, socket = Socket}) -> + ok = Transport:setopts(Socket, [{active, true}]), + {noreply, State}; + handle_info(Info, State) -> - logger:warning("[sdlan_channel] get a unknown message: ~p, channel will closed, state: ~p", [Info, State]), + logger:warning("[sdlan_channel] get a unknown message: ~p, state: ~p", [Info, State]), {noreply, State}. terminate(Reason, #state{}) -> @@ -265,14 +270,14 @@ handle_cast_frame(#'CastFrame'{body = {event_stream, CastMessage}}, HostPid, Sta handle_event_stream_frame(CastMessage), {noreply, State}; handle_cast_frame(#'CastFrame'{body = Body}, _HostPid, State) -> - logger:warning("[tcp_channel] unsupported cast message type: command, body: ~p", [Body]), + logger:warning("[ssl_channel] unsupported cast message type: command, body: ~p", [Body]), {noreply, State}. -spec handle_event_stream_frame(message_pb:'TaskEventStream'()) -> any(). handle_event_stream_frame(#'TaskEventStream'{task_id = TaskId, type = Type0, stream = Reason0}) when Type0 =:= <<"close">> -> iot_event_stream_observer:stream_close(TaskId, iolist_to_binary(Reason0)); handle_event_stream_frame(#'TaskEventStream'{task_id = TaskId, type = Type, stream = Stream}) -> - logger:debug("[tcp_channel] get task_id: ~p, type: ~ts, stream: ~ts", [TaskId, Type, Stream]), + logger:debug("[ssl_channel] get task_id: ~p, type: ~ts, stream: ~ts", [TaskId, Type, Stream]), iot_event_stream_observer:stream_data(TaskId, Type, Stream). -spec handle_response_frame(message_pb:'ResponseFrame'(), map(), #state{}) -> diff --git a/todo_docs/mysql.md b/todo_docs/mysql.md deleted file mode 100644 index c737b39..0000000 --- a/todo_docs/mysql.md +++ /dev/null @@ -1,37 +0,0 @@ -# 待完成 - -## 1. endpoint需要存储在数据库 -```mysql - -CREATE TABLE `endpoint` ( -`id` bigint unsigned NOT NULL AUTO_INCREMENT, -`name` varchar(191) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL DEFAULT '' COMMENT '名称,路由时基于名称', -`title` varchar(191) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL DEFAULT '' COMMENT '序列号', -`type` varchar(60) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci DEFAULT '' COMMENT '类型', -`config_json` text CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci COMMENT '配置信息，基于json格式存储', -`status` smallint NOT NULL DEFAULT '-1', -`creator` smallint NOT NULL DEFAULT '0' COMMENT '创建人', -`created_at` timestamp NULL DEFAULT NULL, -`updated_at` timestamp NULL DEFAULT NULL, -PRIMARY KEY (`id`), -KEY `idx_name` (`name`) -) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci -``` - -## 2. service_config微服务配置也要存在于数据库 -```mysql -CREATE TABLE `service_config` ( -`id` bigint unsigned NOT NULL AUTO_INCREMENT, -`service_id` bigint unsigned NOT NULL COMMENT '服务的id', -`host_uuid` varchar(191) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL DEFAULT '' COMMENT '名称,路由时基于名称', -`config_json` text CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci COMMENT '配置信息，基于json格式存储', -`last_config_json` text CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci COMMENT '配置信息，基于json格式存储', -`creator` smallint NOT NULL DEFAULT '0' COMMENT '创建人', -`created_at` timestamp NULL DEFAULT NULL, -`updated_at` timestamp NULL DEFAULT NULL, -PRIMARY KEY (`id`), -KEY `idx_service_id` (`service_id`) -) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci - - -``` \ No newline at end of file