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}