feat: new BlockReader implementation for block-as-file (#385)

Signed-off-by: Atanas Atanasov <a.v.atanasov98@gmail.com>

server/src/main/java/com/hedera/block/server/persistence/PersistenceInjectionModule.java

@@ -92,14 +92,17 @@ static BlockWriter<List<BlockItemUnparsed>> providesBlockWriter(
      *
      * @param config the persistence storage configuration needed to build the
      * block reader
+     * @param blockPathResolver the block path resolver needed to build
+     * the block reader
      * @return a block reader singleton
      */
     @Provides
     @Singleton
-    static BlockReader<BlockUnparsed> providesBlockReader(@NonNull final PersistenceStorageConfig config) {
+    static BlockReader<BlockUnparsed> providesBlockReader(
+            @NonNull final PersistenceStorageConfig config, @NonNull final BlockPathResolver blockPathResolver) {
         final StorageType persistenceType = config.type();
         return switch (persistenceType) {
-            case BLOCK_AS_LOCAL_FILE -> BlockAsLocalFileReader.newInstance();
+            case BLOCK_AS_LOCAL_FILE -> BlockAsLocalFileReader.of(blockPathResolver);
             case BLOCK_AS_LOCAL_DIRECTORY -> BlockAsLocalDirReader.of(config);
             case NO_OP -> NoOpBlockReader.newInstance();
         };

server/src/main/java/com/hedera/block/server/persistence/storage/read/BlockAsLocalFileReader.java

@@ -17,34 +17,61 @@
 package com.hedera.block.server.persistence.storage.read;
 
 import com.hedera.block.common.utils.Preconditions;
+import com.hedera.block.server.persistence.storage.path.BlockPathResolver;
 import com.hedera.hapi.block.BlockUnparsed;
+import com.hedera.pbj.runtime.ParseException;
+import com.hedera.pbj.runtime.io.stream.ReadableStreamingData;
 import edu.umd.cs.findbugs.annotations.NonNull;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Objects;
 import java.util.Optional;
 
 /**
  * A Block reader that reads block-as-file.
  */
 public final class BlockAsLocalFileReader implements LocalBlockReader<BlockUnparsed> {
+    private final BlockPathResolver pathResolver;
+
     /**
      * Constructor.
+     *
+     * @param pathResolver valid, {@code non-null} instance of
+     * {@link BlockPathResolver} used to resolve paths to block files
      */
-    private BlockAsLocalFileReader() {}
+    private BlockAsLocalFileReader(@NonNull final BlockPathResolver pathResolver) {
+        this.pathResolver = Objects.requireNonNull(pathResolver);
+    }
 
     /**
      * This method creates and returns a new instance of
      * {@link BlockAsLocalFileReader}.
      *
+     * @param pathResolver valid, {@code non-null} instance of
+     * {@link BlockPathResolver} used to resolve paths to block files
      * @return a new, fully initialized instance of
      * {@link BlockAsLocalFileReader}
      */
-    public static BlockAsLocalFileReader newInstance() {
-        return new BlockAsLocalFileReader();
+    public static BlockAsLocalFileReader of(@NonNull final BlockPathResolver pathResolver) {
+        return new BlockAsLocalFileReader(pathResolver);
     }
 
     @NonNull
     @Override
-    public Optional<BlockUnparsed> read(final long blockNumber) {
+    public Optional<BlockUnparsed> read(final long blockNumber) throws IOException, ParseException {
         Preconditions.requireWhole(blockNumber);
-        throw new UnsupportedOperationException("Not implemented yet");
+        final Path resolvedBlockPath = pathResolver.resolvePathToBlock(blockNumber);
+        if (Files.exists(resolvedBlockPath)) {
+            return Optional.of(doRead(resolvedBlockPath));
+        } else {
+            return Optional.empty();
+        }
+    }
+
+    private BlockUnparsed doRead(final Path resolvedBlockPath) throws IOException, ParseException {
+        try (final ReadableStreamingData data = new ReadableStreamingData(resolvedBlockPath)) {
+            return BlockUnparsed.PROTOBUF.parse(data);
+        }
     }
 }

server/src/test/java/com/hedera/block/server/persistence/PersistenceInjectionModuleTest.java

@@ -179,7 +179,7 @@ void testProvidesBlockReader(final StorageType storageType) {
         when(persistenceStorageConfigMock.type()).thenReturn(storageType);
 
         final BlockReader<BlockUnparsed> actual =
-                PersistenceInjectionModule.providesBlockReader(persistenceStorageConfigMock);
+                PersistenceInjectionModule.providesBlockReader(persistenceStorageConfigMock, blockPathResolverMock);
 
         final Class<?> targetInstanceType =
                 switch (storageType) {

server/src/test/java/com/hedera/block/server/persistence/storage/read/BlockAsLocalFileReaderTest.java

@@ -16,11 +16,32 @@
 
 package com.hedera.block.server.persistence.storage.read;
 
+import static com.hedera.block.server.util.PersistTestUtils.PERSISTENCE_STORAGE_LIVE_ROOT_PATH_KEY;
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
-import static org.assertj.core.api.AssertionsForClassTypes.assertThatExceptionOfType;
+import static org.assertj.core.api.Assertions.from;
+import static org.mockito.Mockito.spy;
 
+import com.hedera.block.server.config.BlockNodeContext;
+import com.hedera.block.server.persistence.storage.PersistenceStorageConfig;
+import com.hedera.block.server.persistence.storage.path.BlockAsLocalFilePathResolver;
+import com.hedera.block.server.persistence.storage.write.BlockAsLocalFileWriter;
+import com.hedera.block.server.util.PersistTestUtils;
+import com.hedera.block.server.util.TestConfigUtil;
+import com.hedera.hapi.block.BlockItemUnparsed;
+import com.hedera.hapi.block.BlockUnparsed;
+import com.hedera.hapi.block.stream.output.BlockHeader;
+import com.hedera.pbj.runtime.ParseException;
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
 import java.util.stream.Stream;
+import org.assertj.core.api.InstanceOfAssertFactories;
 import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.io.TempDir;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.MethodSource;
@@ -29,25 +50,107 @@
  * Tests for the {@link BlockAsLocalFileReader} class.
  */
 class BlockAsLocalFileReaderTest {
+    private BlockAsLocalFileWriter blockAsLocalFileWriterMock;
     private BlockAsLocalFileReader toTest;
 
+    @TempDir
+    private Path testLiveRootPath;
+
     @BeforeEach
-    void setUp() {
-        toTest = BlockAsLocalFileReader.newInstance();
+    void setUp() throws IOException {
+        final BlockNodeContext blockNodeContext = TestConfigUtil.getTestBlockNodeContext(
+                Map.of(PERSISTENCE_STORAGE_LIVE_ROOT_PATH_KEY, testLiveRootPath.toString()));
+        final PersistenceStorageConfig testConfig =
+                blockNodeContext.configuration().getConfigData(PersistenceStorageConfig.class);
+
+        final String testConfigLiveRootPath = testConfig.liveRootPath();
+        assertThat(testConfigLiveRootPath).isEqualTo(testLiveRootPath.toString());
+
+        final BlockAsLocalFilePathResolver blockAsLocalFileResolverMock =
+                spy(BlockAsLocalFilePathResolver.of(testLiveRootPath));
+        blockAsLocalFileWriterMock = spy(BlockAsLocalFileWriter.of(blockNodeContext, blockAsLocalFileResolverMock));
+        toTest = BlockAsLocalFileReader.of(blockAsLocalFileResolverMock);
     }
 
     /**
      * This test aims to verify that the
-     * {@link BlockAsLocalFileReader#read(long)} correctly reads a block with
-     * a given block number.
-     *
-     * @param toRead parameterized, block number
+     * {@link BlockAsLocalFileReader#read(long)} correctly reads a block with a
+     * given block number and returns a {@code non-empty} {@link Optional}.
+     */
+    @ParameterizedTest
+    @MethodSource("validBlockNumbers")
+    void testSuccessfulBlockRead(final long blockNumber) throws IOException, ParseException {
+        final List<BlockItemUnparsed> blockItemUnparsed =
+                PersistTestUtils.generateBlockItemsUnparsedForWithBlockNumber(blockNumber);
+        final Optional<List<BlockItemUnparsed>> written = blockAsLocalFileWriterMock.write(blockItemUnparsed);
+
+        // writing the test data is successful
+        assertThat(written).isNotNull().isPresent();
+
+        final Optional<BlockUnparsed> actual = toTest.read(blockNumber);
+        assertThat(actual)
+                .isNotNull()
+                .isPresent()
+                .get(InstanceOfAssertFactories.type(BlockUnparsed.class))
+                .isNotNull()
+                .isExactlyInstanceOf(BlockUnparsed.class)
+                .returns(blockNumber, from(blockUnparsed -> {
+                    try {
+                        return BlockHeader.PROTOBUF
+                                .parse(Objects.requireNonNull(
+                                        blockUnparsed.blockItems().getFirst().blockHeader()))
+                                .number();
+                    } catch (final ParseException e) {
+                        throw new RuntimeException(e);
+                    }
+                }))
+                .extracting(BlockUnparsed::blockItems)
+                .asList()
+                .isNotNull()
+                .isNotEmpty();
+    }
+
+    /**
+     * This test aims to verify that the
+     * {@link BlockAsLocalFileReader#read(long)} correctly reads a block with a
+     * given block number and has the same contents as the block that has been
+     * persisted.
+     */
+    @ParameterizedTest
+    @MethodSource("validBlockNumbers")
+    void testSuccessfulBlockReadContents(final long blockNumber) throws IOException, ParseException {
+        final List<BlockItemUnparsed> blockItemUnparsed =
+                PersistTestUtils.generateBlockItemsUnparsedForWithBlockNumber(blockNumber);
+        final Optional<List<BlockItemUnparsed>> written = blockAsLocalFileWriterMock.write(blockItemUnparsed);
+
+        // writing the test data is successful
+        assertThat(written).isNotNull().isPresent();
+
+        final Optional<BlockUnparsed> actual = toTest.read(blockNumber);
+        assertThat(actual)
+                .isNotNull()
+                .isPresent()
+                .get(InstanceOfAssertFactories.type(BlockUnparsed.class))
+                .isNotNull()
+                .extracting(BlockUnparsed::blockItems)
+                .asList()
+                .isNotNull()
+                .isNotEmpty()
+                .hasSize(blockItemUnparsed.size())
+                .containsExactlyElementsOf(blockItemUnparsed);
+    }
+
+    /**
+     * This test aims to verify that the
+     * {@link BlockAsLocalFileReader#read(long) correctly returns an empty
+     * {@link Optional} when no block file is found for the given valid block
+     * number.
      */
     @ParameterizedTest
     @MethodSource("validBlockNumbers")
-    void testSuccessfulBlockRead(final long toRead) {
-        // todo currently throws UnsupportedOperationException
-        assertThatExceptionOfType(UnsupportedOperationException.class).isThrownBy(() -> toTest.read(toRead));
+    void testEmptyOptWhenNoBLockFileFound(final long blockNumber) throws IOException, ParseException {
+        final Optional<BlockUnparsed> actual = toTest.read(blockNumber);
+        assertThat(actual).isNotNull().isEmpty();
     }
 
     /**

server/src/test/java/com/hedera/block/server/util/PersistTestUtils.java

@@ -51,45 +51,73 @@ public static void writeBytesToPath(final Path path, final byte[] bytes) throws
         }
     }
 
-    public static List<BlockItemUnparsed> generateBlockItemsUnparsed(int numOfBlocks) {
-
-        List<BlockItemUnparsed> blockItems = new ArrayList<>();
-        for (int i = 1; i <= numOfBlocks; i++) {
-            for (int j = 1; j <= 10; j++) {
-                switch (j) {
-                    case 1:
-                        // First block is always the header
-                        blockItems.add(BlockItemUnparsed.newBuilder()
-                                .blockHeader(BlockHeader.PROTOBUF.toBytes(BlockHeader.newBuilder()
-                                        .number(i)
-                                        .softwareVersion(SemanticVersion.newBuilder()
-                                                .major(1)
-                                                .minor(0)
-                                                .build())
-                                        .build()))
-                                .build());
-                        break;
-                    case 10:
-                        // Last block is always the state proof
-                        blockItems.add(BlockItemUnparsed.newBuilder()
-                                .blockProof(BlockProof.PROTOBUF.toBytes(
-                                        BlockProof.newBuilder().block(i).build()))
-                                .build());
-                        break;
-                    default:
-                        // Middle blocks are events
-                        blockItems.add(BlockItemUnparsed.newBuilder()
-                                .eventHeader(EventHeader.PROTOBUF.toBytes(EventHeader.newBuilder()
-                                        .eventCore(EventCore.newBuilder()
-                                                .creatorNodeId(i)
-                                                .build())
-                                        .build()))
-                                .build());
-                        break;
-                }
+    /**
+     * This method generates a list of {@link BlockItemUnparsed} with the input
+     * blockNumber used to generate the block items for. It generates 10 block
+     * items starting with the block header, followed by 8 events and ending
+     * with the block proof.
+     *
+     * @param blockNumber the block number to generate the block items for
+     *
+     * @return a list of {@link BlockItemUnparsed} with the input blockNumber
+     * used to generate the block items for
+     */
+    public static List<BlockItemUnparsed> generateBlockItemsUnparsedForWithBlockNumber(final long blockNumber) {
+        final List<BlockItemUnparsed> result = new ArrayList<>();
+        for (int j = 1; j <= 10; j++) {
+            switch (j) {
+                case 1:
+                    // First block is always the header
+                    result.add(BlockItemUnparsed.newBuilder()
+                            .blockHeader(BlockHeader.PROTOBUF.toBytes(BlockHeader.newBuilder()
+                                    .number(blockNumber)
+                                    .softwareVersion(SemanticVersion.newBuilder()
+                                            .major(1)
+                                            .minor(0)
+                                            .build())
+                                    .build()))
+                            .build());
+                    break;
+                case 10:
+                    // Last block is always the state proof
+                    result.add(BlockItemUnparsed.newBuilder()
+                            .blockProof(BlockProof.PROTOBUF.toBytes(
+                                    BlockProof.newBuilder().block(blockNumber).build()))
+                            .build());
+                    break;
+                default:
+                    // Middle blocks are events
+                    result.add(BlockItemUnparsed.newBuilder()
+                            .eventHeader(EventHeader.PROTOBUF.toBytes(EventHeader.newBuilder()
+                                    .eventCore(EventCore.newBuilder()
+                                            .creatorNodeId(blockNumber)
+                                            .build())
+                                    .build()))
+                            .build());
+                    break;
             }
         }
+        return result;
+    }
 
+    /**
+     * This method generates a list of {@link BlockItemUnparsed} for as many
+     * blocks as specified by the input parameter numOfBlocks. For each block
+     * number from 1 to numOfBlocks, it generates 10 block items starting with
+     * the block header, followed by 8 events and ending with the block proof.
+     * In a way, this simulates a stream of block items. Each 10 items in the
+     * list represent a block.
+     *
+     * @param numOfBlocks the number of blocks to generate block items for
+     *
+     * @return a list of {@link BlockItemUnparsed} for as many blocks as
+     * specified by the input parameter numOfBlocks
+     */
+    public static List<BlockItemUnparsed> generateBlockItemsUnparsed(int numOfBlocks) {
+        final List<BlockItemUnparsed> blockItems = new ArrayList<>();
+        for (int i = 1; i <= numOfBlocks; i++) {
+            blockItems.addAll(generateBlockItemsUnparsedForWithBlockNumber(i));
+        }
         return blockItems;
     }