diff --git a/.gitignore b/.gitignore index 06f7647..5195982 100644 --- a/.gitignore +++ b/.gitignore @@ -9,3 +9,4 @@ out/ ci/pipeline.yml stats-restore .vscode/ +.DS_Store diff --git a/integration/spring-amqp-rabbit/README.adoc b/integration/spring-amqp-rabbit/README.adoc new file mode 100644 index 0000000..2b4c1a6 --- /dev/null +++ b/integration/spring-amqp-rabbit/README.adoc @@ -0,0 +1 @@ +Tests if Spring AMQP with RabbitMQ is working. diff --git a/integration/spring-amqp-rabbit/build.gradle b/integration/spring-amqp-rabbit/build.gradle new file mode 100644 index 0000000..d2a2f1a --- /dev/null +++ b/integration/spring-amqp-rabbit/build.gradle @@ -0,0 +1,21 @@ +plugins { + id "java" + id "org.springframework.boot" + id "org.springframework.cr.smoke-test" +} + +dependencies { + implementation(platform(org.springframework.boot.gradle.plugin.SpringBootPlugin.BOM_COORDINATES)) + implementation("org.springframework.boot:spring-boot-starter-amqp") + + // TODO: update to non snapshot spring-rabbit 3.0.8+ version when available. + implementation("org.springframework.amqp:spring-rabbit:3.0.8-SNAPSHOT") + + implementation("org.crac:crac:$cracVersion") + implementation(project(":cr-listener")) + + testImplementation("org.springframework.boot:spring-boot-starter-test") + + appTestImplementation(project(":cr-smoke-test-support")) + appTestImplementation("org.awaitility:awaitility:4.2.0") +} diff --git a/integration/spring-amqp-rabbit/docker-compose.yml b/integration/spring-amqp-rabbit/docker-compose.yml new file mode 100644 index 0000000..628ebf7 --- /dev/null +++ b/integration/spring-amqp-rabbit/docker-compose.yml @@ -0,0 +1,7 @@ +version: '3' +services: + rabbitmq: + image: 'rabbitmq:management' + ports: + - '5672' + - '15672' diff --git a/integration/spring-amqp-rabbit/src/appTest/java/com/example/amqp/AmqpRabbitApplicationCrTests.java b/integration/spring-amqp-rabbit/src/appTest/java/com/example/amqp/AmqpRabbitApplicationCrTests.java new file mode 100644 index 0000000..bcc7031 --- /dev/null +++ b/integration/spring-amqp-rabbit/src/appTest/java/com/example/amqp/AmqpRabbitApplicationCrTests.java @@ -0,0 +1,38 @@ +/* + * Copyright 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 com.example.amqp; + +import java.time.Duration; + +import org.awaitility.Awaitility; +import org.junit.jupiter.api.Test; + +import org.springframework.cr.smoketest.support.assertj.AssertableOutput; +import org.springframework.cr.smoketest.support.junit.ApplicationTest; + +import static org.assertj.core.api.Assertions.assertThat; + +@ApplicationTest +class AmqpRabbitApplicationCrTests { + + @Test + void rabbitListenerMethodReceivesMessageAndSendsResponse(AssertableOutput output) { + Awaitility.await() + .atMost(Duration.ofSeconds(60)) + .untilAsserted(() -> assertThat(output).hasLineContaining("++++++ Received: ONEtwo")); + } + +} diff --git a/integration/spring-amqp-rabbit/src/main/java/com/example/amqp/AmqpRabbitApplication.java b/integration/spring-amqp-rabbit/src/main/java/com/example/amqp/AmqpRabbitApplication.java new file mode 100644 index 0000000..c501627 --- /dev/null +++ b/integration/spring-amqp-rabbit/src/main/java/com/example/amqp/AmqpRabbitApplication.java @@ -0,0 +1,106 @@ +/* + * Copyright 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 com.example.amqp; + +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.springframework.amqp.AmqpRejectAndDontRequeueException; +import org.springframework.amqp.core.Queue; +import org.springframework.amqp.rabbit.annotation.RabbitListener; +import org.springframework.amqp.rabbit.connection.CorrelationData; +import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.context.SmartLifecycle; +import org.springframework.context.annotation.Bean; +import org.springframework.messaging.handler.annotation.SendTo; +import org.springframework.scheduling.annotation.EnableScheduling; +import org.springframework.scheduling.annotation.Scheduled; + +@SpringBootApplication +@EnableScheduling +public class AmqpRabbitApplication implements SmartLifecycle { + + public static void main(String[] args) throws Exception { + SpringApplication.run(AmqpRabbitApplication.class, args); + } + + @Autowired + RabbitTemplate template; + + @RabbitListener(id = "cf1", queues = "cf1") + @SendTo + public String upperCaseIt(String in) { + try { + String two = sendWithConfirms(); + return in.toUpperCase() + two; + } + catch (Exception e) { + throw new AmqpRejectAndDontRequeueException("fail"); + } + } + + @RabbitListener(id = "cf2", queues = "cf2") + @SendTo + public String lowerCaseIt(String in) { + return in.toLowerCase(); + } + + private String sendWithConfirms() throws Exception { + CorrelationData data = new CorrelationData(); + String result = (String) this.template.convertSendAndReceive("", "cf2", "TWO", data); + data.getFuture().get(10, TimeUnit.SECONDS); + return result; + } + + @Bean + public Queue queue1() { + return new Queue("cf1"); + } + + @Bean + public Queue queue2() { + return new Queue("cf2"); + } + + @Scheduled(fixedDelay = 1, timeUnit = TimeUnit.SECONDS) + public void doSendMessage() { + if (this.isRunning.get()) { + System.out.println("++++++ Received: " + template.convertSendAndReceive("", "cf1", "one")); + } + } + + // Use the SmartLifecycle to synch the message sending with the checkpoint/restore. + private AtomicBoolean isRunning = new AtomicBoolean(false); + + @Override + public void start() { + this.isRunning.set(true); + } + + @Override + public void stop() { + this.isRunning.set(false); + } + + @Override + public boolean isRunning() { + return this.isRunning.get(); + } + +} diff --git a/integration/spring-amqp-rabbit/src/main/resources/application.properties b/integration/spring-amqp-rabbit/src/main/resources/application.properties new file mode 100644 index 0000000..c3c3592 --- /dev/null +++ b/integration/spring-amqp-rabbit/src/main/resources/application.properties @@ -0,0 +1,3 @@ +spring.rabbitmq.host=${RABBITMQ_HOST:localhost} +spring.rabbitmq.port=${RABBITMQ_PORT_5672:5672} +spring.rabbitmq.publisher-confirm-type=correlated