diff --git a/spring-pulsar-sample-apps/sample-pulsar-binder/build.gradle b/spring-pulsar-sample-apps/sample-pulsar-binder/build.gradle index eac7e653..2f8eaa6a 100644 --- a/spring-pulsar-sample-apps/sample-pulsar-binder/build.gradle +++ b/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' } diff --git a/spring-pulsar-sample-apps/sample-pulsar-binder/compose.yaml b/spring-pulsar-sample-apps/sample-pulsar-binder/compose.yaml new file mode 100644 index 00000000..39df55a5 --- /dev/null +++ b/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' diff --git a/spring-pulsar-sample-apps/sample-pulsar-binder/src/main/java/com/example/SpringPulsarBinderSampleApp.java b/spring-pulsar-sample-apps/sample-pulsar-binder/src/main/java/com/example/SpringPulsarBinderSampleApp.java new file mode 100644 index 00000000..943ffb61 --- /dev/null +++ b/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 fooProcessor() { + return (foo) -> { + var bar = new Bar(foo); + LOG.info("++++++PROCESSOR {} --> {}------", foo, bar); + return bar; + }; + } + + @Bean + public Consumer barLogger() { + return (bar) -> LOG.info("++++++SINK {}------", bar); + } + + record Foo(String value) { + } + + record Bar(Foo value) { + } + +} diff --git a/spring-pulsar-sample-apps/sample-pulsar-binder/src/main/java/org/springframework/pulsar/sample/binder/package-info.java b/spring-pulsar-sample-apps/sample-pulsar-binder/src/main/java/com/example/package-info.java similarity index 78% rename from spring-pulsar-sample-apps/sample-pulsar-binder/src/main/java/org/springframework/pulsar/sample/binder/package-info.java rename to spring-pulsar-sample-apps/sample-pulsar-binder/src/main/java/com/example/package-info.java index e2d5e948..5f2a90d1 100644 --- a/spring-pulsar-sample-apps/sample-pulsar-binder/src/main/java/org/springframework/pulsar/sample/binder/package-info.java +++ b/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; diff --git a/spring-pulsar-sample-apps/sample-pulsar-binder/src/main/java/org/springframework/pulsar/sample/binder/SpringPulsarBinderSampleApp.java b/spring-pulsar-sample-apps/sample-pulsar-binder/src/main/java/org/springframework/pulsar/sample/binder/SpringPulsarBinderSampleApp.java deleted file mode 100644 index 4b1ee9e5..00000000 --- a/spring-pulsar-sample-apps/sample-pulsar-binder/src/main/java/org/springframework/pulsar/sample/binder/SpringPulsarBinderSampleApp.java +++ /dev/null @@ -1,104 +0,0 @@ -/* - * Copyright 2022-2023 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 org.springframework.pulsar.sample.binder; - -import java.util.function.Consumer; -import java.util.function.Function; -import java.util.function.Supplier; - -import org.apache.pulsar.common.schema.SchemaType; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import org.springframework.boot.SpringApplication; -import org.springframework.boot.autoconfigure.SpringBootApplication; -import org.springframework.context.annotation.Bean; -import org.springframework.pulsar.annotation.PulsarListener; - -/** - * This sample binder app has an extra consumer that is equipped with Pulsar's DLT feature - * - timeLoggerToDlt. However, this consumer is not part of the - * spring.cloud.function.definition. In order to enable this, add the function - * timeLoggerToDlt to the definition in the application.yml file. When doing this, in - * order to minimize verbose output and just to focus on the DLT feature, comment out the - * regular supplier below (timeSupplier) and then un-comment the ApplicationRunner below. - * The runner only sends a single message whereas the supplier sends a message every - * second. - * - * @author Soby Chacko - */ -@SpringBootApplication -public class SpringPulsarBinderSampleApp { - - private final Logger logger = LoggerFactory.getLogger(this.getClass()); - - public static void main(String[] args) { - SpringApplication.run(SpringPulsarBinderSampleApp.class, args); - } - - @Bean - public Supplier