@@ -70,6 +70,8 @@ public class AMQPMirrorControllerSource extends BasicMirrorController<Sender> im
7070 public static final Symbol QUEUE = Symbol .getSymbol ("x-opt-amq-mr-qu" );
7171 public static final Symbol BROKER_ID = Symbol .getSymbol ("x-opt-amq-bkr-id" );
7272 public static final SimpleString BROKER_ID_SIMPLE_STRING = SimpleString .of (BROKER_ID .toString ());
73+ public static final SimpleString NO_FORWARD_SOURCE = SimpleString .of ("x-opt-amq-no-forward-source" );
74+ public static final SimpleString RECEIVER_ID_FILTER = SimpleString .of ("x-opt-amq-receiver-id-filter" );
7375
7476 // Events:
7577 public static final Symbol ADD_ADDRESS = Symbol .getSymbol ("addAddress" );
@@ -89,9 +91,11 @@ public class AMQPMirrorControllerSource extends BasicMirrorController<Sender> im
8991 // Capabilities
9092 public static final Symbol MIRROR_CAPABILITY = Symbol .getSymbol ("amq.mirror" );
9193 public static final Symbol QPID_DISPATCH_WAYPOINT_CAPABILITY = Symbol .valueOf ("qd.waypoint" );
94+ public static final Symbol NO_FORWARD = Symbol .getSymbol ("amq.no.forward" );
9295
9396 public static final SimpleString INTERNAL_ID_EXTRA_PROPERTY = SimpleString .of (INTERNAL_ID .toString ());
9497 public static final SimpleString INTERNAL_BROKER_ID_EXTRA_PROPERTY = SimpleString .of (BROKER_ID .toString ());
98+ public static final SimpleString INTERNAL_NO_FORWARD = SimpleString .of (NO_FORWARD .toString ());
9599
96100 private static final ThreadLocal <RoutingContext > mirrorControlRouting = ThreadLocal .withInitial (() -> new RoutingContextImpl (null ));
97101
@@ -230,12 +234,17 @@ public void addAddress(AddressInfo addressInfo) throws Exception {
230234 public void deleteAddress (AddressInfo addressInfo ) throws Exception {
231235 logger .trace ("{} deleteAddress {}" , server , addressInfo );
232236
237+ if (isBlockedByNoForward ()) {
238+ return ;
239+ }
240+
233241 if (invalidTarget (getControllerInUse ()) || addressInfo .isInternal ()) {
234242 return ;
235243 }
236244 if (ignoreAddress (addressInfo .getName ())) {
237245 return ;
238246 }
247+
239248 if (deleteQueues ) {
240249 Message message = createMessage (addressInfo .getName (), null , DELETE_ADDRESS , null , addressInfo .toJSON ());
241250 routeMirrorCommand (server , message );
@@ -246,6 +255,10 @@ public void deleteAddress(AddressInfo addressInfo) throws Exception {
246255 public void createQueue (QueueConfiguration queueConfiguration ) throws Exception {
247256 logger .trace ("{} createQueue {}" , server , queueConfiguration );
248257
258+ if (isBlockedByNoForward ()) {
259+ return ;
260+ }
261+
249262 if (invalidTarget (getControllerInUse ()) || queueConfiguration .isInternal ()) {
250263 if (logger .isTraceEnabled ()) {
251264 logger .trace ("Rejecting ping pong on create {} as isInternal={} and mirror target = {}" , queueConfiguration , queueConfiguration .isInternal (), getControllerInUse ());
@@ -264,6 +277,7 @@ public void createQueue(QueueConfiguration queueConfiguration) throws Exception
264277 }
265278 return ;
266279 }
280+
267281 if (addQueues ) {
268282 Message message = createMessage (queueConfiguration .getAddress (), queueConfiguration .getName (), CREATE_QUEUE , null , queueConfiguration .toJSON ());
269283 routeMirrorCommand (server , message );
@@ -276,6 +290,10 @@ public void deleteQueue(SimpleString address, SimpleString queue) throws Excepti
276290 logger .trace ("{} deleteQueue {}/{}" , server , address , queue );
277291 }
278292
293+ if (isBlockedByNoForward ()) {
294+ return ;
295+ }
296+
279297 if (invalidTarget (getControllerInUse ())) {
280298 return ;
281299 }
@@ -310,6 +328,14 @@ private boolean invalidTarget(MirrorController controller) {
310328 return controller != null && sameNode (getRemoteMirrorId (), controller .getRemoteMirrorId ());
311329 }
312330
331+ private boolean isBlockedByNoForward () {
332+ return getControllerInUse () != null && getControllerInUse ().isNoForward ();
333+ }
334+
335+ private boolean isBlockedByNoForward (Message message ) {
336+ return isBlockedByNoForward () || Boolean .TRUE .equals (message .getBrokerProperty (INTERNAL_NO_FORWARD ));
337+ }
338+
313339 private boolean ignoreAddress (SimpleString address ) {
314340 if (address .startsWith (server .getConfiguration ().getManagementAddress ())) {
315341 return true ;
@@ -338,6 +364,11 @@ Message copyMessageForPaging(Message message) {
338364 public void sendMessage (Transaction tx , Message message , RoutingContext context ) {
339365 SimpleString address = context .getAddress (message );
340366
367+ if (isBlockedByNoForward (message )) {
368+ logger .trace ("sendMessage::server {} is discarding the message because its source is setting a noForward policy" , server );
369+ return ;
370+ }
371+
341372 if (context .isInternal ()) {
342373 logger .trace ("sendMessage::server {} is discarding send to avoid sending to internal queue" , server );
343374 return ;
@@ -353,6 +384,8 @@ public void sendMessage(Transaction tx, Message message, RoutingContext context)
353384 return ;
354385 }
355386
387+ logger .trace ("sendMessage::{} send message {}" , server , message );
388+
356389 try {
357390 context .setReusable (false );
358391
@@ -467,6 +500,28 @@ public static void validateProtocolData(ReferenceIDSupplier referenceIDSupplier,
467500 }
468501 }
469502
503+ /**
504+ * Checks if the message ref should be filtered or not.
505+ * @param ref the message to check
506+ * @return true if the INTERNAL_RECEIVER_ID_FILTER annotation of the message is set to a different value
507+ * than the remoteMirrorID, false otherwise.
508+ */
509+ public boolean filterMessage (MessageReference ref ) {
510+ Object filterID = ref .getMessage ().getAnnotation (RECEIVER_ID_FILTER );
511+ if (filterID != null ) {
512+ String remoteMirrorId = getRemoteMirrorId ();
513+ if (remoteMirrorId != null ) {
514+ if (remoteMirrorId .equals (filterID )) {
515+ return false ;
516+ } else {
517+ return true ;
518+ }
519+ }
520+ return false ;
521+ }
522+ return false ;
523+ }
524+
470525 /** This method will return the brokerID used by the message */
471526 private static String setProtocolData (ReferenceIDSupplier referenceIDSupplier , MessageReference ref ) {
472527 String brokerID = referenceIDSupplier .getServerID (ref );
@@ -543,6 +598,17 @@ public void preAcknowledge(final Transaction tx, final MessageReference ref, fin
543598 logger .trace ("preAcknowledge::tx={}, ref={}, reason={}" , tx , ref , reason );
544599 }
545600
601+ SimpleString noForwardSource = null ;
602+ if (Boolean .TRUE .equals (ref .getMessage ().getBooleanProperty (INTERNAL_NO_FORWARD ))) {
603+ noForwardSource = (SimpleString ) ref .getMessage ().getBrokerProperty (NO_FORWARD_SOURCE );
604+ String remoteMirrorId = getRemoteMirrorId ();
605+ if (remoteMirrorId != null ) {
606+ if (!SimpleString .of (remoteMirrorId ).equals (noForwardSource )) {
607+ return ;
608+ }
609+ }
610+ }
611+
546612 MirrorController controllerInUse = getControllerInUse ();
547613
548614 // Retried ACKs are not forwarded.
@@ -578,6 +644,9 @@ public void preAcknowledge(final Transaction tx, final MessageReference ref, fin
578644 String nodeID = idSupplier .getServerID (ref ); // notice the brokerID will be null for any message generated on this broker.
579645 long internalID = idSupplier .getID (ref );
580646 Message messageCommand = createMessage (ref .getQueue ().getAddress (), ref .getQueue ().getName (), POST_ACK , nodeID , internalID , reason );
647+ if (noForwardSource != null ) {
648+ messageCommand .setBrokerProperty (RECEIVER_ID_FILTER , noForwardSource );
649+ }
581650 if (sync ) {
582651 OperationContext operationContext ;
583652 operationContext = OperationContextImpl .getContext (server .getExecutorFactory ());
0 commit comments