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.converter.stream;
018
019 import java.io.IOException;
020 import java.io.InputStream;
021 import java.io.Reader;
022 import java.io.StringReader;
023
024 import javax.xml.transform.TransformerException;
025 import javax.xml.transform.sax.SAXSource;
026 import javax.xml.transform.stream.StreamSource;
027
028 import org.apache.camel.Converter;
029 import org.apache.camel.Exchange;
030 import org.apache.camel.converter.IOConverter;
031 import org.apache.camel.converter.jaxp.BytesSource;
032 import org.apache.camel.converter.jaxp.StringSource;
033 import org.apache.camel.converter.jaxp.XmlConverter;
034 import org.apache.camel.util.IOHelper;
035 import org.apache.commons.logging.Log;
036 import org.apache.commons.logging.LogFactory;
037
038 /**
039 * A set of {@link Converter} methods for wrapping stream-based messages in a {@link StreamCache}
040 * implementation to ensure message re-readability (eg multicasting, retrying)
041 */
042 @Converter
043 public class StreamCacheConverter {
044 private static final transient Log LOG = LogFactory.getLog(StreamCacheConverter.class);
045
046 private XmlConverter converter = new XmlConverter();
047
048 @Converter
049 public StreamCache convertToStreamCache(StreamSource source, Exchange exchange) throws IOException {
050 return new StreamSourceCache(source, exchange);
051 }
052
053 @Converter
054 public StreamCache convertToStreamCache(StringSource source) {
055 //no need to do stream caching for a StringSource
056 return null;
057 }
058
059 @Converter
060 public StreamCache convertToStreamCache(BytesSource source) {
061 //no need to do stream caching for a BytesSource
062 return null;
063 }
064
065 @Converter
066 public StreamCache convertToStreamCache(SAXSource source) throws TransformerException {
067 return new SourceCache(converter.toString(source));
068 }
069
070 @Converter
071 public StreamCache convertToStreamCache(InputStream stream, Exchange exchange) throws IOException {
072 // set up CachedOutputStream with the properties
073 CachedOutputStream cos = new CachedOutputStream(exchange.getContext().getProperties());
074 IOHelper.copyAndCloseInput(stream, cos);
075 return cos.getStreamCache();
076 }
077
078 @Converter
079 public StreamCache convertToStreamCache(Reader reader) throws IOException {
080 return new ReaderCache(IOConverter.toString(reader));
081 }
082
083 /*
084 * {@link StreamCache} implementation for {@link Source}s
085 */
086 private class SourceCache extends StringSource implements StreamCache {
087
088 private static final long serialVersionUID = 4147248494104812945L;
089
090 public SourceCache() {
091 }
092
093 public SourceCache(String text) {
094 super(text);
095 }
096
097 public void reset() {
098 // do nothing here
099 }
100
101 }
102
103 /*
104 * {@link StreamCache} implementation for Cache the StreamSource {@link StreamSource}s
105 */
106 private class StreamSourceCache extends StreamSource implements StreamCache {
107 StreamCache streamCache;
108 ReaderCache readCache;
109
110 public StreamSourceCache(StreamSource source, Exchange exchange) throws IOException {
111 if (source.getInputStream() != null) {
112 // set up CachedOutputStream with the properties
113 CachedOutputStream cos = new CachedOutputStream(exchange.getContext().getProperties());
114 IOHelper.copyAndCloseInput(source.getInputStream(), cos);
115 streamCache = cos.getStreamCache();
116 setInputStream((InputStream)streamCache);
117 setSystemId(source.getSystemId());
118 }
119 if (source.getReader() != null) {
120 readCache = new ReaderCache(IOConverter.toString(source.getReader()));
121 setReader(readCache);
122 }
123 }
124 public void reset() {
125 if (streamCache != null) {
126 streamCache.reset();
127 }
128 if (readCache != null) {
129 readCache.reset();
130 }
131 }
132
133 }
134
135 private class ReaderCache extends StringReader implements StreamCache {
136
137 public ReaderCache(String s) {
138 super(s);
139 }
140
141 public void reset() {
142 try {
143 super.reset();
144 } catch (IOException e) {
145 LOG.warn("Exception is thrown when resets the ReaderCache", e);
146 }
147 }
148
149 public void close() {
150 // Do not release the string for caching
151 }
152
153 }
154
155
156
157
158 }