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 */ 039public class CompositeDestinationFilter extends DestinationFilter { 040 041 private Collection forwardDestinations; 042 private boolean forwardOnly; 043 private boolean concurrentSend = false; 044 045 public CompositeDestinationFilter(Destination next, Collection forwardDestinations, boolean forwardOnly, boolean concurrentSend) { 046 super(next); 047 this.forwardDestinations = forwardDestinations; 048 this.forwardOnly = forwardOnly; 049 this.concurrentSend = concurrentSend; 050 } 051 052 @Override 053 public void send(final ProducerBrokerExchange context, final Message message) throws Exception { 054 MessageEvaluationContext messageContext = null; 055 056 Collection<ActiveMQDestination> matchingDestinations = new LinkedList<ActiveMQDestination>(); 057 for (Iterator iter = forwardDestinations.iterator(); iter.hasNext();) { 058 ActiveMQDestination destination = null; 059 Object value = iter.next(); 060 061 if (value instanceof FilteredDestination) { 062 FilteredDestination filteredDestination = (FilteredDestination)value; 063 if (messageContext == null) { 064 messageContext = new NonCachedMessageEvaluationContext(); 065 messageContext.setMessageReference(message); 066 } 067 messageContext.setDestination(filteredDestination.getDestination()); 068 if (filteredDestination.matches(messageContext)) { 069 destination = filteredDestination.getDestination(); 070 } 071 } else if (value instanceof ActiveMQDestination) { 072 destination = (ActiveMQDestination)value; 073 } 074 if (destination == null) { 075 continue; 076 } 077 matchingDestinations.add(destination); 078 } 079 080 final CountDownLatch concurrent = new CountDownLatch(concurrentSend ? matchingDestinations.size() : 0); 081 final AtomicReference<Exception> exceptionAtomicReference = new AtomicReference<Exception>(); 082 final BrokerService brokerService = context.getConnectionContext().getBroker().getBrokerService(); 083 for (final ActiveMQDestination destination : matchingDestinations) { 084 if (concurrent.getCount() > 0) { 085 brokerService.getTaskRunnerFactory().execute(new Runnable() { 086 @Override 087 public void run() { 088 try { 089 if (exceptionAtomicReference.get() == null) { 090 doForward(context.copy(), message, brokerService.getRegionBroker(), destination); 091 } 092 } catch (Exception e) { 093 exceptionAtomicReference.set(e); 094 } finally { 095 concurrent.countDown(); 096 } 097 } 098 }); 099 } else { 100 doForward(context, message, brokerService.getRegionBroker(), destination); 101 } 102 } 103 if (!forwardOnly) { 104 super.send(context, message); 105 } 106 concurrent.await(); 107 if (exceptionAtomicReference.get() != null) { 108 throw exceptionAtomicReference.get(); 109 } 110 } 111 112 private void doForward(ProducerBrokerExchange context, Message message, Broker regionBroker, ActiveMQDestination destination) throws Exception { 113 Message forwardedMessage = message.copy(); 114 115 forwardedMessage.setOriginalDestination( message.getDestination() ); 116 forwardedMessage.setDestination(destination); 117 118 // Send it back through the region broker for routing. 119 context.setMutable(true); 120 regionBroker.send(context, forwardedMessage); 121 } 122}