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.util.regex.Matcher;
020import java.util.regex.Pattern;
021
022import org.apache.activemq.broker.Broker;
023import org.apache.activemq.broker.ConnectionContext;
024import org.apache.activemq.broker.region.Destination;
025import org.apache.activemq.command.ActiveMQDestination;
026import org.apache.activemq.command.ActiveMQQueue;
027import org.apache.activemq.command.ActiveMQTopic;
028import org.apache.activemq.filter.DestinationFilter;
029
030/**
031 * Creates <a href="http://activemq.org/site/virtual-destinations.html">Virtual
032 * Topics</a> using a prefix and postfix. The virtual destination creates a
033 * wildcard that is then used to look up all active queue subscriptions which
034 * match.
035 *
036 * @org.apache.xbean.XBean
037 */
038public class VirtualTopic implements VirtualDestination {
039
040    private String prefix = "Consumer.*.";
041    private String postfix = "";
042    private String name = ">";
043    private boolean selectorAware = false;
044    private boolean local = false;
045    private boolean concurrentSend = false;
046
047    @Override
048    public ActiveMQDestination getVirtualDestination() {
049        return new ActiveMQTopic(getName());
050    }
051
052    @Override
053    public Destination intercept(Destination destination) {
054        return selectorAware ? new SelectorAwareVirtualTopicInterceptor(destination, this) :
055                new VirtualTopicInterceptor(destination, this);
056    }
057
058    @Override
059    public ActiveMQDestination getMappedDestinations() {
060        return new ActiveMQQueue(prefix + name + postfix);
061    }
062
063    @Override
064    public Destination interceptMappedDestination(Destination destination) {
065        // do a reverse map from destination to get actual virtual destination
066        final String physicalName = destination.getActiveMQDestination().getPhysicalName();
067        final Pattern pattern = Pattern.compile(getRegex(prefix) + "(.*)" + getRegex(postfix));
068        final Matcher matcher = pattern.matcher(physicalName);
069        if (matcher.matches()) {
070            final String virtualName = matcher.group(1);
071            return new MappedQueueFilter(new ActiveMQTopic(virtualName), destination);
072        }
073        return destination;
074    }
075
076    private String getRegex(String part) {
077        StringBuilder builder = new StringBuilder();
078        for (char c : part.toCharArray()) {
079            switch (c) {
080                case '.':
081                    builder.append("\\.");
082                    break;
083                case '*':
084                    builder.append("[^\\.]*");
085                    break;
086                default:
087                    builder.append(c);
088            }
089        }
090        return builder.toString();
091    }
092
093    @Override
094    public void create(Broker broker, ConnectionContext context, ActiveMQDestination destination) throws Exception {
095        if (destination.isQueue() && destination.isPattern()) {
096            DestinationFilter filter = DestinationFilter.parseFilter(new ActiveMQQueue(prefix + DestinationFilter.ANY_DESCENDENT));
097            if (filter.matches(destination)) {
098                broker.addDestination(context, destination, false);
099
100            }
101        }
102    }
103
104    @Override
105    public void remove(Destination destination) {
106    }
107
108    // Properties
109    // -------------------------------------------------------------------------
110
111    public String getPostfix() {
112        return postfix;
113    }
114
115    /**
116     * Sets any postix used to identify the queue consumers
117     */
118    public void setPostfix(String postfix) {
119        this.postfix = postfix;
120    }
121
122    public String getPrefix() {
123        return prefix;
124    }
125
126    /**
127     * Sets the prefix wildcard used to identify the queue consumers for a given
128     * topic
129     */
130    public void setPrefix(String prefix) {
131        this.prefix = prefix;
132    }
133
134    public String getName() {
135        return name;
136    }
137
138    public void setName(String name) {
139        this.name = name;
140    }
141
142    /**
143     * Indicates whether the selectors of consumers are used to determine
144     * dispatch to a virtual destination, when true only messages matching an
145     * existing consumer will be dispatched.
146     *
147     * @param selectorAware
148     *            when true take consumer selectors into consideration
149     */
150    public void setSelectorAware(boolean selectorAware) {
151        this.selectorAware = selectorAware;
152    }
153
154    public boolean isSelectorAware() {
155        return selectorAware;
156    }
157
158    public boolean isLocal() {
159        return local;
160    }
161
162    public void setLocal(boolean local) {
163        this.local = local;
164    }
165
166    @Override
167    public String toString() {
168        return new StringBuilder("VirtualTopic:").append(prefix).append(',').append(name).append(',').
169                                                  append(postfix).append(',').append(selectorAware).
170                                                  append(',').append(local).toString();
171    }
172
173    public boolean isConcurrentSend() {
174        return concurrentSend;
175    }
176
177    /**
178     * When true, dispatch to matching destinations in parallel (in multiple threads)
179     * @param concurrentSend
180     */
181    public void setConcurrentSend(boolean concurrentSend) {
182        this.concurrentSend = concurrentSend;
183    }
184}