Implemented simple emmitter for catching exceptions during the message consumption.

Author: majusko

pom.xml

@@ -23,6 +23,7 @@
 		<testcontainers.version>1.14.3</testcontainers.version>
 		<junit-jupiter.version>5.6.2</junit-jupiter.version>
 		<awaitility.version>4.0.3</awaitility.version>
+		<reactor.core.version>3.3.6.RELEASE</reactor.core.version>
 		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
 	</properties>
 
@@ -38,6 +39,13 @@
 			<version>${apache.pulsar.client.version}</version>
 		</dependency>
 
+		<dependency>
+			<groupId>io.projectreactor</groupId>
+			<artifactId>reactor-core</artifactId>
+			<version>${reactor.core.version}</version>
+		</dependency>
+
+		<!--	TESTS	-->
 		<dependency>
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-starter-test</artifactId>
@@ -63,7 +71,6 @@
 			<scope>test</scope>
 		</dependency>
 
-        <!--	Pulsar integration tests	-->
 		<dependency>
 			<groupId>org.testcontainers</groupId>
 			<artifactId>testcontainers</artifactId>

src/main/java/io/github/majusko/pulsar/consumer/ConsumerBuilder.java

@@ -2,12 +2,16 @@
 
 import io.github.majusko.pulsar.collector.ConsumerCollector;
 import io.github.majusko.pulsar.collector.ConsumerHolder;
+import io.github.majusko.pulsar.error.FailedMessage;
+import io.github.majusko.pulsar.error.exception.ConsumerInitException;
 import org.apache.pulsar.client.api.Consumer;
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.api.Schema;
 import org.springframework.context.annotation.DependsOn;
 import org.springframework.stereotype.Component;
+import reactor.core.Disposable;
+import reactor.core.publisher.EmitterProcessor;
 
 import javax.annotation.PostConstruct;
 import java.lang.reflect.Method;
@@ -18,10 +22,11 @@
 @DependsOn({"pulsarClient", "consumerCollector"})
 public class ConsumerBuilder {
 
+    private final EmitterProcessor<FailedMessage> exceptionEmitter = EmitterProcessor.create();
     private final ConsumerCollector consumerCollector;
     private final PulsarClient pulsarClient;
 
-    private List<Consumer> consumers;
+    private List<Consumer<?>> consumers;
 
     public ConsumerBuilder(ConsumerCollector consumerCollector, PulsarClient pulsarClient) {
         this.consumerCollector = consumerCollector;
@@ -53,15 +58,19 @@ private Consumer<?> subscribe(String name, ConsumerHolder holder) {
                         consumer.acknowledge(msg);
                     } catch (Exception e) {
                         consumer.negativeAcknowledge(msg);
-                        throw new RuntimeException("TODO Custom Exception!", e);
+                        exceptionEmitter.onNext(new FailedMessage(e, consumer, msg));
                     }
                 }).subscribe();
         } catch (PulsarClientException e) {
-            throw new RuntimeException("TODO Custom Exception!", e);
+            throw new ConsumerInitException("Failed to init consumer.", e);
         }
     }
 
-    public List<Consumer> getConsumers() {
+    public List<Consumer<?>> getConsumers() {
         return consumers;
     }
+
+    public Disposable onError(java.util.function.Consumer<? super FailedMessage> consumer) {
+        return exceptionEmitter.subscribe(consumer);
+    }
 }

src/main/java/io/github/majusko/pulsar/error/FailedMessage.java

@@ -0,0 +1,28 @@
+package io.github.majusko.pulsar.error;
+
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Message;
+
+public class FailedMessage {
+    private final Exception exception;
+    private final Consumer<?> consumer;
+    private final Message<?> message;
+
+    public FailedMessage(Exception exception, Consumer<?> consumer, Message<?> message) {
+        this.exception = exception;
+        this.consumer = consumer;
+        this.message = message;
+    }
+
+    public Exception getException() {
+        return exception;
+    }
+
+    public Consumer<?> getConsumer() {
+        return consumer;
+    }
+
+    public Message<?> getMessage() {
+        return message;
+    }
+}

src/main/java/io/github/majusko/pulsar/error/exception/ConsumerInitException.java

@@ -0,0 +1,7 @@
+package io.github.majusko.pulsar.error.exception;
+
+public class ConsumerInitException extends RuntimeException {
+    public ConsumerInitException(String message, Throwable cause) {
+        super(message, cause);
+    }
+}

src/test/java/io/github/majusko/pulsar/PulsarJavaSpringBootStarterApplicationTests.java

@@ -64,11 +64,11 @@ void testProducerSendMethod() throws PulsarClientException {
 
     @Test
     void testConsumerRegistration1() throws Exception {
-        final List<Consumer> consumers = consumerBuilder.getConsumers();
+        final List<Consumer<?>> consumers = consumerBuilder.getConsumers();
 
         Assertions.assertEquals(1, consumers.size());
 
-        final Consumer consumer = consumers.stream().findFirst().orElseThrow(Exception::new);
+        final Consumer<?> consumer = consumers.stream().findFirst().orElseThrow(Exception::new);
 
         Assertions.assertNotNull(consumer);
         Assertions.assertEquals("topic-one", consumer.getTopic());