@Test public void recoverTest() throws IOException { JournalPersistence store = new PositioningStore(); store.recover(path, new Properties()); int size = 10; int maxLength = 999; long start = store.max(); List<byte[]> journals = ByteUtils.createRandomSizeByteList(maxLength, size); int length = journals.stream().mapToInt(journal -> journal.length).sum(); long writePosition = 0L; for (byte[] journal : journals) { writePosition = store.append(journal); } Assert.assertEquals(length + start, writePosition); Assert.assertEquals(writePosition, store.max()); while (store.flushed() < store.max()) { store.flush(); } store.close(); store = new PositioningStore(); store.recover(path, new Properties()); byte[] readBytes = store.read(start, length); byte[] writeBytes = ByteUtils.concatBytes(journals); Assert.assertArrayEquals(writeBytes, readBytes); store.close(); }
public void recover(Path path, long min, Properties properties) throws IOException { Files.createDirectories(path); this.base = path.toFile(); this.config = toConfig(properties); bufferPool.addPreLoad(config.getFileDataSize(), config.getCachedFileCoreCount(), config.getCachedFileMaxCount()); recoverFileMap(min); long recoverPosition = this.storeFileMap.isEmpty() ? min : this.storeFileMap.lastKey() + this.storeFileMap.lastEntry().getValue().fileDataSize(); flushPosition.set(recoverPosition); writePosition.set(recoverPosition); leftPosition.set(this.storeFileMap.isEmpty() ? min : this.storeFileMap.firstKey()); resetWriteStoreFile(); if (logger.isDebugEnabled()) { logger.debug("Store loaded, left: {}, right: {}, base: {}.", ThreadSafeFormat.formatWithComma(min()), ThreadSafeFormat.formatWithComma(max()), base.getAbsolutePath()); } }
PositioningStore implements JournalPersistence, MonitoredPersistence, Closeable { public void recover(Path path, long min, Properties properties) throws IOException { Files.createDirectories(path); this.base = path.toFile(); this.config = toConfig(properties); bufferPool.addPreLoad(config.getFileDataSize(), config.getCachedFileCoreCount(), config.getCachedFileMaxCount()); recoverFileMap(min); long recoverPosition = this.storeFileMap.isEmpty() ? min : this.storeFileMap.lastKey() + this.storeFileMap.lastEntry().getValue().fileDataSize(); flushPosition.set(recoverPosition); writePosition.set(recoverPosition); leftPosition.set(this.storeFileMap.isEmpty() ? min : this.storeFileMap.firstKey()); resetWriteStoreFile(); if (logger.isDebugEnabled()) { logger.debug("Store loaded, left: {}, right: {}, base: {}.", ThreadSafeFormat.formatWithComma(min()), ThreadSafeFormat.formatWithComma(max()), base.getAbsolutePath()); } } }
PositioningStore implements JournalPersistence, MonitoredPersistence, Closeable { public void recover(Path path, long min, Properties properties) throws IOException { Files.createDirectories(path); this.base = path.toFile(); this.config = toConfig(properties); bufferPool.addPreLoad(config.getFileDataSize(), config.getCachedFileCoreCount(), config.getCachedFileMaxCount()); recoverFileMap(min); long recoverPosition = this.storeFileMap.isEmpty() ? min : this.storeFileMap.lastKey() + this.storeFileMap.lastEntry().getValue().fileDataSize(); flushPosition.set(recoverPosition); writePosition.set(recoverPosition); leftPosition.set(this.storeFileMap.isEmpty() ? min : this.storeFileMap.firstKey()); resetWriteStoreFile(); if (logger.isDebugEnabled()) { logger.debug("Store loaded, left: {}, right: {}, base: {}.", ThreadSafeFormat.formatWithComma(min()), ThreadSafeFormat.formatWithComma(max()), base.getAbsolutePath()); } } PositioningStore(); }
PositioningStore implements JournalPersistence, MonitoredPersistence, Closeable { public void recover(Path path, long min, Properties properties) throws IOException { Files.createDirectories(path); this.base = path.toFile(); this.config = toConfig(properties); bufferPool.addPreLoad(config.getFileDataSize(), config.getCachedFileCoreCount(), config.getCachedFileMaxCount()); recoverFileMap(min); long recoverPosition = this.storeFileMap.isEmpty() ? min : this.storeFileMap.lastKey() + this.storeFileMap.lastEntry().getValue().fileDataSize(); flushPosition.set(recoverPosition); writePosition.set(recoverPosition); leftPosition.set(this.storeFileMap.isEmpty() ? min : this.storeFileMap.firstKey()); resetWriteStoreFile(); if (logger.isDebugEnabled()) { logger.debug("Store loaded, left: {}, right: {}, base: {}.", ThreadSafeFormat.formatWithComma(min()), ThreadSafeFormat.formatWithComma(max()), base.getAbsolutePath()); } } PositioningStore(); void truncate(long givenMax); void delete(); void recover(Path path, long min, Properties properties); @Override long append(byte[] bytes); @Override long append(List<byte[]> bytesList ); @Override long min(); @Override long physicalMin(); @Override long max(); @Override long flushed(); @Override void flush(); byte[] read(long position, int length); Long readLong(long position); long compact(long givenMin); @Override Path getBasePath(); @Override void close(); @Override Path getPath(); @Override long getFreeSpace(); @Override long getTotalSpace(); @Override String toString(); }
PositioningStore implements JournalPersistence, MonitoredPersistence, Closeable { public void recover(Path path, long min, Properties properties) throws IOException { Files.createDirectories(path); this.base = path.toFile(); this.config = toConfig(properties); bufferPool.addPreLoad(config.getFileDataSize(), config.getCachedFileCoreCount(), config.getCachedFileMaxCount()); recoverFileMap(min); long recoverPosition = this.storeFileMap.isEmpty() ? min : this.storeFileMap.lastKey() + this.storeFileMap.lastEntry().getValue().fileDataSize(); flushPosition.set(recoverPosition); writePosition.set(recoverPosition); leftPosition.set(this.storeFileMap.isEmpty() ? min : this.storeFileMap.firstKey()); resetWriteStoreFile(); if (logger.isDebugEnabled()) { logger.debug("Store loaded, left: {}, right: {}, base: {}.", ThreadSafeFormat.formatWithComma(min()), ThreadSafeFormat.formatWithComma(max()), base.getAbsolutePath()); } } PositioningStore(); void truncate(long givenMax); void delete(); void recover(Path path, long min, Properties properties); @Override long append(byte[] bytes); @Override long append(List<byte[]> bytesList ); @Override long min(); @Override long physicalMin(); @Override long max(); @Override long flushed(); @Override void flush(); byte[] read(long position, int length); Long readLong(long position); long compact(long givenMin); @Override Path getBasePath(); @Override void close(); @Override Path getPath(); @Override long getFreeSpace(); @Override long getTotalSpace(); @Override String toString();  }