// Path: src/main/java/mirror/Utils.java
// public class Utils {
// 
//   private static final Logger log = LoggerFactory.getLogger(Utils.class);
// 
//   @FunctionalInterface
//   public interface InterruptedRunnable {
//     void run() throws InterruptedException;
//   }
// 
//   /** grpc-java doesn't support timeouts yet, so we have to set a per-call deadline. */
//   public static MirrorStub withTimeout(MirrorStub s) {
//     // over VPN, ~100k files can take 30 seconds.
//     return s.withDeadlineAfter(10, TimeUnit.MINUTES);
//   }
// 
//   public static void logConnectionError(Logger log, Throwable t) {
//     if (t instanceof StatusRuntimeException) {
//       log.info("Connection status: " + niceToString(((StatusRuntimeException) t).getStatus()));
//     } else if (t instanceof StatusException) {
//       log.info("Connection status: " + niceToString(((StatusException) t).getStatus()));
//     } else {
//       log.error("Error from stream: " + t.getMessage(), t);
//     }
//   }
// 
//   private static String niceToString(Status status) {
//     // the default Status.toString has a stack trace in it, which is ugly
//     return MoreObjects
//       .toStringHelper(status)
//       .add("code", status.getCode().name())
//       .add("description", status.getDescription())
//       .add("cause", status.getCause() != null ? status.getCause().getMessage() : null)
//       .toString();
//   }
// 
//   public static void resetIfInterrupted(InterruptedRunnable r) {
//     try {
//       r.run();
//     } catch (InterruptedException e) {
//       Thread.currentThread().interrupt();
//       throw new RuntimeException(e);
//     }
//   }
// 
//   public static String debugString(Update u) {
//     return "[" + TextFormat.shortDebugString(u).replace(": ", ":") + "]";
//   }
// 
//   public static void time(Logger log, String action, Runnable r) {
//     log.info("Starting " + action);
//     long start = System.currentTimeMillis();
//     r.run();
//     long stop = System.currentTimeMillis();
//     log.info("Completed " + action + ": " + (stop - start) + "ms");
//   }
// 
//   /** Waits until {@code path} is not longer being actively written to (best-effort). */
//   public static void ensureSettled(FileAccess fileAccess, Path path) throws InterruptedException {
//     // do some gyrations to ensure the file writer has completely written the file
//     boolean shouldBeComplete = false;
//     try {
//       while (!shouldBeComplete) {
//         long localModTime = fileAccess.getModifiedTime(path);
//         if (fileWasJustModified(localModTime)) {
//           long size1 = fileAccess.getFileSize(path);
//           Thread.sleep(500); // 100ms was too small
//           long size2 = fileAccess.getFileSize(path);
//           shouldBeComplete = size1 == size2;
//           if (!shouldBeComplete) {
//             log.debug("{} not settled {} {}", path, size1, size2);
//           }
//         } else {
//           shouldBeComplete = true; // no need to check
//         }
//       }
//     } catch (IOException io) {
//       // assume the file disappeared, and we'll catch it later
//       return;
//     }
//   }
// 
//   /** @return whether localModTime was within the last 2 seconds. */
//   private static boolean fileWasJustModified(long localModTime) {
//     return (System.currentTimeMillis() - localModTime) <= 2000;
//   }
// 
//   public static String abbreviatePath(String path) {
//     if (StringUtils.countMatches(path, '/') < 2) {
//       return path;
//     } else {
//       return StringUtils.substringBefore(path, "/") + "/.../" + StringUtils.substringAfterLast(path, "/");
//     }
//   }
// 
// }


import java.time.Duration;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import mirror.Utils;
package mirror.tasks;





/**
 * Provides some basic "start up a thread and poll a queue" abstractions.
 *
 * Kind of actor-like. Ish.
 */
class ThreadBasedTask {

  private static int nextThread = 0;
  protected final Logger log = LoggerFactory.getLogger(getClass());
  private final AtomicBoolean shutdown = new AtomicBoolean(false);
  private final CountDownLatch isStarted = new CountDownLatch(1);
  private final CountDownLatch isShutdown = new CountDownLatch(1);
  private final Thread thread;
  private final TaskLogic task;
  private final Runnable onFailure;
  private final Runnable onStop;

  private static synchronized int nextThreadId() {
    return nextThread++;
  }

  protected ThreadBasedTask(TaskLogic task, Runnable onFailure, Runnable onStop) {
    this.task = task;
    this.onFailure = onFailure;
    this.onStop = onStop;
    thread = new ThreadFactoryBuilder().setDaemon(true).setNameFormat(nextThreadId() + "-" + task.getName() + "-%s").build().newThread(() -> run());
  }

  void start() {
    thread.start();
    Utils.resetIfInterrupted(() -> isStarted.await());
  }

  void stop() {
    if (shutdown.compareAndSet(false, true)) {
      Utils.resetIfInterrupted(() -> isStarted.await());
      thread.interrupt();
      task.onInterrupt();
// Path: src/main/java/mirror/Utils.java
// public class Utils {
// 
//   private static final Logger log = LoggerFactory.getLogger(Utils.class);
// 
//   @FunctionalInterface
//   public interface InterruptedRunnable {
//     void run() throws InterruptedException;
//   }
// 
//   /** grpc-java doesn't support timeouts yet, so we have to set a per-call deadline. */
//   public static MirrorStub withTimeout(MirrorStub s) {
//     // over VPN, ~100k files can take 30 seconds.
//     return s.withDeadlineAfter(10, TimeUnit.MINUTES);
//   }
// 
//   public static void logConnectionError(Logger log, Throwable t) {
//     if (t instanceof StatusRuntimeException) {
//       log.info("Connection status: " + niceToString(((StatusRuntimeException) t).getStatus()));
//     } else if (t instanceof StatusException) {
//       log.info("Connection status: " + niceToString(((StatusException) t).getStatus()));
//     } else {
//       log.error("Error from stream: " + t.getMessage(), t);
//     }
//   }
// 
//   private static String niceToString(Status status) {
//     // the default Status.toString has a stack trace in it, which is ugly
//     return MoreObjects
//       .toStringHelper(status)
//       .add("code", status.getCode().name())
//       .add("description", status.getDescription())
//       .add("cause", status.getCause() != null ? status.getCause().getMessage() : null)
//       .toString();
//   }
// 
//   public static void resetIfInterrupted(InterruptedRunnable r) {
//     try {
//       r.run();
//     } catch (InterruptedException e) {
//       Thread.currentThread().interrupt();
//       throw new RuntimeException(e);
//     }
//   }
// 
//   public static String debugString(Update u) {
//     return "[" + TextFormat.shortDebugString(u).replace(": ", ":") + "]";
//   }
// 
//   public static void time(Logger log, String action, Runnable r) {
//     log.info("Starting " + action);
//     long start = System.currentTimeMillis();
//     r.run();
//     long stop = System.currentTimeMillis();
//     log.info("Completed " + action + ": " + (stop - start) + "ms");
//   }
// 
//   /** Waits until {@code path} is not longer being actively written to (best-effort). */
//   public static void ensureSettled(FileAccess fileAccess, Path path) throws InterruptedException {
//     // do some gyrations to ensure the file writer has completely written the file
//     boolean shouldBeComplete = false;
//     try {
//       while (!shouldBeComplete) {
//         long localModTime = fileAccess.getModifiedTime(path);
//         if (fileWasJustModified(localModTime)) {
//           long size1 = fileAccess.getFileSize(path);
//           Thread.sleep(500); // 100ms was too small
//           long size2 = fileAccess.getFileSize(path);
//           shouldBeComplete = size1 == size2;
//           if (!shouldBeComplete) {
//             log.debug("{} not settled {} {}", path, size1, size2);
//           }
//         } else {
//           shouldBeComplete = true; // no need to check
//         }
//       }
//     } catch (IOException io) {
//       // assume the file disappeared, and we'll catch it later
//       return;
//     }
//   }
// 
//   /** @return whether localModTime was within the last 2 seconds. */
//   private static boolean fileWasJustModified(long localModTime) {
//     return (System.currentTimeMillis() - localModTime) <= 2000;
//   }
// 
//   public static String abbreviatePath(String path) {
//     if (StringUtils.countMatches(path, '/') < 2) {
//       return path;
//     } else {
//       return StringUtils.substringBefore(path, "/") + "/.../" + StringUtils.substringAfterLast(path, "/");
//     }
//   }
// 
// }


// Path: src/main/java/mirror/tasks/ThreadBasedTask.java
import java.time.Duration;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import mirror.Utils;

package mirror.tasks;





/**
 * Provides some basic "start up a thread and poll a queue" abstractions.
 *
 * Kind of actor-like. Ish.
 */
class ThreadBasedTask {

  private static int nextThread = 0;
  protected final Logger log = LoggerFactory.getLogger(getClass());
  private final AtomicBoolean shutdown = new AtomicBoolean(false);
  private final CountDownLatch isStarted = new CountDownLatch(1);
  private final CountDownLatch isShutdown = new CountDownLatch(1);
  private final Thread thread;
  private final TaskLogic task;
  private final Runnable onFailure;
  private final Runnable onStop;

  private static synchronized int nextThreadId() {
    return nextThread++;
  }

  protected ThreadBasedTask(TaskLogic task, Runnable onFailure, Runnable onStop) {
    this.task = task;
    this.onFailure = onFailure;
    this.onStop = onStop;
    thread = new ThreadFactoryBuilder().setDaemon(true).setNameFormat(nextThreadId() + "-" + task.getName() + "-%s").build().newThread(() -> run());
  }

  void start() {
    thread.start();
    Utils.resetIfInterrupted(() -> isStarted.await());
  }

  void stop() {
    if (shutdown.compareAndSet(false, true)) {
      Utils.resetIfInterrupted(() -> isStarted.await());
      thread.interrupt();
      task.onInterrupt();
      Utils.resetIfInterrupted(() -> isShutdown.await());