From 768c5d6206d9cbf193a0391fba8f83ae482cb98a Mon Sep 17 00:00:00 2001 From: anlicheng <244108715@qq.com> Date: Thu, 8 Jan 2026 16:02:20 +0800 Subject: [PATCH] add publisher --- apps/aircon/src/aircon_mqtt_publisher.erl | 165 +++++++++++++++++++++ apps/aircon/src/aircon_mqtt_subscriber.erl | 3 +- apps/aircon/src/aircon_sup.erl | 9 ++ 3 files changed, 175 insertions(+), 2 deletions(-) create mode 100644 apps/aircon/src/aircon_mqtt_publisher.erl diff --git a/apps/aircon/src/aircon_mqtt_publisher.erl b/apps/aircon/src/aircon_mqtt_publisher.erl new file mode 100644 index 0000000..220884c --- /dev/null +++ b/apps/aircon/src/aircon_mqtt_publisher.erl @@ -0,0 +1,165 @@ +%%%------------------------------------------------------------------- +%%% @author aresei +%%% @copyright (C) 2023, +%%% @doc +%%% 1. 需要考虑集群部署的相关问题，上行的数据可能在集群中共享 +%%% 2. host进程不能直接去监听topic，这样涉及到新增和下线的很多问题 +%%% @end +%%% Created : 12. 3月 2023 21:27 +%%%------------------------------------------------------------------- +-module(aircon_mqtt_publisher). +-author("aresei"). + +-behaviour(gen_server). + +%% API +-export([start_link/0]). +-export([publish/3]). +-export([test/0]). + +%% gen_server callbacks +-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). + +-define(SERVER, ?MODULE). + +-record(state, { + conn_pid :: pid() +}). + +%%%=================================================================== +%%% API +%%%=================================================================== + +test() -> + Topic = <<"/aircon/20525456021829/data">>, + Payload = <<"{\"key\":\"OnOff\",\"value\":0,\"unit\":10,\"type\":\"AI\",\"timestamp\":1767747017459,\"name\":\"开关状态\",\"label\":\"开关状态\",\"sessionid\":760143116038215}">>, + publish(Topic, Payload, 2). + +-spec publish(Topic :: binary(), Payload :: binary(), Qos :: integer()) -> no_return(). +publish(Topic, Payload, Qos) when is_binary(Topic), is_binary(Payload), is_integer(Qos) -> + gen_server:cast(?SERVER, {publish, Topic, Payload, Qos}). + +%% @doc Spawns the server and registers the local name (unique) +-spec(start_link() -> + {ok, Pid :: pid()} | ignore | {error, Reason :: term()}). +start_link() -> + gen_server:start_link({local, ?SERVER}, ?MODULE, [], []). + +%%%=================================================================== +%%% 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([]) -> + %% 建立到emqx服务器的连接 + Opts = emqx_opts(<<"aircon-data-publisher">>), + logger:debug("[mqtt_publisher] opts is: ~p", [Opts]), + case emqtt:start_link(Opts) of + {ok, ConnPid} -> + logger:debug("[mqtt_publisher] start conntecting, pid: ~p", [ConnPid]), + {ok, _} = emqtt:connect(ConnPid), + logger:debug("[mqtt_publisher] connect success"), + {ok, #state{conn_pid = ConnPid}}; + ignore -> + logger:debug("[mqtt_publisher] connect emqx get ignore"), + {stop, ignore}; + {error, Reason} -> + logger:debug("[mqtt_publisher] connect emqx get error: ~p", [Reason]), + {stop, Reason} + end. + +%% @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({publish, Topic, Payload, Qos}, State = #state{conn_pid = ConnPid}) -> + Res = emqtt:publish(ConnPid, Topic, Payload, Qos), + logger:debug("[mqtt_publisher] publish result is: ~p", [Res]), + {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({puback, Packet = #{packet_id := _PacketId}}, State = #state{}) -> + logger:debug("[mqtt_publisher] receive puback packet: ~p", [Packet]), + {noreply, State}; + +handle_info(Info, State = #state{}) -> + logger:debug("[mqtt_publisher] 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) -> + ok = emqtt:disconnect(ConnPid), + logger:debug("[iot_mqtt_publisher] terminate with reason: ~p", [Reason]), + ok; +terminate(Reason, _State) -> + logger:debug("[iot_mqtt_publisher] 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 emqx_opts(ClientSuffix :: binary()) -> list(). +emqx_opts(ClientSuffix) when is_binary(ClientSuffix) -> + %% 建立到emqx服务器的连接 + {ok, Props} = application:get_env(aircon, emqx_server), + EMQXHost = proplists:get_value(host, Props), + EMQXPort = proplists:get_value(port, Props, 1883), + Username = proplists:get_value(username, Props), + Password = proplists:get_value(password, Props), + RetryInterval = proplists:get_value(retry_interval, Props, 5), + Keepalive = proplists:get_value(keepalive, Props, 86400), + + Node = atom_to_binary(node()), + ClientId = <<"mqtt-client-", Node/binary, "-", ClientSuffix/binary>>, + [ + {clientid, ClientId}, + {host, EMQXHost}, + {port, EMQXPort}, + {owner, self()}, + {tcp_opts, []}, + {username, Username}, + {password, Password}, + {keepalive, Keepalive}, + {auto_ack, true}, + {proto_ver, v3}, + {retry_interval, RetryInterval} + ]. \ No newline at end of file diff --git a/apps/aircon/src/aircon_mqtt_subscriber.erl b/apps/aircon/src/aircon_mqtt_subscriber.erl index f9721b8..8108959 100644 --- a/apps/aircon/src/aircon_mqtt_subscriber.erl +++ b/apps/aircon/src/aircon_mqtt_subscriber.erl @@ -51,13 +51,12 @@ start_link() -> init([]) -> %% 建立到emqx服务器的连接 Opts = emqx_opts(<<"aircon-data-subscriber">>), - logger:debug("[opts] is: ~p", [Opts]), + logger:debug("[mqtt_subscriber] opts is: ~p", [Opts]), case emqtt:start_link(Opts) of {ok, ConnPid} -> logger:debug("[mqtt_subscriber] start conntecting, pid: ~p", [ConnPid]), {ok, _} = emqtt:connect(ConnPid), logger:debug("[mqtt_subscriber] connect success"), - %% 监听和设备的全部事件 SubscribeResult = emqtt:subscribe(ConnPid, ?Topics), logger:debug("[mqtt_subscriber] subscribe topics: ~p, result is: ~p", [?Topics, SubscribeResult]), diff --git a/apps/aircon/src/aircon_sup.erl b/apps/aircon/src/aircon_sup.erl index 7553938..7bce29f 100644 --- a/apps/aircon/src/aircon_sup.erl +++ b/apps/aircon/src/aircon_sup.erl @@ -72,6 +72,15 @@ init([]) -> modules => ['aircon_device_sup'] }, + #{ + id => 'aircon_mqtt_publisher', + start => {'aircon_mqtt_publisher', start_link, []}, + restart => permanent, + shutdown => 2000, + type => worker, + modules => ['aircon_mqtt_publisher'] + }, + #{ id => 'aircon_mqtt_subscriber', start => {'aircon_mqtt_subscriber', start_link, []},