Merge branch 'stable'

deps/amqp_client/src/amqp_channel.erl

@@ -103,7 +103,7 @@
                 %% true | false, only relevant in the direct
                 %% client case.
                 %% when true, consumers will manually notify
-                %% queue pids using rabbit_amqqueue:notify_sent/2
+                %% queue pids using rabbit_amqqueue_common:notify_sent/2
                 %% to prevent the queue from overwhelming slow
                 %% consumers that use automatic acknowledgement
                 %% mode.
@@ -425,7 +425,7 @@ handle_cast(enable_delivery_flow_control, State) ->
     {noreply, State#state{delivery_flow_control = true}};
 %% @private
 handle_cast({send_notify, {QPid, ChPid}}, State) ->
-    rabbit_amqqueue:notify_sent(QPid, ChPid),
+    rabbit_amqqueue_common:notify_sent(QPid, ChPid),
     {noreply, State};
 %% @private
 handle_cast({cast, Method, AmqpMsg, Sender, noflow}, State) ->
@@ -489,7 +489,7 @@ handle_info({send_command_and_notify, QPid, ChPid,
             State = #state{delivery_flow_control = MFC}) ->
     case MFC of
         false -> handle_method_from_server(Method, Content, State),
-                 rabbit_amqqueue:notify_sent(QPid, ChPid);
+                 rabbit_amqqueue_common:notify_sent(QPid, ChPid);
         true  -> handle_method_from_server(Method, Content,
                                            {self(), QPid, ChPid}, State)
     end,
@@ -530,7 +530,7 @@ handle_info({'DOWN', _, process, FlowHandler, Reason},
               "Reason: ~p~n", [self(), FlowHandler, Reason]),
     {noreply, State#state{flow_handler = none}};
 handle_info({'DOWN', _, process, QPid, _Reason}, State) ->
-    rabbit_amqqueue:notify_sent_queue_down(QPid),
+    rabbit_amqqueue_common:notify_sent_queue_down(QPid),
     {noreply, State};
 handle_info({confirm_timeout, From}, State = #state{waiting_set = WSet}) ->
     case gb_trees:lookup(From, WSet) of

deps/amqp_client/src/amqp_direct_connection.erl

@@ -217,11 +217,11 @@ ssl_cert_info(Sock) ->
     case rabbit_net:peercert(Sock) of
         {ok, Cert} ->
             [{peer_cert_issuer,   list_to_binary(
-                                    rabbit_ssl:peer_cert_issuer(Cert))},
+                                    rabbit_cert_info:issuer(Cert))},
              {peer_cert_subject,  list_to_binary(
-                                    rabbit_ssl:peer_cert_subject(Cert))},
+                                    rabbit_cert_info:subject(Cert))},
              {peer_cert_validity, list_to_binary(
-                                    rabbit_ssl:peer_cert_validity(Cert))}];
+                                    rabbit_cert_info:validity(Cert))}];
         _ ->
             []
     end.

deps/amqp_client/src/amqp_gen_connection.erl

@@ -224,6 +224,14 @@ handle_info({'DOWN', _, process, BlockHandler, Reason},
     ?LOG_WARN("Connection (~p): Unregistering block handler ~p because it died. "
               "Reason: ~p~n", [self(), BlockHandler, Reason]),
     {noreply, State#state{block_handler = none}};
+handle_info({'EXIT', BlockHandler, Reason},
+            State = #state{block_handler = {BlockHandler, Ref}}) ->
+    ?LOG_WARN("Connection (~p): Unregistering block handler ~p because it died. "
+              "Reason: ~p~n", [self(), BlockHandler, Reason]),
+    erlang:demonitor(Ref, [flush]),
+    {noreply, State#state{block_handler = none}};
+handle_info({'EXIT', _Pid, _Reason}, State) ->
+    {noreply, State};
 handle_info(Info, State) ->
     callback(handle_message, [Info], State).
 

deps/amqp_client/src/amqp_network_connection.erl

@@ -144,7 +144,7 @@ do_connect({Addr, Family},
                          [Family | ?RABBIT_TCP_OPTS] ++ ExtraOpts,
                          Timeout) of
         {ok, Sock} ->
-            SslOpts = rabbit_networking:fix_ssl_options(
+            SslOpts = rabbit_ssl_options:fix(
                         orddict:to_list(
                           orddict:merge(fun (_, _A, B) -> B end,
                                         orddict:from_list(GlobalSslOpts),