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.component.file;
018    
019    import java.io.File;
020    import java.io.FileOutputStream;
021    import java.io.IOException;
022    import java.io.InputStream;
023    import java.io.RandomAccessFile;
024    import java.nio.ByteBuffer;
025    import java.nio.channels.FileChannel;
026    
027    import org.apache.camel.Exchange;
028    import org.apache.camel.Message;
029    import org.apache.camel.Producer;
030    import org.apache.camel.impl.DefaultProducer;
031    import org.apache.camel.util.ExchangeHelper;
032    import org.apache.commons.logging.Log;
033    import org.apache.commons.logging.LogFactory;
034    
035    /**
036     * A {@link Producer} implementation for File
037     *
038     * @version $Revision: 37863 $
039     */
040    public class FileProducer extends DefaultProducer {
041        private static final transient Log LOG = LogFactory.getLog(FileProducer.class);
042        private FileEndpoint endpoint;
043    
044        public FileProducer(FileEndpoint endpoint) {
045            super(endpoint);
046            this.endpoint = endpoint;
047        }
048    
049        public FileEndpoint getEndpoint() {
050            return (FileEndpoint) super.getEndpoint();
051        }
052    
053        public void process(Exchange exchange) throws Exception {
054            // TODO is it really worth using a FileExchange as the core type?
055            FileExchange fileExchange = endpoint.createExchange(exchange);
056            process(fileExchange);
057            ExchangeHelper.copyResults(exchange, fileExchange);
058        }
059    
060        public void process(FileExchange exchange) throws Exception {
061            if (ExchangeHelper.isOutCapable(exchange)) {
062                // lets poll the file
063                Message out = exchange.getOut(true);
064                endpoint.configureMessage(endpoint.getFile(), out);
065                return;
066            }
067    
068            InputStream in = ExchangeHelper.getMandatoryInBody(exchange, InputStream.class);
069            File file = createFileName(exchange.getIn());
070            buildDirectory(file);
071    
072            if (LOG.isDebugEnabled()) {
073                LOG.debug("About to write to: " + file + " from exchange: " + exchange);
074            }
075            FileChannel fc = null;
076            try {
077                if (getEndpoint().isAppend()) {
078                    fc = new RandomAccessFile(file, "rw").getChannel();
079                    fc.position(fc.size());
080                } else {
081                    fc = new FileOutputStream(file).getChannel();
082                }
083    
084                int size = getEndpoint().getBufferSize();
085                byte[] buffer = new byte[size];
086                ByteBuffer byteBuffer = ByteBuffer.wrap(buffer);
087                while (true) {
088                    int count = in.read(buffer);
089                    if (count <= 0) {
090                        break;
091                    } else if (count < size) {
092                        byteBuffer = ByteBuffer.wrap(buffer, 0, count);
093                        fc.write(byteBuffer);
094                        break;
095                    } else {
096                        fc.write(byteBuffer);
097                        byteBuffer.clear();
098                    }
099                }
100            } finally {
101                if (in != null) {
102                    try {
103                        in.close();
104                    } catch (IOException e) {
105                        LOG.warn("Failed to close input: " + e, e);
106                    }
107                }
108                if (fc != null) {
109                    try {
110                        fc.close();
111                    } catch (IOException e) {
112                        LOG.warn("Failed to close output: " + e, e);
113                    }
114                }
115            }
116        }
117    
118        protected File createFileName(Message message) {
119            File answer;
120    
121            String name = null;
122            if (!endpoint.isIgnoreFileNameHeader()) {
123                name = message.getHeader(FileComponent.HEADER_FILE_NAME, String.class);
124            }
125    
126            File endpointFile = endpoint.getFile();
127            if (endpointFile.isDirectory()) {
128                if (name != null) {
129                    answer = new File(endpointFile, name);
130                    if (answer.isDirectory()) {
131                        answer = new File(answer, message.getMessageId());
132                    }
133                } else {
134                    answer = new File(endpointFile, message.getMessageId());
135                }
136            } else {
137                if (name == null) {
138                    answer = endpointFile;
139                } else {
140                    answer = new File(endpointFile, name);
141                }
142            }
143    
144            return answer;
145        }
146    
147        private void buildDirectory(File file) {
148            String dirName = file.getAbsolutePath();
149            int index = dirName.lastIndexOf(File.separatorChar);
150            if (index > 0) {
151                dirName = dirName.substring(0, index);
152                File dir = new File(dirName);
153                dir.mkdirs();
154            }
155        }
156    
157    }