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.camel.processor;
018
019import java.util.Iterator;
020
021import org.apache.camel.AsyncCallback;
022import org.apache.camel.AsyncProcessor;
023import org.apache.camel.AsyncProducerCallback;
024import org.apache.camel.CamelContext;
025import org.apache.camel.Endpoint;
026import org.apache.camel.ErrorHandlerFactory;
027import org.apache.camel.Exchange;
028import org.apache.camel.ExchangePattern;
029import org.apache.camel.Expression;
030import org.apache.camel.FailedToCreateProducerException;
031import org.apache.camel.Message;
032import org.apache.camel.Processor;
033import org.apache.camel.Producer;
034import org.apache.camel.Traceable;
035import org.apache.camel.builder.ExpressionBuilder;
036import org.apache.camel.impl.DefaultExchange;
037import org.apache.camel.impl.EmptyProducerCache;
038import org.apache.camel.impl.ProducerCache;
039import org.apache.camel.spi.EndpointUtilizationStatistics;
040import org.apache.camel.spi.IdAware;
041import org.apache.camel.spi.RouteContext;
042import org.apache.camel.support.ServiceSupport;
043import org.apache.camel.util.AsyncProcessorHelper;
044import org.apache.camel.util.ExchangeHelper;
045import org.apache.camel.util.MessageHelper;
046import org.apache.camel.util.ObjectHelper;
047import org.apache.camel.util.ServiceHelper;
048import org.slf4j.Logger;
049import org.slf4j.LoggerFactory;
050
051import static org.apache.camel.processor.PipelineHelper.continueProcessing;
052import static org.apache.camel.util.ObjectHelper.notNull;
053
054/**
055 * Implements a <a href="http://camel.apache.org/routing-slip.html">Routing Slip</a>
056 * pattern where the list of actual endpoints to send a message exchange to are
057 * dependent on the value of a message header.
058 * <p/>
059 * This implementation mirrors the logic from the {@link org.apache.camel.processor.Pipeline} in the async variation
060 * as the failover load balancer is a specialized pipeline. So the trick is to keep doing the same as the
061 * pipeline to ensure it works the same and the async routing engine is flawless.
062 */
063public class RoutingSlip extends ServiceSupport implements AsyncProcessor, Traceable, IdAware {
064    protected final Logger log = LoggerFactory.getLogger(getClass());
065    protected String id;
066    protected ProducerCache producerCache;
067    protected int cacheSize;
068    protected boolean ignoreInvalidEndpoints;
069    protected String header;
070    protected Expression expression;
071    protected String uriDelimiter;
072    protected final CamelContext camelContext;
073
074    /**
075     * The iterator to be used for retrieving the next routing slip(s) to be used.
076     */
077    protected interface RoutingSlipIterator {
078
079        /**
080         * Are the more routing slip(s)?
081         *
082         * @param exchange the current exchange
083         * @return <tt>true</tt> if more slips, <tt>false</tt> otherwise.
084         */
085        boolean hasNext(Exchange exchange);
086
087        /**
088         * Returns the next routing slip(s).
089         *
090         * @param exchange the current exchange
091         * @return the slip(s).
092         */
093        Object next(Exchange exchange);
094
095    }
096
097    public RoutingSlip(CamelContext camelContext) {
098        notNull(camelContext, "camelContext");
099        this.camelContext = camelContext;
100    }
101
102    public RoutingSlip(CamelContext camelContext, Expression expression, String uriDelimiter) {
103        notNull(camelContext, "camelContext");
104        notNull(expression, "expression");
105        
106        this.camelContext = camelContext;
107        this.expression = expression;
108        this.uriDelimiter = uriDelimiter;
109        this.header = null;
110    }
111
112    public String getId() {
113        return id;
114    }
115
116    public void setId(String id) {
117        this.id = id;
118    }
119
120    public Expression getExpression() {
121        return expression;
122    }
123
124    public String getUriDelimiter() {
125        return uriDelimiter;
126    }
127
128    public void setDelimiter(String delimiter) {
129        this.uriDelimiter = delimiter;
130    }
131    
132    public boolean isIgnoreInvalidEndpoints() {
133        return ignoreInvalidEndpoints;
134    }
135    
136    public void setIgnoreInvalidEndpoints(boolean ignoreInvalidEndpoints) {
137        this.ignoreInvalidEndpoints = ignoreInvalidEndpoints;
138    }
139
140    public int getCacheSize() {
141        return cacheSize;
142    }
143
144    public void setCacheSize(int cacheSize) {
145        this.cacheSize = cacheSize;
146    }
147
148    @Override
149    public String toString() {
150        return "RoutingSlip[expression=" + expression + " uriDelimiter=" + uriDelimiter + "]";
151    }
152
153    public String getTraceLabel() {
154        return "routingSlip[" + expression + "]";
155    }
156
157    public void process(Exchange exchange) throws Exception {
158        AsyncProcessorHelper.process(this, exchange);
159    }
160
161    public boolean process(Exchange exchange, AsyncCallback callback) {
162        if (!isStarted()) {
163            exchange.setException(new IllegalStateException("RoutingSlip has not been started: " + this));
164            callback.done(true);
165            return true;
166        }
167
168        return doRoutingSlipWithExpression(exchange, this.expression, callback);
169    }
170
171    public boolean doRoutingSlip(Exchange exchange, Object routingSlip, AsyncCallback callback) {
172        if (routingSlip instanceof Expression) {
173            return doRoutingSlipWithExpression(exchange, (Expression) routingSlip, callback);
174        } else {
175            return doRoutingSlipWithExpression(exchange, ExpressionBuilder.constantExpression(routingSlip), callback);
176        }
177    }
178
179    /**
180     * Creates the route slip iterator to be used.
181     *
182     * @param exchange the exchange
183     * @param expression the expression
184     * @return the iterator, should never be <tt>null</tt>
185     */
186    protected RoutingSlipIterator createRoutingSlipIterator(final Exchange exchange, final Expression expression) throws Exception {
187        Object slip = expression.evaluate(exchange, Object.class);
188        if (exchange.getException() != null) {
189            // force any exceptions occurred during evaluation to be thrown
190            throw exchange.getException();
191        }
192
193        final Iterator<Object> delegate = ObjectHelper.createIterator(slip, uriDelimiter);
194
195        return new RoutingSlipIterator() {
196            public boolean hasNext(Exchange exchange) {
197                return delegate.hasNext();
198            }
199
200            public Object next(Exchange exchange) {
201                return delegate.next();
202            }
203        };
204    }
205
206    private boolean doRoutingSlipWithExpression(final Exchange exchange, final Expression expression, final AsyncCallback callback) {
207        Exchange current = exchange;
208        RoutingSlipIterator iter;
209        try {
210            iter = createRoutingSlipIterator(exchange, expression);
211        } catch (Exception e) {
212            exchange.setException(e);
213            callback.done(true);
214            return true;
215        }
216
217        // ensure the slip is empty when we start
218        if (current.hasProperties()) {
219            current.setProperty(Exchange.SLIP_ENDPOINT, null);
220        }
221
222        while (iter.hasNext(current)) {
223            Endpoint endpoint;
224            try {
225                endpoint = resolveEndpoint(iter, exchange);
226                // if no endpoint was resolved then try the next
227                if (endpoint == null) {
228                    continue;
229                }
230            } catch (Exception e) {
231                // error resolving endpoint so we should break out
232                current.setException(e);
233                break;
234            }
235
236            //process and prepare the routing slip
237            boolean sync = processExchange(endpoint, current, exchange, callback, iter);
238            current = prepareExchangeForRoutingSlip(current, endpoint);
239            
240            if (!sync) {
241                log.trace("Processing exchangeId: {} is continued being processed asynchronously", exchange.getExchangeId());
242                // the remainder of the routing slip will be completed async
243                // so we break out now, then the callback will be invoked which then continue routing from where we left here
244                return false;
245            }
246
247            log.trace("Processing exchangeId: {} is continued being processed synchronously", exchange.getExchangeId());
248
249            // we ignore some kind of exceptions and allow us to continue
250            if (isIgnoreInvalidEndpoints()) {
251                FailedToCreateProducerException e = current.getException(FailedToCreateProducerException.class);
252                if (e != null) {
253                    if (log.isDebugEnabled()) {
254                        log.debug("Endpoint uri is invalid: " + endpoint + ". This exception will be ignored.", e);
255                    }
256                    current.setException(null);
257                }
258            }
259
260            // Decide whether to continue with the recipients or not; similar logic to the Pipeline
261            // check for error if so we should break out
262            if (!continueProcessing(current, "so breaking out of the routing slip", log)) {
263                break;
264            }
265        }
266
267        // logging nextExchange as it contains the exchange that might have altered the payload and since
268        // we are logging the completion if will be confusing if we log the original instead
269        // we could also consider logging the original and the nextExchange then we have *before* and *after* snapshots
270        log.trace("Processing complete for exchangeId: {} >>> {}", exchange.getExchangeId(), current);
271
272        // copy results back to the original exchange
273        ExchangeHelper.copyResults(exchange, current);
274
275        callback.done(true);
276        return true;
277    }
278
279    protected Endpoint resolveEndpoint(RoutingSlipIterator iter, Exchange exchange) throws Exception {
280        Object nextRecipient = iter.next(exchange);
281        Endpoint endpoint = null;
282        try {
283            endpoint = ExchangeHelper.resolveEndpoint(exchange, nextRecipient);
284        } catch (Exception e) {
285            if (isIgnoreInvalidEndpoints()) {
286                log.info("Endpoint uri is invalid: " + nextRecipient + ". This exception will be ignored.", e);
287            } else {
288                throw e;
289            }
290        }
291        return endpoint;
292    }
293
294    protected Exchange prepareExchangeForRoutingSlip(Exchange current, Endpoint endpoint) {
295        Exchange copy = new DefaultExchange(current);
296        // we must use the same id as this is a snapshot strategy where Camel copies a snapshot
297        // before processing the next step in the pipeline, so we have a snapshot of the exchange
298        // just before. This snapshot is used if Camel should do redeliveries (re try) using
299        // DeadLetterChannel. That is why it's important the id is the same, as it is the *same*
300        // exchange being routed.
301        copy.setExchangeId(current.getExchangeId());
302        copyOutToIn(copy, current);
303
304        // ensure stream caching is reset
305        MessageHelper.resetStreamCache(copy.getIn());
306
307        return copy;
308    }
309
310    protected AsyncProcessor createErrorHandler(RouteContext routeContext, Exchange exchange, AsyncProcessor processor, Endpoint endpoint) {
311        AsyncProcessor answer = processor;
312
313        boolean tryBlock = exchange.getProperty(Exchange.TRY_ROUTE_BLOCK, false, boolean.class);
314
315        // do not wrap in error handler if we are inside a try block
316        if (!tryBlock && routeContext != null) {
317            // wrap the producer in error handler so we have fine grained error handling on
318            // the output side instead of the input side
319            // this is needed to support redelivery on that output alone and not doing redelivery
320            // for the entire routingslip/dynamic-router block again which will start from scratch again
321
322            log.trace("Creating error handler for: {}", processor);
323            ErrorHandlerFactory builder = routeContext.getRoute().getErrorHandlerBuilder();
324            // create error handler (create error handler directly to keep it light weight,
325            // instead of using ProcessorDefinition.wrapInErrorHandler)
326            try {
327                answer = (AsyncProcessor) builder.createErrorHandler(routeContext, processor);
328
329                // must start the error handler
330                ServiceHelper.startServices(answer);
331
332            } catch (Exception e) {
333                throw ObjectHelper.wrapRuntimeCamelException(e);
334            }
335        }
336
337        return answer;
338    }
339
340    protected boolean processExchange(final Endpoint endpoint, final Exchange exchange, final Exchange original,
341                                      final AsyncCallback callback, final RoutingSlipIterator iter) {
342
343        // this does the actual processing so log at trace level
344        log.trace("Processing exchangeId: {} >>> {}", exchange.getExchangeId(), exchange);
345
346        boolean sync = producerCache.doInAsyncProducer(endpoint, exchange, null, callback, new AsyncProducerCallback() {
347            public boolean doInAsyncProducer(Producer producer, AsyncProcessor asyncProducer, final Exchange exchange,
348                                             ExchangePattern exchangePattern, final AsyncCallback callback) {
349
350                // rework error handling to support fine grained error handling
351                RouteContext routeContext = exchange.getUnitOfWork() != null ? exchange.getUnitOfWork().getRouteContext() : null;
352                AsyncProcessor target = createErrorHandler(routeContext, exchange, asyncProducer, endpoint);
353
354                // set property which endpoint we send to
355                exchange.setProperty(Exchange.TO_ENDPOINT, endpoint.getEndpointUri());
356                exchange.setProperty(Exchange.SLIP_ENDPOINT, endpoint.getEndpointUri());
357
358                boolean answer = target.process(exchange, new AsyncCallback() {
359                    public void done(boolean doneSync) {
360                        // we only have to handle async completion of the routing slip
361                        if (doneSync) {
362                            callback.done(doneSync);
363                            return;
364                        }
365
366                        // continue processing the routing slip asynchronously
367                        Exchange current = prepareExchangeForRoutingSlip(exchange, endpoint);
368
369                        while (iter.hasNext(current)) {
370
371                            // we ignore some kind of exceptions and allow us to continue
372                            if (isIgnoreInvalidEndpoints()) {
373                                FailedToCreateProducerException e = current.getException(FailedToCreateProducerException.class);
374                                if (e != null) {
375                                    if (log.isDebugEnabled()) {
376                                        log.debug("Endpoint uri is invalid: " + endpoint + ". This exception will be ignored.", e);
377                                    }
378                                    current.setException(null);
379                                }
380                            }
381
382                            // Decide whether to continue with the recipients or not; similar logic to the Pipeline
383                            // check for error if so we should break out
384                            if (!continueProcessing(current, "so breaking out of the routing slip", log)) {
385                                break;
386                            }
387
388                            Endpoint endpoint;
389                            try {
390                                endpoint = resolveEndpoint(iter, exchange);
391                                // if no endpoint was resolved then try the next
392                                if (endpoint == null) {
393                                    continue;
394                                }
395                            } catch (Exception e) {
396                                // error resolving endpoint so we should break out
397                                exchange.setException(e);
398                                break;
399                            }
400
401                            // prepare and process the routing slip
402                            boolean sync = processExchange(endpoint, current, original, callback, iter);
403                            current = prepareExchangeForRoutingSlip(current, endpoint);
404
405                            if (!sync) {
406                                log.trace("Processing exchangeId: {} is continued being processed asynchronously", original.getExchangeId());
407                                return;
408                            }
409                        }
410
411                        // logging nextExchange as it contains the exchange that might have altered the payload and since
412                        // we are logging the completion if will be confusing if we log the original instead
413                        // we could also consider logging the original and the nextExchange then we have *before* and *after* snapshots
414                        log.trace("Processing complete for exchangeId: {} >>> {}", original.getExchangeId(), current);
415
416                        // copy results back to the original exchange
417                        ExchangeHelper.copyResults(original, current);
418
419                        if (target instanceof DeadLetterChannel) {
420                            Processor deadLetter = ((DeadLetterChannel) target).getDeadLetter();
421                            try {
422                                ServiceHelper.stopService(deadLetter);
423                            } catch (Exception e) {
424                                log.warn("Error stopping DeadLetterChannel error handler on routing slip. This exception is ignored.", e);
425                            }
426                        }
427
428                        callback.done(false);
429                    }
430                });
431
432                // stop error handler if we completed synchronously
433                if (answer) {
434                    if (target instanceof DeadLetterChannel) {
435                        Processor deadLetter = ((DeadLetterChannel) target).getDeadLetter();
436                        try {
437                            ServiceHelper.stopService(deadLetter);
438                        } catch (Exception e) {
439                            log.warn("Error stopping DeadLetterChannel error handler on routing slip. This exception is ignored.", e);
440                        }
441                    }
442                }
443
444                return answer;
445            }
446        });
447
448        return sync;
449    }
450
451    protected void doStart() throws Exception {
452        if (producerCache == null) {
453            if (cacheSize < 0) {
454                producerCache = new EmptyProducerCache(this, camelContext);
455                log.debug("RoutingSlip {} is not using ProducerCache", this);
456            } else if (cacheSize == 0) {
457                producerCache = new ProducerCache(this, camelContext);
458                log.debug("RoutingSlip {} using ProducerCache with default cache size", this);
459            } else {
460                producerCache = new ProducerCache(this, camelContext, cacheSize);
461                log.debug("RoutingSlip {} using ProducerCache with cacheSize={}", this, cacheSize);
462            }
463        }
464        ServiceHelper.startService(producerCache);
465    }
466
467    protected void doStop() throws Exception {
468        ServiceHelper.stopServices(producerCache);
469    }
470
471    protected void doShutdown() throws Exception {
472        ServiceHelper.stopAndShutdownServices(producerCache);
473    }
474
475    public EndpointUtilizationStatistics getEndpointUtilizationStatistics() {
476        return producerCache.getEndpointUtilizationStatistics();
477    }
478
479    /**
480     * Returns the outbound message if available. Otherwise return the inbound message.
481     */
482    private Message getResultMessage(Exchange exchange) {
483        if (exchange.hasOut()) {
484            return exchange.getOut();
485        } else {
486            // if this endpoint had no out (like a mock endpoint) just take the in
487            return exchange.getIn();
488        }
489    }
490
491    /**
492     * Copy the outbound data in 'source' to the inbound data in 'result'.
493     */
494    private void copyOutToIn(Exchange result, Exchange source) {
495        result.setException(source.getException());
496        result.setIn(getResultMessage(source));
497
498        result.getProperties().clear();
499        result.getProperties().putAll(source.getProperties());
500    }
501}