first impl. busy wait holds thread so onNext is not called

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

Author: mattp-swirldslabs

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

@@ -20,7 +20,7 @@
 import com.hedera.block.protos.BlockStreamServiceGrpcProto;
 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.stub.StreamObserver;
 import io.helidon.webserver.grpc.GrpcService;
@@ -42,19 +42,18 @@ public class BlockStreamService implements GrpcService {
     private final System.Logger LOGGER = System.getLogger(getClass().getName());
 
     private final long timeoutThresholdMillis;
-    private final StreamMediator<BlockStreamServiceGrpcProto.Block, BlockStreamServiceGrpcProto.BlockResponse> streamMediator;
+    private final BlockPersistenceHandler<BlockStreamServiceGrpcProto.Block> blockPersistenceHandler;
 
     /**
      * Constructor for the BlockStreamService class.
      *
      * @param timeoutThresholdMillis the timeout threshold in milliseconds
-     * @param streamMediator the stream mediator
      */
     public BlockStreamService(final long timeoutThresholdMillis,
-                              final StreamMediator<BlockStreamServiceGrpcProto.Block, BlockStreamServiceGrpcProto.BlockResponse> streamMediator) {
+                              final BlockPersistenceHandler<BlockStreamServiceGrpcProto.Block> blockPersistenceHandler) {
 
         this.timeoutThresholdMillis = timeoutThresholdMillis;
-        this.streamMediator = streamMediator;
+        this.blockPersistenceHandler = blockPersistenceHandler;
     }
 
     /**
@@ -101,7 +100,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 ProducerBlockStreamObserver(blockPersistenceHandler, responseStreamObserver);
     }
 
     /**
@@ -117,17 +116,14 @@ 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 LiveStreamObserver<BlockStreamServiceGrpcProto.Block, BlockStreamServiceGrpcProto.BlockResponse> streamObserver = new LiveStreamObserverImpl(
+        LiveStreamObserver<BlockStreamServiceGrpcProto.BlockResponse> liveStreamObserver = new LiveStreamObserverImpl(
                 timeoutThresholdMillis,
                 Clock.systemDefaultZone(),
-                Clock.systemDefaultZone(),
-                streamMediator,
+                blockPersistenceHandler,
                 responseStreamObserver);
 
-        // Subscribe the observer to the mediator
-        streamMediator.subscribe(streamObserver);
-
-        return streamObserver;
+        liveStreamObserver.emitBlocks();
+        return liveStreamObserver;
     }
 }
 

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

@@ -17,7 +17,6 @@
 package com.hedera.block.server;
 
 import com.hedera.block.protos.BlockStreamServiceGrpcProto;
-import com.hedera.block.server.mediator.LiveStreamMediatorImpl;
 import com.hedera.block.server.persistence.WriteThroughCacheHandler;
 import com.hedera.block.server.persistence.storage.BlockStorage;
 import com.hedera.block.server.persistence.storage.FileSystemBlockStorage;
@@ -64,7 +63,7 @@ public static void main(final String[] args) {
 
             // TODO: Make timeoutThresholdMillis configurable
             final BlockStreamService blockStreamService = new BlockStreamService(1500,
-                    new LiveStreamMediatorImpl(new WriteThroughCacheHandler(blockStorage)));
+                    new WriteThroughCacheHandler(blockStorage));
 
             // Start the web server
             WebServer.builder()

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

@@ -24,15 +24,11 @@
  * the implementation of this interface can receive and process inbound messages with different types from
  * the producer and response messages from the consumer.
  *
- * @param <U> the type of the block
  * @param <V> the type of the StreamObserver
  */
-public interface LiveStreamObserver<U, V> extends StreamObserver<V> {
+public interface LiveStreamObserver<V> extends StreamObserver<V> {
 
     /**
-     * Pass the block to the observer.
-     *
-     * @param block - the block to be passed to the observer
      */
-    void notify(final U block);
+    void emitBlocks();
 }

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

@@ -17,73 +17,46 @@
 package com.hedera.block.server.consumer;
 
 import com.hedera.block.protos.BlockStreamServiceGrpcProto;
-import com.hedera.block.server.mediator.StreamMediator;
+import com.hedera.block.server.persistence.BlockPersistenceHandler;
 import io.grpc.stub.StreamObserver;
 
 import java.time.Clock;
 import java.time.InstantSource;
+import java.util.concurrent.atomic.AtomicLong;
 
 /**
  * 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 LiveStreamObserverImpl implements LiveStreamObserver<BlockStreamServiceGrpcProto.Block, BlockStreamServiceGrpcProto.BlockResponse> {
+public class LiveStreamObserverImpl implements LiveStreamObserver<BlockStreamServiceGrpcProto.BlockResponse> {
 
     private final System.Logger LOGGER = System.getLogger(getClass().getName());
 
-    private final StreamMediator<BlockStreamServiceGrpcProto.Block, BlockStreamServiceGrpcProto.BlockResponse> mediator;
+    private final BlockPersistenceHandler<BlockStreamServiceGrpcProto.Block> blockPersistenceHandler;
     private final StreamObserver<BlockStreamServiceGrpcProto.Block> responseStreamObserver;
 
     private final long timeoutThresholdMillis;
-
-    private final InstantSource producerLivenessClock;
-    private long producerLivenessMillis;
-
-    private final InstantSource consumerLivenessClock;
-    private long consumerLivenessMillis;
+    private final AtomicLong consumerLivenessMillis;
+    private final AtomicLong counter;
 
     /**
      * Constructor for the LiveStreamObserverImpl class.
      *
-     * @param mediator the mediator
      * @param responseStreamObserver the response stream observer
      */
     public LiveStreamObserverImpl(
             final long timeoutThresholdMillis,
-            final InstantSource producerLivenessClock,
             final InstantSource consumerLivenessClock,
-            final StreamMediator<BlockStreamServiceGrpcProto.Block, BlockStreamServiceGrpcProto.BlockResponse> mediator,
+            final BlockPersistenceHandler<BlockStreamServiceGrpcProto.Block> blockPersistenceHandler,
             final StreamObserver<BlockStreamServiceGrpcProto.Block> responseStreamObserver) {
 
         this.timeoutThresholdMillis = timeoutThresholdMillis;
-        this.producerLivenessClock = producerLivenessClock;
-        this.consumerLivenessClock = consumerLivenessClock;
-        this.mediator = mediator;
+        this.blockPersistenceHandler = blockPersistenceHandler;
         this.responseStreamObserver = responseStreamObserver;
 
-        this.producerLivenessMillis = producerLivenessClock.millis();
-        this.consumerLivenessMillis = consumerLivenessClock.millis();
-    }
-
-    /**
-     * Pass the block to the observer provided by Helidon
-     *
-     * @param block the block to be passed to the observer
-     */
-    @Override
-    public void notify(final BlockStreamServiceGrpcProto.Block block) {
-
-        // Check if the consumer has timed out.  If so, unsubscribe the observer from the mediator.
-        if (consumerLivenessClock.millis() - consumerLivenessMillis > timeoutThresholdMillis) {
-            if (mediator.isSubscribed(this)) {
-                LOGGER.log(System.Logger.Level.DEBUG, "Consumer timeout threshold exceeded.  Unsubscribing observer.");
-                mediator.unsubscribe(this);
-            }
-        } else {
-            // Refresh the producer liveness and pass the block to the observer.
-            producerLivenessMillis = producerLivenessClock.millis();
-            responseStreamObserver.onNext(block);
-        }
+        this.consumerLivenessMillis = new AtomicLong(consumerLivenessClock.millis());
+//        this.counter = new AtomicLong(blockPersistenceHandler.getLastBlockId());
+        this.counter = new AtomicLong(1);
     }
 
     /**
@@ -94,12 +67,21 @@ public void notify(final BlockStreamServiceGrpcProto.Block block) {
     @Override
     public void onNext(final BlockStreamServiceGrpcProto.BlockResponse blockResponse) {
 
-        if (producerLivenessClock.millis() - producerLivenessMillis > timeoutThresholdMillis) {
-            LOGGER.log(System.Logger.Level.DEBUG, "Producer timeout threshold exceeded.  Unsubscribing observer.");
-            mediator.unsubscribe(this);
-        } else {
-            LOGGER.log(System.Logger.Level.DEBUG, "Received response block " + blockResponse);
-            consumerLivenessMillis = consumerLivenessClock.millis();
+        // update the consumer liveness clock
+        consumerLivenessMillis.set(Clock.systemDefaultZone().millis());
+
+        if (!isThresholdExceeded()) {
+            emitBlocks();
+        }
+    }
+
+    @Override
+    public void emitBlocks() {
+        final long latestBlockId = counter.get();
+        for (BlockStreamServiceGrpcProto.Block block : blockPersistenceHandler.readRange(latestBlockId, latestBlockId + 1)) {
+            LOGGER.log(System.Logger.Level.INFO, "Thread: {0}-{1}, Emitting block: {2}", Thread.currentThread().threadId(), Thread.currentThread().getName(), block.getId());
+            responseStreamObserver.onNext(block);
+            counter.incrementAndGet();
         }
     }
 
@@ -112,7 +94,6 @@ public void onNext(final BlockStreamServiceGrpcProto.BlockResponse blockResponse
     @Override
     public void onError(final Throwable t) {
         LOGGER.log(System.Logger.Level.ERROR, "Unexpected consumer stream communication failure: %s".formatted(t), t);
-        mediator.unsubscribe(this);
     }
 
     /**
@@ -121,8 +102,17 @@ public void onError(final Throwable t) {
      */
     @Override
     public void onCompleted() {
-        LOGGER.log(System.Logger.Level.DEBUG, "gRPC connection completed.  Unsubscribing observer.");
-        mediator.unsubscribe(this);
-        LOGGER.log(System.Logger.Level.DEBUG, "Unsubscribed observer.");
+        LOGGER.log(System.Logger.Level.INFO, "gRPC connection completed.");
+    }
+
+    private boolean isThresholdExceeded() {
+        final long currentTimeMillis = Clock.systemDefaultZone().millis();
+        final long elapsedMillis = currentTimeMillis - consumerLivenessMillis.get();
+        if (elapsedMillis> timeoutThresholdMillis) {
+            LOGGER.log(System.Logger.Level.INFO, "Elapsed milliseconds: " + elapsedMillis + ", timeout threshold: " + timeoutThresholdMillis);
+            return true;
+        }
+
+        return false;
     }
 }