removed import

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

Author: mattp-swirldslabs

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

@@ -18,8 +18,8 @@
 
 import com.google.protobuf.Descriptors;
 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.consumer.ConsumerBlockStreamObserver;
+import com.hedera.block.server.consumer.ConsumerBlockStreamObserverImpl;
 import com.hedera.block.server.persistence.BlockPersistenceHandler;
 import com.hedera.block.server.producer.ProducerBlockStreamObserver;
 import io.grpc.stub.StreamObserver;
@@ -116,14 +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.
-        LiveStreamObserver<BlockStreamServiceGrpcProto.BlockResponse> liveStreamObserver = new LiveStreamObserverImpl(
+        final ConsumerBlockStreamObserver<BlockStreamServiceGrpcProto.BlockResponse> consumerBlockStreamObserver = new ConsumerBlockStreamObserverImpl(
                 timeoutThresholdMillis,
                 Clock.systemDefaultZone(),
                 blockPersistenceHandler,
                 responseStreamObserver);
 
-        liveStreamObserver.emitBlocks();
-        return liveStreamObserver;
+        consumerBlockStreamObserver.emitBlocks();
+        return consumerBlockStreamObserver;
     }
 }
 

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

@@ -25,7 +25,6 @@
 import io.helidon.config.Config;
 import io.helidon.webserver.WebServer;
 import io.helidon.webserver.grpc.GrpcRouting;
-import io.helidon.webserver.http.HttpRouting;
 
 import java.io.IOException;
 import java.util.stream.Stream;
@@ -38,7 +37,7 @@
 public class Server {
 
     // Function stubs to satisfy the bidi routing param signatures.  The implementations are in the service class.
-    private static ServerCalls.BidiStreamingMethod<Stream<BlockStreamServiceGrpcProto.Block>, StreamObserver<BlockStreamServiceGrpcProto.Block>> clientBidiStreamingMethod;
+    private static ServerCalls.BidiStreamingMethod<Stream<BlockStreamServiceGrpcProto.Block>, StreamObserver<BlockStreamServiceGrpcProto.BlockResponse>> clientBidiStreamingMethod;
     private static ServerCalls.BidiStreamingMethod<Stream<BlockStreamServiceGrpcProto.BlockResponse>, StreamObserver<BlockStreamServiceGrpcProto.Block>> serverBidiStreamingMethod;
 
     private static final System.Logger LOGGER = System.getLogger(Server.class.getName());

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

@@ -26,7 +26,7 @@
  *
  * @param <V> the type of the StreamObserver
  */
-public interface LiveStreamObserver<V> extends StreamObserver<V> {
+public interface ConsumerBlockStreamObserver<V> extends StreamObserver<V> {
 
     /**
      */

server/src/main/java/com/hedera/block/server/consumer/LiveStreamObserverImpl.java renamed to server/src/main/java/com/hedera/block/server/consumer/ConsumerBlockStreamObserverImpl.java

@@ -28,7 +28,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 LiveStreamObserverImpl implements LiveStreamObserver<BlockStreamServiceGrpcProto.BlockResponse> {
+public class ConsumerBlockStreamObserverImpl implements ConsumerBlockStreamObserver<BlockStreamServiceGrpcProto.BlockResponse> {
 
     private final System.Logger LOGGER = System.getLogger(getClass().getName());
 
@@ -37,14 +37,14 @@ public class LiveStreamObserverImpl implements LiveStreamObserver<BlockStreamSer
 
     private final long timeoutThresholdMillis;
     private final AtomicLong consumerLivenessMillis;
-    private final AtomicLong counter;
+    private long counter;
 
     /**
      * Constructor for the LiveStreamObserverImpl class.
      *
      * @param responseStreamObserver the response stream observer
      */
-    public LiveStreamObserverImpl(
+    public ConsumerBlockStreamObserverImpl(
             final long timeoutThresholdMillis,
             final InstantSource consumerLivenessClock,
             final BlockPersistenceHandler<BlockStreamServiceGrpcProto.Block> blockPersistenceHandler,
@@ -55,8 +55,7 @@ public LiveStreamObserverImpl(
         this.responseStreamObserver = responseStreamObserver;
 
         this.consumerLivenessMillis = new AtomicLong(consumerLivenessClock.millis());
-//        this.counter = new AtomicLong(blockPersistenceHandler.getLastBlockId());
-        this.counter = new AtomicLong(1);
+        this.counter = blockPersistenceHandler.getLastBlockId();
     }
 
     /**
@@ -77,11 +76,13 @@ public void onNext(final BlockStreamServiceGrpcProto.BlockResponse blockResponse
 
     @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();
+        long latestBlockId = blockPersistenceHandler.getLastBlockId();
+        if (counter <= latestBlockId) {
+            for (BlockStreamServiceGrpcProto.Block block : blockPersistenceHandler.readRange(counter, latestBlockId)) {
+                LOGGER.log(System.Logger.Level.INFO, "Thread: {0}-{1}, Emitting block: {2}", Thread.currentThread().threadId(), Thread.currentThread().getName(), block.getId());
+                responseStreamObserver.onNext(block);
+                counter += 1;
+            }
         }
     }
 
@@ -108,7 +109,7 @@ public void onCompleted() {
     private boolean isThresholdExceeded() {
         final long currentTimeMillis = Clock.systemDefaultZone().millis();
         final long elapsedMillis = currentTimeMillis - consumerLivenessMillis.get();
-        if (elapsedMillis> timeoutThresholdMillis) {
+        if (elapsedMillis > timeoutThresholdMillis) {
             LOGGER.log(System.Logger.Level.INFO, "Elapsed milliseconds: " + elapsedMillis + ", timeout threshold: " + timeoutThresholdMillis);
             return true;
         }

server/src/main/java/com/hedera/block/server/persistence/WriteThroughCacheHandler.java

@@ -68,13 +68,9 @@ public Long getLastBlockId() {
     @Override
     public Queue<BlockStreamServiceGrpcProto.Block> readRange(final long startBlockId, final long endBlockId) {
         final Queue<BlockStreamServiceGrpcProto.Block> blocks = new LinkedList<>();
-
-        long count = startBlockId;
-        Optional<BlockStreamServiceGrpcProto.Block> blockOpt = read(count);
-        while (count <= endBlockId && blockOpt.isPresent()) {
-            final BlockStreamServiceGrpcProto.Block block = blockOpt.get();
-            blocks.add(block);
-            blockOpt = read(++count);
+        for (long count = startBlockId; count <= endBlockId; count++) {
+            final Optional<BlockStreamServiceGrpcProto.Block> blockOpt = read(count);
+            blockOpt.ifPresent(blocks::add);
         }
 
         return blocks;

server/src/main/java/com/hedera/block/server/persistence/storage/FileSystemBlockStorage.java

@@ -25,7 +25,6 @@
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
-import java.nio.file.StandardOpenOption;
 import java.util.Optional;
 import java.util.concurrent.atomic.AtomicLong;