diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/credit_flow.erl | 17 | ||||
| -rw-r--r-- | src/rabbit_amqqueue_process.erl | 2 | ||||
| -rw-r--r-- | src/rabbit_channel.erl | 2 | ||||
| -rw-r--r-- | src/rabbit_mirror_queue_slave.erl | 2 | ||||
| -rw-r--r-- | src/rabbit_reader.erl | 2 |
5 files changed, 11 insertions, 14 deletions
diff --git a/src/credit_flow.erl b/src/credit_flow.erl index 606441109d..7df6c92a6d 100644 --- a/src/credit_flow.erl +++ b/src/credit_flow.erl @@ -29,7 +29,7 @@ -define(MORE_CREDIT_AT, 150). -export([ack/1, handle_bump_msg/1, blocked/0, send/1]). --export([sender_down/1, receiver_down/1]). +-export([peer_down/1]). %%---------------------------------------------------------------------------- @@ -41,8 +41,7 @@ -spec(handle_bump_msg/1 :: (bump_msg()) -> 'ok'). -spec(blocked/0 :: () -> boolean()). -spec(send/1 :: (pid()) -> 'ok'). --spec(sender_down/1 :: (pid()) -> 'ok'). --spec(receiver_down/1 :: (pid()) -> 'ok'). +-spec(peer_down/1 :: (pid()) -> 'ok'). -endif. @@ -85,15 +84,13 @@ send(From) -> end, put({credit_from, From}, Credit). -sender_down(To) -> - %% In theory we could remove it from credit_deferred here, but it +peer_down(Peer) -> + %% In theory we could also remove it from credit_deferred here, but it %% doesn't really matter; at some point later we will drain %% credit_deferred and thus send messages into the void... - erase({credit_to, To}). - -receiver_down(From) -> - unblock(From), - erase({credit_from, From}). + unblock(Peer), + erase({credit_from, Peer}), + erase({credit_to, Peer}). %% -------------------------------------------------------------------------- diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl index 0a42673309..390c78cbd8 100644 --- a/src/rabbit_amqqueue_process.erl +++ b/src/rabbit_amqqueue_process.erl @@ -644,7 +644,7 @@ handle_ch_publisher_down(DownPid) -> case lookup_ch_publisher(DownPid) of not_found -> ok; _ -> erase_ch_record_publisher(DownPid), - credit_flow:sender_down(DownPid) + credit_flow:peer_down(DownPid) end. check_exclusive_access({_ChPid, _ConsumerTag}, _ExclusiveConsume, _State) -> diff --git a/src/rabbit_channel.erl b/src/rabbit_channel.erl index e695ac0e96..c8c319ee47 100644 --- a/src/rabbit_channel.erl +++ b/src/rabbit_channel.erl @@ -335,7 +335,7 @@ 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), - credit_flow:receiver_down(QPid), + credit_flow:peer_down(QPid), erase_queue_stats(QPid), noreply(State3#ch{queue_monitors = sets:del_element(QPid, State3#ch.queue_monitors)}); diff --git a/src/rabbit_mirror_queue_slave.erl b/src/rabbit_mirror_queue_slave.erl index 5a2a7e66ab..bc29b0806c 100644 --- a/src/rabbit_mirror_queue_slave.erl +++ b/src/rabbit_mirror_queue_slave.erl @@ -602,7 +602,7 @@ ensure_monitoring(ChPid, State = #state { known_senders = KS }) -> local_sender_death(ChPid, State = #state { known_senders = KS }) -> ok = case dict:is_key(ChPid, KS) of false -> ok; - true -> credit_flow:sender_down(ChPid), + true -> credit_flow:peer_down(ChPid), confirm_sender_death(ChPid) end, State. diff --git a/src/rabbit_reader.erl b/src/rabbit_reader.erl index e1e3eb6ab1..059f88ba54 100644 --- a/src/rabbit_reader.erl +++ b/src/rabbit_reader.erl @@ -419,7 +419,7 @@ handle_dependent_exit(ChPid, Reason, State) -> channel_cleanup(ChPid) -> case get({ch_pid, ChPid}) of undefined -> undefined; - {Channel, MRef} -> credit_flow:receiver_down(ChPid), + {Channel, MRef} -> credit_flow:peer_down(ChPid), erase({channel, Channel}), erase({ch_pid, ChPid}), erlang:demonitor(MRef, [flush]), |