Replace use of deprecated DirectProcessor

Signed-off-by: Saai Syvendra (Github key) <[email protected]>
hashgraph · Oct 31, 2024 · 722cb6c · 722cb6c
1 parent f2c78f3
commit 722cb6c
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 9 deletions.
diff --git a/hedera-mirror-grpc/src/main/java/com/hedera/mirror/grpc/listener/SharedTopicListener.java b/hedera-mirror-grpc/src/main/java/com/hedera/mirror/grpc/listener/SharedTopicListener.java
@@ -35,16 +35,11 @@ public abstract class SharedTopicListener implements TopicListener {
     @Override
     public Flux<TopicMessage> listen(TopicMessageFilter filter) {
         Sinks.Many<TopicMessage> sink = Sinks.many().unicast().onBackpressureBuffer();
-        Flux<TopicMessage> overflowProcessor = sink.asFlux();
 
-        // moving publishOn from after onBackpressureBuffer to after Flux.merge reduces CPU usage by up to 40%
-        Flux<TopicMessage> topicMessageFlux = getSharedListener(filter)
+        return getSharedListener(filter)
                 .doOnSubscribe(s -> log.info("Subscribing: {}", filter))
-                .onBackpressureBuffer(
-                        listenerProperties.getMaxBufferSize(), t -> sink.tryEmitError(Exceptions.failWithOverflow()))
-                .doFinally(s -> sink.tryEmitComplete());
-
-        return Flux.merge(listenerProperties.getPrefetch(), topicMessageFlux, overflowProcessor)
+                .onBackpressureBuffer(listenerProperties.getMaxBufferSize(), BufferOverflowStrategy.ERROR)
+                .doFinally(s -> sink.tryEmitComplete())
                 .publishOn(Schedulers.boundedElastic(), false, listenerProperties.getPrefetch());
     }
 

diff --git a/...r-grpc/src/test/java/com/hedera/mirror/grpc/listener/AbstractSharedTopicListenerTest.java b/...r-grpc/src/test/java/com/hedera/mirror/grpc/listener/AbstractSharedTopicListenerTest.java
@@ -80,7 +80,7 @@ void slowSubscriberOverflowException() {
                 .expectNext(1L, 2L)
                 .thenAwait(Duration.ofMillis(500L)) // stall to overrun backpressure buffer
                 .thenRequest(Long.MAX_VALUE)
-                .thenConsumeWhile(n -> n < numMessages)
+                .expectNextCount(maxBufferSize)
                 .expectErrorMatches(Exceptions::isOverflow)
                 .verify(Duration.ofMillis(1000L));