merge default into bug21429bug21429

22 files changed, 490 insertions, 308 deletions

diff --git a/Makefile b/Makefile
index 1dcf7362..f0702756 100644
--- a/Makefile
+++ b/Makefile
@@ -148,7 +148,7 @@ srcdist: distclean
 	rm -rf $(TARGET_SRC_DIR)
 
 distclean: clean
-	make -C $(AMQP_CODEGEN_DIR) distclean
+	$(MAKE) -C $(AMQP_CODEGEN_DIR) distclean
 	rm -rf dist
 	find . -regex '.*\(~\|#\|\.swp\|\.dump\)' -exec rm {} \;
 
diff --git a/calculate-relative b/calculate-relative
index 3c3e2b1f..3af18e8f 100755
--- a/calculate-relative
+++ b/calculate-relative
@@ -1,4 +1,4 @@
-#!/usr/bin/python
+#!/usr/bin/env python
 #
 # relpath.py
 # R.Barran 30/08/2004
diff --git a/packaging/RPMS/Fedora/Makefile b/packaging/RPMS/Fedora/Makefile
index 89b73841..fa2844fd 100644
--- a/packaging/RPMS/Fedora/Makefile
+++ b/packaging/RPMS/Fedora/Makefile
@@ -29,8 +29,11 @@ prepare:
 	sed -i 's|%%VERSION%%|$(VERSION)|;s|%%REQUIRES%%|$(REQUIRES)|' \
 		SPECS/rabbitmq-server.spec
 
-	cp init.d SOURCES/rabbitmq-server.init
 	cp ${COMMON_DIR}/* SOURCES/
+	sed -i \
+	    -e 's|^DEFAULTS_FILE=.*$$|DEFAULTS_FILE=/etc/sysconfig/rabbitmq|' \
+	    -e 's|^LOCK_FILE=.*$$|LOCK_FILE=/var/lock/subsys/$$NAME|' \
+	    SOURCES/rabbitmq-server.init
 	cp rabbitmq-server.logrotate SOURCES/rabbitmq-server.logrotate
 
 server: prepare
diff --git a/packaging/RPMS/Fedora/init.d b/packaging/common/rabbitmq-server.init
index 21019c70..e71562f8 100644
--- a/packaging/RPMS/Fedora/init.d
+++ b/packaging/common/rabbitmq-server.init
@@ -8,10 +8,10 @@
 
 ### BEGIN INIT INFO
 # Provides:          rabbitmq-server
-# Default-Start:
-# Default-Stop:
 # Required-Start:    $remote_fs $network
 # Required-Stop:     $remote_fs $network
+# Default-Start:
+# Default-Stop:
 # Description:       RabbitMQ broker
 # Short-Description: Enable AMQP service provided by RabbitMQ broker
 ### END INIT INFO
@@ -24,13 +24,14 @@ USER=rabbitmq
 NODE_COUNT=1
 ROTATE_SUFFIX=
 
-LOCK_FILE=/var/lock/subsys/$NAME
+DEFAULTS_FILE= # This is filled in when building packages
+LOCK_FILE= # This is filled in when building packages
 
 test -x $DAEMON || exit 0
 
 # Include rabbitmq defaults if available
-if [ -f /etc/sysconfig/rabbitmq ] ; then
-	. /etc/sysconfig/rabbitmq
+if [ -f "$DEFAULTS_FILE" ] ; then
+	. $DEFAULTS_FILE
 fi
 
 RETVAL=0
@@ -41,7 +42,8 @@ start_rabbitmq () {
     $DAEMON start_all ${NODE_COUNT} > /var/log/rabbitmq/startup_log 2> /var/log/rabbitmq/startup_err
     case "$?" in
       0)
-        echo SUCCESS && touch $LOCK_FILE
+        echo SUCCESS
+        [ -n "$LOCK_FILE" ] && touch $LOCK_FILE
         RETVAL=0
         ;;
       1)
@@ -52,7 +54,7 @@ start_rabbitmq () {
         echo FAILED - check /var/log/rabbitmq/startup_log, _err
         RETVAL=1
         ;;
-    esac 
+    esac
     set -e
 }
 
@@ -65,7 +67,7 @@ stop_rabbitmq () {
         if [ $RETVAL = 0 ] ; then
             # Try to stop epmd if run by the rabbitmq user
             pkill -u rabbitmq epmd || :
-            rm -rf $LOCK_FILE
+            [ -n "$LOCK_FILE" ] && rm -rf $LOCK_FILE
         else
             echo FAILED - check /var/log/rabbitmq/shutdown_log, _err
         fi
@@ -121,19 +123,14 @@ case "$1" in
         echo -n "Rotating log files for $DESC: "
         rotate_logs_rabbitmq
         ;;
-    force-reload|reload|restart)
-        echo -n "Restarting $DESC: "
-        restart_rabbitmq
-        echo "$NAME."
-        ;;
-    condrestart|try-restart)
+    force-reload|reload|restart|condrestart|try-restart)
         echo -n "Restarting $DESC: "
         restart_rabbitmq
         echo "$NAME."
         ;;
     *)
         echo "Usage: $0 {start|stop|status|rotate-logs|restart|condrestart|try-restart|reload|force-reload}" >&2
-        RETVAL=2
+        RETVAL=1
         ;;
 esac
 
diff --git a/packaging/debs/Debian/Makefile b/packaging/debs/Debian/Makefile
index 7ab8b659..dafaf9ce 100644
--- a/packaging/debs/Debian/Makefile
+++ b/packaging/debs/Debian/Makefile
@@ -22,6 +22,10 @@ package: clean
 	tar -zxvf $(DEBIAN_ORIG_TARBALL)
 	cp -r debian $(UNPACKED_DIR)
 	cp $(COMMON_DIR)/* $(UNPACKED_DIR)/debian/
+	sed -i \
+	    -e 's|^DEFAULTS_FILE=.*$$|DEFAULTS_FILE=/etc/default/rabbitmq|' \
+	    -e 's|^LOCK_FILE=.*$$|LOCK_FILE=|' \
+	    $(UNPACKED_DIR)/debian/rabbitmq-server.init
 	chmod a+x $(UNPACKED_DIR)/debian/rules
 	UNOFFICIAL_RELEASE=$(UNOFFICIAL_RELEASE) VERSION=$(VERSION) ./check-changelog.sh rabbitmq-server $(UNPACKED_DIR)
 	cd $(UNPACKED_DIR); GNUPGHOME=$(GNUPG_PATH)/.gnupg dpkg-buildpackage -rfakeroot $(SIGNING)
diff --git a/packaging/debs/Debian/debian/init.d b/packaging/debs/Debian/debian/init.d
deleted file mode 100644
index 4a7909c5..00000000
--- a/packaging/debs/Debian/debian/init.d
+++ /dev/null
@@ -1,125 +0,0 @@
-#!/bin/sh
-### BEGIN INIT INFO
-# Provides:          rabbitmq
-# Required-Start:    $remote_fs $network
-# Required-Stop:     $remote_fs $network
-# Default-Start:     2 3 4 5
-# Default-Stop:      0 1 6
-# Description:       RabbitMQ broker
-# Short-Description: Enable AMQP service provided by RabbitMQ broker
-### END INIT INFO
-
-PATH=/sbin:/usr/sbin:/bin:/usr/bin
-DAEMON=/usr/sbin/rabbitmq-multi
-NAME=rabbitmq-server
-DESC=rabbitmq-server
-USER=rabbitmq
-NODE_COUNT=1
-ROTATE_SUFFIX=
-
-test -x $DAEMON || exit 0
-
-# Include rabbitmq defaults if available
-if [ -f /etc/default/rabbitmq ] ; then
-	. /etc/default/rabbitmq
-fi
-
-RETVAL=0
-set -e
-
-start_rabbitmq () {
-    set +e
-    $DAEMON start_all ${NODE_COUNT} > /var/log/rabbitmq/startup_log 2> /var/log/rabbitmq/startup_err
-    case "$?" in
-      0)
-        echo SUCCESS
-        RETVAL=0
-        ;;
-      1)
-        echo TIMEOUT - check /var/log/rabbitmq/startup_\{log,err\}
-        RETVAL=1
-        ;;
-      *)
-        echo FAILED - check /var/log/rabbitmq/startup_log, _err
-        RETVAL=1
-        ;;
-    esac
-    set -e
-}
-
-stop_rabbitmq () {
-    set +e
-    status_rabbitmq quiet
-    if [ $RETVAL = 0 ] ; then
-        $DAEMON stop_all > /var/log/rabbitmq/shutdown_log 2> /var/log/rabbitmq/shutdown_err
-        RETVAL=$?
-        if [ $RETVAL = 0 ] ; then
-            # Try to stop epmd if run by the rabbitmq user
-            pkill -u rabbitmq epmd || :
-        else
-            echo FAILED - check /var/log/rabbitmq/shutdown_log, _err
-        fi
-    else
-        echo No nodes running 
-        RETVAL=0
-    fi
-    set -e
-}
-
-status_rabbitmq() {
-    set +e
-    if [ "$1" != "quiet" ] ; then
-        $DAEMON status 2>&1
-    else
-        $DAEMON status > /dev/null 2>&1
-    fi
-    if [ $? != 0 ] ; then
-        RETVAL=1
-    fi
-    set -e
-}
-
-rotate_logs_rabbitmq() {
-    set +e
-    $DAEMON rotate_logs ${ROTATE_SUFFIX}
-    if [ $? != 0 ] ; then
-        RETVAL=1
-    fi
-    set -e
-}
-
-restart_rabbitmq() {
-    stop_rabbitmq	    
-    start_rabbitmq
-}
-
-case "$1" in
-    start)
-        echo -n "Starting $DESC: "
-        start_rabbitmq
-        echo "$NAME."
-        ;;
-    stop)
-        echo -n "Stopping $DESC: "
-        stop_rabbitmq
-        echo "$NAME."
-        ;;
-    status)
-        status_rabbitmq
-        ;;
-    rotate-logs)
-        echo -n "Rotating log files for $DESC: "
-        rotate_logs_rabbitmq
-        ;;
-    force-reload|restart)
-        echo -n "Restarting $DESC: "
-        restart_rabbitmq
-        echo "$NAME."
-        ;;
-    *)
-        echo "Usage: $0 {start|stop|status|rotate-logs|restart|force-reload}" >&2
-        RETVAL=1
-        ;;
-esac
-
-exit $RETVAL
diff --git a/packaging/generic-unix/Makefile b/packaging/generic-unix/Makefile
index b3988696..4eade6c7 100644
--- a/packaging/generic-unix/Makefile
+++ b/packaging/generic-unix/Makefile
@@ -4,10 +4,10 @@ TARGET_DIR=rabbitmq_server-$(VERSION)
 TARGET_TARBALL=rabbitmq-server-generic-unix-$(VERSION)
 
 dist:
-	make -C ../.. VERSION=$(VERSION) srcdist
+	$(MAKE) -C ../.. VERSION=$(VERSION) srcdist
 	tar -zxvf ../../dist/$(SOURCE_DIR).tar.gz
 
-	make -C $(SOURCE_DIR) \
+	$(MAKE) -C $(SOURCE_DIR) \
 		TARGET_DIR=`pwd`/$(TARGET_DIR) \
 		SBIN_DIR=`pwd`/$(TARGET_DIR)/sbin \
 		MAN_DIR=`pwd`/$(TARGET_DIR)/share/man \
diff --git a/packaging/windows/Makefile b/packaging/windows/Makefile
index 28f79319..387becb3 100644
--- a/packaging/windows/Makefile
+++ b/packaging/windows/Makefile
@@ -4,9 +4,9 @@ TARGET_DIR=rabbitmq_server-$(VERSION)
 TARGET_ZIP=rabbitmq-server-windows-$(VERSION)
 
 dist:
-	make -C ../.. VERSION=$(VERSION) srcdist
+	$(MAKE) -C ../.. VERSION=$(VERSION) srcdist
 	tar -zxvf ../../dist/$(SOURCE_DIR).tar.gz
-	make -C $(SOURCE_DIR)
+	$(MAKE) -C $(SOURCE_DIR)
 
 	mkdir $(SOURCE_DIR)/sbin
 	mv $(SOURCE_DIR)/scripts/rabbitmq-server.bat $(SOURCE_DIR)/sbin
diff --git a/scripts/rabbitmq-activate-plugins.bat b/scripts/rabbitmq-activate-plugins.bat
index 8bef4ad2..3540bf2d 100644
--- a/scripts/rabbitmq-activate-plugins.bat
+++ b/scripts/rabbitmq-activate-plugins.bat
@@ -30,10 +30,6 @@ REM
 REM   Contributor(s): ______________________________________.

 REM

 

-if "%ERLANG_HOME%"=="" (

-    set ERLANG_HOME=%~dp0%..\..\..

-)

-

 if not exist "%ERLANG_HOME%\bin\erl.exe" (

     echo.

     echo ******************************

diff --git a/scripts/rabbitmq-multi.bat b/scripts/rabbitmq-multi.bat
index a30c0889..8abf13f1 100755
--- a/scripts/rabbitmq-multi.bat
+++ b/scripts/rabbitmq-multi.bat
@@ -49,10 +49,6 @@ if "%RABBITMQ_NODE_PORT%"=="" (
 set RABBITMQ_PIDS_FILE=%RABBITMQ_BASE%\rabbitmq.pids

 set RABBITMQ_SCRIPT_HOME=%~sdp0%

 

-if "%ERLANG_HOME%"=="" (

-    set ERLANG_HOME=%~dp0%..\..\..

-)

-

 if not exist "%ERLANG_HOME%\bin\erl.exe" (

     echo.

     echo ******************************

diff --git a/scripts/rabbitmq-server.bat b/scripts/rabbitmq-server.bat
index b4868841..a784fee3 100755
--- a/scripts/rabbitmq-server.bat
+++ b/scripts/rabbitmq-server.bat
@@ -46,10 +46,6 @@ if "%RABBITMQ_NODE_PORT%"=="" (
     set RABBITMQ_NODE_PORT=5672

 )

 

-if "%ERLANG_HOME%"=="" (

-    set ERLANG_HOME=%~dp0%..\..\..

-)

-

 if not exist "%ERLANG_HOME%\bin\erl.exe" (

     echo.

     echo ******************************

diff --git a/scripts/rabbitmqctl.bat b/scripts/rabbitmqctl.bat
index e4dccfba..5111724f 100755
--- a/scripts/rabbitmqctl.bat
+++ b/scripts/rabbitmqctl.bat
@@ -30,10 +30,6 @@ REM
 REM   Contributor(s): ______________________________________.

 REM

 

-if "%ERLANG_HOME%"=="" (

-    set ERLANG_HOME=%~dp0%..\..\..

-)

-

 if not exist "%ERLANG_HOME%\bin\erl.exe" (

     echo.

     echo ******************************

diff --git a/src/priority_queue.erl b/src/priority_queue.erl
index 732757c4..c74b39a9 100644
--- a/src/priority_queue.erl
+++ b/src/priority_queue.erl
@@ -55,7 +55,8 @@
 
 -module(priority_queue).
 
--export([new/0, is_queue/1, is_empty/1, len/1, to_list/1, in/2, in/3, out/1]).
+-export([new/0, is_queue/1, is_empty/1, len/1, to_list/1, in/2, in/3,
+         out/1, join/2]).
 
 %%----------------------------------------------------------------------------
 
@@ -73,6 +74,7 @@
 -spec(in/2 :: (any(), pqueue()) -> pqueue()).
 -spec(in/3 :: (any(), priority(), pqueue()) -> pqueue()).
 -spec(out/1 :: (pqueue()) -> {empty | {value, any()}, pqueue()}).
+-spec(join/2 :: (pqueue(), pqueue()) -> pqueue()).
 
 -endif.
 
@@ -147,6 +149,42 @@ out({pqueue, [{P, Q} | Queues]}) ->
            end,
     {R, NewQ}.
 
+join(A, {queue, [], []}) ->
+    A;
+join({queue, [], []}, B) ->
+    B;
+join({queue, AIn, AOut}, {queue, BIn, BOut}) ->
+    {queue, BIn, AOut ++ lists:reverse(AIn, BOut)};
+join(A = {queue, _, _}, {pqueue, BPQ}) ->
+    {Pre, Post} = lists:splitwith(fun ({P, _}) -> P < 0 end, BPQ),
+    Post1 = case Post of
+                []                        -> [ {0, A} ];
+                [ {0, ZeroQueue} | Rest ] -> [ {0, join(A, ZeroQueue)} | Rest ];
+                _                         -> [ {0, A} | Post ]
+            end,
+    {pqueue, Pre ++ Post1};
+join({pqueue, APQ}, B = {queue, _, _}) ->
+    {Pre, Post} = lists:splitwith(fun ({P, _}) -> P < 0 end, APQ),
+    Post1 = case Post of
+                []                        -> [ {0, B} ];
+                [ {0, ZeroQueue} | Rest ] -> [ {0, join(ZeroQueue, B)} | Rest ];
+                _                         -> [ {0, B} | Post ]
+            end,
+    {pqueue, Pre ++ Post1};
+join({pqueue, APQ}, {pqueue, BPQ}) ->
+    {pqueue, merge(APQ, BPQ, [])}.
+
+merge([], BPQ, Acc) ->
+    lists:reverse(Acc, BPQ);
+merge(APQ, [], Acc) ->
+    lists:reverse(Acc, APQ);
+merge([{P, A}|As], [{P, B}|Bs], Acc) ->
+    merge(As, Bs, [ {P, join(A, B)} | Acc ]);
+merge([{PA, A}|As], Bs = [{PB, _}|_], Acc) when PA < PB ->
+    merge(As, Bs, [ {PA, A} | Acc ]);
+merge(As = [{_, _}|_], [{PB, B}|Bs], Acc) ->
+    merge(As, Bs, [ {PB, B} | Acc ]).
+
 r2f([])      -> {queue, [], []};
 r2f([_] = R) -> {queue, [], R};
 r2f([X,Y])   -> {queue, [X], [Y]};
diff --git a/src/rabbit_alarm.erl b/src/rabbit_alarm.erl
index 21999f16..309c9a0e 100644
--- a/src/rabbit_alarm.erl
+++ b/src/rabbit_alarm.erl
@@ -41,7 +41,7 @@
 -define(MEMSUP_CHECK_INTERVAL, 1000).
 
 %% OSes on which we know memory alarms to be trustworthy
--define(SUPPORTED_OS, [{unix, linux}]).
+-define(SUPPORTED_OS, [{unix, linux}, {unix, darwin}]).
 
 -record(alarms, {alertees, system_memory_high_watermark = false}).
 
@@ -136,33 +136,35 @@ code_change(_OldVsn, State, _Extra) ->
 %%----------------------------------------------------------------------------
 
 start_memsup() ->
-    Mod = case os:type() of
-              %% memsup doesn't take account of buffers or cache when
-              %% considering "free" memory - therefore on Linux we can
-              %% get memory alarms very easily without any pressure
-              %% existing on memory at all. Therefore we need to use
-              %% our own simple memory monitor.
-              %%
-              {unix, linux} -> rabbit_memsup_linux;
-
-              %% Start memsup programmatically rather than via the
-              %% rabbitmq-server script. This is not quite the right
-              %% thing to do as os_mon checks to see if memsup is
-              %% available before starting it, but as memsup is
-              %% available everywhere (even on VXWorks) it should be
-              %% ok.
-              %%
-              %% One benefit of the programmatic startup is that we
-              %% can add our alarm_handler before memsup is running,
-              %% thus ensuring that we notice memory alarms that go
-              %% off on startup.
-              %%
-              _             -> memsup
-          end,
+    {Mod, Args} =
+        case os:type() of
+            %% memsup doesn't take account of buffers or cache when
+            %% considering "free" memory - therefore on Linux we can
+            %% get memory alarms very easily without any pressure
+            %% existing on memory at all. Therefore we need to use
+            %% our own simple memory monitor.
+            %%
+            {unix, linux}  -> {rabbit_memsup, [rabbit_memsup_linux]};
+            {unix, darwin} -> {rabbit_memsup, [rabbit_memsup_darwin]};
+
+            %% Start memsup programmatically rather than via the
+            %% rabbitmq-server script. This is not quite the right
+            %% thing to do as os_mon checks to see if memsup is
+            %% available before starting it, but as memsup is
+            %% available everywhere (even on VXWorks) it should be
+            %% ok.
+            %%
+            %% One benefit of the programmatic startup is that we
+            %% can add our alarm_handler before memsup is running,
+            %% thus ensuring that we notice memory alarms that go
+            %% off on startup.
+            %%
+            _              -> {memsup, []}
+        end,
     %% This is based on os_mon:childspec(memsup, true)
     {ok, _} = supervisor:start_child(
                 os_mon_sup,
-                {memsup, {Mod, start_link, []},
+                {memsup, {Mod, start_link, Args},
                  permanent, 2000, worker, [Mod]}),
     ok.
 
diff --git a/src/rabbit_amqqueue.erl b/src/rabbit_amqqueue.erl
index 4903c2c5..f05f7880 100644
--- a/src/rabbit_amqqueue.erl
+++ b/src/rabbit_amqqueue.erl
@@ -303,10 +303,10 @@ basic_cancel(#amqqueue{pid = QPid}, ChPid, ConsumerTag, OkMsg) ->
                           infinity).
 
 notify_sent(QPid, ChPid) ->
-    gen_server2:cast(QPid, {notify_sent, ChPid}).
+    gen_server2:pcast(QPid, 8, {notify_sent, ChPid}).
 
 unblock(QPid, ChPid) ->
-    gen_server2:cast(QPid, {unblock, ChPid}).
+    gen_server2:pcast(QPid, 8, {unblock, ChPid}).
 
 internal_delete(QueueName) ->
     rabbit_misc:execute_mnesia_transaction(
diff --git a/src/rabbit_guid.erl b/src/rabbit_guid.erl
index 2be00503..b789fbd1 100644
--- a/src/rabbit_guid.erl
+++ b/src/rabbit_guid.erl
@@ -42,6 +42,7 @@
          terminate/2, code_change/3]).
 
 -define(SERVER, ?MODULE).
+-define(SERIAL_FILENAME, "rabbit_serial").
 
 -record(state, {serial}).
 
@@ -59,17 +60,28 @@
 %%----------------------------------------------------------------------------
 
 start_link() ->
-    %% The persister can get heavily loaded, and we don't want that to
-    %% impact guid generation.  We therefore keep the serial in a
-    %% separate process rather than calling rabbit_persister:serial/0
-    %% directly in the functions below.
     gen_server:start_link({local, ?SERVER}, ?MODULE,
-                          [rabbit_persister:serial()], []).
+                          [update_disk_serial()], []).
+
+update_disk_serial() ->
+    Filename = filename:join(rabbit_mnesia:dir(), ?SERIAL_FILENAME),
+    Serial = case rabbit_misc:read_term_file(Filename) of
+                 {ok, [Num]}     -> Num;
+                 {error, enoent} -> rabbit_persister:serial();
+                 {error, Reason} ->
+                     throw({error, {cannot_read_serial_file, Filename, Reason}})
+             end,
+    case rabbit_misc:write_term_file(Filename, [Serial + 1]) of
+        ok -> ok;
+        {error, Reason1} ->
+            throw({error, {cannot_write_serial_file, Filename, Reason1}})
+    end,
+    Serial.
 
 %% generate a guid that is monotonically increasing per process.
 %%
 %% The id is only unique within a single cluster and as long as the
-%% persistent message store hasn't been deleted.
+%% serial store hasn't been deleted.
 guid() ->
     %% We don't use erlang:now() here because a) it may return
     %% duplicates when the system clock has been rewound prior to a
@@ -77,7 +89,7 @@ guid() ->
     %% now() to move ahead of the system time), and b) it is really
     %% slow since it takes a global lock and makes a system call.
     %%
-    %% rabbit_persister:serial/0, in combination with self/0 (which
+    %% A persisted serial number, in combination with self/0 (which
     %% includes the node name) uniquely identifies a process in space
     %% and time. We combine that with a process-local counter to give
     %% us a GUID that is monotonically increasing per process.
diff --git a/src/rabbit_memsup.erl b/src/rabbit_memsup.erl
new file mode 100644
index 00000000..b0d57cb2
--- /dev/null
+++ b/src/rabbit_memsup.erl
@@ -0,0 +1,142 @@
+%%   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_memsup).
+
+-behaviour(gen_server).
+
+-export([start_link/1]).
+
+-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
+         terminate/2, code_change/3]).
+
+-export([update/0]).
+
+-record(state, {memory_fraction,
+                timeout,
+                timer,
+                mod,
+                mod_state,
+                alarmed
+               }).
+
+-define(SERVER, memsup). %% must be the same as the standard memsup
+
+-define(DEFAULT_MEMORY_CHECK_INTERVAL, 1000).
+
+%%----------------------------------------------------------------------------
+
+-ifdef(use_specs).
+
+-spec(start_link/1 :: (atom()) -> {'ok', pid()} | 'ignore' | {'error', any()}).
+-spec(update/0 :: () -> 'ok').
+     
+-endif.
+
+%%----------------------------------------------------------------------------
+
+start_link(Args) ->
+    gen_server:start_link({local, ?SERVER}, ?MODULE, [Args], []).
+
+update() ->
+    gen_server:cast(?SERVER, update).
+
+%%----------------------------------------------------------------------------
+
+init([Mod]) -> 
+    Fraction = os_mon:get_env(memsup, system_memory_high_watermark),
+    TRef = start_timer(?DEFAULT_MEMORY_CHECK_INTERVAL),
+    InitState = Mod:init(),
+    State = #state { memory_fraction = Fraction,
+                     timeout = ?DEFAULT_MEMORY_CHECK_INTERVAL,
+                     timer = TRef,
+                     mod = Mod,
+                     mod_state = InitState,
+                     alarmed = false },
+    {ok, internal_update(State)}.
+
+start_timer(Timeout) ->
+    {ok, TRef} = timer:apply_interval(Timeout, ?MODULE, update, []),
+    TRef.
+
+%% Export the same API as the real memsup. Note that
+%% get_sysmem_high_watermark gives an int in the range 0 - 100, while
+%% set_sysmem_high_watermark takes a float in the range 0.0 - 1.0.
+handle_call(get_sysmem_high_watermark, _From, State) ->
+    {reply, trunc(100 * State#state.memory_fraction), State};
+
+handle_call({set_sysmem_high_watermark, Float}, _From, State) ->
+    {reply, ok, State#state{memory_fraction = Float}};
+
+handle_call(get_check_interval, _From, State) ->
+    {reply, State#state.timeout, State};
+
+handle_call({set_check_interval, Timeout}, _From, State) ->
+    {ok, cancel} = timer:cancel(State#state.timer),
+    {reply, ok, State#state{timeout = Timeout, timer = start_timer(Timeout)}};
+
+handle_call(get_memory_data, _From,
+            State = #state { mod = Mod, mod_state = ModState }) ->
+    {reply, Mod:get_memory_data(ModState), State};
+
+handle_call(_Request, _From, State) ->
+    {noreply, State}.
+
+handle_cast(update, State) ->
+    {noreply, internal_update(State)};
+
+handle_cast(_Request, State) -> 
+    {noreply, State}.
+
+handle_info(_Info, State) -> 
+    {noreply, State}.
+
+terminate(_Reason, _State) -> 
+    ok.
+
+code_change(_OldVsn, State, _Extra) -> 
+    {ok, State}.
+ 
+internal_update(State = #state { memory_fraction = MemoryFraction,
+                                 alarmed = Alarmed,
+                                 mod = Mod, mod_state = ModState }) ->
+    ModState1 = Mod:update(ModState),
+    {MemTotal, MemUsed, _BigProc} = Mod:get_memory_data(ModState1),
+    NewAlarmed = MemUsed / MemTotal > MemoryFraction,
+    case {Alarmed, NewAlarmed} of
+        {false, true} ->
+            alarm_handler:set_alarm({system_memory_high_watermark, []});
+        {true, false} ->
+            alarm_handler:clear_alarm(system_memory_high_watermark);
+        _ ->
+            ok
+    end,    
+    State #state { mod_state = ModState1, alarmed = NewAlarmed }.
diff --git a/src/rabbit_memsup_darwin.erl b/src/rabbit_memsup_darwin.erl
new file mode 100644
index 00000000..3de2d843
--- /dev/null
+++ b/src/rabbit_memsup_darwin.erl
@@ -0,0 +1,88 @@
+%%   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_memsup_darwin).
+
+-export([init/0, update/1, get_memory_data/1]).
+
+-record(state, {total_memory,
+                allocated_memory}).
+
+%%----------------------------------------------------------------------------
+
+-ifdef(use_specs).
+
+-type(state() :: #state { total_memory :: ('undefined' | non_neg_integer()),
+                          allocated_memory :: ('undefined' | non_neg_integer())
+                        }).
+
+-spec(init/0 :: () -> state()).
+-spec(update/1 :: (state()) -> state()).
+-spec(get_memory_data/1 :: (state()) -> {non_neg_integer(), non_neg_integer(),
+                                         ('undefined' | pid())}).
+
+-endif.
+
+%%----------------------------------------------------------------------------
+
+init() -> 
+    #state{total_memory = undefined,
+           allocated_memory = undefined}.
+
+update(State) ->
+    File = os:cmd("/usr/bin/vm_stat"),
+    Lines = string:tokens(File, "\n"),
+    Dict = dict:from_list(lists:map(fun parse_line/1, Lines)),
+    [PageSize, Inactive, Active, Free, Wired] =
+        [dict:fetch(Key, Dict) ||
+            Key <- [page_size, 'Pages inactive', 'Pages active', 'Pages free',
+                    'Pages wired down']],
+    MemTotal = PageSize * (Inactive + Active + Free + Wired),
+    MemUsed = PageSize * (Active + Wired),
+    State#state{total_memory = MemTotal, allocated_memory = MemUsed}.
+
+get_memory_data(State) ->
+    {State#state.total_memory, State#state.allocated_memory, undefined}.
+
+%%----------------------------------------------------------------------------
+
+%% A line looks like "Foo bar: 123456."
+parse_line(Line) ->
+    [Name, RHS | _Rest] = string:tokens(Line, ":"),
+    case Name of
+        "Mach Virtual Memory Statistics" ->
+            ["(page", "size", "of", PageSize, "bytes)"] =
+                string:tokens(RHS, " "),
+            {page_size, list_to_integer(PageSize)};
+        _ ->
+            [Value | _Rest1] = string:tokens(RHS, " ."),
+            {list_to_atom(Name), list_to_integer(Value)}
+    end.
diff --git a/src/rabbit_memsup_linux.erl b/src/rabbit_memsup_linux.erl
index ffdc7e99..ca942d7c 100644
--- a/src/rabbit_memsup_linux.erl
+++ b/src/rabbit_memsup_linux.erl
@@ -31,104 +31,44 @@
 
 -module(rabbit_memsup_linux).
 
--behaviour(gen_server).
+-export([init/0, update/1, get_memory_data/1]).
 
--export([start_link/0]).
-
--export([init/1, handle_call/3, handle_cast/2, handle_info/2,
-         terminate/2, code_change/3]).
-
--export([update/0]).
-
--define(SERVER, memsup). %% must be the same as the standard memsup
-
--define(DEFAULT_MEMORY_CHECK_INTERVAL, 1000).
-
--record(state, {memory_fraction, alarmed, timeout, timer}).   
+-record(state, {total_memory,
+                allocated_memory}).
 
 %%----------------------------------------------------------------------------
 
 -ifdef(use_specs).
 
--spec(start_link/0 :: () -> {'ok', pid()} | 'ignore' | {'error', any()}).
--spec(update/0 :: () -> 'ok').
-     
--endif.
-
-%%----------------------------------------------------------------------------
-
-start_link() ->
-    gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).
+-type(state() :: #state { total_memory :: ('undefined' | non_neg_integer()),
+                          allocated_memory :: ('undefined' | non_neg_integer())
+                        }).
 
+-spec(init/0 :: () -> state()).
+-spec(update/1 :: (state()) -> state()).
+-spec(get_memory_data/1 :: (state()) -> {non_neg_integer(), non_neg_integer(),
+                                         ('undefined' | pid())}).
 
-update() ->
-    gen_server:cast(?SERVER, update).
+-endif.
 
 %%----------------------------------------------------------------------------
 
-init(_Args) -> 
-    Fraction = os_mon:get_env(memsup, system_memory_high_watermark),
-    TRef = start_timer(?DEFAULT_MEMORY_CHECK_INTERVAL),
-    {ok, #state{alarmed = false, 
-                memory_fraction = Fraction, 
-                timeout = ?DEFAULT_MEMORY_CHECK_INTERVAL,
-                timer = TRef}}.
-
-start_timer(Timeout) ->
-    {ok, TRef} = timer:apply_interval(Timeout, ?MODULE, update, []),
-    TRef.
-
-%% Export the same API as the real memsup. Note that
-%% get_sysmem_high_watermark gives an int in the range 0 - 100, while
-%% set_sysmem_high_watermark takes a float in the range 0.0 - 1.0.
-handle_call(get_sysmem_high_watermark, _From, State) ->
-    {reply, trunc(100 * State#state.memory_fraction), State};
-
-handle_call({set_sysmem_high_watermark, Float}, _From, State) ->
-    {reply, ok, State#state{memory_fraction = Float}};
+init() -> 
+    #state{total_memory = undefined,
+           allocated_memory = undefined}.
 
-handle_call(get_check_interval, _From, State) ->
-    {reply, State#state.timeout, State};
-
-handle_call({set_check_interval, Timeout}, _From, State) ->
-    {ok, cancel} = timer:cancel(State#state.timer),
-    {reply, ok, State#state{timeout = Timeout, timer = start_timer(Timeout)}};
-
-handle_call(_Request, _From, State) -> 
-    {noreply, State}.
-
-handle_cast(update, State = #state{alarmed = Alarmed,
-                                   memory_fraction = MemoryFraction}) -> 
+update(State) ->
     File = read_proc_file("/proc/meminfo"),
     Lines = string:tokens(File, "\n"),
     Dict = dict:from_list(lists:map(fun parse_line/1, Lines)),
-    MemTotal = dict:fetch('MemTotal', Dict),
-    MemUsed = MemTotal
-        - dict:fetch('MemFree', Dict)
-        - dict:fetch('Buffers', Dict)
-        - dict:fetch('Cached', Dict),
-    NewAlarmed = MemUsed / MemTotal > MemoryFraction,
-    case {Alarmed, NewAlarmed} of
-        {false, true} ->
-            alarm_handler:set_alarm({system_memory_high_watermark, []});
-        {true, false} ->
-            alarm_handler:clear_alarm(system_memory_high_watermark);
-        _ ->
-            ok
-    end,
-    {noreply,  State#state{alarmed = NewAlarmed}};
-
-handle_cast(_Request, State) -> 
-    {noreply, State}.
-
-handle_info(_Info, State) -> 
-    {noreply, State}.
-
-terminate(_Reason, _State) -> 
-    ok.
-
-code_change(_OldVsn, State, _Extra) -> 
-    {ok, State}.
+    [MemTotal, MemFree, Buffers, Cached] =
+        [dict:fetch(Key, Dict) ||
+            Key <- ['MemTotal', 'MemFree', 'Buffers', 'Cached']],
+    MemUsed = MemTotal - MemFree - Buffers - Cached,
+    State#state{total_memory = MemTotal, allocated_memory = MemUsed}.
+
+get_memory_data(State) ->
+    {State#state.total_memory, State#state.allocated_memory, undefined}.
 
 %%----------------------------------------------------------------------------
 
@@ -152,5 +92,10 @@ read_proc_file(IoDevice, Acc) ->
 
 %% A line looks like "FooBar: 123456 kB"
 parse_line(Line) ->
-    [Name, Value | _] = string:tokens(Line, ": "),   
-    {list_to_atom(Name), list_to_integer(Value)}.
+    [Name, RHS | _Rest] = string:tokens(Line, ":"),
+    [Value | UnitsRest] = string:tokens(RHS, " "),
+    Value1 = case UnitsRest of
+                 [] -> list_to_integer(Value); %% no units
+                 ["kB"] -> list_to_integer(Value) * 1024
+             end,
+    {list_to_atom(Name), Value1}.
diff --git a/src/rabbit_misc.erl b/src/rabbit_misc.erl
index 13a2aa05..95a274e3 100644
--- a/src/rabbit_misc.erl
+++ b/src/rabbit_misc.erl
@@ -50,9 +50,11 @@
 -export([intersperse/2, upmap/2, map_in_order/2]).
 -export([table_foreach/2]).
 -export([dirty_read_all/1, dirty_foreach_key/2, dirty_dump_log/1]).
+-export([read_term_file/1, write_term_file/2]).
 -export([append_file/2, ensure_parent_dirs_exist/1]).
 -export([format_stderr/2]).
 -export([start_applications/1, stop_applications/1]).
+-export([unfold/2, ceil/1]).
 
 -import(mnesia).
 -import(lists).
@@ -65,6 +67,8 @@
 
 -include_lib("kernel/include/inet.hrl").
 
+-type(ok_or_error() :: 'ok' | {'error', any()}).
+
 -spec(method_record_type/1 :: (tuple()) -> atom()).
 -spec(polite_pause/0 :: () -> 'done').
 -spec(polite_pause/1 :: (non_neg_integer()) -> 'done').
@@ -88,9 +92,9 @@
 -spec(r_arg/4 :: (vhost() | r(atom()), K, amqp_table(), binary()) ->
              undefined | r(K)  when is_subtype(K, atom())).
 -spec(rs/1 :: (r(atom())) -> string()).
--spec(enable_cover/0 :: () -> 'ok' | {'error', any()}).
+-spec(enable_cover/0 :: () -> ok_or_error()).
 -spec(report_cover/0 :: () -> 'ok').
--spec(enable_cover/1 :: (string()) -> 'ok' | {'error', any()}).
+-spec(enable_cover/1 :: (string()) -> ok_or_error()).
 -spec(report_cover/1 :: (string()) -> 'ok').
 -spec(throw_on_error/2 ::
       (atom(), thunk({error, any()} | {ok, A} | A)) -> A). 
@@ -100,7 +104,7 @@
 -spec(with_vhost/2 :: (vhost(), thunk(A)) -> A).
 -spec(with_user_and_vhost/3 :: (username(), vhost(), thunk(A)) -> A).
 -spec(execute_mnesia_transaction/1 :: (thunk(A)) -> A).
--spec(ensure_ok/2 :: ('ok' | {'error', any()}, atom()) -> 'ok').
+-spec(ensure_ok/2 :: (ok_or_error(), atom()) -> 'ok').
 -spec(localnode/1 :: (atom()) -> erlang_node()).
 -spec(tcp_name/3 :: (atom(), ip_address(), ip_port()) -> atom()).
 -spec(intersperse/2 :: (A, [A]) -> [A]).
@@ -110,12 +114,16 @@
 -spec(dirty_read_all/1 :: (atom()) -> [any()]).
 -spec(dirty_foreach_key/2 :: (fun ((any()) -> any()), atom()) ->
              'ok' | 'aborted').
--spec(dirty_dump_log/1 :: (string()) -> 'ok' | {'error', any()}).
--spec(append_file/2 :: (string(), string()) -> 'ok' | {'error', any()}).
+-spec(dirty_dump_log/1 :: (string()) -> ok_or_error()).
+-spec(read_term_file/1 :: (string()) -> {'ok', [any()]} | {'error', any()}).
+-spec(write_term_file/2 :: (string(), [any()]) -> ok_or_error()).
+-spec(append_file/2 :: (string(), string()) -> ok_or_error()).
 -spec(ensure_parent_dirs_exist/1 :: (string()) -> 'ok').
 -spec(format_stderr/2 :: (string(), [any()]) -> 'ok').
 -spec(start_applications/1 :: ([atom()]) -> 'ok').
 -spec(stop_applications/1 :: ([atom()]) -> 'ok').
+-spec(unfold/2  :: (fun ((A) -> ({'true', B, A} | 'false')), A) -> {[B], A}).
+-spec(ceil/1 :: (number()) -> number()).
 
 -endif.
 
@@ -376,6 +384,12 @@ dirty_dump_log1(LH, {K, Terms, BadBytes}) ->
     dirty_dump_log1(LH, disk_log:chunk(LH, K)).
 
 
+read_term_file(File) -> file:consult(File).
+
+write_term_file(File, Terms) ->
+    file:write_file(File, list_to_binary([io_lib:format("~w.~n", [Term]) ||
+                                             Term <- Terms])).
+
 append_file(File, Suffix) ->
     case file:read_file_info(File) of
         {ok, FInfo}     -> append_file(File, FInfo#file_info.size, Suffix);
@@ -446,3 +460,18 @@ stop_applications(Apps) ->
                         cannot_stop_application,
                         Apps).
 
+unfold(Fun, Init) ->
+    unfold(Fun, [], Init).
+
+unfold(Fun, Acc, Init) ->
+    case Fun(Init) of
+        {true, E, I} -> unfold(Fun, [E|Acc], I);
+        false -> {Acc, Init}
+    end.
+
+ceil(N) ->
+    T = trunc(N),
+    case N - T of
+        0 -> N;
+        _ -> 1 + T
+    end.
diff --git a/src/rabbit_mnesia.erl b/src/rabbit_mnesia.erl
index 65b23559..37e20335 100644
--- a/src/rabbit_mnesia.erl
+++ b/src/rabbit_mnesia.erl
@@ -197,28 +197,16 @@ cluster_nodes_config_filename() ->
 
 create_cluster_nodes_config(ClusterNodes) ->
     FileName = cluster_nodes_config_filename(),
-    Handle = case file:open(FileName, [write]) of
-                 {ok, Device} -> Device;
-                 {error, Reason} ->
-                     throw({error, {cannot_create_cluster_nodes_config,
-                                    FileName, Reason}})
-             end,
-    try
-        ok = io:write(Handle, ClusterNodes),
-        ok = io:put_chars(Handle, [$.])
-    after
-        case file:close(Handle) of
-            ok -> ok;
-            {error, Reason1} ->
-                throw({error, {cannot_close_cluster_nodes_config,
-                               FileName, Reason1}})
-        end
-    end,
-    ok.
+    case rabbit_misc:write_term_file(FileName, [ClusterNodes]) of
+        ok -> ok;
+        {error, Reason} ->
+            throw({error, {cannot_create_cluster_nodes_config,
+                           FileName, Reason}})
+    end.
 
 read_cluster_nodes_config() ->
     FileName = cluster_nodes_config_filename(),
-    case file:consult(FileName) of
+    case rabbit_misc:read_term_file(FileName) of
         {ok, [ClusterNodes]} -> ClusterNodes;
         {error, enoent} ->
             case application:get_env(cluster_config) of
diff --git a/src/rabbit_tests.erl b/src/rabbit_tests.erl
index 4f207fbb..b4cd30bc 100644
--- a/src/rabbit_tests.erl
+++ b/src/rabbit_tests.erl
@@ -49,6 +49,7 @@ test_content_prop_roundtrip(Datum, Binary) ->
 
 all_tests() ->
     passed = test_priority_queue(),
+    passed = test_unfold(),
     passed = test_parsing(),
     passed = test_topic_matching(),
     passed = test_log_management(),
@@ -75,7 +76,8 @@ test_priority_queue() ->
 
     %% 1-element priority Q
     Q1 = priority_queue:in(foo, 1, priority_queue:new()),
-    {true, false, 1, [{1, foo}], [foo]} = test_priority_queue(Q1),
+    {true, false, 1, [{1, foo}], [foo]} =
+        test_priority_queue(Q1),
 
     %% 2-element same-priority Q
     Q2 = priority_queue:in(bar, 1, Q1),
@@ -91,6 +93,71 @@ test_priority_queue() ->
     Q4 = priority_queue:in(foo, -1, priority_queue:new()),
     {true, false, 1, [{-1, foo}], [foo]} = test_priority_queue(Q4),
 
+    %% merge 2 * 1-element no-priority Qs
+    Q5 = priority_queue:join(priority_queue:in(foo, Q),
+                             priority_queue:in(bar, Q)),
+    {true, false, 2, [{0, foo}, {0, bar}], [foo, bar]} =
+        test_priority_queue(Q5),
+
+    %% merge 1-element no-priority Q with 1-element priority Q
+    Q6 = priority_queue:join(priority_queue:in(foo, Q),
+                             priority_queue:in(bar, 1, Q)),
+    {true, false, 2, [{1, bar}, {0, foo}], [bar, foo]} =
+        test_priority_queue(Q6),
+
+    %% merge 1-element priority Q with 1-element no-priority Q 
+    Q7 = priority_queue:join(priority_queue:in(foo, 1, Q),
+                             priority_queue:in(bar, Q)),
+    {true, false, 2, [{1, foo}, {0, bar}], [foo, bar]} =
+        test_priority_queue(Q7),
+
+    %% merge 2 * 1-element same-priority Qs
+    Q8 = priority_queue:join(priority_queue:in(foo, 1, Q),
+                             priority_queue:in(bar, 1, Q)),
+    {true, false, 2, [{1, foo}, {1, bar}], [foo, bar]} =
+        test_priority_queue(Q8),
+
+    %% merge 2 * 1-element different-priority Qs
+    Q9 = priority_queue:join(priority_queue:in(foo, 1, Q),
+                             priority_queue:in(bar, 2, Q)),
+    {true, false, 2, [{2, bar}, {1, foo}], [bar, foo]} =
+        test_priority_queue(Q9),
+
+    %% merge 2 * 1-element different-priority Qs (other way around)
+    Q10 = priority_queue:join(priority_queue:in(bar, 2, Q),
+                              priority_queue:in(foo, 1, Q)),
+    {true, false, 2, [{2, bar}, {1, foo}], [bar, foo]} =
+        test_priority_queue(Q10),
+
+    %% merge 2 * 2-element multi-different-priority Qs
+    Q11 = priority_queue:join(Q6, Q5),
+    {true, false, 4, [{1, bar}, {0, foo}, {0, foo}, {0, bar}],
+     [bar, foo, foo, bar]} = test_priority_queue(Q11),
+
+    %% and the other way around
+    Q12 = priority_queue:join(Q5, Q6),
+    {true, false, 4, [{1, bar}, {0, foo}, {0, bar}, {0, foo}],
+     [bar, foo, bar, foo]} = test_priority_queue(Q12),
+
+    %% merge with negative priorities
+    Q13 = priority_queue:join(Q4, Q5),
+    {true, false, 3, [{0, foo}, {0, bar}, {-1, foo}], [foo, bar, foo]} =
+        test_priority_queue(Q13),
+
+    %% and the other way around
+    Q14 = priority_queue:join(Q5, Q4),
+    {true, false, 3, [{0, foo}, {0, bar}, {-1, foo}], [foo, bar, foo]} =
+        test_priority_queue(Q14),
+
+    %% joins with empty queues:
+    Q1 = priority_queue:join(Q, Q1),
+    Q1 = priority_queue:join(Q1, Q),
+
+    %% insert with priority into non-empty zero-priority queue
+    Q15 = priority_queue:in(baz, 1, Q5),
+    {true, false, 3, [{1, baz}, {0, foo}, {0, bar}], [baz, foo, bar]} =
+        test_priority_queue(Q15),
+
     passed.
 
 priority_queue_in_all(Q, L) ->
@@ -116,6 +183,14 @@ test_simple_n_element_queue(N) ->
     {true, false, N, ToListRes, Items} = test_priority_queue(Q),
     passed.
 
+test_unfold() ->
+    {[], test} = rabbit_misc:unfold(fun (_V) -> false end, test),
+    List = lists:seq(2,20,2),
+    {List, 0} = rabbit_misc:unfold(fun (0) -> false;
+                                       (N) -> {true, N*2, N-1}
+                                   end, 10),
+    passed.
+
 test_parsing() ->
     passed = test_content_properties(),
     passed.