push to streaming might be fixed

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

Author: mattp-swirldslabs

consumer.sh

@@ -6,24 +6,29 @@ echo "Starting consumer..."
 GRPC_SERVER="localhost:8080"
 GRPC_METHOD="BlockStreamGrpc/StreamSource"
 
-# Function to handle received messages and send replies
-handle_streaming() {
-  while read -r line; do
-    id=$(echo "$line" | jq -r '.id')
-    if [[ $id != "null" ]]; then
-      echo "{\"id\": $id}"
-    fi
-  done
-}
-
-# Trap SIGINT to clean up
 trap "echo 'Received SIGINT, stopping...'; exit 0" SIGINT
 
 payload="{\"id\": 0}"
 
+(
+  iter=$1
+  while true; do
+    echo "{\"id\": $iter}"
+
+    if [ $iter -eq $2 ]; then
+      exit 0
+    fi
+
+    ((iter++))
+
+    sleep 0.25
+
+  done
+) | grpcurl -plaintext -proto ./protos/src/main/protobuf/blockstream.proto -d @ $GRPC_SERVER $GRPC_METHOD
+
 #grpcurl -vv -plaintext -proto ./protos/src/main/protobuf/blockstream.proto -d "$payload" localhost:8080 BlockStreamGrpc/StreamSource
 
 # Initiate the bidirectional streaming connection
-grpcurl -vv -plaintext -proto ./protos/src/main/protobuf/blockstream.proto -d "$payload" $GRPC_SERVER $GRPC_METHOD
-#{} # Initial empty request if needed
-EOF | handle_streaming | grpcurl -plaintext -d @ $GRPC_SERVER $GRPC_METHOD --format json
+#grpcurl -vv -plaintext -proto ./protos/src/main/protobuf/blockstream.proto -d "$payload" $GRPC_SERVER $GRPC_METHOD | handle_streaming | grpcurl -plaintext -d @ $GRPC_SERVER $GRPC_METHOD
+#grpcurl -plaintext -proto ./protos/src/main/protobuf/blockstream.proto -d @ $GRPC_SERVER $GRPC_METHOD
+#handle_streaming | grpcurl -plaintext -d @ $GRPC_SERVER $GRPC_METHOD --format json

producer.sh

@@ -51,7 +51,7 @@ trap cleanup SIGINT
 
     ((iter++))
 
-    sleep 1
+    sleep 0.25
   done
 ) | grpcurl -vv -plaintext -proto ./protos/src/main/protobuf/blockstream.proto -d @ $GRPC_SERVER $GRPC_METHOD &
 

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

@@ -25,7 +25,6 @@
 import com.hedera.block.server.observe.mediate.StreamMediatorImpl;
 import com.hedera.block.server.observe.subscribe.HistoricStreamObserver;
 import com.hedera.block.server.observe.subscribe.LiveStreamObserver;
-import com.hedera.block.server.persistence.BlockPersistenceHandler;
 import com.hedera.block.server.persistence.WriteThroughCacheHandler;
 import com.hedera.block.server.persistence.cache.BlockCache;
 import com.hedera.block.server.persistence.cache.LRUCache;
@@ -43,8 +42,7 @@
 public class BlockStreamService implements GrpcService {
 
     private final Logger logger = Logger.getLogger(getClass().getName());
-    private final BlockPersistenceHandler<BlockStreamServiceGrpcProto.Block> blockPersistenceHandler;
-    private final StreamMediator<BlockStreamServiceGrpcProto.Block> streamMediator;
+    private final StreamMediator<BlockStreamServiceGrpcProto.BlockResponse, BlockStreamServiceGrpcProto.Block> streamMediator;
 
     public BlockStreamService() {
         try {
@@ -54,8 +52,7 @@ public BlockStreamService() {
 
             BlockStorage<BlockStreamServiceGrpcProto.Block> blockStorage = new FileSystemBlockStorage(BLOCKNODE_STORAGE_ROOT_PATH_KEY, config);
             BlockCache<BlockStreamServiceGrpcProto.Block> blockCache = new LRUCache(10L);
-            this.blockPersistenceHandler = new WriteThroughCacheHandler(blockStorage, blockCache);
-            this.streamMediator = new StreamMediatorImpl(blockPersistenceHandler);
+            this.streamMediator = new StreamMediatorImpl(new WriteThroughCacheHandler(blockStorage, blockCache));
 
         } catch (IOException e) {
             throw new RuntimeException(e);
@@ -85,12 +82,11 @@ private StreamObserver<BlockStreamServiceGrpcProto.Block> streamSink(StreamObser
     }
 
     private StreamObserver<BlockStreamServiceGrpcProto.BlockResponse> streamSource(StreamObserver<BlockStreamServiceGrpcProto.Block> responseStreamObserver) {
-        logger.info("Executing streamSource method");
+        logger.info("Executing bidirectional streamSource method");
 
         LiveStreamObserver<BlockStreamServiceGrpcProto.BlockResponse, BlockStreamServiceGrpcProto.Block> streamObserver = new HistoricStreamObserver(
                 streamMediator,
-                responseStreamObserver,
-                blockPersistenceHandler);
+                responseStreamObserver);
         streamMediator.subscribe(streamObserver);
 
         return streamObserver;

server/src/main/java/com/hedera/block/server/observe/InboundBlockStreamObserver.java

@@ -27,14 +27,14 @@
 
 public class InboundBlockStreamObserver implements StreamObserver<BlockStreamServiceGrpcProto.Block> {
 
-    private final StreamMediator<BlockStreamServiceGrpcProto.Block> streamMediator;
-    private final StreamObserver<BlockStreamServiceGrpcProto.BlockResponse> outboundBlockStreamObserver;
+    private final StreamMediator<BlockStreamServiceGrpcProto.BlockResponse, BlockStreamServiceGrpcProto.Block> streamMediator;
+    private final StreamObserver<BlockStreamServiceGrpcProto.BlockResponse> responseStreamObserver;
     private final Logger logger = Logger.getLogger(getClass().getName());
 
-    public InboundBlockStreamObserver(StreamMediator<BlockStreamServiceGrpcProto.Block> streamMediator,
-                                      StreamObserver<BlockStreamServiceGrpcProto.BlockResponse> outboundBlockStreamObserver) {
+    public InboundBlockStreamObserver(StreamMediator<BlockStreamServiceGrpcProto.BlockResponse, BlockStreamServiceGrpcProto.Block> streamMediator,
+                                      StreamObserver<BlockStreamServiceGrpcProto.BlockResponse> responseStreamObserver) {
         this.streamMediator = streamMediator;
-        this.outboundBlockStreamObserver = outboundBlockStreamObserver;
+        this.responseStreamObserver = responseStreamObserver;
     }
 
     @Override
@@ -44,18 +44,17 @@ public void onNext(BlockStreamServiceGrpcProto.Block block) {
         streamMediator.onNext(block);
 
         BlockStreamServiceGrpcProto.BlockResponse blockResponse = BlockStreamServiceGrpcProto.BlockResponse.newBuilder().setId(block.getId()).build();
-        this.outboundBlockStreamObserver.onNext(blockResponse);
+        this.responseStreamObserver.onNext(blockResponse);
     }
 
     @Override
     public void onError(Throwable t) {
         logger.severe("onError called: " + t.getMessage());
-        this.outboundBlockStreamObserver.onError(t);
     }
 
     @Override
     public void onCompleted() {
-        logger.fine("onCompleted called");
-        this.outboundBlockStreamObserver.onCompleted();
+        logger.info("InboundBlockStreamObserver onCompleted called");
+        this.streamMediator.unsubscribeAll();
     }
 }

server/src/main/java/com/hedera/block/server/observe/mediate/StreamMediator.java

@@ -25,7 +25,9 @@
 
 import java.util.function.Consumer;
 
-public interface StreamMediator<V> extends StreamObserver<V> {
-    void subscribe(LiveStreamObserver<BlockStreamServiceGrpcProto.BlockResponse, BlockStreamServiceGrpcProto.Block> c);
-    void unsubscribe(LiveStreamObserver<BlockStreamServiceGrpcProto.BlockResponse, BlockStreamServiceGrpcProto.Block> c);
+public interface StreamMediator<V, S> {
+    void subscribe(LiveStreamObserver<V, S> observer);
+    void unsubscribe(LiveStreamObserver<V, S> observer);
+    void unsubscribeAll();
+    void onNext(S block);
 }

server/src/main/java/com/hedera/block/server/observe/mediate/StreamMediatorImpl.java

@@ -25,24 +25,30 @@
 import java.util.Queue;
 import java.util.concurrent.LinkedBlockingQueue;
 
-public class StreamMediatorImpl implements StreamMediator<BlockStreamServiceGrpcProto.Block> {
+public class StreamMediatorImpl implements StreamMediator<BlockStreamServiceGrpcProto.BlockResponse, BlockStreamServiceGrpcProto.Block> {
 
-    private final Queue<LiveStreamObserver<BlockStreamServiceGrpcProto.BlockResponse, BlockStreamServiceGrpcProto.Block>> subscribers;
-    private final BlockPersistenceHandler<BlockStreamServiceGrpcProto.Block> blockPersistenceHandler;
+    private final Queue<LiveStreamObserver<BlockStreamServiceGrpcProto.BlockResponse, BlockStreamServiceGrpcProto.Block>> subscribers = new LinkedBlockingQueue<>();
+//    private final BlockPersistenceHandler<BlockStreamServiceGrpcProto.Block> blockPersistenceHandler;
 
     public StreamMediatorImpl(BlockPersistenceHandler<BlockStreamServiceGrpcProto.Block> blockPersistenceHandler) {
-        this.blockPersistenceHandler = blockPersistenceHandler;
-        this.subscribers = new LinkedBlockingQueue<>();
+//        this.blockPersistenceHandler = blockPersistenceHandler;
     }
 
     @Override
     public void subscribe(LiveStreamObserver<BlockStreamServiceGrpcProto.BlockResponse, BlockStreamServiceGrpcProto.Block> liveStreamObserver) {
-        subscribers.add(liveStreamObserver);
+        this.subscribers.add(liveStreamObserver);
     }
 
     @Override
     public void unsubscribe(LiveStreamObserver<BlockStreamServiceGrpcProto.BlockResponse, BlockStreamServiceGrpcProto.Block> liveStreamObserver) {
+        if (!this.subscribers.isEmpty()) {
+            this.subscribers.remove(liveStreamObserver);
+        }
+    }
 
+    @Override
+    public void unsubscribeAll() {
+        this.subscribers.clear();
     }
 
     @Override
@@ -56,18 +62,4 @@ public void onNext(BlockStreamServiceGrpcProto.Block block) {
         // Persist the block
 //        blockPersistenceHandler.persist(block);
     }
-
-    @Override
-    public void onError(Throwable t) {
-//        for (StreamObserver<BlockStreamServiceGrpcProto.BlockResponse> subscriber : subscribers) {
-//            subscriber.onError(t);
-//        }
-    }
-
-    @Override
-    public void onCompleted() {
-//        for (StreamObserver<BlockStreamServiceGrpcProto.BlockResponse> subscriber : subscribers) {
-//            subscriber.onCompleted();
-//        }
-    }
 }

server/src/main/java/com/hedera/block/server/observe/subscribe/HistoricStreamObserver.java

@@ -24,158 +24,46 @@
 import io.grpc.stub.StreamObserver;
 
 import java.time.Duration;
-import java.util.LinkedList;
-import java.util.Optional;
-import java.util.Queue;
+import java.util.*;
 import java.util.concurrent.LinkedBlockingQueue;
+import java.util.function.Consumer;
 import java.util.logging.Logger;
 
 public class HistoricStreamObserver implements LiveStreamObserver<BlockStreamServiceGrpcProto.BlockResponse, BlockStreamServiceGrpcProto.Block> {
 
     private final Logger logger = Logger.getLogger(getClass().getName());
 
-    private final StreamMediator<BlockStreamServiceGrpcProto.Block> mediator;
+    private final StreamMediator<BlockStreamServiceGrpcProto.BlockResponse, BlockStreamServiceGrpcProto.Block> mediator;
     private final StreamObserver<BlockStreamServiceGrpcProto.Block> responseStreamObserver;
-    private final BlockPersistenceHandler<BlockStreamServiceGrpcProto.Block> blockPersistenceHandler;
-
-    private final Queue<BlockStreamServiceGrpcProto.Block> historicStream;
-
-    private final Queue<BlockStreamServiceGrpcProto.Block> blocks = new LinkedBlockingQueue<>();
 
     public HistoricStreamObserver(
-            StreamMediator<BlockStreamServiceGrpcProto.Block> mediator,
-            StreamObserver<BlockStreamServiceGrpcProto.Block> responseStreamObserver,
-            BlockPersistenceHandler<BlockStreamServiceGrpcProto.Block> blockPersistenceHandler) {
+            StreamMediator<BlockStreamServiceGrpcProto.BlockResponse, BlockStreamServiceGrpcProto.Block> mediator,
+            StreamObserver<BlockStreamServiceGrpcProto.Block> responseStreamObserver) {
 
         this.mediator = mediator;
         this.responseStreamObserver = responseStreamObserver;
-        this.blockPersistenceHandler = blockPersistenceHandler;
-
-        this.historicStream = new LinkedList<>();
     }
 
     @Override
     public void notify(BlockStreamServiceGrpcProto.Block block) {
-        blocks.add(block);
+        this.responseStreamObserver.onNext(block);
     }
 
     @Override
     public void onNext(BlockStreamServiceGrpcProto.BlockResponse blockResponse) {
         logger.info("Received response block " + blockResponse);
-
-        while (true) {
-            while (!blocks.isEmpty()) {
-                responseStreamObserver.onNext(blocks.poll());
-            }
-
-            try {
-                Thread.sleep(Duration.ofMillis(50));
-            } catch (InterruptedException e) {
-                throw new RuntimeException(e);
-            }
-        }
     }
 
-//    @Override
-//    public void onNext(BlockStreamServiceGrpcProto.Block block) {
-//        if (firstLiveBlockId == -1) {
-//            firstLiveBlockId = block.getId();
-//        }
-//
-//        // Buffer the live blocks
-//        liveStream.add(block);
-//
-//        if (counter < firstLiveBlockId - 1) {
-//
-//            // Fetch the historic blocks
-//            historicStream.addAll(blockPersistenceHandler.readRange(1, firstLiveBlockId - 1));
-//            while (historicStream.peek() != null) {
-//                // Send the historic blocks downstream to the subscriber
-//                outboundStreamObserver.onNext(historicStream.poll());
-//                counter++;
-//            }
-//        } else {
-//
-//            while (liveStream.peek() != null) {
-//                // Send the buffered blocks downstream to the subscriber
-//                outboundStreamObserver.onNext(liveStream.poll());
-//            }
-//        }
-//
-//        this.mediator.subscribe(this::onNext);
-//    }
-
     @Override
     public void onError(Throwable t) {
-//        outboundStreamObserver.onError(t);
+        logger.severe("onError: " + t.getMessage());
+        mediator.unsubscribe(this);
     }
 
     @Override
     public void onCompleted() {
-        logger.info("Completed");
-//        outboundStreamObserver.onCompleted();
+        logger.info("gRPC connection completed.  Unsubscribing observer.");
+        mediator.unsubscribe(this);
+        logger.info("Unsubscribed observer.");
     }
-
-
-//    private static class StreamGenerator {
-//
-//        private final int inactivityLimit;
-//        private final BlockPersistenceHandler blockPersistenceHandler;
-//        private final StreamObserver<BlockStreamServiceGrpcProto.Block> observer;
-//
-//        StreamGenerator(int inactivityLimit,
-//                        BlockPersistenceHandler blockPersistenceHandler,
-//                        StreamObserver<BlockStreamServiceGrpcProto.Block> observer) {
-//            this.inactivityLimit = inactivityLimit;
-//            this.blockPersistenceHandler = blockPersistenceHandler;
-//            this.observer = observer;
-//        }
-//
-//        public void generate(long blockId) {
-//
-//            AtomicInteger inactivityCounter = new AtomicInteger();
-//            while (true) {
-//                Queue<BlockStreamServiceGrpcProto.Block> blocks = blockPersistenceHandler.readFrom(blockId, x -> true);
-//                blockId = generate(blockId, inactivityCounter, blocks);
-//                if (blockId == -1) {
-//                    break;
-//                }
-//            }
-//        }
-//
-//        private long generate(long blockId, AtomicInteger inactivityCounter, Queue<BlockStreamServiceGrpcProto.Block> blocks) {
-//
-//            if (blocks.isEmpty()) {
-//                if (inactivityCounter.get() >= inactivityLimit) {
-////                logger.debug("Thread inactivity limit reached.  Disconnecting client.");
-////                logger.debug("Object: " + this);
-//                    observer.onCompleted();
-//                    return -1;
-//                }
-//
-//                try {
-//                    Thread.sleep(1000);
-//                    inactivityCounter.incrementAndGet();
-//                } catch (InterruptedException e) {
-//                    throw new RuntimeException(e);
-//                }
-//
-//                return blockId;
-//            }
-//
-////    logger.debug("Generator: Id: " + state);
-////    logger.debug("Generator: block count: " + blocks.size());
-//
-//            inactivityCounter.set(0);
-//
-//            for (BlockStreamServiceGrpcProto.Block block : blocks) {
-//                observer.onNext(block);
-//                blockId = block.getId();
-//            }
-//
-//            blockId++;
-//
-//            return blockId;
-//        }
-//    }
 }