Adding backpressure pattern #3233

Author: sanurah

backpressure/README.md

@@ -11,10 +11,27 @@
 
     <artifactId>backpressure</artifactId>
 
-    <properties>
-        <maven.compiler.source>21</maven.compiler.source>
-        <maven.compiler.target>21</maven.compiler.target>
-        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-    </properties>
+    <dependencies>
+        <dependency>
+            <groupId>io.projectreactor</groupId>
+            <artifactId>reactor-core</artifactId>
+            <version>3.8.0-M1</version>
+        </dependency>
+        <dependency>
+            <groupId>ch.qos.logback</groupId>
+            <artifactId>logback-classic</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-engine</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>io.projectreactor</groupId>
+            <artifactId>reactor-test</artifactId>
+            <version>3.8.0-M1</version>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
 
 </project>

backpressure/pom.xml

@@ -0,0 +1,55 @@
+package com.iluwatar.backpressure;
+
+import java.util.concurrent.CountDownLatch;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * The Backpressure pattern is a flow control mechanism. It allows a consumer to signal to a
+ * producer to slow down or stop sending data when it's overwhelmed.
+ * <li>Prevents memory overflow, CPU thrashing, and resource exhaustion.
+ * <li>Ensures fair usage of resources in distributed systems.
+ * <li>Avoids buffer bloat and latency spikes.
+ * Key concepts of this design paradigm involves
+ * <li>Publisher/Producer: Generates data.
+ * <li>Subscriber/Consumer: Receives and processes data.
+ *
+ * <p>In this example we will create a {@link Publisher} and a {@link Subscriber}. Publisher will
+ * emit a stream of integer values with a predefined delay. Subscriber takes 500 ms to process one
+ * integer. Since the subscriber can't process the items fast enough we apply backpressure to the
+ * publisher so that it will request 10 items first, process 5 items and request for the next 5
+ * again. After processing 5 items subscriber will keep requesting for another 5 until the stream
+ * ends.
+ */
+@Slf4j
+public class App {
+
+  protected static CountDownLatch latch;
+
+  /**
+   * Program entry point.
+   *
+   * @param args command line args
+   */
+  public static void main(String[] args) throws InterruptedException {
+
+    /*
+     * This custom subscriber applies backpressure:
+     * - Has a processing delay of 0.5 milliseconds
+     * - Requests 10 items initially
+     * - Process 5 items and request for the next 5 items
+     */
+    Subscriber sub = new Subscriber();
+    // slow publisher emit 15 numbers with a delay of 200 milliseconds
+    Publisher.publish(1, 17, 200).subscribe(sub);
+
+    latch = new CountDownLatch(1);
+    latch.await();
+
+    sub = new Subscriber();
+    // fast publisher emit 15 numbers with a delay of 1 millisecond
+    Publisher.publish(1, 17, 1).subscribe(sub);
+
+    latch = new CountDownLatch(1);
+    latch.await();
+  }
+}

backpressure/src/main/java/com/iluwatar/App.java

@@ -0,0 +1,22 @@
+package com.iluwatar.backpressure;
+
+import java.time.Duration;
+import reactor.core.publisher.Flux;
+
+/**
+ * This class is the publisher that generates the data stream.
+ */
+public class Publisher {
+
+  /**
+   * On message method will trigger when the subscribed event is published.
+   *
+   * @param start starting integer
+   * @param count how many integers to emit
+   * @param delay delay between each item in milliseconds
+   * @return a flux stream of integers
+   */
+  public static Flux<Integer> publish(int start, int count, int delay) {
+    return Flux.range(start, count).delayElements(Duration.ofMillis(delay)).log();
+  }
+}

backpressure/src/main/java/com/iluwatar/backpressure/App.java

@@ -0,0 +1,46 @@
+package com.iluwatar.backpressure;
+
+import lombok.NonNull;
+import lombok.extern.slf4j.Slf4j;
+import org.reactivestreams.Subscription;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import reactor.core.publisher.BaseSubscriber;
+
+/**
+ * This class is the custom subscriber that subscribes to the data stream.
+ */
+@Slf4j
+public class Subscriber extends BaseSubscriber<Integer> {
+
+  private static final Logger logger = LoggerFactory.getLogger(Subscriber.class);
+
+  @Override
+  protected void hookOnSubscribe(@NonNull Subscription subscription) {
+    logger.info("subscribe()");
+    request(10); //request 10 items initially
+  }
+
+  @Override
+  protected void hookOnNext(@NonNull Integer value) {
+    processItem();
+    logger.info("process({})", value);
+    if (value % 5 == 0) {
+      // request for the next 5 items after processing first 5
+      request(5);
+    }
+  }
+
+  @Override
+  protected void hookOnComplete() {
+    App.latch.countDown();
+  }
+
+  private void processItem() {
+    try {
+      Thread.sleep(500); // simulate slow processing
+    } catch (InterruptedException e) {
+      logger.error(e.getMessage(), e);
+    }
+  }
+}

backpressure/src/main/java/com/iluwatar/backpressure/Publisher.java

@@ -0,0 +1,13 @@
+package com.iluwatar.backpressure;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+
+import org.junit.jupiter.api.Test;
+
+public class AppTest {
+
+  @Test
+  void shouldExecuteApplicationWithoutException() {
+    assertDoesNotThrow(() -> App.main(new String[] {}));
+  }
+}

backpressure/src/main/java/com/iluwatar/backpressure/Subscriber.java

@@ -0,0 +1,64 @@
+/*
+ * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt).
+ *
+ * The MIT License
+ * Copyright © 2014-2022 Ilkka Seppälä
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+ * THE SOFTWARE.
+ */
+package com.iluwatar.backpressure;
+
+import ch.qos.logback.classic.Logger;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.core.read.ListAppender;
+import java.util.List;
+import java.util.stream.Collectors;
+import org.junit.jupiter.api.extension.AfterEachCallback;
+import org.junit.jupiter.api.extension.BeforeEachCallback;
+import org.junit.jupiter.api.extension.ExtensionContext;
+import org.slf4j.LoggerFactory;
+
+public class LoggerExtension implements BeforeEachCallback, AfterEachCallback {
+
+  private final ListAppender<ILoggingEvent> listAppender = new ListAppender<>();
+  private final Logger logger = (Logger) LoggerFactory.getLogger(Logger.ROOT_LOGGER_NAME);
+
+  @Override
+  public void afterEach(ExtensionContext extensionContext) throws Exception {
+    listAppender.stop();
+    listAppender.list.clear();
+    logger.detachAppender(listAppender);
+  }
+
+  @Override
+  public void beforeEach(ExtensionContext extensionContext) throws Exception {
+    logger.addAppender(listAppender);
+    listAppender.start();
+  }
+
+  public List<String> getMessages() {
+    return listAppender.list.stream().map(e -> e.getMessage()).collect(Collectors.toList());
+  }
+
+  public List<String> getFormattedMessages() {
+    return listAppender.list.stream()
+        .map(e -> e.getFormattedMessage())
+        .collect(Collectors.toList());
+  }
+}

backpressure/src/test/java/com/iluwatar/backpressure/AppTest.java

@@ -0,0 +1,27 @@
+package com.iluwatar.backpressure;
+
+import static com.iluwatar.backpressure.Publisher.publish;
+
+import java.time.Duration;
+import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Flux;
+import reactor.test.StepVerifier;
+
+public class PublisherTest {
+
+  @Test
+  public void testPublish() {
+
+    Flux<Integer>  flux = publish(1, 3, 200);
+
+    StepVerifier.withVirtualTime(() -> flux)
+        .expectSubscription()
+        .expectNoEvent(Duration.ofMillis(200))
+        .expectNext(1)
+        .expectNoEvent(Duration.ofSeconds(200))
+        .expectNext(2)
+        .expectNoEvent(Duration.ofSeconds(200))
+        .expectNext(3)
+        .verifyComplete();
+  }
+}

backpressure/src/test/java/com/iluwatar/backpressure/LoggerExtension.java

@@ -0,0 +1,26 @@
+package com.iluwatar.backpressure;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.util.concurrent.CountDownLatch;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+public class SubscriberTest {
+
+  @RegisterExtension
+  public LoggerExtension loggerExtension = new LoggerExtension();
+
+  @Test
+  public void testSubscribe() throws InterruptedException {
+    App.latch = new CountDownLatch(1);
+    Subscriber sub = new Subscriber();
+    Publisher.publish(1, 8, 100).subscribe(sub);
+
+    App.latch.await();
+    assertEquals(22, loggerExtension.getFormattedMessages().size());
+    assertEquals("subscribe()", loggerExtension.getFormattedMessages().get(2));
+    assertEquals("request(10)", loggerExtension.getFormattedMessages().get(3));
+    assertEquals("request(5)", loggerExtension.getFormattedMessages().get(14));
+  }
+}