summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorMatthias Radestock <matthias@lshift.net>2009-12-17 11:16:21 +0000
committerMatthias Radestock <matthias@lshift.net>2009-12-17 11:16:21 +0000
commitbb173a12eb269eeca191369b8ac71dffce8292a1 (patch)
tree3d9b7b242bcc9468f30f0dd5cfbbcb94ccb06741
parent5482a87d0ae83fb39df473931b15c63c62677b60 (diff)
parent937626a88075d2936cb1cc49e84841a409b6cc5a (diff)
downloadrabbitmq-server-bb173a12eb269eeca191369b8ac71dffce8292a1.tar.gz
merge bug22013 into default
-rw-r--r--Makefile13
-rw-r--r--codegen.py25
-rw-r--r--docs/rabbitmqctl.1.pod11
-rw-r--r--include/rabbit.hrl2
-rw-r--r--src/rabbit.erl1
-rw-r--r--src/rabbit_control.erl46
-rw-r--r--src/rabbit_exchange.erl18
-rw-r--r--src/rabbit_exchange_type.erl107
-rw-r--r--src/rabbit_networking.erl48
-rw-r--r--src/rabbit_plugin_activator.erl24
-rw-r--r--src/rabbit_reader.erl50
-rw-r--r--src/vm_memory_monitor.erl83
12 files changed, 282 insertions, 146 deletions
diff --git a/Makefile b/Makefile
index a9845607..1cd48df2 100644
--- a/Makefile
+++ b/Makefile
@@ -15,7 +15,20 @@ TARGETS=$(EBIN_DIR)/rabbit.app $(BEAM_TARGETS)
WEB_URL=http://stage.rabbitmq.com/
MANPAGES=$(patsubst %.pod, %.gz, $(wildcard docs/*.[0-9].pod))
+ifeq ($(shell python -c 'import simplejson' 2>/dev/null && echo yes),yes)
PYTHON=python
+else
+ifeq ($(shell python2.6 -c 'import simplejson' 2>/dev/null && echo yes),yes)
+PYTHON=python2.6
+else
+ifeq ($(shell python2.5 -c 'import simplejson' 2>/dev/null && echo yes),yes)
+PYTHON=python2.5
+else
+# Hmm. Missing simplejson?
+PYTHON=python
+endif
+endif
+endif
BASIC_PLT=basic.plt
RABBIT_PLT=rabbit.plt
diff --git a/codegen.py b/codegen.py
index 1c5b2b54..20bfc947 100644
--- a/codegen.py
+++ b/codegen.py
@@ -244,19 +244,10 @@ def genErl(spec):
print 'lookup_amqp_exception(%s) -> {%s, ?%s, <<"%s">>};' % \
(n.lower(), hardErrorBoolStr, n, n)
- def genIsAmqpHardErrorCode(c,v,cls):
- mCls = messageConstantClass(cls)
- if mCls == 'SOFT_ERROR' : genIsAmqpHardErrorCode1(c,'false')
- elif mCls == 'HARD_ERROR' : genIsAmqpHardErrorCode1(c,'true')
- elif mCls == '' : pass
- else: raise 'Unkown constant class', cls
-
- def genIsAmqpHardErrorCode1(c,hardErrorBoolStr):
+ def genAmqpException(c,v,cls):
n = erlangConstantName(c)
- print 'is_amqp_hard_error_code(?%s) -> %s;' % \
- (n, hardErrorBoolStr)
- print 'is_amqp_hard_error_code(%s) -> %s;' % \
- (n.lower(), hardErrorBoolStr)
+ print 'amqp_exception(?%s) -> %s;' % \
+ (n, n.lower())
methods = spec.allMethods()
@@ -274,7 +265,7 @@ def genErl(spec):
-export([encode_method_fields/1]).
-export([encode_properties/1]).
-export([lookup_amqp_exception/1]).
--export([is_amqp_hard_error_code/1]).
+-export([amqp_exception/1]).
bitvalue(true) -> 1;
bitvalue(false) -> 0;
@@ -313,12 +304,8 @@ bitvalue(undefined) -> 0.
print " rabbit_log:warning(\"Unknown AMQP error code '~p'~n\", [Code]),"
print " {true, ?INTERNAL_ERROR, <<\"INTERNAL_ERROR\">>}."
- for(c,v,cls) in spec.constants: genIsAmqpHardErrorCode(c,v,cls)
- print "is_amqp_hard_error_code(Code) when is_integer(Code) ->"
- print " true;"
- print "is_amqp_hard_error_code(Code) ->"
- print " rabbit_log:warning(\"Unknown AMQP error code '~p'~n\", [Code]),"
- print " true."
+ for(c,v,cls) in spec.constants: genAmqpException(c,v,cls)
+ print "amqp_exception(_Code) -> undefined."
def genHrl(spec):
def erlType(domain):
diff --git a/docs/rabbitmqctl.1.pod b/docs/rabbitmqctl.1.pod
index 6b420872..5255be28 100644
--- a/docs/rabbitmqctl.1.pod
+++ b/docs/rabbitmqctl.1.pod
@@ -198,9 +198,9 @@ whether the queue will be deleted when no longer used
queue arguments
-=item node
+=item pid
-node on which the process associated with the queue resides
+id of the Erlang process associated with the queue
=item messages_ready
@@ -297,7 +297,7 @@ I<user>, I<peer_address>, I<peer_port> and I<state> are assumed.
=item node
-node on which the process associated with the connection resides
+id of the Erlang process associated with the connection
=item address
@@ -340,6 +340,11 @@ connection timeout
maximum frame size (bytes)
+=item client_properties
+
+informational properties transmitted by the client during connection
+establishment
+
=item recv_oct
octets received
diff --git a/include/rabbit.hrl b/include/rabbit.hrl
index 5703d0d6..4b157cbc 100644
--- a/include/rabbit.hrl
+++ b/include/rabbit.hrl
@@ -36,7 +36,7 @@
-record(vhost, {virtual_host, dummy}).
--record(connection, {user, timeout_sec, frame_max, vhost}).
+-record(connection, {user, timeout_sec, frame_max, vhost, client_properties}).
-record(content,
{class_id,
diff --git a/src/rabbit.erl b/src/rabbit.erl
index c6dde385..3293927a 100644
--- a/src/rabbit.erl
+++ b/src/rabbit.erl
@@ -134,6 +134,7 @@ start(normal, []) ->
fun () -> ok = rabbit_mnesia:init() end},
{"core processes",
fun () ->
+ ok = start_child(rabbit_exchange_type),
ok = start_child(rabbit_log),
ok = rabbit_hooks:start(),
diff --git a/src/rabbit_control.erl b/src/rabbit_control.erl
index 19579729..ddd0c002 100644
--- a/src/rabbit_control.erl
+++ b/src/rabbit_control.erl
@@ -173,7 +173,7 @@ The list_queues, list_exchanges and list_bindings commands accept an optional
virtual host parameter for which to display results. The default value is \"/\".
<QueueInfoItem> must be a member of the list [name, durable, auto_delete,
-arguments, node, messages_ready, messages_unacknowledged, messages_uncommitted,
+arguments, pid, messages_ready, messages_unacknowledged, messages_uncommitted,
messages, acks_uncommitted, consumers, transactions, memory]. The default is
to display name and (number of) messages.
@@ -183,10 +183,10 @@ auto_delete, arguments]. The default is to display name and type.
The output format for \"list_bindings\" is a list of rows containing
exchange name, queue name, routing key and arguments, in that order.
-<ConnectionInfoItem> must be a member of the list [node, address, port,
+<ConnectionInfoItem> must be a member of the list [pid, address, port,
peer_address, peer_port, state, channels, user, vhost, timeout, frame_max,
-recv_oct, recv_cnt, send_oct, send_cnt, send_pend]. The default is to display
-user, peer_address, peer_port and state.
+client_properties, recv_oct, recv_cnt, send_oct, send_cnt, send_pend].
+The default is to display user, peer_address, peer_port and state.
"),
halt(1).
@@ -268,8 +268,7 @@ action(list_user_permissions, Node, Args = [_Username], Inform) ->
action(list_queues, Node, Args, Inform) ->
Inform("Listing queues", []),
{VHostArg, RemainingArgs} = parse_vhost_flag_bin(Args),
- ArgAtoms = list_replace(node, pid,
- default_if_empty(RemainingArgs, [name, messages])),
+ ArgAtoms = default_if_empty(RemainingArgs, [name, messages]),
display_info_list(rpc_call(Node, rabbit_amqqueue, info_all,
[VHostArg, ArgAtoms]),
ArgAtoms);
@@ -294,9 +293,7 @@ action(list_bindings, Node, Args, Inform) ->
action(list_connections, Node, Args, Inform) ->
Inform("Listing connections", []),
- ArgAtoms = list_replace(node, pid,
- default_if_empty(Args, [user, peer_address,
- peer_port, state])),
+ ArgAtoms = default_if_empty(Args, [user, peer_address, peer_port, state]),
display_info_list(rpc_call(Node, rabbit_networking, connection_info_all,
[ArgAtoms]),
ArgAtoms);
@@ -358,12 +355,15 @@ format_info_item(Key, Items) ->
is_tuple(Value) ->
inet_parse:ntoa(Value);
Value when is_pid(Value) ->
- atom_to_list(node(Value));
+ pid_to_string(Value);
Value when is_binary(Value) ->
escape(Value);
Value when is_atom(Value) ->
- escape(atom_to_list(Value));
- Value ->
+ escape(atom_to_list(Value));
+ Value = [{TableEntryKey, TableEntryType, _TableEntryValue} | _]
+ when is_binary(TableEntryKey) andalso is_atom(TableEntryType) ->
+ io_lib:format("~1000000000000p", [prettify_amqp_table(Value)]);
+ Value ->
io_lib:format("~w", [Value])
end.
@@ -388,14 +388,14 @@ rpc_call(Node, Mod, Fun, Args) ->
%% characters. We don't escape characters above 127, since they may
%% form part of UTF-8 strings.
-escape(Bin) when binary(Bin) ->
+escape(Bin) when is_binary(Bin) ->
escape(binary_to_list(Bin));
escape(L) when is_list(L) ->
escape_char(lists:reverse(L), []).
escape_char([$\\ | T], Acc) ->
escape_char(T, [$\\, $\\ | Acc]);
-escape_char([X | T], Acc) when X > 32, X /= 127 ->
+escape_char([X | T], Acc) when X >= 32, X /= 127 ->
escape_char(T, [X | Acc]);
escape_char([X | T], Acc) ->
escape_char(T, [$\\, $0 + (X bsr 6), $0 + (X band 8#070 bsr 3),
@@ -403,6 +403,20 @@ escape_char([X | T], Acc) ->
escape_char([], Acc) ->
Acc.
-list_replace(Find, Replace, List) ->
- [case X of Find -> Replace; _ -> X end || X <- List].
+prettify_amqp_table(Table) ->
+ [{escape(K), prettify_typed_amqp_value(T, V)} || {K, T, V} <- Table].
+prettify_typed_amqp_value(Type, Value) ->
+ case Type of
+ longstr -> escape(Value);
+ table -> prettify_amqp_table(Value);
+ array -> [prettify_typed_amqp_value(T, V) || {T, V} <- Value];
+ _ -> Value
+ end.
+
+%% see http://erlang.org/doc/apps/erts/erl_ext_dist.html (8.10 and 8.7)
+pid_to_string(Pid) ->
+ <<131,103,100,NodeLen:16,NodeBin:NodeLen/binary,Id:32,Ser:32,_Cre:8>>
+ = term_to_binary(Pid),
+ Node = binary_to_term(<<131,100,NodeLen:16,NodeBin:NodeLen/binary>>),
+ lists:flatten(io_lib:format("<~w.~B.~B>", [Node, Id, Ser])).
diff --git a/src/rabbit_exchange.erl b/src/rabbit_exchange.erl
index 09ea1e96..be73e818 100644
--- a/src/rabbit_exchange.erl
+++ b/src/rabbit_exchange.erl
@@ -134,18 +134,18 @@ declare(ExchangeName, Type, Durable, AutoDelete, Args) ->
end
end).
-typename_to_plugin_module(T) when is_binary(T) ->
- case catch list_to_existing_atom("rabbit_exchange_type_" ++ binary_to_list(T)) of
- {'EXIT', {badarg, _}} ->
+typename_to_plugin_module(T) ->
+ case rabbit_exchange_type:lookup_module(T) of
+ {ok, Module} ->
+ Module;
+ {error, not_found} ->
rabbit_misc:protocol_error(
- command_invalid, "invalid exchange type '~s'", [T]);
- Module ->
- Module
+ command_invalid, "invalid exchange type '~s'", [T])
end.
-plugin_module_to_typename(M) when is_atom(M) ->
- "rabbit_exchange_type_" ++ S = atom_to_list(M),
- list_to_binary(S).
+plugin_module_to_typename(M) ->
+ {ok, TypeName} = rabbit_exchange_type:lookup_name(M),
+ TypeName.
check_type(T) ->
Module = typename_to_plugin_module(T),
diff --git a/src/rabbit_exchange_type.erl b/src/rabbit_exchange_type.erl
new file mode 100644
index 00000000..58dcfbb6
--- /dev/null
+++ b/src/rabbit_exchange_type.erl
@@ -0,0 +1,107 @@
+%% 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-2009 LShift
+%% Ltd. Portions created by Cohesive Financial Technologies LLC are
+%% Copyright (C) 2007-2009 Cohesive Financial Technologies
+%% LLC. Portions created by Rabbit Technologies Ltd are Copyright
+%% (C) 2007-2009 Rabbit Technologies Ltd.
+%%
+%% All Rights Reserved.
+%%
+%% Contributor(s): ______________________________________.
+%%
+
+-module(rabbit_exchange_type).
+
+-behaviour(gen_server).
+
+-export([start_link/0]).
+
+-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
+ terminate/2, code_change/3]).
+
+-export([register/2, lookup_module/1, lookup_name/1]).
+
+-define(SERVER, ?MODULE).
+
+%%---------------------------------------------------------------------------
+
+start_link() ->
+ gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).
+
+%%---------------------------------------------------------------------------
+
+register(TypeName, ModuleName) ->
+ gen_server:call(?SERVER, {register, TypeName, ModuleName}).
+
+lookup_module(T) when is_binary(T) ->
+ case ets:lookup(rabbit_exchange_type_modules, T) of
+ [{_, Module}] ->
+ {ok, Module};
+ [] ->
+ {error, not_found}
+ end.
+
+lookup_name(M) when is_atom(M) ->
+ [{_, TypeName}] = ets:lookup(rabbit_exchange_type_names, M),
+ {ok, TypeName}.
+
+%%---------------------------------------------------------------------------
+
+internal_register(TypeName, ModuleName)
+ when is_binary(TypeName), is_atom(ModuleName) ->
+ true = ets:insert(rabbit_exchange_type_modules, {TypeName, ModuleName}),
+ true = ets:insert(rabbit_exchange_type_names, {ModuleName, TypeName}),
+ ok.
+
+%%---------------------------------------------------------------------------
+
+init([]) ->
+ rabbit_exchange_type_modules =
+ ets:new(rabbit_exchange_type_modules, [protected, set, named_table]),
+ rabbit_exchange_type_names =
+ ets:new(rabbit_exchange_type_names, [protected, set, named_table]),
+
+ %% TODO: split out into separate boot startup steps.
+ ok = internal_register(<<"direct">>, rabbit_exchange_type_direct),
+ ok = internal_register(<<"fanout">>, rabbit_exchange_type_fanout),
+ ok = internal_register(<<"headers">>, rabbit_exchange_type_headers),
+ ok = internal_register(<<"topic">>, rabbit_exchange_type_topic),
+
+ {ok, none}.
+
+handle_call({register, TypeName, ModuleName}, _From, State) ->
+ ok = internal_register(TypeName, ModuleName),
+ {reply, ok, State};
+handle_call(Request, _From, State) ->
+ {stop, {unhandled_call, Request}, State}.
+
+handle_cast(Request, State) ->
+ {stop, {unhandled_cast, Request}, State}.
+
+handle_info(Message, State) ->
+ {stop, {unhandled_info, Message}, State}.
+
+terminate(_Reason, _State) ->
+ ok.
+
+code_change(_OldVsn, State, _Extra) ->
+ {ok, State}.
diff --git a/src/rabbit_networking.erl b/src/rabbit_networking.erl
index 1bc17a32..3a0f9240 100644
--- a/src/rabbit_networking.erl
+++ b/src/rabbit_networking.erl
@@ -53,6 +53,9 @@
%% {delay_send, true},
{exit_on_close, false}
]).
+
+-define(SSL_TIMEOUT, 5). %% seconds
+
%%----------------------------------------------------------------------------
-ifdef(use_specs).
@@ -160,36 +163,31 @@ node_listeners(Node) ->
on_node_down(Node) ->
ok = mnesia:dirty_delete(rabbit_listener, Node).
-start_client(Sock) ->
+start_client(Sock, SockTransform) ->
{ok, Child} = supervisor:start_child(rabbit_tcp_client_sup, []),
ok = rabbit_net:controlling_process(Sock, Child),
- Child ! {go, Sock},
+ Child ! {go, Sock, SockTransform},
Child.
+start_client(Sock) ->
+ start_client(Sock, fun (S) -> {ok, S} end).
+
start_ssl_client(SslOpts, Sock) ->
- case rabbit_net:peername(Sock) of
- {ok, {PeerAddress, PeerPort}} ->
- PeerIp = inet_parse:ntoa(PeerAddress),
- case ssl:ssl_accept(Sock, SslOpts) of
- {ok, SslSock} ->
- rabbit_log:info("upgraded TCP connection "
- "from ~s:~p to SSL~n",
- [PeerIp, PeerPort]),
- RabbitSslSock = #ssl_socket{tcp = Sock, ssl = SslSock},
- start_client(RabbitSslSock);
- {error, Reason} ->
- gen_tcp:close(Sock),
- rabbit_log:error("failed to upgrade TCP connection "
- "from ~s:~p to SSL: ~n~p~n",
- [PeerIp, PeerPort, Reason]),
- {error, Reason}
- end;
- {error, Reason} ->
- gen_tcp:close(Sock),
- rabbit_log:error("failed to upgrade TCP connection to SSL: ~p~n",
- [Reason]),
- {error, Reason}
- end.
+ start_client(
+ Sock,
+ fun (Sock1) ->
+ case catch ssl:ssl_accept(Sock1, SslOpts, ?SSL_TIMEOUT * 1000) of
+ {ok, SslSock} ->
+ rabbit_log:info("upgraded TCP connection ~p to SSL~n",
+ [self()]),
+ {ok, #ssl_socket{tcp = Sock1, ssl = SslSock}};
+ {error, Reason} ->
+ {error, {ssl_upgrade_error, Reason}};
+ {'EXIT', Reason} ->
+ {error, {ssl_upgrade_failure, Reason}}
+
+ end
+ end).
connections() ->
[Pid || {_, Pid, _, _} <- supervisor:which_children(
diff --git a/src/rabbit_plugin_activator.erl b/src/rabbit_plugin_activator.erl
index a2ac74ef..9f787920 100644
--- a/src/rabbit_plugin_activator.erl
+++ b/src/rabbit_plugin_activator.erl
@@ -44,7 +44,7 @@
-ifdef(use_specs).
--spec(start/0 :: () -> no_return()).
+-spec(start/0 :: () -> no_return()).
-spec(stop/0 :: () -> 'ok').
-endif.
@@ -73,7 +73,7 @@ start() ->
%% Build the entire set of dependencies - this will load the
%% applications along the way
AllApps = case catch sets:to_list(expand_dependencies(RequiredApps)) of
- {failed_to_load_app, App, Err} ->
+ {failed_to_load_app, App, Err} ->
error("failed to load application ~s:~n~p", [App, Err]);
AppList ->
AppList
@@ -82,8 +82,8 @@ start() ->
{rabbit, RabbitVersion} = proplists:lookup(rabbit, AppVersions),
%% Build the overall release descriptor
- RDesc = {release,
- {"rabbit", RabbitVersion},
+ RDesc = {release,
+ {"rabbit", RabbitVersion},
{erts, erlang:system_info(version)},
AppVersions},
@@ -93,15 +93,15 @@ start() ->
%% Compile the script
ScriptFile = RootName ++ ".script",
case systools:make_script(RootName, [local, silent]) of
- {ok, Module, Warnings} ->
+ {ok, Module, Warnings} ->
%% This gets lots of spurious no-source warnings when we
%% have .ez files, so we want to supress them to prevent
%% hiding real issues.
WarningStr = Module:format_warning(
- [W || W <- Warnings,
- case W of
- {warning, {source_not_found, _}} -> false;
- _ -> true
+ [W || W <- Warnings,
+ case W of
+ {warning, {source_not_found, _}} -> false;
+ _ -> true
end]),
case length(WarningStr) of
0 -> ok;
@@ -136,8 +136,8 @@ get_env(Key, Default) ->
end.
determine_version(App) ->
- application:load(App),
- {ok, Vsn} = application:get_key(App, vsn),
+ application:load(App),
+ {ok, Vsn} = application:get_key(App, vsn),
{App, Vsn}.
assert_dir(Dir) ->
@@ -236,7 +236,7 @@ post_process_script(ScriptFile) ->
{error, {failed_to_load_script, Reason}}
end.
-process_entries([]) ->
+process_entries([]) ->
[];
process_entries([Entry = {apply,{application,start_boot,[stdlib,permanent]}} |
Rest]) ->
diff --git a/src/rabbit_reader.erl b/src/rabbit_reader.erl
index e21485b5..e78d889d 100644
--- a/src/rabbit_reader.erl
+++ b/src/rabbit_reader.erl
@@ -58,7 +58,7 @@
-define(INFO_KEYS,
[pid, address, port, peer_address, peer_port,
recv_oct, recv_cnt, send_oct, send_cnt, send_pend,
- state, channels, user, vhost, timeout, frame_max]).
+ state, channels, user, vhost, timeout, frame_max, client_properties]).
%% connection lifecycle
%%
@@ -142,7 +142,8 @@ start_link() ->
init(Parent) ->
Deb = sys:debug_options([]),
receive
- {go, Sock} -> start_connection(Parent, Deb, Sock)
+ {go, Sock, SockTransform} ->
+ start_connection(Parent, Deb, Sock, SockTransform)
end.
system_continue(Parent, Deb, State) ->
@@ -192,34 +193,35 @@ teardown_profiling(Value) ->
inet_op(F) -> rabbit_misc:throw_on_error(inet_error, F).
-peername(Sock) ->
- try
- {Address, Port} = inet_op(fun () -> rabbit_net:peername(Sock) end),
- AddressS = inet_parse:ntoa(Address),
- {AddressS, Port}
- catch
- Ex -> rabbit_log:error("error on TCP connection ~p:~p~n",
- [self(), Ex]),
- rabbit_log:info("closing TCP connection ~p", [self()]),
- exit(normal)
+socket_op(Sock, Fun) ->
+ case Fun(Sock) of
+ {ok, Res} -> Res;
+ {error, Reason} -> rabbit_log:error("error on TCP connection ~p:~p~n",
+ [self(), Reason]),
+ rabbit_log:info("closing TCP connection ~p~n",
+ [self()]),
+ exit(normal)
end.
-start_connection(Parent, Deb, ClientSock) ->
+start_connection(Parent, Deb, Sock, SockTransform) ->
process_flag(trap_exit, true),
- {PeerAddressS, PeerPort} = peername(ClientSock),
+ {PeerAddress, PeerPort} = socket_op(Sock, fun rabbit_net:peername/1),
+ PeerAddressS = inet_parse:ntoa(PeerAddress),
+ rabbit_log:info("starting TCP connection ~p from ~s:~p~n",
+ [self(), PeerAddressS, PeerPort]),
+ ClientSock = socket_op(Sock, SockTransform),
+ erlang:send_after(?HANDSHAKE_TIMEOUT * 1000, self(),
+ handshake_timeout),
ProfilingValue = setup_profiling(),
try
- rabbit_log:info("starting TCP connection ~p from ~s:~p~n",
- [self(), PeerAddressS, PeerPort]),
- erlang:send_after(?HANDSHAKE_TIMEOUT * 1000, self(),
- handshake_timeout),
mainloop(Parent, Deb, switch_callback(
#v1{sock = ClientSock,
connection = #connection{
user = none,
timeout_sec = ?HANDSHAKE_TIMEOUT,
frame_max = ?FRAME_MIN_SIZE,
- vhost = none},
+ vhost = none,
+ client_properties = none},
callback = uninitialized_callback,
recv_ref = none,
connection_state = pre_init},
@@ -558,7 +560,8 @@ handle_method0(MethodName, FieldsBin, State) ->
end.
handle_method0(#'connection.start_ok'{mechanism = Mechanism,
- response = Response},
+ response = Response,
+ client_properties = ClientProperties},
State = #v1{connection_state = starting,
connection = Connection,
sock = Sock}) ->
@@ -570,7 +573,9 @@ handle_method0(#'connection.start_ok'{mechanism = Mechanism,
frame_max = 131072,
heartbeat = 0}),
State#v1{connection_state = tuning,
- connection = Connection#connection{user = User}};
+ connection = Connection#connection{
+ user = User,
+ client_properties = ClientProperties}};
handle_method0(#'connection.tune_ok'{channel_max = _ChannelMax,
frame_max = FrameMax,
heartbeat = ClientHeartbeat},
@@ -689,6 +694,9 @@ i(timeout, #v1{connection = #connection{timeout_sec = Timeout}}) ->
Timeout;
i(frame_max, #v1{connection = #connection{frame_max = FrameMax}}) ->
FrameMax;
+i(client_properties, #v1{connection = #connection{
+ client_properties = ClientProperties}}) ->
+ ClientProperties;
i(Item, #v1{}) ->
throw({bad_argument, Item}).
diff --git a/src/vm_memory_monitor.erl b/src/vm_memory_monitor.erl
index 6da47933..8be28f52 100644
--- a/src/vm_memory_monitor.erl
+++ b/src/vm_memory_monitor.erl
@@ -34,7 +34,7 @@
%% has a single process that's consuming all memory. In such a case,
%% during garbage collection, Erlang tries to allocate a huge chunk of
%% continuous memory, which can result in a crash or heavy swapping.
-%%
+%%
%% This module tries to warn Rabbit before such situations occur, so
%% that it has a higher chance to avoid running out of memory.
%%
@@ -42,7 +42,7 @@
-module(vm_memory_monitor).
--behaviour(gen_server2).
+-behaviour(gen_server).
-export([start_link/1]).
@@ -73,9 +73,10 @@
-ifdef(use_specs).
--spec(start_link/1 :: (float()) -> ('ignore' | {error, any()} | {'ok', pid()})).
+-spec(start_link/1 :: (float()) ->
+ ('ignore' | {'error', any()} | {'ok', pid()})).
-spec(update/0 :: () -> 'ok').
--spec(get_total_memory/0 :: () -> (non_neg_integer() | unknown)).
+-spec(get_total_memory/0 :: () -> (non_neg_integer() | 'unknown')).
-spec(get_check_interval/0 :: () -> non_neg_integer()).
-spec(set_check_interval/1 :: (non_neg_integer()) -> 'ok').
-spec(get_vm_memory_high_watermark/0 :: () -> float()).
@@ -85,25 +86,48 @@
%%----------------------------------------------------------------------------
+%% Public API
+%%----------------------------------------------------------------------------
+
+update() ->
+ gen_server:cast(?SERVER, update).
+
+get_total_memory() ->
+ get_total_memory(os:type()).
+
+get_check_interval() ->
+ gen_server:call(?MODULE, get_check_interval).
+
+set_check_interval(Fraction) ->
+ gen_server:call(?MODULE, {set_check_interval, Fraction}).
+
+get_vm_memory_high_watermark() ->
+ gen_server:call(?MODULE, get_vm_memory_high_watermark).
+
+set_vm_memory_high_watermark(Fraction) ->
+ gen_server:call(?MODULE, {set_vm_memory_high_watermark, Fraction}).
+
+%%----------------------------------------------------------------------------
%% gen_server callbacks
%%----------------------------------------------------------------------------
start_link(Args) ->
- gen_server2:start_link({local, ?SERVER}, ?MODULE, [Args], []).
+ gen_server:start_link({local, ?SERVER}, ?MODULE, [Args], []).
-init([MemFraction]) ->
+init([MemFraction]) ->
TotalMemory =
case get_total_memory() of
unknown ->
- rabbit_log:warning(
+ error_logger:warning_msg(
"Unknown total memory size for your OS ~p. "
- "Assuming memory size is ~pMB.~n",
+ "Assuming memory size is ~pMB.~n",
[os:type(), trunc(?MEMORY_SIZE_FOR_UNKNOWN_OS/1048576)]),
?MEMORY_SIZE_FOR_UNKNOWN_OS;
M -> M
end,
MemLimit = get_mem_limit(MemFraction, TotalMemory),
- rabbit_log:info("Memory limit set to ~pMB.~n", [trunc(MemLimit/1048576)]),
+ error_logger:info_msg("Memory limit set to ~pMB.~n",
+ [trunc(MemLimit/1048576)]),
TRef = start_timer(?DEFAULT_MEMORY_CHECK_INTERVAL),
State = #state { total_memory = TotalMemory,
memory_limit = MemLimit,
@@ -117,8 +141,8 @@ handle_call(get_vm_memory_high_watermark, _From, State) ->
handle_call({set_vm_memory_high_watermark, MemFraction}, _From, State) ->
MemLimit = get_mem_limit(MemFraction, State#state.total_memory),
- rabbit_log:info("Memory alarm changed to ~p, ~p bytes.~n",
- [MemFraction, MemLimit]),
+ error_logger:info_msg("Memory alarm changed to ~p, ~p bytes.~n",
+ [MemFraction, MemLimit]),
{reply, ok, State#state{memory_limit = MemLimit}};
handle_call(get_check_interval, _From, State) ->
@@ -134,41 +158,19 @@ handle_call(_Request, _From, State) ->
handle_cast(update, State) ->
{noreply, internal_update(State)};
-handle_cast(_Request, State) ->
+handle_cast(_Request, State) ->
{noreply, State}.
-handle_info(_Info, State) ->
+handle_info(_Info, State) ->
{noreply, State}.
-terminate(_Reason, _State) ->
+terminate(_Reason, _State) ->
ok.
-code_change(_OldVsn, State, _Extra) ->
+code_change(_OldVsn, State, _Extra) ->
{ok, State}.
%%----------------------------------------------------------------------------
-%% Public API
-%%----------------------------------------------------------------------------
-
-update() ->
- gen_server2:cast(?SERVER, update).
-
-get_total_memory() ->
- get_total_memory(os:type()).
-
-get_check_interval() ->
- gen_server2:call(?MODULE, get_check_interval).
-
-set_check_interval(Fraction) ->
- gen_server2:call(?MODULE, {set_check_interval, Fraction}).
-
-get_vm_memory_high_watermark() ->
- gen_server2:call(?MODULE, get_vm_memory_high_watermark).
-
-set_vm_memory_high_watermark(Fraction) ->
- gen_server2:call(?MODULE, {set_vm_memory_high_watermark, Fraction}).
-
-%%----------------------------------------------------------------------------
%% Server Internals
%%----------------------------------------------------------------------------
@@ -189,8 +191,9 @@ internal_update(State = #state { memory_limit = MemLimit,
State #state {alarmed = NewAlarmed}.
emit_update_info(State, MemUsed, MemLimit) ->
- rabbit_log:info("vm_memory_high_watermark ~p. Memory used:~p allowed:~p~n",
- [State, MemUsed, MemLimit]).
+ error_logger:info_msg(
+ "vm_memory_high_watermark ~p. Memory used:~p allowed:~p~n",
+ [State, MemUsed, MemLimit]).
start_timer(Timeout) ->
{ok, TRef} = timer:apply_interval(Timeout, ?MODULE, update, []),
@@ -202,7 +205,7 @@ get_vm_limit() ->
case erlang:system_info(wordsize) of
4 -> 4294967296; %% 4 GB for 32 bits 2^32
8 -> 281474976710656 %% 256 TB for 64 bits 2^48
- %% http://en.wikipedia.org/wiki/X86-64#Virtual_address_space_details
+ %%http://en.wikipedia.org/wiki/X86-64#Virtual_address_space_details
end.
get_mem_limit(MemFraction, TotalMemory) ->