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.io.ByteArrayOutputStream;
020
021 import org.apache.camel.AsyncCallback;
022 import org.apache.camel.AsyncProcessor;
023 import org.apache.camel.CamelContext;
024 import org.apache.camel.CamelContextAware;
025 import org.apache.camel.Exchange;
026 import org.apache.camel.Message;
027 import org.apache.camel.Traceable;
028 import org.apache.camel.converter.stream.CachedOutputStream;
029 import org.apache.camel.spi.DataFormat;
030 import org.apache.camel.support.ServiceSupport;
031 import org.apache.camel.util.AsyncProcessorHelper;
032 import org.apache.camel.util.ObjectHelper;
033 import org.apache.camel.util.ServiceHelper;
034
035 /**
036 * Marshals the body of the incoming message using the given
037 * <a href="http://camel.apache.org/data-format.html">data format</a>
038 *
039 * @version
040 */
041 public class MarshalProcessor extends ServiceSupport implements AsyncProcessor, Traceable, CamelContextAware {
042 private CamelContext camelContext;
043 private final DataFormat dataFormat;
044
045 public MarshalProcessor(DataFormat dataFormat) {
046 this.dataFormat = dataFormat;
047 }
048
049 public void process(Exchange exchange) throws Exception {
050 AsyncProcessorHelper.process(this, exchange);
051 }
052
053 public boolean process(Exchange exchange, AsyncCallback callback) {
054 ObjectHelper.notNull(dataFormat, "dataFormat");
055
056 // if stream caching is enabled then use that so we can stream accordingly
057 // for example to overflow to disk for big streams
058 CachedOutputStream cos;
059 ByteArrayOutputStream os;
060 if (exchange.getContext().getStreamCachingStrategy().isEnabled()) {
061 cos = new CachedOutputStream(exchange);
062 os = null;
063 } else {
064 cos = null;
065 os = new ByteArrayOutputStream();
066 }
067
068 Message in = exchange.getIn();
069 Object body = in.getBody();
070
071 // lets setup the out message before we invoke the dataFormat
072 // so that it can mutate it if necessary
073 Message out = exchange.getOut();
074 out.copyFrom(in);
075
076 try {
077 if (cos != null) {
078 dataFormat.marshal(exchange, body, cos);
079 out.setBody(cos.newStreamCache());
080 } else {
081 dataFormat.marshal(exchange, body, os);
082 byte[] data = os.toByteArray();
083 out.setBody(data);
084 }
085 } catch (Exception e) {
086 // remove OUT message, as an exception occurred
087 exchange.setOut(null);
088 exchange.setException(e);
089 }
090
091 callback.done(true);
092 return true;
093 }
094
095 @Override
096 public String toString() {
097 return "Marshal[" + dataFormat + "]";
098 }
099
100 public String getTraceLabel() {
101 return "marshal[" + dataFormat + "]";
102 }
103
104 public CamelContext getCamelContext() {
105 return camelContext;
106 }
107
108 public void setCamelContext(CamelContext camelContext) {
109 this.camelContext = camelContext;
110 }
111
112 @Override
113 protected void doStart() throws Exception {
114 // inject CamelContext on data format
115 if (dataFormat instanceof CamelContextAware) {
116 ((CamelContextAware) dataFormat).setCamelContext(camelContext);
117 }
118 ServiceHelper.startService(dataFormat);
119 }
120
121 @Override
122 protected void doStop() throws Exception {
123 ServiceHelper.stopService(dataFormat);
124 }
125 }