<commit_before>package mirror;

import static org.apache.commons.lang3.StringUtils.defaultIfEmpty;
import static org.apache.commons.lang3.StringUtils.substringAfterLast;
import static org.jooq.lambda.Seq.seq;
import static org.jooq.lambda.tuple.Tuple.tuple;

import java.io.IOException;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import org.jooq.lambda.tuple.Tuple2;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.annotations.VisibleForTesting;

import mirror.UpdateTreeDiff.DiffResults;
import mirror.tasks.TaskLogic;

/**
 * Implements the steady-state (post-initial sync) two-way sync logic.
 *
 * We poll for changes, either the remote host or our local disk, and
 * either persist it locally or send it out remotely, while also considering
 * whether we've since had a newer/conflicting change.
 */
public class SyncLogic implements TaskLogic {

  private static final Logger log = LoggerFactory.getLogger(SyncLogic.class);
  private final Queues queues;
  private final FileAccess fileAccess;
  private final UpdateTree tree;

  public SyncLogic(Queues queues, FileAccess fileAccess, UpdateTree tree) {
    this.queues = queues;
    this.fileAccess = fileAccess;
    this.tree = tree;
  }

  @Override
  public void onStart() throws InterruptedException {
    diff(); // do an initial diff
  }

  @Override
  public Duration runOneLoop() throws InterruptedException {
    try {
      List<Update> batch = getNextBatchOrBlock();
      logLocalUpdates(batch);
      for (Update u : batch) {
        handleUpdate(u);
      }
      diff();
    } catch (IOException | RuntimeException e) {
      log.error("Exception", e);
    }
    return null;
  }

  // see if we have up to N more updates
  private List<Update> getNextBatchOrBlock() throws InterruptedException {
    List<Update> updates = new ArrayList<>();
    // block for at least one
    updates.add(queues.incomingQueue.take());
    // now go ahead and drain the rest while we're here
    queues.incomingQueue.drainTo(updates);
    return updates;
  }

  @VisibleForTesting
  void poll() throws IOException, InterruptedException {
    Update u = queues.incomingQueue.poll();
    if (u != null) {
      handleUpdate(u);
      diff();
    }
  }

  private void handleUpdate(Update u) throws IOException, InterruptedException {
    if (u.getLocal()) {
      if (isStaleLocalUpdate(u)) {
        return;
      }
      tree.addLocal(ensureSettledAndReadModTime(u));
    } else {
      tree.addRemote(u);
    }
  }

  @VisibleForTesting
  void diff() throws InterruptedException {
    DiffResults r = new UpdateTreeDiff(tree).diff();
    for (Update u : r.saveLocally) {
      queues.saveToLocal.put(u);
    }
    for (Update u : r.sendToRemote) {
      queues.saveToRemote.put(u);
    }
  }
<commit_msg>Change SyncLogic.poll to loop as it makes tests less verbose.<commit_after>package mirror;

import static org.apache.commons.lang3.StringUtils.defaultIfEmpty;
import static org.apache.commons.lang3.StringUtils.substringAfterLast;
import static org.jooq.lambda.Seq.seq;
import static org.jooq.lambda.tuple.Tuple.tuple;

import java.io.IOException;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import org.jooq.lambda.tuple.Tuple2;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.annotations.VisibleForTesting;

import mirror.UpdateTreeDiff.DiffResults;
import mirror.tasks.TaskLogic;

/**
 * Implements the steady-state (post-initial sync) two-way sync logic.
 *
 * We poll for changes, either the remote host or our local disk, and
 * either persist it locally or send it out remotely, while also considering
 * whether we've since had a newer/conflicting change.
 */
public class SyncLogic implements TaskLogic {

  private static final Logger log = LoggerFactory.getLogger(SyncLogic.class);
  private final Queues queues;
  private final FileAccess fileAccess;
  private final UpdateTree tree;

  public SyncLogic(Queues queues, FileAccess fileAccess, UpdateTree tree) {
    this.queues = queues;
    this.fileAccess = fileAccess;
    this.tree = tree;
  }

  @Override
  public void onStart() throws InterruptedException {
    diff(); // do an initial diff
  }

  @Override
  public Duration runOneLoop() throws InterruptedException {
    try {
      List<Update> batch = getNextBatchOrBlock();
      logLocalUpdates(batch);
      for (Update u : batch) {
        handleUpdate(u);
      }
      diff();
    } catch (IOException | RuntimeException e) {
      log.error("Exception", e);
    }
    return null;
  }

  // see if we have up to N more updates
  private List<Update> getNextBatchOrBlock() throws InterruptedException {
    List<Update> updates = new ArrayList<>();
    // block for at least one
    updates.add(queues.incomingQueue.take());
    // now go ahead and drain the rest while we're here
    queues.incomingQueue.drainTo(updates);
    return updates;
  }

  @VisibleForTesting
  void poll() throws IOException, InterruptedException {
    Update u;
    while ((u = queues.incomingQueue.poll()) != null) {
      handleUpdate(u);
    }
    diff();
  }

  private void handleUpdate(Update u) throws IOException, InterruptedException {
    if (u.getLocal()) {
      if (isStaleLocalUpdate(u)) {
        return;
      }
      tree.addLocal(ensureSettledAndReadModTime(u));
    } else {
      tree.addRemote(u);
    }
  }

  @VisibleForTesting
  void diff() throws InterruptedException {
    DiffResults r = new UpdateTreeDiff(tree).diff();
    for (Update u : r.saveLocally) {
      queues.saveToLocal.put(u);
    }
    for (Update u : r.sendToRemote) {
      queues.saveToRemote.put(u);
    }
  }

{'fraction_non_alphanumeric': 0.10056246804158854, 'fraction_numerical': 0.0017044486108743822, 'mean_word_length': 3.479389312977099, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '348082', 'n_tokens_mistral': 1791, 'n_tokens_neox': 1765, 'n_words': 483}