summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorAlexandru Scvortov <alexandru@rabbitmq.com>2010-11-01 13:59:04 +0000
committerAlexandru Scvortov <alexandru@rabbitmq.com>2010-11-01 13:59:04 +0000
commit9e1f0572d052463b7ab94d516c241c3e69191dca (patch)
tree3f38e42f7301cb0abfc7784e8cc1388e103b071c
parentf5a19c98ec202880fe0508b503903df26f101a5e (diff)
parent0a57389f4429357aaaacc663c28ae4a02034e215 (diff)
downloadrabbitmq-server-git-9e1f0572d052463b7ab94d516c241c3e69191dca.tar.gz
merge from default
Diffstat
-rw-r--r--docs/html-to-website-xml.xsl2
-rw-r--r--docs/rabbitmq-server.1.xml12
-rw-r--r--docs/rabbitmq-service.xml12
-rw-r--r--docs/rabbitmqctl.1.xml12
-rw-r--r--docs/remove-namespaces.xsl3
-rw-r--r--ebin/rabbit_app.in2
-rw-r--r--include/rabbit.hrl4
-rw-r--r--packaging/macports/Portfile.in2
-rwxr-xr-xscripts/rabbitmq-server2
-rw-r--r--scripts/rabbitmq-server.bat1
-rw-r--r--scripts/rabbitmq-service.bat1
-rw-r--r--src/delegate.erl23
-rw-r--r--src/delegate_sup.erl10
-rw-r--r--src/rabbit_access_control.erl75
-rw-r--r--src/rabbit_control.erl6
-rw-r--r--src/rabbit_mnesia.erl6
-rw-r--r--src/rabbit_msg_store.erl120
-rw-r--r--src/rabbit_networking.erl25
-rw-r--r--src/rabbit_queue_index.erl48
-rw-r--r--src/rabbit_reader.erl29
-rw-r--r--src/rabbit_ssl.erl1
-rw-r--r--src/rabbit_tests.erl245
-rw-r--r--src/rabbit_variable_queue.erl451
23 files changed, 590 insertions, 502 deletions
diff --git a/docs/html-to-website-xml.xsl b/docs/html-to-website-xml.xsl
index 3e7a53261a..ec8f87e5eb 100644
--- a/docs/html-to-website-xml.xsl
+++ b/docs/html-to-website-xml.xsl
@@ -17,7 +17,7 @@
<!-- Copy the root node, and munge the outer part of the page -->
<xsl:template match="/html">
<xsl:processing-instruction name="xml-stylesheet">type="text/xml" href="page.xsl"</xsl:processing-instruction>
-<html xmlns:doc="http://www.rabbitmq.com/namespaces/ad-hoc/doc">
+<html xmlns:doc="http://www.rabbitmq.com/namespaces/ad-hoc/doc" xmlns="http://www.w3.org/1999/xhtml">
<head>
<title><xsl:value-of select="document($original)/refentry/refnamediv/refname"/><xsl:if test="document($original)/refentry/refmeta/manvolnum">(<xsl:value-of select="document($original)/refentry/refmeta/manvolnum"/>)</xsl:if> manual page</title>
</head>
diff --git a/docs/rabbitmq-server.1.xml b/docs/rabbitmq-server.1.xml
index 921da4f1cc..03e76c795b 100644
--- a/docs/rabbitmq-server.1.xml
+++ b/docs/rabbitmq-server.1.xml
@@ -98,18 +98,6 @@ Defaults to 5672.
</listitem>
</varlistentry>
- <varlistentry>
- <term>RABBITMQ_CLUSTER_CONFIG_FILE</term>
- <listitem>
- <para>
-Defaults to <filename>/etc/rabbitmq/rabbitmq_cluster.config</filename>. If this file is
-present it is used by the server to auto-configure a RabbitMQ cluster.
-See the <ulink url="http://www.rabbitmq.com/clustering.html">clustering guide</ulink>
-for details.
- </para>
- </listitem>
- </varlistentry>
-
</variablelist>
</refsect1>
diff --git a/docs/rabbitmq-service.xml b/docs/rabbitmq-service.xml
index 2b416e3e1e..e95f9889dd 100644
--- a/docs/rabbitmq-service.xml
+++ b/docs/rabbitmq-service.xml
@@ -193,18 +193,6 @@ manager.
</varlistentry>
<varlistentry>
- <term>RABBITMQ_CLUSTER_CONFIG_FILE</term>
- <listitem>
- <para>
-If this file is
-present it is used by the server to auto-configure a RabbitMQ cluster.
-See the <ulink url="http://www.rabbitmq.com/clustering.html">clustering guide</ulink>
-for details.
- </para>
- </listitem>
- </varlistentry>
-
- <varlistentry>
<term>RABBITMQ_CONSOLE_LOG</term>
<listitem>
<para>
diff --git a/docs/rabbitmqctl.1.xml b/docs/rabbitmqctl.1.xml
index 3b7244c701..acb99bc848 100644
--- a/docs/rabbitmqctl.1.xml
+++ b/docs/rabbitmqctl.1.xml
@@ -589,7 +589,7 @@
</varlistentry>
<varlistentry>
- <term><cmdsynopsis><command>set_permissions</command> <arg choice="opt">-p <replaceable>vhostpath</replaceable></arg> <arg choice="opt">-s <replaceable>scope</replaceable></arg> <arg choice="req"><replaceable>user</replaceable></arg> <arg choice="req"><replaceable>conf</replaceable></arg> <arg choice="req"><replaceable>write</replaceable></arg> <arg choice="req"><replaceable>read</replaceable></arg></cmdsynopsis></term>
+ <term><cmdsynopsis><command>set_permissions</command> <arg choice="opt">-p <replaceable>vhostpath</replaceable></arg> <arg choice="req"><replaceable>user</replaceable></arg> <arg choice="req"><replaceable>conf</replaceable></arg> <arg choice="req"><replaceable>write</replaceable></arg> <arg choice="req"><replaceable>read</replaceable></arg></cmdsynopsis></term>
<listitem>
<variablelist>
<varlistentry>
@@ -597,16 +597,6 @@
<listitem><para>The name of the virtual host to which to grant the user access, defaulting to <command>/</command>.</para></listitem>
</varlistentry>
<varlistentry>
- <term>scope</term>
- <listitem><para>Scope of the permissions: either
- <command>client</command> (the default) or
- <command>all</command>. This determines whether
- permissions are checked for server-generated resource
- names (<command>all</command>) or only for
- client-specified resource names
- (<command>client</command>).</para></listitem>
- </varlistentry>
- <varlistentry>
<term>user</term>
<listitem><para>The name of the user to grant access to the specified virtual host.</para></listitem>
</varlistentry>
diff --git a/docs/remove-namespaces.xsl b/docs/remove-namespaces.xsl
index 58a1e826d2..7f7f3c12b2 100644
--- a/docs/remove-namespaces.xsl
+++ b/docs/remove-namespaces.xsl
@@ -1,13 +1,14 @@
<?xml version='1.0'?>
<xsl:stylesheet xmlns:xsl="http://www.w3.org/1999/XSL/Transform"
xmlns:doc="http://www.rabbitmq.com/namespaces/ad-hoc/doc"
+ xmlns="http://www.w3.org/1999/xhtml"
version='1.0'>
<xsl:output method="xml" />
<!-- Copy every element through with local name only -->
<xsl:template match="*">
- <xsl:element name="{local-name()}">
+ <xsl:element name="{local-name()}" namespace="">
<xsl:apply-templates select="@*|node()"/>
</xsl:element>
</xsl:template>
diff --git a/ebin/rabbit_app.in b/ebin/rabbit_app.in
index 4be09c5a39..17d05a9919 100644
--- a/ebin/rabbit_app.in
+++ b/ebin/rabbit_app.in
@@ -29,4 +29,6 @@
{default_user_is_admin, true},
{default_vhost, <<"/">>},
{default_permissions, [<<".*">>, <<".*">>, <<".*">>]},
+ {cluster_nodes, []},
+ {server_properties, []},
{collect_statistics, none}]}]}.
diff --git a/include/rabbit.hrl b/include/rabbit.hrl
index df22b5738b..3de2442d20 100644
--- a/include/rabbit.hrl
+++ b/include/rabbit.hrl
@@ -30,7 +30,7 @@
%%
-record(user, {username, password, is_admin}).
--record(permission, {scope, configure, write, read}).
+-record(permission, {configure, write, read}).
-record(user_vhost, {username, virtual_host}).
-record(user_permission, {user_vhost, permission}).
@@ -63,7 +63,7 @@
-record(binding, {source, key, destination, args = []}).
-record(reverse_binding, {destination, key, source, args = []}).
--record(listener, {node, protocol, host, port}).
+-record(listener, {node, protocol, host, ip_address, port}).
-record(basic_message, {exchange_name, routing_key, content, guid,
is_persistent}).
diff --git a/packaging/macports/Portfile.in b/packaging/macports/Portfile.in
index f30460d3d8..e37a45b387 100644
--- a/packaging/macports/Portfile.in
+++ b/packaging/macports/Portfile.in
@@ -75,7 +75,7 @@ post-destroot {
reinplace -E "s:(/etc/rabbitmq/rabbitmq.conf):${prefix}\\1:g" \
${realsbin}/rabbitmq-env
- foreach var {CONFIG_FILE CLUSTER_CONFIG_FILE LOG_BASE MNESIA_BASE PIDS_FILE} {
+ foreach var {CONFIG_FILE LOG_BASE MNESIA_BASE PIDS_FILE} {
reinplace -E "s:^($var)=/:\\1=${prefix}/:" \
${realsbin}/rabbitmq-multi \
${realsbin}/rabbitmq-server \
diff --git a/scripts/rabbitmq-server b/scripts/rabbitmq-server
index 8e26663a50..ef0a3521c1 100755
--- a/scripts/rabbitmq-server
+++ b/scripts/rabbitmq-server
@@ -35,7 +35,6 @@ NODENAME=rabbit@${HOSTNAME%%.*}
SERVER_ERL_ARGS="+K true +A30 +P 1048576 \
-kernel inet_default_listen_options [{nodelay,true}] \
-kernel inet_default_connect_options [{nodelay,true}]"
-CLUSTER_CONFIG_FILE=/etc/rabbitmq/rabbitmq_cluster.config
CONFIG_FILE=/etc/rabbitmq/rabbitmq
LOG_BASE=/var/log/rabbitmq
MNESIA_BASE=/var/lib/rabbitmq/mnesia
@@ -59,7 +58,6 @@ else
fi
[ "x" = "x$RABBITMQ_NODENAME" ] && RABBITMQ_NODENAME=${NODENAME}
[ "x" = "x$RABBITMQ_SERVER_ERL_ARGS" ] && RABBITMQ_SERVER_ERL_ARGS=${SERVER_ERL_ARGS}
-[ "x" = "x$RABBITMQ_CLUSTER_CONFIG_FILE" ] && RABBITMQ_CLUSTER_CONFIG_FILE=${CLUSTER_CONFIG_FILE}
[ "x" = "x$RABBITMQ_CONFIG_FILE" ] && RABBITMQ_CONFIG_FILE=${CONFIG_FILE}
[ "x" = "x$RABBITMQ_LOG_BASE" ] && RABBITMQ_LOG_BASE=${LOG_BASE}
[ "x" = "x$RABBITMQ_MNESIA_BASE" ] && RABBITMQ_MNESIA_BASE=${MNESIA_BASE}
diff --git a/scripts/rabbitmq-server.bat b/scripts/rabbitmq-server.bat
index 5bcbc6babd..193f1c8a0d 100644
--- a/scripts/rabbitmq-server.bat
+++ b/scripts/rabbitmq-server.bat
@@ -165,7 +165,6 @@ if not "!RABBITMQ_NODE_IP_ADDRESS!"=="" (
-os_mon start_disksup false ^
-os_mon start_memsup false ^
-mnesia dir \""!RABBITMQ_MNESIA_DIR!"\" ^
-!CLUSTER_CONFIG! ^
!RABBITMQ_SERVER_START_ARGS! ^
!STAR!
diff --git a/scripts/rabbitmq-service.bat b/scripts/rabbitmq-service.bat
index 4b3961d43b..51ef6ecca3 100644
--- a/scripts/rabbitmq-service.bat
+++ b/scripts/rabbitmq-service.bat
@@ -232,7 +232,6 @@ set ERLANG_SERVICE_ARGUMENTS= ^
-os_mon start_disksup false ^
-os_mon start_memsup false ^
-mnesia dir \""!RABBITMQ_MNESIA_DIR!"\" ^
-!CLUSTER_CONFIG! ^
!RABBITMQ_SERVER_START_ARGS! ^
!STAR!
diff --git a/src/delegate.erl b/src/delegate.erl
index c8aa3092c7..e50b99f141 100644
--- a/src/delegate.erl
+++ b/src/delegate.erl
@@ -35,7 +35,7 @@
-behaviour(gen_server2).
--export([start_link/1, invoke_no_result/2, invoke/2, process_count/0]).
+-export([start_link/2, invoke_no_result/2, invoke/2, process_count/0]).
-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
terminate/2, code_change/3]).
@@ -44,7 +44,8 @@
-ifdef(use_specs).
--spec(start_link/1 :: (non_neg_integer()) -> {'ok', pid()} | {'error', any()}).
+-spec(start_link/2 ::
+ (atom(), non_neg_integer()) -> {'ok', pid()} | {'error', any()}).
-spec(invoke_no_result/2 ::
(pid() | [pid()], fun ((pid()) -> any())) -> 'ok').
-spec(invoke/2 :: (pid() | [pid()], fun ((pid()) -> A)) -> A).
@@ -60,8 +61,8 @@
%%----------------------------------------------------------------------------
-start_link(Hash) ->
- gen_server2:start_link({local, server(Hash)}, ?MODULE, [], []).
+start_link(Prefix, Hash) ->
+ gen_server2:start_link({local, server(Prefix, Hash)}, ?MODULE, [], []).
invoke(Pid, Fun) when is_pid(Pid) ->
[Res] = invoke_per_node(split_delegate_per_node([Pid]), Fun),
@@ -147,7 +148,8 @@ delegate_per_remote_node(NodePids, Fun, DelegateFun) ->
local_server(Node) ->
case get({delegate_local_server_name, Node}) of
undefined ->
- Name = server(erlang:phash2({self(), Node}, process_count())),
+ Name = server(outgoing,
+ erlang:phash2({self(), Node}, process_count())),
put({delegate_local_server_name, Node}, Name),
Name;
Name -> Name
@@ -160,17 +162,20 @@ remote_server(Node) ->
{badrpc, _} ->
%% Have to return something, if we're just casting
%% then we don't want to blow up
- server(1);
+ server(incoming, 1);
Count ->
- Name = server(erlang:phash2({self(), Node}, Count)),
+ Name = server(incoming,
+ erlang:phash2({self(), Node}, Count)),
put({delegate_remote_server_name, Node}, Name),
Name
end;
Name -> Name
end.
-server(Hash) ->
- list_to_atom("delegate_process_" ++ integer_to_list(Hash)).
+server(Prefix, Hash) ->
+ list_to_atom("delegate_" ++
+ atom_to_list(Prefix) ++ "_" ++
+ integer_to_list(Hash)).
safe_invoke(Pids, Fun) when is_list(Pids) ->
[safe_invoke(Pid, Fun) || Pid <- Pids];
diff --git a/src/delegate_sup.erl b/src/delegate_sup.erl
index ff303ee28c..544546f1b1 100644
--- a/src/delegate_sup.erl
+++ b/src/delegate_sup.erl
@@ -55,9 +55,11 @@ start_link() ->
%%----------------------------------------------------------------------------
init(_Args) ->
- {ok, {{one_for_one, 10, 10},
- [{Hash, {delegate, start_link, [Hash]},
- transient, 16#ffffffff, worker, [delegate]} ||
- Hash <- lists:seq(0, delegate:process_count() - 1)]}}.
+ {ok, {{one_for_one, 10, 10}, specs(incoming) ++ specs(outgoing)}}.
+
+specs(Prefix) ->
+ [{{Prefix, Hash}, {delegate, start_link, [Prefix, Hash]},
+ transient, 16#ffffffff, worker, [delegate]} ||
+ Hash <- lists:seq(0, delegate:process_count() - 1)].
%%----------------------------------------------------------------------------
diff --git a/src/rabbit_access_control.erl b/src/rabbit_access_control.erl
index 73fd6f0e51..85452abfaa 100644
--- a/src/rabbit_access_control.erl
+++ b/src/rabbit_access_control.erl
@@ -38,7 +38,7 @@
-export([add_user/2, delete_user/1, change_password/2, set_admin/1,
clear_admin/1, list_users/0, lookup_user/1]).
-export([add_vhost/1, delete_vhost/1, vhost_exists/1, list_vhosts/0]).
--export([set_permissions/5, set_permissions/6, clear_permissions/2,
+-export([set_permissions/5, clear_permissions/2,
list_permissions/0, list_vhost_permissions/1, list_user_permissions/1,
list_user_vhost_permissions/2]).
@@ -52,9 +52,6 @@
-type(username() :: binary()).
-type(password() :: binary()).
-type(regexp() :: binary()).
--type(scope() :: binary()).
--type(scope_atom() :: 'client' | 'all').
-
-spec(check_login/2 ::
(binary(), binary()) -> rabbit_types:user() |
rabbit_types:channel_exit()).
@@ -82,21 +79,15 @@
-spec(list_vhosts/0 :: () -> [rabbit_types:vhost()]).
-spec(set_permissions/5 ::(username(), rabbit_types:vhost(), regexp(),
regexp(), regexp()) -> 'ok').
--spec(set_permissions/6 ::(scope(), username(), rabbit_types:vhost(),
- regexp(), regexp(), regexp()) -> 'ok').
-spec(clear_permissions/2 :: (username(), rabbit_types:vhost()) -> 'ok').
-spec(list_permissions/0 ::
- () -> [{username(), rabbit_types:vhost(), regexp(), regexp(), regexp(),
- scope_atom()}]).
+ () -> [{username(), rabbit_types:vhost(), regexp(), regexp(), regexp()}]).
-spec(list_vhost_permissions/1 ::
- (rabbit_types:vhost()) -> [{username(), regexp(), regexp(), regexp(),
- scope_atom()}]).
+ (rabbit_types:vhost()) -> [{username(), regexp(), regexp(), regexp()}]).
-spec(list_user_permissions/1 ::
- (username()) -> [{rabbit_types:vhost(), regexp(), regexp(), regexp(),
- scope_atom()}]).
+ (username()) -> [{rabbit_types:vhost(), regexp(), regexp(), regexp()}]).
-spec(list_user_vhost_permissions/2 ::
- (username(), rabbit_types:vhost()) -> [{regexp(), regexp(), regexp(),
- scope_atom()}]).
+ (username(), rabbit_types:vhost()) -> [{regexp(), regexp(), regexp()}]).
-endif.
@@ -188,20 +179,15 @@ check_resource_access(Username,
[] ->
false;
[#user_permission{permission = P}] ->
- case {Name, P} of
- {<<"amq.gen",_/binary>>, #permission{scope = client}} ->
- true;
- _ ->
- PermRegexp =
- case element(permission_index(Permission), P) of
- %% <<"^$">> breaks Emacs' erlang mode
- <<"">> -> <<$^, $$>>;
- RE -> RE
- end,
- case re:run(Name, PermRegexp, [{capture, none}]) of
- match -> true;
- nomatch -> false
- end
+ PermRegexp =
+ case element(permission_index(Permission), P) of
+ %% <<"^$">> breaks Emacs' erlang mode
+ <<"">> -> <<$^, $$>>;
+ RE -> RE
+ end,
+ case re:run(Name, PermRegexp, [{capture, none}]) of
+ match -> true;
+ nomatch -> false
end
end,
if Res -> ok;
@@ -334,7 +320,7 @@ internal_delete_vhost(VHostPath) ->
ok = rabbit_exchange:delete(Name, false)
end,
rabbit_exchange:list(VHostPath)),
- lists:foreach(fun ({Username, _, _, _, _}) ->
+ lists:foreach(fun ({Username, _, _, _}) ->
ok = clear_permissions(Username, VHostPath)
end,
list_vhost_permissions(VHostPath)),
@@ -355,16 +341,7 @@ validate_regexp(RegexpBin) ->
end.
set_permissions(Username, VHostPath, ConfigurePerm, WritePerm, ReadPerm) ->
- set_permissions(<<"client">>, Username, VHostPath, ConfigurePerm,
- WritePerm, ReadPerm).
-
-set_permissions(ScopeBin, Username, VHostPath, ConfigurePerm, WritePerm, ReadPerm) ->
lists:map(fun validate_regexp/1, [ConfigurePerm, WritePerm, ReadPerm]),
- Scope = case ScopeBin of
- <<"client">> -> client;
- <<"all">> -> all;
- _ -> throw({error, {invalid_scope, ScopeBin}})
- end,
rabbit_misc:execute_mnesia_transaction(
rabbit_misc:with_user_and_vhost(
Username, VHostPath,
@@ -374,7 +351,6 @@ set_permissions(ScopeBin, Username, VHostPath, ConfigurePerm, WritePerm, ReadPer
username = Username,
virtual_host = VHostPath},
permission = #permission{
- scope = Scope,
configure = ConfigurePerm,
write = WritePerm,
read = ReadPerm}},
@@ -393,35 +369,34 @@ clear_permissions(Username, VHostPath) ->
end)).
list_permissions() ->
- [{Username, VHostPath, ConfigurePerm, WritePerm, ReadPerm, Scope} ||
- {Username, VHostPath, ConfigurePerm, WritePerm, ReadPerm, Scope} <-
+ [{Username, VHostPath, ConfigurePerm, WritePerm, ReadPerm} ||
+ {Username, VHostPath, ConfigurePerm, WritePerm, ReadPerm} <-
list_permissions(match_user_vhost('_', '_'))].
list_vhost_permissions(VHostPath) ->
- [{Username, ConfigurePerm, WritePerm, ReadPerm, Scope} ||
- {Username, _, ConfigurePerm, WritePerm, ReadPerm, Scope} <-
+ [{Username, ConfigurePerm, WritePerm, ReadPerm} ||
+ {Username, _, ConfigurePerm, WritePerm, ReadPerm} <-
list_permissions(rabbit_misc:with_vhost(
VHostPath, match_user_vhost('_', VHostPath)))].
list_user_permissions(Username) ->
- [{VHostPath, ConfigurePerm, WritePerm, ReadPerm, Scope} ||
- {_, VHostPath, ConfigurePerm, WritePerm, ReadPerm, Scope} <-
+ [{VHostPath, ConfigurePerm, WritePerm, ReadPerm} ||
+ {_, VHostPath, ConfigurePerm, WritePerm, ReadPerm} <-
list_permissions(rabbit_misc:with_user(
Username, match_user_vhost(Username, '_')))].
list_user_vhost_permissions(Username, VHostPath) ->
- [{ConfigurePerm, WritePerm, ReadPerm, Scope} ||
- {_, _, ConfigurePerm, WritePerm, ReadPerm, Scope} <-
+ [{ConfigurePerm, WritePerm, ReadPerm} ||
+ {_, _, ConfigurePerm, WritePerm, ReadPerm} <-
list_permissions(rabbit_misc:with_user_and_vhost(
Username, VHostPath,
match_user_vhost(Username, VHostPath)))].
list_permissions(QueryThunk) ->
- [{Username, VHostPath, ConfigurePerm, WritePerm, ReadPerm, Scope} ||
+ [{Username, VHostPath, ConfigurePerm, WritePerm, ReadPerm} ||
#user_permission{user_vhost = #user_vhost{username = Username,
virtual_host = VHostPath},
- permission = #permission{ scope = Scope,
- configure = ConfigurePerm,
+ permission = #permission{ configure = ConfigurePerm,
write = WritePerm,
read = ReadPerm}} <-
%% TODO: use dirty ops instead
diff --git a/src/rabbit_control.erl b/src/rabbit_control.erl
index 8facaf1606..6b21274529 100644
--- a/src/rabbit_control.erl
+++ b/src/rabbit_control.erl
@@ -39,7 +39,6 @@
-define(QUIET_OPT, "-q").
-define(NODE_OPT, "-n").
-define(VHOST_OPT, "-p").
--define(SCOPE_OPT, "-s").
%%----------------------------------------------------------------------------
@@ -67,7 +66,7 @@ start() ->
{[Command0 | Args], Opts} =
rabbit_misc:get_options(
[{flag, ?QUIET_OPT}, {option, ?NODE_OPT, NodeStr},
- {option, ?VHOST_OPT, "/"}, {option, ?SCOPE_OPT, "client"}],
+ {option, ?VHOST_OPT, "/"}],
FullCommand),
Opts1 = lists:map(fun({K, V}) ->
case K of
@@ -289,10 +288,9 @@ action(list_consumers, Node, _Args, Opts, Inform) ->
action(set_permissions, Node, [Username, CPerm, WPerm, RPerm], Opts, Inform) ->
VHost = proplists:get_value(?VHOST_OPT, Opts),
- Scope = proplists:get_value(?SCOPE_OPT, Opts),
Inform("Setting permissions for user ~p in vhost ~p", [Username, VHost]),
call(Node, {rabbit_access_control, set_permissions,
- [Scope, Username, VHost, CPerm, WPerm, RPerm]});
+ [Username, VHost, CPerm, WPerm, RPerm]});
action(clear_permissions, Node, [Username], Opts, Inform) ->
VHost = proplists:get_value(?VHOST_OPT, Opts),
diff --git a/src/rabbit_mnesia.erl b/src/rabbit_mnesia.erl
index 577d206d1b..8de2f0d664 100644
--- a/src/rabbit_mnesia.erl
+++ b/src/rabbit_mnesia.erl
@@ -340,10 +340,8 @@ read_cluster_nodes_config() ->
case rabbit_misc:read_term_file(FileName) of
{ok, [ClusterNodes]} -> ClusterNodes;
{error, enoent} ->
- case application:get_env(cluster_nodes) of
- undefined -> [];
- {ok, ClusterNodes} -> ClusterNodes
- end;
+ {ok, ClusterNodes} = application:get_env(rabbit, cluster_nodes),
+ ClusterNodes;
{error, Reason} ->
throw({error, {cannot_read_cluster_nodes_config,
FileName, Reason}})
diff --git a/src/rabbit_msg_store.erl b/src/rabbit_msg_store.erl
index ba2af20c4c..d04f679f2a 100644
--- a/src/rabbit_msg_store.erl
+++ b/src/rabbit_msg_store.erl
@@ -34,8 +34,9 @@
-behaviour(gen_server2).
-export([start_link/4, successfully_recovered_state/1,
- client_init/3, client_terminate/2, client_delete_and_terminate/3,
- write/4, read/3, contains/2, remove/2, release/2, sync/3]).
+ client_init/3, client_terminate/1, client_delete_and_terminate/1,
+ client_ref/1,
+ write/3, read/2, contains/2, remove/2, release/2, sync/3]).
-export([sync/1, set_maximum_since_use/2,
has_readers/2, combine_files/3, delete_file/2]). %% internal
@@ -88,7 +89,9 @@
}).
-record(client_msstate,
- { file_handle_cache,
+ { server,
+ client_ref,
+ file_handle_cache,
index_state,
index_module,
dir,
@@ -96,8 +99,7 @@
file_handles_ets,
file_summary_ets,
dedup_cache_ets,
- cur_file_cache_ets,
- client_ref
+ cur_file_cache_ets
}).
-record(file_summary,
@@ -125,8 +127,11 @@
}).
-type(server() :: pid() | atom()).
+-type(client_ref() :: binary()).
-type(file_num() :: non_neg_integer()).
-type(client_msstate() :: #client_msstate {
+ server :: server(),
+ client_ref :: client_ref(),
file_handle_cache :: dict:dictionary(),
index_state :: any(),
index_module :: atom(),
@@ -135,8 +140,7 @@
file_handles_ets :: ets:tid(),
file_summary_ets :: ets:tid(),
dedup_cache_ets :: ets:tid(),
- cur_file_cache_ets :: ets:tid(),
- client_ref :: rabbit_guid:guid()}).
+ cur_file_cache_ets :: ets:tid()}).
-type(startup_fun_state() ::
{(fun ((A) -> 'finished' | {rabbit_guid:guid(), non_neg_integer(), A})),
A}).
@@ -146,18 +150,18 @@
(atom(), file:filename(), [binary()] | 'undefined',
startup_fun_state()) -> rabbit_types:ok_pid_or_error()).
-spec(successfully_recovered_state/1 :: (server()) -> boolean()).
--spec(client_init/3 :: (server(), rabbit_guid:guid(), guid_fun()) -> client_msstate()).
--spec(client_terminate/2 :: (client_msstate(), server()) -> 'ok').
--spec(client_delete_and_terminate/3 ::
- (client_msstate(), server(), binary()) -> 'ok').
--spec(write/4 :: (server(), rabbit_guid:guid(), msg(), client_msstate()) ->
- rabbit_types:ok(client_msstate())).
--spec(read/3 :: (server(), rabbit_guid:guid(), client_msstate()) ->
- {rabbit_types:ok(msg()) | 'not_found', client_msstate()}).
--spec(contains/2 :: (server(), rabbit_guid:guid()) -> boolean()).
--spec(remove/2 :: (server(), [rabbit_guid:guid()]) -> 'ok').
--spec(release/2 :: (server(), [rabbit_guid:guid()]) -> 'ok').
--spec(sync/3 :: (server(), [rabbit_guid:guid()], fun (() -> any())) -> 'ok').
+-spec(client_init/3 :: (server(), client_ref(), guid_fun()) -> client_msstate()).
+-spec(client_terminate/1 :: (client_msstate()) -> 'ok').
+-spec(client_delete_and_terminate/1 :: (client_msstate()) -> 'ok').
+-spec(client_ref/1 :: (client_msstate()) -> client_ref()).
+-spec(write/3 :: (rabbit_guid:guid(), msg(), client_msstate()) -> 'ok').
+-spec(read/2 :: (rabbit_guid:guid(), client_msstate()) ->
+ {rabbit_types:ok(msg()) | 'not_found', client_msstate()}).
+-spec(contains/2 :: (rabbit_guid:guid(), client_msstate()) -> boolean()).
+-spec(remove/2 :: ([rabbit_guid:guid()], client_msstate()) -> 'ok').
+-spec(release/2 :: ([rabbit_guid:guid()], client_msstate()) -> 'ok').
+-spec(sync/3 :: ([rabbit_guid:guid()], fun (() -> any()), client_msstate()) ->
+ 'ok').
-spec(sync/1 :: (server()) -> 'ok').
-spec(set_maximum_since_use/2 :: (server(), non_neg_integer()) -> 'ok').
@@ -338,7 +342,9 @@ client_init(Server, Ref, MsgOnDiskFun) ->
FileHandlesEts, FileSummaryEts, DedupCacheEts, CurFileCacheEts} =
gen_server2:call(Server, {new_client_state, Ref, MsgOnDiskFun},
infinity),
- #client_msstate { file_handle_cache = dict:new(),
+ #client_msstate { server = Server,
+ client_ref = Ref,
+ file_handle_cache = dict:new(),
index_state = IState,
index_module = IModule,
dir = Dir,
@@ -346,24 +352,25 @@ client_init(Server, Ref, MsgOnDiskFun) ->
file_handles_ets = FileHandlesEts,
file_summary_ets = FileSummaryEts,
dedup_cache_ets = DedupCacheEts,
- cur_file_cache_ets = CurFileCacheEts,
- client_ref = Ref }.
+ cur_file_cache_ets = CurFileCacheEts }.
-client_terminate(CState, Server) ->
+client_terminate(CState = #client_msstate { client_ref = Ref }) ->
close_all_handles(CState),
- ok = gen_server2:call(Server, {client_terminate, CState}, infinity).
+ ok = server_call(CState, {client_terminate, Ref}).
-client_delete_and_terminate(CState, Server, Ref) ->
+client_delete_and_terminate(CState = #client_msstate { client_ref = Ref }) ->
close_all_handles(CState),
- ok = gen_server2:cast(Server, {client_delete, Ref}).
+ ok = server_cast(CState, {client_delete, Ref}).
-write(Server, Guid, Msg,
+client_ref(#client_msstate { client_ref = Ref }) -> Ref.
+
+write(Guid, Msg,
CState = #client_msstate { cur_file_cache_ets = CurFileCacheEts,
client_ref = CRef }) ->
ok = update_msg_cache(CurFileCacheEts, Guid, Msg),
- {gen_server2:cast(Server, {write, CRef, Guid}), CState}.
+ ok = server_cast(CState, {write, CRef, Guid}).
-read(Server, Guid,
+read(Guid,
CState = #client_msstate { dedup_cache_ets = DedupCacheEts,
cur_file_cache_ets = CurFileCacheEts }) ->
%% 1. Check the dedup cache
@@ -372,13 +379,12 @@ read(Server, Guid,
%% 2. Check the cur file cache
case ets:lookup(CurFileCacheEts, Guid) of
[] ->
- Defer = fun() -> {gen_server2:call(
- Server, {read, Guid}, infinity),
- CState} end,
+ Defer = fun() ->
+ {server_call(CState, {read, Guid}), CState}
+ end,
case index_lookup_positive_ref_count(Guid, CState) of
not_found -> Defer();
- MsgLocation -> client_read1(Server, MsgLocation, Defer,
- CState)
+ MsgLocation -> client_read1(MsgLocation, Defer, CState)
end;
[{Guid, Msg, _CacheRefCount}] ->
%% Although we've found it, we don't know the
@@ -389,12 +395,12 @@ read(Server, Guid,
{{ok, Msg}, CState}
end.
-contains(Server, Guid) -> gen_server2:call(Server, {contains, Guid}, infinity).
-remove(_Server, []) -> ok;
-remove(Server, Guids) -> gen_server2:cast(Server, {remove, Guids}).
-release(_Server, []) -> ok;
-release(Server, Guids) -> gen_server2:cast(Server, {release, Guids}).
-sync(Server, Guids, K) -> gen_server2:cast(Server, {sync, Guids, K}).
+contains(Guid, CState) -> server_call(CState, {contains, Guid}).
+remove([], _CState) -> ok;
+remove(Guids, CState) -> server_cast(CState, {remove, Guids}).
+release([], _CState) -> ok;
+release(Guids, CState) -> server_cast(CState, {release, Guids}).
+sync(Guids, K, CState) -> server_cast(CState, {sync, Guids, K}).
sync(Server) ->
gen_server2:cast(Server, sync).
@@ -406,18 +412,22 @@ set_maximum_since_use(Server, Age) ->
%% Client-side-only helpers
%%----------------------------------------------------------------------------
-client_read1(Server,
- #msg_location { guid = Guid, file = File } = MsgLocation,
- Defer,
+server_call(#client_msstate { server = Server }, Msg) ->
+ gen_server2:call(Server, Msg, infinity).
+
+server_cast(#client_msstate { server = Server }, Msg) ->
+ gen_server2:cast(Server, Msg).
+
+client_read1(#msg_location { guid = Guid, file = File } = MsgLocation, Defer,
CState = #client_msstate { file_summary_ets = FileSummaryEts }) ->
case ets:lookup(FileSummaryEts, File) of
[] -> %% File has been GC'd and no longer exists. Go around again.
- read(Server, Guid, CState);
+ read(Guid, CState);
[#file_summary { locked = Locked, right = Right }] ->
- client_read2(Server, Locked, Right, MsgLocation, Defer, CState)
+ client_read2(Locked, Right, MsgLocation, Defer, CState)
end.
-client_read2(_Server, false, undefined, _MsgLocation, Defer, _CState) ->
+client_read2(false, undefined, _MsgLocation, Defer, _CState) ->
%% Although we've already checked both caches and not found the
%% message there, the message is apparently in the
%% current_file. We can only arrive here if we are trying to read
@@ -428,12 +438,12 @@ client_read2(_Server, false, undefined, _MsgLocation, Defer, _CState) ->
%% contents of the current file, thus reads from the current file
%% will end up here and will need to be deferred.
Defer();
-client_read2(_Server, true, _Right, _MsgLocation, Defer, _CState) ->
+client_read2(true, _Right, _MsgLocation, Defer, _CState) ->
%% Of course, in the mean time, the GC could have run and our msg
%% is actually in a different file, unlocked. However, defering is
%% the safest and simplest thing to do.
Defer();
-client_read2(Server, false, _Right,
+client_read2(false, _Right,
MsgLocation = #msg_location { guid = Guid, file = File },
Defer,
CState = #client_msstate { file_summary_ets = FileSummaryEts }) ->
@@ -442,10 +452,10 @@ client_read2(Server, false, _Right,
%% finished.
safe_ets_update_counter(
FileSummaryEts, File, {#file_summary.readers, +1},
- fun (_) -> client_read3(Server, MsgLocation, Defer, CState) end,
- fun () -> read(Server, Guid, CState) end).
+ fun (_) -> client_read3(MsgLocation, Defer, CState) end,
+ fun () -> read(Guid, CState) end).
-client_read3(Server, #msg_location { guid = Guid, file = File }, Defer,
+client_read3(#msg_location { guid = Guid, file = File }, Defer,
CState = #client_msstate { file_handles_ets = FileHandlesEts,
file_summary_ets = FileSummaryEts,
dedup_cache_ets = DedupCacheEts,
@@ -469,7 +479,7 @@ client_read3(Server, #msg_location { guid = Guid, file = File }, Defer,
%% too).
case ets:lookup(FileSummaryEts, File) of
[] -> %% GC has deleted our file, just go round again.
- read(Server, Guid, CState);
+ read(Guid, CState);
[#file_summary { locked = true }] ->
%% If we get a badarg here, then the GC has finished and
%% deleted our file. Try going around again. Otherwise,
@@ -480,7 +490,7 @@ client_read3(Server, #msg_location { guid = Guid, file = File }, Defer,
%% unlocks the dest)
try Release(),
Defer()
- catch error:badarg -> read(Server, Guid, CState)
+ catch error:badarg -> read(Guid, CState)
end;
[#file_summary { locked = false }] ->
%% Ok, we're definitely safe to continue - a GC involving
@@ -505,7 +515,7 @@ client_read3(Server, #msg_location { guid = Guid, file = File }, Defer,
{{ok, Msg}, CState2};
#msg_location {} = MsgLocation -> %% different file!
Release(), %% this MUST NOT fail with badarg
- client_read1(Server, MsgLocation, Defer, CState);
+ client_read1(MsgLocation, Defer, CState);
not_found -> %% it seems not to exist. Defer, just to be sure.
try Release() %% this can badarg, same as locked case, above
catch error:badarg -> ok
@@ -657,7 +667,7 @@ handle_call({new_client_state, CRef, Callback}, _From,
State #msstate { client_refs = sets:add_element(CRef, ClientRefs),
client_ondisk_callback = CODC1 });
-handle_call({client_terminate, #client_msstate { client_ref = CRef }}, _From,
+handle_call({client_terminate, CRef}, _From,
State) ->
reply(ok, clear_client_callback(CRef, State));
diff --git a/src/rabbit_networking.erl b/src/rabbit_networking.erl
index db5c71f62a..03a9b38679 100644
--- a/src/rabbit_networking.erl
+++ b/src/rabbit_networking.erl
@@ -41,7 +41,7 @@
%%used by TCP-based transports, e.g. STOMP adapter
-export([check_tcp_listener_address/3]).
--export([tcp_listener_started/2, tcp_listener_stopped/2,
+-export([tcp_listener_started/3, tcp_listener_stopped/3,
start_client/1, start_ssl_client/2]).
-include("rabbit.hrl").
@@ -160,14 +160,14 @@ check_tcp_listener_address(NamePrefix, Host, Port) ->
{IPAddress, Name}.
start_tcp_listener(Host, Port) ->
- start_listener(Host, Port, "TCP Listener",
+ start_listener(Host, Port, amqp, "TCP Listener",
{?MODULE, start_client, []}).
start_ssl_listener(Host, Port, SslOpts) ->
- start_listener(Host, Port, "SSL Listener",
+ start_listener(Host, Port, 'amqp/ssl', "SSL Listener",
{?MODULE, start_ssl_client, [SslOpts]}).
-start_listener(Host, Port, Label, OnConnect) ->
+start_listener(Host, Port, Protocol, Label, OnConnect) ->
{IPAddress, Name} =
check_tcp_listener_address(rabbit_tcp_listener_sup, Host, Port),
{ok,_} = supervisor:start_child(
@@ -175,8 +175,8 @@ start_listener(Host, Port, Label, OnConnect) ->
{Name,
{tcp_listener_sup, start_link,
[IPAddress, Port, ?RABBIT_TCP_OPTS ,
- {?MODULE, tcp_listener_started, []},
- {?MODULE, tcp_listener_stopped, []},
+ {?MODULE, tcp_listener_started, [Protocol]},
+ {?MODULE, tcp_listener_stopped, [Protocol]},
OnConnect, Label]},
transient, infinity, supervisor, [tcp_listener_sup]}),
ok.
@@ -188,20 +188,25 @@ stop_tcp_listener(Host, Port) ->
ok = supervisor:delete_child(rabbit_sup, Name),
ok.
-tcp_listener_started(IPAddress, Port) ->
+tcp_listener_started(Protocol, IPAddress, Port) ->
+ %% We need the ip to distinguish e.g. 0.0.0.0 and 127.0.0.1
+ %% We need the host so we can distinguish multiple instances of the above
+ %% in a cluster.
ok = mnesia:dirty_write(
rabbit_listener,
#listener{node = node(),
- protocol = tcp,
+ protocol = Protocol,
host = tcp_host(IPAddress),
+ ip_address = IPAddress,
port = Port}).
-tcp_listener_stopped(IPAddress, Port) ->
+tcp_listener_stopped(Protocol, IPAddress, Port) ->
ok = mnesia:dirty_delete_object(
rabbit_listener,
#listener{node = node(),
- protocol = tcp,
+ protocol = Protocol,
host = tcp_host(IPAddress),
+ ip_address = IPAddress,
port = Port}).
active_listeners() ->
diff --git a/src/rabbit_queue_index.erl b/src/rabbit_queue_index.erl
index b58c5a60d0..8a4bb801d9 100644
--- a/src/rabbit_queue_index.erl
+++ b/src/rabbit_queue_index.erl
@@ -31,8 +31,9 @@
-module(rabbit_queue_index).
--export([init/5, terminate/2, delete_and_terminate/1, publish/5,
- deliver/2, ack/2, sync/2, flush/1, read/3,
+-export([init/2, shutdown_terms/1, recover/5,
+ terminate/2, delete_and_terminate/1,
+ publish/5, deliver/2, ack/2, sync/2, flush/1, read/3,
next_segment_boundary/1, bounds/1, recover/1]).
-define(CLEAN_FILENAME, "clean.dot").
@@ -202,10 +203,13 @@
-type(startup_fun_state() ::
{fun ((A) -> 'finished' | {rabbit_guid:guid(), non_neg_integer(), A}),
A}).
+-type(shutdown_terms() :: [any()]).
--spec(init/5 :: (rabbit_amqqueue:name(), boolean(), boolean(),
- fun ((rabbit_guid:guid()) -> boolean()), on_sync_fun()) ->
- {'undefined' | non_neg_integer(), [any()], qistate()}).
+-spec(init/2 :: (rabbit_amqqueue:name(), on_sync_fun()) -> qistate()).
+-spec(shutdown_terms/1 :: (rabbit_amqqueue:name()) -> shutdown_terms()).
+-spec(recover/5 :: (rabbit_amqqueue:name(), shutdown_terms(), boolean(),
+ fun ((rabbit_guid:guid()) -> boolean()), on_sync_fun()) ->
+ {'undefined' | non_neg_integer(), qistate()}).
-spec(terminate/2 :: ([any()], qistate()) -> qistate()).
-spec(delete_and_terminate/1 :: (qistate()) -> qistate()).
-spec(publish/5 :: (rabbit_guid:guid(), seq_id(),
@@ -222,8 +226,8 @@
-spec(next_segment_boundary/1 :: (seq_id()) -> seq_id()).
-spec(bounds/1 :: (qistate()) ->
{non_neg_integer(), non_neg_integer(), qistate()}).
--spec(recover/1 ::
- ([rabbit_amqqueue:name()]) -> {[[any()]], startup_fun_state()}).
+-spec(recover/1 :: ([rabbit_amqqueue:name()]) ->
+ {[[any()]], startup_fun_state()}).
-endif.
@@ -232,25 +236,27 @@
%% public API
%%----------------------------------------------------------------------------
-init(Name, false, _MsgStoreRecovered, _ContainsCheckFun, OnSyncFun) ->
+init(Name, OnSyncFun) ->
State = #qistate { dir = Dir } = blank_state(Name),
false = filelib:is_file(Dir), %% is_file == is file or dir
- {0, [], State #qistate { on_sync = OnSyncFun }};
+ State #qistate { on_sync = OnSyncFun }.
-init(Name, true, MsgStoreRecovered, ContainsCheckFun, OnSyncFun) ->
+shutdown_terms(Name) ->
+ #qistate { dir = Dir } = blank_state(Name),
+ case read_shutdown_terms(Dir) of
+ {error, _} -> [];
+ {ok, Terms1} -> Terms1
+ end.
+
+recover(Name, Terms, MsgStoreRecovered, ContainsCheckFun, OnSyncFun) ->
State = #qistate { dir = Dir } = blank_state(Name),
- Terms = case read_shutdown_terms(Dir) of
- {error, _} -> [];
- {ok, Terms1} -> Terms1
- end,
+ State1 = State #qistate { on_sync = OnSyncFun },
CleanShutdown = detect_clean_shutdown(Dir),
- {Count, State1} =
- case CleanShutdown andalso MsgStoreRecovered of
- true -> RecoveredCounts = proplists:get_value(segments, Terms, []),
- init_clean(RecoveredCounts, State);
- false -> init_dirty(CleanShutdown, ContainsCheckFun, State)
- end,
- {Count, Terms, State1 #qistate { on_sync = OnSyncFun }}.
+ case CleanShutdown andalso MsgStoreRecovered of
+ true -> RecoveredCounts = proplists:get_value(segments, Terms, []),
+ init_clean(RecoveredCounts, State1);
+ false -> init_dirty(CleanShutdown, ContainsCheckFun, State1)
+ end.
terminate(Terms, State) ->
{SegmentCounts, State1 = #qistate { dir = Dir }} = terminate(State),
diff --git a/src/rabbit_reader.erl b/src/rabbit_reader.erl
index 29004bd5fb..58d29008a8 100644
--- a/src/rabbit_reader.erl
+++ b/src/rabbit_reader.erl
@@ -235,12 +235,29 @@ conserve_memory(Pid, Conserve) ->
server_properties() ->
{ok, Product} = application:get_key(rabbit, id),
{ok, Version} = application:get_key(rabbit, vsn),
- [{list_to_binary(K), longstr, list_to_binary(V)} ||
- {K, V} <- [{"product", Product},
- {"version", Version},
- {"platform", "Erlang/OTP"},
- {"copyright", ?COPYRIGHT_MESSAGE},
- {"information", ?INFORMATION_MESSAGE}]].
+
+ %% Get any configuration-specified server properties
+ {ok, RawConfigServerProps} = application:get_env(rabbit,
+ server_properties),
+
+ %% Normalize the simplifed (2-tuple) and unsimplified (3-tuple) forms
+ %% from the config and merge them with the generated built-in properties
+ NormalizedConfigServerProps =
+ [case X of
+ {KeyAtom, Value} -> {atom_to_binary(KeyAtom, latin1),
+ longstr,
+ list_to_binary(Value)};
+ {BinKey, Type, Value} -> {BinKey, Type, Value}
+ end || X <- RawConfigServerProps ++
+ [{product, Product},
+ {version, Version},
+ {platform, "Erlang/OTP"},
+ {copyright, ?COPYRIGHT_MESSAGE},
+ {information, ?INFORMATION_MESSAGE}]],
+
+ %% Filter duplicated properties in favor of config file provided values
+ lists:usort(fun ({K1,_,_}, {K2,_,_}) -> K1 =< K2 end,
+ NormalizedConfigServerProps).
inet_op(F) -> rabbit_misc:throw_on_error(inet_error, F).
diff --git a/src/rabbit_ssl.erl b/src/rabbit_ssl.erl
index be451af631..4335dd2e40 100644
--- a/src/rabbit_ssl.erl
+++ b/src/rabbit_ssl.erl
@@ -34,7 +34,6 @@
-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]).
diff --git a/src/rabbit_tests.erl b/src/rabbit_tests.erl
index 5986e11ed8..8a3005273c 100644
--- a/src/rabbit_tests.erl
+++ b/src/rabbit_tests.erl
@@ -73,6 +73,7 @@ all_tests() ->
passed = test_user_management(),
passed = test_server_status(),
passed = maybe_run_cluster_dependent_tests(),
+ passed = test_configurable_server_properties(),
passed.
maybe_run_cluster_dependent_tests() ->
@@ -962,9 +963,6 @@ test_user_management() ->
control_action(list_permissions, [], [{"-p", "/testhost"}]),
{error, {invalid_regexp, _, _}} =
control_action(set_permissions, ["guest", "+foo", ".*", ".*"]),
- {error, {invalid_scope, _}} =
- control_action(set_permissions, ["guest", "foo", ".*", ".*"],
- [{"-s", "cilent"}]),
%% user creation
ok = control_action(add_user, ["foo", "bar"]),
@@ -987,9 +985,7 @@ test_user_management() ->
ok = control_action(set_permissions, ["foo", ".*", ".*", ".*"],
[{"-p", "/testhost"}]),
ok = control_action(set_permissions, ["foo", ".*", ".*", ".*"],
- [{"-p", "/testhost"}, {"-s", "client"}]),
- ok = control_action(set_permissions, ["foo", ".*", ".*", ".*"],
- [{"-p", "/testhost"}, {"-s", "all"}]),
+ [{"-p", "/testhost"}]),
ok = control_action(list_permissions, [], [{"-p", "/testhost"}]),
ok = control_action(list_permissions, [], [{"-p", "/testhost"}]),
ok = control_action(list_user_permissions, ["foo"]),
@@ -1297,7 +1293,7 @@ info_action(Command, Args, CheckVHost) ->
{bad_argument, dummy} = control_action(Command, ["dummy"]),
ok.
-default_options() -> [{"-s", "client"}, {"-p", "/"}, {"-q", "false"}].
+default_options() -> [{"-p", "/"}, {"-q", "false"}].
expand_options(As, Bs) ->
lists:foldl(fun({K, _}=A, R) ->
@@ -1431,17 +1427,17 @@ restart_msg_store_empty() ->
guid_bin(X) ->
erlang:md5(term_to_binary(X)).
-msg_store_contains(Atom, Guids) ->
+msg_store_contains(Atom, Guids, MSCState) ->
Atom = lists:foldl(
fun (Guid, Atom1) when Atom1 =:= Atom ->
- rabbit_msg_store:contains(?PERSISTENT_MSG_STORE, Guid) end,
+ rabbit_msg_store:contains(Guid, MSCState) end,
Atom, Guids).
-msg_store_sync(Guids) ->
+msg_store_sync(Guids, MSCState) ->
Ref = make_ref(),
Self = self(),
- ok = rabbit_msg_store:sync(?PERSISTENT_MSG_STORE, Guids,
- fun () -> Self ! {sync, Ref} end),
+ ok = rabbit_msg_store:sync(Guids, fun () -> Self ! {sync, Ref} end,
+ MSCState),
receive
{sync, Ref} -> ok
after
@@ -1453,55 +1449,65 @@ msg_store_sync(Guids) ->
msg_store_read(Guids, MSCState) ->
lists:foldl(fun (Guid, MSCStateM) ->
{{ok, Guid}, MSCStateN} = rabbit_msg_store:read(
- ?PERSISTENT_MSG_STORE,
Guid, MSCStateM),
MSCStateN
end, MSCState, Guids).
msg_store_write(Guids, MSCState) ->
- lists:foldl(fun (Guid, {ok, MSCStateN}) ->
- rabbit_msg_store:write(?PERSISTENT_MSG_STORE,
- Guid, Guid, MSCStateN)
- end, {ok, MSCState}, Guids).
+ ok = lists:foldl(
+ fun (Guid, ok) -> rabbit_msg_store:write(Guid, Guid, MSCState) end,
+ ok, Guids).
+
+msg_store_remove(Guids, MSCState) ->
+ rabbit_msg_store:remove(Guids, MSCState).
-msg_store_remove(Guids) ->
- rabbit_msg_store:remove(?PERSISTENT_MSG_STORE, Guids).
+msg_store_remove(MsgStore, Ref, Guids) ->
+ with_msg_store_client(MsgStore, Ref,
+ fun (MSCStateM) ->
+ ok = msg_store_remove(Guids, MSCStateM),
+ MSCStateM
+ end).
+
+with_msg_store_client(MsgStore, Ref, Fun) ->
+ rabbit_msg_store:client_terminate(
+ Fun(rabbit_msg_store:client_init(MsgStore, Ref, undefined))).
foreach_with_msg_store_client(MsgStore, Ref, Fun, L) ->
rabbit_msg_store:client_terminate(
- lists:foldl(fun (Guid, MSCState) -> Fun(Guid, MsgStore, MSCState) end,
- rabbit_msg_store:client_init(MsgStore, Ref, undefined), L), MsgStore).
+ lists:foldl(fun (Guid, MSCState) -> Fun(Guid, MSCState) end,
+ rabbit_msg_store:client_init(MsgStore, Ref, undefined), L)).
test_msg_store() ->
restart_msg_store_empty(),
Self = self(),
Guids = [guid_bin(M) || M <- lists:seq(1,100)],
{Guids1stHalf, Guids2ndHalf} = lists:split(50, Guids),
- %% check we don't contain any of the msgs we're about to publish
- false = msg_store_contains(false, Guids),
Ref = rabbit_guid:guid(),
- MSCState = rabbit_msg_store:client_init(?PERSISTENT_MSG_STORE, Ref, undefined),
+ MSCState = rabbit_msg_store:client_init(?PERSISTENT_MSG_STORE, Ref,
+ undefined),
+ %% check we don't contain any of the msgs we're about to publish
+ false = msg_store_contains(false, Guids, MSCState),
%% publish the first half
- {ok, MSCState1} = msg_store_write(Guids1stHalf, MSCState),
+ ok = msg_store_write(Guids1stHalf, MSCState),
%% sync on the first half
- ok = msg_store_sync(Guids1stHalf),
+ ok = msg_store_sync(Guids1stHalf, MSCState),
%% publish the second half
- {ok, MSCState2} = msg_store_write(Guids2ndHalf, MSCState1),
+ ok = msg_store_write(Guids2ndHalf, MSCState),
%% sync on the first half again - the msg_store will be dirty, but
%% we won't need the fsync
- ok = msg_store_sync(Guids1stHalf),
+ ok = msg_store_sync(Guids1stHalf, MSCState),
%% check they're all in there
- true = msg_store_contains(true, Guids),
+ true = msg_store_contains(true, Guids, MSCState),
%% publish the latter half twice so we hit the caching and ref count code
- {ok, MSCState3} = msg_store_write(Guids2ndHalf, MSCState2),
+ ok = msg_store_write(Guids2ndHalf, MSCState),
%% check they're still all in there
- true = msg_store_contains(true, Guids),
+ true = msg_store_contains(true, Guids, MSCState),
%% sync on the 2nd half, but do lots of individual syncs to try
%% and cause coalescing to happen
ok = lists:foldl(
fun (Guid, ok) -> rabbit_msg_store:sync(
- ?PERSISTENT_MSG_STORE,
- [Guid], fun () -> Self ! {sync, Guid} end)
+ [Guid], fun () -> Self ! {sync, Guid} end,
+ MSCState)
end, ok, Guids2ndHalf),
lists:foldl(
fun(Guid, ok) ->
@@ -1516,24 +1522,24 @@ test_msg_store() ->
end, ok, Guids2ndHalf),
%% it's very likely we're not dirty here, so the 1st half sync
%% should hit a different code path
- ok = msg_store_sync(Guids1stHalf),
+ ok = msg_store_sync(Guids1stHalf, MSCState),
%% read them all
- MSCState4 = msg_store_read(Guids, MSCState3),
+ MSCState1 = msg_store_read(Guids, MSCState),
%% read them all again - this will hit the cache, not disk
- MSCState5 = msg_store_read(Guids, MSCState4),
+ MSCState2 = msg_store_read(Guids, MSCState1),
%% remove them all
- ok = rabbit_msg_store:remove(?PERSISTENT_MSG_STORE, Guids),
+ ok = rabbit_msg_store:remove(Guids, MSCState2),
%% check first half doesn't exist
- false = msg_store_contains(false, Guids1stHalf),
+ false = msg_store_contains(false, Guids1stHalf, MSCState2),
%% check second half does exist
- true = msg_store_contains(true, Guids2ndHalf),
+ true = msg_store_contains(true, Guids2ndHalf, MSCState2),
%% read the second half again
- MSCState6 = msg_store_read(Guids2ndHalf, MSCState5),
+ MSCState3 = msg_store_read(Guids2ndHalf, MSCState2),
%% release the second half, just for fun (aka code coverage)
- ok = rabbit_msg_store:release(?PERSISTENT_MSG_STORE, Guids2ndHalf),
+ ok = rabbit_msg_store:release(Guids2ndHalf, MSCState3),
%% read the second half again, just for fun (aka code coverage)
- MSCState7 = msg_store_read(Guids2ndHalf, MSCState6),
- ok = rabbit_msg_store:client_terminate(MSCState7, ?PERSISTENT_MSG_STORE),
+ MSCState4 = msg_store_read(Guids2ndHalf, MSCState3),
+ ok = rabbit_msg_store:client_terminate(MSCState4),
%% stop and restart, preserving every other msg in 2nd half
ok = rabbit_variable_queue:stop_msg_store(),
ok = rabbit_variable_queue:start_msg_store(
@@ -1544,22 +1550,29 @@ test_msg_store() ->
([Guid|GuidsTail]) ->
{Guid, 0, GuidsTail}
end, Guids2ndHalf}),
+ MSCState5 = rabbit_msg_store:client_init(?PERSISTENT_MSG_STORE, Ref,
+ undefined),
%% check we have the right msgs left
lists:foldl(
fun (Guid, Bool) ->
- not(Bool = rabbit_msg_store:contains(?PERSISTENT_MSG_STORE, Guid))
+ not(Bool = rabbit_msg_store:contains(Guid, MSCState5))
end, false, Guids2ndHalf),
+ ok = rabbit_msg_store:client_terminate(MSCState5),
%% restart empty
restart_msg_store_empty(),
+ MSCState6 = rabbit_msg_store:client_init(?PERSISTENT_MSG_STORE, Ref,
+ undefined),
%% check we don't contain any of the msgs
- false = msg_store_contains(false, Guids),
+ false = msg_store_contains(false, Guids, MSCState6),
%% publish the first half again
- MSCState8 = rabbit_msg_store:client_init(?PERSISTENT_MSG_STORE, Ref, undefined),
- {ok, MSCState9} = msg_store_write(Guids1stHalf, MSCState8),
+ ok = msg_store_write(Guids1stHalf, MSCState6),
%% this should force some sort of sync internally otherwise misread
ok = rabbit_msg_store:client_terminate(
- msg_store_read(Guids1stHalf, MSCState9), ?PERSISTENT_MSG_STORE),
- ok = rabbit_msg_store:remove(?PERSISTENT_MSG_STORE, Guids1stHalf),
+ msg_store_read(Guids1stHalf, MSCState6)),
+ MSCState7 = rabbit_msg_store:client_init(?PERSISTENT_MSG_STORE, Ref,
+ undefined),
+ ok = rabbit_msg_store:remove(Guids1stHalf, MSCState7),
+ ok = rabbit_msg_store:client_terminate(MSCState7),
%% restart empty
restart_msg_store_empty(), %% now safe to reuse guids
%% push a lot of msgs in... at least 100 files worth
@@ -1568,31 +1581,39 @@ test_msg_store() ->
BigCount = trunc(100 * FileSize / (PayloadSizeBits div 8)),
GuidsBig = [guid_bin(X) || X <- lists:seq(1, BigCount)],
Payload = << 0:PayloadSizeBits >>,
- ok = foreach_with_msg_store_client(
+ ok = with_msg_store_client(
?PERSISTENT_MSG_STORE, Ref,
- fun (Guid, MsgStore, MSCStateM) ->
- {ok, MSCStateN} = rabbit_msg_store:write(
- MsgStore, Guid, Payload, MSCStateM),
- MSCStateN
- end, GuidsBig),
+ fun (MSCStateM) ->
+ [ok = rabbit_msg_store:write(Guid, Payload, MSCStateM) ||
+ Guid <- GuidsBig],
+ MSCStateM
+ end),
%% now read them to ensure we hit the fast client-side reading
ok = foreach_with_msg_store_client(
?PERSISTENT_MSG_STORE, Ref,
- fun (Guid, MsgStore, MSCStateM) ->
+ fun (Guid, MSCStateM) ->
{{ok, Payload}, MSCStateN} = rabbit_msg_store:read(
- MsgStore, Guid, MSCStateM),
+ Guid, MSCStateM),
MSCStateN
end, GuidsBig),
%% .., then 3s by 1...
- ok = msg_store_remove([guid_bin(X) || X <- lists:seq(BigCount, 1, -3)]),
+ ok = msg_store_remove(?PERSISTENT_MSG_STORE, Ref,
+ [guid_bin(X) || X <- lists:seq(BigCount, 1, -3)]),
%% .., then remove 3s by 2, from the young end first. This hits
%% GC (under 50% good data left, but no empty files. Must GC).
- ok = msg_store_remove([guid_bin(X) || X <- lists:seq(BigCount-1, 1, -3)]),
+ ok = msg_store_remove(?PERSISTENT_MSG_STORE, Ref,
+ [guid_bin(X) || X <- lists:seq(BigCount-1, 1, -3)]),
%% .., then remove 3s by 3, from the young end first. This hits
%% GC...
- ok = msg_store_remove([guid_bin(X) || X <- lists:seq(BigCount-2, 1, -3)]),
+ ok = msg_store_remove(?PERSISTENT_MSG_STORE, Ref,
+ [guid_bin(X) || X <- lists:seq(BigCount-2, 1, -3)]),
%% ensure empty
- false = msg_store_contains(false, GuidsBig),
+ ok = with_msg_store_client(
+ ?PERSISTENT_MSG_STORE, Ref,
+ fun (MSCStateM) ->
+ false = msg_store_contains(false, GuidsBig, MSCStateM),
+ MSCStateM
+ end),
%% restart empty
restart_msg_store_empty(),
passed.
@@ -1604,14 +1625,18 @@ test_queue() ->
queue_name(<<"test">>).
init_test_queue() ->
- rabbit_queue_index:init(
- test_queue(), true, false,
- fun (Guid) ->
- rabbit_msg_store:contains(?PERSISTENT_MSG_STORE, Guid)
- end,
- fun (_) ->
- ok %% Sync!
- end).
+ TestQueue = test_queue(),
+ Terms = rabbit_queue_index:shutdown_terms(TestQueue),
+ PRef = proplists:get_value(persistent_ref, Terms, rabbit_guid:guid()),
+ PersistentClient = rabbit_msg_store:client_init(?PERSISTENT_MSG_STORE,
+ PRef, undefined),
+ Res = rabbit_queue_index:recover(
+ TestQueue, Terms, false,
+ fun (Guid) ->
+ rabbit_msg_store:contains(Guid, PersistentClient)
+ end),
+ ok = rabbit_msg_store:client_delete_and_terminate(PersistentClient),
+ Res.
restart_test_queue(Qi) ->
_ = rabbit_queue_index:terminate([], Qi),
@@ -1622,13 +1647,13 @@ restart_test_queue(Qi) ->
empty_test_queue() ->
ok = rabbit_variable_queue:stop(),
ok = rabbit_variable_queue:start([]),
- {0, _Terms, Qi} = init_test_queue(),
+ {0, Qi} = init_test_queue(),
_ = rabbit_queue_index:delete_and_terminate(Qi),
ok.
with_empty_test_queue(Fun) ->
ok = empty_test_queue(),
- {0, _Terms, Qi} = init_test_queue(),
+ {0, Qi} = init_test_queue(),
rabbit_queue_index:delete_and_terminate(Fun(Qi)).
queue_index_publish(SeqIds, Persistent, Qi) ->
@@ -1637,18 +1662,17 @@ queue_index_publish(SeqIds, Persistent, Qi) ->
true -> ?PERSISTENT_MSG_STORE;
false -> ?TRANSIENT_MSG_STORE
end,
- {A, B, MSCStateEnd} =
+ MSCState = rabbit_msg_store:client_init(MsgStore, Ref, undefined),
+ {A, B} =
lists:foldl(
- fun (SeqId, {QiN, SeqIdsGuidsAcc, MSCStateN}) ->
+ fun (SeqId, {QiN, SeqIdsGuidsAcc}) ->
Guid = rabbit_guid:guid(),
QiM = rabbit_queue_index:publish(
Guid, SeqId, #message_properties{}, Persistent, QiN),
- {ok, MSCStateM} = rabbit_msg_store:write(MsgStore, Guid,
- Guid, MSCStateN),
- {QiM, [{SeqId, Guid} | SeqIdsGuidsAcc], MSCStateM}
- end, {Qi, [], rabbit_msg_store:client_init(MsgStore, Ref, undefined)}, SeqIds),
- ok = rabbit_msg_store:client_delete_and_terminate(
- MSCStateEnd, MsgStore, Ref),
+ ok = rabbit_msg_store:write(Guid, Guid, MSCState),
+ {QiM, [{SeqId, Guid} | SeqIdsGuidsAcc]}
+ end, {Qi, []}, SeqIds),
+ ok = rabbit_msg_store:client_delete_and_terminate(MSCState),
{A, B}.
verify_read_with_published(_Delivered, _Persistent, [], _) ->
@@ -1694,7 +1718,7 @@ test_queue_index() ->
ok = verify_read_with_published(false, false, ReadA,
lists:reverse(SeqIdsGuidsA)),
%% should get length back as 0, as all the msgs were transient
- {0, _Terms1, Qi6} = restart_test_queue(Qi4),
+ {0, Qi6} = restart_test_queue(Qi4),
{0, 0, Qi7} = rabbit_queue_index:bounds(Qi6),
{Qi8, SeqIdsGuidsB} = queue_index_publish(SeqIdsB, true, Qi7),
{0, TwoSegs, Qi9} = rabbit_queue_index:bounds(Qi8),
@@ -1703,7 +1727,7 @@ test_queue_index() ->
lists:reverse(SeqIdsGuidsB)),
%% should get length back as MostOfASegment
LenB = length(SeqIdsB),
- {LenB, _Terms2, Qi12} = restart_test_queue(Qi10),
+ {LenB, Qi12} = restart_test_queue(Qi10),
{0, TwoSegs, Qi13} = rabbit_queue_index:bounds(Qi12),
Qi14 = rabbit_queue_index:deliver(SeqIdsB, Qi13),
{ReadC, Qi15} = rabbit_queue_index:read(0, SegmentSize, Qi14),
@@ -1715,7 +1739,7 @@ test_queue_index() ->
{0, 0, Qi18} = rabbit_queue_index:bounds(Qi17),
%% should get length back as 0 because all persistent
%% msgs have been acked
- {0, _Terms3, Qi19} = restart_test_queue(Qi18),
+ {0, Qi19} = restart_test_queue(Qi18),
Qi19
end),
@@ -1787,11 +1811,11 @@ test_queue_index() ->
true, Qi0),
Qi2 = rabbit_queue_index:deliver([0,1,4], Qi1),
Qi3 = rabbit_queue_index:ack([0], Qi2),
- {5, _Terms9, Qi4} = restart_test_queue(Qi3),
+ {5, Qi4} = restart_test_queue(Qi3),
{Qi5, _SeqIdsGuidsF} = queue_index_publish([3,6,8], true, Qi4),
Qi6 = rabbit_queue_index:deliver([2,3,5,6], Qi5),
Qi7 = rabbit_queue_index:ack([1,2,3], Qi6),
- {5, _Terms10, Qi8} = restart_test_queue(Qi7),
+ {5, Qi8} = restart_test_queue(Qi7),
Qi8
end),
@@ -2033,4 +2057,57 @@ test_queue_recover() ->
end),
passed.
+test_configurable_server_properties() ->
+ %% List of the names of the built-in properties do we expect to find
+ BuiltInPropNames = [<<"product">>, <<"version">>, <<"platform">>,
+ <<"copyright">>, <<"information">>],
+
+ %% Verify that the built-in properties are initially present
+ ActualPropNames = [Key ||
+ {Key, longstr, _} <- rabbit_reader:server_properties()],
+ true = lists:all(fun (X) -> lists:member(X, ActualPropNames) end,
+ BuiltInPropNames),
+
+ %% Get the initial server properties configured in the environment
+ {ok, ServerProperties} = application:get_env(rabbit, server_properties),
+
+ %% Helper functions
+ ConsProp = fun (X) -> application:set_env(rabbit,
+ server_properties,
+ [X | ServerProperties]) end,
+ IsPropPresent = fun (X) -> lists:member(X,
+ rabbit_reader:server_properties())
+ end,
+
+ %% Add a wholly new property of the simplified {KeyAtom, StringValue} form
+ NewSimplifiedProperty = {NewHareKey, NewHareVal} = {hare, "soup"},
+ ConsProp(NewSimplifiedProperty),
+ %% Do we find hare soup, appropriately formatted in the generated properties?
+ ExpectedHareImage = {list_to_binary(atom_to_list(NewHareKey)),
+ longstr,
+ list_to_binary(NewHareVal)},
+ true = IsPropPresent(ExpectedHareImage),
+
+ %% Add a wholly new property of the {BinaryKey, Type, Value} form
+ %% and check for it
+ NewProperty = {<<"new-bin-key">>, signedint, -1},
+ ConsProp(NewProperty),
+ %% Do we find the new property?
+ true = IsPropPresent(NewProperty),
+
+ %% Add a property that clobbers a built-in, and verify correct clobbering
+ {NewVerKey, NewVerVal} = NewVersion = {version, "X.Y.Z."},
+ {BinNewVerKey, BinNewVerVal} = {list_to_binary(atom_to_list(NewVerKey)),
+ list_to_binary(NewVerVal)},
+ ConsProp(NewVersion),
+ ClobberedServerProps = rabbit_reader:server_properties(),
+ %% Is the clobbering insert present?
+ true = IsPropPresent({BinNewVerKey, longstr, BinNewVerVal}),
+ %% Is the clobbering insert the only thing with the clobbering key?
+ [{BinNewVerKey, longstr, BinNewVerVal}] =
+ [E || {K, longstr, _V} = E <- ClobberedServerProps, K =:= BinNewVerKey],
+
+ application:set_env(rabbit, server_properties, ServerProperties),
+ passed.
+
nop(_) -> ok.
diff --git a/src/rabbit_variable_queue.erl b/src/rabbit_variable_queue.erl
index bd7be94099..8463d3d55e 100644
--- a/src/rabbit_variable_queue.erl
+++ b/src/rabbit_variable_queue.erl
@@ -382,18 +382,18 @@ init(QueueName, IsDurable, Recover) ->
fun (Guids) -> msgs_written_to_disk(Self, Guids) end,
fun (Guids) -> msg_indices_written_to_disk(Self, Guids) end).
-init(QueueName, IsDurable, Recover,
- MsgOnDiskFun, MsgIdxOnDiskFun) ->
- {DeltaCount, Terms, IndexState} =
- rabbit_queue_index:init(
- QueueName, Recover,
- rabbit_msg_store:successfully_recovered_state(?PERSISTENT_MSG_STORE),
- fun (Guid) ->
- rabbit_msg_store:contains(?PERSISTENT_MSG_STORE, Guid)
- end,
- MsgIdxOnDiskFun),
- {LowSeqId, NextSeqId, IndexState1} = rabbit_queue_index:bounds(IndexState),
+init(QueueName, IsDurable, false, MsgOnDiskFun, MsgIdxOnDiskFun) ->
+ IndexState = rabbit_queue_index:init(QueueName, MsgIdxOnDiskFun),
+ init(IsDurable, IndexState, 0, [],
+ case IsDurable of
+ true -> msg_store_client_init(?PERSISTENT_MSG_STORE,
+ MsgOnDiskFun);
+ false -> undefined
+ end,
+ msg_store_client_init(?TRANSIENT_MSG_STORE, undefined));
+init(QueueName, true, true, MsgOnDiskFun, MsgIdxOnDiskFun) ->
+ Terms = rabbit_queue_index:shutdown_terms(QueueName),
{PRef, TRef, Terms1} =
case [persistent_ref, transient_ref] -- proplists:get_keys(Terms) of
[] -> {proplists:get_value(persistent_ref, Terms),
@@ -401,71 +401,33 @@ init(QueueName, IsDurable, Recover,
Terms};
_ -> {rabbit_guid:guid(), rabbit_guid:guid(), []}
end,
- DeltaCount1 = proplists:get_value(persistent_count, Terms1, DeltaCount),
- Delta = case DeltaCount1 == 0 andalso DeltaCount /= undefined of
- true -> ?BLANK_DELTA;
- false -> #delta { start_seq_id = LowSeqId,
- count = DeltaCount1,
- end_seq_id = NextSeqId }
- end,
- Now = now(),
-
- PersistentClient =
- case IsDurable of
- true -> rabbit_msg_store:client_init(?PERSISTENT_MSG_STORE, PRef,
- MsgOnDiskFun);
- false -> undefined
- end,
- TransientClient =
- rabbit_msg_store:client_init(?TRANSIENT_MSG_STORE, TRef, undefined),
-
- State = #vqstate {
- q1 = queue:new(),
- q2 = bpqueue:new(),
- delta = Delta,
- q3 = bpqueue:new(),
- q4 = queue:new(),
- next_seq_id = NextSeqId,
- pending_ack = dict:new(),
- index_state = IndexState1,
- msg_store_clients = {{PersistentClient, PRef},
- {TransientClient, TRef}},
- on_sync = ?BLANK_SYNC,
- durable = IsDurable,
- transient_threshold = NextSeqId,
-
- len = DeltaCount1,
- persistent_count = DeltaCount1,
-
- duration_target = infinity,
- target_ram_msg_count = infinity,
- ram_msg_count = 0,
- ram_msg_count_prev = 0,
- ram_index_count = 0,
- out_counter = 0,
- in_counter = 0,
- rates = #rates { egress = {Now, 0},
- ingress = {Now, DeltaCount1},
- avg_egress = 0.0,
- avg_ingress = 0.0,
- timestamp = Now },
- msgs_on_disk = gb_sets:new(),
- msg_indices_on_disk = gb_sets:new(),
- unconfirmed = gb_sets:new() },
- a(maybe_deltas_to_betas(State)).
+ PersistentClient = rabbit_msg_store:client_init(?PERSISTENT_MSG_STORE,
+ PRef, MsgOnDiskFun),
+ TransientClient = rabbit_msg_store:client_init(?TRANSIENT_MSG_STORE,
+ TRef, undefined),
+ {DeltaCount, IndexState} =
+ rabbit_queue_index:recover(
+ QueueName, Terms1,
+ rabbit_msg_store:successfully_recovered_state(?PERSISTENT_MSG_STORE),
+ fun (Guid) ->
+ rabbit_msg_store:contains(Guid, PersistentClient)
+ end,
+ MsgIdxOnDiskFun),
+ init(true, IndexState, DeltaCount, Terms1,
+ PersistentClient, TransientClient).
terminate(State) ->
State1 = #vqstate { persistent_count = PCount,
index_state = IndexState,
- msg_store_clients = {{MSCStateP, PRef},
- {MSCStateT, TRef}} } =
+ msg_store_clients = {MSCStateP, MSCStateT} } =
remove_pending_ack(true, tx_commit_index(State)),
- case MSCStateP of
- undefined -> ok;
- _ -> rabbit_msg_store:client_terminate(
- MSCStateP, ?PERSISTENT_MSG_STORE)
- end,
- rabbit_msg_store:client_terminate(MSCStateT, ?TRANSIENT_MSG_STORE),
+ PRef = case MSCStateP of
+ undefined -> undefined;
+ _ -> ok = rabbit_msg_store:client_terminate(MSCStateP),
+ rabbit_msg_store:client_ref(MSCStateP)
+ end,
+ ok = rabbit_msg_store:client_terminate(MSCStateT),
+ TRef = rabbit_msg_store:client_ref(MSCStateT),
Terms = [{persistent_ref, PRef},
{transient_ref, TRef},
{persistent_count, PCount}],
@@ -481,44 +443,44 @@ delete_and_terminate(State) ->
%% deleting it.
{_PurgeCount, State1} = purge(State),
State2 = #vqstate { index_state = IndexState,
- msg_store_clients = {{MSCStateP, PRef},
- {MSCStateT, TRef}} } =
+ msg_store_clients = {MSCStateP, MSCStateT} } =
remove_pending_ack(false, State1),
IndexState1 = rabbit_queue_index:delete_and_terminate(IndexState),
case MSCStateP of
undefined -> ok;
- _ -> rabbit_msg_store:client_delete_and_terminate(
- MSCStateP, ?PERSISTENT_MSG_STORE, PRef)
+ _ -> rabbit_msg_store:client_delete_and_terminate(MSCStateP)
end,
- rabbit_msg_store:client_delete_and_terminate(
- MSCStateT, ?TRANSIENT_MSG_STORE, TRef),
+ rabbit_msg_store:client_delete_and_terminate(MSCStateT),
a(State2 #vqstate { index_state = IndexState1,
msg_store_clients = undefined }).
-purge(State = #vqstate { q4 = Q4,
- index_state = IndexState,
- len = Len,
- persistent_count = PCount }) ->
+purge(State = #vqstate { q4 = Q4,
+ index_state = IndexState,
+ msg_store_clients = MSCState,
+ 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.
{LensByStore, IndexState1} = remove_queue_entries(
fun rabbit_misc:queue_fold/3, Q4,
- orddict:new(), IndexState),
- {LensByStore1, State1 = #vqstate { q1 = Q1, index_state = IndexState2 }} =
+ orddict:new(), IndexState, MSCState),
+ {LensByStore1, State1 = #vqstate { q1 = Q1,
+ index_state = IndexState2,
+ msg_store_clients = MSCState1 }} =
purge_betas_and_deltas(LensByStore,
State #vqstate { q4 = queue:new(),
index_state = IndexState1 }),
{LensByStore2, IndexState3} = remove_queue_entries(
fun rabbit_misc:queue_fold/3, Q1,
- LensByStore1, IndexState2),
+ LensByStore1, IndexState2, MSCState1),
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 = PCount1 })}.
+ {Len, a(State1 #vqstate { q1 = queue:new(),
+ index_state = IndexState3,
+ len = 0,
+ ram_msg_count = 0,
+ ram_index_count = 0,
+ persistent_count = PCount1 })}.
publish(Msg, MsgProps, State) ->
{_SeqId, State1} = publish(Msg, MsgProps, false, false, State),
@@ -599,7 +561,7 @@ read_msg(MsgStatus = #msg_status { msg = undefined,
State = #vqstate { ram_msg_count = RamMsgCount,
msg_store_clients = MSCState}) ->
{{ok, Msg = #basic_message {}}, MSCState1} =
- read_from_msg_store(MSCState, IsPersistent, Guid),
+ msg_store_read(MSCState, IsPersistent, Guid),
{MsgStatus #msg_status { msg = Msg },
State #vqstate { ram_msg_count = RamMsgCount + 1,
msg_store_clients = MSCState1 }};
@@ -614,20 +576,22 @@ internal_fetch(AckRequired, MsgStatus = #msg_status {
is_delivered = IsDelivered,
msg_on_disk = MsgOnDisk,
index_on_disk = IndexOnDisk },
- State = #vqstate {ram_msg_count = RamMsgCount,
- out_counter = OutCount,
- index_state = IndexState,
- len = Len,
- persistent_count = PCount,
- pending_ack = PA }) ->
+ State = #vqstate {ram_msg_count = RamMsgCount,
+ out_counter = OutCount,
+ index_state = IndexState,
+ msg_store_clients = MSCState,
+ len = Len,
+ persistent_count = PCount,
+ pending_ack = PA }) ->
%% 1. Mark it delivered if necessary
IndexState1 = maybe_write_delivered(
IndexOnDisk andalso not IsDelivered,
SeqId, IndexState),
%% 2. Remove from msg_store and queue index, if necessary
- MsgStore = find_msg_store(IsPersistent),
- Rem = fun () -> ok = rabbit_msg_store:remove(MsgStore, [Guid]) end,
+ Rem = fun () ->
+ ok = msg_store_remove(MSCState, IsPersistent, [Guid])
+ end,
Ack = fun () -> rabbit_queue_index:ack([SeqId], IndexState1) end,
IndexState2 =
case {AckRequired, MsgOnDisk, IndexOnDisk, IsPersistent} of
@@ -659,7 +623,7 @@ internal_fetch(AckRequired, MsgStatus = #msg_status {
pending_ack = PA1 })}.
ack(AckTags, State) ->
- {Guids, State1} = ack(fun rabbit_msg_store:remove/2,
+ {Guids, State1} = ack(fun msg_store_remove/3,
fun (_AckEntry, State1) -> State1 end,
AckTags, State),
{Guids, a(State1)}.
@@ -669,30 +633,32 @@ tx_publish(Txn, Msg = #basic_message { is_persistent = IsPersistent }, MsgProps,
msg_store_clients = MSCState }) ->
Tx = #tx { pending_messages = Pubs } = lookup_tx(Txn),
store_tx(Txn, Tx #tx { pending_messages = [{Msg, MsgProps} | Pubs] }),
- a(case IsPersistent andalso IsDurable of
- true -> MsgStatus = msg_status(true, undefined, Msg, MsgProps),
- {#msg_status { msg_on_disk = true }, MSCState1} =
- maybe_write_msg_to_disk(false, MsgStatus, MSCState),
- State #vqstate { msg_store_clients = MSCState1 };
- false -> State
- end).
+ case IsPersistent andalso IsDurable of
+ true -> MsgStatus = msg_status(true, undefined, Msg, MsgProps),
+ #msg_status { msg_on_disk = true } =
+ maybe_write_msg_to_disk(false, MsgStatus, MSCState);
+ false -> ok
+ end,
+ a(State).
tx_ack(Txn, AckTags, State) ->
Tx = #tx { pending_acks = Acks } = lookup_tx(Txn),
store_tx(Txn, Tx #tx { pending_acks = [AckTags | Acks] }),
State.
-tx_rollback(Txn, State = #vqstate { durable = IsDurable }) ->
+tx_rollback(Txn, State = #vqstate { durable = IsDurable,
+ msg_store_clients = MSCState }) ->
#tx { pending_acks = AckTags, pending_messages = Pubs } = lookup_tx(Txn),
erase_tx(Txn),
ok = case IsDurable of
- true -> rabbit_msg_store:remove(?PERSISTENT_MSG_STORE,
- persistent_guids(Pubs));
+ true -> msg_store_remove(MSCState, true, persistent_guids(Pubs));
false -> ok
end,
{lists:append(AckTags), a(State)}.
-tx_commit(Txn, Fun, MsgPropsFun, State = #vqstate { durable = IsDurable }) ->
+tx_commit(Txn, Fun, MsgPropsFun,
+ State = #vqstate { durable = IsDurable,
+ msg_store_clients = MSCState }) ->
#tx { pending_acks = AckTags, pending_messages = Pubs } = lookup_tx(Txn),
erase_tx(Txn),
AckTags1 = lists:append(AckTags),
@@ -700,8 +666,8 @@ tx_commit(Txn, Fun, MsgPropsFun, State = #vqstate { durable = IsDurable }) ->
HasPersistentPubs = PersistentGuids =/= [],
{AckTags1,
a(case IsDurable andalso HasPersistentPubs of
- true -> ok = rabbit_msg_store:sync(
- ?PERSISTENT_MSG_STORE, PersistentGuids,
+ true -> ok = msg_store_sync(
+ MSCState, true, PersistentGuids,
msg_store_callback(PersistentGuids, Pubs, AckTags1,
Fun, MsgPropsFun)),
State;
@@ -711,7 +677,7 @@ tx_commit(Txn, Fun, MsgPropsFun, State = #vqstate { durable = IsDurable }) ->
requeue(AckTags, MsgPropsFun, State) ->
{_Guids, State1} =
- ack(fun rabbit_msg_store:release/2,
+ ack(fun msg_store_release/3,
fun (#msg_status { msg = Msg, msg_props = MsgProps }, State1) ->
{_SeqId, State2} = publish(Msg, MsgPropsFun(MsgProps),
true, false, State1),
@@ -719,7 +685,7 @@ requeue(AckTags, MsgPropsFun, State) ->
({IsPersistent, Guid, MsgProps}, State1) ->
#vqstate { msg_store_clients = MSCState } = State1,
{{ok, Msg = #basic_message{}}, MSCState1} =
- read_from_msg_store(MSCState, IsPersistent, Guid),
+ msg_store_read(MSCState, IsPersistent, Guid),
State2 = State1 #vqstate { msg_store_clients = MSCState1 },
{_SeqId, State3} = publish(Msg, MsgPropsFun(MsgProps),
true, true, State2),
@@ -881,22 +847,47 @@ msg_status(IsPersistent, SeqId, Msg = #basic_message { guid = Guid },
msg_on_disk = false, index_on_disk = false,
msg_props = MsgProps }.
-find_msg_store(true) -> ?PERSISTENT_MSG_STORE;
-find_msg_store(false) -> ?TRANSIENT_MSG_STORE.
-
-with_msg_store_state({{MSCStateP, PRef}, MSCStateT}, true, Fun) ->
- {Result, MSCStateP1} = Fun(?PERSISTENT_MSG_STORE, MSCStateP),
- {Result, {{MSCStateP1, PRef}, MSCStateT}};
-with_msg_store_state({MSCStateP, {MSCStateT, TRef}}, false, Fun) ->
- {Result, MSCStateT1} = Fun(?TRANSIENT_MSG_STORE, MSCStateT),
- {Result, {MSCStateP, {MSCStateT1, TRef}}}.
+with_msg_store_state({MSCStateP, MSCStateT}, true, Fun) ->
+ {Result, MSCStateP1} = Fun(MSCStateP),
+ {Result, {MSCStateP1, MSCStateT}};
+with_msg_store_state({MSCStateP, MSCStateT}, false, Fun) ->
+ {Result, MSCStateT1} = Fun(MSCStateT),
+ {Result, {MSCStateP, MSCStateT1}}.
+
+with_immutable_msg_store_state(MSCState, IsPersistent, Fun) ->
+ {Res, MSCState} = with_msg_store_state(MSCState, IsPersistent,
+ fun (MSCState1) ->
+ {Fun(MSCState1), MSCState1}
+ end),
+ Res.
+
+msg_store_client_init(MsgStore, MsgOnDiskFun) ->
+ rabbit_msg_store:client_init(MsgStore, rabbit_guid:guid(), MsgOnDiskFun).
+
+msg_store_write(MSCState, IsPersistent, Guid, Msg) ->
+ with_immutable_msg_store_state(
+ MSCState, IsPersistent,
+ fun (MSCState1) -> rabbit_msg_store:write(Guid, Msg, MSCState1) end).
-read_from_msg_store(MSCState, IsPersistent, Guid) ->
+msg_store_read(MSCState, IsPersistent, Guid) ->
with_msg_store_state(
MSCState, IsPersistent,
- fun (MsgStore, MSCState1) ->
- rabbit_msg_store:read(MsgStore, Guid, MSCState1)
- end).
+ fun (MSCState1) -> rabbit_msg_store:read(Guid, MSCState1) end).
+
+msg_store_remove(MSCState, IsPersistent, Guids) ->
+ with_immutable_msg_store_state(
+ MSCState, IsPersistent,
+ fun (MCSState1) -> rabbit_msg_store:remove(Guids, MCSState1) end).
+
+msg_store_release(MSCState, IsPersistent, Guids) ->
+ with_immutable_msg_store_state(
+ MSCState, IsPersistent,
+ fun (MCSState1) -> rabbit_msg_store:release(Guids, MCSState1) end).
+
+msg_store_sync(MSCState, IsPersistent, Guids, Callback) ->
+ with_immutable_msg_store_state(
+ MSCState, IsPersistent,
+ fun (MSCState1) -> rabbit_msg_store:sync(Guids, Callback, MSCState1) end).
maybe_write_delivered(false, _SeqId, IndexState) ->
IndexState;
@@ -980,6 +971,52 @@ update_rate(Now, Then, Count, {OThen, OCount}) ->
%% Internal major helpers for Public API
%%----------------------------------------------------------------------------
+init(IsDurable, IndexState, DeltaCount, Terms,
+ PersistentClient, TransientClient) ->
+ {LowSeqId, NextSeqId, IndexState1} = rabbit_queue_index:bounds(IndexState),
+
+ DeltaCount1 = proplists:get_value(persistent_count, Terms, DeltaCount),
+ Delta = case DeltaCount1 == 0 andalso DeltaCount /= undefined of
+ true -> ?BLANK_DELTA;
+ false -> #delta { start_seq_id = LowSeqId,
+ count = DeltaCount1,
+ end_seq_id = NextSeqId }
+ end,
+ Now = now(),
+ State = #vqstate {
+ q1 = queue:new(),
+ q2 = bpqueue:new(),
+ delta = Delta,
+ q3 = bpqueue:new(),
+ q4 = queue:new(),
+ next_seq_id = NextSeqId,
+ pending_ack = dict:new(),
+ index_state = IndexState1,
+ msg_store_clients = {PersistentClient, TransientClient},
+ on_sync = ?BLANK_SYNC,
+ durable = IsDurable,
+ transient_threshold = NextSeqId,
+
+ len = DeltaCount1,
+ persistent_count = DeltaCount1,
+
+ duration_target = infinity,
+ target_ram_msg_count = infinity,
+ ram_msg_count = 0,
+ ram_msg_count_prev = 0,
+ ram_index_count = 0,
+ out_counter = 0,
+ in_counter = 0,
+ msgs_on_disk = gb_sets:new(),
+ msg_indices_on_disk = gb_sets:new(),
+ unconfirmed = gb_sets:new(),
+ rates = #rates { egress = {Now, 0},
+ ingress = {Now, DeltaCount1},
+ avg_egress = 0.0,
+ avg_ingress = 0.0,
+ timestamp = Now }},
+ a(maybe_deltas_to_betas(State)).
+
msg_store_callback(PersistentGuids, Pubs, AckTags, Fun, MsgPropsFun) ->
Self = self(),
F = fun () -> rabbit_amqqueue:maybe_run_queue_via_backing_queue(
@@ -989,13 +1026,17 @@ msg_store_callback(PersistentGuids, Pubs, AckTags, Fun, MsgPropsFun) ->
end)
end,
fun () -> spawn(fun () -> ok = rabbit_misc:with_exit_handler(
- fun () -> rabbit_msg_store:remove(
- ?PERSISTENT_MSG_STORE,
+ fun () -> remove_persistent_messages(
PersistentGuids)
end, F)
end)
end.
+remove_persistent_messages(Guids) ->
+ PersistentClient = msg_store_client_init(?PERSISTENT_MSG_STORE, undefined),
+ ok = rabbit_msg_store:remove(Guids, PersistentClient),
+ rabbit_msg_store:client_delete_and_terminate(PersistentClient).
+
tx_commit_post_msg_store(HasPersistentPubs, Pubs, AckTags, Fun, MsgPropsFun,
State = #vqstate {
on_sync = OnSync = #sync {
@@ -1062,13 +1103,14 @@ tx_commit_index(State = #vqstate { on_sync = #sync {
State1 #vqstate { index_state = IndexState1, on_sync = ?BLANK_SYNC }).
purge_betas_and_deltas(LensByStore,
- State = #vqstate { q3 = Q3,
- index_state = IndexState }) ->
+ State = #vqstate { q3 = Q3,
+ index_state = IndexState,
+ msg_store_clients = MSCState }) ->
case bpqueue:is_empty(Q3) of
true -> {LensByStore, State};
- false -> {LensByStore1, IndexState1} = remove_queue_entries(
- fun beta_fold/3, Q3,
- LensByStore, IndexState),
+ false -> {LensByStore1, IndexState1} =
+ remove_queue_entries(fun beta_fold/3, Q3,
+ LensByStore, IndexState, MSCState),
purge_betas_and_deltas(LensByStore1,
maybe_deltas_to_betas(
State #vqstate {
@@ -1076,11 +1118,11 @@ purge_betas_and_deltas(LensByStore,
index_state = IndexState1 }))
end.
-remove_queue_entries(Fold, Q, LensByStore, IndexState) ->
+remove_queue_entries(Fold, Q, LensByStore, IndexState, MSCState) ->
{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)
+ ok = orddict:fold(fun (IsPersistent, Guids, ok) ->
+ msg_store_remove(MSCState, IsPersistent, Guids)
end, ok, GuidsByStore),
{sum_guids_by_store_to_len(LensByStore, GuidsByStore),
rabbit_queue_index:ack(Acks,
@@ -1092,8 +1134,7 @@ remove_queue_entries1(
index_on_disk = IndexOnDisk, is_persistent = IsPersistent },
{GuidsByStore, Delivers, Acks}) ->
{case MsgOnDisk of
- true -> rabbit_misc:orddict_cons(find_msg_store(IsPersistent), Guid,
- GuidsByStore);
+ true -> rabbit_misc:orddict_cons(IsPersistent, Guid, GuidsByStore);
false -> GuidsByStore
end,
cons_if(IndexOnDisk andalso not IsDelivered, SeqId, Delivers),
@@ -1101,8 +1142,8 @@ remove_queue_entries1(
sum_guids_by_store_to_len(LensByStore, GuidsByStore) ->
orddict:fold(
- fun (MsgStore, Guids, LensByStore1) ->
- orddict:update_counter(MsgStore, length(Guids), LensByStore1)
+ fun (IsPersistent, Guids, LensByStore1) ->
+ orddict:update_counter(IsPersistent, length(Guids), LensByStore1)
end, LensByStore, GuidsByStore).
%%----------------------------------------------------------------------------
@@ -1138,8 +1179,8 @@ publish(Msg = #basic_message { is_persistent = IsPersistent, guid = Guid },
unconfirmed = Unconfirmed1 }}.
maybe_write_msg_to_disk(_Force, MsgStatus = #msg_status {
- msg_on_disk = true }, MSCState) ->
- {MsgStatus, MSCState};
+ msg_on_disk = true }, _MSCState) ->
+ MsgStatus;
maybe_write_msg_to_disk(Force, MsgStatus = #msg_status {
msg = Msg, guid = Guid,
is_persistent = IsPersistent }, MSCState)
@@ -1148,15 +1189,10 @@ maybe_write_msg_to_disk(Force, MsgStatus = #msg_status {
%% don't persist any recoverable decoded properties
content = rabbit_binary_parser:clear_decoded_content(
Msg #basic_message.content)},
- {ok, MSCState1} = with_msg_store_state(
- MSCState, IsPersistent,
- fun (MsgStore, MSCState2) ->
- rabbit_msg_store:write(MsgStore, Guid, Msg1,
- MSCState2)
- end),
- {MsgStatus #msg_status { msg_on_disk = true }, MSCState1};
-maybe_write_msg_to_disk(_Force, MsgStatus, MSCState) ->
- {MsgStatus, MSCState}.
+ ok = msg_store_write(MSCState, IsPersistent, Guid, Msg1),
+ MsgStatus #msg_status { msg_on_disk = true };
+maybe_write_msg_to_disk(_Force, MsgStatus, _MSCState) ->
+ MsgStatus.
maybe_write_index_to_disk(_Force, MsgStatus = #msg_status {
index_on_disk = true }, IndexState) ->
@@ -1180,12 +1216,10 @@ maybe_write_index_to_disk(_Force, MsgStatus, IndexState) ->
maybe_write_to_disk(ForceMsg, ForceIndex, MsgStatus,
State = #vqstate { index_state = IndexState,
msg_store_clients = MSCState }) ->
- {MsgStatus1, MSCState1} = maybe_write_msg_to_disk(
- ForceMsg, MsgStatus, MSCState),
- {MsgStatus2, IndexState1} = maybe_write_index_to_disk(
- ForceIndex, MsgStatus1, IndexState),
- {MsgStatus2, State #vqstate { index_state = IndexState1,
- msg_store_clients = MSCState1 }}.
+ MsgStatus1 = maybe_write_msg_to_disk(ForceMsg, MsgStatus, MSCState),
+ {MsgStatus2, IndexState1} =
+ maybe_write_index_to_disk(ForceIndex, MsgStatus1, IndexState),
+ {MsgStatus2, State #vqstate { index_state = IndexState1 }}.
%%----------------------------------------------------------------------------
%% Internal gubbins for acks
@@ -1203,22 +1237,23 @@ record_pending_ack(#msg_status { seq_id = SeqId,
dict:store(SeqId, AckEntry, PA).
remove_pending_ack(KeepPersistent,
- State = #vqstate { pending_ack = PA,
- index_state = IndexState }) ->
+ State = #vqstate { pending_ack = PA,
+ index_state = IndexState,
+ msg_store_clients = MSCState }) ->
{SeqIds, GuidsByStore} = dict:fold(fun accumulate_ack/3,
{[], orddict:new()}, PA),
State1 = State #vqstate { pending_ack = dict:new() },
case KeepPersistent of
- true -> case orddict:find(?TRANSIENT_MSG_STORE, GuidsByStore) of
+ true -> case orddict:find(false, GuidsByStore) of
error -> State1;
- {ok, Guids} -> ok = rabbit_msg_store:remove(
- ?TRANSIENT_MSG_STORE, Guids),
+ {ok, Guids} -> ok = msg_store_remove(MSCState, false,
+ Guids),
State1
end;
false -> IndexState1 = rabbit_queue_index:ack(SeqIds, IndexState),
ok = orddict:fold(
- fun (MsgStore, Guids, ok) ->
- rabbit_msg_store:remove(MsgStore, Guids)
+ fun (IsPersistent, Guids, ok) ->
+ msg_store_remove(MSCState, IsPersistent, Guids)
end, ok, GuidsByStore),
State1 #vqstate { index_state = IndexState1 }
end.
@@ -1226,8 +1261,10 @@ remove_pending_ack(KeepPersistent,
ack(_MsgStoreFun, _Fun, [], State) ->
{[], State};
ack(MsgStoreFun, Fun, AckTags, State) ->
- {{SeqIds, GuidsByStore}, State1 = #vqstate { index_state = IndexState,
- persistent_count = PCount }} =
+ {{SeqIds, GuidsByStore},
+ State1 = #vqstate { index_state = IndexState,
+ msg_store_clients = MSCState,
+ persistent_count = PCount }} =
lists:foldl(
fun (SeqId, {Acc, State2 = #vqstate { pending_ack = PA }}) ->
{ok, AckEntry} = dict:find(SeqId, PA),
@@ -1237,8 +1274,8 @@ ack(MsgStoreFun, Fun, AckTags, State) ->
end, {{[], orddict:new()}, State}, AckTags),
IndexState1 = rabbit_queue_index:ack(SeqIds, IndexState),
AckdGuids = lists:concat(
- orddict:fold(fun (MsgStore, Guids, Gs) ->
- MsgStoreFun(MsgStore, Guids),
+ orddict:fold(fun (IsPersistent, Guids, Gs) ->
+ MsgStoreFun(MSCState, IsPersistent, Guids),
[Guids | Gs]
end, [], GuidsByStore)),
State2 = remove_confirms(gb_sets:from_list(AckdGuids), State1),
@@ -1253,10 +1290,10 @@ accumulate_ack(_SeqId, #msg_status { is_persistent = false, %% ASSERTIONS
Acc;
accumulate_ack(SeqId, {IsPersistent, Guid, _MsgProps}, {SeqIdsAcc, Dict}) ->
{cons_if(IsPersistent, SeqId, SeqIdsAcc),
- rabbit_misc:orddict_cons(find_msg_store(IsPersistent), Guid, Dict)}.
+ rabbit_misc:orddict_cons(IsPersistent, Guid, Dict)}.
find_persistent_count(LensByStore) ->
- case orddict:find(?PERSISTENT_MSG_STORE, LensByStore) of
+ case orddict:find(true, LensByStore) of
error -> 0;
{ok, Len} -> Len
end.
@@ -1438,46 +1475,40 @@ maybe_deltas_to_betas(State = #vqstate {
delta = Delta,
q3 = Q3,
index_state = IndexState,
- target_ram_msg_count = TargetRamMsgCount,
transient_threshold = TransientThreshold }) ->
- case bpqueue:is_empty(Q3) orelse (TargetRamMsgCount /= 0) of
- false ->
- State;
- true ->
- #delta { start_seq_id = DeltaSeqId,
- count = DeltaCount,
- end_seq_id = DeltaSeqIdEnd } = Delta,
- DeltaSeqId1 =
- lists:min([rabbit_queue_index:next_segment_boundary(DeltaSeqId),
- DeltaSeqIdEnd]),
- {List, IndexState1} =
- rabbit_queue_index:read(DeltaSeqId, DeltaSeqId1, IndexState),
- {Q3a, IndexState2} = betas_from_index_entries(
- List, TransientThreshold, IndexState1),
- State1 = State #vqstate { index_state = IndexState2 },
- case bpqueue:len(Q3a) of
+ #delta { start_seq_id = DeltaSeqId,
+ count = DeltaCount,
+ end_seq_id = DeltaSeqIdEnd } = Delta,
+ DeltaSeqId1 =
+ lists:min([rabbit_queue_index:next_segment_boundary(DeltaSeqId),
+ DeltaSeqIdEnd]),
+ {List, IndexState1} =
+ rabbit_queue_index:read(DeltaSeqId, DeltaSeqId1, IndexState),
+ {Q3a, IndexState2} =
+ betas_from_index_entries(List, TransientThreshold, IndexState1),
+ State1 = State #vqstate { index_state = IndexState2 },
+ case bpqueue:len(Q3a) of
+ 0 ->
+ %% we ignored every message in the segment due to it being
+ %% transient and below the threshold
+ maybe_deltas_to_betas(
+ State1 #vqstate {
+ delta = Delta #delta { start_seq_id = DeltaSeqId1 }});
+ Q3aLen ->
+ Q3b = bpqueue:join(Q3, Q3a),
+ case DeltaCount - Q3aLen of
0 ->
- %% we ignored every message in the segment due to
- %% it being transient and below the threshold
- maybe_deltas_to_betas(
- State #vqstate {
- delta = Delta #delta { start_seq_id = DeltaSeqId1 }});
- Q3aLen ->
- Q3b = bpqueue:join(Q3, Q3a),
- case DeltaCount - Q3aLen of
- 0 ->
- %% delta is now empty, but it wasn't
- %% before, so can now join q2 onto q3
- State1 #vqstate { q2 = bpqueue:new(),
- delta = ?BLANK_DELTA,
- q3 = bpqueue:join(Q3b, Q2) };
- N when N > 0 ->
- Delta1 = #delta { start_seq_id = DeltaSeqId1,
- count = N,
- end_seq_id = DeltaSeqIdEnd },
- State1 #vqstate { delta = Delta1,
- q3 = Q3b }
- end
+ %% delta is now empty, but it wasn't before, so
+ %% can now join q2 onto q3
+ State1 #vqstate { q2 = bpqueue:new(),
+ delta = ?BLANK_DELTA,
+ q3 = bpqueue:join(Q3b, Q2) };
+ N when N > 0 ->
+ Delta1 = #delta { start_seq_id = DeltaSeqId1,
+ count = N,
+ end_seq_id = DeltaSeqIdEnd },
+ State1 #vqstate { delta = Delta1,
+ q3 = Q3b }
end
end.
View Lorry Controller Status