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.util;
018
019 import java.util.ArrayList;
020 import java.util.List;
021
022 import org.apache.camel.AsyncCallback;
023 import org.apache.camel.AsyncProcessor;
024 import org.apache.camel.DelegateProcessor;
025 import org.apache.camel.Exchange;
026 import org.apache.camel.Navigate;
027 import org.apache.camel.Processor;
028 import org.apache.camel.Service;
029
030 /**
031 * A simple converter that can convert any {@link Processor} to an {@link AsyncProcessor}.
032 * Processing will still occur synchronously but it will provide the required
033 * notifications that the caller expects.
034 *
035 * @version
036 */
037 public final class AsyncProcessorConverterHelper {
038
039 private AsyncProcessorConverterHelper() {
040 // Helper class
041 }
042
043 /**
044 * Creates a {@link AsyncProcessor} that delegates to the given processor.
045 * It is important that this implements {@link DelegateProcessor}
046 */
047 private static final class ProcessorToAsyncProcessorBridge implements DelegateProcessor, AsyncProcessor, Navigate<Processor>, Service {
048 protected Processor processor;
049
050 private ProcessorToAsyncProcessorBridge(Processor processor) {
051 this.processor = processor;
052 }
053
054 public boolean process(Exchange exchange, AsyncCallback callback) {
055 if (processor == null) {
056 // no processor then we are done
057 callback.done(true);
058 return true;
059 }
060 try {
061 processor.process(exchange);
062 } catch (Throwable e) {
063 // must catch throwable so we catch all
064 exchange.setException(e);
065 } finally {
066 // we are bridging a sync processor as async so callback with true
067 callback.done(true);
068 }
069 return true;
070 }
071
072 @Override
073 public String toString() {
074 if (processor != null) {
075 return processor.toString();
076 } else {
077 return "Processor is null";
078 }
079 }
080
081 public void process(Exchange exchange) throws Exception {
082 processNext(exchange);
083 }
084
085 protected void processNext(Exchange exchange) throws Exception {
086 if (processor != null) {
087 processor.process(exchange);
088 }
089 }
090
091 public void start() throws Exception {
092 ServiceHelper.startServices(processor);
093 }
094
095 public void stop() throws Exception {
096 ServiceHelper.stopServices(processor);
097 }
098
099 public boolean hasNext() {
100 return processor != null;
101 }
102
103 public List<Processor> next() {
104 if (!hasNext()) {
105 return null;
106 }
107 List<Processor> answer = new ArrayList<Processor>(1);
108 answer.add(processor);
109 return answer;
110 }
111
112 @Override
113 public Processor getProcessor() {
114 return processor;
115 }
116 }
117
118 public static AsyncProcessor convert(Processor value) {
119 if (value instanceof AsyncProcessor) {
120 return (AsyncProcessor)value;
121 }
122 return new ProcessorToAsyncProcessorBridge(value);
123 }
124 }