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.strategy;
018
019 import org.apache.camel.Exchange;
020 import org.apache.camel.LoggingLevel;
021 import org.apache.camel.component.file.GenericFile;
022 import org.apache.camel.component.file.GenericFileEndpoint;
023 import org.apache.camel.component.file.GenericFileExclusiveReadLockStrategy;
024 import org.apache.camel.component.file.GenericFileOperations;
025 import org.apache.camel.util.CamelLogger;
026 import org.apache.camel.util.StopWatch;
027 import org.slf4j.Logger;
028 import org.slf4j.LoggerFactory;
029
030 /**
031 * Acquires exclusive read lock to the given file. Will wait until the lock is granted.
032 * After granting the read lock it is released, we just want to make sure that when we start
033 * consuming the file its not currently in progress of being written by third party.
034 */
035 public class GenericFileRenameExclusiveReadLockStrategy<T> implements GenericFileExclusiveReadLockStrategy<T> {
036 private static final Logger LOG = LoggerFactory.getLogger(GenericFileRenameExclusiveReadLockStrategy.class);
037 private long timeout;
038 private long checkInterval;
039 private LoggingLevel readLockLoggingLevel = LoggingLevel.WARN;
040
041 @Override
042 public void prepareOnStartup(GenericFileOperations<T> operations, GenericFileEndpoint<T> endpoint) throws Exception {
043 // noop
044 }
045
046 @Override
047 public boolean acquireExclusiveReadLock(GenericFileOperations<T> operations, GenericFile<T> file,
048 Exchange exchange) throws Exception {
049 LOG.trace("Waiting for exclusive read lock to file: {}", file);
050
051 // the trick is to try to rename the file, if we can rename then we have exclusive read
052 // since its a Generic file we cannot use java.nio to get a RW lock
053 String newName = file.getFileName() + ".camelExclusiveReadLock";
054
055 // make a copy as result and change its file name
056 GenericFile<T> newFile = file.copyFrom(file);
057 newFile.changeFileName(newName);
058 StopWatch watch = new StopWatch();
059
060 boolean exclusive = false;
061 while (!exclusive) {
062 // timeout check
063 if (timeout > 0) {
064 long delta = watch.taken();
065 if (delta > timeout) {
066 CamelLogger.log(LOG, readLockLoggingLevel,
067 "Cannot acquire read lock within " + timeout + " millis. Will skip the file: " + file);
068 // we could not get the lock within the timeout period, so return false
069 return false;
070 }
071 }
072
073 exclusive = operations.renameFile(file.getAbsoluteFilePath(), newFile.getAbsoluteFilePath());
074 if (exclusive) {
075 LOG.trace("Acquired exclusive read lock to file: {}", file);
076 // rename it back so we can read it
077 operations.renameFile(newFile.getAbsoluteFilePath(), file.getAbsoluteFilePath());
078 } else {
079 boolean interrupted = sleep();
080 if (interrupted) {
081 // we were interrupted while sleeping, we are likely being shutdown so return false
082 return false;
083 }
084 }
085 }
086
087 return true;
088 }
089
090 @Override
091 public void releaseExclusiveReadLock(GenericFileOperations<T> operations, GenericFile<T> file,
092 Exchange exchange) throws Exception {
093 // noop
094 }
095
096 private boolean sleep() {
097 LOG.trace("Exclusive read lock not granted. Sleeping for {} millis.", checkInterval);
098 try {
099 Thread.sleep(checkInterval);
100 return false;
101 } catch (InterruptedException e) {
102 LOG.debug("Sleep interrupted while waiting for exclusive read lock, so breaking out");
103 return true;
104 }
105 }
106
107 public long getTimeout() {
108 return timeout;
109 }
110
111 @Override
112 public void setTimeout(long timeout) {
113 this.timeout = timeout;
114 }
115
116 @Override
117 public void setCheckInterval(long checkInterval) {
118 this.checkInterval = checkInterval;
119 }
120
121 @Override
122 public void setReadLockLoggingLevel(LoggingLevel readLockLoggingLevel) {
123 this.readLockLoggingLevel = readLockLoggingLevel;
124 }
125 }