Merge in default

9 files changed, 32 insertions, 24 deletions

diff --git a/packaging/RPMS/Fedora/rabbitmq-server.spec b/packaging/RPMS/Fedora/rabbitmq-server.spec
index 32404057..971e7241 100644
--- a/packaging/RPMS/Fedora/rabbitmq-server.spec
+++ b/packaging/RPMS/Fedora/rabbitmq-server.spec
@@ -130,6 +130,9 @@ done
 rm -rf %{buildroot}
 
 %changelog
+* Mon Aug 11 2014 simon@rabbitmq.com 3.3.5-1
+- New Upstream Release
+
 * Tue Jun 24 2014 simon@rabbitmq.com 3.3.4-1
 - New Upstream Release
 
diff --git a/packaging/debs/Debian/debian/changelog b/packaging/debs/Debian/debian/changelog
index d26991e4..8fde9087 100644
--- a/packaging/debs/Debian/debian/changelog
+++ b/packaging/debs/Debian/debian/changelog
@@ -1,3 +1,10 @@
+rabbitmq-server (3.3.5-1) unstable; urgency=low
+
+  * New Upstream Release
+  * Changed Uploaders from Emile Joubert to Blair Hester
+
+ -- Simon MacMullen <simon@rabbitmq.com>  Mon, 11 Aug 2014 12:23:31 +0100
+
 rabbitmq-server (3.3.4-1) unstable; urgency=low
 
   * New Upstream Release
diff --git a/src/priority_queue.erl b/src/priority_queue.erl
index 9a578aa9..a3573bbd 100644
--- a/src/priority_queue.erl
+++ b/src/priority_queue.erl
@@ -139,7 +139,7 @@ out({queue, [V], [], 1}) ->
     {{value, V}, {queue, [], [], 0}};
 out({queue, [Y|In], [], Len}) ->
     [V|Out] = lists:reverse(In, []),
-    {{value, V}, {queue, [Y], Out}, Len - 1};
+    {{value, V}, {queue, [Y], Out, Len - 1}};
 out({queue, In, [V], Len}) when is_list(In) ->
     {{value,V}, r2f(In, Len - 1)};
 out({queue, In,[V|Out], Len}) when is_list(In) ->
diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl
index 8f44d761..ba1517af 100644
--- a/src/rabbit_amqqueue_process.erl
+++ b/src/rabbit_amqqueue_process.erl
@@ -392,10 +392,8 @@ ensure_ttl_timer(Expiry, State = #q{ttl_timer_ref       = undefined,
 ensure_ttl_timer(Expiry, State = #q{ttl_timer_ref    = TRef,
                                     ttl_timer_expiry = TExpiry})
   when Expiry + 1000 < TExpiry ->
-    case rabbit_misc:cancel_timer(TRef) of
-        false -> State;
-        _     -> ensure_ttl_timer(Expiry, State#q{ttl_timer_ref = undefined})
-    end;
+    rabbit_misc:cancel_timer(TRef),
+    ensure_ttl_timer(Expiry, State#q{ttl_timer_ref = undefined});
 ensure_ttl_timer(_Expiry, State) ->
     State.
 
diff --git a/src/rabbit_mirror_queue_slave.erl b/src/rabbit_mirror_queue_slave.erl
index cc06ae44..6d0064ab 100644
--- a/src/rabbit_mirror_queue_slave.erl
+++ b/src/rabbit_mirror_queue_slave.erl
@@ -271,8 +271,8 @@ handle_cast({sync_start, Ref, Syncer},
            DD, Ref, TRef, Syncer, BQ, BQS,
            fun (BQN, BQSN) ->
                    BQSN1 = update_ram_duration(BQN, BQSN),
-                   TRefN = erlang:send_after(?RAM_DURATION_UPDATE_INTERVAL,
-                                             self(), update_ram_duration),
+                   TRefN = rabbit_misc:send_after(?RAM_DURATION_UPDATE_INTERVAL,
+                                                  self(), update_ram_duration),
                    {TRefN, BQSN1}
            end) of
         denied              -> noreply(State1);
diff --git a/src/rabbit_misc.erl b/src/rabbit_misc.erl
index 09355f3f..a5cd1bdd 100644
--- a/src/rabbit_misc.erl
+++ b/src/rabbit_misc.erl
@@ -92,9 +92,9 @@
       :: rabbit_types:channel_exit() | rabbit_types:connection_exit()).
 -type(digraph_label() :: term()).
 -type(graph_vertex_fun() ::
-        fun ((atom(), [term()]) -> [{digraph:vertex(), digraph_label()}])).
+        fun (({atom(), [term()]}) -> [{digraph:vertex(), digraph_label()}])).
 -type(graph_edge_fun() ::
-        fun ((atom(), [term()]) -> [{digraph:vertex(), digraph:vertex()}])).
+        fun (({atom(), [term()]}) -> [{digraph:vertex(), digraph:vertex()}])).
 -type(tref() :: {'erlang', reference()} | {timer, timer:tref()}).
 
 -spec(method_record_type/1 :: (rabbit_framing:amqp_method_record())
diff --git a/src/rabbit_reader.erl b/src/rabbit_reader.erl
index 2ac24f97..2cdd54a7 100644
--- a/src/rabbit_reader.erl
+++ b/src/rabbit_reader.erl
@@ -611,7 +611,7 @@ handle_exception(State = #v1{connection = #connection{protocol = Protocol},
     State1 = close_connection(terminate_channels(State)),
     ok = send_on_channel0(State1#v1.sock, CloseMethod, Protocol),
     State1;
-handle_exception(State = #v1{connection_state = tuning}, Channel, Reason) ->
+handle_exception(State, Channel, Reason) ->
     %% We don't trust the client at this point - force them to wait
     %% for a bit so they can't DOS us with repeated failed logins etc.
     timer:sleep(?SILENT_CLOSE_DELAY * 1000),
diff --git a/src/rabbit_tests.erl b/src/rabbit_tests.erl
index 9eddb51d..a186fb7a 100644
--- a/src/rabbit_tests.erl
+++ b/src/rabbit_tests.erl
@@ -37,7 +37,7 @@ all_tests() ->
     ok = supervisor2_tests:test_all(),
     passed = gm_tests:all_tests(),
     passed = mirrored_supervisor_tests:all_tests(),
-    application:set_env(rabbit, file_handles_high_watermark, 10, infinity),
+    application:set_env(rabbit, file_handles_high_watermark, 10),
     ok = file_handle_cache:set_limit(10),
     passed = test_version_equivalance(),
     passed = test_file_handle_cache(),
@@ -1870,22 +1870,20 @@ test_backing_queue() ->
         {ok, rabbit_variable_queue} ->
             {ok, FileSizeLimit} =
                 application:get_env(rabbit, msg_store_file_size_limit),
-            application:set_env(rabbit, msg_store_file_size_limit, 512,
-                                infinity),
+            application:set_env(rabbit, msg_store_file_size_limit, 512),
             {ok, MaxJournal} =
                 application:get_env(rabbit, queue_index_max_journal_entries),
-            application:set_env(rabbit, queue_index_max_journal_entries, 128,
-                                infinity),
+            application:set_env(rabbit, queue_index_max_journal_entries, 128),
             passed = test_msg_store(),
             application:set_env(rabbit, msg_store_file_size_limit,
-                                FileSizeLimit, infinity),
+                                FileSizeLimit),
             passed = test_queue_index(),
             passed = test_queue_index_props(),
             passed = test_variable_queue(),
             passed = test_variable_queue_delete_msg_store_files_callback(),
             passed = test_queue_recover(),
             application:set_env(rabbit, queue_index_max_journal_entries,
-                                MaxJournal, infinity),
+                                MaxJournal),
             %% We will have restarted the message store, and thus changed
             %% the order of the children of rabbit_sup. This will cause
             %% problems if there are subsequent failures - see bug 24262.
diff --git a/src/rabbit_variable_queue.erl b/src/rabbit_variable_queue.erl
index 9dcd6c76..0fb2aa12 100644
--- a/src/rabbit_variable_queue.erl
+++ b/src/rabbit_variable_queue.erl
@@ -522,6 +522,7 @@ purge(State = #vqstate { q4                = Q4,
     {Stats2, State1 = #vqstate { q1                = Q1,
                                  index_state       = IndexState2,
                                  msg_store_clients = MSCState1 }} =
+
         purge_betas_and_deltas(
           Stats1, State #vqstate { q4          = ?QUEUE:new(),
                                    index_state = IndexState1 }),
@@ -659,7 +660,7 @@ ack([SeqId], State) ->
      State1 = #vqstate { index_state       = IndexState,
                          msg_store_clients = MSCState,
                          ack_out_counter   = AckOutCount }} =
-        remove_pending_ack(SeqId, State),
+        remove_pending_ack(true, SeqId, State),
     IndexState1 = case IndexOnDisk of
                       true  -> rabbit_queue_index:ack([SeqId], IndexState);
                       false -> IndexState
@@ -678,7 +679,7 @@ ack(AckTags, State) ->
                          ack_out_counter   = AckOutCount }} =
         lists:foldl(
           fun (SeqId, {Acc, State2}) ->
-                  {MsgStatus, State3} = remove_pending_ack(SeqId, State2),
+                  {MsgStatus, State3} = remove_pending_ack(true, SeqId, State2),
                   {accumulate_ack(MsgStatus, Acc), State3}
           end, {accumulate_ack_init(), State}, AckTags),
     IndexState1 = rabbit_queue_index:ack(IndexOnDiskSeqIds, IndexState),
@@ -1347,15 +1348,16 @@ lookup_pending_ack(SeqId, #vqstate { ram_pending_ack  = RPA,
         none       -> gb_trees:get(SeqId, DPA)
     end.
 
-remove_pending_ack(SeqId, State) ->
+%% First parameter = UpdatePersistentCount
+remove_pending_ack(true, SeqId, State) ->
     {MsgStatus, State1 = #vqstate { persistent_count = PCount }} =
-        remove_pending_ack0(SeqId, State),
+        remove_pending_ack(false, SeqId, State),
     PCount1 = PCount - one_if(MsgStatus#msg_status.is_persistent),
     {MsgStatus, upd_bytes(-1, MsgStatus,
                           State1 # vqstate{ persistent_count = PCount1 })}.
 
-remove_pending_ack0(SeqId, State = #vqstate { ram_pending_ack  = RPA,
-                                              disk_pending_ack = DPA }) ->
+remove_pending_ack(false, SeqId, State = #vqstate { ram_pending_ack  = RPA,
+                                                    disk_pending_ack = DPA }) ->
     case gb_trees:lookup(SeqId, RPA) of
         {value, V} -> RPA1 = gb_trees:delete(SeqId, RPA),
                       {V, State #vqstate { ram_pending_ack = RPA1 }};
@@ -1506,7 +1508,7 @@ delta_merge(SeqIds, Delta, MsgIds, State) ->
 %% Mostly opposite of record_pending_ack/2
 msg_from_pending_ack(SeqId, State) ->
     {#msg_status { msg_props = MsgProps } = MsgStatus, State1} =
-        remove_pending_ack0(SeqId, State),
+        remove_pending_ack(false, SeqId, State),
     {MsgStatus #msg_status {
        msg_props = MsgProps #message_properties { needs_confirming = false } },
      State1}.