Removed BlockPersistenceHandler interface and associated materials

Signed-off-by: Matt Peterson <matt.peterson@swirldslabs.com>

Author: mattp-swirldslabs

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

@@ -23,7 +23,7 @@
 import com.hedera.block.server.consumer.ConsumerBlockItemObserver;
 import com.hedera.block.server.data.ObjectEvent;
 import com.hedera.block.server.mediator.StreamMediator;
-import com.hedera.block.server.persistence.BlockPersistenceHandler;
+import com.hedera.block.server.persistence.storage.BlockReader;
 import com.hedera.block.server.producer.ItemAckBuilder;
 import com.hedera.block.server.producer.ProducerBlockItemObserver;
 import io.grpc.stub.StreamObserver;
@@ -50,18 +50,18 @@ public class BlockStreamService implements GrpcService {
     private final ItemAckBuilder itemAckBuilder;
     private final StreamMediator<BlockItem, ObjectEvent<SubscribeStreamResponse>> streamMediator;
     private final ServiceStatus serviceStatus;
-    private final BlockPersistenceHandler<BlockItem, Block> blockPersistenceHandler;
+    private final BlockReader<Block> blockReader;
 
     BlockStreamService(
             final long timeoutThresholdMillis,
             final ItemAckBuilder itemAckBuilder,
             final StreamMediator<BlockItem, ObjectEvent<SubscribeStreamResponse>> streamMediator,
-            final BlockPersistenceHandler<BlockItem, Block> blockPersistenceHandler,
+            final BlockReader<Block> blockReader,
             final ServiceStatus serviceStatus) {
         this.timeoutThresholdMillis = timeoutThresholdMillis;
         this.itemAckBuilder = itemAckBuilder;
         this.streamMediator = streamMediator;
-        this.blockPersistenceHandler = blockPersistenceHandler;
+        this.blockReader = blockReader;
         this.serviceStatus = serviceStatus;
     }
 
@@ -144,7 +144,7 @@ void singleBlock(
         if (serviceStatus.isRunning()) {
             final long blockNumber = singleBlockRequest.getBlockNumber();
             try {
-                final Optional<Block> blockOpt = blockPersistenceHandler.retrieve(blockNumber);
+                final Optional<Block> blockOpt = blockReader.read(blockNumber);
                 if (blockOpt.isPresent()) {
                     LOGGER.log(
                             System.Logger.Level.INFO,

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

@@ -22,10 +22,10 @@
 import com.hedera.block.server.data.ObjectEvent;
 import com.hedera.block.server.mediator.LiveStreamMediatorImpl;
 import com.hedera.block.server.mediator.StreamMediator;
-import com.hedera.block.server.persistence.BlockPersistenceHandler;
-import com.hedera.block.server.persistence.FileSystemPersistenceHandler;
 import com.hedera.block.server.persistence.storage.BlockAsDirReader;
 import com.hedera.block.server.persistence.storage.BlockAsDirWriter;
+import com.hedera.block.server.persistence.storage.BlockReader;
+import com.hedera.block.server.persistence.storage.BlockWriter;
 import com.hedera.block.server.producer.ItemAckBuilder;
 import io.grpc.stub.ServerCalls;
 import io.grpc.stub.StreamObserver;
@@ -70,15 +70,16 @@ public static void main(final String[] args) {
 
         try {
             final ServiceStatus serviceStatus = new ServiceStatusImpl();
-            final BlockPersistenceHandler<BlockItem, Block> blockPersistenceHandler =
-                    new FileSystemPersistenceHandler(
-                            new BlockAsDirReader(BLOCKNODE_STORAGE_ROOT_PATH_KEY, config),
-                            new BlockAsDirWriter(BLOCKNODE_STORAGE_ROOT_PATH_KEY, config));
             final StreamMediator<BlockItem, ObjectEvent<SubscribeStreamResponse>> streamMediator =
-                    buildStreamMediator(serviceStatus, blockPersistenceHandler);
+                    buildStreamMediator(
+                            serviceStatus,
+                            new BlockAsDirWriter(BLOCKNODE_STORAGE_ROOT_PATH_KEY, config));
             final BlockStreamService blockStreamService =
                     buildBlockStreamService(
-                            config, streamMediator, blockPersistenceHandler, serviceStatus);
+                            config,
+                            streamMediator,
+                            new BlockAsDirReader(BLOCKNODE_STORAGE_ROOT_PATH_KEY, config),
+                            serviceStatus);
             final GrpcRouting.Builder grpcRouting = buildGrpcRouting(blockStreamService);
 
             // Build the web server
@@ -97,11 +98,9 @@ public static void main(final String[] args) {
 
     private static StreamMediator<BlockItem, ObjectEvent<SubscribeStreamResponse>>
             buildStreamMediator(
-                    final ServiceStatus serviceStatus,
-                    final BlockPersistenceHandler<BlockItem, Block> blockPersistenceHandler)
+                    final ServiceStatus serviceStatus, final BlockWriter<BlockItem> blockWriter)
                     throws IOException {
-        return new LiveStreamMediatorImpl(
-                new ConcurrentHashMap<>(32), blockPersistenceHandler, serviceStatus);
+        return new LiveStreamMediatorImpl(new ConcurrentHashMap<>(32), blockWriter, serviceStatus);
     }
 
     private static GrpcRouting.Builder buildGrpcRouting(
@@ -129,7 +128,7 @@ private static GrpcRouting.Builder buildGrpcRouting(
     private static BlockStreamService buildBlockStreamService(
             final Config config,
             final StreamMediator<BlockItem, ObjectEvent<SubscribeStreamResponse>> streamMediator,
-            final BlockPersistenceHandler<BlockItem, Block> blockPersistenceHandler,
+            final BlockReader<Block> blockReader,
             final ServiceStatus serviceStatus) {
 
         // Get Timeout threshold from configuration
@@ -140,7 +139,7 @@ private static BlockStreamService buildBlockStreamService(
                 consumerTimeoutThreshold,
                 new ItemAckBuilder(),
                 streamMediator,
-                blockPersistenceHandler,
+                blockReader,
                 serviceStatus);
     }
 }

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

@@ -18,12 +18,32 @@
 
 import io.helidon.webserver.WebServer;
 
+/**
+ * The ServiceStatus interface defines the contract for checking the status of the service and
+ * shutting down the web server.
+ */
 public interface ServiceStatus {
+
+    /**
+     * Checks if the service is running.
+     *
+     * @return true if the service is running, false otherwise
+     */
     boolean isRunning();
 
+    /**
+     * Sets the running status of the service.
+     *
+     * @param running true if the service is running, false otherwise
+     */
     void setRunning(final boolean running);
 
+    /** Sets the web server instance. */
     void setWebServer(final WebServer webServer);
 
-    void stopService();
+    /**
+     * Stops the service and web server. This method is called to shut down the service and the web
+     * server in the event of an error or when the service needs to restart.
+     */
+    void stopWebServer();
 }

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

@@ -35,7 +35,7 @@ public void setWebServer(final WebServer webServer) {
         this.webServer = webServer;
     }
 
-    public void stopService() {
+    public void stopWebServer() {
         isRunning.set(false);
         webServer.stop();
     }

server/src/main/java/com/hedera/block/server/mediator/LiveStreamMediatorImpl.java

@@ -16,12 +16,13 @@
 
 package com.hedera.block.server.mediator;
 
-import static com.hedera.block.protos.BlockStreamService.*;
+import static com.hedera.block.protos.BlockStreamService.BlockItem;
+import static com.hedera.block.protos.BlockStreamService.SubscribeStreamResponse;
 
 import com.hedera.block.server.ServiceStatus;
 import com.hedera.block.server.ServiceStatusImpl;
 import com.hedera.block.server.data.ObjectEvent;
-import com.hedera.block.server.persistence.BlockPersistenceHandler;
+import com.hedera.block.server.persistence.storage.BlockWriter;
 import com.lmax.disruptor.BatchEventProcessor;
 import com.lmax.disruptor.BatchEventProcessorBuilder;
 import com.lmax.disruptor.EventHandler;
@@ -37,7 +38,7 @@
 /**
  * LiveStreamMediatorImpl is the implementation of the StreamMediator interface. It is responsible
  * for managing the subscribe and unsubscribe operations of downstream consumers. It also proxies
- * live blocks to the subscribers as they arrive and persists the blocks to the block persistence
+ * block items to the subscribers as they arrive and persists the blocks to a block persistence
  * store.
  */
 public class LiveStreamMediatorImpl
@@ -53,24 +54,27 @@ public class LiveStreamMediatorImpl
                     BatchEventProcessor<ObjectEvent<SubscribeStreamResponse>>>
             subscribers;
 
-    private final BlockPersistenceHandler<BlockItem, Block> blockPersistenceHandler;
+    private final BlockWriter<BlockItem> blockWriter;
     private final ServiceStatus serviceStatus;
 
     /**
-     * Constructor for the LiveStreamMediatorImpl class.
+     * Constructs a new LiveStreamMediatorImpl instance with the given subscribers, block
+     * persistence handler, and service status.
      *
-     * @param blockPersistenceHandler the block persistence handler
+     * @param subscribers the map of subscribers to their corresponding representation batch event
+     *     processor in the LMAX Disruptor data structure.
+     * @param serviceStatus the service status singleton instance
      */
     public LiveStreamMediatorImpl(
             final Map<
                             EventHandler<ObjectEvent<SubscribeStreamResponse>>,
                             BatchEventProcessor<ObjectEvent<SubscribeStreamResponse>>>
                     subscribers,
-            final BlockPersistenceHandler<BlockItem, Block> blockPersistenceHandler,
+            final BlockWriter<BlockItem> blockWriter,
             final ServiceStatus serviceStatus) {
 
         this.subscribers = subscribers;
-        this.blockPersistenceHandler = blockPersistenceHandler;
+        this.blockWriter = blockWriter;
 
         // Initialize and start the disruptor
         final Disruptor<ObjectEvent<SubscribeStreamResponse>> disruptor =
@@ -80,15 +84,13 @@ public LiveStreamMediatorImpl(
         this.serviceStatus = serviceStatus;
     }
 
-    public LiveStreamMediatorImpl(
-            final BlockPersistenceHandler<BlockItem, Block> blockPersistenceHandler) {
-        this(new ConcurrentHashMap<>(), blockPersistenceHandler, new ServiceStatusImpl());
+    public LiveStreamMediatorImpl(final BlockWriter<BlockItem> blockWriter) {
+        this(new ConcurrentHashMap<>(), blockWriter, new ServiceStatusImpl());
     }
 
     public LiveStreamMediatorImpl(
-            final BlockPersistenceHandler<BlockItem, Block> blockPersistenceHandler,
-            final ServiceStatus serviceStatus) {
-        this(new ConcurrentHashMap<>(), blockPersistenceHandler, serviceStatus);
+            final BlockWriter<BlockItem> blockWriter, final ServiceStatus serviceStatus) {
+        this(new ConcurrentHashMap<>(), blockWriter, serviceStatus);
     }
 
     @Override
@@ -104,7 +106,7 @@ public void publish(final BlockItem blockItem) throws IOException {
 
             try {
                 // Persist the BlockItem
-                blockPersistenceHandler.persist(blockItem);
+                blockWriter.write(blockItem);
             } catch (IOException e) {
                 // Disable BlockItem publication for upstream producers
                 serviceStatus.setRunning(false);

server/src/main/java/com/hedera/block/server/mediator/EventPublisher.java renamed to server/src/main/java/com/hedera/block/server/mediator/Publisher.java

@@ -18,6 +18,20 @@
 
 import java.io.IOException;
 
-public interface EventPublisher<U> {
-    void publish(final U blockItem) throws IOException;
+/**
+ * The Publisher interface defines the contract for publishing items emitted by the producer to
+ * downstream subscribers.
+ *
+ * @param <U> the type of the item to publish
+ */
+public interface Publisher<U> {
+
+    /**
+     * Publishes the given item to the downstream subscribers.
+     *
+     * @param item the item emitted by an upstream producer to publish to downstream subscribers.
+     * @throws IOException thrown if an I/O error occurs while publishing the item to the
+     *     subscribers.
+     */
+    void publish(final U item) throws IOException;
 }

server/src/main/java/com/hedera/block/server/mediator/StreamMediator.java

@@ -17,9 +17,12 @@
 package com.hedera.block.server.mediator;
 
 /**
- * The StreamMediator interface represents a bridge between a bidirectional stream of items from a
- * producer (e.g. a Consensus Node) and N consumers each requesting a gRPC server stream of items.
+ * The StreamMediator marker interface defines the combination of Publisher and SubscriptionHandler
+ * contracts. It defines multiple views of the underlying implementation, allowing producers to
+ * publish items while the service and downstream subscribers can manage which consumers are
+ * subscribed to the stream of events.
  *
- * @param <U> The type of items sent by the upstream producer.
+ * @param <U> the type of the item to publish
+ * @param <V> the type of the event
  */
-public interface StreamMediator<U, V> extends EventPublisher<U>, SubscriptionHandler<V> {}
+public interface StreamMediator<U, V> extends Publisher<U>, SubscriptionHandler<V> {}

server/src/main/java/com/hedera/block/server/mediator/SubscriptionHandler.java

@@ -19,17 +19,32 @@
 import com.lmax.disruptor.EventHandler;
 
 /**
- * The SubscriptionHandler interface represents a bridge between a producer and N consumers each
- * requesting a gRPC server stream of items. The SubscriptionHandler manages adding and removing
- * consumers dynamically to receive items as they arrive from upstream.
+ * The SubscriptionHandler interface defines the contract for subscribing and unsubscribing
+ * downstream consumers to the stream of events.
  *
- * @param <V> The type of the response published to the downstream consumers.
+ * @param <V> the type of the subscription events
  */
 public interface SubscriptionHandler<V> {
 
+    /**
+     * Subscribes the given handler to the stream of items.
+     *
+     * @param handler the handler to subscribe
+     */
     void subscribe(final EventHandler<V> handler);
 
+    /**
+     * Unsubscribes the given handler from the stream of events.
+     *
+     * @param handler the handler to unsubscribe
+     */
     void unsubscribe(final EventHandler<V> handler);
 
+    /**
+     * Checks if the given handler is subscribed to the stream of events.
+     *
+     * @param handler the handler to check
+     * @return true if the handler is subscribed, false otherwise
+     */
     boolean isSubscribed(final EventHandler<V> handler);
 }