Xinwen Deng / aem-src

/*
 * Decompiled with CFR 0_118.
 * 
 * Could not load the following classes:
 *  org.apache.sling.commons.scheduler.Job
 *  org.apache.sling.commons.scheduler.JobContext
 *  org.slf4j.Logger
 *  org.slf4j.LoggerFactory
 */
package com.adobe.aemfd.watchfolder.job;

import com.adobe.aemfd.watchfolder.config.WatchFolderConfiguration;
import com.adobe.aemfd.watchfolder.distribution.JobDistributor;
import com.adobe.aemfd.watchfolder.job.ExpiredFileFilter;
import com.adobe.aemfd.watchfolder.job.FileFilter;
import com.adobe.aemfd.watchfolder.job.FilePathBuilder;
import com.adobe.aemfd.watchfolder.job.FileResultHandlerImpl;
import com.adobe.aemfd.watchfolder.job.FileResultHandlerWrapper;
import com.adobe.aemfd.watchfolder.job.FileSorterByDate;
import com.adobe.aemfd.watchfolder.job.GUIDGenerator;
import com.adobe.aemfd.watchfolder.job.WatchedFolderUtils;
import com.adobe.aemfd.watchfolder.util.LogUtil;
import com.adobe.aemfd.watchfolder.util.TopologyHelper;
import java.io.File;
import java.io.FilenameFilter;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.StringTokenizer;
import org.apache.sling.commons.scheduler.Job;
import org.apache.sling.commons.scheduler.JobContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/*
 * This class specifies class file version 49.0 but uses Java 6 signatures.  Assumed Java 6.
 */
public class FileScanJobImpl
implements Job {
    private static final Logger _logger = LoggerFactory.getLogger(FileScanJobImpl.class);
    public static Integer _jobCounterTotal = 0;
    public static Integer _jobCounterSuccess = 0;
    public static Integer _jobCounterFailure = 0;
    public static final String PATTERN_SEPARATED_CHAR = ";";
    private static final String SCAN_FILE_LOCK = ".scnLock";
    private String m_watchedFolderURL;
    private File m_inFolder;
    private File m_outFolder;
    private File m_stagingFolder;
    private File m_preservedFolder;
    private static boolean isLocked = false;
    private ExpiredFileFilter m_fileFilter;
    private WatchFolderConfiguration config;
    private JobDistributor jobDistributor;
    private boolean runModeActive = false;

    public FileScanJobImpl(WatchFolderConfiguration config, JobDistributor jobDistributor, TopologyHelper topologyHelper) {
        this.config = config;
        this.jobDistributor = jobDistributor;
        String rm = config.getRunModes();
        this.runModeActive = topologyHelper.isRunModeActiveLocally(rm);
        if (!this.runModeActive) {
            _logger.warn("No run modes in [" + rm + "] active on local instance, file-scan job will not run for watch-folder " + config.getId());
        }
    }

    public void execute(JobContext jobContext) {
        if (this.runModeActive) {
            this.execute(jobContext.getConfiguration());
        }
    }

    public void execute(Map aInputMap) {
        try {
            this.doInitialize(aInputMap);
            Object[] _inputList = this.fetchInputs();
            this.processInputs(_inputList);
        }
        catch (Throwable t) {
            this.handleFailure(t, null, null);
        }
    }

    public void doInitialize(Map aInputMap) {
        this.initialize(aInputMap);
        if (_logger.isTraceEnabled()) {
            _logger.trace("FileScanJobImpl ------ Processing WatchedFolder started");
        }
        this.verifyAndInitializeFolderHeirarchy();
        this.cleanupPendingJobs();
    }

    public void processInputs(Object[] aInputList) {
        if (aInputList != null && aInputList.length > 0) {
            for (int _i = 0; _i < aInputList.length; ++_i) {
                try {
                    Map<String, File> _inputMap = this.createInputParamMap(aInputList[_i]);
                    this.invokeJob(aInputList[_i], _inputMap);
                    continue;
                }
                catch (Throwable t) {
                    this.handleFailure(t, null, null);
                }
            }
        }
    }

    /*
     * WARNING - Removed try catching itself - possible behaviour change.
     * Enabled aggressive block sorting
     * Enabled unnecessary exception pruning
     * Enabled aggressive exception aggregation
     */
    public Object[] fetchInputs() {
        Object[] _fetchedFiles = new File[]{};
        long time1 = 0;
        if (_logger.isTraceEnabled()) {
            time1 = System.currentTimeMillis();
        }
        boolean _success = true;
        try {
            File _stageFolder;
            if (_success && (_stageFolder = this.fetchFilesToProcess()) != null) {
                _fetchedFiles = _stageFolder.listFiles();
            }
            Object var7_6 = null;
            if (_success) {
                // empty if block
            }
        }
        catch (Throwable var6_8) {
            Object var7_7 = null;
            if (_success) {
                // empty if block
            }
            throw var6_8;
        }
        if (_logger.isTraceEnabled()) {
            long time2 = System.currentTimeMillis();
            _logger.trace("Time to fetch input is ------" + (time2 - time1));
        }
        return _fetchedFiles;
    }

    public static File[] searchFileForPattern(File fileToSearch, String pattern) {
        File[] filteredFileList = null;
        FileFilter fileFilter = new FileFilter();
        fileFilter.clearExcludeList();
        fileFilter.clearIncludeList();
        if (pattern != null && pattern.length() > 0) {
            String[] patternsArr = FileScanJobImpl.getPatternArray(pattern);
            fileFilter.addIncludeList(patternsArr);
            if (fileToSearch.isDirectory()) {
                filteredFileList = fileToSearch.listFiles(fileFilter);
            } else if (fileFilter.accept(fileToSearch.getParentFile(), fileToSearch.getName())) {
                filteredFileList = new File[]{fileToSearch};
            }
        }
        return filteredFileList;
    }

    private File[] getFilteredFiles(File f, String pattern) {
        File[] currentFileList = null;
        if (f != null) {
            currentFileList = FileScanJobImpl.searchFileForPattern(f, pattern);
        }
        return currentFileList;
    }

    public Map<String, File> createInputParamMap(Object aInput) {
        File currentFile = null;
        HashMap<String, File> documentMap = null;
        File[] filteredFileList = this.getFilteredFiles((File)aInput, this.config.getInputFilePattern());
        if (filteredFileList != null && filteredFileList.length > 0) {
            documentMap = new HashMap<String, File>();
            _logger.debug("Document map .............");
            for (int j = 0; j < filteredFileList.length; ++j) {
                currentFile = filteredFileList[j];
                documentMap.put(currentFile.getName(), currentFile);
                _logger.debug(" document ---- " + currentFile.getName() + " :  " + currentFile.getPath());
            }
        }
        return documentMap;
    }

    public void invokeJob(Object aInputObject, Map<String, File> aInputParamMap) {
        if (aInputParamMap == null) {
            aInputParamMap = Collections.emptyMap();
        }
        try {
            this.jobDistributor.distributeFileProcessing(this.config, (File)aInputObject, aInputParamMap);
        }
        catch (Throwable e) {
            _logger.info(LogUtil.toJSON(!this.config.isAsynch() ? "START_PROCESSOR_FAILURE_SINGLETON" : "CREATE_JOB_FAILURE", ((File)aInputObject).getAbsolutePath(), null, null, e.getMessage()));
            this.handleFailure(e, aInputObject, null);
        }
    }

    private void initialize(Map aParamMap) {
        this.m_watchedFolderURL = this.config.getFolderPath();
        if (_logger.isTraceEnabled()) {
            _logger.trace("JobAgent : Entered the Job Agent for URL" + this.m_watchedFolderURL);
        }
        this.setupFilterForFileFetching();
    }

    private void setupFilterForFileFetching() {
        if (this.m_fileFilter == null) {
            this.m_fileFilter = new ExpiredFileFilter();
        }
        this.m_fileFilter.clearIncludeList();
        this.m_fileFilter.clearExcludeList();
        String[] excludeList = FileScanJobImpl.getPatternArray(this.config.getExcludeFilePattern());
        if (excludeList != null && excludeList.length > 0) {
            this.m_fileFilter.addExcludeList(excludeList);
        }
        this.m_fileFilter.addIncludeList(FileScanJobImpl.getPatternArray(this.config.getIncludeFilePattern()));
        long waitTime = new Long(this.config.getWaitTime());
        this.m_fileFilter.addExpirationDuration(waitTime);
    }

    private void cleanupPendingJobs() {
        long purgeDuration = new Long(this.config.getPurgeDuration());
        long purgeDurationInMilliSeconds = purgeDuration * 24 * 60 * 60 * 1000;
        if (purgeDuration != -1) {
            File _closestResultFolderToPurge = WatchedFolderUtils.searchPercent(new File(this.m_watchedFolderURL).getAbsolutePath(), this.config.getResultFolderName());
            if (_logger.isTraceEnabled()) {
                _logger.trace("watchfolder path" + this.m_watchedFolderURL + "------cleaning RESULT FOLDER" + _closestResultFolderToPurge);
            }
            WatchedFolderUtils.recursiveDeleteExpiredDir(_closestResultFolderToPurge, _closestResultFolderToPurge, purgeDurationInMilliSeconds);
        }
        this.cleanupStageFiles();
    }

    private void clearExpiredFilesForStageFolder(File sf) {
        File[] childFiles;
        if (sf != null && sf.exists() && sf.isDirectory() && (childFiles = sf.listFiles()) != null && childFiles.length > 0) {
            for (File cf : childFiles) {
                File removeFile;
                String fName = cf.getName();
                if (fName.endsWith(".remove") || (removeFile = new File(cf.getAbsolutePath() + ".remove")).exists()) continue;
                _logger.info("Marking input " + cf + " as failure after timeout!");
                String msg = "File not processed after a significant amount of time, marking as failure!";
                new FileResultHandlerImpl(this.config).handleFailure(msg, null, cf, null);
            }
        }
    }

    private void clearExpiredStageFiles() {
        long expDur = this.config.getStageFileExpirationDuration();
        if (expDur <= 0) {
            return;
        }
        if (this.config.isDeleteExpiredStageFileOnlyWhenThrottled() && !this.config.isThrottleOn()) {
            return;
        }
        ExpiredFileFilter filter = new ExpiredFileFilter();
        filter.addExpirationDuration(expDur * 1000);
        File[] stageFolders = this.m_stagingFolder.listFiles(filter);
        if (stageFolders != null && stageFolders.length > 0) {
            for (File sf : stageFolders) {
                this.clearExpiredFilesForStageFolder(sf);
            }
        }
    }

    private void cleanupStageFiles() {
        this.clearExpiredStageFiles();
        ExpiredFileFilter expFileFilter = new ExpiredFileFilter();
        expFileFilter.addExpirationDuration(300000);
        File[] _fileList = this.m_stagingFolder.listFiles(expFileFilter);
        if (_fileList != null && _fileList.length > 0) {
            for (int i = 0; i < _fileList.length; ++i) {
                WatchedFolderUtils.deleteMarkedFiles(_fileList[i]);
            }
        }
    }

    private File fetchFilesToProcess() {
        File _stagingLocation = null;
        int _batchSize = this.config.getBatchSize();
        boolean _throttle = this.config.isThrottleOn();
        int _jobsInProgress = 0;
        _jobsInProgress = _throttle ? this.getInvocationsInProgress() : 0;
        int _numberOfFilesToFetch = 0;
        if (_batchSize != -1) {
            if (_jobsInProgress > _batchSize) {
                if (_logger.isTraceEnabled()) {
                    _logger.trace("There are " + _jobsInProgress + "  more than batch size " + _batchSize);
                }
                _numberOfFilesToFetch = 0;
            } else {
                _numberOfFilesToFetch = _batchSize - _jobsInProgress;
                if (_logger.isTraceEnabled()) {
                    _logger.trace("Number of files to fetch are ---- " + _numberOfFilesToFetch);
                }
            }
        } else {
            _numberOfFilesToFetch = -1;
            if (_logger.isTraceEnabled()) {
                _logger.trace("Pick all the files as batch size is -1");
            }
        }
        if (_numberOfFilesToFetch == -1 || _numberOfFilesToFetch > 0) {
            File[] _matchingFiles = this.m_inFolder.listFiles(this.m_fileFilter);
            FileSorterByDate _fileSorterByDate = new FileSorterByDate();
            FileSorterByDate.FileWrapper[] _sortedFiles = _fileSorterByDate.sortFiles(_matchingFiles);
            int _nFilesToPick = 0;
            if (_sortedFiles != null) {
                _nFilesToPick = _sortedFiles.length;
            }
            if (_numberOfFilesToFetch > -1 && _numberOfFilesToFetch < _nFilesToPick) {
                _nFilesToPick = _numberOfFilesToFetch;
            }
            if (_sortedFiles != null && _sortedFiles.length > 0 && _nFilesToPick > 0) {
                if (_logger.isTraceEnabled()) {
                    _logger.trace("FileScanJobImpl : Found " + _sortedFiles.length + " files");
                }
                _stagingLocation = this.moveFilesToStaging(_sortedFiles, _nFilesToPick);
            }
        }
        return _stagingLocation;
    }

    private File moveFilesToStaging(FileSorterByDate.FileWrapper[] aFileset, int aNumberOfFilesToMove) {
        String _guid = GUIDGenerator.generate('W');
        File _stagingFile = null;
        File _currentSourceFile = null;
        int _filesPicked = 0;
        File _guidFolder = new File(this.m_stagingFolder, _guid);
        boolean _fStageCreated = false;
        if (aFileset != null && aFileset.length > 0) {
            try {
                FilePathBuilder.makeDirectoryPath(_guidFolder);
                _fStageCreated = true;
            }
            catch (RuntimeException _fpre) {
                _fStageCreated = false;
                _logger.error("Cannot create stage folder inside watch folder.  Reason may be 1.) Disk is full or 2.)not enough permissions", (Throwable)_fpre);
            }
            if (_fStageCreated) {
                if (_logger.isTraceEnabled()) {
                    _logger.trace("Files picked are........" + aFileset.length);
                }
                for (int _i = 0; _i < aFileset.length; ++_i) {
                    _currentSourceFile = aFileset[_i].file;
                    if (_logger.isTraceEnabled()) {
                        _logger.trace("File[" + _i + "]...." + _currentSourceFile);
                    }
                    if (_currentSourceFile.renameTo(_stagingFile = new File(_guidFolder, _currentSourceFile.getName()))) {
                        ++_filesPicked;
                        if (_logger.isTraceEnabled()) {
                            _logger.trace("Renamed the file" + _currentSourceFile.getName() + " to " + _stagingFile);
                        }
                        if (_filesPicked < aNumberOfFilesToMove) continue;
                        break;
                    }
                    _logger.error("Cannot rename the file" + _currentSourceFile.getName() + " to " + _stagingFile + ". Reason may be 1.) Disk is full or 2.)it does not exists anymore or service may not have rights to write to the watch folder");
                }
            }
        }
        if (_fStageCreated && _filesPicked <= 0) {
            _guidFolder.delete();
            _logger.debug("No files are picked up");
        }
        return _guidFolder;
    }

    private static String[] getPatternArray(String patternString) {
        String[] patternArray = null;
        if (patternString != null) {
            StringTokenizer st = new StringTokenizer(patternString, ";");
            int numToken = st.countTokens();
            patternArray = new String[numToken];
            for (int i = 0; i < numToken; ++i) {
                patternArray[i] = st.nextToken();
            }
        }
        return patternArray;
    }

    public void handleFailure(Throwable aCause, Object aInfo, String jobId) {
        FileResultHandlerWrapper _resultHandler = new FileResultHandlerWrapper(this.config);
        _resultHandler.handleFailure(aCause, (File)aInfo, jobId);
    }

    private void verifyAndInitializeFolderHeirarchy() {
        String _inputFolder = this.config.getInputFolderName();
        String _stageFolder = this.config.getStageFolderName();
        String _failureFolder = this.config.getFailureFolderName();
        String _resultFolder = this.config.getResultFolderName();
        String _preserveFolder = this.config.getPreserveFolderName();
        this.m_inFolder = WatchedFolderUtils.verifyAndCreateFilePath(this.m_watchedFolderURL, _inputFolder);
        this.m_outFolder = WatchedFolderUtils.verifyAndCreateFilePath(this.m_watchedFolderURL, _resultFolder);
        this.m_preservedFolder = WatchedFolderUtils.verifyAndCreateFilePath(this.m_watchedFolderURL, _preserveFolder);
        File _failureFile = WatchedFolderUtils.verifyAndCreateFilePath(this.m_watchedFolderURL, _failureFolder);
        this.m_stagingFolder = WatchedFolderUtils.verifyAndCreateFilePath(this.m_watchedFolderURL, _stageFolder);
    }

    private int getInvocationsInProgress() {
        int _numberOfInvocationsInProgress = 0;
        File[] _folderList = this.m_stagingFolder.listFiles();
        if (_folderList != null && _folderList.length > 0) {
            if (_logger.isTraceEnabled()) {
                _logger.trace("Files/Foldes in the stage folder......");
            }
            for (int _i = 0; _i < _folderList.length; ++_i) {
                File _currentFolder = _folderList[_i];
                File[] _filesInCurrentFolder = _currentFolder.listFiles();
                if (_filesInCurrentFolder == null || _filesInCurrentFolder.length <= 0) continue;
                int _length = _filesInCurrentFolder.length;
                _numberOfInvocationsInProgress += _length;
                if (!_logger.isTraceEnabled()) continue;
                _logger.trace("_currentFolder" + _currentFolder.toString() + "   has " + _length + " number of files");
            }
        }
        if (_logger.isTraceEnabled()) {
            _logger.trace(" Number of job in progress are ----" + _numberOfInvocationsInProgress);
        }
        return _numberOfInvocationsInProgress;
    }
}