Collapsing 3 commits together.

Signed-off-by: a-saksena <[email protected]>

Author: a-saksena

server/src/main/java/com/hedera/block/server/BlockStreamService.java

@@ -21,11 +21,14 @@
 import com.hedera.block.server.consumer.LiveStreamObserver;
 import com.hedera.block.server.consumer.LiveStreamObserverImpl;
 import com.hedera.block.server.mediator.StreamMediator;
+import com.hedera.block.server.persistence.BlockPersistenceHandler;
 import com.hedera.block.server.producer.ProducerBlockStreamObserver;
+import io.grpc.Status;
 import io.grpc.stub.StreamObserver;
 import io.helidon.webserver.grpc.GrpcService;
 
 import java.time.Clock;
+import java.util.Optional;
 
 import static io.helidon.webserver.grpc.ResponseHelper.complete;
 
@@ -45,6 +48,7 @@ public class BlockStreamService implements GrpcService {
 
     private final long timeoutThresholdMillis;
     private final StreamMediator<BlockStreamServiceGrpcProto.Block, BlockStreamServiceGrpcProto.BlockResponse> streamMediator;
+    private final BlockPersistenceHandler<BlockStreamServiceGrpcProto.Block> blockPersistenceHandler;
 
     /**
      * Constructor for the BlockStreamService class.
@@ -53,10 +57,12 @@ public class BlockStreamService implements GrpcService {
      * @param streamMediator the stream mediator
      */
     public BlockStreamService(final long timeoutThresholdMillis,
-                              final StreamMediator<BlockStreamServiceGrpcProto.Block, BlockStreamServiceGrpcProto.BlockResponse> streamMediator) {
+                              final StreamMediator<BlockStreamServiceGrpcProto.Block, BlockStreamServiceGrpcProto.BlockResponse> streamMediator,
+                              final BlockPersistenceHandler<BlockStreamServiceGrpcProto.Block> blockPersistenceHandler) {
 
         this.timeoutThresholdMillis = timeoutThresholdMillis;
         this.streamMediator = streamMediator;
+        this.blockPersistenceHandler = blockPersistenceHandler;
     }
 
     /**
@@ -133,11 +139,22 @@ private StreamObserver<BlockStreamServiceGrpcProto.BlockResponse> streamSource(f
         return streamObserver;
     }
 
-    private void getBlock(BlockStreamServiceGrpcProto.BlockRequest request, StreamObserver<BlockStreamServiceGrpcProto.Block> responseObserver) {
+    private void getBlock(BlockStreamServiceGrpcProto.Block block, StreamObserver<BlockStreamServiceGrpcProto.Block> responseObserver) {
         String message = "GET BLOCK RESPONSE! ";
         LOGGER.log(System.Logger.Level.INFO, "GetBlock request received");
-        BlockStreamServiceGrpcProto.Block response = BlockStreamServiceGrpcProto.Block.newBuilder().setValue(message).build();
-        complete(responseObserver, response);
+        Optional<BlockStreamServiceGrpcProto.Block> responseBlock = blockPersistenceHandler.read(block.getId());
+        if(responseBlock.isPresent()) {
+            LOGGER.log(System.Logger.Level.INFO, "SENDING BLOCK # " + block.getId());
+            complete(responseObserver, responseBlock.get()); // TODO: Should return int and not quoted string
+        } else {
+            LOGGER.log(System.Logger.Level.INFO, "DID NOT FIND YOUR BLOCK");
+            // TODO: Fix below. It could return gRPC equivalent of 404 NOT FOUND
+            responseObserver.onError(Status.NOT_FOUND
+                    .withDescription("DID NOT FIND YOUR BLOCK")
+                    .asRuntimeException()
+            );
+            // complete(responseObserver, BlockStreamServiceGrpcProto.Block.getDefaultInstance());
+        }
     }
 }
 

server/src/main/java/com/hedera/block/server/Server.java

@@ -63,8 +63,11 @@ public static void main(final String[] args) {
 
             // Initialize the block storage, cache, and service
             final BlockStorage<BlockStreamServiceGrpcProto.Block> blockStorage = new FileSystemBlockStorage(BLOCKNODE_STORAGE_ROOT_PATH_KEY, config);
+
+            // TODO: Make timeoutThresholdMillis configurable
             final BlockStreamService blockStreamService = new BlockStreamService(consumerTimeoutThreshold,
-                    new LiveStreamMediatorImpl(new WriteThroughCacheHandler(blockStorage)));
+                    new LiveStreamMediatorImpl(new WriteThroughCacheHandler(blockStorage)),
+                    new WriteThroughCacheHandler(blockStorage));
 
             // Start the web server
             WebServer.builder()