%% Copyright 2014 Erlio GmbH Basel Switzerland (http://erl.io) %% %% Licensed under the Apache License, Version 2.0 (the "License"); %% you may not use this file except in compliance with the License. %% You may obtain a copy of the License at %% %% http://www.apache.org/licenses/LICENSE-2.0 %% %% Unless required by applicable law or agreed to in writing, software %% distributed under the License is distributed on an "AS IS" BASIS, %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% See the License for the specific language governing permissions and %% limitations under the License. -module(gen_emqtt). -behavior(gen_fsm). -include("vmq_types.hrl"). %startup -export([start_link/3, start_link/4, start/3, start/4, subscribe/2, subscribe/3, unsubscribe/2, publish/4, publish/5, disconnect/1, call/2, cast/2]). %% gen_fsm callbacks -export([init/1, handle_info/3, handle_event/3, handle_sync_event/4, code_change/4, terminate/3]). % fsm state -export([ waiting_for_connack/2, connected/2, connecting/2, disconnecting/2]). -export([behaviour_info/1]). -define(MQTT_PROTO_MAJOR, 3). -define(MQTT_PROTO_MINOR, 1). -record(state, {host :: inet:ip_address(), port :: inet:port_number(), sock :: gen_tcp:socket() | ssl:socket(), msgid = 1 :: non_neg_integer(), username :: binary(), password :: binary(), client :: string() , clean_session = false :: boolean(), last_will_topic :: string() | undefined, last_will_msg :: string() | undefined, last_will_retain :: boolean() | undefined, last_will_qos :: non_neg_integer(), buffer = <<>> :: binary(), reconnect_timeout, keepalive_interval = 60000, retry_interval = 10000, proto_version = ?MQTT_PROTO_MAJOR, %% mod, mod_state=[], transport, parser = <<>>, waiting_acks = dict:new(), info_fun}). start_link(Module, Args, Opts) -> GenFSMOpts = proplists:get_value(gen_fsm, Opts, []), gen_fsm:start_link(?MODULE, [Module, Args, Opts], GenFSMOpts). start_link(Name, Module, Args, Opts) -> GenFSMOpts = proplists:get_value(gen_fsm, Opts, []), gen_fsm:start_link(Name, ?MODULE, [Module, Args, Opts], GenFSMOpts). start(Module, Args, Opts) -> GenFSMOpts = proplists:get_value(gen_fsm, Opts, []), gen_fsm:start(?MODULE, [Module, Args, Opts], GenFSMOpts). start(Name, Module, Args, Opts) -> GenFSMOpts = proplists:get_value(gen_fsm, Opts, []), gen_fsm:start(Name, ?MODULE, [Module, Args, Opts], GenFSMOpts). publish(P, Topic, Payload, Qos) -> publish(P, Topic, Payload, Qos, false). publish(P, Topic, Payload, Qos, Retain) -> gen_fsm:send_event(P, {publish, Topic, Payload, Qos, Retain, false}). subscribe(P, [T|_] = Topics) when is_tuple(T) -> gen_fsm:send_event(P, {subscribe, Topics}). subscribe(P, Topic, QoS) -> subscribe(P, [{Topic, QoS}]). unsubscribe(P, [T|_] = Topics) when is_list(T) -> gen_fsm:send_event(P, {unsubscribe, Topics}); unsubscribe(P, Topic) -> unsubscribe(P, [Topic]). disconnect(P) -> gen_fsm:send_event(P, disconnect). call(P, Req) -> gen_fsm:sync_send_all_state_event(P, Req). cast(P, Req) -> gen_fsm:send_all_state_event(P, Req). wrap_res(StateName, init, Args, #state{mod=Mod} = State) -> case erlang:apply(Mod, init, Args) of {ok, ModState} -> {ok, StateName, State#state{mod_state=ModState}}; {ok, ModState, Extra} -> {ok, StateName, State#state{mod_state=ModState}, Extra}; {stop, Reason} -> {stop, Reason} end; wrap_res(StateName, Function, Args, #state{mod=Mod, mod_state=ModState} = State) -> wrap_res(erlang:apply(Mod, Function, Args ++ [ModState]), StateName, State). wrap_res({ok, ModState}, StateName, State) -> {next_state, StateName, State#state{mod_state=ModState}}; wrap_res({ok, ModState, Extra}, StateName, State) -> {next_state, StateName, State#state{mod_state=ModState}, Extra}; wrap_res({stop, Reason}, _StateName, _State) -> {stop, Reason}; wrap_res({reply, Reply, ModState}, StateName, State) -> {reply, Reply, StateName, State#state{mod_state=ModState}}; wrap_res({reply, Reply, ModState, Extra}, StateName, State) -> {reply, Reply, StateName, State#state{mod_state=ModState}, Extra}; wrap_res({noreply, ModState}, StateName, State) -> {next_state, StateName, State#state{mod_state=ModState}}; wrap_res({noreply, ModState, Extra}, StateName, State) -> {next_state, StateName, State#state{mod_state=ModState}, Extra}; wrap_res({stop, Reason, ModState}, _StateName, State) -> {stop, Reason, State#state{mod_state=ModState}}; wrap_res({stop, Reason, Reply, ModState}, _StateName, State) -> {stop, Reason, Reply, State#state{mod_state=ModState}}; wrap_res({ok}, _StateName, _State) -> ok; wrap_res(ok, _StateName, _State) -> ok. behaviour_info(callbacks) -> [{init,1}, {handle_call, 3}, {handle_cast, 2}, {handle_info, 2}, {terminate, 2}, {code_change, 3}, {on_connect, 1}, {on_connect_error, 2}, {on_disconnect, 1}, {on_subscribe, 2}, {on_unsubscribe, 2}, {on_publish, 3}]; behaviour_info(_Other) -> undefined. %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% %%% State Callbacks %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% connecting(connect, State) -> #state{host = Host, port = Port, transport={Transport, Opts}, client=ClientId, info_fun=InfoFun} = State, case Transport:connect(Host, Port, [binary, {packet, raw}|Opts]) of {ok, Sock} -> NewInfoFun = call_info_fun({connect_out, ClientId}, InfoFun), NewState = State#state{sock=Sock, buffer= <<>>, info_fun=NewInfoFun}, send_connect(NewState), active_once(Transport, Sock), {next_state, waiting_for_connack, NewState}; {error, _Reason} -> gen_fsm:send_event_after(3000, connect), wrap_res(connecting, on_connect_error, [server_not_found], State) end; connecting(disconnect, State) -> {stop, normal, State}; connecting(_Event, State) -> {next_state, connecting, State}. waiting_for_connack(_Event, State) -> {next_state, waiting_for_connack, State}. disconnecting({error, ConnectError}, #state{sock=Sock, transport={Transport,_}} = State) -> Transport:close(Sock), error_logger:error_msg("bridge disconnected due to connection error ~p", [ConnectError]), {stop, ConnectError, State}. connected({subscribe, Topics}, State=#state{transport={Transport,_}, msgid=MsgId, sock=Sock, waiting_acks=WAcks, retry_interval=RetryInterval, info_fun=InfoFun}) -> Frame = #mqtt_subscribe{ message_id = MsgId, topics = Topics }, NewInfoFun = call_info_fun({subscribe_out, MsgId}, InfoFun), send_frame(Transport, Sock, Frame), Key = {subscribe, MsgId}, Ref = gen_fsm:send_event_after(RetryInterval, {retry, Key}), {next_state, connected, State#state{msgid='++'(MsgId), waiting_acks=store(Key, {Ref, {subscribe, Topics}}, WAcks), info_fun=NewInfoFun}}; connected({unsubscribe, Topics}, State=#state{transport={Transport, _}, sock=Sock, msgid=MsgId, waiting_acks=WAcks, retry_interval=RetryInterval, info_fun=InfoFun}) -> Frame = #mqtt_unsubscribe{ message_id = MsgId, topics = Topics }, NewInfoFun = call_info_fun({unsubscribe_out, MsgId}, InfoFun), send_frame(Transport, Sock, Frame), Key = {unsubscribe, MsgId}, Ref = gen_fsm:send_event_after(RetryInterval, {retry, Key}), {next_state, connected, State#state{msgid='++'(MsgId), waiting_acks=store(Key, {Ref, {unsubscribe, Topics}}, WAcks), info_fun=NewInfoFun}}; connected({publish, Topic, Payload, QoS, Retain, Dup}, #state{msgid=MsgId} = State) -> {next_state, connected, send_publish(MsgId, Topic, Payload, QoS, Retain, Dup, State#state{msgid='++'(MsgId)})}; connected({publish, MsgId, Topic, Payload, QoS, Retain, _}, State) -> %% called in case of retry {next_state, connected, send_publish(MsgId, Topic, Payload, QoS, Retain, true, State)}; connected({retry, Key}, #state{transport={Transport,_}, sock=Sock, waiting_acks=WAcks, retry_interval=RetryInterval} = State) -> case dict:find(Key, WAcks) of error -> {next_state, connected, State}; {ok, {_Ref, #mqtt_publish{} = Frame}} -> send_frame(Transport, Sock, Frame#mqtt_publish{dup = true}), NewRef = gen_fsm:send_event_after(RetryInterval, {retry, Key}), NewDict = erase(Key, WAcks), {next_state, connected, State#state{waiting_acks=store(Key, {NewRef, Frame}, NewDict)}}; {ok, {_Ref, #mqtt_pubrel{} = Frame}} -> send_frame(Transport, Sock, Frame), NewRef = gen_fsm:send_event_after(RetryInterval, {retry, Key}), NewDict = erase(Key, WAcks), {next_state, connected, State#state{waiting_acks=store(Key, {NewRef, Frame}, NewDict)}}; {ok, {_Ref, Msg}} -> gen_fsm:send_event(self(), Msg), {next_state, connected, State#state{waiting_acks=erase(Key, WAcks)}} end; connected({ack, _MsgId}, State) -> {next_state, connected, State}; connected(ping, #state{transport={Transport, _}, sock=Sock, waiting_acks=WAcks, keepalive_interval=Int} = State) -> send_ping(Transport, Sock), Ref = gen_fsm:send_event_after(Int, ping), {next_state, connected, State#state{waiting_acks=store(ping, Ref, WAcks)}}; connected(disconnect, State=#state{transport={Transport, _}, sock=Sock}) -> send_disconnect(Transport, Sock), {stop, normal, State}; connected(maybe_reconnect, #state{client=ClientId, reconnect_timeout=Timeout, transport={Transport, _}, info_fun=InfoFun} = State) -> case Timeout of undefined -> {stop, normal, State}; _ -> Transport:close(State#state.sock), gen_fsm:send_event_after(Timeout, connect), NewInfoFun = call_info_fun({reconnect, ClientId}, InfoFun), wrap_res(connecting, on_disconnect, [], State#state{sock=undefined, info_fun=NewInfoFun}) end; connected(_Event, State) -> {stop, unknown_event, State}. %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% %% gen_fsm callbacks %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% init([Mod, Args, Opts]) -> Host = proplists:get_value(host, Opts, "localhost"), Port = proplists:get_value(port, Opts, 1883), Username = proplists:get_value(username, Opts, undefined), Password = proplists:get_value(password, Opts, undefined), ClientId = proplists:get_value(client, Opts, "emqttc"), CleanSession = proplists:get_value(clean_session, Opts, true), LWTopic = proplists:get_value(last_will_topic, Opts, undefined), LWMsg = proplists:get_value(last_will_msg, Opts, undefined), LWQos = proplists:get_value(last_will_qos, Opts, 0), LWRet = proplists:get_value(last_will_retain, Opts, undefined), ReconnectTimeout = proplists:get_value(reconnect_timeout, Opts, undefined), KeepAliveInterval = proplists:get_value(keepalive_interval, Opts, 60), RetryInterval = proplists:get_value(retry_interval, Opts, 10), ProtoVer = proplists:get_value(proto_version, Opts, ?MQTT_PROTO_MAJOR), InfoFun = proplists:get_value(info_fun, Opts, {fun(_,_) -> ok end, []}), {Transport, TransportOpts} = proplists:get_value(transport, Opts, {gen_tcp, []}), State = #state{host = Host, port = Port, proto_version=ProtoVer, username = Username, password = Password, client=ClientId, mod=Mod, clean_session=CleanSession, last_will_topic=LWTopic, last_will_msg=LWMsg, last_will_qos=LWQos, last_will_retain=LWRet, reconnect_timeout=case ReconnectTimeout of undefined -> undefined; _ -> 1000 * ReconnectTimeout end, keepalive_interval=1000 * KeepAliveInterval, retry_interval=1000* RetryInterval, transport={Transport, TransportOpts}, info_fun=InfoFun}, Res = wrap_res(connecting, init, [Args], State), gen_fsm:send_event_after(0, connect), Res. handle_info({ssl, Socket, Bin}, StateName, #state{sock=Socket} = State) -> #state{transport={Transport, _}, buffer=Buffer} = State, active_once(Transport, Socket), process_bytes(<>, StateName, State); handle_info({tcp, Socket, Bin}, StateName, #state{sock=Socket} = State) -> #state{transport={Transport, _}, buffer=Buffer} = State, active_once(Transport, Socket), process_bytes(<>, StateName, State); handle_info({ssl_closed, Sock}, _, State=#state{sock=Sock, reconnect_timeout=undefined}) -> {stop, normal, State}; handle_info({ssl_closed, Sock}, _, State=#state{sock=Sock, reconnect_timeout=Timeout}) -> gen_fsm:send_event_after(Timeout, connect), wrap_res(connecting, on_disconnect, [], State#state{sock=undefined}); handle_info({tcp_closed, Sock}, _, State=#state{sock=Sock, reconnect_timeout=undefined}) -> {stop, normal, State}; handle_info({tcp_closed, Sock}, _, State=#state{sock=Sock, reconnect_timeout=Timeout}) -> gen_fsm:send_event_after(Timeout, connect), wrap_res(connecting, on_disconnect, [], State#state{sock=undefined}); handle_info(Info, StateName, State) -> wrap_res(StateName, handle_info, [Info], State). process_bytes(Bytes, StateName, #state{parser=ParserState} = State) -> Data = <>, case vmq_parser:parse(Data) of {error, _Reason} -> {next_state, StateName, State#state{parser= <<>>}}; more -> {next_state, StateName, State#state{parser=Data}}; {Frame, Rest} -> {next_state, NextStateName, NewState} = handle_frame(StateName, Frame, State), process_bytes(Rest, NextStateName, NewState#state{parser= <<>>}) end. handle_frame(waiting_for_connack, #mqtt_connack{return_code=ReturnCode}, State) -> #state{waiting_acks=WAcks, keepalive_interval=Int, client=ClientId, info_fun=InfoFun} = State, case ReturnCode of ?CONNACK_ACCEPT when Int == 0 -> NewInfoFun = call_info_fun({connack_in, ClientId}, InfoFun), wrap_res(connected, on_connect, [], State#state{info_fun=NewInfoFun}); ?CONNACK_ACCEPT -> NewInfoFun = call_info_fun({connack_in, ClientId}, InfoFun), Ref = gen_fsm:send_event_after(Int, ping), wrap_res(connected, on_connect, [], State#state{waiting_acks=store(ping, Ref, WAcks), info_fun=NewInfoFun}); ?CONNACK_PROTO_VER -> gen_fsm:send_event_after(0, {error, {connect_error, ?CONNACK_PROTO_VER}}), wrap_res(disconnecting, on_connect_error, [wrong_protocol_version], State); ?CONNACK_INVALID_ID -> gen_fsm:send_event_after(0, {error, {connect_error, ?CONNACK_INVALID_ID}}), wrap_res(disconnecting, on_connect_error, [invalid_id], State); ?CONNACK_SERVER -> gen_fsm:send_event_after(0, {error, {connect_error, ?CONNACK_SERVER}}), wrap_res(disconnecting, on_connect_error, [server_not_available], State); ?CONNACK_CREDENTIALS -> gen_fsm:send_event_after(0, {error, {connect_error, ?CONNACK_CREDENTIALS}}), wrap_res(disconnecting, on_connect_error, [invalid_credentials], State); ?CONNACK_AUTH -> gen_fsm:send_event_after(0, {error, {connect_error, ?CONNACK_AUTH}}), wrap_res(disconnecting, on_connect_error, [not_authorized], State) end; handle_frame(connected, #mqtt_suback{message_id=MsgId, qos_table=QoSTable}, State) -> #state{waiting_acks=WAcks, info_fun=InfoFun} = State, Key = {subscribe, MsgId}, case dict:find(Key, WAcks) of {ok, {Ref, {subscribe, Topics}}} -> gen_fsm:cancel_timer(Ref), NewInfoFun = call_info_fun({suback, MsgId}, InfoFun), {TopicNames, _} = lists:unzip(Topics), case length(TopicNames) == length(QoSTable) of true -> wrap_res(connected, on_subscribe, [lists:zip(TopicNames, QoSTable)], State#state{waiting_acks=erase(Key, WAcks), info_fun=NewInfoFun}); false -> wrap_res(connected, on_subscribe, [{error, Topics, QoSTable}], State#state{waiting_acks=erase(Key, WAcks), info_fun=NewInfoFun}) end; error -> {next_state, connected, State} end; handle_frame(connected, #mqtt_unsuback{message_id=MsgId}, State) -> #state{waiting_acks=WAcks, info_fun=InfoFun} = State, Key = {unsubscribe, MsgId}, case dict:find(Key, WAcks) of {ok, {Ref, {unsubscribe, Topics}}} -> NewInfoFun = call_info_fun({unsuback, MsgId}, InfoFun), gen_fsm:cancel_timer(Ref), wrap_res(connected, on_unsubscribe, [Topics], State#state{waiting_acks=erase(Key, WAcks), info_fun=NewInfoFun}); error -> {next_state, connected, State} end; handle_frame(connected, #mqtt_puback{message_id=MessageId}, State) -> #state{waiting_acks=WAcks, info_fun=InfoFun} = State, %% qos1 flow Key = {publish, MessageId}, case dict:find(Key, WAcks) of {ok, {Ref, _}} -> NewInfoFun = call_info_fun({puback_in, MessageId}, InfoFun), gen_fsm:cancel_timer(Ref), {next_state, connected, State#state{waiting_acks=erase(Key, WAcks), info_fun=NewInfoFun}}; error -> {next_state, connected, State} end; handle_frame(connected, #mqtt_pubrec{message_id=MessageId}, State) -> #state{transport={Transport, _}, waiting_acks=WAcks, sock=Socket, retry_interval=RetryInterval, info_fun=InfoFun} = State, %% qos2 flow Key = {publish, MessageId}, %% is retried case dict:find(Key, WAcks) of {ok, {Ref, _}} -> NewInfoFun0 = call_info_fun({pubrec_in, MessageId}, InfoFun), gen_fsm:cancel_timer(Ref), % cancel republish timer PubRelFrame = #mqtt_pubrel{message_id=MessageId}, NewInfoFun1 = call_info_fun({pubrel_out, MessageId}, NewInfoFun0), send_frame(Transport, Socket, PubRelFrame), NewWAcks = erase(Key, WAcks), NewKey = {pubrel, MessageId}, NewRef = gen_fsm:send_event_after(RetryInterval, {retry, NewKey}), {next_state, connected, State#state{ waiting_acks=store(NewKey, {NewRef, PubRelFrame}, NewWAcks), info_fun=NewInfoFun1}}; error -> {next_state, connected, State} end; handle_frame(connected, #mqtt_pubrel{message_id=MessageId}, State) -> #state{transport={Transport,_}, waiting_acks=WAcks, sock=Socket, info_fun=InfoFun} = State, %% qos2 flow case dict:find({store, MessageId}, WAcks) of {ok, {Topic, Payload}} -> NewInfoFun0 = call_info_fun({pubrel_in, MessageId}, InfoFun), {next_state, connected, NewState} = wrap_res(connected, on_publish, [Topic, Payload], State), NewInfoFun1 = call_info_fun({pubcomp_out, MessageId}, NewInfoFun0), PubCompFrame = #mqtt_pubcomp{message_id=MessageId}, send_frame(Transport, Socket, PubCompFrame), {next_state, connected, NewState#state{ waiting_acks=erase({store, MessageId}, WAcks), info_fun=NewInfoFun1}}; error -> {next_state, connected, State} end; handle_frame(connected, #mqtt_pubcomp{message_id=MessageId}, State) -> #state{waiting_acks=WAcks, info_fun=InfoFun} = State, %% qos2 flow Key = {pubrel, MessageId}, case dict:find(Key, WAcks) of {ok, {Ref, _}} -> NewInfoFun = call_info_fun({pubcomp_in, MessageId}, InfoFun), gen_fsm:cancel_timer(Ref), % cancel republish timer {next_state, connected, State#state{waiting_acks=erase(Key, WAcks), info_fun=NewInfoFun}}; error -> {next_state, connected, State} end; handle_frame(connected, #mqtt_publish{message_id=MessageId, topic=Topic, qos=QoS, payload=Payload}, State) -> #state{transport={Transport, _}, sock=Socket, waiting_acks=WAcks, info_fun=InfoFun} = State, NewInfoFun = call_info_fun({publish_in, MessageId, Payload, QoS}, InfoFun), case QoS of 0 -> wrap_res(connected, on_publish, [Topic, Payload], State#state{info_fun=NewInfoFun}); 1 -> PubAckFrame = #mqtt_puback{message_id=MessageId}, NewInfoFun1 = call_info_fun({puback_out, MessageId}, NewInfoFun), Res = wrap_res(connected, on_publish, [Topic, Payload], State#state{info_fun=NewInfoFun1}), send_frame(Transport, Socket, PubAckFrame), Res; 2 -> PubRecFrame = #mqtt_pubrec{message_id=MessageId}, NewInfoFun1 = call_info_fun({pubrec_out, MessageId}, NewInfoFun), send_frame(Transport, Socket, PubRecFrame), {next_state, connected, State#state{waiting_acks=store({store, MessageId}, {Topic, Payload}, WAcks), info_fun=NewInfoFun1}} end; handle_frame(connected, #mqtt_pingresp{}, State) -> {next_state, connected, State}; handle_frame(connected, #mqtt_disconnect{}, State) -> gen_fsm:send_event_after(5000, connect), wrap_res(connecting, on_disconnect, [], State). handle_event(Event, StateName, State) -> wrap_res(StateName, handle_cast, [Event], State). handle_sync_event(Req, From, StateName, State) -> wrap_res(StateName, handle_call, [Req, From], State). terminate(Reason, StateName, State) -> wrap_res(StateName, terminate, [Reason], State). code_change(OldVsn, StateName, State, Extra) -> #state{mod=Mod, mod_state=ModState} = State, {ok, NewModState} = erlang:apply(Mod, code_change, [OldVsn, ModState, Extra]), {ok, StateName, State#state{mod_state=NewModState}}. %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% %%% INTERNAL %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% send_connect(State= #state{transport={Transport, _}, sock=Sock, username=Username, password=Password, client=ClientId, clean_session=CleanSession, last_will_topic=LWTopic, last_will_msg=LWMsg, last_will_retain=LWRet, last_will_qos=LWQoS, proto_version=ProtoVer, keepalive_interval=Int}) -> Frame = #mqtt_connect{ proto_ver = ProtoVer, username = Username, password = Password, clean_session = CleanSession, keep_alive = Int div 1000, client_id = ClientId, will_retain = LWRet, will_qos = case (LWTopic /= undefined) and (LWMsg /= undefined) of true when is_integer(LWQoS) -> LWQoS; _ -> 0 end, will_topic = LWTopic, will_msg = LWMsg }, send_frame(Transport, Sock, Frame), State. send_publish(MsgId, Topic, Payload, QoS, Retain, Dup, State) -> #state{transport={Transport, _}, sock = Sock, waiting_acks=WAcks, retry_interval=RetryInterval, info_fun=InfoFun} = State, Frame = #mqtt_publish{ message_id = if QoS == 0 -> undefined; true -> MsgId end, topic = Topic, qos = QoS, retain = Retain, dup = Dup, payload = Payload }, NewInfoFun = call_info_fun({publish_out, MsgId, QoS}, InfoFun), send_frame(Transport, Sock, Frame), case QoS of 0 -> State#state{info_fun=NewInfoFun}; _ -> Msg = {publish, MsgId, Topic, Payload, QoS, Retain, true}, Key = {publish, MsgId}, Ref = gen_fsm:send_event_after(RetryInterval, {retry, Key}), State#state{waiting_acks=store(Key, {Ref, Msg}, WAcks), info_fun=NewInfoFun} end. send_disconnect(Transport, Sock) -> send_frame(Transport, Sock, #mqtt_disconnect{}). send_ping(Transport, Sock) -> send_frame(Transport, Sock, #mqtt_pingreq{}). send_frame(Transport, Sock, Frame) -> case Transport:send(Sock, vmq_parser:serialise(Frame)) of ok -> ok; {error, _} -> gen_fsm:send_event(self(), maybe_reconnect) end. store(K,V,D) -> dict:store(K,V,D). erase(K,D) -> dict:erase(K,D). '++'(65535) -> 1; '++'(N) -> N + 1. active_once(gen_tcp, Sock) -> ok = inet:setopts(Sock, [{active, once}]); active_once(ssl, Sock) -> ok = ssl:setopts(Sock, [{active, once}]). call_info_fun(Info, {Fun, FunState}) -> {Fun, Fun(Info, FunState)}.