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 */ 017 package org.apache.camel.processor; 018 019 import java.io.Closeable; 020 import java.io.IOException; 021 import java.util.ArrayList; 022 import java.util.Collection; 023 import java.util.Collections; 024 import java.util.Iterator; 025 import java.util.List; 026 import java.util.Scanner; 027 import java.util.concurrent.ExecutorService; 028 029 import org.apache.camel.AsyncCallback; 030 import org.apache.camel.AsyncProcessor; 031 import org.apache.camel.CamelContext; 032 import org.apache.camel.Exchange; 033 import org.apache.camel.Expression; 034 import org.apache.camel.Message; 035 import org.apache.camel.Processor; 036 import org.apache.camel.RuntimeCamelException; 037 import org.apache.camel.Traceable; 038 import org.apache.camel.processor.aggregate.AggregationStrategy; 039 import org.apache.camel.processor.aggregate.UseOriginalAggregationStrategy; 040 import org.apache.camel.spi.RouteContext; 041 import org.apache.camel.util.ExchangeHelper; 042 import org.apache.camel.util.IOHelper; 043 import org.apache.camel.util.ObjectHelper; 044 import org.slf4j.Logger; 045 import org.slf4j.LoggerFactory; 046 047 import static org.apache.camel.util.ObjectHelper.notNull; 048 049 /** 050 * Implements a dynamic <a 051 * href="http://camel.apache.org/splitter.html">Splitter</a> pattern 052 * where an expression is evaluated to iterate through each of the parts of a 053 * message and then each part is then send to some endpoint. 054 * 055 * @version 056 */ 057 public class Splitter extends MulticastProcessor implements AsyncProcessor, Traceable { 058 private static final transient Logger LOG = LoggerFactory.getLogger(Splitter.class); 059 060 private final Expression expression; 061 062 public Splitter(CamelContext camelContext, Expression expression, Processor destination, AggregationStrategy aggregationStrategy) { 063 this(camelContext, expression, destination, aggregationStrategy, false, null, false, false, false, 0, null, false); 064 } 065 066 public Splitter(CamelContext camelContext, Expression expression, Processor destination, AggregationStrategy aggregationStrategy, 067 boolean parallelProcessing, ExecutorService executorService, boolean shutdownExecutorService, 068 boolean streaming, boolean stopOnException, long timeout, Processor onPrepare, boolean useSubUnitOfWork) { 069 super(camelContext, Collections.singleton(destination), aggregationStrategy, parallelProcessing, executorService, 070 shutdownExecutorService, streaming, stopOnException, timeout, onPrepare, useSubUnitOfWork); 071 this.expression = expression; 072 notNull(expression, "expression"); 073 notNull(destination, "destination"); 074 } 075 076 @Override 077 public String toString() { 078 return "Splitter[on: " + expression + " to: " + getProcessors().iterator().next() + " aggregate: " + getAggregationStrategy() + "]"; 079 } 080 081 @Override 082 public String getTraceLabel() { 083 return "split[" + expression + "]"; 084 } 085 086 @Override 087 public boolean process(Exchange exchange, final AsyncCallback callback) { 088 final AggregationStrategy strategy = getAggregationStrategy(); 089 090 // if no custom aggregation strategy is being used then fallback to keep the original 091 // and propagate exceptions which is done by a per exchange specific aggregation strategy 092 // to ensure it supports async routing 093 if (strategy == null) { 094 UseOriginalAggregationStrategy original = new UseOriginalAggregationStrategy(exchange, true); 095 setAggregationStrategyOnExchange(exchange, original); 096 } 097 098 return super.process(exchange, callback); 099 } 100 101 @Override 102 protected Iterable<ProcessorExchangePair> createProcessorExchangePairs(Exchange exchange) throws Exception { 103 Object value = expression.evaluate(exchange, Object.class); 104 if (exchange.getException() != null) { 105 // force any exceptions occurred during evaluation to be thrown 106 throw exchange.getException(); 107 } 108 109 Iterable<ProcessorExchangePair> answer; 110 if (isStreaming()) { 111 answer = createProcessorExchangePairsIterable(exchange, value); 112 } else { 113 answer = createProcessorExchangePairsList(exchange, value); 114 } 115 if (exchange.getException() != null) { 116 // force any exceptions occurred during creation of exchange paris to be thrown 117 // before returning the answer; 118 throw exchange.getException(); 119 } 120 121 return answer; 122 } 123 124 private Iterable<ProcessorExchangePair> createProcessorExchangePairsIterable(final Exchange exchange, final Object value) { 125 final Iterator<?> iterator = ObjectHelper.createIterator(value); 126 return new Iterable<ProcessorExchangePair>() { 127 // create a copy which we use as master to copy during splitting 128 // this avoids any side effect reflected upon the incoming exchange 129 private final Exchange copy = copyExchangeNoAttachments(exchange, true); 130 private final RouteContext routeContext = exchange.getUnitOfWork() != null ? exchange.getUnitOfWork().getRouteContext() : null; 131 132 public Iterator<ProcessorExchangePair> iterator() { 133 return new Iterator<ProcessorExchangePair>() { 134 private int index; 135 private boolean closed; 136 137 public boolean hasNext() { 138 if (closed) { 139 return false; 140 } 141 142 boolean answer = iterator.hasNext(); 143 if (!answer) { 144 // we are now closed 145 closed = true; 146 // nothing more so we need to close the expression value in case it needs to be 147 if (value instanceof Closeable) { 148 IOHelper.close((Closeable) value, value.getClass().getName(), LOG); 149 } else if (value instanceof Scanner) { 150 // special for Scanner as it does not implement Closeable 151 Scanner scanner = (Scanner) value; 152 scanner.close(); 153 154 IOException ioException = scanner.ioException(); 155 if (ioException != null) { 156 throw new RuntimeCamelException("Scanner aborted because of an IOException!", ioException); 157 } 158 } 159 } 160 return answer; 161 } 162 163 public ProcessorExchangePair next() { 164 Object part = iterator.next(); 165 // create a correlated copy as the new exchange to be routed in the splitter from the copy 166 // and do not share the unit of work 167 Exchange newExchange = ExchangeHelper.createCorrelatedCopy(copy, false); 168 // if we share unit of work, we need to prepare the child exchange 169 if (isShareUnitOfWork()) { 170 prepareSharedUnitOfWork(newExchange, copy); 171 } 172 if (part instanceof Message) { 173 newExchange.setIn((Message) part); 174 } else { 175 Message in = newExchange.getIn(); 176 in.setBody(part); 177 } 178 return createProcessorExchangePair(index++, getProcessors().iterator().next(), newExchange, routeContext); 179 } 180 181 public void remove() { 182 throw new UnsupportedOperationException("Remove is not supported by this iterator"); 183 } 184 }; 185 } 186 187 }; 188 } 189 190 private Iterable<ProcessorExchangePair> createProcessorExchangePairsList(Exchange exchange, Object value) { 191 List<ProcessorExchangePair> result = new ArrayList<ProcessorExchangePair>(); 192 193 // reuse iterable and add it to the result list 194 Iterable<ProcessorExchangePair> pairs = createProcessorExchangePairsIterable(exchange, value); 195 for (ProcessorExchangePair pair : pairs) { 196 result.add(pair); 197 } 198 199 return result; 200 } 201 202 @Override 203 protected void updateNewExchange(Exchange exchange, int index, Iterable<ProcessorExchangePair> allPairs, 204 Iterator<ProcessorExchangePair> it) { 205 // do not share unit of work 206 exchange.setUnitOfWork(null); 207 208 exchange.setProperty(Exchange.SPLIT_INDEX, index); 209 if (allPairs instanceof Collection) { 210 // non streaming mode, so we know the total size already 211 exchange.setProperty(Exchange.SPLIT_SIZE, ((Collection<?>) allPairs).size()); 212 } 213 if (it.hasNext()) { 214 exchange.setProperty(Exchange.SPLIT_COMPLETE, Boolean.FALSE); 215 } else { 216 exchange.setProperty(Exchange.SPLIT_COMPLETE, Boolean.TRUE); 217 // streaming mode, so set total size when we are complete based on the index 218 exchange.setProperty(Exchange.SPLIT_SIZE, index + 1); 219 } 220 } 221 222 @Override 223 protected Integer getExchangeIndex(Exchange exchange) { 224 return exchange.getProperty(Exchange.SPLIT_INDEX, Integer.class); 225 } 226 227 public Expression getExpression() { 228 return expression; 229 } 230 231 private static Exchange copyExchangeNoAttachments(Exchange exchange, boolean preserveExchangeId) { 232 Exchange answer = ExchangeHelper.createCopy(exchange, preserveExchangeId); 233 // we do not want attachments for the splitted sub-messages 234 answer.getIn().setAttachments(null); 235 return answer; 236 } 237 }