erlang内置支持ssl
This commit is contained in:
parent
ae97b5374c
commit
ecfbdae8f4
@ -7,7 +7,7 @@
|
||||
{backlog, 10240}
|
||||
]},
|
||||
|
||||
{tcp_server, [
|
||||
{ssl_server, [
|
||||
{port, 18092},
|
||||
{acceptors, 500},
|
||||
{max_connections, 10240},
|
||||
|
||||
@ -7,6 +7,13 @@
|
||||
{backlog, 10240}
|
||||
]},
|
||||
|
||||
{ssl_server, [
|
||||
{port, 18092},
|
||||
{acceptors, 500},
|
||||
{max_connections, 10240},
|
||||
{backlog, 10240}
|
||||
]},
|
||||
|
||||
{redis_server, [
|
||||
{port, 16379},
|
||||
{acceptors, 500},
|
||||
|
||||
@ -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}};
|
||||
|
||||
@ -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() ->
|
||||
|
||||
@ -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, <<?FRAME_REQUEST, FrameBin/binary>>}, State = #state{transport = Transport, socket = Socket}) ->
|
||||
RequestFrame = message_pb:decode_msg(FrameBin, 'RequestFrame'),
|
||||
handle_request_frame(RequestFrame, Transport, Socket, State);
|
||||
|
||||
handle_info({tcp, Socket, <<?FRAME_CAST, FrameBin/binary>>}, State = #state{socket = Socket, host_pid = HostPid}) ->
|
||||
CastFrame = message_pb:decode_msg(FrameBin, 'CastFrame'),
|
||||
handle_cast_frame(CastFrame, HostPid, State);
|
||||
|
||||
%% 主机端的消息响应
|
||||
handle_info({tcp, Socket, <<?FRAME_RESPONSE, FrameBin/binary>>}, 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, <<?FRAME_REQUEST, FrameBin/binary>>},
|
||||
State = #state{transport = Transport, socket = Socket}) ->
|
||||
RequestFrame = message_pb:decode_msg(FrameBin, 'RequestFrame'),
|
||||
handle_request_frame(RequestFrame, Transport, Socket, State);
|
||||
|
||||
handle_info({ssl, Socket, <<?FRAME_CAST, FrameBin/binary>>},
|
||||
State = #state{socket = Socket, host_pid = HostPid}) ->
|
||||
CastFrame = message_pb:decode_msg(FrameBin, 'CastFrame'),
|
||||
handle_cast_frame(CastFrame, HostPid, State);
|
||||
|
||||
handle_info({ssl, Socket, <<?FRAME_RESPONSE, FrameBin/binary>>},
|
||||
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{}) ->
|
||||
@ -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
|
||||
|
||||
|
||||
```
|
||||
Loading…
x
Reference in New Issue
Block a user