From 04184757db0eb855ca9f8da34ac056d2f720db87 Mon Sep 17 00:00:00 2001
From: Matt Peterson <matt.peterson@swirldslabs.com>
Date: Thu, 27 Jun 2024 13:31:47 -0600
Subject: [PATCH] fix: removed unsubscribeall.  we do not want a producer to
 evict all consumers

Signed-off-by: Matt Peterson <matt.peterson@swirldslabs.com>
---
 .../mediator/LiveStreamMediatorImpl.java      | 11 ----------
 .../block/server/mediator/StreamMediator.java |  5 -----
 .../producer/ProducerBlockStreamObserver.java |  7 ++++--
 .../mediator/LiveStreamMediatorImplTest.java  | 22 -------------------
 4 files changed, 5 insertions(+), 40 deletions(-)

diff --git a/server/src/main/java/com/hedera/block/server/mediator/LiveStreamMediatorImpl.java b/server/src/main/java/com/hedera/block/server/mediator/LiveStreamMediatorImpl.java
index 9eedbdcd5..057968037 100644
--- a/server/src/main/java/com/hedera/block/server/mediator/LiveStreamMediatorImpl.java
+++ b/server/src/main/java/com/hedera/block/server/mediator/LiveStreamMediatorImpl.java
@@ -64,8 +64,6 @@ public void subscribe(final LiveStreamObserver<BlockStreamServiceGrpcProto.Block
     public void unsubscribe(final LiveStreamObserver<BlockStreamServiceGrpcProto.Block, BlockStreamServiceGrpcProto.BlockResponse> liveStreamObserver) {
         if (subscribers.remove(liveStreamObserver)) {
             LOGGER.log(System.Logger.Level.DEBUG, "Successfully removed observer from subscription list");
-        } else {
-            LOGGER.log(System.Logger.Level.ERROR, "Failed to remove observer from subscription list");
         }
     }
 
@@ -80,15 +78,6 @@ public boolean isSubscribed(final LiveStreamObserver<BlockStreamServiceGrpcProto
         return subscribers.contains(observer);
     }
 
-    /**
-     * Unsubscribe all observers from the mediator
-     */
-    @Override
-    public void unsubscribeAll() {
-        LOGGER.log(System.Logger.Level.DEBUG, "Unsubscribing all observers from the mediator");
-        subscribers.clear();
-    }
-
     /**
      * Notify all observers of a new block
      *
diff --git a/server/src/main/java/com/hedera/block/server/mediator/StreamMediator.java b/server/src/main/java/com/hedera/block/server/mediator/StreamMediator.java
index 66d1135af..07f448df7 100644
--- a/server/src/main/java/com/hedera/block/server/mediator/StreamMediator.java
+++ b/server/src/main/java/com/hedera/block/server/mediator/StreamMediator.java
@@ -53,11 +53,6 @@ public interface StreamMediator<U, V> {
      */
     boolean isSubscribed(final LiveStreamObserver<U, V> observer);
 
-    /**
-     * Unsubscribes all LiveStreamObservers from the producer
-     */
-    void unsubscribeAll();
-
     /**
      * Passes the newly arrived block to all subscribers
      *
diff --git a/server/src/main/java/com/hedera/block/server/producer/ProducerBlockStreamObserver.java b/server/src/main/java/com/hedera/block/server/producer/ProducerBlockStreamObserver.java
index 9d07645eb..5691cc24e 100644
--- a/server/src/main/java/com/hedera/block/server/producer/ProducerBlockStreamObserver.java
+++ b/server/src/main/java/com/hedera/block/server/producer/ProducerBlockStreamObserver.java
@@ -55,8 +55,11 @@ public ProducerBlockStreamObserver(final StreamMediator<BlockStreamServiceGrpcPr
      */
     @Override
     public void onNext(final BlockStreamServiceGrpcProto.Block block) {
+
+        // Notify all the mediator subscribers
         streamMediator.notifyAll(block);
 
+        // Send a response back to the upstream producer
         final BlockStreamServiceGrpcProto.BlockResponse blockResponse = BlockStreamServiceGrpcProto.BlockResponse.newBuilder().setId(block.getId()).build();
         responseStreamObserver.onNext(blockResponse);
     }
@@ -69,6 +72,7 @@ public void onNext(final BlockStreamServiceGrpcProto.Block block) {
     @Override
     public void onError(final Throwable t) {
         LOGGER.log(System.Logger.Level.ERROR, "onError method invoked with an exception", t);
+        responseStreamObserver.onError(t);
     }
 
     /**
@@ -78,7 +82,6 @@ public void onError(final Throwable t) {
     @Override
     public void onCompleted() {
         LOGGER.log(System.Logger.Level.DEBUG, "ProducerBlockStreamObserver completed");
-        streamMediator.unsubscribeAll();
-        LOGGER.log(System.Logger.Level.DEBUG, "Unsubscribed all downstream consumers");
+        responseStreamObserver.onCompleted();
     }
 }
diff --git a/server/src/test/java/com/hedera/block/server/mediator/LiveStreamMediatorImplTest.java b/server/src/test/java/com/hedera/block/server/mediator/LiveStreamMediatorImplTest.java
index 2db3a4547..2692f69b5 100644
--- a/server/src/test/java/com/hedera/block/server/mediator/LiveStreamMediatorImplTest.java
+++ b/server/src/test/java/com/hedera/block/server/mediator/LiveStreamMediatorImplTest.java
@@ -49,28 +49,6 @@ public class LiveStreamMediatorImplTest {
     @Mock
     private BlockCache<BlockStreamServiceGrpcProto.Block> blockCache;
 
-    @Test
-    public void testUnsubscribeAll() {
-
-        final StreamMediator<BlockStreamServiceGrpcProto.Block, BlockStreamServiceGrpcProto.BlockResponse> streamMediator =
-                new LiveStreamMediatorImpl(new WriteThroughCacheHandler(blockStorage, blockCache));
-
-        // Set up the subscribers
-        streamMediator.subscribe(liveStreamObserver1);
-        streamMediator.subscribe(liveStreamObserver2);
-        streamMediator.subscribe(liveStreamObserver3);
-
-        assertTrue(streamMediator.isSubscribed(liveStreamObserver1), "Expected the mediator to have liveStreamObserver1 subscribed");
-        assertTrue(streamMediator.isSubscribed(liveStreamObserver2), "Expected the mediator to have liveStreamObserver2 subscribed");
-        assertTrue(streamMediator.isSubscribed(liveStreamObserver3), "Expected the mediator to have liveStreamObserver3 subscribed");
-
-        streamMediator.unsubscribeAll();
-
-        assertFalse(streamMediator.isSubscribed(liveStreamObserver1), "Expected the mediator to have unsubscribed liveStreamObserver1");
-        assertFalse(streamMediator.isSubscribed(liveStreamObserver2), "Expected the mediator to have unsubscribed liveStreamObserver2");
-        assertFalse(streamMediator.isSubscribed(liveStreamObserver3), "Expected the mediator to have unsubscribed liveStreamObserver3");
-    }
-
     @Test
     public void testUnsubscribeEach() {