diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/rabbit_amqqueue_process.erl | 15 | ||||
| -rw-r--r-- | src/rabbit_misc.erl | 15 | ||||
| -rw-r--r-- | src/rabbit_reader.erl | 13 |
3 files changed, 31 insertions, 12 deletions
diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl index 3418c663f4..7c7e28fe9e 100644 --- a/src/rabbit_amqqueue_process.erl +++ b/src/rabbit_amqqueue_process.erl @@ -617,6 +617,10 @@ maybe_send_reply(ChPid, Msg) -> ok = rabbit_channel:send_command(ChPid, Msg). qname(#q{q = #amqqueue{name = QName}}) -> QName. +backing_queue_idle_timeout(State = #q{backing_queue = BQ}) -> + maybe_run_queue_via_backing_queue( + fun (BQS) -> {[], BQ:idle_timeout(BQS)} end, State). + maybe_run_queue_via_backing_queue(Fun, State = #q{backing_queue_state = BQS}) -> {Guids, BQS1} = Fun(BQS), run_message_queue( @@ -996,10 +1000,8 @@ handle_call({maybe_run_queue_via_backing_queue, Fun}, _From, State) -> handle_cast({maybe_run_queue_via_backing_queue, Fun}, State) -> noreply(maybe_run_queue_via_backing_queue(Fun, State)); -handle_cast(sync_timeout, State = #q{backing_queue = BQ, - backing_queue_state = BQS}) -> - noreply(State#q{backing_queue_state = BQ:idle_timeout(BQS), - sync_timer_ref = undefined}); +handle_cast(sync_timeout, State) -> + noreply(backing_queue_idle_timeout(State#q{sync_timer_ref = undefined})); handle_cast({deliver, Delivery}, State) -> %% Asynchronous, non-"mandatory", non-"immediate" deliver mode. @@ -1133,9 +1135,8 @@ handle_info({'DOWN', _MonitorRef, process, DownPid, _Reason}, State) -> {stop, NewState} -> {stop, normal, NewState} end; -handle_info(timeout, State = #q{backing_queue = BQ}) -> - noreply(maybe_run_queue_via_backing_queue( - fun (BQS) -> {[], BQ:idle_timeout(BQS)} end, State)); +handle_info(timeout, State) -> + noreply(backing_queue_idle_timeout(State)); handle_info({'EXIT', _Pid, Reason}, State) -> {stop, Reason, State}; diff --git a/src/rabbit_misc.erl b/src/rabbit_misc.erl index 3a4fb024fe..7d9167973d 100644 --- a/src/rabbit_misc.erl +++ b/src/rabbit_misc.erl @@ -240,11 +240,20 @@ assert_args_equivalence1(Orig, New, Name, Key) -> {Same, Same} -> ok; {Orig1, New1} -> protocol_error( precondition_failed, - "inequivalent arg '~s' for ~s: " - "required ~w, received ~w", - [Key, rabbit_misc:rs(Name), New1, Orig1]) + "inequivalent arg '~s' for ~s: " + "received ~s but current is ~s", + [Key, rs(Name), val(New1), val(Orig1)]) end. +val(undefined) -> + "none"; +val({Type, Value}) -> + Fmt = case is_binary(Value) of + true -> "the value '~s' of type '~s'"; + false -> "the value '~w' of type '~s'" + end, + lists:flatten(io_lib:format(Fmt, [Value, Type])). + dirty_read(ReadSpec) -> case mnesia:dirty_read(ReadSpec) of [Result] -> {ok, Result}; diff --git a/src/rabbit_reader.erl b/src/rabbit_reader.erl index 475c415ecf..4bb87f19d5 100644 --- a/src/rabbit_reader.erl +++ b/src/rabbit_reader.erl @@ -72,7 +72,13 @@ %% pre-init: %% receive protocol header -> send connection.start, *starting* %% starting: -%% receive connection.start_ok -> send connection.tune, *tuning* +%% receive connection.start_ok -> *securing* +%% securing: +%% check authentication credentials +%% if authentication success -> send connection.tune, *tuning* +%% if more challenge needed -> send connection.secure, +%% receive connection.secure_ok *securing* +%% otherwise send close, *exit* %% tuning: %% receive connection.tune_ok -> start heartbeats, *opening* %% opening: @@ -351,7 +357,10 @@ mainloop(Deb, State = #v1{parent = Parent, sock= Sock, recv_ref = Ref}) -> throw({handshake_timeout, State#v1.callback}) end; timeout -> - throw({timeout, State#v1.connection_state}); + case State#v1.connection_state of + closed -> mainloop(Deb, State); + S -> throw({timeout, S}) + end; {'$gen_call', From, {shutdown, Explanation}} -> {ForceTermination, NewState} = terminate(Explanation, State), gen_server:reply(From, ok), |