merging in from default

deps/amqp_client/src/amqp_channel.erl

@@ -25,7 +25,7 @@
 
 -behaviour(gen_server).
 
--export([start_link/3, connection_closing/2, open/1]).
+-export([start_link/3, connection_closing/3, open/1]).
 -export([init/1, terminate/2, code_change/3, handle_call/3, handle_cast/2,
          handle_info/2]).
 -export([call/2, call/3, cast/2, cast/3]).
@@ -45,7 +45,8 @@
                 rpc_requests        = queue:new(),
                 anon_sub_requests   = queue:new(),
                 tagged_sub_requests = dict:new(),
-                closing             = false,
+                closing             = false, %% false | just_channel |
+                                             %%   {connection, Reason}
                 writer,
                 return_handler_pid  = none,
                 confirm_handler_pid = none,
@@ -238,8 +239,8 @@ start_link(Driver, ChannelNumber, SWF) ->
     gen_server:start_link(?MODULE, [self(), Driver, ChannelNumber, SWF], []).
 
 %% @private
-connection_closing(Pid, ChannelCloseType) ->
-    gen_server:cast(Pid, {connection_closing, ChannelCloseType}).
+connection_closing(Pid, ChannelCloseType, Reason) ->
+    gen_server:cast(Pid, {connection_closing, ChannelCloseType, Reason}).
 
 %% @private
 open(Pid) ->
@@ -318,14 +319,11 @@ handle_cast({method, Method, Content}, State) ->
 %% beforehand. The channel must block all further RPCs,
 %% flush the RPC queue (optional), and terminate
 %% @private
-handle_cast({connection_closing, CloseType}, State) ->
-    handle_connection_closing(CloseType, State);
+handle_cast({connection_closing, CloseType, Reason}, State) ->
+    handle_connection_closing(CloseType, Reason, State);
 %% @private
-handle_cast({shutdown, {_, 200, _}}, State) ->
-    {stop, normal, State};
-%% @private
-handle_cast({shutdown, Reason}, State) ->
-    {stop, Reason, State}.
+handle_cast({shutdown, Shutdown}, State) ->
+    handle_shutdown(Shutdown, State).
 
 %% Received from rabbit_channel in the direct case
 %% @private
@@ -511,9 +509,11 @@ do_rpc(State = #state{rpc_requests = Q,
             end;
         {empty, NewQ} ->
             case Closing of
-                connection -> gen_server:cast(self(),
-                                              {shutdown, connection_closing});
-                _          -> ok
+                {connection, Reason} ->
+                    gen_server:cast(self(),
+                                    {shutdown, {connection_closing, Reason}});
+                _ ->
+                    ok
             end,
             State#state{rpc_requests = NewQ}
     end.
@@ -643,19 +643,21 @@ handle_method_from_server1(Method, Content, State) ->
 %% Other handle_* functions
 %%---------------------------------------------------------------------------
 
-handle_connection_closing(CloseType, State = #state{rpc_requests = RpcQueue,
-                                                    closing      = Closing}) ->
+handle_connection_closing(CloseType, Reason,
+                          State = #state{rpc_requests = RpcQueue,
+                                         closing      = Closing}) ->
+    NewState = State#state{closing = {connection, Reason}},
     case {CloseType, Closing, queue:is_empty(RpcQueue)} of
         {flush, false, false} ->
             erlang:send_after(?TIMEOUT_FLUSH, self(),
                               timed_out_flushing_channel),
-            {noreply, State#state{closing = connection}};
+            {noreply, NewState};
         {flush, just_channel, false} ->
             erlang:send_after(?TIMEOUT_CLOSE_OK, self(),
                               timed_out_waiting_close_ok),
-            {noreply, State#state{closing = connection}};
+            {noreply, NewState};
         _ ->
-            {stop, connection_closing, State}
+            handle_shutdown({connection_closing, Reason}, NewState)
     end.
 
 handle_channel_exit(Reason, State) ->
@@ -665,14 +667,22 @@ handle_channel_exit(Reason, State) ->
             ?LOG_WARN("Channel (~p) closing: server sent error ~p~n",
                       [self(), Reason]),
             {IsHard, Code, _} = ?PROTOCOL:lookup_amqp_exception(ErrorName),
-            {stop, {if IsHard -> server_initiated_hard_close;
-                       true   -> server_initiated_close
-                    end, Code, Expl}, State};
+            {stop, if IsHard -> {connection_closing,
+                                 {server_initiated_hard_close, Code, Expl}};
+                      true   -> {server_initiated_close, Code, Expl}
+                   end, State};
         %% Unexpected death of a channel infrastructure process
         _ ->
             {stop, {infrastructure_died, Reason}, State}
     end.
 
+handle_shutdown({_, 200, _}, State) ->
+    {stop, normal, State};
+handle_shutdown({connection_closing, normal}, State) ->
+    {stop, normal, State};
+handle_shutdown(Reason, State) ->
+    {stop, Reason, State}.
+
 %%---------------------------------------------------------------------------
 %% Internal plumbing
 %%---------------------------------------------------------------------------
@@ -728,7 +738,7 @@ build_content(#amqp_msg{props = Props, payload = Payload}) ->
 
 check_block(_Method, _AmqpMsg, #state{closing = just_channel}) ->
     closing;
-check_block(_Method, _AmqpMsg, #state{closing = connection}) ->
+check_block(_Method, _AmqpMsg, #state{closing = {connection, _}}) ->
     closing;
 check_block(_Method, none, #state{}) ->
     ok;

deps/amqp_client/src/amqp_channels_manager.erl

@@ -22,7 +22,7 @@
 -behaviour(gen_server).
 
 -export([start_link/2, open_channel/3, set_channel_max/2, is_empty/1,
-         num_channels/1, pass_frame/3, signal_connection_closing/2]).
+         num_channels/1, pass_frame/3, signal_connection_closing/3]).
 -export([init/1, terminate/2, code_change/3, handle_call/3, handle_cast/2,
          handle_info/2]).
 
@@ -55,8 +55,8 @@ num_channels(ChMgr) ->
 pass_frame(ChMgr, ChNumber, Frame) ->
     gen_server:cast(ChMgr, {pass_frame, ChNumber, Frame}).
 
-signal_connection_closing(ChMgr, ChannelCloseType) ->
-    gen_server:cast(ChMgr, {connection_closing, ChannelCloseType}).
+signal_connection_closing(ChMgr, ChannelCloseType, Reason) ->
+    gen_server:cast(ChMgr, {connection_closing, ChannelCloseType, Reason}).
 
 %%---------------------------------------------------------------------------
 %% gen_server callbacks
@@ -83,8 +83,8 @@ handle_cast({set_channel_max, ChannelMax}, State) ->
     {noreply, State#state{channel_max = ChannelMax}};
 handle_cast({pass_frame, ChNumber, Frame}, State) ->
     {noreply, internal_pass_frame(ChNumber, Frame, State)};
-handle_cast({connection_closing, ChannelCloseType}, State) ->
-    handle_connection_closing(ChannelCloseType, State).
+handle_cast({connection_closing, ChannelCloseType, Reason}, State) ->
+    handle_connection_closing(ChannelCloseType, Reason, State).
 
 handle_info({'DOWN', _, process, Pid, Reason}, State) ->
     handle_down(Pid, Reason, State).
@@ -159,11 +159,12 @@ maybe_report_down(_Pid, {app_initiated_close, _, _}, _State) ->
     ok;
 maybe_report_down(_Pid, {server_initiated_close, _, _}, _State) ->
     ok;
-maybe_report_down(_Pid, connection_closing, _State) ->
-    ok;
-maybe_report_down(Pid, {server_initiated_hard_close, _, _} = Reason,
+maybe_report_down(Pid, {connection_closing,
+                        {server_initiated_hard_close, _, _} = Reason},
                   #state{connection = Connection}) ->
     amqp_gen_connection:hard_error_in_channel(Connection, Pid, Reason);
+maybe_report_down(_Pid, {connection_closing, _}, _State) ->
+    ok;
 maybe_report_down(_Pid, {server_misbehaved, AmqpError},
                   #state{connection = Connection}) ->
     amqp_gen_connection:server_misbehaved(Connection, AmqpError);
@@ -179,11 +180,12 @@ check_all_channels_terminated(State = #state{closing = true,
         false -> ok
     end.
 
-handle_connection_closing(ChannelCloseType,
+handle_connection_closing(ChannelCloseType, Reason,
                           State = #state{connection = Connection}) ->
     case internal_is_empty(State) of
         true  -> amqp_gen_connection:channels_terminated(Connection);
-        false -> signal_channels_connection_closing(ChannelCloseType, State)
+        false -> signal_channels_connection_closing(ChannelCloseType, Reason,
+                                                    State)
     end,
     {noreply, State#state{closing = true}}.
 
@@ -233,7 +235,7 @@ internal_lookup_pn(Pid, #state{map_pid_num = MapPN}) ->
 internal_update_npa(Number, Pid, AState, State = #state{map_num_pa = MapNPA}) ->
     State#state{map_num_pa = gb_trees:update(Number, {Pid, AState}, MapNPA)}.
 
-signal_channels_connection_closing(ChannelCloseType,
+signal_channels_connection_closing(ChannelCloseType, Reason,
                                    #state{map_pid_num = MapPN}) ->
-    [amqp_channel:connection_closing(Pid, ChannelCloseType)
+    [amqp_channel:connection_closing(Pid, ChannelCloseType, Reason)
         || Pid <- dict:fetch_keys(MapPN)].

deps/amqp_client/src/amqp_gen_connection.erl

@@ -295,13 +295,15 @@ server_misbehaved_close(AmqpError, State) ->
 set_closing_state(ChannelCloseType, NewClosing,
                   State = #state{channels_manager = ChMgr,
                                  closing = CurClosing}) ->
-    amqp_channels_manager:signal_connection_closing(ChMgr, ChannelCloseType),
     ResClosing =
         case closing_priority(NewClosing) =< closing_priority(CurClosing) of
             true  -> NewClosing;
             false -> CurClosing
         end,
-    callback(closing, [ChannelCloseType, closing_to_reason(ResClosing)],
+    ClosingReason = closing_to_reason(ResClosing),
+    amqp_channels_manager:signal_connection_closing(ChMgr, ChannelCloseType,
+                                                    ClosingReason),
+    callback(closing, [ChannelCloseType, ClosingReason],
              State#state{closing = ResClosing}).
 
 closing_priority(false)                                     -> 99;

deps/amqp_client/test/negative_test_util.erl

@@ -67,23 +67,24 @@ hard_error_test(Connection) ->
     try amqp_channel:call(Channel, Qos) of
         _ -> exit(expected_to_exit)
     catch
-        exit:{connection_closing, _} ->
-            %% Network case
+        %% Network case
+        exit:{{connection_closing,
+               {server_initiated_close, ?NOT_IMPLEMENTED, _}}, _} ->
             ok;
-        exit:Reason ->
-            %% Direct case
-            %% TODO: fix error code in the direct case
-            ?assertMatch({{server_initiated_hard_close, ?NOT_IMPLEMENTED, _}, _},
-                         Reason)
+        %% Direct case
+        exit:{{connection_closing,
+               {server_initiated_hard_close, ?NOT_IMPLEMENTED, _}}, _} ->
+            ok
     end,
-    receive {'DOWN', OtherChannelMonitor, process, OtherChannel, OtherExit} ->
-        case OtherExit of
-            %% Direct case
-            %% TODO fix error code in the direct case
-            killed -> ok;
-            %% Network case
-            _      -> ?assertMatch(connection_closing, OtherExit)
-        end
+    receive
+        %% Direct case
+        %% TODO fix error code in the direct case
+        {'DOWN', OtherChannelMonitor, process, OtherChannel, killed} ->
+            ok;
+        {'DOWN', OtherChannelMonitor, process, OtherChannel, OtherExit} ->
+            ?assertMatch({connection_closing,
+                          {server_initiated_close, ?NOT_IMPLEMENTED, _}},
+                         OtherExit)
     end,
     test_util:wait_for_death(Channel),
     test_util:wait_for_death(Connection).

deps/amqp_client/test/test_util.erl

@@ -279,12 +279,12 @@ channel_multi_open_close_test(Connection) ->
                                           closing            -> ok
                                       catch
                                           exit:{noproc, _}             -> ok;
-                                          exit:{connection_closing, _} -> ok
+                                          exit:{normal, _} -> ok
                                       end;
                 closing            -> ok
             catch
                 exit:{noproc, _}             -> ok;
-                exit:{connection_closing, _} -> ok
+                exit:{normal, _} -> ok
             end
         end) || _ <- lists:seq(1, 50)],
     erlang:yield(),