diff --git a/src/emqttd_client.erl b/src/emqttd_client.erl index 6631b4566..63bac7c8d 100644 --- a/src/emqttd_client.erl +++ b/src/emqttd_client.erl @@ -140,7 +140,8 @@ send_fun(Conn, Peername) -> ?LOG(debug, "SEND ~p", [Data], #client_state{peername = Peername}), emqttd_metrics:inc('bytes/sent', iolist_size(Data)), try Conn:async_send(Data) of - true -> ok + ok -> ok; + {error, Reason} -> Self ! {shutdown, Reason} catch error:Error -> Self ! {shutdown, Error} end diff --git a/src/emqttd_protocol.erl b/src/emqttd_protocol.erl index 384f93225..c35e8ae50 100644 --- a/src/emqttd_protocol.erl +++ b/src/emqttd_protocol.erl @@ -341,13 +341,11 @@ send(Msg, State = #proto_state{client_id = ClientId, emqttd_hooks:run('message.delivered', [ClientId, Username], Msg), send(emqttd_message:to_packet(unmount(MountPoint, clean_retain(IsBridge, Msg))), State); -send(Packet = ?PACKET(Type), - State = #proto_state{sendfun = SendFun, stats_data = Stats}) -> +send(Packet = ?PACKET(Type), State = #proto_state{sendfun = SendFun, stats_data = Stats}) -> trace(send, Packet, State), emqttd_metrics:sent(Packet), SendFun(Packet), - Stats1 = inc_stats(send, Type, Stats), - {ok, State#proto_state{stats_data = Stats1}}. + {ok, State#proto_state{stats_data = inc_stats(send, Type, Stats)}}. trace(recv, Packet, ProtoState) -> ?LOG(debug, "RECV ~s", [emqttd_packet:format(Packet)], ProtoState); diff --git a/src/emqttd_ws.erl b/src/emqttd_ws.erl index c7d0b2119..35a7f9852 100644 --- a/src/emqttd_ws.erl +++ b/src/emqttd_ws.erl @@ -38,6 +38,7 @@ handle_request(Req) -> %%-------------------------------------------------------------------- %% MQTT Over WebSocket %%-------------------------------------------------------------------- + handle_request('GET', "/mqtt", Req) -> lager:debug("WebSocket Connection from: ~s", [Req:get(peer)]), Upgrade = Req:get_header_value("Upgrade"), diff --git a/src/emqttd_ws_client.erl b/src/emqttd_ws_client.erl index b9d25ad3e..206f461bb 100644 --- a/src/emqttd_ws_client.erl +++ b/src/emqttd_ws_client.erl @@ -272,10 +272,14 @@ code_change(_OldVsn, State, _Extra) -> %%-------------------------------------------------------------------- send_fun(ReplyChannel) -> + Self = self(), fun(Packet) -> Data = emqttd_serializer:serialize(Packet), emqttd_metrics:inc('bytes/sent', iolist_size(Data)), - ReplyChannel({binary, Data}) + case ReplyChannel({binary, Data}) of + ok -> ok; + {error, Reason} -> Self ! {shutdown, Reason} + end end. stat_fun(Conn) ->