@Test public void readConsumeLog() throws InterruptedException { writeConsumeLog(); String testPath = getTestPath(); ConsumeArchiveService.ArchiveMappedFileRepository archiveMappedFileRepository = new ConsumeArchiveService.ArchiveMappedFileRepository(testPath); for (int i = 0; i < writeRecordNum; i++) { byte[] bytes = archiveMappedFileRepository.readOne(); if (bytes.length > 0) { ConsumeLog read = ArchiveSerializer.read(ByteBuffer.wrap(bytes)); Assert.assertEquals(i, read.getBrokerId()); } } }
private List<ConsumeLog> readConsumeLog(int count) { readByteCounter.set(0); List<ConsumeLog> list = new LinkedList<>(); for (int i = 0; i < count; i++) { byte[] bytes = repository.readOne(); if (bytes.length > 0) { readByteCounter.addAndGet(1 + 4 + bytes.length); list.add(ArchiveSerializer.read(ByteBuffer.wrap(bytes))); } else { break; } } return list; }
ConsumeArchiveService extends Service { private List<ConsumeLog> readConsumeLog(int count) { readByteCounter.set(0); List<ConsumeLog> list = new LinkedList<>(); for (int i = 0; i < count; i++) { byte[] bytes = repository.readOne(); if (bytes.length > 0) { readByteCounter.addAndGet(1 + 4 + bytes.length); list.add(ArchiveSerializer.read(ByteBuffer.wrap(bytes))); } else { break; } } return list; } }
ConsumeArchiveService extends Service { private List<ConsumeLog> readConsumeLog(int count) { readByteCounter.set(0); List<ConsumeLog> list = new LinkedList<>(); for (int i = 0; i < count; i++) { byte[] bytes = repository.readOne(); if (bytes.length > 0) { readByteCounter.addAndGet(1 + 4 + bytes.length); list.add(ArchiveSerializer.read(ByteBuffer.wrap(bytes))); } else { break; } } return list; } ConsumeArchiveService(ArchiveConfig archiveConfig, ClusterManager clusterManager); }
ConsumeArchiveService extends Service { private List<ConsumeLog> readConsumeLog(int count) { readByteCounter.set(0); List<ConsumeLog> list = new LinkedList<>(); for (int i = 0; i < count; i++) { byte[] bytes = repository.readOne(); if (bytes.length > 0) { readByteCounter.addAndGet(1 + 4 + bytes.length); list.add(ArchiveSerializer.read(ByteBuffer.wrap(bytes))); } else { break; } } return list; } ConsumeArchiveService(ArchiveConfig archiveConfig, ClusterManager clusterManager); long getRemainConsumeLogFileNum(); void appendConsumeLog(Connection connection, MessageLocation[] locations); }
ConsumeArchiveService extends Service { private List<ConsumeLog> readConsumeLog(int count) { readByteCounter.set(0); List<ConsumeLog> list = new LinkedList<>(); for (int i = 0; i < count; i++) { byte[] bytes = repository.readOne(); if (bytes.length > 0) { readByteCounter.addAndGet(1 + 4 + bytes.length); list.add(ArchiveSerializer.read(ByteBuffer.wrap(bytes))); } else { break; } } return list; } ConsumeArchiveService(ArchiveConfig archiveConfig, ClusterManager clusterManager); long getRemainConsumeLogFileNum(); void appendConsumeLog(Connection connection, MessageLocation[] locations);  }