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.FileNotFoundException;
021    
022    import org.apache.camel.Component;
023    import org.apache.camel.Exchange;
024    import org.apache.camel.Processor;
025    import org.apache.camel.processor.idempotent.MemoryIdempotentRepository;
026    import org.apache.camel.util.FileUtil;
027    import org.apache.camel.util.ObjectHelper;
028    
029    /**
030     * File endpoint.
031     */
032    public class FileEndpoint extends GenericFileEndpoint<File> {
033    
034        private FileOperations operations = new FileOperations(this);
035        private File file;
036        private boolean copyAndDeleteOnRenameFail = true;
037        private boolean forceWrites = true;
038    
039        public FileEndpoint() {
040            // use marker file as default exclusive read locks
041            this.readLock = "markerFile";
042        }
043    
044        public FileEndpoint(String endpointUri, Component component) {
045            super(endpointUri, component);
046            // use marker file as default exclusive read locks
047            this.readLock = "markerFile";
048        }
049    
050        public FileConsumer createConsumer(Processor processor) throws Exception {
051            ObjectHelper.notNull(operations, "operations");
052            ObjectHelper.notNull(file, "file");
053    
054            // auto create starting directory if needed
055            if (!file.exists() && !file.isDirectory()) {
056                if (isAutoCreate()) {
057                    log.debug("Creating non existing starting directory: {}", file);
058                    boolean absolute = FileUtil.isAbsolute(file);
059                    boolean created = operations.buildDirectory(file.getPath(), absolute);
060                    if (!created) {
061                        log.warn("Cannot auto create starting directory: {}", file);
062                    }
063                } else if (isStartingDirectoryMustExist()) {
064                    throw new FileNotFoundException("Starting directory does not exist: " + file);
065                }
066            }
067    
068            FileConsumer result = newFileConsumer(processor, operations);
069    
070            if (isDelete() && getMove() != null) {
071                throw new IllegalArgumentException("You cannot set both delete=true and move options");
072            }
073    
074            // if noop=true then idempotent should also be configured
075            if (isNoop() && !isIdempotentSet()) {
076                log.info("Endpoint is configured with noop=true so forcing endpoint to be idempotent as well");
077                setIdempotent(true);
078            }
079    
080            // if idempotent and no repository set then create a default one
081            if (isIdempotentSet() && isIdempotent() && idempotentRepository == null) {
082                log.info("Using default memory based idempotent repository with cache max size: " + DEFAULT_IDEMPOTENT_CACHE_SIZE);
083                idempotentRepository = MemoryIdempotentRepository.memoryIdempotentRepository(DEFAULT_IDEMPOTENT_CACHE_SIZE);
084            }
085    
086            // set max messages per poll
087            result.setMaxMessagesPerPoll(getMaxMessagesPerPoll());
088            result.setEagerLimitMaxMessagesPerPoll(isEagerMaxMessagesPerPoll());
089    
090            configureConsumer(result);
091            return result;
092        }
093    
094        public GenericFileProducer<File> createProducer() throws Exception {
095            ObjectHelper.notNull(operations, "operations");
096    
097            // you cannot use temp prefix and file exists append
098            if (getFileExist() == GenericFileExist.Append && getTempPrefix() != null) {
099                throw new IllegalArgumentException("You cannot set both fileExist=Append and tempPrefix options");
100            }
101    
102            // ensure fileExist and moveExisting is configured correctly if in use
103            if (getFileExist() == GenericFileExist.Move && getMoveExisting() == null) {
104                throw new IllegalArgumentException("You must configure moveExisting option when fileExist=Move");
105            } else if (getMoveExisting() != null && getFileExist() != GenericFileExist.Move) {
106                throw new IllegalArgumentException("You must configure fileExist=Move when moveExisting has been set");
107            }
108    
109            return new GenericFileProducer<File>(this, operations);
110        }
111    
112        public Exchange createExchange(GenericFile<File> file) {
113            Exchange exchange = createExchange();
114            if (file != null) {
115                file.bindToExchange(exchange);
116            }
117            return exchange;
118        }
119    
120        /**
121         * Strategy to create a new {@link FileConsumer}
122         *
123         * @param processor  the given processor
124         * @param operations file operations
125         * @return the created consumer
126         */
127        protected FileConsumer newFileConsumer(Processor processor, GenericFileOperations<File> operations) {
128            return new FileConsumer(this, processor, operations);
129        }
130    
131        public File getFile() {
132            return file;
133        }
134    
135        public void setFile(File file) {
136            this.file = file;
137            // update configuration as well
138            getConfiguration().setDirectory(FileUtil.isAbsolute(file) ? file.getAbsolutePath() : file.getPath());
139        }
140    
141        @Override
142        public String getScheme() {
143            return "file";
144        }
145    
146        @Override
147        protected String createEndpointUri() {
148            return getFile().toURI().toString();
149        }
150    
151        @Override
152        public char getFileSeparator() {       
153            return File.separatorChar;
154        }
155    
156        @Override
157        public boolean isAbsolute(String name) {
158            // relative or absolute path?
159            return FileUtil.isAbsolute(new File(name));
160        }
161    
162        public boolean isCopyAndDeleteOnRenameFail() {
163            return copyAndDeleteOnRenameFail;
164        }
165    
166        public void setCopyAndDeleteOnRenameFail(boolean copyAndDeleteOnRenameFail) {
167            this.copyAndDeleteOnRenameFail = copyAndDeleteOnRenameFail;
168        }
169    
170        public boolean isForceWrites() {
171            return forceWrites;
172        }
173    
174        public void setForceWrites(boolean forceWrites) {
175            this.forceWrites = forceWrites;
176        }
177    }