content: /**
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements.  See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You under the Apache License, Version 2.0
   * (the "License"); you may not use this file except in compliance with
   * the License.  You may obtain a copy of the License at
   *
   *      http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package org.apache.activemq.broker.region.virtual;
  
  import org.apache.activemq.broker.*;
  import org.apache.activemq.broker.region.Destination;
  import org.apache.activemq.broker.region.DestinationFilter;
  import org.apache.activemq.broker.region.DestinationInterceptor;
  import org.apache.activemq.command.ActiveMQDestination;
  import org.apache.activemq.command.ActiveMQTopic;
  import org.apache.activemq.command.Message;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  /**
   * Creates <a href="https://activemq.apache.org/mirrored-queues">Mirrored
   * Queue</a> using a prefix and postfix to define the topic name on which to mirror the queue to.
   *
   * 
   * @org.apache.xbean.XBean
   */
  public class MirroredQueue implements DestinationInterceptor, BrokerServiceAware {
      private static final transient Logger LOG = LoggerFactory.getLogger(MirroredQueue.class);
      private String prefix = "VirtualTopic.Mirror.";
      private String postfix = "";
      private boolean copyMessage = true;
      private BrokerService brokerService;
  
      public Destination intercept(final Destination destination) {
          if (destination.getActiveMQDestination().isQueue()) {
              if (!destination.getActiveMQDestination().isTemporary() || brokerService.isUseTempMirroredQueues()) {
                  try {
                      final Destination mirrorDestination = getMirrorDestination(destination);
                      if (mirrorDestination != null) {
                          return new DestinationFilter(destination) {
                              public void send(ProducerBrokerExchange context, Message message) throws Exception {
                                  message.setDestination(mirrorDestination.getActiveMQDestination());
                                  mirrorDestination.send(context, message);
      
                                  if (isCopyMessage()) {
                                      message = message.copy();
                                  }
                                  message.setDestination(destination.getActiveMQDestination());
                                  message.setMemoryUsage(null); // set this to null so that it will use the queue memoryUsage instance instead of the topic.
                                  super.send(context, message);
                              }
                          };
                      }
                  }
                  catch (Exception e) {
                      LOG.error("Failed to lookup the mirror destination for: {}", destination, e);
                  }
              }
          }
          return destination;
      }
      
  
      public void remove(Destination destination) {
          if (brokerService == null) {
              throw new IllegalArgumentException("No brokerService injected!");
          }
          ActiveMQDestination topic = getMirrorTopic(destination.getActiveMQDestination());
          if (topic != null) {
              try {
                  brokerService.removeDestination(topic);
              } catch (Exception e) {
                  LOG.error("Failed to remove mirror destination for {}", destination, e);
              }
          }
          
      }
  
      public void create(Broker broker, ConnectionContext context, ActiveMQDestination destination) {}
  
      // Properties
      // -------------------------------------------------------------------------
  
      public String getPostfix() {
          return postfix;
      }
  
      /**
       * Sets any postix used to identify the queue consumers
       */
      public void setPostfix(String postfix) {
          this.postfix = postfix;
      }
  
      public String getPrefix() {
          return prefix;
      }
  
      /**
       * Sets the prefix wildcard used to identify the queue consumers for a given
       * topic
       */
      public void setPrefix(String prefix) {
          this.prefix = prefix;
      }
  
      public boolean isCopyMessage() {
          return copyMessage;
      }
  
      /**
       * Sets whether a copy of the message will be sent to each destination.
       * Defaults to true so that the forward destination is set as the
       * destination of the message
       */
      public void setCopyMessage(boolean copyMessage) {
          this.copyMessage = copyMessage;
      }
  
      public void setBrokerService(BrokerService brokerService) {
          this.brokerService = brokerService;
      }
  
      // Implementation methods
      //-------------------------------------------------------------------------
      protected Destination getMirrorDestination(Destination destination) throws Exception {
          if (brokerService == null) {
              throw new IllegalArgumentException("No brokerService injected!");
          }
          ActiveMQDestination topic = getMirrorTopic(destination.getActiveMQDestination());
          return brokerService.getDestination(topic);
      }
  
      protected ActiveMQDestination getMirrorTopic(ActiveMQDestination original) {
          return new ActiveMQTopic(prefix + original.getPhysicalName() + postfix);
      }
  
  }
  