Shovel AMQP091 bugfix: basic.cancel is a source message

It is the source side of the shovel that must handle the basic.cancel
message from the source queue. This fix makes 091-10 and 091-local work,
handling the basic.cancel and restarting the shovel when required.

(cherry picked from commit 402e5b7)

Author: dcorbacho

deps/rabbitmq_shovel/src/rabbit_amqp091_shovel.erl

@@ -229,6 +229,10 @@ handle_source({#'basic.deliver'{delivery_tag = Tag,
     % forward to destination
     rabbit_shovel_behaviour:forward(Tag, Msg, State);
 
+handle_source(#'basic.cancel'{}, #{name := Name}) ->
+    ?LOG_WARNING("Shovel ~tp received a 'basic.cancel' from the server", [Name]),
+    {stop, {shutdown, restart}};
+
 handle_source({'EXIT', Conn, Reason},
               #{source := #{current := {Conn, _, _}}}) ->
     {stop, {inbound_conn_died, Reason}};
@@ -251,10 +255,6 @@ handle_dest(#'basic.nack'{delivery_tag = Seq, multiple = Multiple},
                                rabbit_shovel_behaviour:nack(Tag, Multi, StateX)
                        end, Seq, Multiple, State);
 
-handle_dest(#'basic.cancel'{}, #{name := Name}) ->
-    ?LOG_WARNING("Shovel ~tp received a 'basic.cancel' from the server", [Name]),
-    {stop, {shutdown, restart}};
-
 handle_dest({'EXIT', Conn, Reason}, #{dest := #{current := {Conn, _, _}}}) ->
     {stop, {outbound_conn_died, Reason}};