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.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.Exchange;
025 import org.apache.camel.Navigate;
026 import org.apache.camel.Processor;
027 import org.apache.camel.support.ServiceSupport;
028 import org.apache.camel.util.ServiceHelper;
029
030 /**
031 * A Delegate pattern which delegates synchronous processing to a nested {@link org.apache.camel.Processor} which can
032 * be useful for implementation inheritance when writing an {@link org.apache.camel.spi.Policy}
033 * <p/>
034 * <b>Important:</b> This implementation <b>does</b> support the asynchronous routing engine, <b>only</b>.
035 * if the logic in the {@link #process(org.apache.camel.Exchange)} does not invoke EIPs; as it forces using
036 * synchronous processing during the {@link #process(org.apache.camel.Exchange)} method call.
037 * If you are implementing a EIP pattern please use this as the delegate, for simple EIPs.
038 *
039 * @version
040 * @see DelegateAsyncProcessor
041 * @see org.apache.camel.processor.DelegateProcessor
042 */
043 public class DelegateSyncProcessor extends ServiceSupport implements org.apache.camel.DelegateProcessor, AsyncProcessor, Navigate<Processor> {
044 protected Processor processor;
045
046 public DelegateSyncProcessor(Processor processor) {
047 this.processor = processor;
048 }
049
050 @Override
051 public String toString() {
052 return "DelegateSync[" + processor + "]";
053 }
054
055 public Processor getProcessor() {
056 return processor;
057 }
058
059 @Override
060 public boolean process(Exchange exchange, AsyncCallback callback) {
061 // force calling the sync method
062 try {
063 processor.process(exchange);
064 } catch (Throwable e) {
065 // must catch throwable so we catch all
066 exchange.setException(e);
067 } finally {
068 // we are bridging a sync processor as async so callback with true
069 callback.done(true);
070 }
071 return true;
072 }
073
074 @Override
075 public void process(Exchange exchange) throws Exception {
076 processor.process(exchange);
077 }
078
079 @Override
080 public boolean hasNext() {
081 return processor != null;
082 }
083
084 @Override
085 public List<Processor> next() {
086 if (!hasNext()) {
087 return null;
088 }
089 List<Processor> answer = new ArrayList<Processor>(1);
090 answer.add(processor);
091 return answer;
092 }
093
094 @Override
095 protected void doStart() throws Exception {
096 ServiceHelper.startService(processor);
097 }
098
099 @Override
100 protected void doStop() throws Exception {
101 ServiceHelper.stopServices(processor);
102 }
103 }