Merge pull request #49 from TrashCoder96/master

Avro serialization support

Author: majusko

src/main/java/io/github/majusko/pulsar/PulsarAutoConfiguration.java

@@ -3,7 +3,6 @@
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.PulsarClientException;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.ComponentScan;

src/main/java/io/github/majusko/pulsar/PulsarSpringStarterUtils.java

@@ -0,0 +1,23 @@
+package io.github.majusko.pulsar;
+
+import io.github.majusko.pulsar.constant.Serialization;
+import io.github.majusko.pulsar.error.exception.ProducerInitException;
+import org.apache.pulsar.client.api.Schema;
+
+public class PulsarSpringStarterUtils {
+
+    public static <T> Schema<?> getSchema(Serialization serialization, Class<T> clazz) throws RuntimeException {
+        switch (serialization) {
+            case JSON: {
+                return Schema.JSON(clazz);
+            }
+            case AVRO: {
+                return Schema.AVRO(clazz);
+            }
+            default: {
+                throw new ProducerInitException("Unknown producer schema.");
+            }
+        }
+    }
+
+}

src/main/java/io/github/majusko/pulsar/constant/Serialization.java

@@ -1,5 +1,6 @@
 package io.github.majusko.pulsar.constant;
 
 public enum Serialization {
-    JSON
+    JSON,
+    AVRO
 }

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

@@ -1,13 +1,13 @@
 package io.github.majusko.pulsar.consumer;
 
+import io.github.majusko.pulsar.PulsarSpringStarterUtils;
 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.EmbeddedValueResolverAware;
 import org.springframework.context.annotation.DependsOn;
 import org.springframework.stereotype.Component;
@@ -46,7 +46,7 @@ private void init() {
     private Consumer<?> subscribe(String name, ConsumerHolder holder) {
         try {
             return pulsarClient
-                .newConsumer(Schema.JSON(holder.getAnnotation().clazz()))
+                .newConsumer(PulsarSpringStarterUtils.getSchema(holder.getAnnotation().serialization(), holder.getAnnotation().clazz()))
                 .consumerName("consumer-" + name)
                 .subscriptionName("subscription-" + name)
                 .topic(stringValueResolver.resolveStringValue(holder.getAnnotation().topic()))

src/main/java/io/github/majusko/pulsar/producer/ProducerCollector.java

@@ -1,8 +1,8 @@
 package io.github.majusko.pulsar.producer;
 
+import io.github.majusko.pulsar.PulsarSpringStarterUtils;
 import io.github.majusko.pulsar.annotation.PulsarProducer;
 import io.github.majusko.pulsar.collector.ProducerHolder;
-import io.github.majusko.pulsar.constant.Serialization;
 import io.github.majusko.pulsar.error.exception.ProducerInitException;
 import org.apache.pulsar.client.api.Producer;
 import org.apache.pulsar.client.api.PulsarClient;
@@ -55,10 +55,7 @@ private Producer<?> buildProducer(ProducerHolder holder) {
     }
 
     private <T> Schema<?> getSchema(ProducerHolder holder) throws RuntimeException {
-        if (holder.getSerialization().equals(Serialization.JSON)) {
-            return Schema.JSON(holder.getClazz());
-        }
-        throw new ProducerInitException("Unknown producer schema.");
+        return PulsarSpringStarterUtils.getSchema(holder.getSerialization(), holder.getClazz());
     }
 
     Map<String, Producer> getProducers() {

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

@@ -0,0 +1,14 @@
+package io.github.majusko.pulsar;
+
+public class AvroMsg {
+
+    private String data;
+
+    public String getData() {
+        return data;
+    }
+
+    public void setData(String data) {
+        this.data = data;
+    }
+}

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

@@ -20,6 +20,7 @@
 import org.testcontainers.junit.jupiter.Container;
 import org.testcontainers.junit.jupiter.Testcontainers;
 
+import java.time.Duration;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
@@ -48,6 +49,9 @@ class PulsarJavaSpringBootStarterApplicationTests {
     @Autowired
     private PulsarTemplate<String> producerForError;
 
+    @Autowired
+    private PulsarTemplate<AvroMsg> producerForAvroTopic;
+
     @Container
     static PulsarContainer pulsarContainer = new PulsarContainer();
 
@@ -70,7 +74,7 @@ void testProducerSendMethod() throws PulsarClientException {
     void testConsumerRegistration1() throws Exception {
         final List<Consumer> consumers = consumerBuilder.getConsumers();
 
-        Assertions.assertEquals(2, consumers.size());
+        Assertions.assertEquals(3, consumers.size());
 
         final Consumer<?> consumer = consumers.stream().filter( $-> $.getTopic().equals("topic-one")).findFirst().orElseThrow(Exception::new);
 
@@ -94,7 +98,7 @@ void testProducerRegistration() {
 
         final Map<String, ImmutablePair<Class<?>, Serialization>> topics = producerFactory.getTopics();
 
-        Assertions.assertEquals(3, topics.size());
+        Assertions.assertEquals(4, topics.size());
 
         final Set<String> topicNames = new HashSet<>(topics.keySet());
 
@@ -119,4 +123,12 @@ void testMessageErrorHandling() throws PulsarClientException {
 
         await().untilTrue(receivedError);
     }
+
+    @Test
+    void avroSerializationTestOk() throws Exception {
+        AvroMsg testAvroMsg = new AvroMsg();
+        testAvroMsg.setData("avro-test");
+        producerForAvroTopic.send("topic-avro", testAvroMsg);
+        await().atMost(Duration.ofSeconds(10)).until(() -> testConsumers.avroTopicReceived.get());
+    }
 }

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

@@ -11,6 +11,7 @@
 public class TestConsumers {
 
     public AtomicBoolean mockTopicListenerReceived = new AtomicBoolean(false);
+    public AtomicBoolean avroTopicReceived = new AtomicBoolean(false);
 
     @PulsarConsumer(topic = "topic-one", clazz = MyMsg.class, serialization = Serialization.JSON)
     public void topicOneListener(MyMsg myMsg) {
@@ -21,4 +22,10 @@ public void topicOneListener(MyMsg myMsg) {
     @PulsarConsumer(topic = "topic-for-error", clazz = String.class, serialization = Serialization.JSON)
     public void topicForErrorListener(Integer myMsg) {
     }
+
+    @PulsarConsumer(topic = "topic-avro", clazz = AvroMsg.class, serialization = Serialization.AVRO)
+    public void avroTopic(AvroMsg avroMsg) {
+        Assertions.assertNotNull(avroMsg);
+        avroTopicReceived.set(true);
+    }
 }

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

@@ -13,6 +13,7 @@ public ProducerFactory producerFactory() {
         return new ProducerFactory()
             .addProducer("topic-for-error", String.class)
             .addProducer("topic-one", MyMsg.class)
-            .addProducer("topic-two", MyMsg2.class, Serialization.JSON);
+            .addProducer("topic-two", MyMsg2.class, Serialization.JSON)
+            .addProducer("topic-avro", AvroMsg.class, Serialization.AVRO);
     }
 }