wip: refactored class names

Signed-off-by: Matt Peterson <[email protected]>

Author: mattp-swirldslabs

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

@@ -18,20 +18,14 @@
 
 import com.google.protobuf.Descriptors;
 import com.hedera.block.protos.BlockStreamServiceGrpcProto;
-import com.hedera.block.server.consumer.ConsumerBlockStreamObserver;
+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.producer.ProducerBlockStreamObserver;
-import com.lmax.disruptor.BatchEventProcessorBuilder;
-import com.lmax.disruptor.RingBuffer;
-import com.lmax.disruptor.dsl.Disruptor;
-import com.lmax.disruptor.util.DaemonThreadFactory;
+import com.hedera.block.server.producer.ProducerBlockItemObserver;
 import io.grpc.stub.StreamObserver;
 import io.helidon.webserver.grpc.GrpcService;
 
 import java.time.Clock;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
 
 import static com.hedera.block.server.Constants.*;
 
@@ -105,7 +99,7 @@ private StreamObserver<BlockStreamServiceGrpcProto.Block> streamSink(
             final StreamObserver<BlockStreamServiceGrpcProto.BlockResponse> responseStreamObserver) {
         LOGGER.log(System.Logger.Level.DEBUG, "Executing bidirectional streamSink method");
 
-        return new ProducerBlockStreamObserver(streamMediator, responseStreamObserver);
+        return new ProducerBlockItemObserver(streamMediator, responseStreamObserver);
     }
 
     /**
@@ -121,7 +115,7 @@ private StreamObserver<BlockStreamServiceGrpcProto.BlockResponse> streamSource(f
         LOGGER.log(System.Logger.Level.DEBUG, "Executing bidirectional streamSource method");
 
         // Return a custom StreamObserver to handle streaming blocks from the producer.
-        final var streamObserver = new ConsumerBlockStreamObserver(
+        final var streamObserver = new ConsumerBlockItemObserver(
                 timeoutThresholdMillis,
                 Clock.systemDefaultZone(),
                 Clock.systemDefaultZone(),

server/src/main/java/com/hedera/block/server/consumer/BlockEventHandler.java renamed to server/src/main/java/com/hedera/block/server/consumer/BlockItemEventHandler.java

@@ -18,6 +18,6 @@
 import com.lmax.disruptor.EventHandler;
 import io.grpc.stub.StreamObserver;
 
-public interface BlockEventHandler<U, V> extends StreamObserver<V>, EventHandler<U> {
+public interface BlockItemEventHandler<U, V> extends StreamObserver<V>, EventHandler<U> {
     void awaitShutdown() throws InterruptedException;
 }

server/src/main/java/com/hedera/block/server/consumer/ConsumerBlockStreamObserver.java renamed to server/src/main/java/com/hedera/block/server/consumer/ConsumerBlockItemObserver.java

@@ -29,7 +29,7 @@
  * The LiveStreamObserverImpl class implements the LiveStreamObserver interface to pass blocks to the downstream consumer
  * via the notify method and manage the bidirectional stream to the consumer via the onNext, onError, and onCompleted methods.
  */
-public class ConsumerBlockStreamObserver implements BlockEventHandler<ObjectEvent<BlockStreamServiceGrpcProto.Block>, BlockStreamServiceGrpcProto.BlockResponse> {
+public class ConsumerBlockItemObserver implements BlockItemEventHandler<ObjectEvent<BlockStreamServiceGrpcProto.Block>, BlockStreamServiceGrpcProto.BlockResponse> {
 
     private final System.Logger LOGGER = System.getLogger(getClass().getName());
 
@@ -52,7 +52,7 @@ public class ConsumerBlockStreamObserver implements BlockEventHandler<ObjectEven
      *
      * @param responseStreamObserver the response stream observer
      */
-    public ConsumerBlockStreamObserver(
+    public ConsumerBlockItemObserver(
             final long timeoutThresholdMillis,
             final InstantSource producerLivenessClock,
             final InstantSource consumerLivenessClock,

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

@@ -17,7 +17,7 @@
 package com.hedera.block.server.mediator;
 
 import com.hedera.block.protos.BlockStreamServiceGrpcProto;
-import com.hedera.block.server.consumer.BlockEventHandler;
+import com.hedera.block.server.consumer.BlockItemEventHandler;
 import com.hedera.block.server.data.ObjectEvent;
 import com.hedera.block.server.persistence.BlockPersistenceHandler;
 import com.lmax.disruptor.BatchEventProcessor;
@@ -43,7 +43,7 @@ public class LiveStreamMediatorImpl implements StreamMediator<ObjectEvent<BlockS
     private final RingBuffer<ObjectEvent<BlockStreamServiceGrpcProto.Block>> ringBuffer;
     private final ExecutorService executor;
 
-    private final Map<BlockEventHandler<ObjectEvent<BlockStreamServiceGrpcProto.Block>, BlockStreamServiceGrpcProto.BlockResponse>,
+    private final Map<BlockItemEventHandler<ObjectEvent<BlockStreamServiceGrpcProto.Block>, BlockStreamServiceGrpcProto.BlockResponse>,
         BatchEventProcessor<ObjectEvent<BlockStreamServiceGrpcProto.Block>>> subscribers = new HashMap<>();
 
     private final BlockPersistenceHandler<BlockStreamServiceGrpcProto.Block> blockPersistenceHandler;
@@ -74,7 +74,7 @@ public void publishEvent(BlockStreamServiceGrpcProto.Block block) {
     }
 
     @Override
-    public void subscribe(final BlockEventHandler<ObjectEvent<BlockStreamServiceGrpcProto.Block>, BlockStreamServiceGrpcProto.BlockResponse> handler) {
+    public void subscribe(final BlockItemEventHandler<ObjectEvent<BlockStreamServiceGrpcProto.Block>, BlockStreamServiceGrpcProto.BlockResponse> handler) {
 
         // Initialize the batch event processor and set it on the ring buffer
         final var batchEventProcessor = new BatchEventProcessorBuilder()
@@ -87,7 +87,7 @@ public void subscribe(final BlockEventHandler<ObjectEvent<BlockStreamServiceGrpc
     }
 
     @Override
-    public void unsubscribe(final BlockEventHandler<ObjectEvent<BlockStreamServiceGrpcProto.Block>, BlockStreamServiceGrpcProto.BlockResponse> handler) {
+    public void unsubscribe(final BlockItemEventHandler<ObjectEvent<BlockStreamServiceGrpcProto.Block>, BlockStreamServiceGrpcProto.BlockResponse> handler) {
         final var batchEventProcessor = subscribers.remove(handler);
 
         // Stop the processor

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

@@ -17,9 +17,7 @@
 package com.hedera.block.server.mediator;
 
 import com.hedera.block.protos.BlockStreamServiceGrpcProto;
-import com.hedera.block.server.consumer.BlockEventHandler;
-import com.hedera.block.server.data.ObjectEvent;
-import com.lmax.disruptor.EventTranslator;
+import com.hedera.block.server.consumer.BlockItemEventHandler;
 
 /**
  * The StreamMediator interface represents a one-to-many bridge between a bidirectional stream of blocks from a
@@ -36,7 +34,7 @@ public interface StreamMediator<U, V> {
 
     void publishEvent(final BlockStreamServiceGrpcProto.Block block);
 
-    void subscribe(final BlockEventHandler<U, V> handler);
+    void subscribe(final BlockItemEventHandler<U, V> handler);
 
-    void unsubscribe(final BlockEventHandler<U, V> handler);
+    void unsubscribe(final BlockItemEventHandler<U, V> handler);
 }

server/src/main/java/com/hedera/block/server/producer/ProducerBlockStreamObserver.java renamed to server/src/main/java/com/hedera/block/server/producer/ProducerBlockItemObserver.java

@@ -19,7 +19,6 @@
 import com.hedera.block.protos.BlockStreamServiceGrpcProto;
 import com.hedera.block.server.data.ObjectEvent;
 import com.hedera.block.server.mediator.StreamMediator;
-import com.lmax.disruptor.RingBuffer;
 import io.grpc.stub.StreamObserver;
 
 /**
@@ -28,7 +27,7 @@
  * with the connection to the upstream producer (e.g. blocks streamed from the Consensus Node to
  * the server).
  */
-public class ProducerBlockStreamObserver implements StreamObserver<BlockStreamServiceGrpcProto.Block> {
+public class ProducerBlockItemObserver implements StreamObserver<BlockStreamServiceGrpcProto.Block> {
 
     private final System.Logger LOGGER = System.getLogger(getClass().getName());
 
@@ -42,8 +41,8 @@ public class ProducerBlockStreamObserver implements StreamObserver<BlockStreamSe
      *
      * @param responseStreamObserver the response stream observer
      */
-    public ProducerBlockStreamObserver(final StreamMediator<ObjectEvent<BlockStreamServiceGrpcProto.Block>, BlockStreamServiceGrpcProto.BlockResponse> streamMediator,
-                                       final StreamObserver<BlockStreamServiceGrpcProto.BlockResponse> responseStreamObserver) {
+    public ProducerBlockItemObserver(final StreamMediator<ObjectEvent<BlockStreamServiceGrpcProto.Block>, BlockStreamServiceGrpcProto.BlockResponse> streamMediator,
+                                     final StreamObserver<BlockStreamServiceGrpcProto.BlockResponse> responseStreamObserver) {
         this.streamMediator = streamMediator;
         this.responseStreamObserver = responseStreamObserver;
     }