fix: removed bidi interface requirements after server streaming change

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

Author: mattp-swirldslabs

server/src/main/java/com/hedera/block/server/consumer/BlockItemEventHandler.java

@@ -17,8 +17,7 @@
 package com.hedera.block.server.consumer;
 
 import com.lmax.disruptor.EventHandler;
-import io.grpc.stub.StreamObserver;
 
-public interface BlockItemEventHandler<U, V> extends StreamObserver<V>, EventHandler<U> {
+public interface BlockItemEventHandler<U> extends EventHandler<U> {
     void awaitShutdown() throws InterruptedException;
 }

server/src/main/java/com/hedera/block/server/consumer/ConsumerBlockItemObserver.java

@@ -31,8 +31,7 @@
  * the downstream consumer via the notify method and manage the bidirectional stream to the consumer
  * via the onNext, onError, and onCompleted methods.
  */
-public class ConsumerBlockItemObserver
-        implements BlockItemEventHandler<ObjectEvent<BlockItem>, SubscribeStreamRequest> {
+public class ConsumerBlockItemObserver implements BlockItemEventHandler<ObjectEvent<BlockItem>> {
 
     private final System.Logger LOGGER = System.getLogger(getClass().getName());
 
@@ -89,55 +88,19 @@ public ConsumerBlockItemObserver(
     @Override
     public void onEvent(final ObjectEvent<BlockItem> event, final long l, final boolean b) {
 
-        // Refresh the producer liveness and pass the block to the observer.
-        producerLivenessMillis = producerLivenessClock.millis();
-
-        final BlockItem blockItem = event.get();
-        final SubscribeStreamResponse subscribeStreamResponse =
-                SubscribeStreamResponse.newBuilder().setBlockItem(blockItem).build();
-
-        subscribeStreamResponseObserver.onNext(subscribeStreamResponse);
-    }
-
-    /**
-     * The onNext() method is triggered by Helidon when a consumer sends a blockResponse via the
-     * bidirectional stream.
-     */
-    @Override
-    public void onNext(final SubscribeStreamRequest subscribeStreamRequest) {
-
-        // Check if the producer has timed out.  If so, unsubscribe the observer from the mediator.
         if (isThresholdExceeded(producerLivenessMillis)) {
-            LOGGER.log(
-                    System.Logger.Level.DEBUG,
-                    "Producer timeout threshold exceeded.  Unsubscribing observer.");
             streamMediator.unsubscribe(this);
-        }
-    }
+        } else {
 
-    /**
-     * The onError() method is triggered by Helidon when an error occurs on the bidirectional stream
-     * to the downstream consumer. Unsubscribe the observer from the mediator.
-     *
-     * @param t the error occurred on the stream
-     */
-    @Override
-    public void onError(final Throwable t) {
-        LOGGER.log(
-                System.Logger.Level.ERROR,
-                "Unexpected consumer stream communication failure: %s".formatted(t),
-                t);
-    }
+            // Refresh the producer liveness and pass the block to the observer.
+            producerLivenessMillis = producerLivenessClock.millis();
 
-    /**
-     * The onCompleted() method is triggered by Helidon when the bidirectional stream to the
-     * downstream consumer is completed. This implementation will then unsubscribe the observer from
-     * the mediator.
-     */
-    @Override
-    public void onCompleted() {
-        LOGGER.log(System.Logger.Level.DEBUG, "gRPC connection completed.");
-        streamMediator.unsubscribe(this);
+            final BlockItem blockItem = event.get();
+            final SubscribeStreamResponse subscribeStreamResponse =
+                    SubscribeStreamResponse.newBuilder().setBlockItem(blockItem).build();
+
+            subscribeStreamResponseObserver.onNext(subscribeStreamResponse);
+        }
     }
 
     private boolean isThresholdExceeded(long livenessMillis) {

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

@@ -46,7 +46,7 @@ public class LiveStreamMediatorImpl
     private final ExecutorService executor;
 
     private final Map<
-                    BlockItemEventHandler<ObjectEvent<BlockItem>, SubscribeStreamRequest>,
+                    BlockItemEventHandler<ObjectEvent<BlockItem>>,
                     BatchEventProcessor<ObjectEvent<BlockItem>>>
             subscribers = new HashMap<>();
 
@@ -80,8 +80,7 @@ public void publishEvent(BlockItem blockItem) {
     }
 
     @Override
-    public void subscribe(
-            final BlockItemEventHandler<ObjectEvent<BlockItem>, SubscribeStreamRequest> handler) {
+    public void subscribe(final BlockItemEventHandler<ObjectEvent<BlockItem>> handler) {
 
         // Initialize the batch event processor and set it on the ring buffer
         final var batchEventProcessor =
@@ -95,8 +94,7 @@ public void subscribe(
     }
 
     @Override
-    public void unsubscribe(
-            final BlockItemEventHandler<ObjectEvent<BlockItem>, SubscribeStreamRequest> handler) {
+    public void unsubscribe(final BlockItemEventHandler<ObjectEvent<BlockItem>> handler) {
         final var batchEventProcessor = subscribers.remove(handler);
 
         // Stop the processor

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

@@ -36,7 +36,7 @@ public interface StreamMediator<U, V> {
 
     void publishEvent(final BlockItem blockItem);
 
-    void subscribe(final BlockItemEventHandler<U, V> handler);
+    void subscribe(final BlockItemEventHandler<U> handler);
 
-    void unsubscribe(final BlockItemEventHandler<U, V> handler);
+    void unsubscribe(final BlockItemEventHandler<U> handler);
 }