diff options
author | Tim Watson <tim@rabbitmq.com> | 2014-01-22 14:56:28 +0000 |
---|---|---|
committer | Tim Watson <tim@rabbitmq.com> | 2014-01-22 14:56:28 +0000 |
commit | 6ff4b5be52e55629c9a3ffa034c4cc8e651e491a (patch) | |
tree | 128b88ade335110c99632297fa27cb66e20665c9 | |
parent | a96acf84a727312950c632ac1ecee2c1721312f0 (diff) | |
parent | aaec8e3c296797cc0c83a2ab2e408111df6f078e (diff) | |
download | rabbitmq-server-6ff4b5be52e55629c9a3ffa034c4cc8e651e491a.tar.gz |
merge default into bug25853
-rw-r--r-- | docs/rabbitmqctl.1.xml | 8 | ||||
-rw-r--r-- | src/rabbit_access_control.erl | 29 | ||||
-rw-r--r-- | src/rabbit_amqqueue.erl | 5 | ||||
-rw-r--r-- | src/rabbit_amqqueue_process.erl | 4 | ||||
-rw-r--r-- | src/rabbit_channel.erl | 102 | ||||
-rw-r--r-- | src/rabbit_limiter.erl | 66 | ||||
-rw-r--r-- | src/rabbit_reader.erl | 18 |
7 files changed, 71 insertions, 161 deletions
diff --git a/docs/rabbitmqctl.1.xml b/docs/rabbitmqctl.1.xml index d2a3f7c7..d19acd00 100644 --- a/docs/rabbitmqctl.1.xml +++ b/docs/rabbitmqctl.1.xml @@ -1559,14 +1559,6 @@ <term>prefetch_count</term> <listitem><para>QoS prefetch count limit in force, 0 if unlimited.</para></listitem> </varlistentry> - <varlistentry> - <term>client_flow_blocked</term> - <listitem><para>True if the client issued a - <command>channel.flow{active=false}</command> - command, blocking the server from delivering - messages to the channel's consumers. - </para></listitem> - </varlistentry> </variablelist> <para> If no <command>channelinfoitem</command>s are specified then pid, diff --git a/src/rabbit_access_control.erl b/src/rabbit_access_control.erl index d54c2a8d..19171659 100644 --- a/src/rabbit_access_control.erl +++ b/src/rabbit_access_control.erl @@ -52,18 +52,31 @@ check_user_pass_login(Username, Password) -> check_user_login(Username, AuthProps) -> {ok, Modules} = application:get_env(rabbit, auth_backends), lists:foldl( - fun(Module, {refused, _, _}) -> - case Module:check_user_login(Username, AuthProps) of - {error, E} -> - {refused, "~s failed authenticating ~s: ~p~n", - [Module, Username, E]}; - Else -> - Else + fun ({ModN, ModZ}, {refused, _, _}) -> + %% Different modules for authN vs authZ. So authenticate + %% with authN module, then if that succeeds do + %% passwordless (i.e pre-authenticated) login with authZ + %% module, and use the #user{} the latter gives us. + case try_login(ModN, Username, AuthProps) of + {ok, _} -> try_login(ModZ, Username, []); + Else -> Else end; - (_, {ok, User}) -> + (Mod, {refused, _, _}) -> + %% Same module for authN and authZ. Just take the result + %% it gives us + try_login(Mod, Username, AuthProps); + (_, {ok, User}) -> + %% We've successfully authenticated. Skip to the end... {ok, User} end, {refused, "No modules checked '~s'", [Username]}, Modules). +try_login(Module, Username, AuthProps) -> + case Module:check_user_login(Username, AuthProps) of + {error, E} -> {refused, "~s failed authenticating ~s: ~p~n", + [Module, Username, E]}; + Else -> Else + end. + check_vhost_access(User = #user{ username = Username, auth_backend = Module }, VHostPath) -> check_access( diff --git a/src/rabbit_amqqueue.erl b/src/rabbit_amqqueue.erl index 282113a4..c66c8981 100644 --- a/src/rabbit_amqqueue.erl +++ b/src/rabbit_amqqueue.erl @@ -27,7 +27,7 @@ -export([force_event_refresh/0, notify_policy_changed/1]). -export([consumers/1, consumers_all/1, consumer_info_keys/0]). -export([basic_get/4, basic_consume/10, basic_cancel/4, notify_decorators/1]). --export([notify_sent/2, notify_sent_queue_down/1, resume/2, flush_all/2]). +-export([notify_sent/2, notify_sent_queue_down/1, resume/2]). -export([notify_down_all/2, activate_limit_all/2, credit/5]). -export([on_node_down/1]). -export([update/2, store_queue/1, policy_changed/2]). @@ -160,7 +160,6 @@ -spec(notify_sent/2 :: (pid(), pid()) -> 'ok'). -spec(notify_sent_queue_down/1 :: (pid()) -> 'ok'). -spec(resume/2 :: (pid(), pid()) -> 'ok'). --spec(flush_all/2 :: (qpids(), pid()) -> 'ok'). -spec(internal_delete/1 :: (name()) -> rabbit_types:ok_or_error('not_found') | rabbit_types:connection_exit() | @@ -604,8 +603,6 @@ notify_sent_queue_down(QPid) -> resume(QPid, ChPid) -> delegate:cast(QPid, {resume, ChPid}). -flush_all(QPids, ChPid) -> delegate:cast(QPids, {flush, ChPid}). - internal_delete1(QueueName) -> ok = mnesia:delete({rabbit_queue, QueueName}), %% this 'guarded' delete prevents unnecessary writes to the mnesia diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl index 08509c96..ac4bfe8b 100644 --- a/src/rabbit_amqqueue_process.erl +++ b/src/rabbit_amqqueue_process.erl @@ -1081,10 +1081,6 @@ handle_cast({activate_limit, ChPid}, State) -> noreply(possibly_unblock(rabbit_queue_consumers:activate_limit_fun(), ChPid, State)); -handle_cast({flush, ChPid}, State) -> - ok = rabbit_channel:flushed(ChPid, self()), - noreply(State); - handle_cast({set_ram_duration_target, Duration}, State = #q{backing_queue = BQ, backing_queue_state = BQS}) -> BQS1 = BQ:set_ram_duration_target(Duration, BQS), diff --git a/src/rabbit_channel.erl b/src/rabbit_channel.erl index 39c2f0b0..7f9ff827 100644 --- a/src/rabbit_channel.erl +++ b/src/rabbit_channel.erl @@ -21,8 +21,7 @@ -behaviour(gen_server2). -export([start_link/11, do/2, do/3, do_flow/3, flush/1, shutdown/1]). --export([send_command/2, deliver/4, send_credit_reply/2, send_drained/2, - flushed/2]). +-export([send_command/2, deliver/4, send_credit_reply/2, send_drained/2]). -export([list/0, info_keys/0, info/1, info/2, info_all/0, info_all/1]). -export([refresh_config_local/0, ready_for_close/1]). -export([force_event_refresh/0]). @@ -37,7 +36,7 @@ conn_name, limiter, tx, next_tag, unacked_message_q, user, virtual_host, most_recently_declared_queue, queue_names, queue_monitors, consumer_mapping, - blocking, queue_consumers, delivering_queues, + queue_consumers, delivering_queues, queue_collector_pid, stats_timer, confirm_enabled, publish_seqno, unconfirmed, confirmed, capabilities, trace_state}). @@ -53,7 +52,6 @@ messages_uncommitted, acks_uncommitted, prefetch_count, - client_flow_blocked, state]). -define(CREATION_EVENT_KEYS, @@ -99,7 +97,6 @@ -spec(send_credit_reply/2 :: (pid(), non_neg_integer()) -> 'ok'). -spec(send_drained/2 :: (pid(), [{rabbit_types:ctag(), non_neg_integer()}]) -> 'ok'). --spec(flushed/2 :: (pid(), pid()) -> 'ok'). -spec(list/0 :: () -> [pid()]). -spec(list_local/0 :: () -> [pid()]). -spec(info_keys/0 :: () -> rabbit_types:info_keys()). @@ -149,9 +146,6 @@ send_credit_reply(Pid, Len) -> send_drained(Pid, CTagCredit) -> gen_server2:cast(Pid, {send_drained, CTagCredit}). -flushed(Pid, QPid) -> - gen_server2:cast(Pid, {flushed, QPid}). - list() -> rabbit_misc:append_rpc_all_nodes(rabbit_mnesia:cluster_nodes(running), rabbit_channel, list_local, []). @@ -213,7 +207,6 @@ init([Channel, ReaderPid, WriterPid, ConnPid, ConnName, Protocol, User, VHost, queue_names = dict:new(), queue_monitors = pmon:new(), consumer_mapping = dict:new(), - blocking = sets:new(), queue_consumers = dict:new(), delivering_queues = sets:new(), queue_collector_pid = CollectorPid, @@ -291,9 +284,6 @@ handle_cast({method, Method, Content, Flow}, {stop, {Reason, erlang:get_stacktrace()}, State} end; -handle_cast({flushed, QPid}, State) -> - {noreply, queue_blocked(QPid, State), hibernate}; - handle_cast(ready_for_close, State = #ch{state = closing, writer_pid = WriterPid}) -> ok = rabbit_writer:send_command_sync(WriterPid, #'channel.close_ok'{}), @@ -368,8 +358,7 @@ handle_info(emit_stats, State) -> handle_info({'DOWN', _MRef, process, QPid, Reason}, State) -> State1 = handle_publishing_queue_down(QPid, Reason, State), - State2 = queue_blocked(QPid, State1), - State3 = handle_consuming_queue_down(QPid, State2), + State3 = handle_consuming_queue_down(QPid, State1), State4 = handle_delivering_queue_down(QPid, State3), credit_flow:peer_down(QPid), #ch{queue_names = QNames, queue_monitors = QMons} = State4, @@ -480,9 +469,8 @@ check_resource_access(User, Resource, Perm) -> put(permission_cache, [V | CacheTail]) end. -clear_permission_cache() -> - erase(permission_cache), - ok. +clear_permission_cache() -> erase(permission_cache), + ok. check_configure_permitted(Resource, #ch{user = User}) -> check_resource_access(User, Resource, configure). @@ -537,8 +525,7 @@ name_to_resource(Type, NameBin, #ch{virtual_host = VHostPath}) -> rabbit_misc:r(VHostPath, Type, NameBin). expand_queue_name_shortcut(<<>>, #ch{most_recently_declared_queue = <<>>}) -> - rabbit_misc:protocol_error( - not_found, "no previously declared queue", []); + rabbit_misc:protocol_error(not_found, "no previously declared queue", []); expand_queue_name_shortcut(<<>>, #ch{most_recently_declared_queue = MRDQ}) -> MRDQ; expand_queue_name_shortcut(QueueNameBin, _) -> @@ -546,8 +533,7 @@ expand_queue_name_shortcut(QueueNameBin, _) -> expand_routing_key_shortcut(<<>>, <<>>, #ch{most_recently_declared_queue = <<>>}) -> - rabbit_misc:protocol_error( - not_found, "no previously declared queue", []); + rabbit_misc:protocol_error(not_found, "no previously declared queue", []); expand_routing_key_shortcut(<<>>, <<>>, #ch{most_recently_declared_queue = MRDQ}) -> MRDQ; @@ -601,20 +587,6 @@ check_name(Kind, NameBin = <<"amq.", _/binary>>) -> check_name(_Kind, NameBin) -> NameBin. -queue_blocked(QPid, State = #ch{blocking = Blocking}) -> - case sets:is_element(QPid, Blocking) of - false -> State; - true -> maybe_send_flow_ok( - State#ch{blocking = sets:del_element(QPid, Blocking)}) - end. - -maybe_send_flow_ok(State = #ch{blocking = Blocking}) -> - case sets:size(Blocking) of - 0 -> ok = send(#'channel.flow_ok'{active = false}, State); - _ -> ok - end, - State. - record_confirms([], State) -> State; record_confirms(MXs, State = #ch{confirmed = C}) -> @@ -884,8 +856,13 @@ handle_method(#'basic.qos'{prefetch_count = PrefetchCount}, %% unacked messages from basic.get too. Pretty obscure though. Limiter1 = rabbit_limiter:limit_prefetch(Limiter, PrefetchCount, queue:len(UAMQ)), - {reply, #'basic.qos_ok'{}, - maybe_limit_queues(Limiter, Limiter1, State#ch{limiter = Limiter1})}; + case ((not rabbit_limiter:is_active(Limiter)) andalso + rabbit_limiter:is_active(Limiter1)) of + true -> rabbit_amqqueue:activate_limit_all( + consumer_queues(State#ch.consumer_mapping), self()); + false -> ok + end, + {reply, #'basic.qos_ok'{}, State#ch{limiter = Limiter1}}; handle_method(#'basic.recover_async'{requeue = true}, _, State = #ch{unacked_message_q = UAMQ, limiter = Limiter}) -> @@ -1174,36 +1151,11 @@ handle_method(#'confirm.select'{nowait = NoWait}, _, State) -> return_ok(State#ch{confirm_enabled = true}, NoWait, #'confirm.select_ok'{}); -handle_method(#'channel.flow'{active = true}, - _, State = #ch{limiter = Limiter}) -> - Limiter1 = rabbit_limiter:unblock(Limiter), - {reply, #'channel.flow_ok'{active = true}, - maybe_limit_queues(Limiter, Limiter1, State#ch{limiter = Limiter1})}; - -handle_method(#'channel.flow'{active = false}, - _, State = #ch{consumer_mapping = Consumers, - limiter = Limiter}) -> - case rabbit_limiter:is_blocked(Limiter) of - true -> {noreply, maybe_send_flow_ok(State)}; - false -> Limiter1 = rabbit_limiter:block(Limiter), - State1 = maybe_limit_queues(Limiter, Limiter1, - State#ch{limiter = Limiter1}), - %% The semantics of channel.flow{active=false} - %% require that no messages are delivered after the - %% channel.flow_ok has been sent. We accomplish that - %% by "flushing" all messages in flight from the - %% consumer queues to us. To do this we tell all the - %% queues to invoke rabbit_channel:flushed/2, which - %% will send us a {flushed, ...} message that appears - %% *after* all the {deliver, ...} messages. We keep - %% track of all the QPids thus asked, and once all of - %% them have responded (or died) we send the - %% channel.flow_ok. - QPids = consumer_queues(Consumers), - ok = rabbit_amqqueue:flush_all(QPids, self()), - {noreply, maybe_send_flow_ok( - State1#ch{blocking = sets:from_list(QPids)})} - end; +handle_method(#'channel.flow'{active = true}, _, State) -> + {reply, #'channel.flow_ok'{active = true}, State}; + +handle_method(#'channel.flow'{active = false}, _, _State) -> + rabbit_misc:protocol_error(not_implemented, "active=false", []); handle_method(#'basic.credit'{consumer_tag = CTag, credit = Credit, @@ -1465,15 +1417,6 @@ foreach_per_queue(F, UAL) -> end, gb_trees:empty(), UAL), rabbit_misc:gb_trees_foreach(F, T). -maybe_limit_queues(OldLimiter, NewLimiter, State) -> - case ((not rabbit_limiter:is_active(OldLimiter)) andalso - rabbit_limiter:is_active(NewLimiter)) of - true -> Queues = consumer_queues(State#ch.consumer_mapping), - rabbit_amqqueue:activate_limit_all(Queues, self()); - false -> ok - end, - State. - consumer_queues(Consumers) -> lists:usort([QPid || {_Key, #amqqueue{pid = QPid}} <- dict:to_list(Consumers)]). @@ -1485,7 +1428,7 @@ consumer_queues(Consumers) -> notify_limiter(Limiter, Acked) -> %% optimisation: avoid the potentially expensive 'foldl' in the %% common case. - case rabbit_limiter:is_prefetch_limited(Limiter) of + case rabbit_limiter:is_active(Limiter) of false -> ok; true -> case lists:foldl(fun ({_, none, _}, Acc) -> Acc; ({_, _, _}, Acc) -> Acc + 1 @@ -1653,8 +1596,6 @@ i(state, #ch{state = running}) -> credit_flow:state(); i(state, #ch{state = State}) -> State; i(prefetch_count, #ch{limiter = Limiter}) -> rabbit_limiter:get_prefetch_limit(Limiter); -i(client_flow_blocked, #ch{limiter = Limiter}) -> - rabbit_limiter:is_blocked(Limiter); i(Item, _) -> throw({bad_argument, Item}). @@ -1675,8 +1616,7 @@ update_measures(Type, Key, Inc, Measure) -> end, put({Type, Key}, orddict:store(Measure, Cur + Inc, Measures)). -emit_stats(State) -> - emit_stats(State, []). +emit_stats(State) -> emit_stats(State, []). emit_stats(State, Extra) -> Coarse = infos(?STATISTICS_KEYS, State), diff --git a/src/rabbit_limiter.erl b/src/rabbit_limiter.erl index da728033..d37b356c 100644 --- a/src/rabbit_limiter.erl +++ b/src/rabbit_limiter.erl @@ -17,8 +17,7 @@ %% The purpose of the limiter is to stem the flow of messages from %% queues to channels, in order to act upon various protocol-level %% flow control mechanisms, specifically AMQP 0-9-1's basic.qos -%% prefetch_count and channel.flow, and AMQP 1.0's link (aka consumer) -%% credit mechanism. +%% prefetch_count and AMQP 1.0's link (aka consumer) credit mechanism. %% %% Each channel has an associated limiter process, created with %% start_link/1, which it passes to queues on consumer creation with @@ -65,11 +64,9 @@ %% %% 1. Channels tell the limiter about basic.qos prefetch counts - %% that's what the limit_prefetch/3, unlimit_prefetch/1, -%% is_prefetch_limited/1, get_prefetch_limit/1 API functions are -%% about - and channel.flow blocking - that's what block/1, -%% unblock/1 and is_blocked/1 are for. They also tell the limiter -%% queue state (via the queue) about consumer credit changes - -%% that's what credit/5 is for. +%% get_prefetch_limit/1 API functions are about. They also tell the +%% limiter queue state (via the queue) about consumer credit +%% changes - that's what credit/5 is for. %% %% 2. Queues also tell the limiter queue state about the queue %% becoming empty (via drained/1) and consumers leaving (via @@ -83,12 +80,11 @@ %% %% 5. Queues ask the limiter for permission (with can_send/3) whenever %% they want to deliver a message to a channel. The limiter checks -%% whether a) the channel isn't blocked by channel.flow, b) the -%% volume has not yet reached the prefetch limit, and c) whether -%% the consumer has enough credit. If so it increments the volume -%% and tells the queue to proceed. Otherwise it marks the queue as -%% requiring notification (see below) and tells the queue not to -%% proceed. +%% whether a) the volume has not yet reached the prefetch limit, +%% and b) whether the consumer has enough credit. If so it +%% increments the volume and tells the queue to proceed. Otherwise +%% it marks the queue as requiring notification (see below) and +%% tells the queue not to proceed. %% %% 6. A queue that has been told to proceed (by the return value of %% can_send/3) sends the message to the channel. Conversely, a @@ -123,8 +119,7 @@ -export([start_link/1]). %% channel API --export([new/1, limit_prefetch/3, unlimit_prefetch/1, block/1, unblock/1, - is_prefetch_limited/1, is_blocked/1, is_active/1, +-export([new/1, limit_prefetch/3, unlimit_prefetch/1, is_active/1, get_prefetch_limit/1, ack/2, pid/1]). %% queue API -export([client/1, activate/1, can_send/3, resume/1, deactivate/1, @@ -136,14 +131,13 @@ %%---------------------------------------------------------------------------- --record(lstate, {pid, prefetch_limited, blocked}). +-record(lstate, {pid, prefetch_limited}). -record(qstate, {pid, state, credits}). -ifdef(use_specs). -type(lstate() :: #lstate{pid :: pid(), - prefetch_limited :: boolean(), - blocked :: boolean()}). + prefetch_limited :: boolean()}). -type(qstate() :: #qstate{pid :: pid(), state :: 'dormant' | 'active' | 'suspended'}). @@ -154,10 +148,6 @@ -spec(limit_prefetch/3 :: (lstate(), non_neg_integer(), non_neg_integer()) -> lstate()). -spec(unlimit_prefetch/1 :: (lstate()) -> lstate()). --spec(block/1 :: (lstate()) -> lstate()). --spec(unblock/1 :: (lstate()) -> lstate()). --spec(is_prefetch_limited/1 :: (lstate()) -> boolean()). --spec(is_blocked/1 :: (lstate()) -> boolean()). -spec(is_active/1 :: (lstate()) -> boolean()). -spec(get_prefetch_limit/1 :: (lstate()) -> non_neg_integer()). -spec(ack/2 :: (lstate(), non_neg_integer()) -> 'ok'). @@ -183,7 +173,6 @@ -record(lim, {prefetch_count = 0, ch_pid, - blocked = false, queues = orddict:new(), % QPid -> {MonitorRef, Notify} volume = 0}). %% 'Notify' is a boolean that indicates whether a queue should be @@ -201,7 +190,7 @@ start_link(ProcName) -> gen_server2:start_link(?MODULE, [ProcName], []). new(Pid) -> %% this a 'call' to ensure that it is invoked at most once. ok = gen_server:call(Pid, {new, self()}, infinity), - #lstate{pid = Pid, prefetch_limited = false, blocked = false}. + #lstate{pid = Pid, prefetch_limited = false}. limit_prefetch(L, PrefetchCount, UnackedCount) when PrefetchCount > 0 -> ok = gen_server:call( @@ -213,19 +202,7 @@ unlimit_prefetch(L) -> ok = gen_server:call(L#lstate.pid, unlimit_prefetch, infinity), L#lstate{prefetch_limited = false}. -block(L) -> - ok = gen_server:call(L#lstate.pid, block, infinity), - L#lstate{blocked = true}. - -unblock(L) -> - ok = gen_server:call(L#lstate.pid, unblock, infinity), - L#lstate{blocked = false}. - -is_prefetch_limited(#lstate{prefetch_limited = Limited}) -> Limited. - -is_blocked(#lstate{blocked = Blocked}) -> Blocked. - -is_active(L) -> is_prefetch_limited(L) orelse is_blocked(L). +is_active(#lstate{prefetch_limited = Limited}) -> Limited. get_prefetch_limit(#lstate{prefetch_limited = false}) -> 0; get_prefetch_limit(L) -> @@ -349,19 +326,10 @@ handle_call(unlimit_prefetch, _From, State) -> {reply, ok, maybe_notify(State, State#lim{prefetch_count = 0, volume = 0})}; -handle_call(block, _From, State) -> - {reply, ok, State#lim{blocked = true}}; - -handle_call(unblock, _From, State) -> - {reply, ok, maybe_notify(State, State#lim{blocked = false})}; - handle_call(get_prefetch_limit, _From, State = #lim{prefetch_count = PrefetchCount}) -> {reply, PrefetchCount, State}; -handle_call({can_send, QPid, _AckRequired}, _From, - State = #lim{blocked = true}) -> - {reply, false, limit_queue(QPid, State)}; handle_call({can_send, QPid, AckRequired}, _From, State = #lim{volume = Volume}) -> case prefetch_limit_reached(State) of @@ -397,8 +365,8 @@ code_change(_, State, _) -> %%---------------------------------------------------------------------------- maybe_notify(OldState, NewState) -> - case (prefetch_limit_reached(OldState) orelse blocked(OldState)) andalso - not (prefetch_limit_reached(NewState) orelse blocked(NewState)) of + case prefetch_limit_reached(OldState) andalso + not prefetch_limit_reached(NewState) of true -> notify_queues(NewState); false -> NewState end. @@ -406,8 +374,6 @@ maybe_notify(OldState, NewState) -> prefetch_limit_reached(#lim{prefetch_count = Limit, volume = Volume}) -> Limit =/= 0 andalso Volume >= Limit. -blocked(#lim{blocked = Blocked}) -> Blocked. - remember_queue(QPid, State = #lim{queues = Queues}) -> case orddict:is_key(QPid, Queues) of false -> MRef = erlang:monitor(process, QPid), diff --git a/src/rabbit_reader.erl b/src/rabbit_reader.erl index 8553e36d..64debcab 100644 --- a/src/rabbit_reader.erl +++ b/src/rabbit_reader.erl @@ -285,8 +285,11 @@ recvloop(Deb, Buf, BufLen, State = #v1{connection_state = {become, F}}) -> throw({become, F(Deb, Buf, BufLen, State)}); recvloop(Deb, Buf, BufLen, State = #v1{sock = Sock, recv_len = RecvLen}) when BufLen < RecvLen -> - ok = rabbit_net:setopts(Sock, [{active, once}]), - mainloop(Deb, Buf, BufLen, State#v1{pending_recv = true}); + case rabbit_net:setopts(Sock, [{active, once}]) of + ok -> mainloop(Deb, Buf, BufLen, + State#v1{pending_recv = true}); + {error, Reason} -> stop(Reason, State) + end; recvloop(Deb, [B], _BufLen, State) -> {Rest, State1} = handle_input(State#v1.callback, B, State), recvloop(Deb, [Rest], size(Rest), State1); @@ -312,11 +315,9 @@ mainloop(Deb, Buf, BufLen, State = #v1{sock = Sock}) -> closed when State#v1.connection_state =:= closed -> ok; closed -> - maybe_emit_stats(State), - throw(connection_closed_abruptly); + stop(closed, State); {error, Reason} -> - maybe_emit_stats(State), - throw({inet_error, Reason}); + stop(Reason, State); {other, {system, From, Request}} -> sys:handle_system_msg(Request, From, State#v1.parent, ?MODULE, Deb, {Buf, BufLen, State}); @@ -327,6 +328,11 @@ mainloop(Deb, Buf, BufLen, State = #v1{sock = Sock}) -> end end. +stop(closed, State) -> maybe_emit_stats(State), + throw(connection_closed_abruptly); +stop(Reason, State) -> maybe_emit_stats(State), + throw({inet_error, Reason}). + handle_other({conserve_resources, Source, Conserve}, State = #v1{throttle = Throttle = #throttle{alarmed_by = CR}}) -> |