FileBasedReceivedSnapshot implements ReceivedSnapshot { @Override public PersistedSnapshot persist() { if (snapshotStore.hasSnapshotId(metadata.getSnapshotIdAsString())) { abort(); return snapshotStore.getLatestSnapshot().orElseThrow(); } final var files = directory.toFile().listFiles(); Objects.requireNonNull(files, "No chunks have been applied yet"); if (files.length != expectedTotalCount) { throw new IllegalStateException( String.format( "Expected '%d' chunk files for this snapshot, but found '%d'. Files are: %s.", expectedSnapshotChecksum, files.length, Arrays.toString(files))); } final var filePaths = Arrays.stream(files).sorted().map(File::toPath).collect(Collectors.toList()); final long actualSnapshotChecksum; try { actualSnapshotChecksum = ChecksumUtil.createCombinedChecksum(filePaths); } catch (final IOException e) { throw new UncheckedIOException("Unexpected exception on calculating snapshot checksum.", e); } if (actualSnapshotChecksum != expectedSnapshotChecksum) { throw new IllegalStateException( String.format( "Expected snapshot checksum %d, but calculated %d.", expectedSnapshotChecksum, actualSnapshotChecksum)); } return snapshotStore.newSnapshot(metadata, directory); } FileBasedReceivedSnapshot(
      final FileBasedSnapshotMetadata metadata,
      final Path directory,
      final FileBasedSnapshotStore snapshotStore); @Override long index(); @Override boolean containsChunk(final ByteBuffer chunkId); @Override boolean isExpectedChunk(final ByteBuffer chunkId); @Override void setNextExpected(final ByteBuffer nextChunkId); @Override boolean apply(final SnapshotChunk snapshotChunk); @Override void abort(); @Override PersistedSnapshot persist(); Path getPath(); @Override String toString();  }
@Test public void shouldPersistSnapshot() throws Exception { final var index = 1L; final var term = 0L; final var receivedSnapshot = takeAndReceiveSnapshot(index, term); final var snapshot = receivedSnapshot.persist(); assertThat(snapshot).isNotNull(); assertThat(receiverPendingSnapshotsDir.toFile().listFiles()).isEmpty(); assertThat(receiverSnapshotsDir).exists(); final var files = receiverSnapshotsDir.toFile().listFiles(); assertThat(files).isNotNull().hasSize(1); final var dir = files[0]; assertThat(dir).hasName(snapshot.getId()); final var snapshotFileList = dir.listFiles(); assertThat(snapshotFileList).isNotNull().extracting(File::getName).containsExactly("file1.txt"); }
@Test public void shouldNotDeletePersistedSnapshotOnPurgePendingOnStore() throws Exception { final var index = 1L; final var term = 0L; final var snapshot = takeAndReceiveSnapshot(index, term).persist(); receiverSnapshotStore.purgePendingSnapshots(); assertThat(receiverPendingSnapshotsDir.toFile().listFiles()).isEmpty(); assertThat(receiverSnapshotsDir).exists(); final var files = receiverSnapshotsDir.toFile().listFiles(); assertThat(files).isNotNull().hasSize(1); final var dir = files[0]; assertThat(dir).hasName(snapshot.getId()); final var snapshotFileList = dir.listFiles(); assertThat(snapshotFileList).isNotNull().extracting(File::getName).containsExactly("file1.txt"); }
@Test public void shouldReplaceSnapshotOnNextSnapshot() throws Exception { final var index = 1L; final var term = 0L; takeAndReceiveSnapshot(index, term).persist(); takeAndReceiveSnapshot(index + 1, term).persist(); assertThat(receiverPendingSnapshotsDir.toFile().listFiles()).isEmpty(); final var snapshotDirs = receiverSnapshotsDir.toFile().listFiles(); assertThat(snapshotDirs).isNotNull().hasSize(1); final var committedSnapshotDir = snapshotDirs[0]; assertThat( FileBasedSnapshotMetadata.ofFileName(committedSnapshotDir.getName()).get().getIndex()) .isEqualTo(2); assertThat(committedSnapshotDir.listFiles()) .isNotNull() .extracting(File::getName) .containsExactly("file1.txt"); }
@Test public void shouldRemovePendingSnapshotOnCommittingSnapshot() throws Exception { final var index = 1L; final var term = 0L; takeAndReceiveSnapshot(index, term); takeAndReceiveSnapshot(index + 1, term).persist(); assertThat(receiverPendingSnapshotsDir.toFile().listFiles()).isEmpty(); final var snapshotDirs = receiverSnapshotsDir.toFile().listFiles(); assertThat(snapshotDirs).isNotNull().hasSize(1); final var committedSnapshotDir = snapshotDirs[0]; assertThat( FileBasedSnapshotMetadata.ofFileName(committedSnapshotDir.getName()).get().getIndex()) .isEqualTo(2); assertThat(committedSnapshotDir.listFiles()) .isNotNull() .extracting(File::getName) .containsExactly("file1.txt"); }
@Test public void shouldNotRemovePendingSnapshotOnCommittingSnapshotWhenHigher() throws Exception { final var index = 1L; final var term = 0L; final FileBasedSnapshotStoreFactory fileBasedSnapshotStoreFactory = new FileBasedSnapshotStoreFactory(); fileBasedSnapshotStoreFactory.createReceivableSnapshotStore( temporaryFolder.newFolder("other").toPath(), "1"); final var otherStore = fileBasedSnapshotStoreFactory.getConstructableSnapshotStore("1"); final var olderTransient = otherStore.newTransientSnapshot(index, term, 1, 0).orElseThrow(); olderTransient.take(this::takeSnapshot); final var olderPersistedSnapshot = olderTransient.persist(); final PersistedSnapshot newPersistedSnapshot = takeSnapshot(index + 1, term); receiveSnapshot(newPersistedSnapshot); receiveSnapshot(olderPersistedSnapshot).persist(); final var pendingSnapshotDirs = receiverPendingSnapshotsDir.toFile().listFiles(); assertThat(pendingSnapshotDirs).isNotNull().hasSize(1); final var pendingSnapshotDir = pendingSnapshotDirs[0]; assertThat(FileBasedSnapshotMetadata.ofFileName(pendingSnapshotDir.getName()).get().getIndex()) .isEqualTo(2); assertThat(pendingSnapshotDir.listFiles()) .isNotNull() .extracting(File::getName) .containsExactly("file1.txt"); final var snapshotDirs = receiverSnapshotsDir.toFile().listFiles(); assertThat(snapshotDirs).isNotNull().hasSize(1); final var committedSnapshotDir = snapshotDirs[0]; assertThat(committedSnapshotDir.getName()).isEqualTo(olderPersistedSnapshot.getId()); assertThat(committedSnapshotDir.listFiles()) .isNotNull() .extracting(File::getName) .containsExactly("file1.txt"); }
@Test public void shouldNotifyListenersOnNewSnapshot() throws Exception { final var listener = mock(PersistedSnapshotListener.class); final var index = 1L; final var term = 0L; receiverSnapshotStore.addSnapshotListener(listener); final var persistedSnapshot = takeAndReceiveSnapshot(index, term).persist(); verify(listener, times(1)).onNewSnapshot(eq(persistedSnapshot)); }
@Test public void shouldNotNotifyListenersOnNewSnapshotWhenDeregistered() throws Exception { final var listener = mock(PersistedSnapshotListener.class); final var index = 1L; final var term = 0L; senderSnapshotStore.addSnapshotListener(listener); senderSnapshotStore.removeSnapshotListener(listener); final var persistedSnapshot = takeAndReceiveSnapshot(index, term).persist(); verify(listener, times(0)).onNewSnapshot(eq(persistedSnapshot)); }
@Test public void shouldReceiveConcurrentlyAndPersist() throws Exception { final var index = 1L; final var term = 0L; final var persistedSnapshot = takeSnapshot(index, term); final var receivedSnapshot = receiveSnapshot(persistedSnapshot); final var otherReceivedSnapshot = receiveSnapshot(persistedSnapshot); final var receivedPersisted = receivedSnapshot.persist(); final var otherReceivedPersisted = otherReceivedSnapshot.persist(); assertThat(receivedPersisted).isEqualTo(otherReceivedPersisted); final var snapshotDirs = receiverSnapshotsDir.toFile().listFiles(); assertThat(snapshotDirs).isNotNull().hasSize(1); final var committedSnapshotDir = snapshotDirs[0]; assertThat(committedSnapshotDir.getName()).isEqualTo(persistedSnapshot.getId()); assertThat(committedSnapshotDir.listFiles()) .isNotNull() .extracting(File::getName) .containsExactly("file1.txt"); assertThat(receiverPendingSnapshotsDir.toFile().listFiles()).isEmpty(); }
@Test public void shouldReceiveConcurrentlyAndPersistDoesnotDependOnTheOrder() throws Exception { final var index = 1L; final var term = 0L; final var persistedSnapshot = takeSnapshot(index, term); final var receivedSnapshot = receiveSnapshot(persistedSnapshot); final var otherReceivedSnapshot = receiveSnapshot(persistedSnapshot); final var otherReceivedPersisted = otherReceivedSnapshot.persist(); final var receivedPersisted = receivedSnapshot.persist(); assertThat(receivedPersisted).isEqualTo(otherReceivedPersisted); final var snapshotDirs = receiverSnapshotsDir.toFile().listFiles(); assertThat(snapshotDirs).isNotNull().hasSize(1); final var committedSnapshotDir = snapshotDirs[0]; assertThat(committedSnapshotDir.getName()).isEqualTo(persistedSnapshot.getId()); assertThat(committedSnapshotDir.listFiles()) .isNotNull() .extracting(File::getName) .containsExactly("file1.txt"); assertThat(receiverPendingSnapshotsDir.toFile().listFiles()).isEmpty(); }