diff --git a/apps/endpoint/include/endpoint.hrl b/apps/endpoint/include/endpoint.hrl index df99a1b..e455155 100644 --- a/apps/endpoint/include/endpoint.hrl +++ b/apps/endpoint/include/endpoint.hrl @@ -10,16 +10,18 @@ -record(http_endpoint, { url = <<>> :: binary(), - pool_size = 10 :: integer() -}). + %% post | put + method = post :: atom(), + %% json | web_form, + data_format = json :: atom(), --record(ws_endpoint, { - url = <<>> :: binary() + pool_size = 10 :: integer() }). -record(mqtt_endpoint, { host = <<>> :: binary(), port = 0 :: integer(), + client_id = <<>> :: binary(), username = <<>> :: binary(), password = <<>> :: binary(), topic = <<>> :: binary(), @@ -45,8 +47,6 @@ -record(endpoint, { id :: integer(), - %% 不同的对端名字要唯一 - name = <<>> :: binary(), %% 标题描述 title = <<>> :: binary(), mapper = <<>> :: binary(), @@ -55,17 +55,9 @@ %% fun(LocationCode :: binary(), Fields :: [{<<"key">> => <<>>, <<"value">> => <<>>, <<"unit">> => <<>>}], Timestamp :: integer()) mapper_fun = fun(_, Fields) -> Fields end :: fun(), %% 配置项, 格式: #{<<"protocol">> => <<"http|https|ws|kafka|mqtt">>, <<"args">> => #{}} - config = #http_endpoint{} :: #http_endpoint{} | #ws_endpoint{} | #mqtt_endpoint{} | #kafka_endpoint{} | #mysql_endpoint{}, + config = #http_endpoint{} :: #http_endpoint{} | #mqtt_endpoint{} | #kafka_endpoint{} | #mysql_endpoint{}, %% 更新时间 updated_at = 0 :: integer(), %% 创建时间 created_at = 0 :: integer() -}). - - - --record(post_data, { - id, - location_code, - body }). \ No newline at end of file diff --git a/apps/endpoint/src/endpoint.erl b/apps/endpoint/src/endpoint.erl index adf3c51..1f9fe00 100644 --- a/apps/endpoint/src/endpoint.erl +++ b/apps/endpoint/src/endpoint.erl @@ -10,349 +10,49 @@ -module(endpoint). -include("endpoint.hrl"). --behaviour(gen_statem). - %% API --export([start_link/2]). --export([get_name/1, get_pid/1, forward/4, get_stat/1, reload/2, clean_up/1, get_mapper_fun/1]). - -%% gen_statem callbacks --export([init/1, handle_event/4, terminate/3, code_change/4, callback_mode/0]). - -%% 消息重发间隔 --define(RETRY_INTERVAL, 5000). - --record(state, { - endpoint, - mp, - postman_pid :: undefined | pid(), - %% 队列 - queue, - %% 定时器 - timer_map = #{}, - %% 窗口大小,允许最大的未确认消息数 - window_size = 10, - %% 未确认的消息数 - flight_num = 0, - %% 记录成功处理的消息数 - acc_num = 0 -}). +-export([start_link/1]). +-export([get_name/1, get_pid/1, forward/4, reload/2, clean_up/1]). %%%=================================================================== %%% API %%%=================================================================== --spec get_name(Name :: binary() | #endpoint{}) -> atom(). -get_name(#endpoint{name = Name}) when is_binary(Name) -> - get_name(Name); -get_name(EndpointName) when is_binary(EndpointName) -> - binary_to_atom(<<"iot_endpoint:", EndpointName/binary>>). +-spec start_link(Endpoint :: #endpoint{}) -> {'ok', pid()} | 'ignore' | {'error', term()}. +start_link(Endpoint = #endpoint{id = Id, config = #http_endpoint{}}) -> + Name = get_name(Id), + endpoint_http:start_link(Name, Endpoint); +start_link(Endpoint = #endpoint{id = Id, config = #mqtt_endpoint{}}) -> + Name = get_name(Id), + endpoint_mqtt:start_link(Name, Endpoint); +start_link(Endpoint = #endpoint{id = Id, config = #mysql_endpoint{}}) -> + Name = get_name(Id), + endpoint_mysql:start_link(Name, Endpoint). --spec get_pid(Name :: binary()) -> undefined | pid(). -get_pid(Name) when is_binary(Name) -> - whereis(get_name(Name)). +-spec get_name(Name :: binary() | #endpoint{}) -> atom(). +get_name(Id) when is_integer(Id) -> + list_to_atom("endpoint:" ++ integer_to_list(Id)). + +-spec get_pid(Id :: integer()) -> undefined | pid(). +get_pid(Id) when is_integer(Id) -> + whereis(get_name(Id)). -spec forward(Pid :: undefined | pid(), LocationCode :: binary(), Fields :: list(), Timestamp :: integer()) -> no_return(). forward(undefined, _, _, _) -> ok; forward(Pid, LocationCode, Fields, Timestamp) when is_pid(Pid), is_binary(LocationCode), is_list(Fields); is_binary(Fields), is_integer(Timestamp) -> - gen_statem:cast(Pid, {forward, LocationCode, Fields, Timestamp}). + Pid ! {forward, LocationCode, Fields, Timestamp}. reload(Pid, NEndpoint = #endpoint{}) when is_pid(Pid) -> gen_statem:cast(Pid, {reload, NEndpoint}). --spec get_stat(Pid :: pid()) -> {ok, Stat :: #{}}. -get_stat(Pid) when is_pid(Pid) -> - gen_statem:call(Pid, get_stat, 5000). - -spec clean_up(Pid :: pid()) -> ok. clean_up(Pid) when is_pid(Pid) -> gen_statem:call(Pid, clean_up, 5000). --spec get_mapper_fun(Pid :: pid()) -> fun(). -get_mapper_fun(Pid) when is_pid(Pid) -> - gen_statem:call(Pid, get_mapper_fun). - -%% @doc Creates a gen_statem process which calls Module:init/1 to -%% initialize. To ensure a synchronized start-up procedure, this -%% function does not return until Module:init/1 has returned. -start_link(Name, Endpoint = #endpoint{}) -> - gen_statem:start_link({local, Name}, ?MODULE, [Endpoint], []). - -%%%=================================================================== -%%% gen_statem callbacks -%%%=================================================================== - -%% @private -%% @doc Whenever a gen_statem is started using gen_statem:start/[3,4] or -%% gen_statem:start_link/[3,4], this function is called by the new -%% process to initialize. -init([Endpoint = #endpoint{matcher = Regexp}]) -> - erlang:process_flag(trap_exit, true), - - %% 编译正则表达式 - {ok, MP} = re:compile(Regexp), - %% 创建转发器, 避免阻塞当前进程的创建,因此采用了延时初始化的机制 - erlang:start_timer(0, self(), create_postman), - - {ok, disconnected, #state{endpoint = Endpoint, mp = MP, queue = queue:new(), postman_pid = undefined}}. - -%% @private -%% @doc This function is called by a gen_statem when it needs to find out -%% the callback mode of the callback module. -callback_mode() -> - handle_event_function. - -%% @private -%% @doc There should be one instance of this function for each possible -%% state name. If callback_mode is state_functions, one of these -%% functions is called when gen_statem receives and event from -%% call/2, cast/2, or as a normal process message. - -%% 重新加载新的终端配置 -handle_event(cast, {reload, NEndpoint}, disconnected, State = #state{endpoint = Endpoint}) -> - lager:warning("[iot_endpoint] state_name: disconnected, reload endpoint, old: ~p, new: ~p", [Endpoint, NEndpoint]), - {keep_state, State#state{endpoint = NEndpoint}}; - -handle_event(cast, {reload, NEndpoint = #endpoint{name = Name}}, connected, State = #state{endpoint = Endpoint, timer_map = TimerMap, postman_pid = PostmanPid}) -> - lager:debug("[iot_endpoint] state_name: connected, reload endpoint, old: ~p, new: ~p", [Endpoint, NEndpoint]), - case config_equals(NEndpoint#endpoint.config, Endpoint#endpoint.config) of - true -> - lager:debug("[iot_endpoint] reload endpoint: ~p, config equals", [Name]), - {keep_state, State#state{endpoint = NEndpoint}}; - false -> - %% 解除和postman的link关系 - unlink(PostmanPid), - %% 关闭postman进程 - catch PostmanPid ! stop, - %% 未确认的消息需要暂存 - lists:foreach(fun({_, TimerRef}) -> catch erlang:cancel_timer(TimerRef) end, maps:to_list(TimerMap)), - %% 重新建立新的postman - erlang:start_timer(0, self(), create_postman), - - {next_state, disconnected, State#state{endpoint = NEndpoint, timer_map = maps:new(), postman_pid = undefined}} - end; - -handle_event(cast, {forward, LocationCode, Fields, Timestamp}, StateName, State = #state{mp = MP, queue = Q, window_size = WindowSize, flight_num = FlightNum, endpoint = #endpoint{name = Name}}) -> - case re:run(LocationCode, MP, [{capture, all, list}]) of - nomatch -> - {keep_state, State}; - {match, _} -> - lager:debug("[iot_endpoint] name: ~p, match location_code: ~p", [Name, LocationCode]), - Q1 = queue:in(#north_data{location_code = LocationCode, fields = Fields, timestamp = Timestamp}, Q), - %% 避免不必要的内部消息 - Actions = case StateName =:= connected andalso FlightNum < WindowSize of - true -> [{next_event, info, fetch_next}]; - false -> [] - end, - {keep_state, State#state{queue = Q1}, Actions} - end; - -%% 触发读取下一条数据 -handle_event(info, fetch_next, disconnected, State = #state{endpoint = #endpoint{name = Name}}) -> - lager:debug("[iot_endpoint] fetch_next endpoint: ~p, postman offline, data in queue", [Name]), - {keep_state, State}; -handle_event(info, fetch_next, connected, State = #state{flight_num = FlightNum, window_size = WindowSize}) when FlightNum >= WindowSize -> - {keep_state, State}; -handle_event(info, fetch_next, connected, State = #state{queue = Q, endpoint = #endpoint{name = Name}, timer_map = TimerMap, flight_num = FlightNum}) -> - case queue:out(Q) of - {{value, NorthData = #north_data{id = Id}}, Q1} -> - lager:debug("[iot_endpoint] endpoint: ~p, fetch_next success, north data is: ~p", [Name, NorthData]), - case do_post(NorthData, State) of - error -> - {keep_state, State}; - {ok, TimerRef} -> - {keep_state, State#state{timer_map = maps:put(Id, TimerRef, TimerMap), flight_num = FlightNum + 1}} - end; - {empty, Q1} -> - {keep_state, State#state{queue = Q1}} - end; - -%% 收到确认消息 -handle_event(info, {ack, Id}, StateName, State = #state{tab_name = TabName, endpoint = #endpoint{name = Name}, timer_map = TimerMap, acc_num = AccNum, flight_num = FlightNum}) -> - ok = mnesia_queue:delete(TabName, Id), - lager:debug("[iot_endpoint] endpoint: ~p, get ack: ~p, delete from mnesia", [Name, Id]), - Actions = case StateName =:= connected of - true -> [{next_event, info, fetch_next}]; - false -> [] - end, - {keep_state, State#state{timer_map = remove_timer(Id, TimerMap), acc_num = AccNum + 1, flight_num = FlightNum - 1}, Actions}; - -%% 收到重发过期请求 -handle_event(info, {timeout, _, {repost_ticker, NorthData = #north_data{id = Id}}}, connected, State = #state{endpoint = #endpoint{name = Name}, timer_map = TimerMap}) -> - lager:debug("[iot_endpoint] endpoint: ~p, repost data: ~p", [Name, Id]), - case do_post(NorthData, State) of - error -> - {keep_state, State}; - {ok, TimerRef} -> - {keep_state, State#state{timer_map = maps:put(Id, TimerRef, TimerMap)}} - end; - -%% 离线时,忽略超时逻辑 -handle_event(info, {timeout, _, {repost_ticker, _}}, disconnected, State) -> - {keep_state, State}; - -handle_event(info, {timeout, _, create_postman}, disconnected, State = #state{endpoint = Endpoint = #endpoint{name = Name, config = Config}, window_size = WindowSize}) -> - lager:debug("[iot_endpoint] endpoint: ~p, create postman", [Name]), - try - {ok, PostmanPid} = create_postman(Endpoint), - %% 最多允许window_size - Actions = lists:map(fun(_) -> {next_event, info, fetch_next} end, lists:seq(1, WindowSize)), - {next_state, connected, State#state{endpoint = Endpoint, postman_pid = PostmanPid, timer_map = maps:new(), flight_num = 0}, Actions} - catch _:Error -> - lager:warning("[iot_endpoint] endpoint: ~p, config: ~p, create postman get error: ~p", [Name, Config, Error]), - erlang:start_timer(?RETRY_INTERVAL, self(), create_postman), - - {keep_state, State#state{endpoint = Endpoint, postman_pid = undefined}} - end; - -%% 删除时需要清理 -handle_event({call, From}, clean_up, _, State = #state{tab_name = TabName}) -> - mnesia:delete_table(TabName), - {keep_state, State, [{reply, From, ok}]}; - -handle_event({call, From}, get_mapper_fun, _, State = #state{endpoint = #endpoint{mapper_fun = F}}) -> - {keep_state, State, [{reply, From, F}]}; - -%% 获取当前统计信息 -handle_event({call, From}, get_stat, StateName, State = #state{acc_num = AccNum, tab_name = TabName}) -> - Stat = #{ - <<"acc_num">> => AccNum, - <<"queue_num">> => mnesia_queue:table_size(TabName), - <<"state_name">> => atom_to_binary(StateName) - }, - {keep_state, State, [{reply, From, Stat}]}; - -%% postman进程挂掉时,重新建立新的 -handle_event(info, {'EXIT', PostmanPid, Reason}, connected, State = #state{endpoint = #endpoint{name = Name}, timer_map = TimerMap, postman_pid = PostmanPid}) -> - lager:warning("[iot_endpoint] endpoint: ~p, postman exited with reason: ~p", [Name, Reason]), - lists:foreach(fun({_, TimerRef}) -> catch erlang:cancel_timer(TimerRef) end, maps:to_list(TimerMap)), - erlang:start_timer(?RETRY_INTERVAL, self(), create_postman), - - {next_state, disconnected, State#state{timer_map = maps:new(), postman_pid = undefined}}; - -%% @private -%% @doc If callback_mode is handle_event_function, then whenever a -%% gen_statem receives an event from call/2, cast/2, or as a normal -%% process message, this function is called. -handle_event(EventType, Event, StateName, State) -> - lager:warning("[iot_endpoint] unknown message, event_type: ~p, event: ~p, state_name: ~p, state: ~p", [EventType, Event, StateName, State]), - {keep_state, State}. - -%% @private -%% @doc This function is called by a gen_statem when it is about to -%% terminate. It should be the opposite of Module:init/1 and do any -%% necessary cleaning up. When it returns, the gen_statem terminates with -%% Reason. The return value is ignored. -terminate(Reason, _StateName, #state{endpoint = #endpoint{name = Name}}) -> - lager:debug("[iot_endpoint] endpoint: ~p, terminate with reason: ~p", [Name, Reason]), - ok. - -%% @private -%% @doc Convert process state when code is changed -code_change(_OldVsn, StateName, State = #state{}, _Extra) -> - {ok, StateName, State}. - -%%%=================================================================== -%%% Internal functions -%%%=================================================================== - --spec remove_timer(Id :: integer(), TimerMap :: #{}) -> NTimerMap :: #{}. -remove_timer(Id, TimerMap) when is_integer(Id), is_map(TimerMap) -> - case maps:take(Id, TimerMap) of - error -> - TimerMap; - {TimerRef, NTimerMap} -> - is_reference(TimerRef) andalso erlang:cancel_timer(TimerRef), - NTimerMap - end. - -%% 对http和https协议的支持 -create_postman(#endpoint{config = #http_endpoint{url = Url, pool_size = PoolSize}}) -> - WorkerArgs = [{url, Url}], - broker_postman:start_link(http_postman, WorkerArgs, PoolSize); - -%% 对mqtt协议的支持, 只需要建立单个链接 -create_postman(#endpoint{name = Name, config = #mqtt_endpoint{host = Host, port = Port, username = Username, password = Password, topic = Topic, qos = Qos}}) -> - Node = atom_to_binary(node()), - ClientId = <<"mqtt-client-", Node/binary, "-", Name/binary>>, - Opts = [ - {clientid, ClientId}, - {host, binary_to_list(Host)}, - {port, Port}, - {tcp_opts, []}, - {username, binary_to_list(Username)}, - {password, binary_to_list(Password)}, - {keepalive, 86400}, - {auto_ack, true}, - {connect_timeout, 5000}, - {proto_ver, v5}, - {retry_interval, 5000} - ], - - mqtt_postman:start_link(Opts, Topic, Qos); - -%% 对mysql协议的支持 -create_postman(#endpoint{config = #mysql_endpoint{host = Host, port = Port, username = Username, password = Password, database = Database, table_name = TableName, pool_size = PoolSize}}) -> - WorkerArgs = [ - {mysql_opts, [ - {host, binary_to_list(Host)}, - {port, Port}, - {user, binary_to_list(Username)}, - {password, binary_to_list(Password)}, - {keep_alive, true}, - {database, binary_to_list(Database)}, - {queries, [<<"set names utf8">>]} - ]}, - {table, TableName} - ], - broker_postman:start_link(mysql_postman, WorkerArgs, PoolSize); -create_postman(#endpoint{}) -> - throw(<<"not supported">>). - --spec do_post(NorthData :: #north_data{}, State :: #state{}) -> error | {ok, TimerRef :: reference()}. -do_post(NorthData = #north_data{id = Id}, #state{postman_pid = PostmanPid, tab_name = TabName, endpoint = #endpoint{name = Name, mapper_fun = MapperFun}}) -> - lager:debug("[iot_endpoint] endpoint: ~p, fetch_next success, north data is: ~p", [Name, NorthData]), - case safe_invoke_mapper(MapperFun, NorthData) of - {ok, Body} -> - PostmanPid ! {post, self(), make_post_data(NorthData, Body)}, - %% 重发机制, 在发送的过程中mapper可能会改变 - TimerRef = erlang:start_timer(?RETRY_INTERVAL, self(), {repost_ticker, NorthData}), - {ok, TimerRef}; - {error, Error} -> - lager:debug("[iot_endpoint] forward endpoint: ~p, mapper get error: ~p, message discard", [Name, Error]), - mnesia_queue:delete(TabName, Id), - error; - ignore -> - lager:debug("[iot_endpoint] forward endpoint: ~p, mapper ignore, messge discard", [Name]), - mnesia_queue:delete(TabName, Id), - error - end. - --spec safe_invoke_mapper(MapperFun :: fun(), NorthData :: #north_data{}) -> - {ok, Body :: any()} | ignore | {error, Reason :: any()}. -safe_invoke_mapper(MapperFun, #north_data{location_code = LocationCode, fields = Fields, timestamp = Timestamp}) -> - try - if - is_function(MapperFun, 2) -> - MapperFun(LocationCode, Fields); - is_function(MapperFun, 3) -> - MapperFun(LocationCode, Fields, Timestamp) - end - catch _:Error -> - {error, Error} - end. - --spec make_post_data(NorthData :: #north_data{}, Body :: any()) -> PostData :: #post_data{}. -make_post_data(#north_data{id = Id, location_code = LocationCode}, Body) -> - #post_data{id = Id, location_code = LocationCode, body = Body}. - - -spec config_equals(any(), any()) -> boolean(). config_equals(#http_endpoint{url = Url}, #http_endpoint{url = Url}) -> true; -config_equals(#ws_endpoint{url = Url}, #ws_endpoint{url = Url}) -> - true; config_equals(#kafka_endpoint{username = Username, password = Password, bootstrap_servers = BootstrapServers, topic = Topic}, #kafka_endpoint{username = Username, password = Password, bootstrap_servers = BootstrapServers, topic = Topic}) -> true; diff --git a/apps/endpoint/src/endpoint_buffer.erl b/apps/endpoint/src/endpoint_buffer.erl index 20c0108..1a7b3a1 100644 --- a/apps/endpoint/src/endpoint_buffer.erl +++ b/apps/endpoint/src/endpoint_buffer.erl @@ -72,7 +72,7 @@ trigger_n(Buffer = #buffer{window_size = WindowSize}) -> %% 触发读取下一条数据 -spec trigger_next(Buffer :: #buffer{}) -> NBuffer :: #buffer{}. -trigger_next(Buffer = #buffer{tid = Tid, cursor = Cursor, timer_pid = TimerPid, flight_num = FlightNum, endpoint = #endpoint{name = Name, mapper_fun = MapperFun}}) -> +trigger_next(Buffer = #buffer{tid = Tid, cursor = Cursor, timer_pid = TimerPid, flight_num = FlightNum, endpoint = #endpoint{title = Title, mapper_fun = MapperFun}}) -> case ets:next(Tid, Cursor) of '$end_of_table' -> Buffer; @@ -88,11 +88,11 @@ trigger_next(Buffer = #buffer{tid = Tid, cursor = Cursor, timer_pid = TimerPid, Buffer#buffer{flight_num = FlightNum + 1}; {error, Error} -> - lager:debug("[iot_endpoint] forward endpoint: ~p, mapper get error: ~p, message discard", [Name, Error]), + lager:debug("[iot_endpoint] forward endpoint: ~p, mapper get error: ~p, message discard", [Title, Error]), ets:delete(Tid, Id), Buffer; ignore -> - lager:debug("[iot_endpoint] forward endpoint: ~p, mapper ignore, messge discard", [Name]), + lager:debug("[iot_endpoint] forward endpoint: ~p, mapper ignore, messge discard", [Title]), ets:delete(Tid, Id), Buffer end diff --git a/apps/endpoint/src/endpoint_http.erl b/apps/endpoint/src/endpoint_http.erl index d564211..3f9ccb1 100644 --- a/apps/endpoint/src/endpoint_http.erl +++ b/apps/endpoint/src/endpoint_http.erl @@ -53,7 +53,7 @@ cleanup(Pid) when is_pid(Pid) -> gen_server:cast(Pid, cleanup). %% @doc Spawns the server and registers the local name (unique) --spec(start_link(Name, Endpoint :: #http_endpoint{}) -> +-spec(start_link(Name :: atom(), Endpoint :: #http_endpoint{}) -> {ok, Pid :: pid()} | ignore | {error, Reason :: term()}). start_link(Name, Endpoint = #endpoint{config = #http_endpoint{}}) when is_atom(Name) -> gen_server:start_link({local, Name}, ?MODULE, [Endpoint], []). @@ -105,7 +105,7 @@ handle_cast(cleanup, State = #state{buffer = Buffer}) -> {noreply, NewState :: #state{}} | {noreply, NewState :: #state{}, timeout() | hibernate} | {stop, Reason :: term(), NewState :: #state{}}). -handle_info({next_data, Id, _LocationCode, Body}, State = #state{buffer = Buffer, http_endpoint = #http_endpoint{url = Url}}) -> +handle_info({next_data, Id, _LocationCode, Body}, State = #state{buffer = Buffer, endpoint = #http_endpoint{url = Url}}) -> Headers = [ {<<"content-type">>, <<"application/json">>} ], diff --git a/apps/endpoint/src/endpoint_mqtt.erl b/apps/endpoint/src/endpoint_mqtt.erl index 0be1cb6..2f44ec9 100644 --- a/apps/endpoint/src/endpoint_mqtt.erl +++ b/apps/endpoint/src/endpoint_mqtt.erl @@ -98,21 +98,13 @@ callback_mode() -> %% functions is called when gen_statem receives and event from %% call/2, cast/2, or as a normal process message. -handle_event(cast, {forward, LocationCode, Fields, Timestamp}, _StateName, State = #state{buffer = Buffer, endpoint = #endpoint{name = Name}}) -> - lager:debug("[iot_endpoint] name: ~p, match location_code: ~p", [Name, LocationCode]), +handle_event(cast, {forward, LocationCode, Fields, Timestamp}, _StateName, State = #state{buffer = Buffer}) -> NBuffer = endpoint_buffer:append({LocationCode, Fields, Timestamp}, Buffer), - {keep_state, State#state{buffer = NBuffer}}; - - -handle_event(info, {timeout, _, create_postman}, disconnected, - State = #state{buffer = Buffer, endpoint = #endpoint{name = Name, config = #mqtt_endpoint{host = Host, port = Port, username = Username, password = Password, topic = Topic, qos = Qos}}}) -> - - lager:debug("[iot_endpoint] endpoint: ~p, create postman", [Name]), - - Node = atom_to_binary(node()), - ClientId = <<"mqtt-client-", Node/binary, "-", Name/binary>>, +handle_event(info, {timeout, _, create_postman}, disconnected, State = #state{buffer = Buffer, + endpoint = #endpoint{title = Title, config = #mqtt_endpoint{host = Host, port = Port, username = Username, password = Password, client_id = ClientId}}}) -> + lager:debug("[iot_endpoint] endpoint: ~p, create postman", [Title]), Opts = [ {owner, self()}, {clientid, ClientId}, @@ -190,8 +182,8 @@ handle_event(info, {puback, #{packet_id := PacketId}}, connected, State = #state end; %% postman进程挂掉时,重新建立新的 -handle_event(info, {'EXIT', ConnPid, Reason}, connected, State = #state{endpoint = #endpoint{name = Name}, conn_pid = ConnPid}) -> - lager:warning("[enpoint_mqtt] endpoint: ~p, conn pid exit with reason: ~p", [Name, Reason]), +handle_event(info, {'EXIT', ConnPid, Reason}, connected, State = #state{endpoint = #endpoint{title = Title}, conn_pid = ConnPid}) -> + lager:warning("[enpoint_mqtt] endpoint: ~p, conn pid exit with reason: ~p", [Title, Reason]), erlang:start_timer(?RETRY_INTERVAL, self(), create_postman), {next_state, disconnected, State#state{conn_pid = undefined}}; @@ -208,8 +200,8 @@ handle_event(EventType, Event, StateName, State) -> %% terminate. It should be the opposite of Module:init/1 and do any %% necessary cleaning up. When it returns, the gen_statem terminates with %% Reason. The return value is ignored. -terminate(Reason, _StateName, #state{endpoint = #endpoint{name = Name}, buffer = Buffer}) -> - lager:debug("[iot_endpoint] endpoint: ~p, terminate with reason: ~p", [Name, Reason]), +terminate(Reason, _StateName, #state{endpoint = #endpoint{title = Title}, buffer = Buffer}) -> + lager:debug("[iot_endpoint] endpoint: ~p, terminate with reason: ~p", [Title, Reason]), endpoint_buffer:cleanup(Buffer), ok. diff --git a/apps/endpoint/src/endpoint_mysql.erl b/apps/endpoint/src/endpoint_mysql.erl index f069027..86d0b0a 100644 --- a/apps/endpoint/src/endpoint_mysql.erl +++ b/apps/endpoint/src/endpoint_mysql.erl @@ -92,8 +92,7 @@ callback_mode() -> %% functions is called when gen_statem receives and event from %% call/2, cast/2, or as a normal process message. -handle_event(cast, {forward, LocationCode, Fields, Timestamp}, _StateName, State = #state{buffer = Buffer, endpoint = #endpoint{name = Name}}) -> - lager:debug("[iot_endpoint] name: ~p, match location_code: ~p", [Name, LocationCode]), +handle_event(cast, {forward, LocationCode, Fields, Timestamp}, _StateName, State = #state{buffer = Buffer}) -> NBuffer = endpoint_buffer:append({LocationCode, Fields, Timestamp}, Buffer), {keep_state, State#state{buffer = NBuffer}}; @@ -102,8 +101,8 @@ handle_event(cast, cleanup, _StateName, State = #state{buffer = Buffer}) -> {keep_state, State}; handle_event(info, {timeout, _, create_postman}, disconnected, - State = #state{buffer = Buffer, endpoint = #endpoint{name = Name, config = #mysql_endpoint{host = Host, port = Port, username = Username, password = Password, database = Database, table_name = TableName, pool_size = PoolSize}}}) -> - lager:debug("[iot_endpoint] endpoint: ~p, create postman", [Name]), + State = #state{buffer = Buffer, endpoint = #endpoint{title = Title, config = #mysql_endpoint{host = Host, port = Port, username = Username, password = Password, database = Database, table_name = TableName, pool_size = PoolSize}}}) -> + lager:debug("[iot_endpoint] endpoint: ~p, create postman", [Title]), WorkerArgs = [ {host, binary_to_list(Host)}, {port, Port}, @@ -139,7 +138,7 @@ handle_event(info, {next_data, _Id, _LocationCode, _Message}, disconnected, Stat {keep_state, State}; %% 发送数据到mqtt服务器 handle_event(info, {next_data, Id, _LocationCode, Fields}, connected, - State = #state{pool_pid = PoolPid, buffer = Buffer, endpoint = #endpoint{name = Name, config = #mysql_endpoint{table_name = Table}}}) -> + State = #state{pool_pid = PoolPid, buffer = Buffer, endpoint = #endpoint{title = Title, config = #mysql_endpoint{table_name = Table}}}) -> {ok, InsertSql, Values} = insert_sql(Table, Fields), case poolboy:transaction(PoolPid, fun(ConnPid) -> mysql:query(ConnPid, InsertSql, Values) end) of @@ -147,13 +146,13 @@ handle_event(info, {next_data, Id, _LocationCode, Fields}, connected, NBuffer = endpoint_buffer:ack(Id, Buffer), {keep_state, State#state{buffer = NBuffer}}; Error -> - lager:warning("[endpoint_mysql] endpoint: ~p, insert mysql get error: ~p", [Name, Error]), + lager:warning("[endpoint_mysql] endpoint: ~p, insert mysql get error: ~p", [Title, Error]), {keep_state, State} end; %% postman进程挂掉时,重新建立新的 -handle_event(info, {'EXIT', PoolPid, Reason}, connected, State = #state{endpoint = #endpoint{name = Name}, pool_pid = PoolPid}) -> - lager:warning("[enpoint_mqtt] endpoint: ~p, conn pid exit with reason: ~p", [Name, Reason]), +handle_event(info, {'EXIT', PoolPid, Reason}, connected, 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), {next_state, disconnected, State#state{pool_pid = undefined}}; @@ -170,8 +169,8 @@ handle_event(EventType, Event, StateName, State) -> %% terminate. It should be the opposite of Module:init/1 and do any %% necessary cleaning up. When it returns, the gen_statem terminates with %% Reason. The return value is ignored. -terminate(Reason, _StateName, #state{endpoint = #endpoint{name = Name}, buffer = Buffer}) -> - lager:debug("[iot_endpoint] endpoint: ~p, terminate with reason: ~p", [Name, Reason]), +terminate(Reason, _StateName, #state{endpoint = #endpoint{title = Title}, buffer = Buffer}) -> + lager:debug("[iot_endpoint] endpoint: ~p, terminate with reason: ~p", [Title, Reason]), endpoint_buffer:cleanup(Buffer), ok.