From 4b72498d3a78a54d63e5eb4ae9a33bfe3a3681f9 Mon Sep 17 00:00:00 2001 From: anlicheng <244108715@qq.com> Date: Fri, 22 Dec 2023 17:28:24 +0800 Subject: [PATCH] add mqtt consumer --- apps/iot/src/consumer/iot_zd_consumer.erl | 201 ++++++++++++++++++++++ 1 file changed, 201 insertions(+) create mode 100644 apps/iot/src/consumer/iot_zd_consumer.erl diff --git a/apps/iot/src/consumer/iot_zd_consumer.erl b/apps/iot/src/consumer/iot_zd_consumer.erl new file mode 100644 index 0000000..cb10871 --- /dev/null +++ b/apps/iot/src/consumer/iot_zd_consumer.erl @@ -0,0 +1,201 @@ +%%%------------------------------------------------------------------- +%%% @author aresei +%%% @copyright (C) 2023, +%%% @doc +%%% 1. 需要考虑集群部署的相关问题，上行的数据可能在集群中共享 +%%% 2. host进程不能直接去监听topic，这样涉及到新增和下线的很多问题 +%%% @end +%%% Created : 12. 3月 2023 21:27 +%%%------------------------------------------------------------------- +-module(iot_zd_consumer). +-author("aresei"). +-include("iot.hrl"). + +-behaviour(gen_server). + +%% API +-export([start_link/1]). + +%% gen_server callbacks +-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). + +-define(SERVER, ?MODULE). +-define(RETRY_INTERVAL, 5000). + +%% 需要订阅的主题信息 +-define(Topics,[ + {<<"CET/NX/upload">>, 2} +]). + +-record(state, { + conn_pid :: pid(), + logger_pid :: pid(), + mqtt_props :: list(), + is_connected = false +}). + +%%%=================================================================== +%%% API +%%%=================================================================== + +%% @doc Spawns the server and registers the local name (unique) +-spec(start_link(Props :: list()) -> + {ok, Pid :: pid()} | ignore | {error, Reason :: term()}). +start_link(Props) when is_list(Props) -> + gen_server:start_link({local, ?MODULE}, ?MODULE, [Props], []). + +%%%=================================================================== +%%% gen_server callbacks +%%%=================================================================== + +%% @private +%% @doc Initializes the server +-spec(init(Args :: term()) -> + {ok, State :: #state{}} | {ok, State :: #state{}, timeout() | hibernate} | + {stop, Reason :: term()} | ignore). +init([Props]) -> + erlang:process_flag(trap_exit, true), + + %% 创建转发器, 避免阻塞当前进程的创建，因此采用了延时初始化的机制 + erlang:start_timer(0, self(), create_consumer), + %% 启动日志记录器 + {ok, LoggerPid} = iot_logger:start_link("directive_data"), + + {ok, disconnected, #state{mqtt_props = Props, conn_pid = undefined, logger_pid = LoggerPid, is_connected = false}}. + +%% @private +%% @doc Handling call messages +-spec(handle_call(Request :: term(), From :: {pid(), Tag :: term()}, + State :: #state{}) -> + {reply, Reply :: term(), NewState :: #state{}} | + {reply, Reply :: term(), NewState :: #state{}, timeout() | hibernate} | + {noreply, NewState :: #state{}} | + {noreply, NewState :: #state{}, timeout() | hibernate} | + {stop, Reason :: term(), Reply :: term(), NewState :: #state{}} | + {stop, Reason :: term(), NewState :: #state{}}). +handle_call(_Info, _From, State = #state{conn_pid = _ConnPid}) -> + {reply, ok, State}. + +%% @private +%% @doc Handling cast messages +-spec(handle_cast(Request :: term(), State :: #state{}) -> + {noreply, NewState :: #state{}} | + {noreply, NewState :: #state{}, timeout() | hibernate} | + {stop, Reason :: term(), NewState :: #state{}}). +handle_cast(_Request, State = #state{}) -> + {noreply, State}. + +%% @private +%% @doc Handling all non call/cast messages +-spec(handle_info(Info :: timeout() | term(), State :: #state{}) -> + {noreply, NewState :: #state{}} | + {noreply, NewState :: #state{}, timeout() | hibernate} | + {stop, Reason :: term(), NewState :: #state{}}). +handle_info({disconnect, ReasonCode, Properties}, State) -> + lager:debug("[iot_zd_consumer] Recv a DISONNECT packet - ReasonCode: ~p, Properties: ~p", [ReasonCode, Properties]), + {stop, disconnected, State}; +%% 必须要做到消息的快速分发，数据的json反序列需要在host进程进行 +handle_info({publish, #{packet_id := _PacketId, payload := Payload, qos := Qos, topic := Topic}}, State) -> + lager:debug("[iot_zd_consumer] Recv a topic: ~p, publish packet: ~p, qos: ~p", [Topic, Payload, Qos]), + %% 将消息分发到对应的host进程去处理 + {noreply, State}; +handle_info({puback, Packet = #{packet_id := _PacketId}}, State = #state{}) -> + lager:debug("[iot_zd_consumer] receive puback packet: ~p", [Packet]), + {noreply, State}; + +handle_info({timeout, _, create_consumer}, State = #state{mqtt_props = Props, is_connected = false}) -> + try + {ok, ConnPid} = create_consumer(Props), + {noreply, State#state{conn_pid = ConnPid}} + catch _:Error:Stack -> + lager:warning("[iot_zd_consumer] config: ~p, create consumer get error: ~p, stack: ~p", [Props, Error, Stack]), + erlang:start_timer(?RETRY_INTERVAL, self(), create_consumer), + {noreply, State#state{conn_pid = undefined}} + end; + +%% postman进程挂掉时，重新建立新的 +handle_info({'EXIT', ConnPid, Reason}, State = #state{conn_pid = ConnPid}) -> + lager:warning("[iot_zd_consumer] consumer exited with reason: ~p", [Reason]), + erlang:start_timer(?RETRY_INTERVAL, self(), create_consumer), + + {next_state, disconnected, State#state{conn_pid = undefined}}; + +handle_info(Info, State = #state{}) -> + lager:debug("[iot_zd_consumer] get info: ~p", [Info]), + {noreply, State}. + +%% @private +%% @doc This function is called by a gen_server 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_server terminates +%% with Reason. The return value is ignored. +-spec(terminate(Reason :: (normal | shutdown | {shutdown, term()} | term()), + State :: #state{}) -> term()). +terminate(Reason, _State = #state{conn_pid = ConnPid}) when is_pid(ConnPid) -> + %% 取消topic的订阅 + TopicNames = lists:map(fun({Name, _}) -> Name end, ?Topics), + {ok, _Props, _ReasonCode} = emqtt:unsubscribe(ConnPid, #{}, TopicNames), + + ok = emqtt:disconnect(ConnPid), + lager:debug("[iot_zd_consumer] terminate with reason: ~p", [Reason]), + ok; +terminate(Reason, _State) -> + lager:debug("[iot_zd_consumer] terminate with reason: ~p", [Reason]), + ok. + +%% @private +%% @doc Convert process state when code is changed +-spec(code_change(OldVsn :: term() | {down, term()}, State :: #state{}, + Extra :: term()) -> + {ok, NewState :: #state{}} | {error, Reason :: term()}). +code_change(_OldVsn, State = #state{}, _Extra) -> + {ok, State}. + +%%%=================================================================== +%%% Internal functions +%%%=================================================================== + +-spec create_consumer(Props :: list()) -> {ok, ConnPid :: pid()} | {error, Reason :: any()}. +create_consumer(Props) when is_list(Props) -> + Node = atom_to_binary(node()), + ClientId = <<"mqtt-client-", Node/binary, "-zhongdian_mqtt_consumer">>, + + %% 建立到emqx服务器的连接 + Host = proplists:get_value(host, Props), + Port = proplists:get_value(port, Props, 18080), + Username = proplists:get_value(username, Props), + Password = proplists:get_value(password, Props), + Keepalive = proplists:get_value(keepalive, Props, 86400), + + Opts = [ + {clientid, ClientId}, + {host, Host}, + {port, Port}, + {owner, self()}, + {tcp_opts, []}, + {username, Username}, + {password, Password}, + {keepalive, Keepalive}, + {auto_ack, true}, + {connect_timeout, 5000}, + {proto_ver, v5}, + {retry_interval, 5000} + ], + + %% 建立到emqx服务器的连接 + lager:debug("[iot_zd_consumer] opts is: ~p", [Opts]), + case emqtt:start_link(Opts) of + {ok, ConnPid} -> + %% 监听和host相关的全部事件 + lager:debug("[iot_zd_consumer] start conntecting, pid: ~p", [ConnPid]), + {ok, _} = emqtt:connect(ConnPid), + lager:debug("[iot_zd_consumer] connect success, pid: ~p", [ConnPid]), + SubscribeResult = emqtt:subscribe(ConnPid, ?Topics), + lager:debug("[iot_zd_consumer] subscribe topics: ~p, result is: ~p", [?Topics, SubscribeResult]), + + {ok, ConnPid}; + ignore -> + {error, ignore}; + {error, Reason} -> + {error, Reason} + end. \ No newline at end of file