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.List;
021import java.util.Set;
022
023import org.apache.activemq.broker.Broker;
024import org.apache.activemq.broker.ProducerBrokerExchange;
025import org.apache.activemq.broker.region.Destination;
026import org.apache.activemq.broker.region.Subscription;
027import org.apache.activemq.command.ActiveMQDestination;
028import org.apache.activemq.command.Message;
029import org.apache.activemq.filter.MessageEvaluationContext;
030import org.apache.activemq.filter.NonCachedMessageEvaluationContext;
031
032public class SelectorAwareVirtualTopicInterceptor extends VirtualTopicInterceptor {
033
034    public SelectorAwareVirtualTopicInterceptor(Destination next, String prefix, String postfix) {
035        super(next, prefix, postfix);
036    }
037
038    /**
039     * Respect the selectors of the subscriptions to ensure only matched messages are dispatched to
040     * the virtual queues, hence there is no build up of unmatched messages on these destinations
041     */
042    @Override
043    protected void send(ProducerBrokerExchange context, Message message, ActiveMQDestination destination) throws Exception {
044        Broker broker = context.getConnectionContext().getBroker();
045        Set<Destination> destinations = broker.getDestinations(destination);
046
047        for (Destination dest : destinations) {
048            if (matchesSomeConsumer(message, dest)) {
049                dest.send(context, message.copy());
050            }
051        }
052    }
053    
054    private boolean matchesSomeConsumer(Message message, Destination dest) throws IOException {
055        boolean matches = false;
056        MessageEvaluationContext msgContext = new NonCachedMessageEvaluationContext();
057        msgContext.setDestination(dest.getActiveMQDestination());
058        msgContext.setMessageReference(message);
059        List<Subscription> subs = dest.getConsumers();
060        for (Subscription sub: subs) {
061            if (sub.matches(message, msgContext)) {
062                matches = true;
063                break;
064            }
065        }
066        return matches;
067    }
068}