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.Message;
023 import org.apache.camel.impl.DefaultMessage;
024 import org.apache.camel.support.ServiceSupport;
025 import org.apache.camel.util.AsyncProcessorHelper;
026 import org.apache.camel.util.IOHelper;
027 import org.apache.camel.util.ObjectHelper;
028
029 /**
030 * A processor which converts the payload of the input message to be of the given type
031 * <p/>
032 * If the conversion fails an {@link org.apache.camel.InvalidPayloadException} is thrown.
033 *
034 * @version
035 */
036 public class ConvertBodyProcessor extends ServiceSupport implements AsyncProcessor {
037 private final Class<?> type;
038 private final String charset;
039
040 public ConvertBodyProcessor(Class<?> type) {
041 ObjectHelper.notNull(type, "type", this);
042 this.type = type;
043 this.charset = null;
044 }
045
046 public ConvertBodyProcessor(Class<?> type, String charset) {
047 ObjectHelper.notNull(type, "type", this);
048 this.type = type;
049 this.charset = IOHelper.normalizeCharset(charset);
050 }
051
052 @Override
053 public String toString() {
054 return "convertBodyTo[" + type.getCanonicalName() + "]";
055 }
056
057 public void process(Exchange exchange) throws Exception {
058 AsyncProcessorHelper.process(this, exchange);
059 }
060
061 @Override
062 public boolean process(Exchange exchange, AsyncCallback callback) {
063 Message in = exchange.getIn();
064 if (in.getBody() == null) {
065 // only convert if the is a body
066 callback.done(true);
067 return true;
068 }
069
070 if (charset != null) {
071 // override existing charset with configured charset as that is what the user
072 // have explicit configured and expects to be used
073 exchange.setProperty(Exchange.CHARSET_NAME, charset);
074 }
075 // use mandatory conversion
076 Object value;
077 try {
078 value = in.getMandatoryBody(type);
079 } catch (Exception e) {
080 exchange.setException(e);
081 callback.done(true);
082 return true;
083 }
084
085 // create a new message container so we do not drag specialized message objects along
086 Message msg = new DefaultMessage();
087 msg.copyFrom(in);
088 msg.setBody(value);
089
090 if (exchange.getPattern().isOutCapable()) {
091 exchange.setOut(msg);
092 } else {
093 exchange.setIn(msg);
094 }
095
096 // remove charset when we are done as we should not propagate that,
097 // as that can lead to double converting later on
098 if (charset != null) {
099 exchange.removeProperty(Exchange.CHARSET_NAME);
100 }
101
102 callback.done(true);
103 return true;
104 }
105
106 public Class<?> getType() {
107 return type;
108 }
109
110 @Override
111 protected void doStart() throws Exception {
112 // noop
113 }
114
115 @Override
116 protected void doStop() throws Exception {
117 // noop
118 }
119 }