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 org.apache.camel.AsyncCallback;
020 import org.apache.camel.AsyncProcessor;
021 import org.apache.camel.Exchange;
022 import org.apache.camel.Processor;
023 import org.apache.camel.support.ServiceSupport;
024 import org.apache.camel.util.AsyncProcessorConverterHelper;
025 import org.apache.camel.util.ServiceHelper;
026
027 /**
028 * A bridge to have regular interceptors implemented as {@link org.apache.camel.Processor}
029 * work with the asynchronous routing engine without causing side effects.
030 *
031 * @version
032 */
033 public class InterceptorToAsyncProcessorBridge extends ServiceSupport implements AsyncProcessor {
034
035 private final AsyncProcessor interceptor;
036 private volatile AsyncProcessor target;
037 private volatile ThreadLocal<AsyncCallback> callback = new ThreadLocal<AsyncCallback>();
038 private volatile ThreadLocal<Boolean> interceptorDone = new ThreadLocal<Boolean>();
039
040 /**
041 * Constructs the bridge
042 *
043 * @param interceptor the interceptor to bridge
044 */
045 public InterceptorToAsyncProcessorBridge(Processor interceptor) {
046 this(interceptor, null);
047 }
048
049 /**
050 * Constructs the bridge
051 *
052 * @param interceptor the interceptor to bridge
053 * @param target the target
054 */
055 public InterceptorToAsyncProcessorBridge(Processor interceptor, AsyncProcessor target) {
056 this.interceptor = AsyncProcessorConverterHelper.convert(interceptor);
057 this.target = target;
058 }
059
060 /**
061 * Process invoked by the interceptor
062 * @param exchange the message exchange
063 * @throws Exception
064 */
065 public void process(Exchange exchange) throws Exception {
066 // invoke when interceptor wants to invoke
067 boolean done = interceptor.process(exchange, callback.get());
068 interceptorDone.set(done);
069 }
070
071 public boolean process(Exchange exchange, AsyncCallback callback) {
072 // remember the callback to be used by the interceptor
073 this.callback.set(callback);
074 try {
075 // invoke the target
076 boolean done = target.process(exchange, callback);
077 if (interceptorDone.get() != null) {
078 // return the result from the interceptor if it was invoked
079 return interceptorDone.get();
080 } else {
081 // otherwise from the target
082 return done;
083 }
084 } finally {
085 // cleanup
086 this.callback.remove();
087 this.interceptorDone.remove();
088 }
089 }
090
091 public void setTarget(Processor target) {
092 this.target = AsyncProcessorConverterHelper.convert(target);
093 }
094
095 @Override
096 public String toString() {
097 return "AsyncBridge[" + interceptor.toString() + "]";
098 }
099
100 @Override
101 protected void doStart() throws Exception {
102 ServiceHelper.startServices(target, interceptor);
103 }
104
105 @Override
106 protected void doStop() throws Exception {
107 callback.remove();
108 interceptorDone.remove();
109 ServiceHelper.stopServices(interceptor, target);
110 }
111 }