Java 类java.nio.file.ClosedWatchServiceException 实例源码

@Override
protected String nextEvent() throws IOException, InterruptedException {
    WatchKey key;
    try {
        key = watcher.take();
    } catch (ClosedWatchServiceException cwse) { // #238261
        @SuppressWarnings({"ThrowableInstanceNotThrown"})
        InterruptedException ie = new InterruptedException();
        throw (InterruptedException) ie.initCause(cwse);
    }
    Path dir = (Path)key.watchable();

    String res = dir.toAbsolutePath().toString();
    for (WatchEvent<?> event: key.pollEvents()) {
        if (event.kind() == OVERFLOW) {
            // full rescan
            res = null;
        }
    }
    key.reset();
    return res;
}

/**
 * Keep polling for a short time: when (multiple) directories get deleted the watch keys might
 * arrive just a bit later
 */
private void pollForMoreChanges() throws ClosedWatchServiceException, InterruptedException {
    boolean keepPolling = true;
    List<WatchKey> polledKeys = new ArrayList<>();
    final long startPolling = System.currentTimeMillis();
    while (keepPolling) {
        log.debug("Waiting {} ms for more changes...", POLLING_TIME_MILLIS);
        WatchKey key = watcher.poll(POLLING_TIME_MILLIS, TimeUnit.MILLISECONDS);
        if (key == null) {
            keepPolling = false;
        } else {
            log.debug("Found change for '{}' found during extra polling time", key.watchable());
            polledKeys.add(key);
        }
    }
    log.debug("Polled '{}' more changes during '{}' ms", polledKeys.size(), String.valueOf(System.currentTimeMillis() - startPolling));
    for (WatchKey polledKey : polledKeys) {
        processWatchKey(polledKey);
    }
}

/**
 * Polls the given WatchService in a tight loop. This keeps the event
 * queue drained, it also hogs a CPU core which seems necessary to
 * tickle the original bug.
 */
static void poll(WatchService watcher) {
    try {
        for (;;) {
            WatchKey key = watcher.take();
            if (key != null) {
                key.pollEvents();
                key.reset();
            }
        }
    } catch (ClosedWatchServiceException expected) {
        // nothing to do
    } catch (Exception e) {
        e.printStackTrace();
        failed = true;
    }
}

/**
 * Polls the given WatchService in a tight loop. This keeps the event
 * queue drained, it also hogs a CPU core which seems necessary to
 * tickle the original bug.
 */
static void poll(int id, WatchService watcher) {
    System.out.printf("begin poll %d%n", id);
    try {
        for (;;) {
            WatchKey key = watcher.take();
            if (key != null) {
                key.pollEvents();
                key.reset();
            }
        }
    } catch (ClosedWatchServiceException expected) {
        // nothing to do but print
        System.out.printf("poll %d expected exception %s%n", id, expected);
    } catch (Exception e) {
        e.printStackTrace();
        failed = true;
    }
    System.out.printf("end poll %d%n", id);
}

/**
 * Polls the given WatchService in a tight loop. This keeps the event
 * queue drained, it also hogs a CPU core which seems necessary to
 * tickle the original bug.
 */
static void poll(WatchService watcher) {
    try {
        for (;;) {
            WatchKey key = watcher.take();
            if (key != null) {
                key.pollEvents();
                key.reset();
            }
        }
    } catch (ClosedWatchServiceException expected) {
        // nothing to do
    } catch (Exception e) {
        e.printStackTrace();
        failed = true;
    }
}

/**
 * Polls the given WatchService in a tight loop. This keeps the event
 * queue drained, it also hogs a CPU core which seems necessary to
 * tickle the original bug.
 */
static void poll(WatchService watcher) {
    try {
        for (;;) {
            WatchKey key = watcher.take();
            if (key != null) {
                key.pollEvents();
                key.reset();
            }
        }
    } catch (ClosedWatchServiceException expected) {
        // nothing to do
    } catch (Exception e) {
        e.printStackTrace();
        failed = true;
    }
}

/**
 * Polls the given WatchService in a tight loop. This keeps the event
 * queue drained, it also hogs a CPU core which seems necessary to
 * tickle the original bug.
 */
static void poll(WatchService watcher) {
    try {
        for (;;) {
            WatchKey key = watcher.take();
            if (key != null) {
                key.pollEvents();
                key.reset();
            }
        }
    } catch (ClosedWatchServiceException expected) {
        // nothing to do
    } catch (Exception e) {
        e.printStackTrace();
        failed = true;
    }
}

@Test
@Category( { SlowTest.class, Watchable.class, Writable.class } )
@SuppressWarnings( "PMD.EmptyCatchBlock" )
public void testWatchServiceTakeBlocks() throws Exception {
    Path dir = dirTA();
    final WatchService watcher = dir.getFileSystem().newWatchService();
    dir.register( watcher, ENTRY_CREATE );

    final Ref<Boolean> interrupted = Ref.valueOf( false );

    new Thread( () -> {
        try {
            watcher.take();
        } catch( InterruptedException | ClosedWatchServiceException e ) {
            // nothing to do
        } finally {
            interrupted.set( true );

        }
    } ).start();

    Thread.sleep( 1000 );

    assertThat( interrupted.get() ).isFalse();

}

private synchronized void handleChanges() {
    final WatchKey key;
    try {
        key = this.watchService.take();
    } catch (final InterruptedException | ClosedWatchServiceException e) {
        if (!ConfigLoaderImpl.this.closed) {
            LOG.warn(INTERRUPTED, e);
            Thread.currentThread().interrupt();
        }
        return;
    }

    if (key != null) {
        for (final WatchEvent<?> event : key.pollEvents()) {
            handleEvent(event.context().toString());
        }
        final boolean reset = key.reset();
        if (!reset) {
            LOG.warn("Could not reset the watch key.");
        }
    }
}

public Watcher(final BiConsumer<Kind<?>, Path> listener, final Path... dirs)
    throws IOException {
  this.watcher = FileSystems.getDefault().newWatchService();
  this.keys = new HashMap<WatchKey, Path>();
  this.listener = listener;
  for (Path dir : dirs) {
    registerAll(dir);
  }

  this.scanner = new Thread(() -> {
    boolean process = true;
    listener.accept(ENTRY_MODIFY, dirs[0]);
    try {
    while (process) {
      process = processEvents();
    }
    } catch (ClosedWatchServiceException ex) {
      log.trace("watch service closed", ex);
    }
  }, "asset-compiler");

  scanner.setDaemon(true);
}

/**
 * Tests {@link FileSystemFactory#newWatchService()}.
 *
 * @throws ClosedWatchServiceException if the newly-created service is closed
 * @throws IOException if the test fails
 * @throws InterruptedException hopefully never
 */
@Test
public void testNewSingleThreadExecutor() throws ClosedWatchServiceException, IOException, InterruptedException {
    WatchService service = new FileSystemFactory().newWatchService();
    try {
        assertNotNull(service);
        assertNull(service.poll()); // verifies the service is not closed
    } finally {
        if (service != null) {
            try {
                service.close();
            }catch(IOException ex) {
                //trap
            }
        }
    }
}

@Override
public WatchKey poll() {
    if (!running) {
        throw new ClosedWatchServiceException();
    }
    WatchKey pending = popPending();
    if (null != pending) {
        return pending;
    }
    for (GlusterWatchKey k : paths) {
        if (k.isValid() && k.isReady() && k.update()) {
            pendingPaths.add(k);
        }
    }
    return popPending();
}

@Override
public WatchKey poll(long timeout, TimeUnit unit) {
    long timeoutMillis = timeoutToMillis(timeout, unit);
    long loops = 0;
    while (running) {
        WatchKey key = poll();
        if (key != null) {
            return key;
        }
        if ((loops * PERIOD) > timeoutMillis) {
            return null;
        }
        loops++;
        try {
            Thread.sleep(PERIOD);
        } catch (InterruptedException e) {
        }
    }
    throw new ClosedWatchServiceException();
}

private void pumpEvents() throws InterruptedException {
    while (isRunning()) {
        try {
            List<FileWatcherEvent> events = poller.takeEvents();
            if (events != null) {
                deliverEvents(events);
            }
        } catch (ClosedWatchServiceException e) {
            LOGGER.debug("Received ClosedWatchServiceException, stopping");
            stop();
        }
    }
}

@Override
protected WatchKey addWatch(String pathStr) throws IOException {
    Path path = Paths.get(pathStr);
    try {
        WatchKey key = path.register(watcher, ENTRY_CREATE, ENTRY_DELETE, ENTRY_MODIFY);
        return key;
    } catch (ClosedWatchServiceException ex) {
        throw new IOException(ex);
    }
}

@Override
protected void removeWatch(WatchKey key) throws IOException {
    try {
        key.cancel();
    } catch (ClosedWatchServiceException ex) {
        throw new IOException(ex);
    }
}

protected void scanner ()
{
    logger.trace ( "Watching for events" );
    while ( true )
    {
        WatchKey key = null;
        try
        {
            key = this.ws.take ();
            logger.trace ( "Took events: {}", key.watchable () );

            final List<WatchEvent<?>> events = key.pollEvents ();
            for ( final WatchEvent<?> evt : events )
            {
                processEvent ( evt );
            }
        }
        catch ( final InterruptedException | ClosedWatchServiceException e )
        {
            return;
        }
        finally
        {
            if ( key != null )
            {
                key.reset ();
            }
        }
    }
}

@Override
public void run() {
    try {
        log.info("Watch started");
        while (true) {
            processChanges();
        }
    } catch (ClosedWatchServiceException e) {
        log.info("Watch closed", e);
    } finally {
        IOUtils.closeQuietly(watcher);
    }
}

private PollingWatchKey doPrivilegedRegister(Path path,
                                             Set<? extends WatchEvent.Kind<?>> events,
                                             int sensitivityInSeconds)
    throws IOException
{
    // check file is a directory and get its file key if possible
    BasicFileAttributes attrs = Files.readAttributes(path, BasicFileAttributes.class);
    if (!attrs.isDirectory()) {
        throw new NotDirectoryException(path.toString());
    }
    Object fileKey = attrs.fileKey();
    if (fileKey == null)
        throw new AssertionError("File keys must be supported");

    // grab close lock to ensure that watch service cannot be closed
    synchronized (closeLock()) {
        if (!isOpen())
            throw new ClosedWatchServiceException();

        PollingWatchKey watchKey;
        synchronized (map) {
            watchKey = map.get(fileKey);
            if (watchKey == null) {
                // new registration
                watchKey = new PollingWatchKey(path, this, fileKey);
                map.put(fileKey, watchKey);
            } else {
                // update to existing registration
                watchKey.disable();
            }
        }
        watchKey.enable(events, sensitivityInSeconds);
        return watchKey;
    }

}

/**
 * Creates a WatchService and registers the given directory
 */
private void setupWatch(String initialText) throws IOException {
    this.watcher = FileSystems.getDefault().newWatchService();
    this.dir = Files.createTempDirectory("extedit");
    this.tmpfile = Files.createTempFile(dir, null, ".java");
    Files.write(tmpfile, initialText.getBytes(Charset.forName("UTF-8")));
    dir.register(watcher,
            ENTRY_CREATE,
            ENTRY_DELETE,
            ENTRY_MODIFY);
    watchedThread = new Thread(() -> {
        for (;;) {
            WatchKey key;
            try {
                key = watcher.take();
            } catch (ClosedWatchServiceException ex) {
                // The watch service has been closed, we are done
                break;
            } catch (InterruptedException ex) {
                // tolerate an interrupt
                continue;
            }

            if (!key.pollEvents().isEmpty()) {
                saveFile();
            }

            boolean valid = key.reset();
            if (!valid) {
                // The watch service has been closed, we are done
                break;
            }
        }
    });
    watchedThread.start();
}

/**
 * Returns a task that updates the registration of a directory with
 * a WatchService.
 */
static Callable<Boolean> newRegisterTask(WatchService watcher, Path dir) {
    return () -> {
        try {
            dir.register(watcher, StandardWatchEventKinds.ENTRY_DELETE);
            return true;
        } catch (ClosedWatchServiceException e) {
            return false;
        } catch (IOException ioe) {
            throw new UncheckedIOException(ioe);
        }
    };
}

/**
 * Creates a WatchService and registers the given directory
 */
private void setupWatch(final String initialText) throws IOException {
    this.watcher = FileSystems.getDefault().newWatchService();
    this.dir = Files.createTempDirectory("REPL");
    this.tmpfile = Files.createTempFile(dir, null, ".js");
    Files.write(tmpfile, initialText.getBytes(Charset.forName("UTF-8")));
    dir.register(watcher,
            ENTRY_CREATE,
            ENTRY_DELETE,
            ENTRY_MODIFY);
    watchedThread = new Thread(() -> {
        for (;;) {
            WatchKey key;
            try {
                key = watcher.take();
            } catch (final ClosedWatchServiceException ex) {
                break;
            } catch (final InterruptedException ex) {
                continue; // tolerate an intrupt
            }

            if (!key.pollEvents().isEmpty()) {
                if (!input.terminalEditorRunning()) {
                    saveFile();
                }
            }

            boolean valid = key.reset();
            if (!valid) {
                errorHandler.accept("Invalid key");
                break;
            }
        }
    });
    watchedThread.start();
}

/**
 * Creates a WatchService and registers the given directory
 */
private void setupWatch(String initialText) throws IOException {
    this.watcher = FileSystems.getDefault().newWatchService();
    this.dir = Files.createTempDirectory("REPL");
    this.tmpfile = Files.createTempFile(dir, null, ".repl");
    Files.write(tmpfile, initialText.getBytes(Charset.forName("UTF-8")));
    dir.register(watcher,
            ENTRY_CREATE,
            ENTRY_DELETE,
            ENTRY_MODIFY);
    watchedThread = new Thread(() -> {
        for (;;) {
            WatchKey key;
            try {
                key = watcher.take();
            } catch (ClosedWatchServiceException ex) {
                break;
            } catch (InterruptedException ex) {
                continue; // tolerate an intrupt
            }

            if (!key.pollEvents().isEmpty()) {
                if (!input.terminalEditorRunning()) {
                    saveFile();
                }
            }

            boolean valid = key.reset();
            if (!valid) {
                errorHandler.accept("Invalid key");
                break;
            }
        }
    });
    watchedThread.start();
}

/**
 * Returns a task that updates the registration of a directory with
 * a WatchService.
 */
static Callable<Boolean> newRegisterTask(WatchService watcher, Path dir) {
    return () -> {
        try {
            dir.register(watcher, StandardWatchEventKinds.ENTRY_DELETE);
            return true;
        } catch (ClosedWatchServiceException e) {
            return false;
        } catch (IOException ioe) {
            throw new UncheckedIOException(ioe);
        }
    };
}

/**
 * Creates a WatchService and registers the given directory
 */
private void setupWatch(final String initialText) throws IOException {
    this.watcher = FileSystems.getDefault().newWatchService();
    this.dir = Files.createTempDirectory("REPL");
    this.tmpfile = Files.createTempFile(dir, null, ".js");
    Files.write(tmpfile, initialText.getBytes(Charset.forName("UTF-8")));
    dir.register(watcher,
            ENTRY_CREATE,
            ENTRY_DELETE,
            ENTRY_MODIFY);
    watchedThread = new Thread(() -> {
        for (;;) {
            WatchKey key;
            try {
                key = watcher.take();
            } catch (final ClosedWatchServiceException ex) {
                break;
            } catch (final InterruptedException ex) {
                continue; // tolerate an intrupt
            }

            if (!key.pollEvents().isEmpty()) {
                if (!input.terminalEditorRunning()) {
                    saveFile();
                }
            }

            boolean valid = key.reset();
            if (!valid) {
                errorHandler.accept("Invalid key");
                break;
            }
        }
    });
    watchedThread.start();
}

public void processEvents() {
    try {
        for (WatchDir entry : watchers) {
            entry.processEvents();
        }
    } catch (ClosedWatchServiceException e) {
        // Ignore these rare temporary event
    }
}

/**
 * Creates a WatchService and registers the given directory
 */
private void setupWatch(String initialText) throws IOException {
    this.watcher = FileSystems.getDefault().newWatchService();
    this.dir = Files.createTempDirectory("REPL");
    this.tmpfile = Files.createTempFile(dir, null, ".repl");
    Files.write(tmpfile, initialText.getBytes(Charset.forName("UTF-8")));
    dir.register(watcher,
            ENTRY_CREATE,
            ENTRY_DELETE,
            ENTRY_MODIFY);
    watchedThread = new Thread(() -> {
        for (;;) {
            WatchKey key;
            try {
                key = watcher.take();
            } catch (ClosedWatchServiceException ex) {
                break;
            } catch (InterruptedException ex) {
                continue; // tolerate an intrupt
            }

            if (!key.pollEvents().isEmpty()) {
                if (!input.terminalEditorRunning()) {
                    saveFile();
                }
            }

            boolean valid = key.reset();
            if (!valid) {
                errorHandler.accept("Invalid key");
                break;
            }
        }
    });
    watchedThread.start();
}

@Override
public WatchKey poll()
{
  if (_isClosed.get()) {
    throw new ClosedWatchServiceException();
  }

  return _eventQueue.poll();
}

@Override
public WatchKey poll(long timeout, TimeUnit unit) throws InterruptedException
{
  if (_isClosed.get()) {
    throw new ClosedWatchServiceException();
  }

  return _eventQueue.poll(timeout, unit);
}

@Override
public WatchKey take() throws InterruptedException
{
  if (_isClosed.get()) {
    throw new ClosedWatchServiceException();
  }

  return _eventQueue.take();
}

/**
 * Starts the watcher service and registers watches in all of the sub-folders of
 * the given root folder.
 *
 * <p>This method calls the {@link #beforeStart()} method before everything else.
 * Subclasses may execute their own commands there. Before the watch thread is started,
 * {@link #beforePollEventLoop()} is called. And in the watch thread loop,
 * {@link #pollEvents()} is called.
 *
 * <p><b>Important:</b> This method returns immediately, even though the watches
 * might not be in place yet. For large file trees, it might take several seconds
 * until all directories are being monitored. For normal cases (1-100 folders), this
 * should not take longer than a few milliseconds.
 */
public void start() throws Exception {
    // Call before-start hook
    beforeStart();

    // Start watcher thread
    watchThread = new Thread(() -> {
           running.set(true);
           beforePollEventLoop(); // Call before-loop hook

           while (running.get()) {
               try {
                   boolean relevantEvents = pollEvents();

                   if (relevantEvents) {
                       restartWaitSettlementTimer();
                   }
               }
               catch (InterruptedException e) {
                   logger.log(Level.FINE, "Could not poll the events. EXITING watcher.", e);
                   running.set(false);
               }
               catch (ClosedWatchServiceException e) {
                   logger.log(Level.FINE, "Watch closed or polling failed. EXITING watcher.", e);
                   running.set(false);
               }
           }
       }, "Watcher/" + root.toFile().getName());

    watchThread.start();
}

/**
 * Returns the given key, throwing an exception if it's the poison.
 */
@Nullable
private WatchKey check(@Nullable WatchKey key) {
  if (key == poison) {
    // ensure other blocking threads get the poison
    queue.offer(poison);
    throw new ClosedWatchServiceException();
  }
  return key;
}

@Override
public WatchKey poll( long timeout, TimeUnit unit ) throws InterruptedException {
    try {
        Thread current = Thread.currentThread();
        waiting.add( current );
        WatchKey ret = que.pollLast( timeout, unit );
        waiting.remove( current );
        return ret;
    } catch( InterruptedException ex ) {
        throw new ClosedWatchServiceException();
    }
}

@Override
public WatchKey take() throws InterruptedException {
    try {
        Thread current = Thread.currentThread();
        waiting.add( current );
        WatchKey ret = que.takeLast();
        waiting.remove( current );
        return ret;
    } catch( InterruptedException ex ) {
        throw new ClosedWatchServiceException();
    }

}

@Test(expected=ClosedWatchServiceException.class)
@Ignore
public void testSimpleEx() throws IOException {
  Path rootPath = Paths.get(clusterUri);

  WatchService watcher = rootPath.getFileSystem().newWatchService();
  rootPath.register(watcher, 
        new WatchEvent.Kind<?>[] { ENTRY_MODIFY });
  watcher.close();
  // Should throw ClosedWatchServiceException
  watcher.poll();
}

@Test( expected = ClosedWatchServiceException.class )
@Category( { Watchable.class, Writable.class } )
public void testRegisterOnClosedWatchService() throws IOException {
    WatchService watcher = FS.newWatchService();
    watcher.close();
    dirTAB().register( watcher, ENTRY_CREATE );
}

/**
 * Returns the given key, throwing an exception if it's the poison.
 */
@Nullable
private WatchKey check(@Nullable WatchKey key) {
  if (key == poison) {
    // ensure other blocking threads get the poison
    queue.offer(poison);
    throw new ClosedWatchServiceException();
  }
  return key;
}

public WatchKey registerPath(GlusterPath path, WatchEvent.Kind... kinds) {
    if (!running) {
        throw new ClosedWatchServiceException();
    }
    for (GlusterWatchKey k : paths) {
        if (k.getPath().equals(path)) {
            k.setKinds(kinds);
            return k;
        }
    }
    GlusterWatchKey key = new GlusterWatchKey(path, kinds);
    paths.add(key);
    return key;
}

@Override
public WatchKey take() {
    while (running) {
        WatchKey key = poll();
        if (key != null) {
            return key;
        }
        try {
            Thread.sleep(PERIOD);
        } catch (InterruptedException e) {
        }
    }
    throw new ClosedWatchServiceException();
}

@Test(expected = ClosedWatchServiceException.class)
public void testPollTimeout_whenClosed() {
    long timeout = 150L;
    TimeUnit unit = TimeUnit.MILLISECONDS;
    doReturn(timeout).when(watchService).timeoutToMillis(timeout, unit);
    watchService.setRunning(false);
    watchService.poll(timeout, unit);
}