public class C5CommonTestUtil {
  protected static final Logger LOG = LoggerFactory.getLogger(C5CommonTestUtil.class);

  public C5CommonTestUtil() {
  }

  /**
   * System property key to get base test directory value
   */
  public static final String BASE_TEST_DIRECTORY_KEY =
      "test.build.data.basedirectory";

  /**
   * Default base directory for test output.
   */
  public static final String DEFAULT_BASE_TEST_DIRECTORY = "target/test-data";

  /**
   * Directory where we put the data for this instance of C5CommonTestUtil
   */
  private File dataTestDir = null;

  /**
   * @return Where to write test data on local filesystem, specific to
   * the test. Useful for tests that do not use a cluster.
   * Creates it if it does not exist already.
   */
  public Path getDataTestDir() {
    if (this.dataTestDir == null) {
      setupDataTestDir();
    }
    return Paths.get(this.dataTestDir.getAbsolutePath());
  }

  /**
   * @param subdirName Test subdir name
   * @return Path to a subdirectory named <code>subdirName</code> under
   * {@link #getDataTestDir()}.
   * Does *NOT* create it if it does not exist.
   */
  public Path getDataTestDir(final String subdirName) {
    return Paths.get(getDataTestDir().toString(), subdirName);
  }

  /**
   * Sets up a directory for a test to use.
   *
   * @return New directory path, if created.
   */
  protected Path setupDataTestDir() {
    if (this.dataTestDir != null) {
      LOG.warn("Data test dir already setup in " +
          dataTestDir.getAbsolutePath());
      return null;
    }

    String randomStr = UUID.randomUUID().toString();
    Path testPath = Paths.get(getBaseTestDir().toString(), randomStr);

    this.dataTestDir = new File(testPath.toString()).getAbsoluteFile();
    // Set this property so if mapreduce jobs run, they will use this as their home dir.
    System.setProperty("test.build.dir", this.dataTestDir.toString());
    if (deleteOnExit()) {
      this.dataTestDir.deleteOnExit();
    }

    createSubDir("c5.local.dir", testPath, "c5-local-dir");

    return testPath;
  }

  protected void createSubDir(String propertyName, Path parent, String subDirName) {
    Path newPath = Paths.get(parent.toString(), subDirName);
    File newDir = new File(newPath.toString()).getAbsoluteFile();
    if (deleteOnExit()) {
      newDir.deleteOnExit();
    }
  }

  /**
   * @return True if we should delete testing dirs on exit.
   */
  boolean deleteOnExit() {
    String v = System.getProperty("c5.testing.preserve.testdir");
    // Let default be true, to delete on exit.
    return v == null || !Boolean.parseBoolean(v);
  }

  /**
   * @return True if we removed the test dirs
   * @throws java.io.IOException
   */
  public boolean cleanupTestDir() throws IOException {
    if (deleteDir(this.dataTestDir)) {
      this.dataTestDir = null;
      return true;
    }
    return false;
  }

  /**
   * @param subdir Test subdir name.
   * @return True if we removed the test dir
   * @throws java.io.IOException
   */
  boolean cleanupTestDir(final String subdir) throws IOException {
    if (this.dataTestDir == null) {
      return false;
    }
    return deleteDir(new File(this.dataTestDir, subdir));
  }

  /**
   * @return Where to write test data on local filesystem; usually
   * {@link #DEFAULT_BASE_TEST_DIRECTORY}
   * Should not be used by the unit tests, hence it's private.
   * Unit test will use a subdirectory of this directory.
   * @see #setupDataTestDir()
   */
  private Path getBaseTestDir() {
    String PathName = System.getProperty(
        BASE_TEST_DIRECTORY_KEY, DEFAULT_BASE_TEST_DIRECTORY);

    return Paths.get(PathName);
  }

  /**
   * @param dir Directory to delete
   * @return True if we deleted it.
   * @throws java.io.IOException
   */
  boolean deleteDir(final File dir) throws IOException {
    if (dir == null || !dir.exists()) {
      return true;
    }
    try {
      if (deleteOnExit()) {
        Files.delete(dir.toPath());
      }
      return true;
    } catch (IOException ex) {
      LOG.warn("Failed to delete " + dir.getAbsolutePath());
      return false;
    }
  }
} public final class QuorumConfiguration {

  public final boolean isTransitional;

  private final Set<Long> allPeers;
  private final Set<Long> prevPeers;
  private final Set<Long> nextPeers;

  public static final QuorumConfiguration EMPTY = new QuorumConfiguration(new HashSet<>());

  public static QuorumConfiguration of(Collection<Long> peerCollection) {
    return new QuorumConfiguration(peerCollection);
  }

  public static QuorumConfiguration fromProtostuff(c5db.replication.generated.QuorumConfigurationMessage message) {
    if (message.getTransitional()) {
      return new QuorumConfiguration(message.getPrevPeersList(), message.getNextPeersList());
    } else {
      return new QuorumConfiguration(message.getAllPeersList());
    }
  }

  public QuorumConfiguration getTransitionalConfiguration(Collection<Long> newPeerCollection) {
    if (isTransitional) {
      return new QuorumConfiguration(prevPeers, newPeerCollection);
    } else {
      return new QuorumConfiguration(allPeers, newPeerCollection);
    }
  }

  public QuorumConfiguration getCompletedConfiguration() {
    assert isTransitional;

    return new QuorumConfiguration(nextPeers);
  }

  public QuorumConfigurationMessage toProtostuff() {
    return new QuorumConfigurationMessage(
        isTransitional,
        Lists.newArrayList(allPeers),
        Lists.newArrayList(prevPeers),
        Lists.newArrayList(nextPeers));
  }

  public Set<Long> allPeers() {
    return allPeers;
  }

  public Set<Long> prevPeers() {
    return prevPeers;
  }

  public Set<Long> nextPeers() {
    return nextPeers;
  }

  public boolean isEmpty() {
    return allPeers.size() == 0
        && prevPeers.size() == 0
        && nextPeers.size() == 0;
  }

  /**
   * Determine if the peers in sourceSet include a majority of the peers in this configuration.
   */
  public boolean setContainsMajority(Set<Long> sourceSet) {
    if (isTransitional) {
      return setComprisesMajorityOfAnotherSet(sourceSet, prevPeers)
          && setComprisesMajorityOfAnotherSet(sourceSet, nextPeers);
    } else {
      return setComprisesMajorityOfAnotherSet(sourceSet, allPeers);
    }
  }

  /**
   * Given a map which tells the last acknowledged entry index for different peers, find the maximum
   * index value which is less than or equal to a majority of this configuration's peers' indexes.
   */
  public long calculateCommittedIndex(Map<Long, Long> peersLastAckedIndex) {
    if (isTransitional) {
      return Math.min(
          getGreatestIndexCommittedByMajority(prevPeers, peersLastAckedIndex),
          getGreatestIndexCommittedByMajority(nextPeers, peersLastAckedIndex));
    } else {
      return getGreatestIndexCommittedByMajority(allPeers, peersLastAckedIndex);
    }
  }


  private QuorumConfiguration(Collection<Long> peers) {
    this.isTransitional = false;
    allPeers = ImmutableSet.copyOf(peers);
    prevPeers = nextPeers = ImmutableSet.of();
  }

  private QuorumConfiguration(Collection<Long> prevPeers, Collection<Long> nextPeers) {
    this.isTransitional = true;
    this.prevPeers = ImmutableSet.copyOf(prevPeers);
    this.nextPeers = ImmutableSet.copyOf(nextPeers);
    this.allPeers = Sets.union(this.prevPeers, this.nextPeers).immutableCopy();
  }


  private static long getGreatestIndexCommittedByMajority(Set<Long> peers, Map<Long, Long> peersLastAckedIndex) {
    SortedMultiset<Long> committedIndexes = TreeMultiset.create();
    committedIndexes.addAll(peers.stream().map(peerId
        -> peersLastAckedIndex.getOrDefault(peerId, 0L)).collect(Collectors.toList()));
    return Iterables.get(committedIndexes.descendingMultiset(), calculateNumericalMajority(peers.size()) - 1);
  }

  private static <T> boolean setComprisesMajorityOfAnotherSet(Set<T> sourceSet, Set<T> destinationSet) {
    return Sets.intersection(sourceSet, destinationSet).size() >= calculateNumericalMajority(destinationSet.size());
  }

  private static int calculateNumericalMajority(int setSize) {
    return (setSize / 2) + 1;
  }

  @Override
  public String toString() {
    return "QuorumConfiguration{" +
        "isTransitional=" + isTransitional +
        ", allPeers=" + allPeers +
        ", prevPeers=" + prevPeers +
        ", nextPeers=" + nextPeers +
        '}';
  }

  @Override
  public boolean equals(Object o) {
    if (this == o) {
      return true;
    }
    if (o == null || getClass() != o.getClass()) {
      return false;
    }

    QuorumConfiguration that = (QuorumConfiguration) o;

    return isTransitional == that.isTransitional
        && allPeers.equals(that.allPeers)
        && nextPeers.equals(that.nextPeers)
        && prevPeers.equals(that.prevPeers);
  }

  @Override
  public int hashCode() {
    int result = (isTransitional ? 1 : 0);
    result = 31 * result + allPeers.hashCode();
    result = 31 * result + prevPeers.hashCode();
    result = 31 * result + nextPeers.hashCode();
    return result;
  }
} public interface ReplicatorLog {
  /**
   * Log entries to the log.  The entries are in order, and should start from getLastIndex() + 1.
   * <p>
   * The implementation should feel free to verify this.
   * <p>
   * After this call returns, the log implementation should mark these entries as "to be committed"
   * and calls to 'getLastIndex()' should return the last entry in "entries".  The implementation will
   * then return a future that will be set true upon successful sync to durable storage, or else
   * set with an exception.
   * <p>
   * Note that over time, multiple calls to logEntries() may be issued before the prior call has signaled
   * full sync to the client.  This also implies that once this call returns, calls to the other methods
   * of this interface must now return data from these entries.  For example calling getLogTerm(long) should
   * return data from these entries even if they haven't been quite sync'ed to disk yet.
   *
   * @param entries new log entries
   * @return an future that indicates success.
   */
  ListenableFuture<Boolean> logEntries(List<LogEntry> entries);

  /**
   * Get a future which will return the entries in a specified range of indexes from start, inclusive, to end,
   * exclusive. If start and end are equal, the returned list will be empty.
   *
   * @param start the index of the low endpoint of the range (inclusive)
   * @param end   the index of the high endpoint of the range (exclusive)
   * @return a future which will yield the requested entries, or an exception if an error occurs.
   */
  ListenableFuture<List<LogEntry>> getLogEntries(long start, long end);

  /**
   * Get the term for a given log index. If the given index is not present in the log, then this
   * will return 0. A term value of 0 should be considered invalid. This is expected to be fast,
   * so it's a synchronous interface.
   *
   * @param index the index to look up the term for
   * @return the term value at 'index', or 0 if no such entry
   */
  long getLogTerm(long index);

  /**
   * Gets the term value for the last entry in the log. if the log is empty, then this will return
   * 0. A term value of 0 should never be valid.
   *
   * @return the last term or 0 if no such entry
   */
  long getLastTerm();

  /**
   * Gets the index of the most recent log entry.  An index is like a log sequence number, but there are
   * no holes.
   *
   * @return the index or 0 if the log is empty. This implies log entries start at 1.
   */
  long getLastIndex();

  /**
   * Delete all log entries after and including the specified index.
   * <p>
   * To persist the deletion, this might take a few so use a future.
   *
   * @param entryIndex the index entry to truncate log from.
   * @return A future set to true upon completion, or set with an exception upon failure.
   */
  ListenableFuture<Boolean> truncateLog(long entryIndex);

  /**
   * Get the latest, and thus current, configuration of the replicator's peers. (The configuration is
   * logged just like any other entry.)
   *
   * @return The latest configuration; or if there is none, an empty configuration.
   */
  QuorumConfiguration getLastConfiguration();

  /**
   * Get the index of the entry with the latest configuration of the replicator's peers.
   *
   * @return The greatest index of any configuration entry in the log, or zero if there is none.
   */
  long getLastConfigurationIndex();
} public interface CheckedConsumer<T, E extends Throwable> {

  void accept(T t) throws E;
} static Matcher<List<OLogEntry>> aListOfEntriesWithConsecutiveSeqNums(long start, long end) {
  return new TypeSafeMatcher<List<OLogEntry>>() {
    @Override
    protected boolean matchesSafely(List<OLogEntry> entries) {
      if (entries.size() != (end - start)) {
        return false;
      }
      long expectedSeqNum = start;
      for (OLogEntry entry : entries) {
        if (entry.getSeqNum() != expectedSeqNum) {
          return false;
        }
        expectedSeqNum++;
      }
      return true;
    }

    @Override
    public void describeTo(Description description) {
      description.appendText("a list of OLogEntry with consecutive sequence numbers from ")
          .appendValue(start).appendText(" inclusive to ").appendValue(end).appendText(" exclusive");
    }
  };
} public static List<OLogEntry> makeSingleEntryList(long seqNum, long term, String stringData) {
  return Lists.newArrayList(makeEntry(seqNum, term, stringData));
} public static List<OLogEntry> someConsecutiveEntries(long start, long end) {
  List<OLogEntry> entries = new ArrayList<>();
  for (long i = start; i < end; i++) {
    entries.add(makeEntry(i, LongMath.divide(i + 1, 2, RoundingMode.CEILING), someData()));
  }
  return entries;
} public static long seqNum(long seqNum) {
  return seqNum;
} public static ByteBuffer someData() {
  byte[] bytes = new byte[10];
  deterministicDataSequence.nextBytes(bytes);
  return ByteBuffer.wrap(bytes);
} public static long term(long term) {
  return term;
} public static LogSequenceBuilder entries() {
  return new LogSequenceBuilder();
}
import c5db.C5CommonTestUtil;
import c5db.interfaces.replication.QuorumConfiguration;
import c5db.interfaces.replication.ReplicatorLog;
import c5db.util.CheckedConsumer;
import c5db.util.WrappingKeySerializingExecutor;
import com.google.common.collect.Lists;
import org.hamcrest.core.Is;
import org.junit.Test;
import java.nio.file.Path;
import java.util.concurrent.Executors;
import static c5db.FutureMatchers.resultsIn;
import static c5db.log.LogMatchers.aListOfEntriesWithConsecutiveSeqNums;
import static c5db.log.LogTestUtil.makeSingleEntryList;
import static c5db.log.LogTestUtil.someConsecutiveEntries;
import static c5db.log.ReplicatorLogGenericTestUtil.seqNum;
import static c5db.log.ReplicatorLogGenericTestUtil.someData;
import static c5db.log.ReplicatorLogGenericTestUtil.term;
import static c5db.replication.ReplicatorTestUtil.entries;
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.equalTo;
/*
 * Copyright 2014 WANdisco
 *
 *  WANdisco licenses this file to you 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 c5db.log;




/**
 * Test the chain Mooring -> QuorumDelegatingLog -> EncodedSequentialLog -> (disk), ensuring that
 * persisted information can be reconstructed from the information left on disk, so that a replicator
 * can pick up where it left off after a crash.
 */
public class LogIntegratedPersistenceTest {
  private static final String QUORUM_ID = "LogIntegratedPersistenceTest";
  private static final int NUM_THREADS = 3;

  private final Path testDirectory = (new C5CommonTestUtil()).getDataTestDir("olog");

  @Test
  public void anEmptyLogReturnsDefaultValuesForIndexesAndTermsAndConfigurations() throws Exception {
    withReplicatorLog((log) -> {
      assertThat(log.getLastIndex(), is(equalTo(0L)));
      assertThat(log.getLastTerm(), is(equalTo(0L)));
      assertThat(log.getLastConfigurationIndex(), is(equalTo(0L)));
      assertThat(log.getLastConfiguration(), is(equalTo(QuorumConfiguration.EMPTY)));
    });
  }

  @Test
  public void termAndIndexInformationIsPersisted() throws Exception {
    withReplicatorLog((log) -> {
      log.logEntries(
          entries()
              .term(77).indexes(1, 2)
              .term(78).indexes(3, 4).build());

    });

    withReplicatorLog((log) -> {
      assertThat(log.getLastIndex(), is(equalTo(4L)));
      assertThat(log.getLastTerm(), is(equalTo(78L)));

      assertThat(log.getLogTerm(3), is(equalTo(78L)));
      assertThat(log.getLogTerm(2), is(equalTo(77L)));
    });
  }

  @Test
  public void configurationInformationIsPersisted() throws Exception {
    final QuorumConfiguration firstConfig = QuorumConfiguration
        .of(Lists.newArrayList(1L, 2L, 3L))
        .getTransitionalConfiguration(Lists.newArrayList(2L, 3L, 4L));
    final QuorumConfiguration secondConfig = firstConfig.getCompletedConfiguration();

    withReplicatorLog((log) -> {
      log.logEntries(
          entries()
              .term(97).indexes(1, 2)
              .term(98).configurationAndIndex(firstConfig, 3)
              .term(99).indexes(4, 5)
              .term(99).configurationAndIndex(secondConfig, 6).build());

    });

    withReplicatorLog((log) -> {
      assertThat(log.getLastConfiguration(), is(equalTo(secondConfig)));
      assertThat(log.getLastConfigurationIndex(), is(equalTo(6L)));

      log.truncateLog(seqNum(4));

      assertThat(log.getLastConfiguration(), is(equalTo(firstConfig)));
      assertThat(log.getLastConfigurationIndex(), is(equalTo(3L)));

      log.truncateLog(seqNum(2));

      assertThat(log.getLastConfiguration(), is(equalTo(QuorumConfiguration.EMPTY)));
      assertThat(log.getLastConfigurationIndex(), is(equalTo(0L)));
    });
  }

  @Test(expected = RuntimeException.class)
  public void informationAboutWhatSequenceNumberComesNextIsPersistedSoThatAnIncorrectSeqNumCanBeCaught()
      throws Exception {
    withReplicatorLog((log) -> {
      log.logEntries(
          entries().term(1).indexes(1, 2, 3).build());
    });

    withReplicatorLog((log) -> {
      log.logEntries(
          entries().term(2).indexes(5, 6, 7).build());
    });
  }

  @Test(timeout = 3000)
  public void headerInformationIsPersistedAfterARollOperation() throws Exception {
    final QuorumConfiguration configuration = QuorumConfiguration.of(Lists.newArrayList(1L, 2L, 3L));
    final long lastTerm = 97;

    try (OLog oLog = getOLog()) {
      new Mooring(oLog, QUORUM_ID)
          .logEntries(entries()
              .term(lastTerm)
              .configurationAndIndex(configuration, 1).build());
      oLog.roll(QUORUM_ID).get();
    }

    withReplicatorLog((log) -> {
      assertThat(log.getLastTerm(), is(equalTo(lastTerm)));
      assertThat(log.getLastConfiguration(), is(equalTo(configuration)));
    });
  }

  @Test(timeout = 3000)
  public void rollsKeepingTrackOfTermCorrectly() throws Exception {
    withOpenOLog((oLog) -> {
      oLog.logEntries(makeSingleEntryList(seqNum(1), term(17), someData()), QUORUM_ID);
      oLog.roll(QUORUM_ID);
      oLog.logEntries(makeSingleEntryList(seqNum(2), term(56), someData()), QUORUM_ID);
    });

    withOpenOLog((oLog) -> {
      assertThat(oLog.getLastTerm(QUORUM_ID), Is.is(equalTo(56L)));

      oLog.truncateLog(seqNum(2), QUORUM_ID);
      assertThat(oLog.getLastTerm(QUORUM_ID), Is.is(equalTo(17L)));
    });
  }

  @Test(timeout = 3000)
  public void performsMultiLogFetchesCorrectlyFromPersistence() throws Exception {
    withOpenOLog((oLog) -> {
      oLog.logEntries(someConsecutiveEntries(1, 6), QUORUM_ID);
      oLog.roll(QUORUM_ID);
      oLog.logEntries(someConsecutiveEntries(6, 11), QUORUM_ID);
      oLog.roll(QUORUM_ID);
      oLog.logEntries(someConsecutiveEntries(11, 16), QUORUM_ID);
    });

    withOpenOLog((oLog) ->
        assertThat(oLog.getLogEntries(3, 15, QUORUM_ID), resultsIn(aListOfEntriesWithConsecutiveSeqNums(3, 15))));