fix: only send blockitems after the beginning of the next block is reached

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

Author: mattp-swirldslabs

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

@@ -45,6 +45,8 @@ public class ConsumerBlockItemObserver implements BlockItemEventHandler<ObjectEv
 
     private final StreamMediator<ObjectEvent<BlockItem>, BlockItem> streamMediator;
 
+    private boolean streamStarted;
+
     /**
      * Constructor for the LiveStreamObserverImpl class.
      *
@@ -96,11 +98,19 @@ public void onEvent(final ObjectEvent<BlockItem> event, final long l, final bool
             // Refresh the producer liveness and pass the BlockItem to the downstream observer.
             producerLivenessMillis = currentMillis;
 
-            // Construct a response
-            final SubscribeStreamResponse subscribeStreamResponse =
-                    SubscribeStreamResponse.newBuilder().setBlockItem(event.get()).build();
+            // Only start sending BlockItems after we've reached
+            // the beginning of a block.
+            final BlockItem blockItem = event.get();
+            if (!streamStarted && blockItem.hasHeader()) {
+                streamStarted = true;
+            }
+
+            if (streamStarted) {
+                final SubscribeStreamResponse subscribeStreamResponse =
+                        SubscribeStreamResponse.newBuilder().setBlockItem(blockItem).build();
 
-            subscribeStreamResponseObserver.onNext(subscribeStreamResponse);
+                subscribeStreamResponseObserver.onNext(subscribeStreamResponse);
+            }
         }
     }
 

server/src/test/java/com/hedera/block/server/consumer/ConsumerBlockItemObserverTest.java

@@ -22,6 +22,7 @@
 import static com.hedera.block.server.util.TestClock.buildClockOutsideWindow;
 import static org.mockito.Mockito.*;
 
+import com.hedera.block.protos.BlockStreamService;
 import com.hedera.block.server.data.ObjectEvent;
 import com.hedera.block.server.mediator.StreamMediator;
 import io.grpc.stub.StreamObserver;
@@ -51,7 +52,9 @@ public void testProducerTimeoutWithinWindow() {
                         streamMediator,
                         responseStreamObserver);
 
-        final BlockItem blockItem = BlockItem.newBuilder().build();
+        final BlockStreamService.BlockHeader blockHeader =
+                BlockStreamService.BlockHeader.newBuilder().setBlockNumber(1).build();
+        final BlockItem blockItem = BlockItem.newBuilder().setHeader(blockHeader).build();
         when(objectEvent.get()).thenReturn(blockItem);
 
         final SubscribeStreamResponse subscribeStreamResponse =

server/src/test/java/com/hedera/block/server/mediator/LiveStreamMediatorImplTest.java

@@ -146,7 +146,8 @@ public void testMediatorPublishEventToSubscribers() throws InterruptedException
                 streamMediator.isSubscribed(concreteObserver3),
                 "Expected the mediator to have observer3 subscribed");
 
-        final BlockItem blockItem = BlockItem.newBuilder().build();
+        final BlockHeader blockHeader = BlockHeader.newBuilder().setBlockNumber(1).build();
+        final BlockItem blockItem = BlockItem.newBuilder().setHeader(blockHeader).build();
         final SubscribeStreamResponse subscribeStreamResponse =
                 SubscribeStreamResponse.newBuilder().setBlockItem(blockItem).build();