merge bug23939 into default

11 files changed, 338 insertions, 127 deletions

diff --git a/docs/rabbitmqctl.1.xml b/docs/rabbitmqctl.1.xml
index 3550e5ea..62869158 100644
--- a/docs/rabbitmqctl.1.xml
+++ b/docs/rabbitmqctl.1.xml
@@ -1264,6 +1264,67 @@
         </varlistentry>
       </variablelist>
     </refsect2>
+
+    <refsect2>
+      <title>Configuration variables</title>
+      <para>
+        Some configuration values can be changed at run time. Note
+        that this does not apply to all variables; many are only read
+        at startup - changing them will have no effect.
+      </para>
+      <variablelist>
+        <varlistentry>
+          <term><cmdsynopsis><command>set_env</command> <arg choice="req"><replaceable>variable</replaceable></arg> <arg choice="req"><replaceable>value</replaceable></arg></cmdsynopsis></term>
+          <listitem>
+            <variablelist>
+              <varlistentry>
+                <term>variable</term>
+                <listitem><para>The name of the variable to set, as the string form of an Erlang term.</para></listitem>
+              </varlistentry>
+              <varlistentry>
+                <term>value</term>
+                <listitem><para>The value to set it to, as the string form of an Erlang term.</para></listitem>
+              </varlistentry>
+            </variablelist>
+            <para>
+              Set the value of a configuration variable.
+            </para>
+          </listitem>
+        </varlistentry>
+
+        <varlistentry>
+          <term><cmdsynopsis><command>get_env</command> <arg choice="req"><replaceable>variable</replaceable></arg></cmdsynopsis></term>
+          <listitem>
+            <variablelist>
+              <varlistentry>
+                <term>variable</term>
+                <listitem><para>The name of the variable to get, as the string form of an Erlang term.</para></listitem>
+              </varlistentry>
+            </variablelist>
+            <para>
+              Get the value of a configuration variable, printing either
+              {ok,<command>Value</command>} or undefined.
+            </para>
+          </listitem>
+        </varlistentry>
+
+        <varlistentry>
+          <term><cmdsynopsis><command>unset_env</command> <arg choice="req"><replaceable>variable</replaceable></arg></cmdsynopsis></term>
+          <listitem>
+            <variablelist>
+              <varlistentry>
+                <term>variable</term>
+                <listitem><para>The name of the variable to clear, as the string form of an Erlang term.</para></listitem>
+              </varlistentry>
+            </variablelist>
+            <para>
+              Clear the value of a configuration variable.
+            </para>
+          </listitem>
+        </varlistentry>
+
+      </variablelist>
+    </refsect2>
   </refsect1>
 
 </refentry>
diff --git a/src/rabbit_access_control.erl b/src/rabbit_access_control.erl
index b0b57af4..59c00848 100644
--- a/src/rabbit_access_control.erl
+++ b/src/rabbit_access_control.erl
@@ -18,7 +18,7 @@
 
 -include("rabbit.hrl").
 
--export([user_pass_login/2, check_user_pass_login/2, check_user_login/2,
+-export([check_user_pass_login/2, check_user_login/2,
          check_vhost_access/2, check_resource_access/3, list_vhosts/2]).
 
 %%----------------------------------------------------------------------------
@@ -30,9 +30,6 @@
 -type(permission_atom() :: 'configure' | 'read' | 'write').
 -type(vhost_permission_atom() :: 'read' | 'write').
 
--spec(user_pass_login/2 ::
-        (rabbit_types:username(), rabbit_types:password())
-        -> rabbit_types:user() | rabbit_types:channel_exit()).
 -spec(check_user_pass_login/2 ::
         (rabbit_types:username(), rabbit_types:password())
         -> {'ok', rabbit_types:user()} | {'refused', string(), [any()]}).
@@ -49,16 +46,6 @@
 
 %%----------------------------------------------------------------------------
 
-user_pass_login(User, Pass) ->
-    ?LOGDEBUG("Login with user ~p pass ~p~n", [User, Pass]),
-    case check_user_pass_login(User, Pass) of
-        {refused, Msg, Args} ->
-            rabbit_misc:protocol_error(
-              access_refused, "login refused: ~s", [io_lib:format(Msg, Args)]);
-        {ok, U} ->
-            U
-    end.
-
 check_user_pass_login(Username, Password) ->
     check_user_login(Username, [{password, Password}]).
 
diff --git a/src/rabbit_channel.erl b/src/rabbit_channel.erl
index 0c12614c..f0788862 100644
--- a/src/rabbit_channel.erl
+++ b/src/rabbit_channel.erl
@@ -281,6 +281,7 @@ handle_cast({deliver, ConsumerTag, AckRequired,
                          true  -> deliver;
                          false -> deliver_no_ack
                      end, State),
+    rabbit_trace:tap_trace_out(Msg),
     noreply(State1#ch{next_tag = DeliveryTag + 1});
 
 handle_cast(emit_stats, State = #ch{stats_timer = StatsTimer}) ->
@@ -303,7 +304,10 @@ handle_info({'DOWN', MRef, process, QPid, Reason},
               handle_publishing_queue_down(QPid, Reason, State);
           {ok, ConsumerTag} ->
               handle_consuming_queue_down(MRef, ConsumerTag, State)
-      end).
+      end);
+
+handle_info({'EXIT', _Pid, Reason}, State) ->
+    {stop, Reason, State}.
 
 handle_pre_hibernate(State = #ch{stats_timer = StatsTimer}) ->
     ok = clear_permission_cache(),
@@ -604,6 +608,7 @@ handle_method(#'basic.publish'{exchange    = ExchangeNameBin,
         end,
     case rabbit_basic:message(ExchangeName, RoutingKey, DecodedContent) of
         {ok, Message} ->
+            rabbit_trace:tap_trace_in(Message),
             {RoutingRes, DeliveredQPids} =
                 rabbit_exchange:publish(
                   Exchange,
@@ -672,6 +677,7 @@ handle_method(#'basic.get'{queue = QueueNameBin,
                                  true  -> get_no_ack;
                                  false -> get
                              end, State),
+            rabbit_trace:tap_trace_out(Msg),
             ok = rabbit_writer:send_command(
                    WriterPid,
                    #'basic.get_ok'{delivery_tag = DeliveryTag,
diff --git a/src/rabbit_control.erl b/src/rabbit_control.erl
index 1af91f4c..6ab07111 100644
--- a/src/rabbit_control.erl
+++ b/src/rabbit_control.erl
@@ -282,6 +282,19 @@ action(list_consumers, Node, _Args, Opts, Inform) ->
         Other             -> Other
     end;
 
+action(set_env, Node, [Var, Term], _Opts, Inform) ->
+    Inform("Setting control variable ~s for node ~p to ~s", [Var, Node, Term]),
+    rpc_call(Node, application, set_env, [rabbit, parse(Var), parse(Term)]);
+
+action(get_env, Node, [Var], _Opts, Inform) ->
+    Inform("Getting control variable ~s for node ~p", [Var, Node]),
+    Val = rpc_call(Node, application, get_env, [rabbit, parse(Var)]),
+    io:format("~p~n", [Val]);
+
+action(unset_env, Node, [Var], _Opts, Inform) ->
+    Inform("Clearing control variable ~s for node ~p", [Var, Node]),
+    rpc_call(Node, application, unset_env, [rabbit, parse(Var)]);
+
 action(set_permissions, Node, [Username, CPerm, WPerm, RPerm], Opts, Inform) ->
     VHost = proplists:get_value(?VHOST_OPT, Opts),
     Inform("Setting permissions for user ~p in vhost ~p", [Username, VHost]),
@@ -325,6 +338,11 @@ default_if_empty(List, Default) when is_list(List) ->
        true       -> [list_to_atom(X) || X <- List]
     end.
 
+parse(Str) ->
+    {ok, Tokens, _} = erl_scan:string(Str ++ "."),
+    {ok, Term} = erl_parse:parse_term(Tokens),
+    Term.
+
 display_info_list(Results, InfoItemKeys) when is_list(Results) ->
     lists:foreach(
       fun (Result) -> display_row(
diff --git a/src/rabbit_direct.erl b/src/rabbit_direct.erl
index 0dac18d1..7ff534ee 100644
--- a/src/rabbit_direct.erl
+++ b/src/rabbit_direct.erl
@@ -16,7 +16,7 @@
 
 -module(rabbit_direct).
 
--export([boot/0, connect/5, start_channel/8, disconnect/1]).
+-export([boot/0, connect/4, start_channel/8, disconnect/1]).
 
 -include("rabbit.hrl").
 
@@ -25,8 +25,8 @@
 -ifdef(use_specs).
 
 -spec(boot/0 :: () -> 'ok').
--spec(connect/5 :: (binary(), binary(), binary(), rabbit_types:protocol(),
-                    rabbit_event:event_props()) ->
+-spec(connect/4 :: (rabbit_types:username(), rabbit_types:vhost(),
+                    rabbit_types:protocol(), rabbit_event:event_props()) ->
                         {'ok', {rabbit_types:user(),
                                 rabbit_framing:amqp_table()}}).
 -spec(start_channel/8 ::
@@ -53,11 +53,11 @@ boot() ->
 
 %%----------------------------------------------------------------------------
 
-connect(Username, Password, VHost, Protocol, Infos) ->
+connect(Username, VHost, Protocol, Infos) ->
     case lists:keymember(rabbit, 1, application:which_applications()) of
         true  ->
-            try rabbit_access_control:user_pass_login(Username, Password) of
-                #user{} = User ->
+            case rabbit_access_control:check_user_login(Username, []) of
+                {ok, User} ->
                     try rabbit_access_control:check_vhost_access(User, VHost) of
                         ok -> rabbit_event:notify(connection_created, Infos),
                               {ok, {User,
@@ -65,9 +65,9 @@ connect(Username, Password, VHost, Protocol, Infos) ->
                     catch
                         exit:#amqp_error{name = access_refused} ->
                             {error, access_refused}
-                    end
-            catch
-                exit:#amqp_error{name = access_refused} -> {error, auth_failure}
+                    end;
+                {refused, _Msg, _Args} ->
+                    {error, auth_failure}
             end;
         false ->
             {error, broker_not_found_on_node}
diff --git a/src/rabbit_net.erl b/src/rabbit_net.erl
index c500548a..b944ec81 100644
--- a/src/rabbit_net.erl
+++ b/src/rabbit_net.erl
@@ -18,7 +18,7 @@
 -include("rabbit.hrl").
 
 -export([is_ssl/1, ssl_info/1, controlling_process/2, getstat/2,
-         async_recv/3, port_command/2, send/2, close/1,
+         recv/1, async_recv/3, port_command/2, setopts/2, send/2, close/1,
          sockname/1, peername/1, peercert/1]).
 
 %%---------------------------------------------------------------------------
@@ -28,8 +28,8 @@
 -export_type([socket/0]).
 
 -type(stat_option() ::
-	'recv_cnt' | 'recv_max' | 'recv_avg' | 'recv_oct' | 'recv_dvi' |
-	'send_cnt' | 'send_max' | 'send_avg' | 'send_oct' | 'send_pend').
+        'recv_cnt' | 'recv_max' | 'recv_avg' | 'recv_oct' | 'recv_dvi' |
+        'send_cnt' | 'send_max' | 'send_avg' | 'send_oct' | 'send_pend').
 -type(ok_val_or_error(A) :: rabbit_types:ok_or_error2(A, any())).
 -type(ok_or_any_error() :: rabbit_types:ok_or_error(any())).
 -type(socket() :: port() | #ssl_socket{}).
@@ -42,9 +42,15 @@
 -spec(getstat/2 ::
         (socket(), [stat_option()])
         -> ok_val_or_error([{stat_option(), integer()}])).
+-spec(recv/1 :: (socket()) ->
+                     {'data', [char()] | binary()} | 'closed' |
+                     rabbit_types:error(any()) | {'other', any()}).
 -spec(async_recv/3 ::
         (socket(), integer(), timeout()) -> rabbit_types:ok(any())).
 -spec(port_command/2 :: (socket(), iolist()) -> 'true').
+-spec(setopts/2 :: (socket(), [{atom(), any()} |
+                               {raw, non_neg_integer(), non_neg_integer(),
+                                binary()}]) -> ok_or_any_error()).
 -spec(send/2 :: (socket(), binary() | iolist()) -> ok_or_any_error()).
 -spec(close/1 :: (socket()) -> ok_or_any_error()).
 -spec(sockname/1 ::
@@ -80,6 +86,19 @@ getstat(Sock, Stats) when ?IS_SSL(Sock) ->
 getstat(Sock, Stats) when is_port(Sock) ->
     inet:getstat(Sock, Stats).
 
+recv(Sock) when ?IS_SSL(Sock) ->
+    recv(Sock#ssl_socket.ssl, {ssl, ssl_closed, ssl_error});
+recv(Sock) when is_port(Sock) ->
+    recv(Sock, {tcp, tcp_closed, tcp_error}).
+
+recv(S, {DataTag, ClosedTag, ErrorTag}) ->
+    receive
+        {DataTag, S, Data}    -> {data, Data};
+        {ClosedTag, S}        -> closed;
+        {ErrorTag, S, Reason} -> {error, Reason};
+        Other                 -> {other, Other}
+    end.
+
 async_recv(Sock, Length, Timeout) when ?IS_SSL(Sock) ->
     Pid = self(),
     Ref = make_ref(),
@@ -103,6 +122,11 @@ port_command(Sock, Data) when ?IS_SSL(Sock) ->
 port_command(Sock, Data) when is_port(Sock) ->
     erlang:port_command(Sock, Data).
 
+setopts(Sock, Options) when ?IS_SSL(Sock) ->
+    ssl:setopts(Sock#ssl_socket.ssl, Options);
+setopts(Sock, Options) when is_port(Sock) ->
+    inet:setopts(Sock, Options).
+
 send(Sock, Data) when ?IS_SSL(Sock) -> ssl:send(Sock#ssl_socket.ssl, Data);
 send(Sock, Data) when is_port(Sock) -> gen_tcp:send(Sock, Data).
 
diff --git a/src/rabbit_reader.erl b/src/rabbit_reader.erl
index 42af91a8..f5214a77 100644
--- a/src/rabbit_reader.erl
+++ b/src/rabbit_reader.erl
@@ -38,10 +38,10 @@
 
 %%--------------------------------------------------------------------------
 
--record(v1, {parent, sock, connection, callback, recv_length, recv_ref,
+-record(v1, {parent, sock, connection, callback, recv_len, pending_recv,
              connection_state, queue_collector, heartbeater, stats_timer,
-             channel_sup_sup_pid, start_heartbeat_fun, auth_mechanism,
-             auth_state}).
+             channel_sup_sup_pid, start_heartbeat_fun, buf, buf_len,
+             auth_mechanism, auth_state}).
 
 -define(STATISTICS_KEYS, [pid, recv_oct, recv_cnt, send_oct, send_cnt,
                           send_pend, state, channels]).
@@ -192,7 +192,7 @@ start_connection(Parent, ChannelSupSupPid, Collector, StartHeartbeatFun, Deb,
     erlang:send_after(?HANDSHAKE_TIMEOUT * 1000, self(),
                       handshake_timeout),
     try
-        mainloop(Deb, switch_callback(
+        recvloop(Deb, switch_callback(
                         #v1{parent              = Parent,
                             sock                = ClientSock,
                             connection          = #connection{
@@ -204,8 +204,8 @@ start_connection(Parent, ChannelSupSupPid, Collector, StartHeartbeatFun, Deb,
                               client_properties  = none,
                               capabilities       = []},
                             callback            = uninitialized_callback,
-                            recv_length         = 0,
-                            recv_ref            = none,
+                            recv_len            = 0,
+                            pending_recv        = false,
                             connection_state    = pre_init,
                             queue_collector     = Collector,
                             heartbeater         = none,
@@ -213,6 +213,8 @@ start_connection(Parent, ChannelSupSupPid, Collector, StartHeartbeatFun, Deb,
                                 rabbit_event:init_stats_timer(),
                             channel_sup_sup_pid = ChannelSupSupPid,
                             start_heartbeat_fun = StartHeartbeatFun,
+                            buf                 = [],
+                            buf_len             = 0,
                             auth_mechanism      = none,
                             auth_state          = none
                            },
@@ -237,92 +239,104 @@ start_connection(Parent, ChannelSupSupPid, Collector, StartHeartbeatFun, Deb,
     end,
     done.
 
-mainloop(Deb, State = #v1{parent = Parent, sock= Sock, recv_ref = Ref}) ->
-    receive
-        {inet_async, Sock, Ref, {ok, Data}} ->
-            mainloop(Deb, handle_input(State#v1.callback, Data,
-                                       State#v1{recv_ref = none}));
-        {inet_async, Sock, Ref, {error, closed}} ->
-            if State#v1.connection_state =:= closed ->
-                    State;
-               true ->
-                    throw(connection_closed_abruptly)
-            end;
-        {inet_async, Sock, Ref, {error, Reason}} ->
-            throw({inet_error, Reason});
-        {conserve_memory, Conserve} ->
-            mainloop(Deb, internal_conserve_memory(Conserve, State));
-        {channel_closing, ChPid} ->
-            ok = rabbit_channel:ready_for_close(ChPid),
-            channel_cleanup(ChPid),
-            mainloop(Deb, State);
-        {'EXIT', Parent, Reason} ->
-            terminate(io_lib:format("broker forced connection closure "
-                                    "with reason '~w'", [Reason]), State),
-            %% this is what we are expected to do according to
-            %% http://www.erlang.org/doc/man/sys.html
-            %%
-            %% If we wanted to be *really* nice we should wait for a
-            %% while for clients to close the socket at their end,
-            %% just as we do in the ordinary error case. However,
-            %% since this termination is initiated by our parent it is
-            %% probably more important to exit quickly.
-            exit(Reason);
-        {channel_exit, _Channel, E = {writer, send_failed, _Error}} ->
-            throw(E);
-        {channel_exit, Channel, Reason} ->
-            mainloop(Deb, handle_exception(State, Channel, Reason));
-        {'DOWN', _MRef, process, ChPid, Reason} ->
-            mainloop(Deb, handle_dependent_exit(ChPid, Reason, State));
-        terminate_connection ->
-            State;
-        handshake_timeout ->
-            if ?IS_RUNNING(State) orelse
-               State#v1.connection_state =:= closing orelse
-               State#v1.connection_state =:= closed ->
-                    mainloop(Deb, State);
-               true ->
-                    throw({handshake_timeout, State#v1.callback})
-            end;
-        timeout ->
-            case State#v1.connection_state of
-                closed -> mainloop(Deb, State);
-                S      -> throw({timeout, S})
-            end;
-        {'$gen_call', From, {shutdown, Explanation}} ->
-            {ForceTermination, NewState} = terminate(Explanation, State),
-            gen_server:reply(From, ok),
-            case ForceTermination of
-                force  -> ok;
-                normal -> mainloop(Deb, NewState)
-            end;
-        {'$gen_call', From, info} ->
-            gen_server:reply(From, infos(?INFO_KEYS, State)),
-            mainloop(Deb, State);
-        {'$gen_call', From, {info, Items}} ->
-            gen_server:reply(From, try {ok, infos(Items, State)}
-                                   catch Error -> {error, Error}
-                                   end),
-            mainloop(Deb, State);
-        {'$gen_cast', emit_stats} ->
-            State1 = internal_emit_stats(State),
-            mainloop(Deb, State1);
-        {system, From, Request} ->
-            sys:handle_system_msg(Request, From,
-                                  Parent, ?MODULE, Deb, State);
-        Other ->
-            %% internal error -> something worth dying for
-            exit({unexpected_message, Other})
+recvloop(Deb, State = #v1{pending_recv = true}) ->
+    mainloop(Deb, State);
+recvloop(Deb, State = #v1{connection_state = blocked}) ->
+    mainloop(Deb, State);
+recvloop(Deb, State = #v1{sock = Sock, recv_len = RecvLen, buf_len = BufLen})
+  when BufLen < RecvLen ->
+    ok = rabbit_net:setopts(Sock, [{active, once}]),
+    mainloop(Deb, State#v1{pending_recv = true});
+recvloop(Deb, State = #v1{recv_len = RecvLen, buf = Buf, buf_len = BufLen}) ->
+    {Data, Rest} = split_binary(case Buf of
+                                    [B] -> B;
+                                    _   -> list_to_binary(lists:reverse(Buf))
+                                end, RecvLen),
+    recvloop(Deb, handle_input(State#v1.callback, Data,
+                               State#v1{buf = [Rest],
+                                        buf_len = BufLen - RecvLen})).
+
+mainloop(Deb, State = #v1{sock = Sock, buf = Buf, buf_len = BufLen}) ->
+    case rabbit_net:recv(Sock) of
+        {data, Data}    -> recvloop(Deb, State#v1{buf = [Data | Buf],
+                                                  buf_len = BufLen + size(Data),
+                                                  pending_recv = false});
+        closed          -> if State#v1.connection_state =:= closed ->
+                                   State;
+                              true ->
+                                   throw(connection_closed_abruptly)
+                           end;
+        {error, Reason} -> throw({inet_error, Reason});
+        {other, Other}  -> handle_other(Other, Deb, State)
     end.
 
+handle_other({conserve_memory, Conserve}, Deb, State) ->
+    recvloop(Deb, internal_conserve_memory(Conserve, State));
+handle_other({channel_closing, ChPid}, Deb, State) ->
+    ok = rabbit_channel:ready_for_close(ChPid),
+    channel_cleanup(ChPid),
+    mainloop(Deb, State);
+handle_other({'EXIT', Parent, Reason}, _Deb, State = #v1{parent = Parent}) ->
+    terminate(io_lib:format("broker forced connection closure "
+                            "with reason '~w'", [Reason]), State),
+    %% this is what we are expected to do according to
+    %% http://www.erlang.org/doc/man/sys.html
+    %%
+    %% If we wanted to be *really* nice we should wait for a while for
+    %% clients to close the socket at their end, just as we do in the
+    %% ordinary error case. However, since this termination is
+    %% initiated by our parent it is probably more important to exit
+    %% quickly.
+    exit(Reason);
+handle_other({channel_exit, _Channel, E = {writer, send_failed, _Error}},
+             _Deb, _State) ->
+    throw(E);
+handle_other({channel_exit, Channel, Reason}, Deb, State) ->
+    mainloop(Deb, handle_exception(State, Channel, Reason));
+handle_other({'DOWN', _MRef, process, ChPid, Reason}, Deb, State) ->
+    mainloop(Deb, handle_dependent_exit(ChPid, Reason, State));
+handle_other(terminate_connection, _Deb, State) ->
+    State;
+handle_other(handshake_timeout, Deb, State)
+  when ?IS_RUNNING(State) orelse
+       State#v1.connection_state =:= closing orelse
+       State#v1.connection_state =:= closed ->
+    mainloop(Deb, State);
+handle_other(handshake_timeout, _Deb, State) ->
+    throw({handshake_timeout, State#v1.callback});
+handle_other(timeout, Deb, State = #v1{connection_state = closed}) ->
+    mainloop(Deb, State);
+handle_other(timeout, _Deb, #v1{connection_state = S}) ->
+    throw({timeout, S});
+handle_other({'$gen_call', From, {shutdown, Explanation}}, Deb, State) ->
+    {ForceTermination, NewState} = terminate(Explanation, State),
+    gen_server:reply(From, ok),
+    case ForceTermination of
+        force  -> ok;
+        normal -> mainloop(Deb, NewState)
+    end;
+handle_other({'$gen_call', From, info}, Deb, State) ->
+    gen_server:reply(From, infos(?INFO_KEYS, State)),
+    mainloop(Deb, State);
+handle_other({'$gen_call', From, {info, Items}}, Deb, State) ->
+    gen_server:reply(From, try {ok, infos(Items, State)}
+                           catch Error -> {error, Error}
+                           end),
+    mainloop(Deb, State);
+handle_other({'$gen_cast', emit_stats}, Deb, State) ->
+    mainloop(Deb, internal_emit_stats(State));
+handle_other({system, From, Request}, Deb, State = #v1{parent = Parent}) ->
+    sys:handle_system_msg(Request, From, Parent, ?MODULE, Deb, State);
+handle_other(Other, _Deb, _State) ->
+    %% internal error -> something worth dying for
+    exit({unexpected_message, Other}).
+
 switch_callback(State = #v1{connection_state = blocked,
                             heartbeater = Heartbeater}, Callback, Length) ->
     ok = rabbit_heartbeat:pause_monitor(Heartbeater),
-    State#v1{callback = Callback, recv_length = Length, recv_ref = none};
+    State#v1{callback = Callback, recv_len = Length};
 switch_callback(State, Callback, Length) ->
-    Ref = inet_op(fun () -> rabbit_net:async_recv(
-                              State#v1.sock, Length, infinity) end),
-    State#v1{callback = Callback, recv_length = Length, recv_ref = Ref}.
+    State#v1{callback = Callback, recv_len = Length}.
 
 terminate(Explanation, State) when ?IS_RUNNING(State) ->
     {normal, send_exception(State, 0,
@@ -336,12 +350,9 @@ internal_conserve_memory(true,  State = #v1{connection_state = running}) ->
 internal_conserve_memory(false, State = #v1{connection_state = blocking}) ->
     State#v1{connection_state = running};
 internal_conserve_memory(false, State = #v1{connection_state = blocked,
-                                            heartbeater      = Heartbeater,
-                                            callback         = Callback,
-                                            recv_length      = Length,
-                                            recv_ref         = none}) ->
+                                            heartbeater      = Heartbeater}) ->
     ok = rabbit_heartbeat:resume_monitor(Heartbeater),
-    switch_callback(State#v1{connection_state = running}, Callback, Length);
+    State#v1{connection_state = running};
 internal_conserve_memory(_Conserve, State) ->
     State.
 
@@ -513,8 +524,8 @@ handle_input({frame_payload, Type, Channel, PayloadSize},
              PayloadAndMarker, State) ->
     case PayloadAndMarker of
         <<Payload:PayloadSize/binary, ?FRAME_END>> ->
-            handle_frame(Type, Channel, Payload,
-                         switch_callback(State, frame_header, 7));
+            switch_callback(handle_frame(Type, Channel, Payload, State),
+                            frame_header, 7);
         _ ->
             throw({bad_payload, Type, Channel, PayloadSize, PayloadAndMarker})
     end;
diff --git a/src/rabbit_tests.erl b/src/rabbit_tests.erl
index 48119fae..e8e13212 100644
--- a/src/rabbit_tests.erl
+++ b/src/rabbit_tests.erl
@@ -1553,7 +1553,7 @@ test_logs_working(MainLogFile, SaslLogFile) ->
     ok = rabbit_log:error("foo bar"),
     ok = error_logger:error_report(crash_report, [foo, bar]),
     %% give the error loggers some time to catch up
-    timer:sleep(50),
+    timer:sleep(100),
     [true, true] = non_empty_files([MainLogFile, SaslLogFile]),
     ok.
 
@@ -2074,7 +2074,7 @@ test_queue_index() ->
 
 variable_queue_init(Q, Recover) ->
     rabbit_variable_queue:init(
-      Q, Recover, fun nop/1, fun nop/1, fun nop/2, fun nop/1).
+      Q, Recover, fun nop/2, fun nop/2, fun nop/2, fun nop/1).
 
 variable_queue_publish(IsPersistent, Count, VQ) ->
     lists:foldl(
diff --git a/src/rabbit_trace.erl b/src/rabbit_trace.erl
new file mode 100644
index 00000000..2d15e7fc
--- /dev/null
+++ b/src/rabbit_trace.erl
@@ -0,0 +1,101 @@
+%% 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 Developer of the Original Code is VMware, Inc.
+%% Copyright (c) 2007-2011 VMware, Inc.  All rights reserved.
+%%
+
+-module(rabbit_trace).
+
+-export([tap_trace_in/1, tap_trace_out/1]).
+
+-include("rabbit.hrl").
+-include("rabbit_framing.hrl").
+
+%%----------------------------------------------------------------------------
+
+-ifdef(use_specs).
+
+-spec(tap_trace_in/1 :: (rabbit_types:basic_message()) -> 'ok').
+-spec(tap_trace_out/1 :: (rabbit_amqqueue:qmsg()) -> 'ok').
+
+-endif.
+
+%%----------------------------------------------------------------------------
+
+tap_trace_in(Msg) ->
+    maybe_trace(Msg, <<"publish">>, xname(Msg), []).
+
+tap_trace_out({#resource{name = QName}, _QPid, _QMsgId, Redelivered, Msg}) ->
+    RedeliveredNum = case Redelivered of true -> 1; false -> 0 end,
+    maybe_trace(Msg, <<"deliver">>, QName,
+                [{<<"redelivered">>, signedint, RedeliveredNum}]).
+
+xname(#basic_message{exchange_name = #resource{name         = XName}}) -> XName.
+vhost(#basic_message{exchange_name = #resource{virtual_host = VHost}}) -> VHost.
+
+maybe_trace(Msg, RKPrefix, RKSuffix, Extra) ->
+    XName = xname(Msg),
+    case trace_exchange(vhost(Msg)) of
+        none   -> ok;
+        XName  -> ok;
+        TraceX -> case catch trace(TraceX, Msg, RKPrefix, RKSuffix, Extra) of
+                      {'EXIT', R} -> rabbit_log:info("Trace died: ~p~n", [R]);
+                      ok          -> ok
+                  end
+    end.
+
+trace_exchange(VHost) ->
+    case application:get_env(rabbit, trace_exchanges) of
+        undefined  -> none;
+        {ok, Xs}   -> proplists:get_value(VHost, Xs, none)
+    end.
+
+trace(TraceX, Msg0, RKPrefix, RKSuffix, Extra) ->
+    Msg = ensure_content_decoded(Msg0),
+    rabbit_basic:publish(rabbit_misc:r(vhost(Msg), exchange, TraceX),
+                         <<RKPrefix/binary, ".", RKSuffix/binary>>,
+                         #'P_basic'{headers = msg_to_table(Msg) ++ Extra},
+                         payload(Msg)),
+    ok.
+
+msg_to_table(#basic_message{exchange_name = #resource{name = XName},
+                            routing_keys  = RoutingKeys,
+                            content       = #content{properties = Props}}) ->
+    {PropsTable, _Ix} =
+        lists:foldl(
+          fun (K, {L, Ix}) ->
+                  V = element(Ix, Props),
+                  NewL = case V of
+                             undefined -> L;
+                             _         -> [{a2b(K), type(V), V} | L]
+                         end,
+                  {NewL, Ix + 1}
+          end, {[], 2}, record_info(fields, 'P_basic')),
+    [{<<"exchange_name">>, longstr, XName},
+     {<<"routing_keys">>,  array,   [{longstr, K} || K <- RoutingKeys]},
+     {<<"properties">>,    table,   PropsTable},
+     {<<"node">>,          longstr, a2b(node())}].
+
+payload(#basic_message{content = #content{payload_fragments_rev = PFR}}) ->
+    list_to_binary(lists:reverse(PFR)).
+
+ensure_content_decoded(Msg = #basic_message{content = Content}) ->
+    Msg#basic_message{content = rabbit_binary_parser:ensure_content_decoded(
+                                  Content)}.
+
+a2b(A) ->
+    list_to_binary(atom_to_list(A)).
+
+type(V) when is_list(V)    -> table;
+type(V) when is_integer(V) -> signedint;
+type(_V)                   -> longstr.
diff --git a/src/test_sup.erl b/src/test_sup.erl
index 150235da..84c4121c 100644
--- a/src/test_sup.erl
+++ b/src/test_sup.erl
@@ -33,10 +33,10 @@ test_supervisor_delayed_restart() ->
 test_supervisor_delayed_restart(SupPid) ->
     ok = ping_child(SupPid),
     ok = exit_child(SupPid),
-    timer:sleep(10),
+    timer:sleep(100),
     ok = ping_child(SupPid),
     ok = exit_child(SupPid),
-    timer:sleep(10),
+    timer:sleep(100),
     timeout = ping_child(SupPid),
     timer:sleep(1010),
     ok = ping_child(SupPid),
diff --git a/src/vm_memory_monitor.erl b/src/vm_memory_monitor.erl
index dcc6aff5..fb2fa267 100644
--- a/src/vm_memory_monitor.erl
+++ b/src/vm_memory_monitor.erl
@@ -239,10 +239,13 @@ get_total_memory({unix,darwin}) ->
     PageSize * (Inactive + Active + Free + Wired);
 
 get_total_memory({unix,freebsd}) ->
-    PageSize  = freebsd_sysctl("vm.stats.vm.v_page_size"),
-    PageCount = freebsd_sysctl("vm.stats.vm.v_page_count"),
+    PageSize  = sysctl("vm.stats.vm.v_page_size"),
+    PageCount = sysctl("vm.stats.vm.v_page_count"),
     PageCount * PageSize;
 
+get_total_memory({unix,openbsd}) ->
+    sysctl("hw.usermem");
+
 get_total_memory({win32,_OSname}) ->
     %% Due to the Erlang print format bug, on Windows boxes the memory
     %% size is broken. For example Windows 7 64 bit with 4Gigs of RAM
@@ -342,7 +345,7 @@ parse_line_aix(Line) ->
          false -> list_to_integer(Value)
      end}.
 
-freebsd_sysctl(Def) ->
+sysctl(Def) ->
     list_to_integer(cmd("/sbin/sysctl -n " ++ Def) -- "\n").
 
 %% file:read_file does not work on files in /proc as it seems to get