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.impl;
018
019 import java.util.ArrayList;
020 import java.util.List;
021 import java.util.Map;
022 import java.util.concurrent.ConcurrentHashMap;
023
024 import org.apache.camel.CamelContext;
025 import org.apache.camel.Endpoint;
026 import org.apache.camel.Exchange;
027 import org.apache.camel.ExchangePattern;
028 import org.apache.camel.Message;
029 import org.apache.camel.MessageHistory;
030 import org.apache.camel.spi.Synchronization;
031 import org.apache.camel.spi.UnitOfWork;
032 import org.apache.camel.util.ExchangeHelper;
033 import org.apache.camel.util.ObjectHelper;
034
035 /**
036 * A default implementation of {@link Exchange}
037 *
038 * @version
039 */
040 public final class DefaultExchange implements Exchange {
041
042 protected final CamelContext context;
043 private Map<String, Object> properties;
044 private Message in;
045 private Message out;
046 private Exception exception;
047 private String exchangeId;
048 private UnitOfWork unitOfWork;
049 private ExchangePattern pattern;
050 private Endpoint fromEndpoint;
051 private String fromRouteId;
052 private List<Synchronization> onCompletions;
053
054 public DefaultExchange(CamelContext context) {
055 this(context, ExchangePattern.InOnly);
056 }
057
058 public DefaultExchange(CamelContext context, ExchangePattern pattern) {
059 this.context = context;
060 this.pattern = pattern;
061 }
062
063 public DefaultExchange(Exchange parent) {
064 this(parent.getContext(), parent.getPattern());
065 this.fromEndpoint = parent.getFromEndpoint();
066 this.fromRouteId = parent.getFromRouteId();
067 this.unitOfWork = parent.getUnitOfWork();
068 }
069
070 public DefaultExchange(Endpoint fromEndpoint) {
071 this(fromEndpoint, ExchangePattern.InOnly);
072 }
073
074 public DefaultExchange(Endpoint fromEndpoint, ExchangePattern pattern) {
075 this(fromEndpoint.getCamelContext(), pattern);
076 this.fromEndpoint = fromEndpoint;
077 }
078
079 @Override
080 public String toString() {
081 return "Exchange[" + (out == null ? in : out) + "]";
082 }
083
084 public Exchange copy() {
085 DefaultExchange exchange = new DefaultExchange(this);
086
087 if (hasProperties()) {
088 exchange.setProperties(safeCopy(getProperties()));
089 }
090
091 exchange.setIn(getIn().copy());
092 if (hasOut()) {
093 exchange.setOut(getOut().copy());
094 }
095 exchange.setException(getException());
096 return exchange;
097 }
098
099 @SuppressWarnings("unchecked")
100 private static Map<String, Object> safeCopy(Map<String, Object> properties) {
101 if (properties == null) {
102 return null;
103 }
104
105 Map<String, Object> answer = new ConcurrentHashMap<String, Object>(properties);
106
107 // safe copy message history using a defensive copy
108 List<MessageHistory> history = (List<MessageHistory>) answer.remove(Exchange.MESSAGE_HISTORY);
109 if (history != null) {
110 answer.put(Exchange.MESSAGE_HISTORY, new ArrayList<MessageHistory>(history));
111 }
112
113 return answer;
114 }
115
116 public CamelContext getContext() {
117 return context;
118 }
119
120 public Object getProperty(String name) {
121 if (properties != null) {
122 return properties.get(name);
123 }
124 return null;
125 }
126
127 public Object getProperty(String name, Object defaultValue) {
128 Object answer = getProperty(name);
129 return answer != null ? answer : defaultValue;
130 }
131
132 @SuppressWarnings("unchecked")
133 public <T> T getProperty(String name, Class<T> type) {
134 Object value = getProperty(name);
135 if (value == null) {
136 // lets avoid NullPointerException when converting to boolean for null values
137 if (boolean.class.isAssignableFrom(type)) {
138 return (T) Boolean.FALSE;
139 }
140 return null;
141 }
142
143 // eager same instance type test to avoid the overhead of invoking the type converter
144 // if already same type
145 if (type.isInstance(value)) {
146 return type.cast(value);
147 }
148
149 return ExchangeHelper.convertToType(this, type, value);
150 }
151
152 @SuppressWarnings("unchecked")
153 public <T> T getProperty(String name, Object defaultValue, Class<T> type) {
154 Object value = getProperty(name, defaultValue);
155 if (value == null) {
156 // lets avoid NullPointerException when converting to boolean for null values
157 if (boolean.class.isAssignableFrom(type)) {
158 return (T) Boolean.FALSE;
159 }
160 return null;
161 }
162
163 // eager same instance type test to avoid the overhead of invoking the type converter
164 // if already same type
165 if (type.isInstance(value)) {
166 return type.cast(value);
167 }
168
169 return ExchangeHelper.convertToType(this, type, value);
170 }
171
172 public void setProperty(String name, Object value) {
173 if (value != null) {
174 // avoid the NullPointException
175 getProperties().put(name, value);
176 } else {
177 // if the value is null, we just remove the key from the map
178 if (name != null) {
179 getProperties().remove(name);
180 }
181 }
182 }
183
184 public Object removeProperty(String name) {
185 if (!hasProperties()) {
186 return null;
187 }
188 return getProperties().remove(name);
189 }
190
191 public Map<String, Object> getProperties() {
192 if (properties == null) {
193 properties = new ConcurrentHashMap<String, Object>();
194 }
195 return properties;
196 }
197
198 public boolean hasProperties() {
199 return properties != null && !properties.isEmpty();
200 }
201
202 public void setProperties(Map<String, Object> properties) {
203 this.properties = properties;
204 }
205
206 public Message getIn() {
207 if (in == null) {
208 in = new DefaultMessage();
209 configureMessage(in);
210 }
211 return in;
212 }
213
214 public <T> T getIn(Class<T> type) {
215 Message in = getIn();
216
217 // eager same instance type test to avoid the overhead of invoking the type converter
218 // if already same type
219 if (type.isInstance(in)) {
220 return type.cast(in);
221 }
222
223 // fallback to use type converter
224 return context.getTypeConverter().convertTo(type, this, in);
225 }
226
227 public void setIn(Message in) {
228 this.in = in;
229 configureMessage(in);
230 }
231
232 public Message getOut() {
233 // lazy create
234 if (out == null) {
235 out = (in != null && in instanceof MessageSupport)
236 ? ((MessageSupport)in).newInstance() : new DefaultMessage();
237 configureMessage(out);
238 }
239 return out;
240 }
241
242 public <T> T getOut(Class<T> type) {
243 if (!hasOut()) {
244 return null;
245 }
246
247 Message out = getOut();
248
249 // eager same instance type test to avoid the overhead of invoking the type converter
250 // if already same type
251 if (type.isInstance(out)) {
252 return type.cast(out);
253 }
254
255 // fallback to use type converter
256 return context.getTypeConverter().convertTo(type, this, out);
257 }
258
259 public boolean hasOut() {
260 return out != null;
261 }
262
263 public void setOut(Message out) {
264 this.out = out;
265 configureMessage(out);
266 }
267
268 public Exception getException() {
269 return exception;
270 }
271
272 public <T> T getException(Class<T> type) {
273 return ObjectHelper.getException(type, exception);
274 }
275
276 public void setException(Throwable t) {
277 if (t == null) {
278 this.exception = null;
279 } else if (t instanceof Exception) {
280 this.exception = (Exception) t;
281 } else {
282 // wrap throwable into an exception
283 this.exception = ObjectHelper.wrapCamelExecutionException(this, t);
284 }
285 }
286
287 public ExchangePattern getPattern() {
288 return pattern;
289 }
290
291 public void setPattern(ExchangePattern pattern) {
292 this.pattern = pattern;
293 }
294
295 public Endpoint getFromEndpoint() {
296 return fromEndpoint;
297 }
298
299 public void setFromEndpoint(Endpoint fromEndpoint) {
300 this.fromEndpoint = fromEndpoint;
301 }
302
303 public String getFromRouteId() {
304 return fromRouteId;
305 }
306
307 public void setFromRouteId(String fromRouteId) {
308 this.fromRouteId = fromRouteId;
309 }
310
311 public String getExchangeId() {
312 if (exchangeId == null) {
313 exchangeId = createExchangeId();
314 }
315 return exchangeId;
316 }
317
318 public void setExchangeId(String id) {
319 this.exchangeId = id;
320 }
321
322 public boolean isFailed() {
323 return (hasOut() && getOut().isFault()) || getException() != null;
324 }
325
326 public boolean isTransacted() {
327 UnitOfWork uow = getUnitOfWork();
328 if (uow != null) {
329 return uow.isTransacted();
330 } else {
331 return false;
332 }
333 }
334
335 public Boolean isExternalRedelivered() {
336 Boolean answer = null;
337
338 // check property first, as the implementation details to know if the message
339 // was externally redelivered is message specific, and thus the message implementation
340 // could potentially change during routing, and therefore later we may not know if the
341 // original message was externally redelivered or not, therefore we store this detail
342 // as a exchange property to keep it around for the lifecycle of the exchange
343 if (hasProperties()) {
344 answer = getProperty(Exchange.EXTERNAL_REDELIVERED, null, Boolean.class);
345 }
346
347 if (answer == null) {
348 // lets avoid adding methods to the Message API, so we use the
349 // DefaultMessage to allow component specific messages to extend
350 // and implement the isExternalRedelivered method.
351 DefaultMessage msg = getIn(DefaultMessage.class);
352 if (msg != null) {
353 answer = msg.isTransactedRedelivered();
354 // store as property to keep around
355 setProperty(Exchange.EXTERNAL_REDELIVERED, answer);
356 }
357 }
358
359 return answer;
360 }
361
362 public boolean isRollbackOnly() {
363 return Boolean.TRUE.equals(getProperty(Exchange.ROLLBACK_ONLY)) || Boolean.TRUE.equals(getProperty(Exchange.ROLLBACK_ONLY_LAST));
364 }
365
366 public UnitOfWork getUnitOfWork() {
367 return unitOfWork;
368 }
369
370 public void setUnitOfWork(UnitOfWork unitOfWork) {
371 this.unitOfWork = unitOfWork;
372 if (onCompletions != null) {
373 // now an unit of work has been assigned so add the on completions
374 // we might have registered already
375 for (Synchronization onCompletion : onCompletions) {
376 unitOfWork.addSynchronization(onCompletion);
377 }
378 // cleanup the temporary on completion list as they now have been registered
379 // on the unit of work
380 onCompletions.clear();
381 onCompletions = null;
382 }
383 }
384
385 public void addOnCompletion(Synchronization onCompletion) {
386 if (unitOfWork == null) {
387 // unit of work not yet registered so we store the on completion temporary
388 // until the unit of work is assigned to this exchange by the unit of work
389 if (onCompletions == null) {
390 onCompletions = new ArrayList<Synchronization>();
391 }
392 onCompletions.add(onCompletion);
393 } else {
394 getUnitOfWork().addSynchronization(onCompletion);
395 }
396 }
397
398 public boolean containsOnCompletion(Synchronization onCompletion) {
399 if (unitOfWork != null) {
400 // if there is an unit of work then the completions is moved there
401 return unitOfWork.containsSynchronization(onCompletion);
402 } else {
403 // check temporary completions if no unit of work yet
404 return onCompletions != null && onCompletions.contains(onCompletion);
405 }
406 }
407
408 public void handoverCompletions(Exchange target) {
409 if (onCompletions != null) {
410 for (Synchronization onCompletion : onCompletions) {
411 target.addOnCompletion(onCompletion);
412 }
413 // cleanup the temporary on completion list as they have been handed over
414 onCompletions.clear();
415 onCompletions = null;
416 } else if (unitOfWork != null) {
417 // let unit of work handover
418 unitOfWork.handoverSynchronization(target);
419 }
420 }
421
422 public List<Synchronization> handoverCompletions() {
423 List<Synchronization> answer = null;
424 if (onCompletions != null) {
425 answer = new ArrayList<Synchronization>(onCompletions);
426 onCompletions.clear();
427 onCompletions = null;
428 }
429 return answer;
430 }
431
432 /**
433 * Configures the message after it has been set on the exchange
434 */
435 protected void configureMessage(Message message) {
436 if (message instanceof MessageSupport) {
437 MessageSupport messageSupport = (MessageSupport)message;
438 messageSupport.setExchange(this);
439 }
440 }
441
442 @SuppressWarnings("deprecation")
443 protected String createExchangeId() {
444 String answer = null;
445 if (in != null) {
446 answer = in.createExchangeId();
447 }
448 if (answer == null) {
449 answer = context.getUuidGenerator().generateUuid();
450 }
451 return answer;
452 }
453 }