// Path: src/main/java/mirror/Utils.java
// public static MirrorStub withTimeout(MirrorStub s) {
//   // over VPN, ~100k files can take 30 seconds.
//   return s.withDeadlineAfter(10, TimeUnit.MINUTES);
// }
// 
// Path: src/main/java/mirror/tasks/TaskFactory.java
// public interface TaskFactory {
// 
//   default TaskHandle runTask(TaskLogic logic) {
//     return runTask(logic, null);
//   }
// 
//   TaskHandle runTask(TaskLogic logic, Runnable onFailure);
// 
//   void stopTask(TaskLogic logic);
// 
//   default TaskPool newTaskPool() {
//     return new TaskPool(this);
//   }
// 
// }
// 
// Path: src/main/java/mirror/tasks/TaskLogic.java
// public interface TaskLogic {
// 
//   /** Run one interaction, and return how long the task wants to sleep. */
//   Duration runOneLoop() throws InterruptedException;
// 
//   /** Called on the task thread, before we start calling {@link #runOneLoop()} in a loop. */
//   default void onStart() throws InterruptedException {
//   }
// 
//   default void onFailure() throws InterruptedException {
//   }
// 
//   /** Called on the task thread, after we've interrupted/stopped calling {@link #runOneLoop()}. */
//   default void onStop() throws InterruptedException {
//   }
// 
//   /**
//    * Called off the task thread, when we're trying to interrupt the task.
//    *
//    * Most threads shouldn't need this because they'll shut down
//    * when we call thread.interrupt().
//    */
//   default void onInterrupt() {
//   }
// 
//   default String getName() {
//     String name = getClass().getSimpleName();
//     // lambdas don't have simple names
//     if (name.equals("")) {
//       name = StringUtils.substringAfterLast(getClass().getName(), ".");
//     }
//     return name;
//   }
// }


import static mirror.Utils.withTimeout;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.time.Duration;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicReference;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.common.util.concurrent.SettableFuture;
import io.grpc.ManagedChannel;
import io.grpc.stub.ClientCallStreamObserver;
import io.grpc.stub.ClientResponseObserver;
import io.grpc.stub.StreamObserver;
import mirror.MirrorGrpc.MirrorStub;
import mirror.tasks.TaskFactory;
import mirror.tasks.TaskLogic;
package mirror;






public class MirrorClient {

  private static final Logger log = LoggerFactory.getLogger(MirrorClient.class);

  private final MirrorPaths paths;
  private final TaskFactory taskFactory;
  private final ConnectionDetector detector;
  private final FileWatcherFactory watcherFactory;
  private final FileAccess fileAccess;
  private final ChannelFactory channelFactory;
  private volatile TaskLogic sessionStarter;
  private volatile MirrorSession session;

  public MirrorClient(
    MirrorPaths paths,
    TaskFactory taskFactory,
    ConnectionDetector detector,
    FileWatcherFactory watcherFactory,
    FileAccess fileAccess,
    ChannelFactory channelFactory) {
    this.paths = paths;
    this.taskFactory = taskFactory;
    this.detector = detector;
    this.watcherFactory = watcherFactory;
    this.fileAccess = fileAccess;
    this.channelFactory = channelFactory;
  }

  /** Connects to the server and starts a sync session. */
  public void startSession() throws InterruptedException {
    CountDownLatch started = new CountDownLatch(1);
    sessionStarter = new SessionStarter(channelFactory, started);
    taskFactory.runTask(sessionStarter);
    started.await();
  }

  private void startSession(ChannelFactory channelFactory, CountDownLatch onFailure) {
    detector.blockUntilConnected();
    log.info("Connected, starting session, version " + Mirror.getVersion());

    ManagedChannel channel = channelFactory.newChannel();
    MirrorStub stub = MirrorGrpc.newStub(channel).withCompression("gzip");

    // 0. Do a time drift check
    boolean outOfSync = logErrorIfTimeOutOfSync(stub);
    if (outOfSync) {
      return;
    }

    session = new MirrorSession(taskFactory, paths, fileAccess, watcherFactory);
    session.addStoppedCallback(channel::shutdownNow);
    // Automatically re-connect when we're disconnected
    session.addStoppedCallback(() -> {
      // Don't call startSession again directly, because then we'll start running
      // our connection code on whatever thread is running this callback. Instead
      // just signal our main client thread that it should try again.
      onFailure.countDown();
    });

    // 1. see what our current state is
    try {
      List<Update> localState = session.calcInitialState();
      log.info("Client has " + localState.size() + " paths");

      // 2. send it to the server, so they can send back any stale/missing paths we have
      SettableFuture<InitialSyncResponse> responseFuture = SettableFuture.create();

      // Ideally this would be a blocking/sync call, but it looks like because
      // one of our RPC methods is streaming, then this one is as well
      InitialSyncRequest.Builder req = InitialSyncRequest
        .newBuilder()
        .setRemotePath(paths.remoteRoot.toString())
        .setClientId(getClientId())
        .setVersion(Mirror.getVersion())
        .addAllState(localState);
      paths.addParameters(req);
// Path: src/main/java/mirror/Utils.java
// public static MirrorStub withTimeout(MirrorStub s) {
//   // over VPN, ~100k files can take 30 seconds.
//   return s.withDeadlineAfter(10, TimeUnit.MINUTES);
// }
// 
// Path: src/main/java/mirror/tasks/TaskFactory.java
// public interface TaskFactory {
// 
//   default TaskHandle runTask(TaskLogic logic) {
//     return runTask(logic, null);
//   }
// 
//   TaskHandle runTask(TaskLogic logic, Runnable onFailure);
// 
//   void stopTask(TaskLogic logic);
// 
//   default TaskPool newTaskPool() {
//     return new TaskPool(this);
//   }
// 
// }
// 
// Path: src/main/java/mirror/tasks/TaskLogic.java
// public interface TaskLogic {
// 
//   /** Run one interaction, and return how long the task wants to sleep. */
//   Duration runOneLoop() throws InterruptedException;
// 
//   /** Called on the task thread, before we start calling {@link #runOneLoop()} in a loop. */
//   default void onStart() throws InterruptedException {
//   }
// 
//   default void onFailure() throws InterruptedException {
//   }
// 
//   /** Called on the task thread, after we've interrupted/stopped calling {@link #runOneLoop()}. */
//   default void onStop() throws InterruptedException {
//   }
// 
//   /**
//    * Called off the task thread, when we're trying to interrupt the task.
//    *
//    * Most threads shouldn't need this because they'll shut down
//    * when we call thread.interrupt().
//    */
//   default void onInterrupt() {
//   }
// 
//   default String getName() {
//     String name = getClass().getSimpleName();
//     // lambdas don't have simple names
//     if (name.equals("")) {
//       name = StringUtils.substringAfterLast(getClass().getName(), ".");
//     }
//     return name;
//   }
// }


// Path: src/main/java/mirror/MirrorClient.java
import static mirror.Utils.withTimeout;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.time.Duration;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicReference;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.common.util.concurrent.SettableFuture;
import io.grpc.ManagedChannel;
import io.grpc.stub.ClientCallStreamObserver;
import io.grpc.stub.ClientResponseObserver;
import io.grpc.stub.StreamObserver;
import mirror.MirrorGrpc.MirrorStub;
import mirror.tasks.TaskFactory;
import mirror.tasks.TaskLogic;

package mirror;






public class MirrorClient {

  private static final Logger log = LoggerFactory.getLogger(MirrorClient.class);

  private final MirrorPaths paths;
  private final TaskFactory taskFactory;
  private final ConnectionDetector detector;
  private final FileWatcherFactory watcherFactory;
  private final FileAccess fileAccess;
  private final ChannelFactory channelFactory;
  private volatile TaskLogic sessionStarter;
  private volatile MirrorSession session;

  public MirrorClient(
    MirrorPaths paths,
    TaskFactory taskFactory,
    ConnectionDetector detector,
    FileWatcherFactory watcherFactory,
    FileAccess fileAccess,
    ChannelFactory channelFactory) {
    this.paths = paths;
    this.taskFactory = taskFactory;
    this.detector = detector;
    this.watcherFactory = watcherFactory;
    this.fileAccess = fileAccess;
    this.channelFactory = channelFactory;
  }

  /** Connects to the server and starts a sync session. */
  public void startSession() throws InterruptedException {
    CountDownLatch started = new CountDownLatch(1);
    sessionStarter = new SessionStarter(channelFactory, started);
    taskFactory.runTask(sessionStarter);
    started.await();
  }

  private void startSession(ChannelFactory channelFactory, CountDownLatch onFailure) {
    detector.blockUntilConnected();
    log.info("Connected, starting session, version " + Mirror.getVersion());

    ManagedChannel channel = channelFactory.newChannel();
    MirrorStub stub = MirrorGrpc.newStub(channel).withCompression("gzip");

    // 0. Do a time drift check
    boolean outOfSync = logErrorIfTimeOutOfSync(stub);
    if (outOfSync) {
      return;
    }

    session = new MirrorSession(taskFactory, paths, fileAccess, watcherFactory);
    session.addStoppedCallback(channel::shutdownNow);
    // Automatically re-connect when we're disconnected
    session.addStoppedCallback(() -> {
      // Don't call startSession again directly, because then we'll start running
      // our connection code on whatever thread is running this callback. Instead
      // just signal our main client thread that it should try again.
      onFailure.countDown();
    });

    // 1. see what our current state is
    try {
      List<Update> localState = session.calcInitialState();
      log.info("Client has " + localState.size() + " paths");

      // 2. send it to the server, so they can send back any stale/missing paths we have
      SettableFuture<InitialSyncResponse> responseFuture = SettableFuture.create();

      // Ideally this would be a blocking/sync call, but it looks like because
      // one of our RPC methods is streaming, then this one is as well
      InitialSyncRequest.Builder req = InitialSyncRequest
        .newBuilder()
        .setRemotePath(paths.remoteRoot.toString())
        .setClientId(getClientId())
        .setVersion(Mirror.getVersion())
        .addAllState(localState);
      paths.addParameters(req);
      withTimeout(stub).initialSync(req.build(), new StreamObserver<InitialSyncResponse>() {