content: /*
   * Copyright (C) 2016 Lightbend Inc. <http://www.lightbend.com>
   */
  
  package akka.stream.contrib;
  
  import akka.NotUsed;
  import akka.japi.Pair;
  import akka.stream.Attributes;
  import akka.stream.Outlet;
  import akka.stream.SourceShape;
  import akka.stream.javadsl.Source;
  import akka.stream.stage.AbstractOutHandler;
  import akka.stream.stage.GraphStage;
  import akka.stream.stage.GraphStageLogic;
  import akka.stream.stage.TimerGraphStageLogic;
  
  import com.sun.nio.file.SensitivityWatchEventModifier;
  
  import scala.Tuple2;
  import scala.concurrent.duration.FiniteDuration;
  
  import java.io.IOException;
  import java.nio.file.*;
  import java.util.ArrayDeque;
  import java.util.Queue;
  
  import static java.nio.file.StandardWatchEventKinds.*;
  
  /**
   * Watches a file system directory and streams change events from it.
   *
   * Note that the JDK watcher is notoriously slow on some platform (up to 1s after event actually happened on OSX for example)
   *
   * @deprecated since 0.10, use Alpakka's implementation instead https://developer.lightbend.com/docs/alpakka/current/file.html#listing-directory-contents
   */
  @Deprecated
  public final class DirectoryChanges extends GraphStage<SourceShape<Pair<Path, DirectoryChanges.Change>>> {
  
    public enum Change {
      Modification,
      Creation,
      Deletion
    }
  
    private final static Attributes DEFAULT_ATTRIBUTES = Attributes.name("DirectoryChanges");
  
    private final Path directoryPath;
    private final FiniteDuration pollInterval;
    private final int maxBufferSize;
    private final Outlet<Pair<Path, Change>> out = Outlet.create("DirectoryChanges.out");
    private final SourceShape<Pair<Path, Change>> shape = SourceShape.of(out);
  
    /**
     * @param directoryPath Directory to watch
     * @param pollInterval Interval between polls to the JDK watch service when a push comes in and there was no changes, if
     *                     the JDK implementation is slow, it will not help lowering this
     * @param maxBufferSize Maximum number of buffered directory changes before the stage fails
     */
    public DirectoryChanges(Path directoryPath, FiniteDuration pollInterval, int maxBufferSize) {
      this.directoryPath = directoryPath;
      this.pollInterval = pollInterval;
      this.maxBufferSize = maxBufferSize;
    }
  
    @Override
    public SourceShape<Pair<Path, Change>> shape() {
      return shape;
    }
  
    @Override
    public Attributes initialAttributes() {
      return DEFAULT_ATTRIBUTES;
    }
  
    @Override
    public GraphStageLogic createLogic(Attributes inheritedAttributes) throws IOException {
      if (!Files.exists(directoryPath)) throw new IllegalArgumentException("The path: '" + directoryPath + "' does not exist");
      if (!Files.isDirectory(directoryPath)) throw new IllegalArgumentException("The path '" + directoryPath + "' is not a directory");
  
      return new TimerGraphStageLogic(shape) {
        private final Queue<Pair<Path, Change>> buffer = new ArrayDeque<>();
        private final WatchService service = directoryPath.getFileSystem().newWatchService();
        private final WatchKey watchKey = directoryPath.register(
          service,
          new WatchEvent.Kind<?>[] { ENTRY_CREATE, ENTRY_MODIFY, ENTRY_DELETE, OVERFLOW },
          // this is com.sun internal, but the service is useless on OSX without it
          SensitivityWatchEventModifier.HIGH
        );
  
        {
          setHandler(out, new AbstractOutHandler(){
  
            @Override
            public void onPull() throws Exception {
              if (!buffer.isEmpty()) {
                pushHead();
              } else {
                doPoll();
                if (!buffer.isEmpty()) {
                  pushHead();
                } else {
                  schedulePoll();
                }
              }
            }
          });
        }
  
        @Override
        public void onTimer(Object timerKey) {
          if (!isClosed(out)) {
            doPoll();
            if (!buffer.isEmpty()) {
              pushHead();
            } else {
              schedulePoll();
            }
          }
        }
  
        @Override
        public void postStop() {
          try {
            if (watchKey.isValid()) watchKey.cancel();
            service.close();
          } catch (Exception ex) {
            // Remove when #21168 is in a release
            throw new RuntimeException(ex);
          }
        }
  
        private void pushHead() {
          final Pair<Path, Change> head = buffer.poll();
          if (head != null) {
            push(out, head);
          }
        }
  
        private void schedulePoll() {
          scheduleOnce("poll", pollInterval);
        }
  
        private void doPoll() {
          try {
            for (WatchEvent<?> event: watchKey.pollEvents()) {
              final WatchEvent.Kind<?> kind = event.kind();
  
              if (OVERFLOW.equals(kind)) {
                // overflow means that some file system change events may have been missed,
                // that may be ok for some scenarios but to make sure it does not pass unnoticed we fail the stage
                failStage(new RuntimeException("Overflow from watch service: '" + directoryPath + "'"));
  
              } else {
                // if it's not an overflow it must be a Path event
                @SuppressWarnings("unchecked")
                final Path path = (Path) event.context();
                final Path absolutePath = directoryPath.resolve(path);
                final Change change = kindToChange(kind);
  
                buffer.add(new Pair<>(absolutePath, change));
                if (buffer.size() > maxBufferSize) {
                  failStage(new RuntimeException("Max event buffer size " +
                    maxBufferSize + " reached for $path"));
                }
              }
  
            }
          } finally {
            if (!watchKey.reset()) {
              // directory no longer accessible
              completeStage();
            }
          }
        }
  
  
  
        // convert from the parametrized API to our much nicer API enum
        private Change kindToChange(WatchEvent.Kind<?> kind) {
          final Change change;
          if (kind.equals(ENTRY_CREATE)) {
            change = Change.Creation;
          } else if (kind.equals(ENTRY_DELETE)) {
            change = Change.Deletion;
          } else if (kind.equals(ENTRY_MODIFY)) {
            change = Change.Modification;
          } else {
            throw new RuntimeException("Unexpected kind of event gotten from watch service for path '" +
              directoryPath + "': " + kind);
          }
          return change;
        }
  
  
      };
    }
  
    @Override
    public String toString() {
      return "DirectoryChanges(" + directoryPath + ')';
    }
  
  
    // factory methods
  
    /**
     * Java API
     *
     * @param directoryPath Directory to watch
     * @param pollInterval Interval between polls to the JDK watch service when a push comes in and there was no changes, if
     *                     the JDK implementation is slow, it will not help lowering this
     * @param maxBufferSize Maximum number of buffered directory changes before the stage fails
     */
    public static Source<Pair<Path, Change>, NotUsed> create(Path directoryPath, FiniteDuration pollInterval, int maxBufferSize) {
      return Source.fromGraph(new DirectoryChanges(directoryPath, pollInterval, maxBufferSize));
    }
  
    /**
     * Scala API
     *
     * @param directoryPath Directory to watch
     * @param pollInterval Interval between polls to the JDK watch service when a push comes in and there was no changes, if
     *                     the JDK implementation is slow, it will not help lowering this
     * @param maxBufferSize Maximum number of buffered directory changes before the stage fails
     */
    public static akka.stream.scaladsl.Source<Tuple2<Path, Change>, NotUsed> apply(Path directoryPath, FiniteDuration pollInterval, int maxBufferSize) {
      return create(directoryPath, pollInterval, maxBufferSize)
        .map((Pair<Path, Change> pair) -> Tuple2.apply(pair.first(), pair.second()))
        .asScala();
    }
  
  }
  