Merge pull request #79 from majusko/feature/string-byte-proto-support

majusko · web-flow · commit 023d36676014 · 2021-01-23T20:20:30.000+01:00
Feature/string byte proto support
diff --git a/src/main/java/io/github/majusko/pulsar/PulsarSpringStarterUtils.java b/src/main/java/io/github/majusko/pulsar/PulsarSpringStarterUtils.java
@@ -1,5 +1,6 @@
 package io.github.majusko.pulsar;
 
+import com.google.protobuf.GeneratedMessageV3;
 import io.github.majusko.pulsar.constant.Serialization;
 import io.github.majusko.pulsar.error.exception.ProducerInitException;
 import org.apache.pulsar.client.api.Schema;
@@ -8,20 +9,49 @@
 
 public class PulsarSpringStarterUtils {
 
-    public static <T> Schema<?> getSchema(Serialization serialization, Class<T> clazz) throws RuntimeException {
+    private PulsarSpringStarterUtils() {
+    }
+
+    private static <T> Schema<?> getGenericSchema(Serialization serialization, Class<T> clazz) throws RuntimeException {
         switch (serialization) {
             case JSON: {
                 return Schema.JSON(clazz);
             }
             case AVRO: {
                 return Schema.AVRO(clazz);
             }
+            case STRING: {
+                return Schema.STRING;
+            }
             default: {
                 throw new ProducerInitException("Unknown producer schema.");
             }
         }
     }
 
+    private static <T extends com.google.protobuf.GeneratedMessageV3> Schema<?> getProtoSchema(Serialization serialization, Class<T> clazz) throws RuntimeException {
+        if (serialization == Serialization.PROTOBUF) {
+            return Schema.PROTOBUF(clazz);
+        }
+        throw new ProducerInitException("Unknown producer schema.");
+    }
+
+    public static Schema<?> getSchema(Serialization serialisation, Class<?> clazz) {
+        if (clazz == byte[].class) {
+            return Schema.BYTES;
+        }
+
+        if (isProto(serialisation)) {
+            return getProtoSchema(serialisation, (Class<? extends GeneratedMessageV3>) clazz);
+        }
+
+        return getGenericSchema(serialisation, clazz);
+    }
+
+    public static boolean isProto(Serialization serialization) {
+        return serialization == Serialization.PROTOBUF;
+    }
+
     public static Class<?> getParameterType(Method method) {
         return method.getParameterTypes()[0];
     }
diff --git a/src/main/java/io/github/majusko/pulsar/annotation/PulsarConsumer.java b/src/main/java/io/github/majusko/pulsar/annotation/PulsarConsumer.java
@@ -13,7 +13,7 @@
 public @interface PulsarConsumer {
     String topic();
 
-    Class<?> clazz();
+    Class<?> clazz() default byte[].class;
 
     Serialization serialization() default Serialization.JSON;
 
diff --git a/src/main/java/io/github/majusko/pulsar/collector/ConsumerCollector.java b/src/main/java/io/github/majusko/pulsar/collector/ConsumerCollector.java
@@ -1,11 +1,12 @@
 package io.github.majusko.pulsar.collector;
 
-import io.github.majusko.pulsar.PulsarSpringStarterUtils;
 import io.github.majusko.pulsar.annotation.PulsarConsumer;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.beans.factory.config.BeanPostProcessor;
 import org.springframework.context.annotation.Configuration;
 
+import java.lang.reflect.Method;
+import java.lang.reflect.Type;
 import java.util.Arrays;
 import java.util.Map;
 import java.util.Optional;
@@ -29,8 +30,9 @@ public Object postProcessBeforeInitialization(Object bean, String beanName) {
         consumers.putAll(Arrays.stream(beanClass.getDeclaredMethods())
             .filter($ -> $.isAnnotationPresent(PulsarConsumer.class))
             .collect(Collectors.toMap(
-                method -> beanClass.getName() + consumerNameDelimiter + method.getName(),
-                method -> new ConsumerHolder(method.getAnnotation(PulsarConsumer.class), method, bean, getParameterType(method)))));
+                method -> getConsumerName(beanClass, method),
+                method -> new ConsumerHolder(method.getAnnotation(PulsarConsumer.class), method, bean,
+                    getParameterType(method)))));
 
         return bean;
     }
@@ -47,4 +49,11 @@ public Map<String, ConsumerHolder> getConsumers() {
     public Optional<ConsumerHolder> getConsumer(String methodDescriptor) {
         return Optional.ofNullable(consumers.get(methodDescriptor));
     }
+
+    public String getConsumerName(Class<?> clazz, Method method) {
+        return clazz.getName() + consumerNameDelimiter + method.getName() + Arrays
+            .stream(method.getGenericParameterTypes())
+            .map(Type::getTypeName)
+            .collect(Collectors.joining(consumerNameDelimiter));
+    }
 }
diff --git a/src/main/java/io/github/majusko/pulsar/constant/Serialization.java b/src/main/java/io/github/majusko/pulsar/constant/Serialization.java
@@ -2,5 +2,8 @@
 
 public enum Serialization {
     JSON,
-    AVRO
+    AVRO,
+    STRING,
+    BYTE,
+    PROTOBUF
 }
diff --git a/src/main/java/io/github/majusko/pulsar/consumer/ConsumerAggregator.java b/src/main/java/io/github/majusko/pulsar/consumer/ConsumerAggregator.java
@@ -1,10 +1,12 @@
 package io.github.majusko.pulsar.consumer;
 
+import com.google.protobuf.GeneratedMessageV3;
 import io.github.majusko.pulsar.ConsumerProperties;
 import io.github.majusko.pulsar.PulsarMessage;
 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.constant.Serialization;
 import io.github.majusko.pulsar.error.FailedMessage;
 import io.github.majusko.pulsar.error.exception.ConsumerInitException;
 import org.apache.pulsar.client.api.*;
diff --git a/src/main/java/io/github/majusko/pulsar/producer/ProducerFactory.java b/src/main/java/io/github/majusko/pulsar/producer/ProducerFactory.java
@@ -12,6 +12,10 @@ public class ProducerFactory implements PulsarProducerFactory {
 
     private final Map<String, ImmutablePair<Class<?>, Serialization>> topics = new HashMap<>();
 
+    public ProducerFactory addProducer(String topic) {
+        return addProducer(topic, byte[].class, Serialization.BYTE);
+    }
+
     public ProducerFactory addProducer(String topic, Class<?> clazz) {
         topics.put(topic, new ImmutablePair<>(clazz, Serialization.JSON));
         return this;
diff --git a/src/test/java/io/github/majusko/pulsar/PulsarJavaSpringBootStarterApplicationTests.java b/src/test/java/io/github/majusko/pulsar/PulsarJavaSpringBootStarterApplicationTests.java
@@ -4,6 +4,9 @@
 import io.github.majusko.pulsar.collector.ConsumerHolder;
 import io.github.majusko.pulsar.constant.Serialization;
 import io.github.majusko.pulsar.consumer.ConsumerAggregator;
+import io.github.majusko.pulsar.msg.AvroMsg;
+import io.github.majusko.pulsar.msg.MyMsg;
+import io.github.majusko.pulsar.msg.ProtoMsg;
 import io.github.majusko.pulsar.producer.ProducerFactory;
 import io.github.majusko.pulsar.producer.PulsarTemplate;
 import org.apache.commons.lang3.tuple.ImmutablePair;
@@ -22,6 +25,7 @@
 import org.testcontainers.junit.jupiter.Testcontainers;
 import reactor.core.Disposable;
 
+import java.nio.charset.StandardCharsets;
 import java.time.Duration;
 import java.util.*;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -53,12 +57,23 @@ class PulsarJavaSpringBootStarterApplicationTests {
     @Autowired
     private PulsarTemplate<AvroMsg> producerForAvroTopic;
 
+    @Autowired
+    private PulsarTemplate<ProtoMsg> producerForProtoTopic;
+
+    @Autowired
+    private PulsarTemplate<byte[]> producerForByteTopic;
+
+    @Autowired
+    private PulsarTemplate<String> producerForStringTopic;
+
     @Container
     static PulsarContainer pulsarContainer = new PulsarContainer();
 
     @Autowired
     private TestConsumers testConsumers;
 
+    public static final String VALIDATION_STRING = "validation-string";
+
     @DynamicPropertySource
     static void propertySettings(DynamicPropertyRegistry registry) {
         registry.add("pulsar.serviceUrl", pulsarContainer::getPulsarBrokerUrl);
@@ -93,11 +108,11 @@ void testProducerSendAsyncMethod() throws PulsarClientException {
     @Test
     void testProducerCreateMessageMethod() throws PulsarClientException {
         producer.createMessage("topic-message", new MyMsg("my-message"))
-                .property("my-key", "my-value")
-                .property("my-other-key", "my-other-value")
-                .sequenceId(123l)
-                .key("my-key")
-                .send();
+            .property("my-key", "my-value")
+            .property("my-other-key", "my-other-value")
+            .sequenceId(123l)
+            .key("my-key")
+            .send();
 
         await().untilTrue(testConsumers.mockTopicMessageListenerReceived);
     }
@@ -106,9 +121,10 @@ void testProducerCreateMessageMethod() throws PulsarClientException {
     void testConsumerRegistration1() throws Exception {
         final List<Consumer> consumers = consumerAggregator.getConsumers();
 
-        Assertions.assertEquals(6, consumers.size());
+        Assertions.assertEquals(9, consumers.size());
 
-        final Consumer<?> consumer = consumers.stream().filter( $-> $.getTopic().equals("topic-one")).findFirst().orElseThrow(Exception::new);
+        final Consumer<?> consumer =
+            consumers.stream().filter($ -> $.getTopic().equals("topic-one")).findFirst().orElseThrow(Exception::new);
 
         Assertions.assertNotNull(consumer);
     }
@@ -117,22 +133,26 @@ void testConsumerRegistration1() throws Exception {
     void testConsumerRegistration2() {
         final Class<TestConsumers> clazz = TestConsumers.class;
         final List<ConsumerHolder> consumerHolders = Arrays.stream(clazz.getMethods())
-            .map($ -> consumerCollector.getConsumer(clazz.getName() + $.getName()).orElse(null))
+            .map($ -> consumerCollector.getConsumer(consumerCollector.getConsumerName(clazz, $)))
+            .filter(Optional::isPresent)
+            .map(Optional::get)
             .collect(Collectors.toList());
 
         Assertions.assertNotNull(consumerHolders);
         Assertions.assertTrue(consumerHolders.stream().anyMatch($ -> $.getAnnotation().topic().equals("topic-one")));
-        Assertions.assertTrue(consumerHolders.stream().anyMatch($ -> $.getAnnotation().topic().equals("topic-for-error")));
+        Assertions.assertTrue(consumerHolders.stream().anyMatch($ -> $.getAnnotation().topic().equals("topic-for" +
+            "-error")));
         Assertions.assertTrue(consumerHolders.stream().anyMatch($ -> $.getBean().getClass().equals(TestConsumers.class)));
-        Assertions.assertTrue(consumerHolders.stream().anyMatch($ -> $.getHandler().getName().equals("topicOneListener")));
+        Assertions.assertTrue(consumerHolders.stream().anyMatch($ -> $.getHandler().getName().equals(
+            "topicOneListener")));
     }
 
     @Test
     void testProducerRegistration() {
 
         final Map<String, ImmutablePair<Class<?>, Serialization>> topics = producerFactory.getTopics();
 
-        Assertions.assertEquals(7, topics.size());
+        Assertions.assertEquals(10, topics.size());
 
         final Set<String> topicNames = new HashSet<>(topics.keySet());
 
@@ -166,4 +186,25 @@ void avroSerializationTestOk() throws Exception {
         producerForAvroTopic.send("topic-avro", testAvroMsg);
         await().atMost(Duration.ofSeconds(10)).until(() -> testConsumers.avroTopicReceived.get());
     }
+
+    @Test
+    void protoSerializationTestOk() throws Exception {
+        final ProtoMsg msg = ProtoMsg.newBuilder().setData(VALIDATION_STRING).build();
+        producerForProtoTopic.send("topic-proto", msg);
+        await().atMost(Duration.ofSeconds(10)).until(() -> testConsumers.protoTopicReceived.get());
+    }
+
+    @Test
+    void byteSerializationTestOk() throws Exception {
+        byte[] data = VALIDATION_STRING.getBytes(StandardCharsets.UTF_8);
+
+        producerForByteTopic.send("topic-byte", data);
+        await().atMost(Duration.ofSeconds(10)).until(() -> testConsumers.byteTopicReceived.get());
+    }
+
+    @Test
+    void stringSerializationTestOk() throws Exception {
+        producerForStringTopic.send("topic-string", VALIDATION_STRING);
+        await().atMost(Duration.ofSeconds(10)).until(() -> testConsumers.stringTopicReceived.get());
+    }
 }
diff --git a/src/test/java/io/github/majusko/pulsar/TestConsumers.java b/src/test/java/io/github/majusko/pulsar/TestConsumers.java
@@ -2,10 +2,14 @@
 
 import io.github.majusko.pulsar.annotation.PulsarConsumer;
 import io.github.majusko.pulsar.constant.Serialization;
+import io.github.majusko.pulsar.msg.AvroMsg;
+import io.github.majusko.pulsar.msg.MyMsg;
+import io.github.majusko.pulsar.msg.ProtoMsg;
 import org.apache.pulsar.client.api.SubscriptionType;
 import org.junit.jupiter.api.Assertions;
 import org.springframework.stereotype.Service;
 
+import java.nio.charset.StandardCharsets;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 
@@ -16,6 +20,9 @@ public class TestConsumers {
     public AtomicBoolean mockTopicAsyncListenerReceived = new AtomicBoolean(false);
     public AtomicBoolean mockTopicMessageListenerReceived = new AtomicBoolean(false);
     public AtomicBoolean avroTopicReceived = new AtomicBoolean(false);
+    public AtomicBoolean protoTopicReceived = new AtomicBoolean(false);
+    public AtomicBoolean byteTopicReceived = new AtomicBoolean(false);
+    public AtomicBoolean stringTopicReceived = new AtomicBoolean(false);
     public AtomicBoolean mockRetryCountListenerReceived = new AtomicBoolean(false);
     public AtomicInteger retryCount = new AtomicInteger(0);
 
@@ -35,6 +42,30 @@ public void avroTopic(AvroMsg avroMsg) {
         avroTopicReceived.set(true);
     }
 
+    @PulsarConsumer(topic = "topic-proto", clazz = ProtoMsg.class, serialization = Serialization.PROTOBUF)
+    public void protoTopic(ProtoMsg protoMsg) {
+        Assertions.assertNotNull(protoMsg);
+        Assertions.assertEquals(PulsarJavaSpringBootStarterApplicationTests.VALIDATION_STRING, protoMsg.getData());
+
+        protoTopicReceived.set(true);
+    }
+
+    @PulsarConsumer(topic = "topic-byte")
+    public void byteTopic(byte[] byteMsg) {
+        Assertions.assertNotNull(byteMsg);
+        Assertions.assertEquals(PulsarJavaSpringBootStarterApplicationTests.VALIDATION_STRING, new String(byteMsg, StandardCharsets.UTF_8));
+
+        byteTopicReceived.set(true);
+    }
+
+    @PulsarConsumer(topic = "topic-string", clazz = String.class, serialization = Serialization.STRING)
+    public void byteTopic(String stringMsg) {
+        Assertions.assertNotNull(stringMsg);
+        Assertions.assertEquals(PulsarJavaSpringBootStarterApplicationTests.VALIDATION_STRING, stringMsg);
+
+        stringTopicReceived.set(true);
+    }
+
     @PulsarConsumer(topic = "topic-async", clazz = MyMsg.class, serialization = Serialization.JSON)
     public void topicAsyncListener(MyMsg myMsg) {
         Assertions.assertNotNull(myMsg);
@@ -63,5 +94,11 @@ public void failTwice(MyMsg myMsg) throws Exception {
         }
         Assertions.assertNotNull(myMsg);
         mockRetryCountListenerReceived.set(true);
+
+
+    }
+
+    public static Serialization aa() {
+        return Serialization.BYTE;
     }
 }
diff --git a/src/test/java/io/github/majusko/pulsar/TestProducerConfiguration.java b/src/test/java/io/github/majusko/pulsar/TestProducerConfiguration.java
@@ -1,6 +1,10 @@
 package io.github.majusko.pulsar;
 
 import io.github.majusko.pulsar.constant.Serialization;
+import io.github.majusko.pulsar.msg.AvroMsg;
+import io.github.majusko.pulsar.msg.MyMsg;
+import io.github.majusko.pulsar.msg.MyMsg2;
+import io.github.majusko.pulsar.msg.ProtoMsg;
 import io.github.majusko.pulsar.producer.ProducerFactory;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -17,6 +21,9 @@ public ProducerFactory producerFactory() {
             .addProducer("topic-avro", AvroMsg.class, Serialization.AVRO)
             .addProducer("topic-async", MyMsg.class)
             .addProducer("topic-message", MyMsg.class)
-            .addProducer("topic-retry", MyMsg.class);
+            .addProducer("topic-retry", MyMsg.class)
+            .addProducer("topic-string", String.class, Serialization.STRING)
+            .addProducer("topic-byte")
+            .addProducer("topic-proto", ProtoMsg.class, Serialization.PROTOBUF);
     }
 }
diff --git a/src/test/java/io/github/majusko/pulsar/msg/AvroMsg.java b/src/test/java/io/github/majusko/pulsar/msg/AvroMsg.java
@@ -1,4 +1,4 @@
-package io.github.majusko.pulsar;
+package io.github.majusko.pulsar.msg;
 
 public class AvroMsg {
 
diff --git a/src/test/java/io/github/majusko/pulsar/msg/Example.java b/src/test/java/io/github/majusko/pulsar/msg/Example.java
diff --git a/src/test/java/io/github/majusko/pulsar/msg/MyMsg.java b/src/test/java/io/github/majusko/pulsar/msg/MyMsg.java
diff --git a/src/test/java/io/github/majusko/pulsar/msg/MyMsg2.java b/src/test/java/io/github/majusko/pulsar/msg/MyMsg2.java
diff --git a/src/test/java/io/github/majusko/pulsar/msg/ProtoMsg.java b/src/test/java/io/github/majusko/pulsar/msg/ProtoMsg.java
diff --git a/src/test/java/io/github/majusko/pulsar/msg/ProtoMsgOrBuilder.java b/src/test/java/io/github/majusko/pulsar/msg/ProtoMsgOrBuilder.java

Original file line number	Diff line number	Diff line change
`@@ -2,5 +2,8 @@`
`2`	`2`
`3`	`3`	`public enum Serialization {`
`4`	`4`	`JSON,`
`5`		`- AVRO`
	`5`	`+ AVRO,`
	`6`	`+ STRING,`
	`7`	`+ BYTE,`
	`8`	`+ PROTOBUF`
`6`	`9`	`}`
Original file line number	Diff line number	Diff line change
`@@ -1,4 +1,4 @@`
`1`		`-package io.github.majusko.pulsar;`
	`1`	`+package io.github.majusko.pulsar.msg;`
`2`	`2`
`3`	`3`	`public class AvroMsg {`
`4`	`4`