Docs and tests for amqps (SSL) integration.

buildSrc/src/main/groovy/io.micronaut.build.internal.rabbitmq-examples.gradle

@@ -10,7 +10,7 @@ dependencies {
     testImplementation mnSerde.micronaut.serde.support
     testImplementation mnSerde.micronaut.serde.jackson
     testImplementation libs.awaitility
-    testImplementation 'io.micronaut.testresources:micronaut-test-resources-client'
+    testImplementation mnTestResources.micronaut.test.resources.client
 }
 configurations.all {
     resolutionStrategy.dependencySubstitution {

settings.gradle

@@ -18,6 +18,8 @@ include 'docs-examples:example-groovy'
 include 'docs-examples:example-java'
 include 'docs-examples:example-kotlin'
 
+include 'tests:rabbitmq-ssl'
+
 enableFeaturePreview 'TYPESAFE_PROJECT_ACCESSORS'
 
 micronautBuild {

tests/rabbitmq-ssl/build.gradle.kts

@@ -0,0 +1,24 @@
+plugins {
+    groovy
+    id("io.micronaut.build.internal.rabbitmq-base")
+    id("io.micronaut.minimal.application")
+}
+
+dependencies {
+    testImplementation(projects.micronautRabbitmq)
+
+    testImplementation(mnSerde.micronaut.serde.support)
+    testImplementation(mnSerde.micronaut.serde.jackson)
+    testImplementation(mn.micronaut.management)
+    testImplementation(mn.reactor)
+
+    testImplementation(mnTest.micronaut.test.spock)
+    testImplementation(mnTestResources.testcontainers.core)
+    testImplementation(mnTestResources.testcontainers.rabbitmq)
+    testImplementation(libs.awaitility)
+}
+
+micronaut {
+    version(libs.versions.micronaut.platform.get())
+    testRuntime("junit5")
+}

tests/rabbitmq-ssl/src/test/groovy/io/micronaut/rabbitmq/ssl/AbstractRabbitMQSSLTest.groovy

@@ -0,0 +1,77 @@
+package io.micronaut.rabbitmq.ssl
+
+import com.rabbitmq.client.ConnectionFactory
+import io.micronaut.context.ApplicationContext
+import org.slf4j.Logger
+import org.slf4j.LoggerFactory
+import org.testcontainers.containers.RabbitMQContainer
+import spock.lang.AutoCleanup
+import spock.lang.Shared
+import spock.lang.Specification
+import spock.util.concurrent.PollingConditions
+
+import javax.net.ssl.KeyManagerFactory
+import javax.net.ssl.SSLContext
+import javax.net.ssl.TrustManagerFactory
+import java.security.KeyStore
+
+abstract class AbstractRabbitMQSSLTest extends Specification {
+    private static final Logger log = LoggerFactory.getLogger(AbstractRabbitMQSSLTest)
+
+    private static final int AMQP_PORT = 5672
+    private static final int AMQPS_PORT = 5671
+
+    public static final String RABBIT_CONTAINER_VERSION = "3.13.1"
+
+    @Shared
+    @AutoCleanup
+    RabbitMQContainer rabbitContainer = new RabbitMQContainer("rabbitmq:" + RABBIT_CONTAINER_VERSION)
+
+    def setupSpec() {
+        rabbitContainer.start()
+    }
+
+    protected ApplicationContext applicationContext
+    protected PollingConditions conditions = new PollingConditions(timeout: 5)
+
+    protected void startContext(Map additionalConfig = [:]) {
+        applicationContext = ApplicationContext.run(
+                ["rabbitmq.port": rabbitContainer.getMappedPort(AMQPS_PORT),
+                 "spec.name": getClass().simpleName] << additionalConfig, "test")
+    }
+
+    protected void waitFor(Closure<?> conditionEvaluator) {
+        conditions.eventually conditionEvaluator
+    }
+
+    void cleanup() {
+        applicationContext?.close()
+    }
+
+    // TODO this is example from rabbitmq TLS docs to integrate, and change test to use the connection
+    // https://www.rabbitmq.com/ssl.html#java-client-connecting-with-peer-verification
+    void configureForSsl(ConnectionFactory factory) {
+        char[] keyPassphrase = "MySecretPassword".toCharArray()
+        KeyStore ks = KeyStore.getInstance("PKCS12")
+        ks.load(new FileInputStream("/path/to/client_key.p12"), keyPassphrase)
+
+        KeyManagerFactory kmf = KeyManagerFactory.getInstance("SunX509")
+        kmf.init(ks, keyPassphrase)
+
+        char[] trustPassphrase = "rabbitstore".toCharArray()
+        KeyStore tks = KeyStore.getInstance("JKS")
+        tks.load(new FileInputStream("/path/to/trustStore"), trustPassphrase)
+
+        TrustManagerFactory tmf = TrustManagerFactory.getInstance("SunX509")
+        tmf.init(tks)
+
+        SSLContext sslContext = SSLContext.getInstance("TLSv1.2")
+        sslContext.init(kmf.getKeyManagers(), tmf.getTrustManagers(), null)
+
+        factory.setHost("localhost")
+        factory.setPort(AMQPS_PORT)
+        factory.enableHostnameVerification()
+        // this is the key part, using no-arg factory.useSslProtocol() is not adequate for prod use
+        factory.useSslProtocol(sslContext)
+    }
+}

tests/rabbitmq-ssl/src/test/groovy/io/micronaut/rabbitmq/ssl/RabbitSSLHealthIndicatorSpec.groovy

@@ -0,0 +1,60 @@
+package io.micronaut.rabbitmq.ssl
+
+import io.micronaut.context.ApplicationContext
+import io.micronaut.health.HealthStatus
+import io.micronaut.management.health.indicator.HealthResult
+import io.micronaut.rabbitmq.health.RabbitMQHealthIndicator
+import reactor.core.publisher.Mono
+
+class RabbitSSLHealthIndicatorSpec extends AbstractRabbitMQSSLTest {
+
+    void "test rabbitmq health indicator"() {
+        given:
+        startContext()
+
+        when:
+        RabbitMQHealthIndicator healthIndicator = applicationContext.getBean(RabbitMQHealthIndicator)
+        HealthResult result = Mono.from(healthIndicator.result).block()
+        def details = result.details
+
+        then:
+        result.status == HealthStatus.UP
+        details.version.toString() == RABBIT_CONTAINER_VERSION
+    }
+
+    void "test rabbitmq health indicator with 2 connections"() {
+        given:
+        applicationContext = ApplicationContext.run([
+                "rabbitmq.servers.one.port": rabbitContainer.getMappedPort(5672),
+                "rabbitmq.servers.two.port": rabbitContainer.getMappedPort(5672)
+        ], "test")
+
+        when:
+        RabbitMQHealthIndicator healthIndicator = applicationContext.getBean(RabbitMQHealthIndicator)
+        HealthResult result = Mono.from(healthIndicator.result).block()
+        def details = result.details
+
+        then:
+        result.status == HealthStatus.UP
+        details.get("connections")[0].get("version").toString() == RABBIT_CONTAINER_VERSION
+        details.get("connections")[1].get("version").toString() == RABBIT_CONTAINER_VERSION
+    }
+
+    void "test rabbitmq health indicator shows down"() {
+        given:
+        startContext()
+
+        when:
+        RabbitMQHealthIndicator healthIndicator = applicationContext.getBean(RabbitMQHealthIndicator)
+        rabbitContainer.stop()
+        HealthResult result = Mono.from(healthIndicator.result).block()
+        def details = result.details
+
+        then:
+        result.status == HealthStatus.DOWN
+        details.get("error").toString().contains("RabbitMQ connection is not open")
+
+        cleanup:
+        rabbitContainer.start()
+    }
+}

tests/rabbitmq-ssl/src/test/groovy/io/micronaut/rabbitmq/ssl/exchange/Animal.groovy

@@ -0,0 +1,12 @@
+package io.micronaut.rabbitmq.ssl.exchange
+
+abstract class Animal {
+
+    String name
+
+    Animal(String name) {
+        this.name = name
+    }
+
+    Animal() {}
+}

tests/rabbitmq-ssl/src/test/groovy/io/micronaut/rabbitmq/ssl/exchange/AnimalClient.groovy

@@ -0,0 +1,16 @@
+package io.micronaut.rabbitmq.ssl.exchange
+
+import io.micronaut.context.annotation.Requires
+import io.micronaut.messaging.annotation.MessageHeader
+import io.micronaut.rabbitmq.annotation.RabbitClient
+
+@Requires(property = "spec.name", value = "CustomSSLExchangeSpec")
+@RabbitClient("animals")
+abstract class AnimalClient {
+
+    abstract void send(@MessageHeader String animalType, Animal animal)
+
+    void send(Animal animal) {
+        send(animal.getClass().simpleName, animal)
+    }
+}

tests/rabbitmq-ssl/src/test/groovy/io/micronaut/rabbitmq/ssl/exchange/AnimalListener.groovy

@@ -0,0 +1,23 @@
+package io.micronaut.rabbitmq.ssl.exchange
+
+import io.micronaut.context.annotation.Requires
+import io.micronaut.rabbitmq.annotation.Queue
+import io.micronaut.rabbitmq.annotation.RabbitListener
+import java.util.concurrent.CopyOnWriteArrayList
+
+@Requires(property = "spec.name", value = "CustomSSLExchangeSpec")
+@RabbitListener
+class AnimalListener {
+
+    CopyOnWriteArrayList<Animal> receivedAnimals = []
+
+    @Queue("cats")
+    void receive(Cat cat) {
+        receivedAnimals << cat
+    }
+
+    @Queue("snakes")
+    void receive(Snake snake) {
+        receivedAnimals << snake
+    }
+}

tests/rabbitmq-ssl/src/test/groovy/io/micronaut/rabbitmq/ssl/exchange/Cat.groovy

@@ -0,0 +1,16 @@
+package io.micronaut.rabbitmq.ssl.exchange
+
+import io.micronaut.serde.annotation.Serdeable
+
+@Serdeable
+class Cat extends Animal {
+
+    int lives
+
+    Cat(String name, int lives) {
+        super(name)
+        this.lives = lives
+    }
+
+    Cat() {}
+}

tests/rabbitmq-ssl/src/test/groovy/io/micronaut/rabbitmq/ssl/exchange/ChannelPoolListener.groovy

@@ -0,0 +1,29 @@
+package io.micronaut.rabbitmq.ssl.exchange
+
+import com.rabbitmq.client.Channel
+import io.micronaut.context.annotation.Requires
+import io.micronaut.rabbitmq.connect.ChannelInitializer
+import jakarta.inject.Singleton
+
+@Requires(property = "spec.name", value = "CustomSSLExchangeSpec")
+@Singleton
+class ChannelPoolListener extends ChannelInitializer {
+
+    @Override
+    void initialize(Channel channel, String name) throws IOException {
+
+        channel.exchangeDeclare("animals", "headers", false)
+        channel.queueDeclare("snakes", false, false, false, null)
+        channel.queueDeclare("cats", false, false, false, null)
+
+        Map<String, Object> catArgs = [:]
+        catArgs.put("x-match", "all")
+        catArgs.put("animalType", "Cat")
+        channel.queueBind("cats", "animals", "", catArgs)
+
+        Map<String, Object> snakeArgs = [:]
+        snakeArgs.put("x-match", "all")
+        snakeArgs.put("animalType", "Snake")
+        channel.queueBind("snakes", "animals", "", snakeArgs)
+    }
+}

tests/rabbitmq-ssl/src/test/groovy/io/micronaut/rabbitmq/ssl/exchange/CustomSSLExchangeSpec.groovy

@@ -0,0 +1,46 @@
+package io.micronaut.rabbitmq.ssl.exchange
+
+import io.micronaut.context.annotation.Property
+import io.micronaut.rabbitmq.ssl.AbstractRabbitMQSSLTest
+
+import static java.util.concurrent.TimeUnit.SECONDS
+import static org.awaitility.Awaitility.await
+
+@Property(name = "spec.name", value = "CustomSSLExchangeSpec")
+class CustomSSLExchangeSpec extends AbstractRabbitMQSSLTest {
+
+    void "test using a custom exchange"() {
+        given:
+        startContext()
+        AnimalClient client = applicationContext.getBean(AnimalClient)
+        AnimalListener listener = applicationContext.getBean(AnimalListener)
+
+        when:
+        client.send(new Cat("Whiskers", 9))
+        client.send(new Cat("Mr. Bigglesworth", 8))
+        client.send(new Snake("Buttercup", false))
+        client.send(new Snake("Monty the Python", true))
+        await().atMost(10, SECONDS).until {
+            listener.receivedAnimals.size() == 4
+        }
+
+        then:
+        assert listener.receivedAnimals.size() == 4
+
+        assert listener.receivedAnimals.find({ animal ->
+            animal instanceof Cat && animal.name == "Whiskers" && ((Cat) animal).lives == 9
+        }) != null
+
+        assert listener.receivedAnimals.find({ animal ->
+            animal instanceof Cat && animal.name == "Mr. Bigglesworth" && ((Cat) animal).lives == 8
+        }) != null
+
+        assert listener.receivedAnimals.find({ animal ->
+            animal instanceof Snake && animal.name == "Buttercup" && !((Snake) animal).venomous
+        }) != null
+
+        assert listener.receivedAnimals.find({ animal ->
+            animal instanceof Snake && animal.name == "Monty the Python" && ((Snake) animal).venomous
+        }) != null
+    }
+}

tests/rabbitmq-ssl/src/test/groovy/io/micronaut/rabbitmq/ssl/exchange/Snake.groovy

@@ -0,0 +1,16 @@
+package io.micronaut.rabbitmq.ssl.exchange
+
+import io.micronaut.serde.annotation.Serdeable
+
+@Serdeable
+class Snake extends Animal {
+
+    boolean venomous
+
+    Snake(String name, boolean venomous) {
+        super(name)
+        this.venomous = venomous
+    }
+
+    Snake() {}
+}

tests/rabbitmq-ssl/src/test/resources/certs/README.md

@@ -0,0 +1,34 @@
+These long-lived certificates were created to test the SSL, they are in no way secure.
+
+To create them, [I followed the instructions here](https://www.rabbitmq.com/ssl.html#automated-certificate-generation-transcript) and [here](https://github.com/rabbitmq/tls-gen):
+
+Which are:
+
+```
+git clone https://github.com/rabbitmq/tls-gen tls-gen
+```
+
+then:
+
+```
+cd tls-gen/basic
+```
+
+then:
+
+```
+# pass a private key password using the PASSWORD variable if needed
+make
+
+## copy or move files to use hostname-neutral filenames
+## such as client_certificate.pem and client_key.pem,
+## this step is optional
+make alias-leaf-artifacts
+```
+
+with output in
+
+```
+# results will be under the ./result directory
+ls -lha ./result
+```