Merge bug23882 (unnecessary coupling and implicit contract between bq and amqqueue_process).

10 files changed, 330 insertions, 515 deletions

diff --git a/ebin/rabbit_app.in b/ebin/rabbit_app.in
index f837684c..014c18b0 100644
--- a/ebin/rabbit_app.in
+++ b/ebin/rabbit_app.in
@@ -20,6 +20,7 @@
          {vm_memory_high_watermark, 0.4},
          {msg_store_index_module, rabbit_msg_store_ets_index},
          {backing_queue_module, rabbit_variable_queue},
+         {frame_max, 131072},
          {persister_max_wrap_entries, 500},
          {persister_hibernate_after, 10000},
          {msg_store_file_size_limit, 16777216},
diff --git a/packaging/windows-exe/lib/EnvVarUpdate.nsh b/packaging/windows-exe/lib/EnvVarUpdate.nsh
deleted file mode 100644
index 839d6a02..00000000
--- a/packaging/windows-exe/lib/EnvVarUpdate.nsh
+++ /dev/null
@@ -1,327 +0,0 @@
-/**

- *  EnvVarUpdate.nsh

- *    : Environmental Variables: append, prepend, and remove entries

- *

- *     WARNING: If you use StrFunc.nsh header then include it before this file

- *              with all required definitions. This is to avoid conflicts

- *

- *  Usage:

- *    ${EnvVarUpdate} "ResultVar" "EnvVarName" "Action" "RegLoc" "PathString"

- *

- *  Credits:

- *  Version 1.0 

- *  * Cal Turney (turnec2)

- *  * Amir Szekely (KiCHiK) and e-circ for developing the forerunners of this

- *    function: AddToPath, un.RemoveFromPath, AddToEnvVar, un.RemoveFromEnvVar,

- *    WriteEnvStr, and un.DeleteEnvStr

- *  * Diego Pedroso (deguix) for StrTok

- *  * Kevin English (kenglish_hi) for StrContains

- *  * Hendri Adriaens (Smile2Me), Diego Pedroso (deguix), and Dan Fuhry  

- *    (dandaman32) for StrReplace

- *

- *  Version 1.1 (compatibility with StrFunc.nsh)

- *  * techtonik

- *

- *  http://nsis.sourceforge.net/Environmental_Variables:_append%2C_prepend%2C_and_remove_entries

- *

- */

-

-

-!ifndef ENVVARUPDATE_FUNCTION

-!define ENVVARUPDATE_FUNCTION

-!verbose push

-!verbose 3

-!include "LogicLib.nsh"

-!include "WinMessages.NSH"

-!include "StrFunc.nsh"

-

-; ---- Fix for conflict if StrFunc.nsh is already includes in main file -----------------------

-!macro _IncludeStrFunction StrFuncName

-  !ifndef ${StrFuncName}_INCLUDED

-    ${${StrFuncName}}

-  !endif

-  !ifndef Un${StrFuncName}_INCLUDED

-    ${Un${StrFuncName}}

-  !endif

-  !define un.${StrFuncName} "${Un${StrFuncName}}"

-!macroend

-

-!insertmacro _IncludeStrFunction StrTok

-!insertmacro _IncludeStrFunction StrStr

-!insertmacro _IncludeStrFunction StrRep

-

-; ---------------------------------- Macro Definitions ----------------------------------------

-!macro _EnvVarUpdateConstructor ResultVar EnvVarName Action Regloc PathString

-  Push "${EnvVarName}"

-  Push "${Action}"

-  Push "${RegLoc}"

-  Push "${PathString}"

-    Call EnvVarUpdate

-  Pop "${ResultVar}"

-!macroend

-!define EnvVarUpdate '!insertmacro "_EnvVarUpdateConstructor"'

- 

-!macro _unEnvVarUpdateConstructor ResultVar EnvVarName Action Regloc PathString

-  Push "${EnvVarName}"

-  Push "${Action}"

-  Push "${RegLoc}"

-  Push "${PathString}"

-    Call un.EnvVarUpdate

-  Pop "${ResultVar}"

-!macroend

-!define un.EnvVarUpdate '!insertmacro "_unEnvVarUpdateConstructor"'

-; ---------------------------------- Macro Definitions end-------------------------------------

- 

-;----------------------------------- EnvVarUpdate start----------------------------------------

-!define hklm_all_users     'HKLM "SYSTEM\CurrentControlSet\Control\Session Manager\Environment"'

-!define hkcu_current_user  'HKCU "Environment"'

- 

-!macro EnvVarUpdate UN

- 

-Function ${UN}EnvVarUpdate

- 

-  Push $0

-  Exch 4

-  Exch $1

-  Exch 3

-  Exch $2

-  Exch 2

-  Exch $3

-  Exch

-  Exch $4

-  Push $5

-  Push $6

-  Push $7

-  Push $8

-  Push $9

-  Push $R0

- 

-  /* After this point:

-  -------------------------

-     $0 = ResultVar     (returned)

-     $1 = EnvVarName    (input)

-     $2 = Action        (input)

-     $3 = RegLoc        (input)

-     $4 = PathString    (input)

-     $5 = Orig EnvVar   (read from registry)

-     $6 = Len of $0     (temp)

-     $7 = tempstr1      (temp)

-     $8 = Entry counter (temp)

-     $9 = tempstr2      (temp)

-     $R0 = tempChar     (temp)  */

- 

-  ; Step 1:  Read contents of EnvVarName from RegLoc

-  ;

-  ; Check for empty EnvVarName

-  ${If} $1 == ""

-    SetErrors

-    DetailPrint "ERROR: EnvVarName is blank"

-    Goto EnvVarUpdate_Restore_Vars

-  ${EndIf}

- 

-  ; Check for valid Action

-  ${If}    $2 != "A"

-  ${AndIf} $2 != "P"

-  ${AndIf} $2 != "R"

-    SetErrors

-    DetailPrint "ERROR: Invalid Action - must be A, P, or R"

-    Goto EnvVarUpdate_Restore_Vars

-  ${EndIf}

- 

-  ${If} $3 == HKLM

-    ReadRegStr $5 ${hklm_all_users} $1     ; Get EnvVarName from all users into $5

-  ${ElseIf} $3 == HKCU

-    ReadRegStr $5 ${hkcu_current_user} $1  ; Read EnvVarName from current user into $5

-  ${Else}

-    SetErrors

-    DetailPrint 'ERROR: Action is [$3] but must be "HKLM" or HKCU"'

-    Goto EnvVarUpdate_Restore_Vars

-  ${EndIf}

- 

-  ; Check for empty PathString

-  ${If} $4 == ""

-    SetErrors

-    DetailPrint "ERROR: PathString is blank"

-    Goto EnvVarUpdate_Restore_Vars

-  ${EndIf}

- 

-  ; Make sure we've got some work to do

-  ${If} $5 == ""

-  ${AndIf} $2 == "R"

-    SetErrors

-    DetailPrint "$1 is empty - Nothing to remove"

-    Goto EnvVarUpdate_Restore_Vars

-  ${EndIf}

- 

-  ; Step 2: Scrub EnvVar

-  ;

-  StrCpy $0 $5                             ; Copy the contents to $0

-  ; Remove spaces around semicolons (NOTE: spaces before the 1st entry or

-  ; after the last one are not removed here but instead in Step 3)

-  ${If} $0 != ""                           ; If EnvVar is not empty ...

-    ${Do}

-      ${${UN}StrStr} $7 $0 " ;"

-      ${If} $7 == ""

-        ${ExitDo}

-      ${EndIf}

-      ${${UN}StrRep} $0  $0 " ;" ";"         ; Remove '<space>;'

-    ${Loop}

-    ${Do}

-      ${${UN}StrStr} $7 $0 "; "

-      ${If} $7 == ""

-        ${ExitDo}

-      ${EndIf}

-      ${${UN}StrRep} $0  $0 "; " ";"         ; Remove ';<space>'

-    ${Loop}

-    ${Do}

-      ${${UN}StrStr} $7 $0 ";;" 

-      ${If} $7 == ""

-        ${ExitDo}

-      ${EndIf}

-      ${${UN}StrRep} $0  $0 ";;" ";"

-    ${Loop}

- 

-    ; Remove a leading or trailing semicolon from EnvVar

-    StrCpy  $7  $0 1 0

-    ${If} $7 == ";"

-      StrCpy $0  $0 "" 1                   ; Change ';<EnvVar>' to '<EnvVar>'

-    ${EndIf}

-    StrLen $6 $0

-    IntOp $6 $6 - 1

-    StrCpy $7  $0 1 $6

-    ${If} $7 == ";"

-     StrCpy $0  $0 $6                      ; Change ';<EnvVar>' to '<EnvVar>'

-    ${EndIf}

-    ; DetailPrint "Scrubbed $1: [$0]"      ; Uncomment to debug

-  ${EndIf}

- 

-  /* Step 3. Remove all instances of the target path/string (even if "A" or "P")

-     $6 = bool flag (1 = found and removed PathString)

-     $7 = a string (e.g. path) delimited by semicolon(s)

-     $8 = entry counter starting at 0

-     $9 = copy of $0

-     $R0 = tempChar      */

- 

-  ${If} $5 != ""                           ; If EnvVar is not empty ...

-    StrCpy $9 $0

-    StrCpy $0 ""

-    StrCpy $8 0

-    StrCpy $6 0

- 

-    ${Do}

-      ${${UN}StrTok} $7 $9 ";" $8 "0"      ; $7 = next entry, $8 = entry counter

- 

-      ${If} $7 == ""                       ; If we've run out of entries,

-        ${ExitDo}                          ;    were done

-      ${EndIf}                             ;

- 

-      ; Remove leading and trailing spaces from this entry (critical step for Action=Remove)

-      ${Do}

-        StrCpy $R0  $7 1

-        ${If} $R0 != " "

-          ${ExitDo}

-        ${EndIf}

-        StrCpy $7   $7 "" 1                ;  Remove leading space

-      ${Loop}

-      ${Do}

-        StrCpy $R0  $7 1 -1

-        ${If} $R0 != " "

-          ${ExitDo}

-        ${EndIf}

-        StrCpy $7   $7 -1                  ;  Remove trailing space

-      ${Loop}

-      ${If} $7 == $4                       ; If string matches, remove it by not appending it

-        StrCpy $6 1                        ; Set 'found' flag

-      ${ElseIf} $7 != $4                   ; If string does NOT match

-      ${AndIf}  $0 == ""                   ;    and the 1st string being added to $0,

-        StrCpy $0 $7                       ;    copy it to $0 without a prepended semicolon

-      ${ElseIf} $7 != $4                   ; If string does NOT match

-      ${AndIf}  $0 != ""                   ;    and this is NOT the 1st string to be added to $0,

-        StrCpy $0 $0;$7                    ;    append path to $0 with a prepended semicolon

-      ${EndIf}                             ;

- 

-      IntOp $8 $8 + 1                      ; Bump counter

-    ${Loop}                                ; Check for duplicates until we run out of paths

-  ${EndIf}

- 

-  ; Step 4:  Perform the requested Action

-  ;

-  ${If} $2 != "R"                          ; If Append or Prepend

-    ${If} $6 == 1                          ; And if we found the target

-      DetailPrint "Target is already present in $1. It will be removed and"

-    ${EndIf}

-    ${If} $0 == ""                         ; If EnvVar is (now) empty

-      StrCpy $0 $4                         ;   just copy PathString to EnvVar

-      ${If} $6 == 0                        ; If found flag is either 0

-      ${OrIf} $6 == ""                     ; or blank (if EnvVarName is empty)

-        DetailPrint "$1 was empty and has been updated with the target"

-      ${EndIf}

-    ${ElseIf} $2 == "A"                    ;  If Append (and EnvVar is not empty),

-      StrCpy $0 $0;$4                      ;     append PathString

-      ${If} $6 == 1

-        DetailPrint "appended to $1"

-      ${Else}

-        DetailPrint "Target was appended to $1"

-      ${EndIf}

-    ${Else}                                ;  If Prepend (and EnvVar is not empty),

-      StrCpy $0 $4;$0                      ;     prepend PathString

-      ${If} $6 == 1

-        DetailPrint "prepended to $1"

-      ${Else}

-        DetailPrint "Target was prepended to $1"

-      ${EndIf}

-    ${EndIf}

-  ${Else}                                  ; If Action = Remove

-    ${If} $6 == 1                          ;   and we found the target

-      DetailPrint "Target was found and removed from $1"

-    ${Else}

-      DetailPrint "Target was NOT found in $1 (nothing to remove)"

-    ${EndIf}

-    ${If} $0 == ""

-      DetailPrint "$1 is now empty"

-    ${EndIf}

-  ${EndIf}

- 

-  ; Step 5:  Update the registry at RegLoc with the updated EnvVar and announce the change

-  ;

-  ClearErrors

-  ${If} $3  == HKLM

-    WriteRegExpandStr ${hklm_all_users} $1 $0     ; Write it in all users section

-  ${ElseIf} $3 == HKCU

-    WriteRegExpandStr ${hkcu_current_user} $1 $0  ; Write it to current user section

-  ${EndIf}

- 

-  IfErrors 0 +4

-    MessageBox MB_OK|MB_ICONEXCLAMATION "Could not write updated $1 to $3"

-    DetailPrint "Could not write updated $1 to $3"

-    Goto EnvVarUpdate_Restore_Vars

- 

-  ; "Export" our change

-  SendMessage ${HWND_BROADCAST} ${WM_WININICHANGE} 0 "STR:Environment" /TIMEOUT=5000

- 

-  EnvVarUpdate_Restore_Vars:

-  ;

-  ; Restore the user's variables and return ResultVar

-  Pop $R0

-  Pop $9

-  Pop $8

-  Pop $7

-  Pop $6

-  Pop $5

-  Pop $4

-  Pop $3

-  Pop $2

-  Pop $1

-  Push $0  ; Push my $0 (ResultVar)

-  Exch

-  Pop $0   ; Restore his $0

- 

-FunctionEnd

- 

-!macroend   ; EnvVarUpdate UN

-!insertmacro EnvVarUpdate ""

-!insertmacro EnvVarUpdate "un."

-;----------------------------------- EnvVarUpdate end----------------------------------------

- 

-!verbose pop

-!endif

diff --git a/packaging/windows-exe/rabbitmq_nsi.in b/packaging/windows-exe/rabbitmq_nsi.in
index 6d79ffd4..1ed4064e 100644
--- a/packaging/windows-exe/rabbitmq_nsi.in
+++ b/packaging/windows-exe/rabbitmq_nsi.in
@@ -4,7 +4,6 @@
 !include WinMessages.nsh
 !include FileFunc.nsh
 !include WordFunc.nsh
-!include lib\EnvVarUpdate.nsh
 
 !define env_hklm 'HKLM "SYSTEM\CurrentControlSet\Control\Session Manager\Environment"'
 !define uninstall "Software\Microsoft\Windows\CurrentVersion\Uninstall\RabbitMQ"
@@ -77,9 +76,6 @@ Section "RabbitMQ Server (required)" Rabbit
   File /r "rabbitmq_server-%%VERSION%%"
   File "rabbitmq.ico"
 
-  ; Add to PATH
-  ${EnvVarUpdate} $0 "PATH" "A" "HKLM" "$INSTDIR\rabbitmq_server-%%VERSION%%\sbin"
-
   ; Write the installation path into the registry
   WriteRegStr HKLM "SOFTWARE\VMware, Inc.\RabbitMQ Server" "Install_Dir" "$INSTDIR"
 
@@ -126,6 +122,9 @@ Section "Start Menu" RabbitStartMenu
   CreateShortCut "$SMPROGRAMS\RabbitMQ Server\Start Service.lnk" "$INSTDIR\rabbitmq_server-%%VERSION%%\sbin\rabbitmq-service.bat" "start" "$INSTDIR\rabbitmq.ico"
   CreateShortCut "$SMPROGRAMS\RabbitMQ Server\Stop Service.lnk" "$INSTDIR\rabbitmq_server-%%VERSION%%\sbin\rabbitmq-service.bat" "stop" "$INSTDIR\rabbitmq.ico"
 
+  SetOutPath "$INSTDIR\rabbitmq_server-%%VERSION%%\sbin"
+  CreateShortCut "$SMPROGRAMS\RabbitMQ Server\Command Prompt (sbin dir).lnk" "$WINDIR\system32\cmd.exe" "" "$WINDIR\system32\cmd.exe"
+  SetOutPath $INSTDIR
 SectionEnd
 
 ;--------------------------------
@@ -157,9 +156,6 @@ Section "Uninstall"
   ExecWait '"$0" /C "$INSTDIR\rabbitmq_server-%%VERSION%%\sbin\rabbitmq-service.bat" stop'
   ExecWait '"$0" /C "$INSTDIR\rabbitmq_server-%%VERSION%%\sbin\rabbitmq-service.bat" remove'
 
-  ; Remove from PATH
-  ${un.EnvVarUpdate} $0 "PATH" "R" "HKLM" "$INSTDIR\rabbitmq_server-%%VERSION%%\sbin"
-
   ; Remove files and uninstaller
   RMDir /r "$INSTDIR\rabbitmq_server-%%VERSION%%"
   Delete "$INSTDIR\rabbitmq.ico"
diff --git a/src/file_handle_cache.erl b/src/file_handle_cache.erl
index 6f8241b3..b26bb988 100644
--- a/src/file_handle_cache.erl
+++ b/src/file_handle_cache.erl
@@ -156,13 +156,6 @@
 -define(SERVER, ?MODULE).
 -define(RESERVED_FOR_OTHERS, 100).
 
-%% Googling around suggests that Windows has a limit somewhere around
-%% 16M, eg
-%% http://blogs.technet.com/markrussinovich/archive/2009/09/29/3283844.aspx
-%% however, it turns out that's only available through the win32
-%% API. Via the C Runtime, we have just 512:
-%% http://msdn.microsoft.com/en-us/library/6e3b887c%28VS.80%29.aspx
--define(FILE_HANDLES_LIMIT_WINDOWS, 512).
 -define(FILE_HANDLES_LIMIT_OTHER, 1024).
 -define(FILE_HANDLES_CHECK_INTERVAL, 2000).
 
@@ -1185,29 +1178,20 @@ track_client(Pid, Clients) ->
         false -> ok
     end.
 
-%% For all unices, assume ulimit exists. Further googling suggests
-%% that BSDs (incl OS X), solaris and linux all agree that ulimit -n
-%% is file handles
+
+%% To increase the number of file descriptors: on Windows set ERL_MAX_PORTS
+%% environment variable, on Linux set `ulimit -n`.
 ulimit() ->
-    case os:type() of
-        {win32, _OsName} ->
-            ?FILE_HANDLES_LIMIT_WINDOWS;
-        {unix, _OsName} ->
-            %% Under Linux, Solaris and FreeBSD, ulimit is a shell
-            %% builtin, not a command. In OS X and AIX it's a command.
-            %% Fortunately, os:cmd invokes the cmd in a shell env, so
-            %% we're safe in all cases.
-            case os:cmd("ulimit -n") of
-                "unlimited" ->
-                    infinity;
-                String = [C|_] when $0 =< C andalso C =< $9 ->
-                    list_to_integer(
-                      lists:takewhile(
-                        fun (D) -> $0 =< D andalso D =< $9 end, String));
-                _ ->
-                    %% probably a variant of
-                    %% "/bin/sh: line 1: ulimit: command not found\n"
-                    unknown
+    case proplists:get_value(max_fds, erlang:system_info(check_io)) of
+        MaxFds when is_integer(MaxFds) andalso MaxFds > 1 ->
+            case os:type() of
+                {win32, _OsName} ->
+                    %% On Windows max_fds is twice the number of open files:
+                    %%   https://github.com/yrashk/erlang/blob/e1282325ed75e52a98d5/erts/emulator/sys/win32/sys.c#L2459-2466
+                    MaxFds div 2;
+                _Any ->
+                    %% For other operating systems trust Erlang.
+                    MaxFds
             end;
         _ ->
             unknown
diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl
index 4f1f50a0..7c4b5190 100644
--- a/src/rabbit_amqqueue_process.erl
+++ b/src/rabbit_amqqueue_process.erl
@@ -440,19 +440,18 @@ gb_trees_cons(Key, Value, Tree) ->
     end.
 
 record_confirm_message(#delivery{msg_seq_no = undefined}, State) ->
-    {no_confirm, State};
+    {never, State};
 record_confirm_message(#delivery{sender     = ChPid,
                                  msg_seq_no = MsgSeqNo,
                                  message    = #basic_message {
                                    is_persistent = true,
                                    id            = MsgId}},
-                       State =
-                           #q{msg_id_to_channel = MTC,
-                              q                 = #amqqueue{durable = true}}) ->
-    {confirm,
+                       State = #q{q                 = #amqqueue{durable = true},
+                                  msg_id_to_channel = MTC}) ->
+    {eventually,
      State#q{msg_id_to_channel = dict:store(MsgId, {ChPid, MsgSeqNo}, MTC)}};
 record_confirm_message(_Delivery, State) ->
-    {no_confirm, State}.
+    {immediately, State}.
 
 run_message_queue(State) ->
     Funs = {fun deliver_from_queue_pred/2,
@@ -468,11 +467,9 @@ attempt_delivery(#delivery{txn        = none,
                            message    = Message,
                            msg_seq_no = MsgSeqNo},
                  {NeedsConfirming, State = #q{backing_queue = BQ}}) ->
-    %% must confirm immediately if it has a MsgSeqNo and not NeedsConfirming
-    case {NeedsConfirming, MsgSeqNo} of
-        {_, undefined}  -> ok;
-        {no_confirm, _} -> rabbit_channel:confirm(ChPid, [MsgSeqNo]);
-        {confirm, _}    -> ok
+    case NeedsConfirming of
+        immediately -> rabbit_channel:confirm(ChPid, [MsgSeqNo]);
+        _           -> ok
     end,
     PredFun = fun (IsEmpty, _State) -> not IsEmpty end,
     DeliverFun =
@@ -484,7 +481,7 @@ attempt_delivery(#delivery{txn        = none,
                     BQ:publish_delivered(
                       AckRequired, Message,
                       (?BASE_MESSAGE_PROPERTIES)#message_properties{
-                        needs_confirming = (NeedsConfirming =:= confirm)},
+                        needs_confirming = (NeedsConfirming =:= eventually)},
                       BQS),
                 {{Message, false, AckTag}, true,
                  State1#q{backing_queue_state = BQS1}}
@@ -504,16 +501,16 @@ attempt_delivery(#delivery{txn     = Txn,
 deliver_or_enqueue(Delivery, State) ->
     case attempt_delivery(Delivery, record_confirm_message(Delivery, State)) of
         {true, _, State1} ->
-            {true, State1};
+            State1;
         {false, NeedsConfirming, State1 = #q{backing_queue = BQ,
                                              backing_queue_state = BQS}} ->
             #delivery{message = Message} = Delivery,
             BQS1 = BQ:publish(Message,
                               (message_properties(State)) #message_properties{
                                 needs_confirming =
-                                    (NeedsConfirming =:= confirm)},
+                                    (NeedsConfirming =:= eventually)},
                               BQS),
-            {false, ensure_ttl_timer(State1#q{backing_queue_state = BQS1})}
+            ensure_ttl_timer(State1#q{backing_queue_state = BQS1})
     end.
 
 requeue_and_run(AckTags, State = #q{backing_queue = BQ, ttl=TTL}) ->
@@ -839,8 +836,7 @@ handle_call({deliver_immediately, Delivery}, _From, State) ->
 handle_call({deliver, Delivery}, From, State) ->
     %% Synchronous, "mandatory" delivery mode. Reply asap.
     gen_server2:reply(From, true),
-    {_Delivered, NewState} = deliver_or_enqueue(Delivery, State),
-    noreply(NewState);
+    noreply(deliver_or_enqueue(Delivery, State));
 
 handle_call({commit, Txn, ChPid}, From, State) ->
     case lookup_ch(ChPid) of
@@ -1002,8 +998,7 @@ handle_cast(sync_timeout, State) ->
 
 handle_cast({deliver, Delivery}, State) ->
     %% Asynchronous, non-"mandatory", non-"immediate" deliver mode.
-    {_Delivered, NewState} = deliver_or_enqueue(Delivery, State),
-    noreply(NewState);
+    noreply(deliver_or_enqueue(Delivery, State));
 
 handle_cast({ack, Txn, AckTags, ChPid},
             State = #q{backing_queue = BQ, backing_queue_state = BQS}) ->
diff --git a/src/rabbit_channel.erl b/src/rabbit_channel.erl
index 526fb428..da103284 100644
--- a/src/rabbit_channel.erl
+++ b/src/rabbit_channel.erl
@@ -33,9 +33,9 @@
              start_limiter_fun, transaction_id, tx_participants, next_tag,
              uncommitted_ack_q, unacked_message_q,
              user, virtual_host, most_recently_declared_queue,
-             consumer_mapping, blocking, queue_collector_pid, stats_timer,
-             confirm_enabled, publish_seqno, unconfirmed_mq, unconfirmed_qm,
-             confirmed, capabilities}).
+             consumer_mapping, blocking, consumer_monitors, queue_collector_pid,
+             stats_timer, confirm_enabled, publish_seqno, unconfirmed_mq,
+             unconfirmed_qm, confirmed, capabilities}).
 
 -define(MAX_PERMISSION_CACHE_SIZE, 12).
 
@@ -176,6 +176,7 @@ init([Channel, ReaderPid, WriterPid, Protocol, User, VHost, Capabilities,
                 most_recently_declared_queue = <<>>,
                 consumer_mapping        = dict:new(),
                 blocking                = dict:new(),
+                consumer_monitors       = dict:new(),
                 queue_collector_pid     = CollectorPid,
                 stats_timer             = StatsTimer,
                 confirm_enabled         = false,
@@ -247,6 +248,11 @@ handle_cast(ready_for_close, State = #ch{state      = closing,
 handle_cast(terminate, State) ->
     {stop, normal, State};
 
+handle_cast({command, #'basic.consume_ok'{consumer_tag = ConsumerTag} = Msg},
+            State = #ch{writer_pid = WriterPid}) ->
+    ok = rabbit_writer:send_command(WriterPid, Msg),
+    noreply(monitor_consumer(ConsumerTag, State));
+
 handle_cast({command, Msg}, State = #ch{writer_pid = WriterPid}) ->
     ok = rabbit_writer:send_command(WriterPid, Msg),
     noreply(State);
@@ -288,23 +294,15 @@ handle_cast({confirm, MsgSeqNos, From}, State) ->
 handle_info(timeout, State) ->
     noreply(State);
 
-handle_info({'DOWN', _MRef, process, QPid, Reason},
-            State = #ch{unconfirmed_qm = UQM}) ->
-    MsgSeqNos = case gb_trees:lookup(QPid, UQM) of
-                    {value, MsgSet} -> gb_sets:to_list(MsgSet);
-                    none            -> []
-                end,
-    %% We remove the MsgSeqNos from UQM before calling
-    %% process_confirms to prevent each MsgSeqNo being removed from
-    %% the set one by one which which would be inefficient
-    State1 = State#ch{unconfirmed_qm = gb_trees:delete_any(QPid, UQM)},
-    {MXs, State2} = process_confirms(MsgSeqNos, QPid, State1),
-    erase_queue_stats(QPid),
-    State3 = (case Reason of
-                  normal -> fun record_confirms/2;
-                  _      -> fun send_nacks/2
-              end)(MXs, State2),
-    noreply(queue_blocked(QPid, State3)).
+handle_info({'DOWN', MRef, process, QPid, Reason},
+            State = #ch{consumer_monitors = ConsumerMonitors}) ->
+    noreply(
+      case dict:find(MRef, ConsumerMonitors) of
+          error ->
+              handle_publishing_queue_down(QPid, Reason, State);
+          {ok, ConsumerTag} ->
+              handle_consuming_queue_down(MRef, ConsumerTag, State)
+      end).
 
 handle_pre_hibernate(State = #ch{stats_timer = StatsTimer}) ->
     ok = clear_permission_cache(),
@@ -513,23 +511,24 @@ record_confirms(MXs, State = #ch{confirmed = C}) ->
 confirm([], _QPid, State) ->
     State;
 confirm(MsgSeqNos, QPid, State) ->
-    {MXs, State1} = process_confirms(MsgSeqNos, QPid, State),
+    {MXs, State1} = process_confirms(MsgSeqNos, QPid, false, State),
     record_confirms(MXs, State1).
 
-process_confirms(MsgSeqNos, QPid, State = #ch{unconfirmed_mq = UMQ,
-                                              unconfirmed_qm = UQM}) ->
+process_confirms(MsgSeqNos, QPid, Nack, State = #ch{unconfirmed_mq = UMQ,
+                                                    unconfirmed_qm = UQM}) ->
     {MXs, UMQ1, UQM1} =
         lists:foldl(
-          fun(MsgSeqNo, {_DMs, UMQ0, _UQM} = Acc) ->
+          fun(MsgSeqNo, {_MXs, UMQ0, _UQM} = Acc) ->
                   case gb_trees:lookup(MsgSeqNo, UMQ0) of
-                      {value, XQ} -> remove_unconfirmed(MsgSeqNo, QPid, XQ, Acc,
-                                                        State);
+                      {value, XQ} -> remove_unconfirmed(MsgSeqNo, QPid, XQ,
+                                                        Acc, Nack, State);
                       none        -> Acc
                   end
           end, {[], UMQ, UQM}, MsgSeqNos),
     {MXs, State#ch{unconfirmed_mq = UMQ1, unconfirmed_qm = UQM1}}.
 
-remove_unconfirmed(MsgSeqNo, QPid, {XName, Qs}, {MXs, UMQ, UQM}, State) ->
+remove_unconfirmed(MsgSeqNo, QPid, {XName, Qs}, {MXs, UMQ, UQM}, Nack,
+                   State) ->
     %% these confirms will be emitted even when a queue dies, but that
     %% should be fine, since the queue stats get erased immediately
     maybe_incr_stats([{{QPid, XName}, 1}], confirm, State),
@@ -544,8 +543,10 @@ remove_unconfirmed(MsgSeqNo, QPid, {XName, Qs}, {MXs, UMQ, UQM}, State) ->
                    UQM
            end,
     Qs1 = gb_sets:del_element(QPid, Qs),
-    case gb_sets:is_empty(Qs1) of
-        true ->
+    %% If QPid somehow died initiating a nack, clear the message from
+    %% internal data-structures.  Also, cleanup empty entries.
+    case (Nack orelse gb_sets:is_empty(Qs1)) of
+        true  ->
             {[{MsgSeqNo, XName} | MXs], gb_trees:delete(MsgSeqNo, UMQ), UQM1};
         false ->
             {MXs, gb_trees:update(MsgSeqNo, {XName, Qs1}, UMQ), UQM1}
@@ -688,9 +689,9 @@ handle_method(#'basic.consume'{queue        = QueueNameBin,
                                no_ack       = NoAck,
                                exclusive    = ExclusiveConsume,
                                nowait       = NoWait},
-              _, State = #ch{reader_pid       = ReaderPid,
-                             limiter_pid      = LimiterPid,
-                             consumer_mapping = ConsumerMapping }) ->
+              _, State = #ch{reader_pid        = ReaderPid,
+                             limiter_pid       = LimiterPid,
+                             consumer_mapping  = ConsumerMapping}) ->
     case dict:find(ConsumerTag, ConsumerMapping) of
         error ->
             QueueName = expand_queue_name_shortcut(QueueNameBin, State),
@@ -707,18 +708,24 @@ handle_method(#'basic.consume'{queue        = QueueNameBin,
             case rabbit_amqqueue:with_exclusive_access_or_die(
                    QueueName, ReaderPid,
                    fun (Q) ->
-                           rabbit_amqqueue:basic_consume(
-                             Q, NoAck, self(), LimiterPid,
-                             ActualConsumerTag, ExclusiveConsume,
-                             ok_msg(NoWait, #'basic.consume_ok'{
-                                      consumer_tag = ActualConsumerTag}))
+                           {rabbit_amqqueue:basic_consume(
+                              Q, NoAck, self(), LimiterPid,
+                              ActualConsumerTag, ExclusiveConsume,
+                              ok_msg(NoWait, #'basic.consume_ok'{
+                                       consumer_tag = ActualConsumerTag})),
+                            Q}
                    end) of
-                ok ->
-                    {noreply, State#ch{consumer_mapping =
-                                           dict:store(ActualConsumerTag,
-                                                      QueueName,
-                                                      ConsumerMapping)}};
-                {error, exclusive_consume_unavailable} ->
+                {ok, Q} ->
+                    State1 = State#ch{consumer_mapping =
+                                          dict:store(ActualConsumerTag,
+                                                     {Q, undefined},
+                                                     ConsumerMapping)},
+                    {noreply,
+                     case NoWait of
+                         true  -> monitor_consumer(ActualConsumerTag, State1);
+                         false -> State1
+                     end};
+                {{error, exclusive_consume_unavailable}, _Q} ->
                     rabbit_misc:protocol_error(
                       access_refused, "~s in exclusive use",
                       [rabbit_misc:rs(QueueName)])
@@ -731,26 +738,31 @@ handle_method(#'basic.consume'{queue        = QueueNameBin,
 
 handle_method(#'basic.cancel'{consumer_tag = ConsumerTag,
                               nowait = NoWait},
-              _, State = #ch{consumer_mapping = ConsumerMapping }) ->
+              _, State = #ch{consumer_mapping = ConsumerMapping,
+                             consumer_monitors = ConsumerMonitors}) ->
     OkMsg = #'basic.cancel_ok'{consumer_tag = ConsumerTag},
     case dict:find(ConsumerTag, ConsumerMapping) of
         error ->
             %% Spec requires we ignore this situation.
             return_ok(State, NoWait, OkMsg);
-        {ok, QueueName} ->
-            NewState = State#ch{consumer_mapping =
-                                    dict:erase(ConsumerTag,
-                                               ConsumerMapping)},
-            case rabbit_amqqueue:with(
-                   QueueName,
-                   fun (Q) ->
-                           %% In order to ensure that no more messages
-                           %% are sent to the consumer after the
-                           %% cancel_ok has been sent, we get the
-                           %% queue process to send the cancel_ok on
-                           %% our behalf. If we were sending the
-                           %% cancel_ok ourselves it might overtake a
-                           %% message sent previously by the queue.
+        {ok, {Q, MRef}} ->
+            ConsumerMonitors1 =
+                case MRef of
+                    undefined -> ConsumerMonitors;
+                    _         -> true = erlang:demonitor(MRef),
+                                 dict:erase(MRef, ConsumerMonitors)
+                end,
+            NewState = State#ch{consumer_mapping  = dict:erase(ConsumerTag,
+                                                               ConsumerMapping),
+                                consumer_monitors = ConsumerMonitors1},
+            %% In order to ensure that no more messages are sent to
+            %% the consumer after the cancel_ok has been sent, we get
+            %% the queue process to send the cancel_ok on our
+            %% behalf. If we were sending the cancel_ok ourselves it
+            %% might overtake a message sent previously by the queue.
+            case rabbit_misc:with_exit_handler(
+                   fun () -> {error, not_found} end,
+                   fun () ->
                            rabbit_amqqueue:basic_cancel(
                              Q, self(), ConsumerTag,
                              ok_msg(NoWait, #'basic.cancel_ok'{
@@ -1080,6 +1092,53 @@ handle_method(_MethodRecord, _Content, _State) ->
 
 %%----------------------------------------------------------------------------
 
+monitor_consumer(ConsumerTag, State = #ch{consumer_mapping = ConsumerMapping,
+                                          consumer_monitors = ConsumerMonitors,
+                                          capabilities = Capabilities}) ->
+    case rabbit_misc:table_lookup(
+           Capabilities, <<"consumer_cancel_notify">>) of
+        {bool, true} ->
+            {#amqqueue{pid = QPid} = Q, undefined} =
+                dict:fetch(ConsumerTag, ConsumerMapping),
+            MRef = erlang:monitor(process, QPid),
+            State#ch{consumer_mapping =
+                         dict:store(ConsumerTag, {Q, MRef}, ConsumerMapping),
+                     consumer_monitors =
+                         dict:store(MRef, ConsumerTag, ConsumerMonitors)};
+        _ ->
+            State
+    end.
+
+handle_publishing_queue_down(QPid, Reason, State = #ch{unconfirmed_qm = UQM}) ->
+    MsgSeqNos = case gb_trees:lookup(QPid, UQM) of
+                    {value, MsgSet} -> gb_sets:to_list(MsgSet);
+                    none            -> []
+                end,
+    %% We remove the MsgSeqNos from UQM before calling
+    %% process_confirms to prevent each MsgSeqNo being removed from
+    %% the set one by one which which would be inefficient
+    State1 = State#ch{unconfirmed_qm = gb_trees:delete_any(QPid, UQM)},
+    {Nack, SendFun} = case Reason of
+                          normal -> {false, fun record_confirms/2};
+                          _      -> {true, fun send_nacks/2}
+                      end,
+    {MXs, State2} = process_confirms(MsgSeqNos, QPid, Nack, State1),
+    erase_queue_stats(QPid),
+    State3 = SendFun(MXs, State2),
+    queue_blocked(QPid, State3).
+
+handle_consuming_queue_down(MRef, ConsumerTag,
+                            State = #ch{consumer_mapping  = ConsumerMapping,
+                                        consumer_monitors = ConsumerMonitors,
+                                        writer_pid        = WriterPid}) ->
+    ConsumerMapping1 = dict:erase(ConsumerTag, ConsumerMapping),
+    ConsumerMonitors1 = dict:erase(MRef, ConsumerMonitors),
+    Cancel = #'basic.cancel'{consumer_tag = ConsumerTag,
+                             nowait       = true},
+    ok = rabbit_writer:send_command(WriterPid, Cancel),
+    State#ch{consumer_mapping = ConsumerMapping1,
+             consumer_monitors = ConsumerMonitors1}.
+
 binding_action(Fun, ExchangeNameBin, DestinationType, DestinationNameBin,
                RoutingKey, Arguments, ReturnMethod, NoWait,
                State = #ch{virtual_host = VHostPath,
@@ -1250,16 +1309,9 @@ limit_queues(LPid, #ch{consumer_mapping = Consumers}) ->
     rabbit_amqqueue:limit_all(consumer_queues(Consumers), self(), LPid).
 
 consumer_queues(Consumers) ->
-    [QPid || QueueName <-
-                 sets:to_list(
-                   dict:fold(fun (_ConsumerTag, QueueName, S) ->
-                                     sets:add_element(QueueName, S)
-                             end, sets:new(), Consumers)),
-             case rabbit_amqqueue:lookup(QueueName) of
-                 {ok, Q} -> QPid = Q#amqqueue.pid, true;
-                 %% queue has been deleted in the meantime
-                 {error, not_found} -> QPid = none, false
-             end].
+    lists:usort([QPid ||
+                    {_Key, {#amqqueue{pid = QPid}, _MRef}}
+                        <- dict:to_list(Consumers)]).
 
 %% tell the limiter about the number of acks that have been received
 %% for messages delivered to subscribed consumers, but not acks for
diff --git a/src/rabbit_exchange_type_topic.erl b/src/rabbit_exchange_type_topic.erl
index f12661d4..ffd1e583 100644
--- a/src/rabbit_exchange_type_topic.erl
+++ b/src/rabbit_exchange_type_topic.erl
@@ -67,17 +67,58 @@ add_binding(true, _Exchange, Binding) ->
 add_binding(false, _Exchange, _Binding) ->
     ok.
 
-remove_bindings(true, _X, Bs) ->
-    lists:foreach(fun remove_binding/1, Bs),
+remove_bindings(true, #exchange{name = X}, Bs) ->
+    %% The remove process is split into two distinct phases. In the
+    %% first phase we gather the lists of bindings and edges to
+    %% delete, then in the second phase we process all the
+    %% deletions. This is to prevent interleaving of read/write
+    %% operations in mnesia that can adversely affect performance.
+    {ToDelete, Paths} =
+       lists:foldl(
+         fun(#binding{source = S, key = K, destination = D}, {Acc, PathAcc}) ->
+                 Path = [{FinalNode, _} | _] =
+                     follow_down_get_path(S, split_topic_key(K)),
+                 {[{FinalNode, D} | Acc],
+                  decrement_bindings(X, Path, maybe_add_path(X, Path, PathAcc))}
+         end, {[], gb_trees:empty()}, Bs),
+
+    [trie_remove_binding(X, FinalNode, D) || {FinalNode, D} <- ToDelete],
+    [trie_remove_edge(X, Parent, Node, W) ||
+        {Node, {Parent, W, {0, 0}}} <- gb_trees:to_list(Paths)],
     ok;
 remove_bindings(false, _X, _Bs) ->
     ok.
 
-remove_binding(#binding{source = X, key = K, destination = D}) ->
-    Path = [{FinalNode, _} | _] = follow_down_get_path(X, split_topic_key(K)),
-    trie_remove_binding(X, FinalNode, D),
-    remove_path_if_empty(X, Path),
-    ok.
+maybe_add_path(_X, [{root, none}], PathAcc) ->
+    PathAcc;
+maybe_add_path(X, [{Node, W}, {Parent, _} | _], PathAcc) ->
+    case gb_trees:is_defined(Node, PathAcc) of
+        true  -> PathAcc;
+        false -> gb_trees:insert(Node, {Parent, W, {trie_binding_count(X, Node),
+                                                    trie_child_count(X, Node)}},
+                                 PathAcc)
+    end.
+
+decrement_bindings(X, Path, PathAcc) ->
+    with_path_acc(X, fun({Bindings, Edges}) -> {Bindings - 1, Edges} end,
+                  Path, PathAcc).
+
+decrement_edges(X, Path, PathAcc) ->
+    with_path_acc(X, fun({Bindings, Edges}) -> {Bindings, Edges - 1} end,
+                  Path, PathAcc).
+
+with_path_acc(_X, _Fun, [{root, none}], PathAcc) ->
+    PathAcc;
+with_path_acc(X, Fun, [{Node, _} | ParentPath], PathAcc) ->
+    {Parent, W, Counts} = gb_trees:get(Node, PathAcc),
+    NewCounts = Fun(Counts),
+    NewPathAcc = gb_trees:update(Node, {Parent, W, NewCounts}, PathAcc),
+    case NewCounts of
+        {0, 0} -> decrement_edges(X, ParentPath,
+                                  maybe_add_path(X, ParentPath, NewPathAcc));
+        _      -> NewPathAcc
+    end.
+
 
 assert_args_equivalence(X, Args) ->
     rabbit_exchange:assert_args_equivalence(X, Args).
@@ -146,15 +187,6 @@ follow_down(X, CurNode, AccFun, Acc, Words = [W | RestW]) ->
         error          -> {error, Acc, Words}
     end.
 
-remove_path_if_empty(_, [{root, none}]) ->
-    ok;
-remove_path_if_empty(X, [{Node, W} | [{Parent, _} | _] = RestPath]) ->
-    case trie_has_any_bindings(X, Node) orelse trie_has_any_children(X, Node) of
-        true  -> ok;
-        false -> trie_remove_edge(X, Parent, Node, W),
-                 remove_path_if_empty(X, RestPath)
-    end.
-
 trie_child(X, Node, Word) ->
     case mnesia:read(rabbit_topic_trie_edge,
                      #trie_edge{exchange_name = X,
@@ -199,21 +231,24 @@ trie_binding_op(X, Node, D, Op) ->
                                            destination   = D}},
             write).
 
-trie_has_any_children(X, Node) ->
-    has_any(rabbit_topic_trie_edge,
+trie_child_count(X, Node) ->
+    count(rabbit_topic_trie_edge,
             #topic_trie_edge{trie_edge = #trie_edge{exchange_name = X,
                                                     node_id       = Node,
                                                     _             = '_'},
                              _         = '_'}).
 
-trie_has_any_bindings(X, Node) ->
-    has_any(rabbit_topic_trie_binding,
+trie_binding_count(X, Node) ->
+    count(rabbit_topic_trie_binding,
             #topic_trie_binding{
               trie_binding = #trie_binding{exchange_name = X,
                                            node_id       = Node,
                                            _             = '_'},
               _            = '_'}).
 
+count(Table, Match) ->
+    length(mnesia:match_object(Table, Match, read)).
+
 trie_remove_all_edges(X) ->
     remove_all(rabbit_topic_trie_edge,
                #topic_trie_edge{trie_edge = #trie_edge{exchange_name = X,
@@ -226,15 +261,6 @@ trie_remove_all_bindings(X) ->
                  trie_binding = #trie_binding{exchange_name = X, _ = '_'},
                  _            = '_'}).
 
-has_any(Table, MatchHead) ->
-    Select = mnesia:select(Table, [{MatchHead, [], ['$_']}], 1, read),
-    select_while_no_result(Select) /= '$end_of_table'.
-
-select_while_no_result({[], Cont}) ->
-    select_while_no_result(mnesia:select(Cont));
-select_while_no_result(Other) ->
-    Other.
-
 remove_all(Table, Pattern) ->
     lists:foreach(fun (R) -> mnesia:delete_object(Table, R, write) end,
                   mnesia:match_object(Table, Pattern, write)).
diff --git a/src/rabbit_networking.erl b/src/rabbit_networking.erl
index 877d2cf7..53be0190 100644
--- a/src/rabbit_networking.erl
+++ b/src/rabbit_networking.erl
@@ -24,7 +24,8 @@
          close_connection/2]).
 
 %%used by TCP-based transports, e.g. STOMP adapter
--export([check_tcp_listener_address/2]).
+-export([check_tcp_listener_address/2,
+         ensure_ssl/0, ssl_transform_fun/1]).
 
 -export([tcp_listener_started/3, tcp_listener_stopped/3,
          start_client/1, start_ssl_client/2]).
@@ -67,7 +68,7 @@
 -spec(close_connection/2 :: (pid(), string()) -> 'ok').
 -spec(on_node_down/1 :: (node()) -> 'ok').
 -spec(check_tcp_listener_address/2 :: (atom(), listener_config())
-                                      -> [{inet:ip_address(), ip_port(), family(), atom()}]).
+        -> [{inet:ip_address(), ip_port(), family(), atom()}]).
 
 -endif.
 
@@ -88,19 +89,8 @@ boot_ssl() ->
         {ok, []} ->
             ok;
         {ok, SslListeners} ->
-            ok = rabbit_misc:start_applications([crypto, public_key, ssl]),
-            {ok, SslOptsConfig} = application:get_env(ssl_options),
-            %% unknown_ca errors are silently ignored  prior to R14B unless we
-            %% supply this verify_fun - remove when at least R14B is required
-            SslOpts =
-                case proplists:get_value(verify, SslOptsConfig, verify_none) of
-                    verify_none -> SslOptsConfig;
-                    verify_peer -> [{verify_fun, fun([])    -> true;
-                                                    ([_|_]) -> false
-                                                 end}
-                                    | SslOptsConfig]
-                end,
-            [start_ssl_listener(Listener, SslOpts) || Listener <- SslListeners],
+            [start_ssl_listener(Listener, ensure_ssl())
+             || Listener <- SslListeners],
             ok
     end.
 
@@ -147,6 +137,34 @@ resolve_family({_,_,_,_,_,_,_,_}, auto) -> inet6;
 resolve_family(IP,                auto) -> throw({error, {strange_family, IP}});
 resolve_family(_,                 F)    -> F.
 
+ensure_ssl() ->
+    ok = rabbit_misc:start_applications([crypto, public_key, ssl]),
+    {ok, SslOptsConfig} = application:get_env(rabbit, ssl_options),
+
+    % unknown_ca errors are silently ignored prior to R14B unless we
+    % supply this verify_fun - remove when at least R14B is required
+    case proplists:get_value(verify, SslOptsConfig, verify_none) of
+        verify_none -> SslOptsConfig;
+        verify_peer -> [{verify_fun, fun([])    -> true;
+                                        ([_|_]) -> false
+                                     end}
+                        | SslOptsConfig]
+    end.
+
+ssl_transform_fun(SslOpts) ->
+    fun (Sock) ->
+            case catch ssl:ssl_accept(Sock, SslOpts, ?SSL_TIMEOUT * 1000) of
+                {ok, SslSock} ->
+                    rabbit_log:info("upgraded TCP connection ~p to SSL~n",
+                                    [self()]),
+                    {ok, #ssl_socket{tcp = Sock, ssl = SslSock}};
+                {error, Reason} ->
+                    {error, {ssl_upgrade_error, Reason}};
+                {'EXIT', Reason} ->
+                    {error, {ssl_upgrade_failure, Reason}}
+            end
+    end.
+
 check_tcp_listener_address(NamePrefix, Port) when is_integer(Port) ->
     check_tcp_listener_address_auto(NamePrefix, Port);
 
@@ -246,21 +264,7 @@ start_client(Sock) ->
     start_client(Sock, fun (S) -> {ok, S} end).
 
 start_ssl_client(SslOpts, Sock) ->
-    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).
+    start_client(Sock, ssl_transform_fun(SslOpts)).
 
 connections() ->
     [rabbit_connection_sup:reader(ConnSup) ||
diff --git a/src/rabbit_reader.erl b/src/rabbit_reader.erl
index 710e6878..5afe5560 100644
--- a/src/rabbit_reader.erl
+++ b/src/rabbit_reader.erl
@@ -35,7 +35,6 @@
 -define(CLOSING_TIMEOUT, 1).
 -define(CHANNEL_TERMINATION_TIMEOUT, 3).
 -define(SILENT_CLOSE_DELAY, 3).
--define(FRAME_MAX, 131072). %% set to zero once QPid fix their negotiation
 
 %%--------------------------------------------------------------------------
 
@@ -165,7 +164,8 @@ server_properties(Protocol) ->
 server_capabilities(rabbit_framing_amqp_0_9_1) ->
     [{<<"publisher_confirms">>,         bool, true},
      {<<"exchange_exchange_bindings">>, bool, true},
-     {<<"basic.nack">>,                 bool, true}];
+     {<<"basic.nack">>,                 bool, true},
+     {<<"consumer_cancel_notify">>,     bool, true}];
 server_capabilities(_) ->
     [].
 
@@ -641,14 +641,15 @@ handle_method0(#'connection.tune_ok'{frame_max = FrameMax,
                            connection = Connection,
                            sock = Sock,
                            start_heartbeat_fun = SHF}) ->
-    if (FrameMax /= 0) and (FrameMax < ?FRAME_MIN_SIZE) ->
+    ServerFrameMax = server_frame_max(),
+    if FrameMax /= 0 andalso FrameMax < ?FRAME_MIN_SIZE ->
             rabbit_misc:protocol_error(
               not_allowed, "frame_max=~w < ~w min size",
               [FrameMax, ?FRAME_MIN_SIZE]);
-       (?FRAME_MAX /= 0) and (FrameMax > ?FRAME_MAX) ->
+       ServerFrameMax /= 0 andalso FrameMax > ServerFrameMax ->
             rabbit_misc:protocol_error(
               not_allowed, "frame_max=~w > ~w max size",
-              [FrameMax, ?FRAME_MAX]);
+              [FrameMax, ServerFrameMax]);
        true ->
             Frame = rabbit_binary_generator:build_heartbeat_frame(),
             SendFun = fun() -> catch rabbit_net:send(Sock, Frame) end,
@@ -706,6 +707,12 @@ handle_method0(_Method, #v1{connection_state = S}) ->
     rabbit_misc:protocol_error(
       channel_error, "unexpected method in connection state ~w", [S]).
 
+%% Compute frame_max for this instance. Could simply use 0, but breaks
+%% QPid Java client.
+server_frame_max() ->
+    {ok, FrameMax} = application:get_env(rabbit, frame_max),
+    FrameMax.
+
 send_on_channel0(Sock, Method, Protocol) ->
     ok = rabbit_writer:internal_send_command(Sock, 0, Method, Protocol).
 
@@ -756,7 +763,7 @@ auth_phase(Response,
             State#v1{auth_state = AuthState1};
         {ok, User} ->
             Tune = #'connection.tune'{channel_max = 0,
-                                      frame_max = ?FRAME_MAX,
+                                      frame_max = server_frame_max(),
                                       heartbeat = 0},
             ok = send_on_channel0(Sock, Tune, Protocol),
             State#v1{connection_state = tuning,
diff --git a/src/rabbit_tests.erl b/src/rabbit_tests.erl
index 05112e8e..930923e8 100644
--- a/src/rabbit_tests.erl
+++ b/src/rabbit_tests.erl
@@ -57,6 +57,7 @@ all_tests() ->
     passed = test_cluster_management(),
     passed = test_user_management(),
     passed = test_server_status(),
+    passed = test_confirms(),
     passed = maybe_run_cluster_dependent_tests(),
     passed = test_configurable_server_properties(),
     passed.
@@ -1225,6 +1226,82 @@ test_statistics_receive_event1(Ch, Matcher) ->
     after 1000 -> throw(failed_to_receive_event)
     end.
 
+test_confirms_receiver(Pid) ->
+    receive
+        shutdown ->
+            ok;
+        {send_command, Method} ->
+            Pid ! Method,
+            test_confirms_receiver(Pid)
+    end.
+
+test_confirms() ->
+    {_Writer, Ch} = test_spawn(fun test_confirms_receiver/1),
+    DeclareBindDurableQueue =
+        fun() ->
+                rabbit_channel:do(Ch, #'queue.declare'{durable = true}),
+                receive #'queue.declare_ok'{queue = Q0} ->
+                        rabbit_channel:do(Ch, #'queue.bind'{
+                                            queue = Q0,
+                                            exchange = <<"amq.direct">>,
+                                            routing_key = "magic" }),
+                        receive #'queue.bind_ok'{} ->
+                                Q0
+                        after 1000 ->
+                                throw(failed_to_bind_queue)
+                        end
+                after 1000 ->
+                        throw(failed_to_declare_queue)
+                end
+        end,
+    %% Declare and bind two queues
+    QName1 = DeclareBindDurableQueue(),
+    QName2 = DeclareBindDurableQueue(),
+    %% Get the first one's pid (we'll crash it later)
+    {ok, Q1} = rabbit_amqqueue:lookup(rabbit_misc:r(<<"/">>, queue, QName1)),
+    QPid1 = Q1#amqqueue.pid,
+    %% Enable confirms
+    rabbit_channel:do(Ch, #'confirm.select'{}),
+    receive #'confirm.select_ok'{} ->
+            ok
+    after 1000 ->
+            throw(failed_to_enable_confirms)
+    end,
+    %% Publish a message
+    rabbit_channel:do(Ch, #'basic.publish'{exchange = <<"amq.direct">>,
+                                           routing_key = "magic"
+                                          },
+                      rabbit_basic:build_content(
+                        #'P_basic'{delivery_mode = 2}, <<"">>)),
+    %% Crash the queue
+    QPid1 ! boom,
+    %% Wait for a nack
+    receive
+        #'basic.nack'{} ->
+            ok;
+        #'basic.ack'{} ->
+            throw(received_ack_instead_of_nack)
+    after 2000 ->
+            throw(did_not_receive_nack)
+    end,
+    receive
+        #'basic.ack'{} ->
+            throw(received_ack_when_none_expected)
+    after 1000 ->
+            ok
+    end,
+    %% Cleanup
+    rabbit_channel:do(Ch, #'queue.delete'{queue = QName2}),
+    receive #'queue.delete_ok'{} ->
+            ok
+    after 1000 ->
+            throw(failed_to_cleanup_queue)
+    end,
+    unlink(Ch),
+    ok = rabbit_channel:shutdown(Ch),
+
+    passed.
+
 test_statistics() ->
     application:set_env(rabbit, collect_statistics, fine),