content: /*
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements.  See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You under the Apache License, Version 2.0
   * (the "License"); you may not use this file except in compliance with
   * the License.  You may obtain a copy of the License at
   *
   *      http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package org.apache.camel.component.file.strategy;
  
  import java.io.File;
  import java.util.Date;
  
  import org.apache.camel.Exchange;
  import org.apache.camel.LoggingLevel;
  import org.apache.camel.component.file.GenericFile;
  import org.apache.camel.component.file.GenericFileOperations;
  import org.apache.camel.spi.CamelLogger;
  import org.apache.camel.util.StopWatch;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  /**
   * Acquires exclusive read lock to the given file by checking whether the file is being changed by scanning the file at
   * different intervals (to detect changes).
   * <p/>
   * Setting the option {@link #setMarkerFiler(boolean)} to <tt>false</tt> allows to turn off using marker files.
   */
  public class FileChangedExclusiveReadLockStrategy extends MarkerFileExclusiveReadLockStrategy {
      private static final Logger LOG = LoggerFactory.getLogger(FileChangedExclusiveReadLockStrategy.class);
      private long timeout;
      private long checkInterval = 1000;
      private long minLength = 1;
      private long minAge;
      private LoggingLevel readLockLoggingLevel = LoggingLevel.DEBUG;
  
      @Override
      public boolean acquireExclusiveReadLock(GenericFileOperations<File> operations, GenericFile<File> file, Exchange exchange)
              throws Exception {
          // must call super
          if (!super.acquireExclusiveReadLock(operations, file, exchange)) {
              return false;
          }
  
          File target = new File(file.getAbsoluteFilePath());
          boolean exclusive = false;
  
          LOG.trace("Waiting for exclusive read lock to file: {}", file);
  
          long lastModified = Long.MIN_VALUE;
          long length = Long.MIN_VALUE;
          StopWatch watch = new StopWatch();
          long startTime = (new Date()).getTime();
  
          while (!exclusive) {
              // timeout check
              if (timeout > 0) {
                  if (isTimedOut(watch, target, timeout, readLockLoggingLevel)) {
                      return false;
                  }
              }
  
              if (!target.exists()) {
                  CamelLogger.log(LOG, readLockLoggingLevel,
                          "Cannot acquire read lock as file no longer exists. Will skip the file: " + file);
                  return false;
              }
  
              long newLastModified = target.lastModified();
              long newLength = target.length();
              long newOlderThan = startTime + watch.taken() - minAge;
  
              LOG.trace("Previous last modified: {}, new last modified: {}", lastModified, newLastModified);
              LOG.trace("Previous length: {}, new length: {}", length, newLength);
              LOG.trace("New older than threshold: {}", newOlderThan);
  
              // CHECKSTYLE:OFF
              if (newLength >= minLength && ((minAge == 0 && newLastModified == lastModified && newLength == length)
                      || (minAge != 0 && newLastModified < newOlderThan))) {
              // CHECKSTYLE:ON
                  LOG.trace("Read lock acquired.");
                  exclusive = true;
              } else {
                  // set new base file change information
                  lastModified = newLastModified;
                  length = newLength;
  
                  boolean interrupted = sleep();
                  if (interrupted) {
                      // we were interrupted while sleeping, we are likely being
                      // shutdown so return false
                      return false;
                  }
              }
          }
  
          return exclusive;
      }
  
      private boolean sleep() {
          LOG.trace("Exclusive read lock not granted. Sleeping for {} millis.", checkInterval);
          try {
              Thread.sleep(checkInterval);
              return false;
          } catch (InterruptedException e) {
              Thread.currentThread().interrupt();
              LOG.debug("Sleep interrupted while waiting for exclusive read lock, so breaking out");
              return true;
          }
      }
  
      public long getTimeout() {
          return timeout;
      }
  
      @Override
      public void setTimeout(long timeout) {
          this.timeout = timeout;
      }
  
      public long getCheckInterval() {
          return checkInterval;
      }
  
      @Override
      public void setCheckInterval(long checkInterval) {
          this.checkInterval = checkInterval;
      }
  
      @Override
      public void setReadLockLoggingLevel(LoggingLevel readLockLoggingLevel) {
          this.readLockLoggingLevel = readLockLoggingLevel;
      }
  
      public long getMinLength() {
          return minLength;
      }
  
      public void setMinLength(long minLength) {
          this.minLength = minLength;
      }
  
      public long getMinAge() {
          return minAge;
      }
  
      public void setMinAge(long minAge) {
          this.minAge = minAge;
      }
  }
  