diff options
| -rw-r--r-- | src/rabbit_networking.erl | 20 | ||||
| -rw-r--r-- | src/rabbit_reader.erl | 2 |
2 files changed, 20 insertions, 2 deletions
diff --git a/src/rabbit_networking.erl b/src/rabbit_networking.erl index a44b3fa334..8b201fd87e 100644 --- a/src/rabbit_networking.erl +++ b/src/rabbit_networking.erl @@ -34,7 +34,8 @@ connections/0, connection_info_keys/0, connection_info/1, connection_info/2, connection_info_all/0, connection_info_all/1, connection_info_all/3, - close_connection/2, force_connection_event_refresh/1, tcp_host/1]). + close_connection/2, force_connection_event_refresh/1, accept_ack/2, + tcp_host/1]). %% Used by TCP-based transports, e.g. STOMP adapter -export([tcp_listener_addresses/1, tcp_listener_spec/9, @@ -87,6 +88,7 @@ 'ok'. -spec close_connection(pid(), string()) -> 'ok'. -spec force_connection_event_refresh(reference()) -> 'ok'. +-spec accept_ack(any(), rabbit_net:socket()) -> ok. -spec on_node_down(node()) -> 'ok'. -spec tcp_listener_addresses(listener_config()) -> [address()]. @@ -313,6 +315,22 @@ force_connection_event_refresh(Ref) -> [rabbit_reader:force_event_refresh(C, Ref) || C <- connections()], ok. +accept_ack(Ref, Sock) -> + ok = ranch:accept_ack(Ref), + case tune_buffer_size(Sock) of + ok -> ok; + {error, _} -> rabbit_net:fast_close(Sock), + exit(normal) + end, + ok = file_handle_cache:obtain(). + +tune_buffer_size(Sock) -> + case rabbit_net:getopts(Sock, [sndbuf, recbuf, buffer]) of + {ok, BufSizes} -> BufSz = lists:max([Sz || {_Opt, Sz} <- BufSizes]), + rabbit_net:setopts(Sock, [{buffer, BufSz}]); + Error -> Error + end. + %%-------------------------------------------------------------------- tcp_host(IPAddress) -> diff --git a/src/rabbit_reader.erl b/src/rabbit_reader.erl index b10e8b15aa..12874f784d 100644 --- a/src/rabbit_reader.erl +++ b/src/rabbit_reader.erl @@ -231,7 +231,7 @@ shutdown(Pid, Explanation) -> gen_server:call(Pid, {shutdown, Explanation}, infinity). init(Parent, HelperSup, Ref, Sock) -> - rabbit_net:accept_ack(Ref, Sock), + rabbit_networking:accept_ack(Ref, Sock), Deb = sys:debug_options([]), start_connection(Parent, HelperSup, Deb, Sock). |