diff options
| -rw-r--r-- | docs/rabbitmqctl.1.xml | 23 | ||||
| -rw-r--r-- | src/file_handle_cache.erl | 2 | ||||
| -rw-r--r-- | src/gen_server2.erl | 3 | ||||
| -rw-r--r-- | src/rabbit_channel.erl | 5 | ||||
| -rw-r--r-- | src/rabbit_control.erl | 2 | ||||
| -rw-r--r-- | src/rabbit_limiter.erl | 19 | ||||
| -rw-r--r-- | src/rabbit_net.erl | 28 | ||||
| -rw-r--r-- | src/rabbit_networking.erl | 23 | ||||
| -rw-r--r-- | src/rabbit_plugin_activator.erl | 2 | ||||
| -rw-r--r-- | src/rabbit_reader.erl | 15 | ||||
| -rw-r--r-- | src/rabbit_ssl.erl | 173 | ||||
| -rw-r--r-- | src/rabbit_variable_queue.erl | 66 |
12 files changed, 296 insertions, 65 deletions
diff --git a/docs/rabbitmqctl.1.xml b/docs/rabbitmqctl.1.xml index 5179eb253c..73882861fc 100644 --- a/docs/rabbitmqctl.1.xml +++ b/docs/rabbitmqctl.1.xml @@ -967,6 +967,21 @@ <listitem><para>Peer port.</para></listitem> </varlistentry> <varlistentry> + <term>peer_cert_subject</term> + <listitem><para>The subject of the peer's SSL + certificate, in RFC4514 form.</para></listitem> + </varlistentry> + <varlistentry> + <term>peer_cert_issuer</term> + <listitem><para>The issuer of the peer's SSL + certificate, in RFC4514 form.</para></listitem> + </varlistentry> + <varlistentry> + <term>peer_cert_validity</term> + <listitem><para>The period for which the peer's SSL + certificate is valid.</para></listitem> + </varlistentry> + <varlistentry> <term>state</term> <listitem><para>Connection state (one of [<command>starting</command>, <command>tuning</command>, <command>opening</command>, <command>running</command>, <command>closing</command>, <command>closed</command>]).</para></listitem> @@ -1101,6 +1116,14 @@ <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/file_handle_cache.erl b/src/file_handle_cache.erl index d2830a25ca..6a948d49b6 100644 --- a/src/file_handle_cache.erl +++ b/src/file_handle_cache.erl @@ -1171,7 +1171,7 @@ ulimit() -> ?FILE_HANDLES_LIMIT_WINDOWS; {unix, _OsName} -> %% Under Linux, Solaris and FreeBSD, ulimit is a shell - %% builtin, not a command. In OS X, it's a command. + %% builtin, not a command. In OS X and AIX it's a command. %% Fortunately, os:cmd invokes the cmd in a shell env, so %% we're safe in all cases. case os:cmd("ulimit -n") of diff --git a/src/gen_server2.erl b/src/gen_server2.erl index b0379b95d1..230d1f2aa1 100644 --- a/src/gen_server2.erl +++ b/src/gen_server2.erl @@ -1128,7 +1128,8 @@ function_exported_or_default(Mod, Fun, Arity, Default) -> %%----------------------------------------------------------------- format_status(Opt, StatusData) -> [PDict, SysState, Parent, Debug, - [Name, State, Mod, _Time, _TimeoutState, Queue]] = StatusData, + #gs2_state{name = Name, state = State, mod = Mod, queue = Queue}] = + StatusData, NameTag = if is_pid(Name) -> pid_to_list(Name); is_atom(Name) -> diff --git a/src/rabbit_channel.erl b/src/rabbit_channel.erl index 40cb1a9503..20a554101c 100644 --- a/src/rabbit_channel.erl +++ b/src/rabbit_channel.erl @@ -60,7 +60,8 @@ consumer_count, messages_unacknowledged, acks_uncommitted, - prefetch_count]). + prefetch_count, + client_flow_blocked]). -define(CREATION_EVENT_KEYS, [pid, @@ -1291,6 +1292,8 @@ i(acks_uncommitted, #ch{uncommitted_ack_q = UAQ}) -> queue:len(UAQ); i(prefetch_count, #ch{limiter_pid = LimiterPid}) -> rabbit_limiter:get_limit(LimiterPid); +i(client_flow_blocked, #ch{limiter_pid = LimiterPid}) -> + rabbit_limiter:is_blocked(LimiterPid); i(Item, _) -> throw({bad_argument, Item}). diff --git a/src/rabbit_control.erl b/src/rabbit_control.erl index a3b6f369e3..57efe7cc51 100644 --- a/src/rabbit_control.erl +++ b/src/rabbit_control.erl @@ -347,6 +347,8 @@ format_info_item([{TableEntryKey, TableEntryType, _TableEntryValue} | _] = Value) when is_binary(TableEntryKey) andalso is_atom(TableEntryType) -> io_lib:format("~1000000000000p", [prettify_amqp_table(Value)]); +format_info_item([C|_] = Value) when is_number(C), C >= 32, C =< 255 -> + Value; format_info_item(Value) -> io_lib:format("~w", [Value]). diff --git a/src/rabbit_limiter.erl b/src/rabbit_limiter.erl index c323d7cef0..be1dcad1a6 100644 --- a/src/rabbit_limiter.erl +++ b/src/rabbit_limiter.erl @@ -37,7 +37,7 @@ handle_info/2, prioritise_call/3]). -export([start_link/2]). -export([limit/2, can_send/3, ack/2, register/2, unregister/2]). --export([get_limit/1, block/1, unblock/1]). +-export([get_limit/1, block/1, unblock/1, is_blocked/1]). %%---------------------------------------------------------------------------- @@ -55,6 +55,7 @@ -spec(get_limit/1 :: (maybe_pid()) -> non_neg_integer()). -spec(block/1 :: (maybe_pid()) -> 'ok'). -spec(unblock/1 :: (maybe_pid()) -> 'ok' | 'stopped'). +-spec(is_blocked/1 :: (maybe_pid()) -> boolean()). -endif. @@ -119,6 +120,11 @@ unblock(undefined) -> unblock(LimiterPid) -> gen_server2:call(LimiterPid, unblock, infinity). +is_blocked(undefined) -> + false; +is_blocked(LimiterPid) -> + gen_server2:call(LimiterPid, is_blocked, infinity). + %%---------------------------------------------------------------------------- %% gen_server callbacks %%---------------------------------------------------------------------------- @@ -157,7 +163,10 @@ handle_call(unblock, _From, State) -> case maybe_notify(State, State#lim{blocked = false}) of {cont, State1} -> {reply, ok, State1}; {stop, State1} -> {stop, normal, stopped, State1} - end. + end; + +handle_call(is_blocked, _From, State) -> + {reply, blocked(State), State}. handle_cast({ack, Count}, State = #lim{volume = Volume}) -> NewVolume = if Volume == 0 -> 0; @@ -186,8 +195,8 @@ code_change(_, State, _) -> %%---------------------------------------------------------------------------- maybe_notify(OldState, NewState) -> - case (limit_reached(OldState) orelse is_blocked(OldState)) andalso - not (limit_reached(NewState) orelse is_blocked(NewState)) of + case (limit_reached(OldState) orelse blocked(OldState)) andalso + not (limit_reached(NewState) orelse blocked(NewState)) of true -> NewState1 = notify_queues(NewState), {case NewState1#lim.prefetch_count of 0 -> stop; @@ -199,7 +208,7 @@ maybe_notify(OldState, NewState) -> limit_reached(#lim{prefetch_count = Limit, volume = Volume}) -> Limit =/= 0 andalso Volume >= Limit. -is_blocked(#lim{blocked = Blocked}) -> Blocked. +blocked(#lim{blocked = Blocked}) -> Blocked. remember_queue(QPid, State = #lim{queues = Queues}) -> case dict:is_key(QPid, Queues) of diff --git a/src/rabbit_net.erl b/src/rabbit_net.erl index 2286896b7b..53d0d5cbf3 100644 --- a/src/rabbit_net.erl +++ b/src/rabbit_net.erl @@ -33,7 +33,7 @@ -include("rabbit.hrl"). -export([async_recv/3, close/1, controlling_process/2, - getstat/2, peername/1, port_command/2, + getstat/2, peername/1, peercert/1, port_command/2, send/2, sockname/1]). %%--------------------------------------------------------------------------- @@ -45,28 +45,29 @@ -type(stat_option() :: 'recv_cnt' | 'recv_max' | 'recv_avg' | 'recv_oct' | 'recv_dvi' | 'send_cnt' | 'send_max' | 'send_avg' | 'send_oct' | 'send_pend'). --type(error() :: rabbit_types:error(any())). +-type(ok_val_or_error(A) :: rabbit_types:ok_or_error2(A, any())). +-type(ok_or_any_error() :: rabbit_types:ok_or_error(any())). -type(socket() :: port() | #ssl_socket{}). -spec(async_recv/3 :: (socket(), integer(), timeout()) -> rabbit_types:ok(any())). --spec(close/1 :: (socket()) -> rabbit_types:ok_or_error(any())). --spec(controlling_process/2 :: - (socket(), pid()) -> rabbit_types:ok_or_error(any())). +-spec(close/1 :: (socket()) -> ok_or_any_error()). +-spec(controlling_process/2 :: (socket(), pid()) -> ok_or_any_error()). -spec(port_command/2 :: (socket(), iolist()) -> 'true'). -spec(send/2 :: - (socket(), binary() | iolist()) -> rabbit_types:ok_or_error(any())). + (socket(), binary() | iolist()) -> ok_or_any_error()). -spec(peername/1 :: (socket()) - -> rabbit_types:ok({inet:ip_address(), rabbit_networking:ip_port()}) | - error()). + -> ok_val_or_error({inet:ip_address(), rabbit_networking:ip_port()})). +-spec(peercert/1 :: + (socket()) + -> 'nossl' | ok_val_or_error(rabbit_ssl:certificate())). -spec(sockname/1 :: (socket()) - -> rabbit_types:ok({inet:ip_address(), rabbit_networking:ip_port()}) | - error()). + -> ok_val_or_error({inet:ip_address(), rabbit_networking:ip_port()})). -spec(getstat/2 :: (socket(), [stat_option()]) - -> rabbit_types:ok([{stat_option(), integer()}]) | error()). + -> ok_val_or_error([{stat_option(), integer()}])). -endif. @@ -108,6 +109,11 @@ peername(Sock) when ?IS_SSL(Sock) -> peername(Sock) when is_port(Sock) -> inet:peername(Sock). +peercert(Sock) when ?IS_SSL(Sock) -> + ssl:peercert(Sock#ssl_socket.ssl); +peercert(Sock) when is_port(Sock) -> + nossl. + port_command(Sock, Data) when ?IS_SSL(Sock) -> case ssl:send(Sock#ssl_socket.ssl, Data) of ok -> self() ! {inet_reply, Sock, ok}, diff --git a/src/rabbit_networking.erl b/src/rabbit_networking.erl index 6dbd54d2bc..db5c71f62a 100644 --- a/src/rabbit_networking.erl +++ b/src/rabbit_networking.erl @@ -46,8 +46,6 @@ -include("rabbit.hrl"). -include_lib("kernel/include/inet.hrl"). --include_lib("ssl/src/ssl_record.hrl"). - -define(RABBIT_TCP_OPTS, [ binary, @@ -120,26 +118,7 @@ boot_ssl() -> end} | SslOptsConfig] end, - % In R13B04 and R14A (at least), rc4 is incorrectly implemented. - CipherSuites = proplists:get_value(ciphers, - SslOpts, - ssl:cipher_suites()), - FilteredCipherSuites = - [C || C <- CipherSuites, - begin - SuiteCode = - if is_tuple(C) -> ssl_cipher:suite(C); - is_list(C) -> ssl_cipher:openssl_suite(C) - end, - SP = ssl_cipher:security_parameters( - SuiteCode, - #security_parameters{}), - SP#security_parameters.bulk_cipher_algorithm =/= ?RC4 - end], - SslOpts1 = [{ciphers, FilteredCipherSuites} - | [{K, V} || {K, V} <- SslOpts, K =/= ciphers]], - [start_ssl_listener(Host, Port, SslOpts1) - || {Host, Port} <- SslListeners], + [start_ssl_listener(Host, Port, SslOpts) || {Host, Port} <- SslListeners], ok end. diff --git a/src/rabbit_plugin_activator.erl b/src/rabbit_plugin_activator.erl index b23776cd74..c38ef8d2f8 100644 --- a/src/rabbit_plugin_activator.erl +++ b/src/rabbit_plugin_activator.erl @@ -90,7 +90,7 @@ start() -> %% Compile the script ScriptFile = RootName ++ ".script", - case systools:make_script(RootName, [local, silent]) of + case systools:make_script(RootName, [local, silent, exref]) of {ok, Module, Warnings} -> %% This gets lots of spurious no-source warnings when we %% have .ez files, so we want to supress them to prevent diff --git a/src/rabbit_reader.erl b/src/rabbit_reader.erl index 745e008349..ff0fb8f778 100644 --- a/src/rabbit_reader.erl +++ b/src/rabbit_reader.erl @@ -66,6 +66,8 @@ send_pend, state, channels]). -define(CREATION_EVENT_KEYS, [pid, address, port, peer_address, peer_port, + peer_cert_subject, peer_cert_issuer, + peer_cert_validity, protocol, user, vhost, timeout, frame_max, client_properties]). @@ -824,6 +826,12 @@ i(peer_address, #v1{sock = Sock}) -> i(peer_port, #v1{sock = Sock}) -> {ok, {_, P}} = rabbit_net:peername(Sock), P; +i(peer_cert_issuer, #v1{sock = Sock}) -> + cert_info(fun rabbit_ssl:peer_cert_issuer/1, Sock); +i(peer_cert_subject, #v1{sock = Sock}) -> + cert_info(fun rabbit_ssl:peer_cert_subject/1, Sock); +i(peer_cert_validity, #v1{sock = Sock}) -> + cert_info(fun rabbit_ssl:peer_cert_validity/1, Sock); i(SockStat, #v1{sock = Sock}) when SockStat =:= recv_oct; SockStat =:= recv_cnt; SockStat =:= send_oct; @@ -858,6 +866,13 @@ i(client_properties, #v1{connection = #connection{ i(Item, #v1{}) -> throw({bad_argument, Item}). +cert_info(F, Sock) -> + case rabbit_net:peercert(Sock) of + nossl -> ''; + {error, no_peercert} -> ''; + {ok, Cert} -> F(Cert) + end. + %%-------------------------------------------------------------------------- send_to_new_channel(Channel, AnalyzedFrame, State) -> diff --git a/src/rabbit_ssl.erl b/src/rabbit_ssl.erl new file mode 100644 index 0000000000..be451af631 --- /dev/null +++ b/src/rabbit_ssl.erl @@ -0,0 +1,173 @@ +%% The contents of this file are subject to the Mozilla Public License +%% Version 1.1 (the "License"); you may not use this file except in +%% compliance with the License. You may obtain a copy of the License at +%% http://www.mozilla.org/MPL/ +%% +%% Software distributed under the License is distributed on an "AS IS" +%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See the +%% License for the specific language governing rights and limitations +%% under the License. +%% +%% The Original Code is RabbitMQ. +%% +%% The Initial Developers of the Original Code are LShift Ltd, +%% Cohesive Financial Technologies LLC, and Rabbit Technologies Ltd. +%% +%% Portions created before 22-Nov-2008 00:00:00 GMT by LShift Ltd, +%% Cohesive Financial Technologies LLC, or Rabbit Technologies Ltd +%% are Copyright (C) 2007-2008 LShift Ltd, Cohesive Financial +%% Technologies LLC, and Rabbit Technologies Ltd. +%% +%% Portions created by LShift Ltd are Copyright (C) 2007-2010 LShift +%% Ltd. Portions created by Cohesive Financial Technologies LLC are +%% Copyright (C) 2007-2010 Cohesive Financial Technologies +%% LLC. Portions created by Rabbit Technologies Ltd are Copyright +%% (C) 2007-2010 Rabbit Technologies Ltd. +%% +%% All Rights Reserved. +%% +%% Contributor(s): ______________________________________. +%% + +-module(rabbit_ssl). + +-include("rabbit.hrl"). + +-include_lib("public_key/include/public_key.hrl"). +-include_lib("ssl/src/ssl_int.hrl"). + +-export([peer_cert_issuer/1, peer_cert_subject/1, peer_cert_validity/1]). + +%%-------------------------------------------------------------------------- + +-ifdef(use_specs). + +-export_type([certificate/0]). + +-type(certificate() :: binary()). + +-spec(peer_cert_issuer/1 :: (certificate()) -> string()). +-spec(peer_cert_subject/1 :: (certificate()) -> string()). +-spec(peer_cert_validity/1 :: (certificate()) -> string()). + +-endif. + +%%-------------------------------------------------------------------------- +%% High-level functions used by reader +%%-------------------------------------------------------------------------- + +%% Return a string describing the certificate's issuer. +peer_cert_issuer(Cert) -> + cert_info(fun(#'OTPCertificate' { + tbsCertificate = #'OTPTBSCertificate' { + issuer = Issuer }}) -> + format_rdn_sequence(Issuer) + end, Cert). + +%% Return a string describing the certificate's subject, as per RFC4514. +peer_cert_subject(Cert) -> + cert_info(fun(#'OTPCertificate' { + tbsCertificate = #'OTPTBSCertificate' { + subject = Subject }}) -> + format_rdn_sequence(Subject) + end, Cert). + +%% Return a string describing the certificate's validity. +peer_cert_validity(Cert) -> + cert_info(fun(#'OTPCertificate' { + tbsCertificate = #'OTPTBSCertificate' { + validity = {'Validity', Start, End} }}) -> + lists:flatten( + io_lib:format("~s - ~s", [format_asn1_value(Start), + format_asn1_value(End)])) + end, Cert). + +%%-------------------------------------------------------------------------- + +cert_info(F, Cert) -> + F(case public_key:pkix_decode_cert(Cert, otp) of + {ok, DecCert} -> DecCert; + DecCert -> DecCert + end). + +%%-------------------------------------------------------------------------- +%% Formatting functions +%%-------------------------------------------------------------------------- + +%% Format and rdnSequence as a RFC4514 subject string. +format_rdn_sequence({rdnSequence, Seq}) -> + lists:flatten( + rabbit_misc:intersperse( + ",", lists:reverse([format_complex_rdn(RDN) || RDN <- Seq]))). + +%% Format an RDN set. +format_complex_rdn(RDNs) -> + lists:flatten( + rabbit_misc:intersperse("+", [format_rdn(RDN) || RDN <- RDNs])). + +%% Format an RDN. If the type name is unknown, use the dotted decimal +%% representation. See RFC4514, section 2.3. +format_rdn(#'AttributeTypeAndValue'{type = T, value = V}) -> + FV = escape_rdn_value(format_asn1_value(V)), + Fmts = [{?'id-at-surname' , "SN"}, + {?'id-at-givenName' , "GIVENNAME"}, + {?'id-at-initials' , "INITIALS"}, + {?'id-at-generationQualifier' , "GENERATIONQUALIFIER"}, + {?'id-at-commonName' , "CN"}, + {?'id-at-localityName' , "L"}, + {?'id-at-stateOrProvinceName' , "ST"}, + {?'id-at-organizationName' , "O"}, + {?'id-at-organizationalUnitName' , "OU"}, + {?'id-at-title' , "TITLE"}, + {?'id-at-countryName' , "C"}, + {?'id-at-serialNumber' , "SERIALNUMBER"}, + {?'id-at-pseudonym' , "PSEUDONYM"}, + {?'id-domainComponent' , "DC"}, + {?'id-emailAddress' , "EMAILADDRESS"}, + {?'street-address' , "STREET"}], + case proplists:lookup(T, Fmts) of + {_, Fmt} -> + io_lib:format(Fmt ++ "=~s", [FV]); + none when is_tuple(T) -> + TypeL = [io_lib:format("~w", [X]) || X <- tuple_to_list(T)], + io_lib:format("~s:~s", [rabbit_misc:intersperse(".", TypeL), FV]); + none -> + io_lib:format("~p:~s", [T, FV]) + end. + +%% Escape a string as per RFC4514. +escape_rdn_value(V) -> + escape_rdn_value(V, start). + +escape_rdn_value([], _) -> + []; +escape_rdn_value([C | S], start) when C =:= $ ; C =:= $# -> + [$\\, C | escape_rdn_value(S, middle)]; +escape_rdn_value(S, start) -> + escape_rdn_value(S, middle); +escape_rdn_value([$ ], middle) -> + [$\\, $ ]; +escape_rdn_value([C | S], middle) when C =:= $"; C =:= $+; C =:= $,; C =:= $;; + C =:= $<; C =:= $>; C =:= $\\ -> + [$\\, C | escape_rdn_value(S, middle)]; +escape_rdn_value([C | S], middle) when C < 32 ; C =:= 127 -> + %% only U+0000 needs escaping, but for display purposes it's handy + %% to escape all non-printable chars + lists:flatten(io_lib:format("\\~2.16.0B", [C])) ++ + escape_rdn_value(S, middle); +escape_rdn_value([C | S], middle) -> + [C | escape_rdn_value(S, middle)]. + +%% Get the string representation of an OTPCertificate field. +format_asn1_value({ST, S}) when ST =:= teletexString; ST =:= printableString; + ST =:= universalString; ST =:= utf8String; + ST =:= bmpString -> + if is_binary(S) -> binary_to_list(S); + true -> S + end; +format_asn1_value({utcTime, [Y1, Y2, M1, M2, D1, D2, H1, H2, + Min1, Min2, S1, S2, $Z]}) -> + io_lib:format("20~c~c-~c~c-~c~cT~c~c:~c~c:~c~cZ", + [Y1, Y2, M1, M2, D1, D2, H1, H2, Min1, Min2, S1, S2]); +format_asn1_value(V) -> + io_lib:format("~p", [V]). diff --git a/src/rabbit_variable_queue.erl b/src/rabbit_variable_queue.erl index 185617d57c..9b8dd23c99 100644 --- a/src/rabbit_variable_queue.erl +++ b/src/rabbit_variable_queue.erl @@ -493,23 +493,30 @@ delete_and_terminate(State) -> a(State2 #vqstate { index_state = IndexState1, msg_store_clients = undefined }). -purge(State = #vqstate { q4 = Q4, index_state = IndexState, len = Len }) -> +purge(State = #vqstate { q4 = Q4, + index_state = IndexState, + len = Len, + persistent_count = PCount }) -> %% TODO: when there are no pending acks, which is a common case, %% we could simply wipe the qi instead of issuing delivers and %% acks for all the messages. - IndexState1 = remove_queue_entries(fun rabbit_misc:queue_fold/3, Q4, - IndexState), - State1 = #vqstate { q1 = Q1, index_state = IndexState2 } = - purge_betas_and_deltas(State #vqstate { q4 = queue:new(), + {LensByStore, IndexState1} = remove_queue_entries( + fun rabbit_misc:queue_fold/3, Q4, + orddict:new(), IndexState), + {LensByStore1, State1 = #vqstate { q1 = Q1, index_state = IndexState2 }} = + purge_betas_and_deltas(LensByStore, + State #vqstate { q4 = queue:new(), index_state = IndexState1 }), - IndexState3 = remove_queue_entries(fun rabbit_misc:queue_fold/3, Q1, - IndexState2), + {LensByStore2, IndexState3} = remove_queue_entries( + fun rabbit_misc:queue_fold/3, Q1, + LensByStore1, IndexState2), + PCount1 = PCount - find_persistent_count(LensByStore2), {Len, a(State1 #vqstate { q1 = queue:new(), index_state = IndexState3, len = 0, ram_msg_count = 0, ram_index_count = 0, - persistent_count = 0 })}. + persistent_count = PCount1 })}. publish(Msg, NeedsConfirming, State) -> {_SeqId, State1} = publish(Msg, false, false, NeedsConfirming, State), @@ -996,26 +1003,30 @@ tx_commit_index(State = #vqstate { on_sync = #sync { reduce_memory_use( State1 #vqstate { index_state = IndexState1, on_sync = ?BLANK_SYNC }). -purge_betas_and_deltas(State = #vqstate { q3 = Q3, +purge_betas_and_deltas(LensByStore, + State = #vqstate { q3 = Q3, index_state = IndexState }) -> case bpqueue:is_empty(Q3) of - true -> State; - false -> IndexState1 = remove_queue_entries(fun beta_fold/3, Q3, - IndexState), - purge_betas_and_deltas( - maybe_deltas_to_betas( - State #vqstate { q3 = bpqueue:new(), - index_state = IndexState1 })) + true -> {LensByStore, State}; + false -> {LensByStore1, IndexState1} = remove_queue_entries( + fun beta_fold/3, Q3, + LensByStore, IndexState), + purge_betas_and_deltas(LensByStore1, + maybe_deltas_to_betas( + State #vqstate { + q3 = bpqueue:new(), + index_state = IndexState1 })) end. -remove_queue_entries(Fold, Q, IndexState) -> +remove_queue_entries(Fold, Q, LensByStore, IndexState) -> {GuidsByStore, Delivers, Acks} = Fold(fun remove_queue_entries1/2, {orddict:new(), [], []}, Q), ok = orddict:fold(fun (MsgStore, Guids, ok) -> rabbit_msg_store:remove(MsgStore, Guids) end, ok, GuidsByStore), - rabbit_queue_index:ack(Acks, - rabbit_queue_index:deliver(Delivers, IndexState)). + {sum_guids_by_store_to_len(LensByStore, GuidsByStore), + rabbit_queue_index:ack(Acks, + rabbit_queue_index:deliver(Delivers, IndexState))}. remove_queue_entries1( #msg_status { guid = Guid, seq_id = SeqId, @@ -1030,6 +1041,12 @@ remove_queue_entries1( cons_if(IndexOnDisk andalso not IsDelivered, SeqId, Delivers), cons_if(IndexOnDisk, SeqId, Acks)}. +sum_guids_by_store_to_len(LensByStore, GuidsByStore) -> + orddict:fold( + fun (MsgStore, Guids, LensByStore1) -> + orddict:update_counter(MsgStore, length(Guids), LensByStore1) + end, LensByStore, GuidsByStore). + %%---------------------------------------------------------------------------- %% Internal gubbins for publishing %%---------------------------------------------------------------------------- @@ -1167,10 +1184,8 @@ ack(MsgStoreFun, Fun, AckTags, State) -> %% the AckTags were removed from State1, so use State in seqids_to_guids State2 = msgs_confirmed(gb_sets:from_list(seqids_to_guids(AckTags, State)), State1), - PCount1 = PCount - case orddict:find(?PERSISTENT_MSG_STORE, GuidsByStore) of - error -> 0; - {ok, Guids} -> length(Guids) - end, + PCount1 = PCount - find_persistent_count(sum_guids_by_store_to_len( + orddict:new(), GuidsByStore)), State2 #vqstate { index_state = IndexState1, persistent_count = PCount1 }. @@ -1182,6 +1197,11 @@ accumulate_ack(SeqId, {IsPersistent, Guid}, {SeqIdsAcc, Dict}) -> {cons_if(IsPersistent, SeqId, SeqIdsAcc), rabbit_misc:orddict_cons(find_msg_store(IsPersistent), Guid, Dict)}. +find_persistent_count(LensByStore) -> + case orddict:find(?PERSISTENT_MSG_STORE, LensByStore) of + error -> 0; + {ok, Len} -> Len + end. %%---------------------------------------------------------------------------- %% Internal plumbing for confirms (aka publisher acks) |