Directory watching for changes in java

Question

I am using WatchService to watch change in directory, in particular creation of new file in directory. Below is my code -

package watcher;

import java.nio.file.*;
import static java.nio.file.StandardWatchEventKinds.ENTRY_CREATE;
import static java.nio.file.StandardWatchEventKinds.OVERFLOW;
import java.io.*;


public class Watch {
    public static void main(String[] args) throws IOException {
        Path dir = Paths.get("c:\\mk\\");
        WatchService service = FileSystems.getDefault().newWatchService();
        WatchKey key = dir.register(service, ENTRY_CREATE);

        System.out.println("Watching directory: "+dir.toString());
        for(;;){
            WatchKey key1;
            try {
                key1 = service.take();
            } catch (InterruptedException x) {
                break;
            }

            for (WatchEvent<?> event: key1.pollEvents()) {
                WatchEvent.Kind<?> kind = event.kind();

                if (kind == OVERFLOW) {
                    continue;
                }

                WatchEvent<Path> ev = (WatchEvent<Path>)event;
                Path filename = ev.context();
                Path child = dir.resolve(filename);
                System.out.println("New file: "+child.toString()+" created.");
                try{
                    FileInputStream in = new FileInputStream(child.toFile());
                    System.out.println("File opened for reading");
                    in.close();
                    System.out.println("File Closed");
                }catch(Exception x){
                    x.printStackTrace();
                }
            }

            boolean valid = key.reset();
            if (!valid) {
                break;
            }
        }
    }
}

When I create file inside "mk" directory, I am getting notification for that. But when I copy some file in this directory, I am getting exception on opening that copied file.

My guess is Windows Copier dialog has still locked that file and I was unable to open that file. So basically I want to know is how to get notified for file has been closed by other process.

Output of above code is like -

Watching directory: c:\mk
New file: c:\mk\New Text Document (2).txt created.
File opened for reading
File Closed
New file: c:\mk\Config.class created.
java.io.FileNotFoundException: c:\mk\Config.class (The process cannot access the file because it is being used by another process)
        at java.io.FileInputStream.open(Native Method)
        at java.io.FileInputStream.<init>(FileInputStream.java:138)
        at watcher.Watch.main(Watch.java:36)
New file: c:\mk\New Text Document (3).txt created.
File opened for reading
File Closed

Files "New Text Document (2).txt" and "New Text Document (3).txt" I have created but file "Config.class" I have copied from other directory.

Please help me on this.

Answer 1

I got this working by implementing algorithm: Watcher thread will put file names in BlockingQueue and other thread will poll this queue, takes file names, try few times to open file. If file gets opened, Windows Copier has released file lock and we can proceed. So when other threads finds file has been unlocked, other thread will put this file name in processed queue, from where my application will retrieve file name. Also the other thread while checking for file unlock by opening file, if it runs long time for unlocking file, we can place back this file name in BlockingQueue and process other file names, former can be processed later.

Solution: Hope this may help to other:

package dirwatch;

import java.nio.file.*;
import static java.nio.file.StandardWatchEventKinds.ENTRY_CREATE;
import static java.nio.file.StandardWatchEventKinds.OVERFLOW;
import static java.nio.file.LinkOption.*;
import java.nio.file.attribute.*;
import java.io.*;
import java.util.*;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;

public class WatchDir {
    private final WatchService watcher;
    private final Map<WatchKey,Path> keys;
    private final boolean recursive;
    private boolean trace = false;

    private BlockingQueue<String> fileProcessingQueue;

    //******* processedFileQueue **** will be used by other threads to retrive unlocked files.. so I have 
    // kept as public final
    public final BlockingQueue<String> processedFileQueue;
    private volatile boolean closeProcessingThread;
    private volatile boolean closeWatcherThread;


    private void processFiles(){
        System.out.println("DirWatchProcessingThread Started");
        String fileName;
        outerLoop: while(!closeProcessingThread || !fileProcessingQueue.isEmpty()){
            try{
                fileName = fileProcessingQueue.poll(1000, TimeUnit.MILLISECONDS);
            }catch(InterruptedException ie){
                fileName = null;
            }

            if(fileName == null || fileName.equals("")){
                continue outerLoop;
            }

            long startTime = System.currentTimeMillis();
            innerLoop: while(true){
                FileInputStream fis = null;
                File file = new File(fileName);
                try{
                    fis = new FileInputStream(fileName);
                    break innerLoop;
                }catch(FileNotFoundException fnfe){
                    if(!file.exists() || file.isDirectory()){
                        System.out.println("File: '"+fileName+"has been deleted in file system or it is not file. Not processing this file.");
                        continue outerLoop;
                    }
                    try{
                        Thread.sleep(WatchDirParameters.millisToPuaseForFileLock);
                    }catch(InterruptedException ie){
                    }
                    if((System.currentTimeMillis() - startTime) > WatchDirParameters.millisToSwapFileForUnlocking){
                        if(fileProcessingQueue.offer(fileName)){
                            continue outerLoop;
                        }else{
                            startTime = System.currentTimeMillis();
                            continue innerLoop;
                        }
                    }
                }finally{
                    if(fis != null){
                        try{
                            fis.close();
                        }catch(IOException ioe){
                            ioe.printStackTrace();
                        }
                    }
                }
            }

            System.out.println("Queuing File: "+fileName);
            processedLoop:while(true){
                try{
                    if(processedFileQueue.offer(fileName, 1000, TimeUnit.MILLISECONDS)){
                        break processedLoop;
                    }
                }catch(InterruptedException ie){
                    //ie.printStackTrace();
                }
            }
        }
        closeWatcherThread = true;
        closeProcessingThread = true;
        System.out.println("DirWatchProcessingThread Exited");
    }

    /**
     * Process all events for keys queued to the watcher
     */
    private void processEvents(){
        System.out.println("DirWatcherThread started.");
        while(!closeWatcherThread) {
            // wait for key to be signalled
            WatchKey key;
            try {
                key = watcher.take();
            } catch (InterruptedException x) {
                // if we are returning from these method, it means we no longer wants to watch directory
                // we must close thread which may be waiting for file names in queue
                continue;
            }catch(ClosedWatchServiceException cwse){
                break;
            }

            Path dir = keys.get(key);
            if (dir == null) {
                System.err.println("WatchKey not recognized!!");
                continue;
            }

            try{
                for (WatchEvent<?> event: key.pollEvents()) {
                    WatchEvent.Kind kind = event.kind();

                    if (kind == OVERFLOW) {
                        continue;
                    }

                    // Context for directory entry event is the file name of entry
                    WatchEvent<Path> ev = cast(event);
                    Path name = ev.context();
                    Path child = dir.resolve(name);
                    if(kind.equals(ENTRY_CREATE)){
                        // if directory is created, and watching recursively, then
                        // register it and its sub-directories
                        if (recursive) {
                            try {
                                if (Files.isDirectory(child, NOFOLLOW_LINKS)) {
                                    registerAll(child);
                                    continue;
                                }
                            } catch (IOException x) {
                                // ignore to keep sample readbale
                            }
                        }
                        while(true){
                            if(fileProcessingQueue.remainingCapacity() < 2){
                                // if only one last can be inserted then don't queue this we need 1 empty space in queue
                                // for swaping file names..
                                // sleep for some time so processing thread may have made some rooms to queue in fileQueue
                                // this logic will not create any problems as only one this thread is inserting in queue
                                try{
                                    Thread.sleep(200);
                                }catch(InterruptedException ie){
                                }
                                continue;
                            }
                            if(!fileProcessingQueue.offer(child.toString())){
                                // couldn't queue this element by whatever reason.. we will try to enqueue again by continuing loop
                                continue;
                            }else{
                                // file name has been queued in queue
                                break;
                            }
                        }
                    }
                }
                // reset key and remove from set if directory no longer accessible
                boolean valid = key.reset();
                if (!valid) {
                    keys.remove(key);

                    // all directories are inaccessible
                    if (keys.isEmpty()) {
                        break;
                    }
                }
            }catch(ClosedWatchServiceException cwse){
                break;
            }

        }
        closeProcessingThread = true;
        closeWatcherThread = true;
        System.out.println("DirWatcherThread exited.");
    }

    public void stopWatching(){
        try{
            watcher.close();
        }catch(IOException ioe){
        }
        closeProcessingThread = true;
        closeWatcherThread = true;
    }

    public static WatchDir watchDirectory(String dirName, boolean recursive) throws InvalidPathException, IOException, Exception{
        try{
            Path dir = Paths.get(dirName);
            final WatchDir watchDir = new WatchDir(dir, recursive);
            watchDir.closeProcessingThread = false;
            watchDir.closeWatcherThread = false;
            new Thread(new Runnable() {
                public void run() {
                    watchDir.processFiles();
                }
            }, "DirWatchProcessingThread").start();
            new Thread(new Runnable() {
                public void run() {
                    watchDir.processEvents();
                }
            }, "DirWatcherThread").start();
            return watchDir;
        }catch(InvalidPathException ipe){
            throw ipe;
        }catch(IOException ioe){
            throw ioe;
        }catch(Exception e){
            throw e;
        }
    }

    @SuppressWarnings("unchecked")
    private static <T> WatchEvent<T> cast(WatchEvent<?> event) {
        return (WatchEvent<T>)event;
    }

    /**
     * Register the given directory with the WatchService
     */
    private void register(Path dir) throws IOException {
        //WatchKey key = dir.register(watcher, ENTRY_CREATE, ENTRY_DELETE, ENTRY_MODIFY);
        WatchKey key = dir.register(watcher, ENTRY_CREATE);
        if (trace) {
            Path prev = keys.get(key);
            if (prev == null) {
                System.out.format("register: %s\n", dir);
            } else {
                if (!dir.equals(prev)) {
                    System.out.format("update: %s -> %s\n", prev, dir);
                }
            }
        }
        keys.put(key, dir);
    }

    /**
     * Register the given directory, and all its sub-directories, with the
     * WatchService.
     */
    private void registerAll(final Path start) throws IOException {
        // register directory and sub-directories
        Files.walkFileTree(start, new SimpleFileVisitor<Path>() {
            @Override
            public FileVisitResult preVisitDirectory(Path dir, BasicFileAttributes attrs) throws IOException {
                register(dir);
                return FileVisitResult.CONTINUE;
            }
        });
    }

    /**
     * Creates a WatchService and registers the given directory
     */
    private WatchDir(Path dir, boolean recursive) throws IOException {
        fileProcessingQueue = new ArrayBlockingQueue<String>(WatchDirParameters.fileQueueSize, false);
        processedFileQueue = new ArrayBlockingQueue<String>(WatchDirParameters.fileQueueSize, false);
        this.watcher = FileSystems.getDefault().newWatchService();
        this.keys = new HashMap<WatchKey,Path>();
        this.recursive = recursive;
        //CreateTxtFile.createFile(dir, 1);
        if (recursive) {
            System.out.format("Scanning %s ...\n", dir);
            registerAll(dir);
            System.out.println("Done.");
        } else {
            register(dir);
        }

        // enable trace after initial registration
        this.trace = true;
    }
}

Parameter Class:

package dirwatch;

public class WatchDirParameters {
    public static final int millisToPuaseForFileLock = 200;
    public static final int fileQueueSize = 500;
    public static final int millisToSwapFileForUnlocking = 2000;
}

Answer 2

Made an updated version of the file provided by @UDPLover that is built for use in a high rate of file access environment I converted the to process queue into a HashMap<String, WatchEvent> to carry the watch event over to pass to an abstract method inside of the file blocking checker itself. Also made a print() method that allows anything printed to console by the WatchCore to be enabled or disabled. The file polling for loop from the original example was updated to use the JDK8 function for loop, made all parts threaded/interrupted. This is untested as of yet, will update with fixes when I can test it.

package filewatcher;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.PrintStream;
import java.nio.file.ClosedWatchServiceException;
import java.nio.file.FileSystems;
import java.nio.file.FileVisitResult;
import java.nio.file.Files;
import static java.nio.file.LinkOption.NOFOLLOW_LINKS;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.SimpleFileVisitor;
import static java.nio.file.StandardWatchEventKinds.ENTRY_CREATE;
import static java.nio.file.StandardWatchEventKinds.OVERFLOW;
import java.nio.file.WatchEvent;
import java.nio.file.WatchKey;
import java.nio.file.WatchService;
import java.nio.file.attribute.BasicFileAttributes;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 *
 * @author Nackloose
 * http://stackoverflow.com/questions/13998379/directory-watching-for-changes-in-java
 */
public abstract class WatchCore extends Thread {

    //make class a thread by default
    /**
     * After the WatchCore recieves an event for a file and deems it unlocked,
     * it will be passed to this function
     *
     * @param e WatchEvent for the file, after it has been affirmed to be
     * unlocked.
     */
    public abstract void onEventAndUnlocked(WatchEvent e);
    private final WatchService watcher;
    private final Map<WatchKey, Path> keys;
    private final boolean recursive;
    private boolean trace = false;
    //converted to HashMap to remove the limitation as I need this in a high rate of file access enviroment.
    //as well as to carry the event passed for that folder into the block check itself.
    //got rid of the finished queue and made events pass to the abstract void above 
    private final HashMap<String, WatchEvent> fileProcessingQueue;
    //create a varible to keep track of the thread checking the file blocking, so we can start and stop it.
    private final WatchBlocker blocker;

    public WatchCore(String dir) throws IOException {
        //defaultly  dont recurse
        this(dir, false);
    }

    public WatchCore(String dir, boolean recursive) throws IOException {
        this(Paths.get(dir), recursive);

    }

    public WatchCore(Path dir) throws IOException {
        //defaultly  dont recurse
        this(dir, false);
    }

    public WatchCore(Path dir, boolean recursive) throws IOException {
        fileProcessingQueue = new HashMap<>();
        this.watcher = FileSystems.getDefault().newWatchService();
        this.keys = new HashMap<>();
        this.recursive = recursive;
        //CreateTxtFile.createFile(dir, 1);
        if (recursive) {
            print("Scanning %s ...", dir);
            registerAll(dir);
            print("Done.");
        } else {
            register(dir);
        }
        // enable trace after initial registration
        this.trace = true;
        //start the thread to process files to be checked for file blocking
        blocker = new WatchBlocker();
    }

    @SuppressWarnings("unchecked")
    private static <T> WatchEvent<T> cast(WatchEvent<?> event) {
        return (WatchEvent<T>) event;
    }

    @Override
    public synchronized void start() {
        //start up our thread _FIRST_
        super.start();
        //then start the blocking thread
        blocker.start();
    }

    @Override
    public void interrupt() {
        //Everything backwards, stop the blocker _FIRST_
        blocker.interrupt();
        //then stop our thread.
        super.interrupt();
    }

    /**
     * Register the given directory with the WatchService
     */
    private void register(Path dir) throws IOException {
        //WatchKey key = dir.register(watcher, ENTRY_CREATE, ENTRY_DELETE, ENTRY_MODIFY);
        WatchKey key = dir.register(watcher, ENTRY_CREATE);
        if (trace) {
            Path prev = keys.get(key);
            if (prev == null) {
                print("register: %s\n", dir);
            } else {
                if (!dir.equals(prev)) {
                    print("update: %s -> %s\n", prev, dir);
                }
            }
        }
        keys.put(key, dir);
    }

    /**
     * Register the given directory, and all its sub-directories, with the
     * WatchService.
     */
    private void registerAll(final Path start) throws IOException {
        // register directory and sub-directories
        Files.walkFileTree(start, new SimpleFileVisitor<Path>() {
            @Override
            public FileVisitResult preVisitDirectory(Path dir, BasicFileAttributes attrs) throws IOException {
                register(dir);
                return FileVisitResult.CONTINUE;
            }
        });
    }

    /**
     * Process all events for keys queued to the watcher
     */
    @Override
    public void run() {
        //this was previous called processEvents()
        //pruned any un-nessicary continues, labels, and labels on breaks, a lot of them
        //were redundant
        print("DirWatcherThread started.");
        //as long as we're not interrupted we keep working
        while (!interrupted()) {
            // wait for key to be signalled
            WatchKey key;
            try {
                key = watcher.take();
            } catch (InterruptedException x) {
                // if we are returning from these method, it means we no longer wants to watch directory
                // we must close thread which may be waiting for file names in queue
                continue;
            } catch (ClosedWatchServiceException cwse) {
                break;
            }

            Path dir = keys.get(key);
            if (dir == null) {
                printe("WatchKey not recognized!!");
                continue;
            }

            try {
                //converted to functional for loop.
                key.pollEvents().stream().filter((event) -> {
                    WatchEvent.Kind kind = event.kind();
                    return !(kind == OVERFLOW); //make sure we do the filter
                }).forEach((event) -> {
                    WatchEvent.Kind kind = event.kind();
                    // Context for directory entry event is the file name of entry
                    WatchEvent<Path> ev = cast(event);
                    Path name = ev.context();
                    Path child = dir.resolve(name);
                    if (kind.equals(ENTRY_CREATE)) {
                        // if directory is created, and watching recursively, then
                        // register it and its sub-directories
                        if (recursive) {
                            try {
                                if (Files.isDirectory(child, NOFOLLOW_LINKS)) {
                                    registerAll(child);
                                    return; //continue;
                                }
                            } catch (IOException x) {
                                // ignore to keep sample readbale
                            }
                        }
                        fileProcessingQueue.put(child.toString(), ev);
                    }
                });
                // reset key and remove from set if directory no longer accessible
                boolean valid = key.reset();
                if (!valid) {
                    keys.remove(key);

                    // all directories are inaccessible
                    if (keys.isEmpty()) {
                        break;
                    }
                }
            } catch (ClosedWatchServiceException cwse) {
                break;
            }

        }
        print("DirWatcherThread exited.");
    }

    /**
     *
     * @author
     * http://stackoverflow.com/questions/13998379/directory-watching-for-changes-in-java
     * Nackloose
     */
    private class WatchBlocker extends Thread {

        @Override
        public synchronized void start() {
            //get it going
            super.start();
        }

        @Override
        public void interrupt() {
            //interupt our thread
            super.interrupt();
        }

        @Override
        public void run() {
            //this was perviously processFiles()
            //pruned any un-nessicary continues, labels, and labels on breaks, a lot of them
            //were redundant
            print("DirWatchProcessingThread Started");
            Entry<String, WatchEvent> fileEvent;
            outerLoop:
            //as long as we're not interrupted we keep working
            while (!interrupted()) {
                if (fileProcessingQueue.isEmpty()) {
                    try {
                        Thread.sleep(WatchCoreParameters.timeToIdle);
                    } catch (InterruptedException ex) {
                        Logger.getLogger(WatchCore.class.getName()).log(Level.SEVERE, null, ex);
                    }
                    continue;
                }
                fileEvent = fileProcessingQueue.entrySet().iterator().next();
                fileProcessingQueue.remove(fileEvent.getKey());

                long startTime = System.currentTimeMillis();
                while (true) {
                    FileInputStream fis = null;
                    File file = new File(fileEvent.getKey());
                    try {
                        fis = new FileInputStream(fileEvent.getKey());
                        break;
                    } catch (FileNotFoundException fnfe) {
                        if (!file.exists() || file.isDirectory()) {
                            print("File: '" + fileEvent + "has been deleted in file system or it is not file. Not processing this file.");
                            continue outerLoop;
                        }
                        try {
                            Thread.sleep(WatchCoreParameters.millisToPauseForFileLock);
                        } catch (InterruptedException ie) {
                        }
                        if ((System.currentTimeMillis() - startTime) > WatchCoreParameters.millisToSwapFileForUnlocking) {
                            fileProcessingQueue.put(fileEvent.getKey(), fileEvent.getValue());
                        }
                    } finally {
                        if (fis != null) {
                            try {
                                fis.close();
                            } catch (IOException ioe) {
                                ioe.printStackTrace();
                            }
                        }
                    }
                }
                print("Queuing File: " + fileEvent);
                //pass the unlocked file event to the abstract method
                onEventAndUnlocked(fileEvent.getValue());
            }
            print("DirWatchProcessingThread Exited");
        }
    }

    /**
     *
     * @author
     * http://stackoverflow.com/questions/13998379/directory-watching-for-changes-in-java
     * Nackloose
     */
    public static class WatchCoreParameters {

        public static int timeToIdle = 2000, millisToPauseForFileLock = 200,
                millisToSwapFileForUnlocking = 2000;
        public static boolean verbose = false;

    }

    //<editor-fold defaultstate="collapsed" desc="Printing methods">
    private void print(String s) {
        //defaultly we're not writing an error
        print(s, false);
    }

    public static final void print(String s, boolean error) {
        //check verbosity, exit if none.
        if (!WatchCoreParameters.verbose) {
            return;
        }
        //if this is an error, assign System.err to a temp varible
        //otherise assign System.out for normal printing
        PrintStream out = (!error ? System.out : System.err);
        if (s.contains("\n")) { // check to see if theirs a new line
            out.print(s); //print accordingly
        } else {
            out.println(s); //print accordingly
        }
    }

    public static final void printe(String s) {
        //shortcut/convenience method for printing an error
        print(s, true);
    }

    public static final void print(String s, Object... formatObj) {
        //check verbosity, exit if none.
        if (!WatchCoreParameters.verbose) {
            return;
        }
        //format the object into the string, and if no newline is there, add it.
        System.out.format(s + (s.contains("\n") ? "" : "\n"), formatObj);
    }
//</editor-fold>
}