jimfs/src/main/java/com/google/common/jimfs/AbstractWatchService.java - platform/external/jimfs - Git at Google

 /*
  * Copyright 2013 Google Inc.
  *
  * Licensed 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 com.google.common.jimfs;

 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkNotNull;
 import static java.nio.file.StandardWatchEventKinds.OVERFLOW;

 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.MoreObjects;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
 import java.io.IOException;
 import java.nio.file.ClosedWatchServiceException;
 import java.nio.file.WatchEvent;
 import java.nio.file.WatchKey;
 import java.nio.file.WatchService;
 import java.nio.file.Watchable;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.Objects;
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 import org.checkerframework.checker.nullness.compatqual.NullableDecl;

 /**
  * Abstract implementation of {@link WatchService}. Provides the means for registering and managing
  * keys but does not handle actually watching. Subclasses should implement the means of watching
  * watchables, posting events to registered keys and queueing keys with the service by signalling
  * them.
  *
  * @author Colin Decker
  */
 abstract class AbstractWatchService implements WatchService {

   private final BlockingQueue<WatchKey> queue = new LinkedBlockingQueue<>();
   private final WatchKey poison = new Key(this, null, ImmutableSet.<WatchEvent.Kind<?>>of());

   private final AtomicBoolean open = new AtomicBoolean(true);

   /**
    * Registers the given watchable with this service, returning a new watch key for it. This
    * implementation just checks that the service is open and creates a key; subclasses may override
    * it to do other things as well.
    */
   public Key register(Watchable watchable, Iterable<? extends WatchEvent.Kind<?>> eventTypes)
       throws IOException {
     checkOpen();
     return new Key(this, watchable, eventTypes);
   }

   /** Returns whether or not this watch service is open. */
   @VisibleForTesting
   public boolean isOpen() {
     return open.get();
   }

   /** Enqueues the given key if the watch service is open; does nothing otherwise. */
   final void enqueue(Key key) {
     if (isOpen()) {
       queue.add(key);
     }
   }

   /** Called when the given key is cancelled. Does nothing by default. */
   public void cancelled(Key key) {}

   @VisibleForTesting
   ImmutableList<WatchKey> queuedKeys() {
     return ImmutableList.copyOf(queue);
   }

   @NullableDecl
   @Override
   public WatchKey poll() {
     checkOpen();
     return check(queue.poll());
   }

   @NullableDecl
   @Override
   public WatchKey poll(long timeout, TimeUnit unit) throws InterruptedException {
     checkOpen();
     return check(queue.poll(timeout, unit));
   }

   @Override
   public WatchKey take() throws InterruptedException {
     checkOpen();
     return check(queue.take());
   }

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

   /** Checks that the watch service is open, throwing {@link ClosedWatchServiceException} if not. */
   protected final void checkOpen() {
     if (!open.get()) {
       throw new ClosedWatchServiceException();
     }
   }

   @Override
   public void close() {
     if (open.compareAndSet(true, false)) {
       queue.clear();
       queue.offer(poison);
     }
   }

   /** A basic implementation of {@link WatchEvent}. */
   static final class Event<T> implements WatchEvent<T> {

     private final Kind<T> kind;
     private final int count;

     @NullableDecl private final T context;

     public Event(Kind<T> kind, int count, @NullableDecl T context) {
       this.kind = checkNotNull(kind);
       checkArgument(count >= 0, "count (%s) must be non-negative", count);
       this.count = count;
       this.context = context;
     }

     @Override
     public Kind<T> kind() {
       return kind;
     }

     @Override
     public int count() {
       return count;
     }

     @NullableDecl
     @Override
     public T context() {
       return context;
     }

     @Override
     public boolean equals(Object obj) {
       if (obj instanceof Event) {
         Event<?> other = (Event<?>) obj;
         return kind().equals(other.kind())
             && count() == other.count()
             && Objects.equals(context(), other.context());
       }
       return false;
     }

     @Override
     public int hashCode() {
       return Objects.hash(kind(), count(), context());
     }

     @Override
     public String toString() {
       return MoreObjects.toStringHelper(this)
           .add("kind", kind())
           .add("count", count())
           .add("context", context())
           .toString();
     }
   }

   /** Implementation of {@link WatchKey} for an {@link AbstractWatchService}. */
   static final class Key implements WatchKey {

     @VisibleForTesting static final int MAX_QUEUE_SIZE = 256;

     private static WatchEvent<Object> overflowEvent(int count) {
       return new Event<>(OVERFLOW, count, null);
     }

     private final AbstractWatchService watcher;
     private final Watchable watchable;
     private final ImmutableSet<WatchEvent.Kind<?>> subscribedTypes;

     private final AtomicReference<State> state = new AtomicReference<>(State.READY);
     private final AtomicBoolean valid = new AtomicBoolean(true);
     private final AtomicInteger overflow = new AtomicInteger();

     private final BlockingQueue<WatchEvent<?>> events = new ArrayBlockingQueue<>(MAX_QUEUE_SIZE);

     public Key(
         AbstractWatchService watcher,
         @NullableDecl Watchable watchable,
         Iterable<? extends WatchEvent.Kind<?>> subscribedTypes) {
       this.watcher = checkNotNull(watcher);
       this.watchable = watchable; // nullable for Watcher poison
       this.subscribedTypes = ImmutableSet.copyOf(subscribedTypes);
     }

     /** Gets the current state of this key, State.READY or SIGNALLED. */
     @VisibleForTesting
     State state() {
       return state.get();
     }

     /** Gets whether or not this key is subscribed to the given type of event. */
     public boolean subscribesTo(WatchEvent.Kind<?> eventType) {
       return subscribedTypes.contains(eventType);
     }

     /**
      * Posts the given event to this key. After posting one or more events, {@link #signal()} must
      * be called to cause the key to be enqueued with the watch service.
      */
     public void post(WatchEvent<?> event) {
       if (!events.offer(event)) {
         overflow.incrementAndGet();
       }
     }

     /**
      * Sets the state to SIGNALLED and enqueues this key with the watcher if it was previously in
      * the READY state.
      */
     public void signal() {
       if (state.getAndSet(State.SIGNALLED) == State.READY) {
         watcher.enqueue(this);
       }
     }

     @Override
     public boolean isValid() {
       return watcher.isOpen() && valid.get();
     }

     @Override
     public List<WatchEvent<?>> pollEvents() {
       // note: it's correct to be able to retrieve more events from a key without calling reset()
       // reset() is ONLY for "returning" the key to the watch service to potentially be retrieved by
       // another thread when you're finished with it
       List<WatchEvent<?>> result = new ArrayList<>(events.size());
       events.drainTo(result);
       int overflowCount = overflow.getAndSet(0);
       if (overflowCount != 0) {
         result.add(overflowEvent(overflowCount));
       }
       return Collections.unmodifiableList(result);
     }

     @Override
     public boolean reset() {
       // calling reset() multiple times without polling events would cause key to be placed in
       // watcher queue multiple times, but not much that can be done about that
       if (isValid() && state.compareAndSet(State.SIGNALLED, State.READY)) {
         // requeue if events are pending
         if (!events.isEmpty()) {
           signal();
         }
       }

       return isValid();
     }

     @Override
     public void cancel() {
       valid.set(false);
       watcher.cancelled(this);
     }

     @Override
     public Watchable watchable() {
       return watchable;
     }

     @VisibleForTesting
     enum State {
       READY,
       SIGNALLED
     }
   }
 }
	/*
	* Copyright 2013 Google Inc.
	*
	* Licensed 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 com.google.common.jimfs;

	import static com.google.common.base.Preconditions.checkArgument;
	import static com.google.common.base.Preconditions.checkNotNull;
	import static java.nio.file.StandardWatchEventKinds.OVERFLOW;

	import com.google.common.annotations.VisibleForTesting;
	import com.google.common.base.MoreObjects;
	import com.google.common.collect.ImmutableList;
	import com.google.common.collect.ImmutableSet;
	import java.io.IOException;
	import java.nio.file.ClosedWatchServiceException;
	import java.nio.file.WatchEvent;
	import java.nio.file.WatchKey;
	import java.nio.file.WatchService;
	import java.nio.file.Watchable;
	import java.util.ArrayList;
	import java.util.Collections;
	import java.util.List;
	import java.util.Objects;
	import java.util.concurrent.ArrayBlockingQueue;
	import java.util.concurrent.BlockingQueue;
	import java.util.concurrent.LinkedBlockingQueue;
	import java.util.concurrent.TimeUnit;
	import java.util.concurrent.atomic.AtomicBoolean;
	import java.util.concurrent.atomic.AtomicInteger;
	import java.util.concurrent.atomic.AtomicReference;
	import org.checkerframework.checker.nullness.compatqual.NullableDecl;

	/**
	* Abstract implementation of {@link WatchService}. Provides the means for registering and managing
	* keys but does not handle actually watching. Subclasses should implement the means of watching
	* watchables, posting events to registered keys and queueing keys with the service by signalling
	* them.
	*
	* @author Colin Decker
	*/
	abstract class AbstractWatchService implements WatchService {

	private final BlockingQueue<WatchKey> queue = new LinkedBlockingQueue<>();
	private final WatchKey poison = new Key(this, null, ImmutableSet.<WatchEvent.Kind<?>>of());

	private final AtomicBoolean open = new AtomicBoolean(true);

	/**
	* Registers the given watchable with this service, returning a new watch key for it. This
	* implementation just checks that the service is open and creates a key; subclasses may override
	* it to do other things as well.
	*/
	public Key register(Watchable watchable, Iterable<? extends WatchEvent.Kind<?>> eventTypes)
	throws IOException {
	checkOpen();
	return new Key(this, watchable, eventTypes);
	}

	/** Returns whether or not this watch service is open. */
	@VisibleForTesting
	public boolean isOpen() {
	return open.get();
	}

	/** Enqueues the given key if the watch service is open; does nothing otherwise. */
	final void enqueue(Key key) {
	if (isOpen()) {
	queue.add(key);
	}
	}

	/** Called when the given key is cancelled. Does nothing by default. */
	public void cancelled(Key key) {}

	@VisibleForTesting
	ImmutableList<WatchKey> queuedKeys() {
	return ImmutableList.copyOf(queue);
	}

	@NullableDecl
	@Override
	public WatchKey poll() {
	checkOpen();
	return check(queue.poll());
	}

	@NullableDecl
	@Override
	public WatchKey poll(long timeout, TimeUnit unit) throws InterruptedException {
	checkOpen();
	return check(queue.poll(timeout, unit));
	}

	@Override
	public WatchKey take() throws InterruptedException {
	checkOpen();
	return check(queue.take());
	}

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

	/** Checks that the watch service is open, throwing {@link ClosedWatchServiceException} if not. */
	protected final void checkOpen() {
	if (!open.get()) {
	throw new ClosedWatchServiceException();
	}
	}

	@Override
	public void close() {
	if (open.compareAndSet(true, false)) {
	queue.clear();
	queue.offer(poison);
	}
	}

	/** A basic implementation of {@link WatchEvent}. */
	static final class Event<T> implements WatchEvent<T> {

	private final Kind<T> kind;
	private final int count;

	@NullableDecl private final T context;

	public Event(Kind<T> kind, int count, @NullableDecl T context) {
	this.kind = checkNotNull(kind);
	checkArgument(count >= 0, "count (%s) must be non-negative", count);
	this.count = count;
	this.context = context;
	}

	@Override
	public Kind<T> kind() {
	return kind;
	}

	@Override
	public int count() {
	return count;
	}

	@NullableDecl
	@Override
	public T context() {
	return context;
	}

	@Override
	public boolean equals(Object obj) {
	if (obj instanceof Event) {
	Event<?> other = (Event<?>) obj;
	return kind().equals(other.kind())
	&& count() == other.count()
	&& Objects.equals(context(), other.context());
	}
	return false;
	}

	@Override
	public int hashCode() {
	return Objects.hash(kind(), count(), context());
	}

	@Override
	public String toString() {
	return MoreObjects.toStringHelper(this)
	.add("kind", kind())
	.add("count", count())
	.add("context", context())
	.toString();
	}
	}

	/** Implementation of {@link WatchKey} for an {@link AbstractWatchService}. */
	static final class Key implements WatchKey {

	@VisibleForTesting static final int MAX_QUEUE_SIZE = 256;

	private static WatchEvent<Object> overflowEvent(int count) {
	return new Event<>(OVERFLOW, count, null);
	}

	private final AbstractWatchService watcher;
	private final Watchable watchable;
	private final ImmutableSet<WatchEvent.Kind<?>> subscribedTypes;

	private final AtomicReference<State> state = new AtomicReference<>(State.READY);
	private final AtomicBoolean valid = new AtomicBoolean(true);
	private final AtomicInteger overflow = new AtomicInteger();

	private final BlockingQueue<WatchEvent<?>> events = new ArrayBlockingQueue<>(MAX_QUEUE_SIZE);

	public Key(
	AbstractWatchService watcher,
	@NullableDecl Watchable watchable,
	Iterable<? extends WatchEvent.Kind<?>> subscribedTypes) {
	this.watcher = checkNotNull(watcher);
	this.watchable = watchable; // nullable for Watcher poison
	this.subscribedTypes = ImmutableSet.copyOf(subscribedTypes);
	}

	/** Gets the current state of this key, State.READY or SIGNALLED. */
	@VisibleForTesting
	State state() {
	return state.get();
	}

	/** Gets whether or not this key is subscribed to the given type of event. */
	public boolean subscribesTo(WatchEvent.Kind<?> eventType) {
	return subscribedTypes.contains(eventType);
	}

	/**
	* Posts the given event to this key. After posting one or more events, {@link #signal()} must
	* be called to cause the key to be enqueued with the watch service.
	*/
	public void post(WatchEvent<?> event) {
	if (!events.offer(event)) {
	overflow.incrementAndGet();
	}
	}

	/**
	* Sets the state to SIGNALLED and enqueues this key with the watcher if it was previously in
	* the READY state.
	*/
	public void signal() {
	if (state.getAndSet(State.SIGNALLED) == State.READY) {
	watcher.enqueue(this);
	}
	}

	@Override
	public boolean isValid() {
	return watcher.isOpen() && valid.get();
	}

	@Override
	public List<WatchEvent<?>> pollEvents() {
	// note: it's correct to be able to retrieve more events from a key without calling reset()
	// reset() is ONLY for "returning" the key to the watch service to potentially be retrieved by
	// another thread when you're finished with it
	List<WatchEvent<?>> result = new ArrayList<>(events.size());
	events.drainTo(result);
	int overflowCount = overflow.getAndSet(0);
	if (overflowCount != 0) {
	result.add(overflowEvent(overflowCount));
	}
	return Collections.unmodifiableList(result);
	}

	@Override
	public boolean reset() {
	// calling reset() multiple times without polling events would cause key to be placed in
	// watcher queue multiple times, but not much that can be done about that
	if (isValid() && state.compareAndSet(State.SIGNALLED, State.READY)) {
	// requeue if events are pending
	if (!events.isEmpty()) {
	signal();
	}
	}

	return isValid();
	}

	@Override
	public void cancel() {
	valid.set(false);
	watcher.cancelled(this);
	}

	@Override
	public Watchable watchable() {
	return watchable;
	}

	@VisibleForTesting
	enum State {
	READY,
	SIGNALLED
	}
	}
	}