ARTEMIS-3753 Prevent sending message to internal queues on mirror

artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/mirror/AMQPMirrorControllerSource.java

@@ -201,6 +201,13 @@ public void sendMessage(Message message, RoutingContext context, List<MessageRef
          return;
       }
 
+      if (context.isInternal()) {
+         if (logger.isTraceEnabled()) {
+            logger.trace("server " + server + " is discarding send to avoid sending to internal queue");
+         }
+         return;
+      }
+
       if (logger.isTraceEnabled()) {
          logger.trace(server + " send message " + message);
       }

artemis-server/src/main/java/org/apache/activemq/artemis/core/server/RoutingContext.java

@@ -41,6 +41,9 @@ public interface RoutingContext {
     *  to avoid*/
    boolean isMirrorController();
 
+   /** return true if every queue routed is internal */
+   boolean isInternal();
+
    MirrorController getMirrorSource();
 
    RoutingContext setMirrorSource(MirrorController mirrorController);

artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/RoutingContextImpl.java

@@ -59,6 +59,8 @@ public class RoutingContextImpl implements RoutingContext {
 
    Boolean reusable = null;
 
+   Boolean internalOnly = null;
+
    volatile int version;
 
    private final Executor executor;
@@ -95,6 +97,11 @@ public boolean isReusable() {
       return reusable != null && reusable;
    }
 
+   @Override
+   public boolean isInternal() {
+      return internalOnly != null && internalOnly;
+   }
+
    @Override
    public int getPreviousBindingsVersion() {
       return version;
@@ -138,6 +145,8 @@ public RoutingContext clear() {
 
       this.reusable = null;
 
+      this.internalOnly = null;
+
       return this;
    }
 
@@ -163,6 +172,13 @@ public void addQueue(final SimpleString address, final Queue queue) {
          listing.getNonDurableQueues().add(queue);
       }
 
+      if (internalOnly == null) {
+         internalOnly = true;
+      }
+
+      // every queue added has to be internal only
+      internalOnly = internalOnly && queue.isInternalQueue();
+
       queueCount++;
    }