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.Collection; 020import java.util.Iterator; 021import java.util.LinkedList; 022import java.util.concurrent.CountDownLatch; 023import java.util.concurrent.atomic.AtomicReference; 024 025import org.apache.activemq.broker.Broker; 026import org.apache.activemq.broker.BrokerService; 027import org.apache.activemq.broker.ProducerBrokerExchange; 028import org.apache.activemq.broker.region.Destination; 029import org.apache.activemq.broker.region.DestinationFilter; 030import org.apache.activemq.command.ActiveMQDestination; 031import org.apache.activemq.command.Message; 032import org.apache.activemq.filter.MessageEvaluationContext; 033import org.apache.activemq.filter.NonCachedMessageEvaluationContext; 034 035/** 036 * Represents a composite {@link Destination} where send()s are replicated to 037 * each Destination instance. 038 * 039 * 040 */ 041public class CompositeDestinationFilter extends DestinationFilter { 042 043 private Collection forwardDestinations; 044 private boolean forwardOnly; 045 private boolean copyMessage; 046 private boolean concurrentSend = false; 047 048 public CompositeDestinationFilter(Destination next, Collection forwardDestinations, boolean forwardOnly, boolean copyMessage, boolean concurrentSend) { 049 super(next); 050 this.forwardDestinations = forwardDestinations; 051 this.forwardOnly = forwardOnly; 052 this.copyMessage = copyMessage; 053 this.concurrentSend = concurrentSend; 054 } 055 056 public void send(final ProducerBrokerExchange context, final Message message) throws Exception { 057 MessageEvaluationContext messageContext = null; 058 059 Collection<ActiveMQDestination> matchingDestinations = new LinkedList<ActiveMQDestination>(); 060 for (Iterator iter = forwardDestinations.iterator(); iter.hasNext();) { 061 ActiveMQDestination destination = null; 062 Object value = iter.next(); 063 064 if (value instanceof FilteredDestination) { 065 FilteredDestination filteredDestination = (FilteredDestination)value; 066 if (messageContext == null) { 067 messageContext = new NonCachedMessageEvaluationContext(); 068 messageContext.setMessageReference(message); 069 } 070 messageContext.setDestination(filteredDestination.getDestination()); 071 if (filteredDestination.matches(messageContext)) { 072 destination = filteredDestination.getDestination(); 073 } 074 } else if (value instanceof ActiveMQDestination) { 075 destination = (ActiveMQDestination)value; 076 } 077 if (destination == null) { 078 continue; 079 } 080 matchingDestinations.add(destination); 081 } 082 083 final CountDownLatch concurrent = new CountDownLatch(concurrentSend ? matchingDestinations.size() : 0); 084 final AtomicReference<Exception> exceptionAtomicReference = new AtomicReference<Exception>(); 085 final BrokerService brokerService = context.getConnectionContext().getBroker().getBrokerService(); 086 for (final ActiveMQDestination destination : matchingDestinations) { 087 if (concurrent.getCount() > 0) { 088 brokerService.getTaskRunnerFactory().execute(new Runnable() { 089 @Override 090 public void run() { 091 try { 092 if (exceptionAtomicReference.get() == null) { 093 doForward(context.copy(), message, brokerService.getRegionBroker(), destination); 094 } 095 } catch (Exception e) { 096 exceptionAtomicReference.set(e); 097 } finally { 098 concurrent.countDown(); 099 } 100 } 101 }); 102 } else { 103 doForward(context, message, brokerService.getRegionBroker(), destination); 104 } 105 } 106 if (!forwardOnly) { 107 super.send(context, message); 108 } 109 concurrent.await(); 110 if (exceptionAtomicReference.get() != null) { 111 throw exceptionAtomicReference.get(); 112 } 113 } 114 115 private void doForward(ProducerBrokerExchange context, Message message, Broker regionBroker, ActiveMQDestination destination) throws Exception { 116 Message forwarded_message; 117 if (copyMessage) { 118 forwarded_message = message.copy(); 119 forwarded_message.setDestination(destination); 120 } 121 else { 122 forwarded_message = message; 123 } 124 125 // Send it back through the region broker for routing. 126 context.setMutable(true); 127 regionBroker.send(context, forwarded_message); 128 } 129}