Merge pull request #19 from majusko/bugfix/issue-13-messages-sending-problem

resolves #13 and added real integration tests without mocking.

Author: majusko

.travis.yml

@@ -2,6 +2,8 @@ language: java
 jdk:
 - oraclejdk8
 dist: trusty
+services:
+- docker
 after_success:
 - bash <(curl -s https://codecov.io/bash)
 before_deploy:

pom.xml

@@ -5,7 +5,7 @@
 	<parent>
 		<groupId>org.springframework.boot</groupId>
 		<artifactId>spring-boot-starter-parent</artifactId>
-		<version>2.2.5.RELEASE</version>
+		<version>2.3.1.RELEASE</version>
 		<relativePath/> <!-- lookup parent from repository -->
 	</parent>
 
@@ -19,8 +19,10 @@
 		<java.version>1.8</java.version>
 		<maven.compiler.target>1.8</maven.compiler.target>
 		<maven.compiler.source>1.8</maven.compiler.source>
-		<apache.pulsar.client.version>2.5.1</apache.pulsar.client.version>
+		<apache.pulsar.client.version>2.6.0</apache.pulsar.client.version>
+		<testcontainers.version>1.14.3</testcontainers.version>
 		<junit-jupiter.version>5.6.2</junit-jupiter.version>
+		<awaitility.version>4.0.3</awaitility.version>
 		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
 	</properties>
 
@@ -60,6 +62,35 @@
 			<version>${junit-jupiter.version}</version>
 			<scope>test</scope>
 		</dependency>
+
+        <!--	Pulsar integration tests	-->
+		<dependency>
+			<groupId>org.testcontainers</groupId>
+			<artifactId>testcontainers</artifactId>
+			<version>${testcontainers.version}</version>
+			<scope>test</scope>
+		</dependency>
+
+		<dependency>
+			<groupId>org.testcontainers</groupId>
+			<artifactId>junit-jupiter</artifactId>
+			<version>${testcontainers.version}</version>
+			<scope>test</scope>
+		</dependency>
+
+		<dependency>
+			<groupId>org.testcontainers</groupId>
+			<artifactId>pulsar</artifactId>
+			<version>${testcontainers.version}</version>
+			<scope>test</scope>
+		</dependency>
+
+		<dependency>
+			<groupId>org.awaitility</groupId>
+			<artifactId>awaitility</artifactId>
+			<version>${awaitility.version}</version>
+			<scope>test</scope>
+		</dependency>
 	</dependencies>
 
 	<build>
@@ -98,7 +129,6 @@
 						<exclude>**/*Properties.*</exclude>
 						<exclude>**/*Configuration.*</exclude>
 						<exclude>**/*Application.*</exclude>
-						<exclude>**/mock/**/*</exclude>
 					</excludes>
 				</configuration>
 				<executions>

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

@@ -48,7 +48,7 @@ private Consumer<?> subscribe(String name, ConsumerHolder holder) {
                         final Method method = holder.getHandler();
 
                         method.setAccessible(true);
-                        method.invoke(holder.getBean(), msg);
+                        method.invoke(holder.getBean(), msg.getValue());
 
                         consumer.acknowledge(msg);
                     } catch (Exception e) {

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

@@ -6,7 +6,14 @@ public MyMsg(String data) {
         this.data = data;
     }
 
+    public MyMsg() {
+    }
+
     public String getData() {
         return data;
     }
+
+    public void setData(String data) {
+        this.data = data;
+    }
 }

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

@@ -14,14 +14,22 @@
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.context.annotation.Import;
+import org.springframework.test.context.DynamicPropertyRegistry;
+import org.springframework.test.context.DynamicPropertySource;
+import org.testcontainers.containers.PulsarContainer;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
 
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 
+import static org.awaitility.Awaitility.await;
+
 @SpringBootTest
-@Import({TestProducerConfiguration.class, TestConsumerConfiguration.class})
+@Import({TestProducerConfiguration.class, TestConsumers.class})
+@Testcontainers
 class PulsarJavaSpringBootStarterApplicationTests {
 
     @Autowired
@@ -36,9 +44,22 @@ class PulsarJavaSpringBootStarterApplicationTests {
     @Autowired
     private PulsarTemplate<MyMsg> producer;
 
+    @Container
+    static PulsarContainer pulsarContainer = new PulsarContainer();
+
+    @Autowired
+    private TestConsumers testConsumers;
+
+    @DynamicPropertySource
+    static void propertySettings(DynamicPropertyRegistry registry) {
+        registry.add("pulsar.serviceUrl", pulsarContainer::getPulsarBrokerUrl);
+    }
+
     @Test
     void testProducerSendMethod() throws PulsarClientException {
         producer.send("topic-one", new MyMsg("bb"));
+
+        await().untilTrue(testConsumers.mockTopicListenerReceived);
     }
 
     @Test
@@ -50,19 +71,19 @@ void testConsumerRegistration1() throws Exception {
         final Consumer consumer = consumers.stream().findFirst().orElseThrow(Exception::new);
 
         Assertions.assertNotNull(consumer);
-        Assertions.assertEquals("mock-topic", consumer.getTopic());
+        Assertions.assertEquals("topic-one", consumer.getTopic());
     }
 
     @Test
     void testConsumerRegistration2() {
-        final Class<TestConsumerConfiguration> clazz = TestConsumerConfiguration.class;
+        final Class<TestConsumers> clazz = TestConsumers.class;
         final String descriptor = clazz.getName() + clazz.getMethods()[0].getName();
         final ConsumerHolder consumerHolder = consumerCollector.getConsumer(descriptor).orElse(null);
 
         Assertions.assertNotNull(consumerHolder);
-        Assertions.assertEquals("mock-topic", consumerHolder.getAnnotation().topic());
-        Assertions.assertEquals(TestConsumerConfiguration.class, consumerHolder.getBean().getClass());
-        Assertions.assertEquals("mockTheListener", consumerHolder.getHandler().getName());
+        Assertions.assertEquals("topic-one", consumerHolder.getAnnotation().topic());
+        Assertions.assertEquals(TestConsumers.class, consumerHolder.getBean().getClass());
+        Assertions.assertEquals("topicOneListener", consumerHolder.getHandler().getName());
     }
 
     @Test

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

@@ -0,0 +1,20 @@
+package io.github.majusko.pulsar;
+
+import io.github.majusko.pulsar.annotation.PulsarConsumer;
+import io.github.majusko.pulsar.constant.Serialization;
+import org.junit.jupiter.api.Assertions;
+import org.springframework.stereotype.Service;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+
+@Service
+public class TestConsumers {
+
+    public AtomicBoolean mockTopicListenerReceived = new AtomicBoolean(false);
+
+    @PulsarConsumer(topic = "topic-one", clazz = MyMsg.class, serialization = Serialization.JSON)
+    public void topicOneListener(MyMsg myMsg) {
+        Assertions.assertNotNull(myMsg);
+        mockTopicListenerReceived.set(true);
+    }
+}