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 */
017package org.apache.activemq.broker.region.virtual;
018
019import java.io.IOException;
020import java.util.Set;
021import java.util.concurrent.CountDownLatch;
022import java.util.concurrent.atomic.AtomicReference;
023import org.apache.activemq.broker.Broker;
024import org.apache.activemq.broker.BrokerService;
025import org.apache.activemq.broker.ConnectionContext;
026import org.apache.activemq.broker.ProducerBrokerExchange;
027import org.apache.activemq.broker.region.Destination;
028import org.apache.activemq.broker.region.DestinationFilter;
029import org.apache.activemq.broker.region.Topic;
030import org.apache.activemq.command.ActiveMQDestination;
031import org.apache.activemq.command.ActiveMQQueue;
032import org.apache.activemq.command.ConnectionId;
033import org.apache.activemq.command.LocalTransactionId;
034import org.apache.activemq.command.Message;
035import org.apache.activemq.util.LRUCache;
036
037/**
038 * A Destination which implements <a href="http://activemq.org/site/virtual-destinations.html">Virtual Topic</a>
039 */
040public class VirtualTopicInterceptor extends DestinationFilter {
041
042    private final String prefix;
043    private final String postfix;
044    private final boolean local;
045    private final boolean concurrentSend;
046    private final LRUCache<ActiveMQDestination, ActiveMQQueue> cache = new LRUCache<ActiveMQDestination, ActiveMQQueue>();
047
048    public VirtualTopicInterceptor(Destination next, VirtualTopic virtualTopic) {
049        super(next);
050        this.prefix = virtualTopic.getPrefix();
051        this.postfix = virtualTopic.getPostfix();
052        this.local = virtualTopic.isLocal();
053        this.concurrentSend = virtualTopic.isConcurrentSend();
054    }
055
056    public Topic getTopic() {
057        return (Topic) this.next;
058    }
059
060    @Override
061    public void send(ProducerBrokerExchange context, Message message) throws Exception {
062        if (!message.isAdvisory() && !(local && message.getBrokerPath() != null)) {
063            ActiveMQDestination queueConsumers = getQueueConsumersWildcard(message.getDestination());
064            send(context, message, queueConsumers);
065        }
066        super.send(context, message);
067    }
068
069    @Override
070    protected void send(final ProducerBrokerExchange context, final Message message, ActiveMQDestination destination) throws Exception {
071        final Broker broker = context.getConnectionContext().getBroker();
072        final Set<Destination> destinations = broker.getDestinations(destination);
073        final int numDestinations = destinations.size();
074
075        final LocalTransactionId localBrokerTransactionToCoalesceJournalSync =
076                beginLocalTransaction(numDestinations, context.getConnectionContext(), message);
077        try {
078            if (concurrentSend && numDestinations > 1) {
079
080                final CountDownLatch concurrent = new CountDownLatch(destinations.size());
081                final AtomicReference<Exception> exceptionAtomicReference = new AtomicReference<Exception>();
082                final BrokerService brokerService = broker.getBrokerService();
083
084                for (final Destination dest : destinations) {
085                    if (shouldDispatch(broker, message, dest)) {
086                        brokerService.getTaskRunnerFactory().execute(new Runnable() {
087                            @Override
088                            public void run() {
089                                try {
090                                    if (exceptionAtomicReference.get() == null) {
091                                        dest.send(context, message.copy());
092                                    }
093                                } catch (Exception e) {
094                                    exceptionAtomicReference.set(e);
095                                } finally {
096                                    concurrent.countDown();
097                                }
098                            }
099                        });
100                    } else {
101                        concurrent.countDown();
102                    }
103                }
104                concurrent.await();
105                if (exceptionAtomicReference.get() != null) {
106                    throw exceptionAtomicReference.get();
107                }
108
109            } else {
110                for (final Destination dest : destinations) {
111                    if (shouldDispatch(broker, message, dest)) {
112                        dest.send(context, message.copy());
113                    }
114                }
115            }
116        } finally {
117            commit(localBrokerTransactionToCoalesceJournalSync, context.getConnectionContext(), message);
118        }
119    }
120
121    private LocalTransactionId beginLocalTransaction(int numDestinations, ConnectionContext connectionContext, Message message) throws Exception {
122        LocalTransactionId result = null;
123        if (numDestinations > 1 && message.isPersistent() && message.getTransactionId() == null) {
124            result = new LocalTransactionId(new ConnectionId(message.getMessageId().getProducerId().toString()), message.getMessageId().getProducerSequenceId());
125            connectionContext.getBroker().beginTransaction(connectionContext, result);
126            connectionContext.setTransaction(connectionContext.getTransactions().get(result));
127            message.setTransactionId(result);
128        }
129        return result;
130    }
131
132    private void commit(LocalTransactionId tx, ConnectionContext connectionContext, Message message) throws Exception {
133        if (tx != null) {
134            connectionContext.getBroker().commitTransaction(connectionContext, tx, true);
135            connectionContext.getTransactions().remove(tx);
136            connectionContext.setTransaction(null);
137            message.setTransactionId(null);
138        }
139    }
140
141    protected boolean shouldDispatch(Broker broker, Message message, Destination dest) throws IOException {
142        return true;
143    }
144
145    protected ActiveMQDestination getQueueConsumersWildcard(ActiveMQDestination original) {
146        ActiveMQQueue queue;
147        synchronized (cache) {
148            queue = cache.get(original);
149            if (queue == null) {
150                queue = new ActiveMQQueue(prefix + original.getPhysicalName() + postfix);
151                cache.put(original, queue);
152            }
153        }
154        return queue;
155    }
156}