001    /**
002     * Licensed to the Apache Software Foundation (ASF) under one or more
003     * contributor license agreements.  See the NOTICE file distributed with
004     * this work for additional information regarding copyright ownership.
005     * The ASF licenses this file to You under the Apache License, Version 2.0
006     * (the "License"); you may not use this file except in compliance with
007     * the License.  You may obtain a copy of the License at
008     *
009     *      http://www.apache.org/licenses/LICENSE-2.0
010     *
011     * Unless required by applicable law or agreed to in writing, software
012     * distributed under the License is distributed on an "AS IS" BASIS,
013     * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
014     * See the License for the specific language governing permissions and
015     * limitations under the License.
016     */
017    package org.apache.activemq.network;
018    
019    import java.io.IOException;
020    import java.util.Iterator;
021    
022    import org.apache.activemq.command.ActiveMQDestination;
023    import org.apache.activemq.command.ConsumerId;
024    import org.apache.activemq.command.ConsumerInfo;
025    import org.apache.activemq.filter.DestinationFilter;
026    import org.apache.activemq.transport.Transport;
027    import org.slf4j.Logger;
028    import org.slf4j.LoggerFactory;
029    
030    /**
031     * Consolidates subscriptions
032     * 
033     * 
034     */
035    public class DurableConduitBridge extends ConduitBridge {
036        private static final Logger LOG = LoggerFactory.getLogger(DurableConduitBridge.class);
037    
038        /**
039         * Constructor
040         * 
041         * @param configuration
042         * 
043         * @param localBroker
044         * @param remoteBroker
045         */
046        public DurableConduitBridge(NetworkBridgeConfiguration configuration, Transport localBroker,
047                                    Transport remoteBroker) {
048            super(configuration, localBroker, remoteBroker);
049        }
050    
051        /**
052         * Subscriptions for these destinations are always created
053         * 
054         */
055        protected void setupStaticDestinations() {
056            super.setupStaticDestinations();
057            ActiveMQDestination[] dests = configuration.isDynamicOnly() ? null : durableDestinations;
058            if (dests != null) {
059                for (int i = 0; i < dests.length; i++) {
060                    ActiveMQDestination dest = dests[i];
061                    if (isPermissableDestination(dest) && !doesConsumerExist(dest)) {
062                        DemandSubscription sub = createDemandSubscription(dest);
063                        if (dest.isTopic()) {
064                            sub.getLocalInfo().setSubscriptionName(getSubscriberName(dest));
065                        }
066                        try {
067                            addSubscription(sub);
068                        } catch (IOException e) {
069                            LOG.error("Failed to add static destination " + dest, e);
070                        }
071                        if (LOG.isTraceEnabled()) {
072                            LOG.trace("Forwarding messages for durable destination: " + dest);
073                        }
074                    }
075                }
076            }
077        }
078    
079        protected DemandSubscription createDemandSubscription(ConsumerInfo info) throws IOException {
080            if (addToAlreadyInterestedConsumers(info)) {
081                return null; // don't want this subscription added
082            }
083            //add our original id to ourselves
084            info.addNetworkConsumerId(info.getConsumerId());
085    
086            if (info.isDurable()) {
087                // set the subscriber name to something reproducible
088                info.setSubscriptionName(getSubscriberName(info.getDestination()));
089                // and override the consumerId with something unique so that it won't
090                // be removed if the durable subscriber (at the other end) goes away
091                info.setConsumerId(new ConsumerId(localSessionInfo.getSessionId(), consumerIdGenerator
092                        .getNextSequenceId()));
093            }
094            info.setSelector(null);
095            return doCreateDemandSubscription(info);
096        }
097    
098        protected String getSubscriberName(ActiveMQDestination dest) {
099            String subscriberName = DURABLE_SUB_PREFIX + configuration.getBrokerName() + "_" + dest.getPhysicalName();
100            return subscriberName;
101        }
102    
103        protected boolean doesConsumerExist(ActiveMQDestination dest) {
104            DestinationFilter filter = DestinationFilter.parseFilter(dest);
105            for (Iterator i = subscriptionMapByLocalId.values().iterator(); i.hasNext();) {
106                DemandSubscription ds = (DemandSubscription)i.next();
107                if (filter.matches(ds.getLocalInfo().getDestination())) {
108                    return true;
109                }
110            }
111            return false;
112        }
113    }