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.Iterator;
021 import java.util.List;
022
023 import org.apache.camel.AsyncCallback;
024 import org.apache.camel.AsyncProcessor;
025 import org.apache.camel.Exchange;
026 import org.apache.camel.Navigate;
027 import org.apache.camel.Processor;
028 import org.apache.camel.Traceable;
029 import org.apache.camel.support.ServiceSupport;
030 import org.apache.camel.util.AsyncProcessorConverterHelper;
031 import org.apache.camel.util.AsyncProcessorHelper;
032 import org.apache.camel.util.ServiceHelper;
033 import org.slf4j.Logger;
034 import org.slf4j.LoggerFactory;
035
036 import static org.apache.camel.processor.PipelineHelper.continueProcessing;
037
038 /**
039 * Implements a Choice structure where one or more predicates are used which if
040 * they are true their processors are used, with a default otherwise clause used
041 * if none match.
042 *
043 * @version
044 */
045 public class ChoiceProcessor extends ServiceSupport implements AsyncProcessor, Navigate<Processor>, Traceable {
046 private static final Logger LOG = LoggerFactory.getLogger(ChoiceProcessor.class);
047 private final List<Processor> filters;
048 private final Processor otherwise;
049
050 public ChoiceProcessor(List<Processor> filters, Processor otherwise) {
051 this.filters = filters;
052 this.otherwise = otherwise;
053 }
054
055 public void process(Exchange exchange) throws Exception {
056 AsyncProcessorHelper.process(this, exchange);
057 }
058
059 public boolean process(final Exchange exchange, final AsyncCallback callback) {
060 Iterator<Processor> processors = next().iterator();
061
062 // callback to restore existing FILTER_MATCHED property on the Exchange
063 final Object existing = exchange.getProperty(Exchange.FILTER_MATCHED);
064 final AsyncCallback choiceCallback = new AsyncCallback() {
065 @Override
066 public void done(boolean doneSync) {
067 if (existing != null) {
068 exchange.setProperty(Exchange.FILTER_MATCHED, existing);
069 } else {
070 exchange.removeProperty(Exchange.FILTER_MATCHED);
071 }
072 callback.done(doneSync);
073 }
074 };
075
076 // as we only pick one processor to process, then no need to have async callback that has a while loop as well
077 // as this should not happen, eg we pick the first filter processor that matches, or the otherwise (if present)
078 // and if not, we just continue without using any processor
079 while (processors.hasNext()) {
080 // get the next processor
081 Processor processor = processors.next();
082
083 // evaluate the predicate on filter predicate early to be faster
084 // and avoid issues when having nested choices
085 // as we should only pick one processor
086 boolean matches = true;
087 if (processor instanceof FilterProcessor) {
088 FilterProcessor filter = (FilterProcessor) processor;
089 try {
090 matches = filter.getPredicate().matches(exchange);
091 exchange.setProperty(Exchange.FILTER_MATCHED, matches);
092 // as we have pre evaluated the predicate then use its processor directly when routing
093 processor = filter.getProcessor();
094 } catch (Throwable e) {
095 exchange.setException(e);
096 }
097 }
098
099 // check for error if so we should break out
100 if (!continueProcessing(exchange, "so breaking out of choice", LOG)) {
101 break;
102 }
103
104 // if we did not match then continue to next filter
105 if (!matches) {
106 continue;
107 }
108
109 // okay we found a filter or its the otherwise we are processing
110 AsyncProcessor async = AsyncProcessorConverterHelper.convert(processor);
111 return async.process(exchange, choiceCallback);
112 }
113
114 // when no filter matches and there is no otherwise, then just continue
115 choiceCallback.done(true);
116 return true;
117 }
118
119 @Override
120 public String toString() {
121 StringBuilder builder = new StringBuilder("choice{");
122 boolean first = true;
123 for (Processor processor : filters) {
124 if (first) {
125 first = false;
126 } else {
127 builder.append(", ");
128 }
129 builder.append("when ");
130 builder.append(processor);
131 }
132 if (otherwise != null) {
133 builder.append(", otherwise: ");
134 builder.append(otherwise);
135 }
136 builder.append("}");
137 return builder.toString();
138 }
139
140 public String getTraceLabel() {
141 return "choice";
142 }
143
144 public List<Processor> getFilters() {
145 return filters;
146 }
147
148 public Processor getOtherwise() {
149 return otherwise;
150 }
151
152 public List<Processor> next() {
153 if (!hasNext()) {
154 return null;
155 }
156 List<Processor> answer = new ArrayList<Processor>();
157 if (filters != null) {
158 answer.addAll(filters);
159 }
160 if (otherwise != null) {
161 answer.add(otherwise);
162 }
163 return answer;
164 }
165
166 public boolean hasNext() {
167 return otherwise != null || (filters != null && !filters.isEmpty());
168 }
169
170 protected void doStart() throws Exception {
171 ServiceHelper.startServices(filters, otherwise);
172 }
173
174 protected void doStop() throws Exception {
175 ServiceHelper.stopServices(otherwise, filters);
176 }
177 }