[Samples] Make sample-pulsar-binder testable

Author: onobc

spring-pulsar-sample-apps/sample-pulsar-binder/build.gradle

@@ -22,6 +22,15 @@ ext['pulsar.version'] = "${pulsarVersion}"
 dependencies {
 	implementation "org.springframework.boot:spring-boot-starter-pulsar"
 	implementation "org.springframework.cloud:spring-cloud-stream-binder-pulsar:${springCloudStreamVersion}"
+	developmentOnly 'org.springframework.boot:spring-boot-docker-compose'
+
+	testImplementation project(':spring-pulsar-test')
+	testRuntimeOnly 'ch.qos.logback:logback-classic'
+	testImplementation "org.springframework.boot:spring-boot-starter-test"
+	testImplementation "org.springframework.boot:spring-boot-testcontainers"
+	testImplementation 'org.testcontainers:junit-jupiter'
+	testImplementation 'org.testcontainers:pulsar'
+
 }
 
 test {
@@ -36,4 +45,6 @@ bootRun {
 			"--add-opens", "java.base/java.util=ALL-UNNAMED",
 			"--add-opens", "java.base/sun.net=ALL-UNNAMED"
 	]
+	// when run from command line, path must be set relative to module dir
+	systemProperty 'spring.docker.compose.file', 'compose.yaml'
 }

spring-pulsar-sample-apps/sample-pulsar-binder/compose.yaml

@@ -0,0 +1,7 @@
+services:
+  pulsar:
+    image: 'apachepulsar/pulsar:3.1.2'
+    ports:
+      - '6650'
+      - '8080'
+    command: 'bin/pulsar standalone'

spring-pulsar-sample-apps/sample-pulsar-binder/src/main/java/com/example/SpringPulsarBinderSampleApp.java

@@ -0,0 +1,74 @@
+/*
+ * Copyright 2022-2024 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.example;
+
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
+import java.util.function.Function;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.springframework.boot.ApplicationRunner;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.cloud.stream.function.StreamBridge;
+import org.springframework.context.annotation.Bean;
+
+@SpringBootApplication
+public class SpringPulsarBinderSampleApp {
+
+	private static final Logger LOG = LoggerFactory.getLogger(SpringPulsarBinderSampleApp.class);
+
+	public static void main(String[] args) {
+		SpringApplication.run(SpringPulsarBinderSampleApp.class, args);
+	}
+
+	private AtomicInteger counter = new AtomicInteger();
+
+	@Bean
+	ApplicationRunner fooSupplier(StreamBridge streamBridge) {
+		return (args) -> {
+			for (int i = 0; i < 10; i++) {
+				var foo = new Foo("fooSupplier:" + i);
+				streamBridge.send("fooSupplier-out-0", foo);
+				LOG.info("++++++SOURCE {}------", foo);
+			}
+		};
+	}
+
+	@Bean
+	public Function<Foo, Bar> fooProcessor() {
+		return (foo) -> {
+			var bar = new Bar(foo);
+			LOG.info("++++++PROCESSOR {} --> {}------", foo, bar);
+			return bar;
+		};
+	}
+
+	@Bean
+	public Consumer<Bar> barLogger() {
+		return (bar) -> LOG.info("++++++SINK {}------", bar);
+	}
+
+	record Foo(String value) {
+	}
+
+	record Bar(Foo value) {
+	}
+
+}

…k/pulsar/sample/binder/package-info.java …/main/java/com/example/package-info.javaspring-pulsar-sample-apps/sample-pulsar-binder/src/main/java/org/springframework/pulsar/sample/binder/package-info.java renamed to spring-pulsar-sample-apps/sample-pulsar-binder/src/main/java/com/example/package-info.java spring-pulsar-sample-apps/sample-pulsar-binder/src/main/java/org/springframework/pulsar/sample/binder/package-info.java renamed to spring-pulsar-sample-apps/sample-pulsar-binder/src/main/java/com/example/package-info.java

@@ -3,7 +3,7 @@
  */
 @NonNullApi
 @NonNullFields
-package org.springframework.pulsar.sample.binder;
+package com.example;
 
 import org.springframework.lang.NonNullApi;
 import org.springframework.lang.NonNullFields;

spring-pulsar-sample-apps/sample-pulsar-binder/src/main/java/org/springframework/pulsar/sample/binder/SpringPulsarBinderSampleApp.java

@@ -1,52 +1,44 @@
 spring:
   cloud:
     function:
-      definition: timeSupplier;timeProcessor;timeLogger
+      definition: fooProcessor;barLogger
     stream:
+      output-bindings: fooSupplier-out-0
       bindings:
-        timeSupplier-out-0:
+        fooSupplier-out-0:
           producer:
             use-native-encoding: true
-        timeProcessor-in-0:
-          destination: timeSupplier-out-0
+        fooProcessor-in-0:
+          destination: fooSupplier-out-0
           consumer:
             use-native-decoding: true
-        timeProcessor-out-0:
-          destination: timeProcessor-out-0
+        fooProcessor-out-0:
+          destination: fooProcessor-out-0
           producer:
             use-native-encoding: true
-        timeLogger-in-0:
-          destination: timeProcessor-out-0
-          consumer:
-            use-native-decoding: true
-        timeLoggerToDlt-in-0:
-          destination: timeProcessor-out-0
+        barLogger-in-0:
+          destination: fooProcessor-out-0
           consumer:
             use-native-decoding: true
       pulsar:
         bindings:
-          timeSupplier-out-0:
+          fooSupplier-out-0:
             producer:
               schema-type: JSON
-              message-type: org.springframework.pulsar.sample.binder.SpringPulsarBinderSampleApp.Time
-          timeProcessor-in-0:
+              message-type: com.example.SpringPulsarBinderSampleApp.Foo
+          fooProcessor-in-0:
             consumer:
               schema-type: JSON
-              message-type: org.springframework.pulsar.sample.binder.SpringPulsarBinderSampleApp.Time
-          timeProcessor-out-0:
+              message-type: com.example.SpringPulsarBinderSampleApp.Foo
+          fooProcessor-out-0:
             producer:
               schema-type: JSON
-              message-type: org.springframework.pulsar.sample.binder.SpringPulsarBinderSampleApp.EnhancedTime
-          timeLogger-in-0:
-            consumer:
-              schema-type: JSON
-              message-type: org.springframework.pulsar.sample.binder.SpringPulsarBinderSampleApp.EnhancedTime
-          timeLoggerToDlt-in-0:
+              message-type: com.example.SpringPulsarBinderSampleApp.Bar
+          barLogger-in-0:
             consumer:
-              subscription-type: Shared
-              negative-ack-redelivery-delay: 1s
-              dead-letter-policy:
-                dead-letter-topic: notification-dlq
-                max-redeliver-count: 5
               schema-type: JSON
-              message-type: org.springframework.pulsar.sample.binder.SpringPulsarBinderSampleApp.EnhancedTime
+              message-type: com.example.SpringPulsarBinderSampleApp.Bar
+  docker:
+    compose:
+      # when run from Intellij via "Run" button, path must be set from project root
+      file: spring-pulsar-sample-apps/sample-pulsar-binder/compose.yaml

spring-pulsar-sample-apps/sample-pulsar-binder/src/main/resources/application.yml

@@ -0,0 +1,62 @@
+/*
+ * Copyright 2012-2024 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.example;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.stream.IntStream;
+
+import com.example.SpringPulsarBinderSampleApp.Bar;
+import com.example.SpringPulsarBinderSampleApp.Foo;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.system.CapturedOutput;
+import org.springframework.boot.test.system.OutputCaptureExtension;
+import org.springframework.pulsar.test.support.PulsarTestContainerSupport;
+import org.springframework.test.context.DynamicPropertyRegistry;
+import org.springframework.test.context.DynamicPropertySource;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@SpringBootTest
+@ExtendWith(OutputCaptureExtension.class)
+class SpringPulsarBinderSampleAppTests implements PulsarTestContainerSupport {
+
+	@DynamicPropertySource
+	static void pulsarProperties(DynamicPropertyRegistry registry) {
+		registry.add("spring.pulsar.client.service-url", PULSAR_CONTAINER::getPulsarBrokerUrl);
+		registry.add("spring.pulsar.admin.service-url", PULSAR_CONTAINER::getHttpServiceUrl);
+	}
+
+	@Test
+	void produceConsumeWithPrimitiveMessageType(CapturedOutput output) {
+		var expectedOutput = new ArrayList<String>();
+		IntStream.range(0, 10).forEachOrdered((i) -> {
+			var foo = new Foo("fooSupplier:" + i);
+			var bar = new Bar(foo);
+			expectedOutput.add("++++++SOURCE %s------".formatted(foo));
+			expectedOutput.add("++++++PROCESSOR %s --> %s------".formatted(foo, bar));
+			expectedOutput.add("++++++SINK %s------".formatted(bar));
+		});
+		Awaitility.waitAtMost(Duration.ofSeconds(15))
+				.untilAsserted(() -> assertThat(output).contains(expectedOutput));
+	}
+
+}

spring-pulsar-sample-apps/sample-pulsar-binder/src/test/java/com/example/SpringPulsarBinderSampleAppTests.java

@@ -0,0 +1,14 @@
+<configuration>
+    <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
+        <encoder>
+            <pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger - %msg%n</pattern>
+        </encoder>
+    </appender>
+    <root level="WARN">
+        <appender-ref ref="STDOUT"/>
+    </root>
+    <logger name="com.example" level="INFO"/>
+    <logger name="com.github.dockerjava" level="ERROR"/>
+    <logger name="org.apache.pulsar.common.util.netty" level="ERROR" />
+    <logger name="org.testcontainers" level="ERROR"/>
+</configuration>