<|start_filename|>docs-examples/example-java/src/test/java/io/micronaut/rabbitmq/docs/consumer/concurrent/ProductClient.java<|end_filename|>
package io.micronaut.rabbitmq.docs.consumer.concurrent;

import io.micronaut.context.annotation.Requires;
import io.micronaut.rabbitmq.annotation.Binding;
import io.micronaut.rabbitmq.annotation.RabbitClient;
// end::imports[]

@Requires(property = "spec.name", value = "ConcurrentSpec")
// tag::clazz[]
@RabbitClient // <1>
public interface ProductClient {

    @Binding("product") // <2>
    void send(byte[] data); // <3>
}
// end::clazz[]


<|start_filename|>docs-examples/example-java/src/test/java/io/micronaut/rabbitmq/docs/quickstart/QuickstartSpec.java<|end_filename|>
package io.micronaut.rabbitmq.docs.quickstart;

import io.micronaut.rabbitmq.AbstractRabbitMQTest;
import org.junit.jupiter.api.Test;

public class QuickstartSpec extends AbstractRabbitMQTest {

    @Test
    void testProductClientAndListener() {
        startContext();

// tag::producer[]
ProductClient productClient = applicationContext.getBean(ProductClient.class);
productClient.send("quickstart".getBytes());
// end::producer[]

        ProductListener productListener = applicationContext.getBean(ProductListener.class);

        waitFor(() ->
                productListener.messageLengths.size() == 1 &&
                productListener.messageLengths.get(0).equals("quickstart")
        );
    }
}


<|start_filename|>rabbitmq/src/test/groovy/io/micronaut/rabbitmq/bind/RabbitMessageHeadersSpec.groovy<|end_filename|>
package io.micronaut.rabbitmq.bind

import io.micronaut.context.annotation.Requires
import io.micronaut.rabbitmq.AbstractRabbitMQTest
import io.micronaut.rabbitmq.annotation.Binding
import io.micronaut.rabbitmq.annotation.Queue
import io.micronaut.rabbitmq.annotation.RabbitClient
import io.micronaut.rabbitmq.annotation.RabbitHeaders
import io.micronaut.rabbitmq.annotation.RabbitListener

class RabbitMessageHeadersSpec extends AbstractRabbitMQTest {

    void "test publishing and consuming with  MessageHeaders and RabbitHeaders"() {

        startContext()

        AnimalProducer producer = applicationContext.getBean(AnimalProducer)
        AnimalListener consumer = applicationContext.getBean(AnimalListener)

        when:
        producer.goWithRabbitHeaders("test", [weight: "2lbs", color: "black"])

        then:
        waitFor {
            consumer.headers == [weight: "2lbs", color: "black"]
        }
    }

    @Requires(property = "spec.name", value = "RabbitMessageHeadersSpec")
    @RabbitClient
    static interface AnimalProducer {
        @Binding("header")
        void goWithRabbitHeaders(String body, @RabbitHeaders Map<String, Object> rabbitHeaders)
    }

    @Requires(property = "spec.name", value = "RabbitMessageHeadersSpec")
    @RabbitListener
    static class AnimalListener {

        Map<String, Object> headers = [:]

        @Queue("header")
        void listen(@RabbitHeaders Map<String, Object> rabbitHeaders) {
            rabbitHeaders.each { k, v ->
                headers[k] = v.toString()
            }
        }
    }
}


<|start_filename|>docs-examples/example-java/src/test/java/io/micronaut/rabbitmq/docs/rpc/ProductClient.java<|end_filename|>
package io.micronaut.rabbitmq.docs.rpc;

import io.micronaut.context.annotation.Requires;
// tag::imports[]
import io.micronaut.rabbitmq.annotation.Binding;
import io.micronaut.rabbitmq.annotation.RabbitClient;
import io.micronaut.rabbitmq.annotation.RabbitProperty;
import org.reactivestreams.Publisher;
// end::imports[]

@Requires(property = "spec.name", value = "RpcUppercaseSpec")
// tag::clazz[]
@RabbitClient
@RabbitProperty(name = "replyTo", value = "amq.rabbitmq.reply-to") // <1>
public interface ProductClient {

    @Binding("product")
    String send(String data); // <2>

    @Binding("product")
    Publisher<String> sendReactive(String data); // <3>
}
// end::clazz[]


<|start_filename|>docs-examples/example-kotlin/src/test/kotlin/io/micronaut/rabbitmq/docs/headers/HeadersSpec.kt<|end_filename|>
package io.micronaut.rabbitmq.docs.headers

import io.kotest.assertions.timing.eventually
import io.kotest.matchers.collections.shouldContain
import io.kotest.matchers.shouldBe
import io.micronaut.rabbitmq.AbstractRabbitMQTest
import kotlin.time.Duration
import kotlin.time.ExperimentalTime

@OptIn(ExperimentalTime::class)
class HeadersSpec : AbstractRabbitMQTest({

    val specName = javaClass.simpleName

    given("A basic producer and consumer") {
        val ctx = startContext(specName)

        `when`("The messages are published") {
            val productListener = ctx.getBean(ProductListener::class.java)

            // tag::producer[]
            val productClient = ctx.getBean(ProductClient::class.java)
            productClient.send("body".toByteArray())
            productClient.send("medium", 20L, "body2".toByteArray())
            productClient.send(null, 30L, "body3".toByteArray())
            productClient.send(mapOf<String, Any>("productSize" to "large", "x-product-count" to 40L), "body4".toByteArray())
            // end::producer[]

            then("The messages are received") {
                eventually(Duration.seconds(10)) {
                    productListener.messageProperties.size shouldBe 4
                    productListener.messageProperties shouldContain "true|10|small"
                    productListener.messageProperties shouldContain "true|20|medium"
                    productListener.messageProperties shouldContain "true|30|null"
                    productListener.messageProperties shouldContain "true|40|large"
                }
            }
        }

        ctx.stop()
    }
})


<|start_filename|>docs-examples/example-java/src/test/java/io/micronaut/rabbitmq/docs/consumer/acknowledge/type/AcknowledgeSpec.java<|end_filename|>
package io.micronaut.rabbitmq.docs.consumer.acknowledge.type;

import io.micronaut.rabbitmq.AbstractRabbitMQTest;
import org.junit.jupiter.api.Test;

public class AcknowledgeSpec extends AbstractRabbitMQTest {

    @Test
    void testAckingWithAcknowledgement() {
        startContext();

// tag::producer[]
ProductClient productClient = applicationContext.getBean(ProductClient.class);
productClient.send("message body".getBytes());
productClient.send("message body".getBytes());
productClient.send("message body".getBytes());
productClient.send("message body".getBytes());
// end::producer[]

        ProductListener productListener = applicationContext.getBean(ProductListener.class);

        waitFor(() ->
                productListener.messageCount.get() == 5 // the first message is rejected and re-queued
        );
    }
}


<|start_filename|>rabbitmq/src/main/java/io/micronaut/rabbitmq/intercept/MutableBasicProperties.java<|end_filename|>
/*
 * Copyright 2017-2020 original 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 io.micronaut.rabbitmq.intercept;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.BasicProperties;

import java.util.Date;
import java.util.Map;

/**
 * A class that stores the basic properties in a mutable state.
 *
 * @author <NAME>
 * @since 1.1.0
 */
public class MutableBasicProperties implements BasicProperties {

    private String contentType;
    private String contentEncoding;
    private Map<String, Object> headers;
    private Integer deliveryMode;
    private Integer priority;
    private String correlationId;
    private String replyTo;
    private String expiration;
    private String messageId;
    private Date timestamp;
    private String type;
    private String userId;
    private String appId;
    private String clusterId;

    @Override
    public String getContentType() {
        return contentType;
    }

    /**
     * Sets the content type.
     *
     * @param contentType The content type
     */
    public void setContentType(String contentType) {
        this.contentType = contentType;
    }

    @Override
    public String getContentEncoding() {
        return contentEncoding;
    }

    /**
     * Sets the content encoding.
     *
     * @param contentEncoding The content encoding
     */
    public void setContentEncoding(String contentEncoding) {
        this.contentEncoding = contentEncoding;
    }

    @Override
    public Map<String, Object> getHeaders() {
        return headers;
    }

    /**
     * Sets the headers.
     *
     * @param headers The headers
     */
    public void setHeaders(Map<String, Object> headers) {
        this.headers = headers;
    }

    @Override
    public Integer getDeliveryMode() {
        return deliveryMode;
    }

    /**
     * Sets the delivery mode.
     *
     * @param deliveryMode The delivery mode
     */
    public void setDeliveryMode(Integer deliveryMode) {
        this.deliveryMode = deliveryMode;
    }

    @Override
    public Integer getPriority() {
        return priority;
    }

    /**
     * Sets the priority.
     *
     * @param priority The priority
     */
    public void setPriority(Integer priority) {
        this.priority = priority;
    }

    @Override
    public String getCorrelationId() {
        return correlationId;
    }

    /**
     * Sets the correlation id.
     *
     * @param correlationId The correlation id
     */
    public void setCorrelationId(String correlationId) {
        this.correlationId = correlationId;
    }

    @Override
    public String getReplyTo() {
        return replyTo;
    }

    /**
     * Sets the reply to.
     *
     * @param replyTo The reply to
     */
    public void setReplyTo(String replyTo) {
        this.replyTo = replyTo;
    }

    @Override
    public String getExpiration() {
        return expiration;
    }

    /**
     * Sets the expiration.
     *
     * @param expiration The expiration
     */
    public void setExpiration(String expiration) {
        this.expiration = expiration;
    }

    @Override
    public String getMessageId() {
        return messageId;
    }

    /**
     * Sets the message id.
     *
     * @param messageId The message id
     */
    public void setMessageId(String messageId) {
        this.messageId = messageId;
    }

    @Override
    public Date getTimestamp() {
        return timestamp;
    }

    /**
     * Sets the timestamp.
     *
     * @param timestamp The timestamp
     */
    public void setTimestamp(Date timestamp) {
        this.timestamp = timestamp;
    }

    @Override
    public String getType() {
        return type;
    }

    /**
     * Sets the type.
     *
     * @param type The type
     */
    public void setType(String type) {
        this.type = type;
    }

    @Override
    public String getUserId() {
        return userId;
    }

    /**
     * Sets the user id.
     *
     * @param userId The user id
     */
    public void setUserId(String userId) {
        this.userId = userId;
    }

    @Override
    public String getAppId() {
        return appId;
    }

    /**
     * Sets the app id.
     *
     * @param appId The app id
     */
    public void setAppId(String appId) {
        this.appId = appId;
    }

    /**
     * @return The cluster id
     */
    public String getClusterId() {
        return clusterId;
    }

    /**
     * Sets the cluster id.
     *
     * @param clusterId The cluster id
     */
    public void setClusterId(String clusterId) {
        this.clusterId = clusterId;
    }

    /**
     * Converts the mutable properties to the RabbitMQ properties.
     *
     * @return The RabbitMQ properties
     */
    public AMQP.BasicProperties toBasicProperties() {
        return new AMQP.BasicProperties.Builder()
                .contentType(contentType)
                .contentEncoding(contentEncoding)
                .headers(headers)
                .deliveryMode(deliveryMode)
                .priority(priority)
                .correlationId(correlationId)
                .replyTo(replyTo)
                .expiration(expiration)
                .messageId(messageId)
                .timestamp(timestamp)
                .type(type)
                .userId(userId)
                .appId(appId)
                .clusterId(clusterId)
                .build();
    }
}


<|start_filename|>docs-examples/example-groovy/src/test/groovy/io/micronaut/rabbitmq/docs/consumer/custom/annotation/DeliveryTagAnnotationBinder.groovy<|end_filename|>
package io.micronaut.rabbitmq.docs.consumer.custom.annotation

import io.micronaut.context.annotation.Requires
// tag::imports[]
import io.micronaut.core.convert.ArgumentConversionContext
import io.micronaut.core.convert.ConversionService
import io.micronaut.rabbitmq.bind.RabbitAnnotatedArgumentBinder
import io.micronaut.rabbitmq.bind.RabbitConsumerState
import jakarta.inject.Singleton
// end::imports[]

@Requires(property = "spec.name", value = "DeliveryTagSpec")
// tag::clazz[]
@Singleton // <1>
class DeliveryTagAnnotationBinder implements RabbitAnnotatedArgumentBinder<DeliveryTag> { // <2>

    private final ConversionService conversionService

    DeliveryTagAnnotationBinder(ConversionService conversionService) { // <3>
        this.conversionService = conversionService
    }

    @Override
    Class<DeliveryTag> getAnnotationType() {
        DeliveryTag
    }

    @Override
    BindingResult<Object> bind(ArgumentConversionContext<Object> context, RabbitConsumerState source) {
        long deliveryTag = source.envelope.deliveryTag // <4>
        return { -> conversionService.convert(deliveryTag, context) } // <5>
    }
}
// end::clazz[]


<|start_filename|>docs-examples/example-kotlin/src/test/kotlin/io/micronaut/rabbitmq/docs/exchange/AnimalClient.kt<|end_filename|>
package io.micronaut.rabbitmq.docs.exchange

import io.micronaut.context.annotation.Requires
// tag::imports[]
import io.micronaut.messaging.annotation.MessageHeader
import io.micronaut.rabbitmq.annotation.RabbitClient
// end::imports[]

@Requires(property = "spec.name", value = "CustomExchangeSpec")
// tag::clazz[]
@RabbitClient("animals") // <1>
abstract class AnimalClient {

    abstract fun send(@MessageHeader animalType: String, animal: Animal)  // <2>

    fun send(animal: Animal) { //<3>
        send(animal.javaClass.simpleName, animal)
    }
}
// end::clazz[]


<|start_filename|>docs-examples/example-java/src/test/java/io/micronaut/rabbitmq/docs/consumer/custom/annotation/DeliveryTagSpec.java<|end_filename|>
package io.micronaut.rabbitmq.docs.consumer.custom.annotation;

import io.micronaut.rabbitmq.AbstractRabbitMQTest;
import org.junit.jupiter.api.Test;

public class DeliveryTagSpec extends AbstractRabbitMQTest {

    @Test
    void testUsingACustomAnnotationBinder() {
        startContext();

// tag::producer[]
        ProductClient productClient = applicationContext.getBean(ProductClient.class);
        productClient.send("body".getBytes());
        productClient.send("body2".getBytes());
        productClient.send("body3".getBytes());
// end::producer[]

        ProductListener productListener = applicationContext.getBean(ProductListener.class);

        waitFor(() -> productListener.messages.size() == 3);
    }
}


<|start_filename|>rabbitmq/src/test/groovy/io/micronaut/rabbitmq/connect/ReactiveChannelSpec.groovy<|end_filename|>
package io.micronaut.rabbitmq.connect

import com.rabbitmq.client.AMQP
import com.rabbitmq.client.Channel
import com.rabbitmq.client.Envelope
import io.micronaut.rabbitmq.AbstractRabbitMQTest
import io.micronaut.rabbitmq.intercept.DefaultConsumer
import io.micronaut.rabbitmq.reactive.RabbitPublishState
import io.micronaut.rabbitmq.reactive.ReactorReactivePublisher
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import spock.lang.Stepwise

import java.time.Duration
import java.util.concurrent.atomic.AtomicInteger

@Stepwise
class ReactiveChannelSpec extends AbstractRabbitMQTest {

    void "test ack multiple"() {
        startContext()

        ChannelPool channelPool = applicationContext.getBean(ChannelPool)

        when:
        Channel consumeChannel = channelPool.channel
        boolean consumerAckd = false
        consumeChannel.basicConsume("abc", false, new DefaultConsumer() {

            AtomicInteger count = new AtomicInteger()

            @Override
            void handleTerminate(String consumerTag) {
                println "consumer terminated"
            }

            @Override
            void handleDelivery(String consumerTag, Envelope envelope,
                                AMQP.BasicProperties properties, byte[] body) throws IOException {
                if (count.incrementAndGet() == 4) {
                    consumeChannel.basicAck(envelope.deliveryTag, true)
                    consumerAckd = true
                }
            }
        })

        ReactorReactivePublisher reactiveChannel = new ReactorReactivePublisher(channelPool, new SingleRabbitConnectionFactoryConfig())
        List<Mono> monos = ["abc", "def", "ghi", "jkl"].collect {
            Mono.from(reactiveChannel.publish(new RabbitPublishState("", "abc", new AMQP.BasicProperties.Builder().build(), it.bytes)))
        }

        then:
        Flux.merge(monos)
            .blockFirst(Duration.ofSeconds(10)) == null

        waitFor {
            consumerAckd
        }

        cleanup:
        channelPool.returnChannel(consumeChannel)
    }

    void "test reinitialization"() {
        startContext()

        ChannelPool channelPool = applicationContext.getBean(ChannelPool)
        AtomicInteger messageCount = new AtomicInteger()
        Channel consumeChannel = channelPool.channel

        consumeChannel.basicConsume("abc", true, new DefaultConsumer() {
            @Override
            void handleTerminate(String consumerTag) {}

            @Override
            void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                messageCount.incrementAndGet()
            }
        })
        ReactorReactivePublisher reactiveChannel = new ReactorReactivePublisher(channelPool, new SingleRabbitConnectionFactoryConfig())

        when:
        reactiveChannel
                .publish(new RabbitPublishState("", "abc", new AMQP.BasicProperties.Builder().build(), "abc".bytes))
                .subscribe()

        then:
        waitFor {
            messageCount.get() == 1
        }

        when:
        reactiveChannel
                .publish(new RabbitPublishState("", "abc", new AMQP.BasicProperties.Builder().build(), "def".bytes))
                .subscribe()


        then:
        waitFor {
            messageCount.get() == 2
        }

        cleanup:
        channelPool.returnChannel(consumeChannel)
    }

    void "test publishing many messages"() {
        startContext("rabbitmq.channelPool.maxIdleChannels": 10)

        ChannelPool channelPool = applicationContext.getBean(ChannelPool)
        AtomicInteger integer = new AtomicInteger(75)
        ReactorReactivePublisher reactiveChannel = new ReactorReactivePublisher(channelPool, new SingleRabbitConnectionFactoryConfig())
        AtomicInteger messageCount = new AtomicInteger()
        Channel consumeChannel = channelPool.channel

        consumeChannel.basicConsume("abc", true, new DefaultConsumer() {
            @Override
            void handleTerminate(String consumerTag) {}

            @Override
            void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                messageCount.incrementAndGet()
            }
        })

        when:
        List<Mono> monos = []
        50.times {
            monos.add(Mono.from(reactiveChannel.publish(new RabbitPublishState("", "abc", null, "abc".bytes))).doOnSuccess({ t ->
                integer.decrementAndGet()
            }))
        }

        List<Mono> monos2 = []
        25.times {
            monos2.add(Mono.from(reactiveChannel.publish(new RabbitPublishState("", "abc", null, "abc".bytes))).doOnSuccess({ t ->
                integer.decrementAndGet()
            }))
        }

        Flux.merge(monos).subscribe()
        sleep 10
        Flux.merge(monos2).subscribe()

        then:
        waitFor {
            messageCount.get() == 75
            integer.get() == 0
        }

        cleanup:
        channelPool.returnChannel(consumeChannel)
    }
}


<|start_filename|>docs-examples/example-kotlin/src/test/kotlin/io/micronaut/rabbitmq/KGenericContainer.kt<|end_filename|>
package io.micronaut.rabbitmq

import org.testcontainers.containers.GenericContainer

class KGenericContainer(imageName: String) : GenericContainer<KGenericContainer>(imageName)


<|start_filename|>docs-examples/example-java/src/test/java/io/micronaut/rabbitmq/docs/properties/ProductClient.java<|end_filename|>
package io.micronaut.rabbitmq.docs.properties;

import io.micronaut.context.annotation.Requires;
// tag::imports[]
import io.micronaut.rabbitmq.annotation.Binding;
import io.micronaut.rabbitmq.annotation.RabbitClient;
import io.micronaut.rabbitmq.annotation.RabbitProperty;
// end::imports[]

@Requires(property = "spec.name", value = "PropertiesSpec")
// tag::clazz[]
@RabbitClient
@RabbitProperty(name = "appId", value = "myApp") // <1>
@RabbitProperty(name = "userId", value = "admin")
public interface ProductClient {

    @Binding("product")
    @RabbitProperty(name = "contentType", value = "application/json") // <2>
    @RabbitProperty(name = "userId", value = "guest")
    void send(byte[] data);

    @Binding("product")
    void send(@RabbitProperty("userId") String user, @RabbitProperty String contentType, byte[] data); // <3>
}
// end::clazz[]


<|start_filename|>docs-examples/example-kotlin/src/test/kotlin/io/micronaut/rabbitmq/docs/consumer/acknowledge/type/ProductListener.kt<|end_filename|>
package io.micronaut.rabbitmq.docs.consumer.acknowledge.type

import io.micronaut.context.annotation.Requires
// tag::imports[]
import io.micronaut.rabbitmq.annotation.Queue
import io.micronaut.rabbitmq.annotation.RabbitListener
import io.micronaut.rabbitmq.bind.RabbitAcknowledgement
import java.util.concurrent.atomic.AtomicInteger
// end::imports[]

@Requires(property = "spec.name", value = "AcknowledgeSpec")
// tag::clazz[]
@RabbitListener
class ProductListener {

    val messageCount = AtomicInteger()

    @Queue(value = "product") // <1>
    fun receive(data: ByteArray, acknowledgement: RabbitAcknowledgement) { // <2>
        val count = messageCount.getAndUpdate { intValue -> intValue + 1 }
        if (count == 0) {
            acknowledgement.nack(false, true) // <3>
        } else if (count > 3) {
            acknowledgement.ack(true) // <4>
        }
    }
}
// end::clazz[]


<|start_filename|>docs-examples/example-kotlin/src/test/kotlin/io/micronaut/rabbitmq/docs/publisher/acknowledge/ProductClient.kt<|end_filename|>
package io.micronaut.rabbitmq.docs.publisher.acknowledge

import io.micronaut.context.annotation.Requires
// tag::imports[]
import io.micronaut.rabbitmq.annotation.Binding
import io.micronaut.rabbitmq.annotation.RabbitClient
import org.reactivestreams.Publisher
import java.util.concurrent.CompletableFuture

// end::imports[]

@Requires(property = "spec.name", value = "PublisherAcknowledgeSpec")
// tag::clazz[]
@RabbitClient
interface ProductClient {

    @Binding("product")
    fun sendPublisher(data: ByteArray): Publisher<Void>  // <1>

    @Binding("product")
    fun sendFuture(data: ByteArray): CompletableFuture<Void>  // <2>

    @Binding("product")
    suspend fun sendSuspend(data: ByteArray) //suspend methods work too!
}
// end::clazz[]


<|start_filename|>rabbitmq/src/main/java/io/micronaut/rabbitmq/annotation/Queue.java<|end_filename|>
/*
 * Copyright 2017-2021 original 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 io.micronaut.rabbitmq.annotation;

import io.micronaut.context.annotation.AliasFor;
import io.micronaut.context.annotation.Executable;
import io.micronaut.messaging.annotation.MessageMapping;

import java.lang.annotation.Documented;
import java.lang.annotation.ElementType;
import java.lang.annotation.Inherited;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;

/**
 * Used to specify which queue messages should be consumed
 * from. The queue must already exist in the broker.
 *
 * @author <NAME>
 * @since 1.1.0
 */
@Documented
@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.METHOD})
@Executable
@Inherited
public @interface Queue {

    /**
     * @return The queue to subscribe to.
     */
    @AliasFor(annotation = MessageMapping.class, member = "value")
    String value();

    /**
     * Only applies to consumer methods that do not inject an
     * {@link io.micronaut.messaging.Acknowledgement} instance.
     *
     * @return Whether nacks should re-queue the message.
     */
    boolean reQueue() default false;

    /**
     * @return Whether the consumer is exclusive to the queue
     */
    boolean exclusive() default false;

    /**
     * @return The unacknowledged message limit
     */
    int prefetch() default 0;

    /**
     * @return The number of consumers used to consumer from a queue concurrently
     */
    int numberOfConsumers() default 1;

    /**
     * @see RabbitConnection#connection()
     * @return The connection to use
     */
    @AliasFor(annotation = RabbitConnection.class, member = "connection")
    String connection() default "";

    /**
     * @see RabbitConnection#executor()
     * @return The executor to use
     */
    @AliasFor(annotation = RabbitConnection.class, member = "executor")
    String executor() default "";
}


<|start_filename|>docs-examples/example-java/src/test/java/io/micronaut/rabbitmq/docs/exchange/AnimalListener.java<|end_filename|>
package io.micronaut.rabbitmq.docs.exchange;

import io.micronaut.context.annotation.Requires;
// tag::imports[]
import io.micronaut.rabbitmq.annotation.Queue;
import io.micronaut.rabbitmq.annotation.RabbitListener;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
// end::imports[]

@Requires(property = "spec.name", value = "CustomExchangeSpec")
// tag::clazz[]
@RabbitListener // <1>
public class AnimalListener {

    List<Animal> receivedAnimals = Collections.synchronizedList(new ArrayList<>());

    @Queue("cats") // <2>
    public void receive(Cat cat) { // <3>
        receivedAnimals.add(cat);
    }

    @Queue("snakes") // <2>
    public void receive(Snake snake) { // <3>
        receivedAnimals.add(snake);
    }
}
// end::clazz[]


<|start_filename|>docs-examples/example-groovy/src/test/groovy/io/micronaut/rabbitmq/docs/consumer/concurrent/ProductListener.groovy<|end_filename|>
package io.micronaut.rabbitmq.docs.consumer.concurrent

// tag::imports[]
import io.micronaut.context.annotation.Requires
import io.micronaut.rabbitmq.annotation.Queue
import io.micronaut.rabbitmq.annotation.RabbitListener

import java.util.concurrent.CopyOnWriteArraySet
// end::imports[]

@Requires(property = "spec.name", value = "ConcurrentSpec")
// tag::clazz[]
@RabbitListener
class ProductListener {

    CopyOnWriteArraySet<String> threads = new CopyOnWriteArraySet<>()

    @Queue(value = "product", numberOfConsumers = 5) // <1>
    void receive(byte[] data) {
        threads << Thread.currentThread().name // <2>
        sleep 500
    }
}
// end::clazz[]


<|start_filename|>docs-examples/example-java/src/test/java/io/micronaut/rabbitmq/docs/exchange/CustomExchangeSpec.java<|end_filename|>
package io.micronaut.rabbitmq.docs.exchange;

import io.micronaut.rabbitmq.AbstractRabbitMQTest;
import org.junit.jupiter.api.Test;

public class CustomExchangeSpec extends AbstractRabbitMQTest {

    @Test
    void testUsingACustomExchange() {
        startContext();

        AnimalClient client = applicationContext.getBean(AnimalClient.class);
        AnimalListener listener = applicationContext.getBean(AnimalListener.class);

        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));

        waitFor(() ->
                listener.receivedAnimals.size() == 4 &&
                listener.receivedAnimals.stream()
                        .filter(Cat.class::isInstance)
                        .anyMatch(cat -> cat.getName().equals("Whiskers") && ((Cat) cat).getLives() == 9) &&
                listener.receivedAnimals.stream()
                        .filter(Cat.class::isInstance)
                        .anyMatch(cat -> cat.getName().equals("Mr. Bigglesworth") && ((Cat) cat).getLives() == 8) &&
                listener.receivedAnimals.stream()
                        .filter(Snake.class::isInstance)
                        .anyMatch(snake -> snake.getName().equals("Buttercup") && !((Snake) snake).isVenomous()) &&
                listener.receivedAnimals.stream()
                        .filter(Snake.class::isInstance)
                        .anyMatch(snake -> snake.getName().equals("Monty the Python") && ((Snake) snake).isVenomous())
        );
    }
}


<|start_filename|>docs-examples/example-groovy/src/test/groovy/io/micronaut/rabbitmq/docs/consumer/acknowledge/type/AcknowledgeSpec.groovy<|end_filename|>
package io.micronaut.rabbitmq.docs.consumer.acknowledge.type

import io.micronaut.rabbitmq.AbstractRabbitMQTest

class AcknowledgeSpec extends AbstractRabbitMQTest {

    void "test acking with an acknowledgement argument"() {
        startContext()

        when:
// tag::producer[]
ProductClient productClient = applicationContext.getBean(ProductClient)
productClient.send("message body".bytes)
productClient.send("message body".bytes)
productClient.send("message body".bytes)
productClient.send("message body".bytes)
// end::producer[]

        ProductListener productListener = applicationContext.getBean(ProductListener)

        then:
        waitFor {
            productListener.messageCount.get() == 5
        }
    }
}


<|start_filename|>rabbitmq/src/main/java/io/micronaut/rabbitmq/reactive/ReactorReactivePublisher.java<|end_filename|>
/*
 * Copyright 2017-2020 original 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 io.micronaut.rabbitmq.reactive;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.ShutdownSignalException;
import io.micronaut.context.annotation.EachBean;
import io.micronaut.context.annotation.Parameter;
import io.micronaut.core.annotation.Internal;
import io.micronaut.messaging.exceptions.MessagingClientException;
import io.micronaut.rabbitmq.bind.RabbitConsumerState;
import io.micronaut.rabbitmq.connect.ChannelPool;
import io.micronaut.rabbitmq.connect.RabbitConnectionFactoryConfig;
import io.micronaut.rabbitmq.exception.RabbitClientException;
import io.micronaut.rabbitmq.intercept.DefaultConsumer;
import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoSink;

import java.io.IOException;
import java.util.Collections;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;

/**
 * A reactive publisher implementation that uses a single channel per publish
 * operation and returns a Reactor {@link Mono}.
 * <p>
 * This is an internal API and may change at any time
 *
 * @author <NAME>
 * @author <NAME>
 * @since 3.0.0
 */
@Internal
@EachBean(ChannelPool.class)
public class ReactorReactivePublisher implements ReactivePublisher {

    private final ChannelPool channelPool;
    private final RabbitConnectionFactoryConfig config;

    /**
     * Default constructor.
     *
     * @param channelPool The channel pool to retrieve channels
     * @param config      Any configuration used in building the publishers
     */
    public ReactorReactivePublisher(@Parameter ChannelPool channelPool,
                                    @Parameter RabbitConnectionFactoryConfig config) {
        this.channelPool = channelPool;
        this.config = config;
    }

    @Override
    public Mono<Void> publishAndConfirm(RabbitPublishState publishState) {
        return Mono.from(getChannel()
                .flatMap(this::initializePublish)
                .flatMap(channel -> publishInternal(channel, publishState))
                .timeout(config.getConfirmTimeout(),
                        Mono.error(new RabbitClientException(String.format("Timed out waiting for publisher confirm for exchange: [%s] and routing key: [%s]", publishState.getExchange(), publishState.getRoutingKey())))
                )
                .then());
//                .toFlowable();
    }

    @Override
    public Mono<Void> publish(RabbitPublishState publishState) {
        return getChannel()
                .flatMap(channel -> publishInternalNoConfirm(channel, publishState))
                .then();
    }

    @Override
    public Flux<RabbitConsumerState> publishAndReply(RabbitPublishState publishState) {
        Flux<RabbitConsumerState> flowable = getChannel()
                .flatMap(channel -> publishRpcInternal(channel, publishState))
                .flux();

        config.getRpc()
                .getTimeout()
                .ifPresent(flowable::timeout);

        return flowable;
    }

    /**
     * Creates a {@link Mono} from a channel, emitting an error
     * if the channel could not be retrieved.
     *
     * @return A {@link Mono} that emits the channel on success
     */
    protected Mono<Channel> getChannel() {
        return Mono.create(emitter -> {
            try {
                Channel channel = channelPool.getChannel();
                emitter.success(channel);
            } catch (IOException e) {
                emitter.error(new RabbitClientException("Failed to retrieve a channel from the pool", e));
            }
        });
    }

    /**
     * Publishes the message to the channel. The {@link Mono} returned from this
     * method should ensure the {@link ConfirmListener} is added to the channel prior
     * to publishing and the {@link ConfirmListener} is removed from the channel after
     * the publish has been acknowledged.
     *
     * @param channel      The channel to publish the message to
     * @param publishState The publishing state
     * @return A completable that terminates when the publish has been acknowledged
     * @see Channel#basicPublish(String, String, AMQP.BasicProperties, byte[])
     */
    protected Mono<Object> publishInternal(Channel channel, RabbitPublishState publishState) {
        return Mono.create(subscriber -> {
            Disposable listener = createListener(channel, subscriber, publishState);
            try {
                channel.basicPublish(
                        publishState.getExchange(),
                        publishState.getRoutingKey(),
                        publishState.getProperties(),
                        publishState.getBody()
                );
            } catch (IOException e) {
                listener.dispose();
                subscriber.error(e);
            }
        }).doFinally(signalType -> returnChannel(channel));
    }

    /**
     * Publishes the message to the channel. The {@link Mono} returned from this
     * method should ensure the {@link ConfirmListener} is added to the channel prior
     * to publishing and the {@link ConfirmListener} is removed from the channel after
     * the publish has been acknowledged.
     *
     * @param channel      The channel to publish the message to
     * @param publishState The publishing state
     * @return A completable that terminates when the publish has been acknowledged
     * @see Channel#basicPublish(String, String, AMQP.BasicProperties, byte[])
     */
    protected Mono<RabbitConsumerState> publishRpcInternal(Channel channel, RabbitPublishState publishState) {
        return Mono.<RabbitConsumerState>create(subscriber -> {
            Disposable listener = null;
            try {
                String correlationId = UUID.randomUUID().toString();
                AMQP.BasicProperties properties = publishState.getProperties().builder().correlationId(correlationId).build();
                listener = createConsumer(channel, publishState, correlationId, subscriber);

                channel.basicPublish(
                        publishState.getExchange(),
                        publishState.getRoutingKey(),
                        properties,
                        publishState.getBody()
                );
            } catch (IOException e) {
                if (listener != null) {
                    listener.dispose();
                }
                subscriber.error(new MessagingClientException("Failed to publish the message", e));
            }
        }).doFinally(signalType -> returnChannel(channel));
    }

    /**
     * Publishes the message to the channel. The {@link Mono} returned from this
     * method should ensure the {@link ConfirmListener} is added to the channel prior
     * to publishing and the {@link ConfirmListener} is removed from the channel after
     * the publish has been acknowledged.
     *
     * @param channel      The channel to publish the message to
     * @param publishState The publishing state
     * @return A completable that terminates when the publish has been acknowledged
     * @see Channel#basicPublish(String, String, AMQP.BasicProperties, byte[])
     */
    protected Mono<Object> publishInternalNoConfirm(Channel channel, RabbitPublishState publishState) {
        return Mono.create(subscriber -> {
            try {
                channel.basicPublish(
                        publishState.getExchange(),
                        publishState.getRoutingKey(),
                        publishState.getProperties(),
                        publishState.getBody()
                );

                subscriber.success();
            } catch (IOException e) {
                subscriber.error(new MessagingClientException("Failed to publish the message", e));
            }
        }).doFinally(signalType -> returnChannel(channel));
    }

    /**
     * Initializes the channel to allow publisher acknowledgements.
     *
     * @param channel The channel to enable acknowledgements.
     * @return A {@link Mono} that will complete according to the
     * success of the operation.
     */
    protected Mono<Channel> initializePublish(Channel channel) {
        return Mono.create(emitter -> {
            try {
                channel.confirmSelect();
                emitter.success(channel);
            } catch (IOException e) {
                emitter.error(new MessagingClientException("Failed to enable publisher confirms on the channel", e));
            }
        });
    }

    /**
     * Removes confirm listeners from the channel and returns the
     * channel to the pool.
     *
     * @param channel The channel to clean and return
     */
    protected void returnChannel(Channel channel) {
        channelPool.returnChannel(channel);
    }

    /**
     * Listens for ack/nack from the broker. The listener auto disposes
     * itself after receiving a response from the broker. If no response is
     * received, the caller is responsible for disposing the listener.
     *
     * @param channel      The channel to listen for confirms
     * @param emitter      The emitter to send the event
     * @param publishState The publishing state
     * @return A disposable to allow cleanup of the listener
     */
    protected Disposable createListener(Channel channel, MonoSink<Object> emitter, RabbitPublishState publishState) {
        AtomicBoolean disposed = new AtomicBoolean();
        Consumer<ConfirmListener> dispose = (listener) -> {
            if (disposed.compareAndSet(false, true)) {
                channel.removeConfirmListener(listener);
            }
        };

        ConfirmListener confirmListener = new ConfirmListener() {
            @Override
            public void handleAck(long deliveryTag, boolean multiple) {
                ackNack(deliveryTag, multiple, true);
            }

            @Override
            public void handleNack(long deliveryTag, boolean multiple) {
                ackNack(deliveryTag, multiple, false);
            }

            private void ackNack(long deliveryTag, boolean multiple, boolean ack) {
                if (ack) {
                    emitter.success();
                } else {
                    emitter.error(new RabbitClientException("Message could not be delivered to the broker", Collections.singletonList(publishState)));
                }
                dispose.accept(this);
            }
        };

        channel.addConfirmListener(confirmListener);

        return new Disposable() {
            @Override
            public void dispose() {
                dispose.accept(confirmListener);
            }

            @Override
            public boolean isDisposed() {
                return disposed.get();
            }
        };
    }

    /**
     * Listens for ack/nack from the broker. The listener auto disposes
     * itself after receiving a response from the broker. If no response is
     * received, the caller is responsible for disposing the listener.
     *
     * @param channel       The channel to listen for confirms
     * @param publishState  The publish state
     * @param correlationId The correlation id
     * @param emitter       The emitter to send the response
     * @return A disposable to allow cleanup of the listener
     * @throws IOException If an error occurred subscribing
     */
    protected Disposable createConsumer(Channel channel, RabbitPublishState publishState, String correlationId, MonoSink<RabbitConsumerState> emitter) throws IOException {
        String replyTo = publishState.getProperties().getReplyTo();
        AtomicBoolean disposed = new AtomicBoolean();

        Consumer<String> dispose = (consumerTag) -> {
            if (disposed.compareAndSet(false, true)) {
                try {
                    channel.basicCancel(consumerTag);
                } catch (IOException e) {
                    //no-op
                }
            }
        };

        DefaultConsumer consumer = new DefaultConsumer() {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                if (replyTo.equals("amq.rabbitmq.reply-to") || correlationId.equals(properties.getCorrelationId())) {
                    dispose.accept(consumerTag);
                    emitter.success(new RabbitConsumerState(envelope, properties, body, channel));
                }
            }

            @Override
            public void handleCancel(String consumerTag) throws IOException {
                handleError(consumerTag);
            }

            @Override
            public void handleShutdownSignal(String consumerTag, ShutdownSignalException sig) {
                handleError(consumerTag);
            }

            private void handleError(String consumerTag) {
                dispose.accept(consumerTag);
                emitter.error(new RabbitClientException("Message was not able to be received from the reply to queue. The consumer was cancelled", Collections.singletonList(publishState)));
            }
        };

        final String consumerTag = channel.basicConsume(replyTo, true, consumer);

        return new Disposable() {
            @Override
            public void dispose() {
                dispose.accept(consumerTag);
            }

            @Override
            public boolean isDisposed() {
                return disposed.get();
            }
        };
    }
}


<|start_filename|>rabbitmq/src/main/java/io/micronaut/rabbitmq/bind/RabbitAcknowledgement.java<|end_filename|>
/*
 * Copyright 2017-2020 original 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 io.micronaut.rabbitmq.bind;

import io.micronaut.messaging.Acknowledgement;
import io.micronaut.messaging.exceptions.MessageAcknowledgementException;

/**
 * A contract for acknowledging or rejecting RabbitMQ messages.
 *
 * @author <NAME>
 * @since 1.1.0
 */
public interface RabbitAcknowledgement extends Acknowledgement {

    @Override
    default void ack() throws MessageAcknowledgementException {
        ack(false);
    }

    @Override
    default void nack() throws MessageAcknowledgementException {
        nack(false, false);
    }

    /**
     * Acknowledges this message.
     *
     * @param multiple If true, also acknowledge previous messages
     * @throws MessageAcknowledgementException If an error occurred
     */
    void ack(boolean multiple) throws MessageAcknowledgementException;

    /**
     * Rejects this message.
     *
     * @param multiple If true, also reject previous messages
     * @param reQueue Re-queue the message to be consumed again
     * @throws MessageAcknowledgementException If an error occurred
     */
    void nack(boolean multiple, boolean reQueue) throws MessageAcknowledgementException;
}


<|start_filename|>rabbitmq/src/main/java/io/micronaut/rabbitmq/bind/RabbitHeadersBinder.java<|end_filename|>
/*
 * Copyright 2017-2021 original 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 io.micronaut.rabbitmq.bind;

import io.micronaut.core.convert.ArgumentConversionContext;
import io.micronaut.core.convert.ConversionService;
import io.micronaut.rabbitmq.annotation.RabbitHeaders;
import jakarta.inject.Singleton;

/**
 * Binds an argument of with the {@link RabbitHeaders} annotation from the {@link RabbitConsumerState}.
 *
 * @author <NAME>
 * @since 2.5.0
 */
@Singleton
public class RabbitHeadersBinder implements RabbitAnnotatedArgumentBinder<RabbitHeaders> {

    private final ConversionService<?> conversionService;

    /**
     * Default constructor.
     *
     * @param conversionService The conversion service to convert the body
     */
    public RabbitHeadersBinder(ConversionService<?> conversionService) {
        this.conversionService = conversionService;
    }

    @Override
    public Class<RabbitHeaders> getAnnotationType() {
        return RabbitHeaders.class;
    }

    @Override
    public BindingResult<Object> bind(ArgumentConversionContext<Object> context, RabbitConsumerState source) {
        return () -> conversionService.convert(source.getProperties().getHeaders(), context);
    }
}


<|start_filename|>docs-examples/example-java/src/test/java/io/micronaut/rabbitmq/docs/serdes/ProductInfoSerDesSpec.java<|end_filename|>
package io.micronaut.rabbitmq.docs.serdes;

import io.micronaut.rabbitmq.AbstractRabbitMQTest;
import org.junit.jupiter.api.Test;

public class ProductInfoSerDesSpec extends AbstractRabbitMQTest {

    @Test
    void testUsingACustomSerDes() {
        startContext();

// tag::producer[]
        ProductClient productClient = applicationContext.getBean(ProductClient.class);
        productClient.send(new ProductInfo("small", 10L, true));
        productClient.send(new ProductInfo("medium", 20L, true));
        productClient.send(new ProductInfo(null, 30L, false));
// end::producer[]

        ProductListener productListener = applicationContext.getBean(ProductListener.class);

        waitFor(() ->
                productListener.messages.size() == 3 &&
                productListener.messages.stream().anyMatch(pi -> pi.getCount() == 10L) &&
                productListener.messages.stream().anyMatch(pi -> pi.getCount() == 20L) &&
                productListener.messages.stream().anyMatch(pi -> pi.getCount() == 30L)
        );
    }
}


<|start_filename|>rabbitmq/src/test/groovy/io/micronaut/rabbitmq/rpc/RpcPublisher.groovy<|end_filename|>
package io.micronaut.rabbitmq.rpc

import io.micronaut.context.annotation.Requires
import io.micronaut.rabbitmq.annotation.Binding
import io.micronaut.rabbitmq.annotation.RabbitClient
import io.micronaut.rabbitmq.annotation.RabbitProperty
import org.reactivestreams.Publisher

@Requires(property = "spec.name", value = "RpcSpec")
@RabbitClient
@RabbitProperty(name = "replyTo", value = "amq.rabbitmq.reply-to")
interface RpcPublisher {

    @Binding("rpc")
    Publisher<String> rpcCall(String data)

    @Binding("rpc")
    String rpcBlocking(String data)
}


<|start_filename|>rabbitmq/src/test/groovy/io/micronaut/rabbitmq/connect/MultipleServersSpec.groovy<|end_filename|>
package io.micronaut.rabbitmq.connect

import com.rabbitmq.client.Connection
import io.micronaut.context.ApplicationContext
import io.micronaut.inject.qualifiers.Qualifiers
import org.testcontainers.containers.GenericContainer
import org.testcontainers.containers.wait.strategy.LogMessageWaitStrategy
import spock.lang.Specification

import java.time.Duration

class MultipleServersSpec extends Specification {

    void "test multiple server configuration"() {
        given:
        GenericContainer rabbit1 = new GenericContainer("library/rabbitmq:3.7")
                .withExposedPorts(5672)
                .waitingFor(new LogMessageWaitStrategy().withRegEx("(?s).*Server startup complete.*"))
        GenericContainer rabbit2 = new GenericContainer("library/rabbitmq:3.7")
                .withExposedPorts(5672)
                .waitingFor(new LogMessageWaitStrategy().withRegEx("(?s).*Server startup complete.*"))

        when:
        rabbit1.start()
        rabbit2.start()
        ApplicationContext context = ApplicationContext.run(
                ["spec.name": getClass().simpleName,
                 "rabbitmq.servers.one.uri": "amqp://localhost:${rabbit1.getMappedPort(5672)}",
                 "rabbitmq.servers.one.channel-pool.max-idle-channels": "10",
                 "rabbitmq.servers.one.rpc.timeout": "10s",
                 "rabbitmq.servers.two.uri": "amqp://localhost:${rabbit2.getMappedPort(5672)}",
                 "rabbitmq.servers.two.channel-pool.max-idle-channels": "20",
                 "rabbitmq.servers.two.rpc.timeout": "20s"])

        then:
        context.getBean(RabbitConnectionFactoryConfig, Qualifiers.byName("one")).port == rabbit1.getMappedPort(5672)
        context.getBean(RabbitConnectionFactoryConfig, Qualifiers.byName("one")).rpc.timeout.get() == Duration.ofSeconds(10)
        context.getBean(RabbitConnectionFactoryConfig, Qualifiers.byName("two")).rpc.timeout.get() == Duration.ofSeconds(20)
        context.getBean(RabbitConnectionFactoryConfig, Qualifiers.byName("two")).port == rabbit2.getMappedPort(5672)
        context.getBean(Connection, Qualifiers.byName("one")).port == rabbit1.getMappedPort(5672)
        context.getBean(Connection, Qualifiers.byName("two")).port == rabbit2.getMappedPort(5672)
        context.getBean(DefaultChannelPool, Qualifiers.byName("one")).connection == context.getBean(Connection, Qualifiers.byName("one"))
        context.getBean(DefaultChannelPool, Qualifiers.byName("two")).connection == context.getBean(Connection, Qualifiers.byName("two"))
        context.getBean(DefaultChannelPool, Qualifiers.byName("one")).channels.remainingCapacity() == 9
        context.getBean(DefaultChannelPool, Qualifiers.byName("two")).channels.remainingCapacity() == 19
        context.getBean(DefaultChannelPool, Qualifiers.byName("two")).name == "two"
        context.getBean(DefaultChannelPool, Qualifiers.byName("one")).name == "one"

        cleanup:
        context.stop()
        rabbit1.stop()
        rabbit2.stop()
    }
}


<|start_filename|>rabbitmq/src/test/groovy/io/micronaut/rabbitmq/rpc/RpcSpec.groovy<|end_filename|>
package io.micronaut.rabbitmq.rpc

import io.micronaut.rabbitmq.AbstractRabbitMQTest
import reactor.core.publisher.Mono

class RpcSpec extends AbstractRabbitMQTest {

    void "test simple RPC call"() {
        startContext()

        RpcPublisher producer = applicationContext.getBean(RpcPublisher)

        expect:
        Mono.from(producer.rpcCall("hello")).block() == "HELLO"
        producer.rpcBlocking("world") == "WORLD"
    }
}


<|start_filename|>docs-examples/example-kotlin/src/test/kotlin/io/micronaut/rabbitmq/docs/serdes/ProductInfo.kt<|end_filename|>
package io.micronaut.rabbitmq.docs.serdes

// tag::clazz[]
class ProductInfo(val size: String?, // <1>
                  val count: Long, // <2>
                  val sealed: Boolean)// <3>
// end::clazz[]

<|start_filename|>rabbitmq/src/main/java/io/micronaut/rabbitmq/exception/RabbitListenerException.java<|end_filename|>
/*
 * Copyright 2017-2020 original 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 io.micronaut.rabbitmq.exception;

import io.micronaut.core.annotation.Nullable;
import io.micronaut.messaging.exceptions.MessageListenerException;
import io.micronaut.rabbitmq.bind.RabbitConsumerState;

import java.util.Optional;

/**
 * Exception thrown when an error occurs processing a RabbitMQ message via a {@link io.micronaut.rabbitmq.annotation.RabbitListener}.
 *
 * @author <NAME>
 * @since 1.1.0
 */
public class RabbitListenerException extends MessageListenerException {

    private final Object listener;
    private final RabbitConsumerState messageState;

    /**
     * Creates a new exception.
     *
     * @param message The message
     * @param listener The listener
     * @param messageState The message
     */
    public RabbitListenerException(String message, Object listener,
                                   @Nullable RabbitConsumerState messageState) {
        super(message);
        this.listener = listener;
        this.messageState = messageState;
    }

    /**
     * Creates a new exception.
     *
     * @param message The message
     * @param cause The cause
     * @param listener The listener
     * @param messageState The message
     */
    public RabbitListenerException(String message, Throwable cause,
                                   Object listener, @Nullable RabbitConsumerState messageState) {
        super(message, cause);
        this.listener = listener;
        this.messageState = messageState;
    }
    
    /**
     * Creates a new exception.
     *
     * @param cause The cause
     * @param listener The listener
     * @param messageState The message
     */
    public RabbitListenerException(Throwable cause, Object listener,
                                   @Nullable RabbitConsumerState messageState) {
        super(cause.getMessage(), cause);
        this.listener = listener;
        this.messageState = messageState;
    }

    /**
     * @return The bean that is the kafka listener
     */
    public Object getListener() {
        return listener;
    }

    /**
     * @return The consumer that produced the error
     */
    public Optional<RabbitConsumerState> getMessageState() {
        return Optional.ofNullable(messageState);
    }
}


<|start_filename|>docs-examples/example-kotlin/src/test/kotlin/io/micronaut/rabbitmq/docs/consumer/custom/type/ProductListener.kt<|end_filename|>
package io.micronaut.rabbitmq.docs.consumer.custom.type

import io.micronaut.context.annotation.Requires
// tag::imports[]
import io.micronaut.rabbitmq.annotation.Queue
import io.micronaut.rabbitmq.annotation.RabbitListener
import java.util.Collections
// end::imports[]

@Requires(property = "spec.name", value = "ProductInfoSpec")
// tag::clazz[]
@RabbitListener
class ProductListener {

    val messages: MutableList<ProductInfo> = Collections.synchronizedList(ArrayList())

    @Queue("product")
    fun receive(data: ByteArray,
                productInfo: ProductInfo) { // <1>
        messages.add(productInfo)
    }
}
// end::clazz[]


<|start_filename|>rabbitmq/src/test/groovy/io/micronaut/rabbitmq/listener/ListenerErrorSpec.groovy<|end_filename|>
package io.micronaut.rabbitmq.listener

import io.micronaut.context.annotation.Primary
import io.micronaut.context.annotation.Replaces
import io.micronaut.context.annotation.Requires
import io.micronaut.rabbitmq.AbstractRabbitMQTest
import io.micronaut.rabbitmq.annotation.Binding
import io.micronaut.rabbitmq.annotation.Queue
import io.micronaut.rabbitmq.annotation.RabbitClient
import io.micronaut.rabbitmq.annotation.RabbitListener
import io.micronaut.rabbitmq.exception.DefaultRabbitListenerExceptionHandler
import io.micronaut.rabbitmq.exception.RabbitListenerException
import io.micronaut.rabbitmq.exception.RabbitListenerExceptionHandler
import jakarta.inject.Singleton

import java.util.concurrent.CopyOnWriteArrayList

class ListenerErrorSpec extends AbstractRabbitMQTest {

    void "test a local error handler"() {
        startContext(global: false)
        MyProducer producer = applicationContext.getBean(MyProducer)
        producer.go("abc")
        producer.go("def")
        producer.go("ghi")

        when:
        MyConsumer consumer = applicationContext.getBean(MyConsumer)

        then:
        waitFor {
            consumer.errors.size() == 3
            consumer.errors.collect { it.cause.message } as Set == ["abc", "def", "ghi"] as Set
        }
    }

    void "test a global error handler"() {
        startContext(global: true)
        MyProducer producer = applicationContext.getBean(MyProducer)
        producer.go("abc")
        producer.go("def")
        producer.go("ghi")

        when:
        MyGlobalErrorHandler handler = applicationContext.getBean(MyGlobalErrorHandler)

        then:
        waitFor {
            handler.errors.size() == 3
            handler.errors.collect { it.cause.message } as Set == ["abc", "def", "ghi"] as Set
        }
    }

    @Requires(property = "spec.name", value = "ListenerErrorSpec")
    @RabbitClient
    static interface MyProducer {
        @Binding("error")
        void go(String data)
    }

    @Requires(property = "spec.name", value = "ListenerErrorSpec")
    @Requires(property = "global", value = "false")
    @RabbitListener
    static class MyConsumer implements RabbitListenerExceptionHandler {

        CopyOnWriteArrayList<Throwable> errors = []

        @Queue("error")
        void listen(String data) {
            throw new RuntimeException(data)
        }

        @Override
        void handle(RabbitListenerException e) {
            errors << e
        }
    }

    @Requires(property = "spec.name", value = "ListenerErrorSpec")
    @Requires(property = "global", value = "true")
    @RabbitListener
    static class MyConsumer2 {

        @Queue("error")
        void listen(String data) {
            throw new RuntimeException(data)
        }
    }

    @Requires(property = "spec.name", value = "ListenerErrorSpec")
    @Requires(property = "global", value = "true")
    @Replaces(DefaultRabbitListenerExceptionHandler)
    @Primary
    @Singleton
    static class MyGlobalErrorHandler implements RabbitListenerExceptionHandler {

        CopyOnWriteArrayList<Throwable> errors = []

        @Override
        void handle(RabbitListenerException e) {
            errors << e
        }
    }
}


<|start_filename|>rabbitmq/src/test/groovy/io/micronaut/rabbitmq/annotation/ColdObservableSpec.groovy<|end_filename|>
package io.micronaut.rabbitmq.annotation

import io.micronaut.context.annotation.Requires
import io.micronaut.rabbitmq.AbstractRabbitMQTest
import org.reactivestreams.Publisher
import org.reactivestreams.Subscriber
import org.reactivestreams.Subscription

class ColdObservableSpec extends AbstractRabbitMQTest {

    void "test publisher acknowledgement returns a cold observable"() {
        startContext()

        MyProducer producer = applicationContext.getBean(MyProducer)
        MyConsumer consumer = applicationContext.getBean(MyConsumer)

        when:
        Publisher<Void> publisher = producer.publisher("def".bytes)

        then:
        waitFor {
            consumer.messages.size() == 0
        }

        when:
        publisher.subscribe(new Subscriber<Void>() {

            @Override
            void onSubscribe(Subscription s) {
                s.request(1)
            }

            @Override void onNext(Void unused) {}
            @Override void onError(Throwable t) {}
            @Override void onComplete() {}
        })

        then:
        waitFor {
            consumer.messages.size() == 1
        }
    }

    @Requires(property = "spec.name", value = "ColdObservableSpec")
    @RabbitClient
    static interface MyProducer {
        @Binding("abc")
        Publisher<Void> publisher(byte[] data)
    }

    @Requires(property = "spec.name", value = "ColdObservableSpec")
    @RabbitListener
    static class MyConsumer {

        static List<byte[]> messages = []

        @Queue("abc")
        void listen(byte[] data) {
            messages << data
        }
    }
}


<|start_filename|>docs-examples/example-groovy/src/test/groovy/io/micronaut/rabbitmq/docs/parameters/ProductListener.groovy<|end_filename|>
package io.micronaut.rabbitmq.docs.parameters

import io.micronaut.context.annotation.Requires
// tag::imports[]
import io.micronaut.rabbitmq.annotation.Queue
import io.micronaut.rabbitmq.annotation.RabbitListener

import java.util.concurrent.CopyOnWriteArrayList
// end::imports[]

@Requires(property = "spec.name", value = "BindingSpec")
// tag::clazz[]
@RabbitListener
class ProductListener {

    CopyOnWriteArrayList<Integer> messageLengths = []

    @Queue("product") // <1>
    void receive(byte[] data) {
        messageLengths << data.length
    }
}
// end::clazz[]


<|start_filename|>docs-examples/example-java/src/test/java/io/micronaut/rabbitmq/docs/consumer/concurrent/ConcurrentSpec.java<|end_filename|>
package io.micronaut.rabbitmq.docs.consumer.concurrent;

import io.micronaut.rabbitmq.AbstractRabbitMQTest;
import org.junit.jupiter.api.Test;

public class ConcurrentSpec extends AbstractRabbitMQTest {

    @Test
    void testConcurrentConsumers() {
        startContext();

        ProductClient productClient = applicationContext.getBean(ProductClient.class);
        for (int i = 0; i < 4; i++) {
            productClient.send("body".getBytes());
        }

        ProductListener productListener = applicationContext.getBean(ProductListener.class);

        waitFor(() -> productListener.threads.size() == 4);
    }
}


<|start_filename|>docs-examples/example-groovy/src/test/groovy/io/micronaut/rabbitmq/docs/consumer/custom/annotation/DeliveryTagSpec.groovy<|end_filename|>
package io.micronaut.rabbitmq.docs.consumer.custom.annotation

import io.micronaut.rabbitmq.AbstractRabbitMQTest

class DeliveryTagSpec extends AbstractRabbitMQTest {

    void "test using a custom annotation binder"() {
        startContext()

        when:
// tag::producer[]
        ProductClient productClient = applicationContext.getBean(ProductClient)
        productClient.send("body".bytes)
        productClient.send("body2".bytes)
        productClient.send("body3".bytes)
// end::producer[]

        ProductListener productListener = applicationContext.getBean(ProductListener)

        then:
        waitFor {
            productListener.messages.size() == 3
        }
    }
}


<|start_filename|>docs-examples/example-groovy/src/test/groovy/io/micronaut/rabbitmq/docs/headers/HeadersSpec.groovy<|end_filename|>
package io.micronaut.rabbitmq.docs.headers

import io.micronaut.rabbitmq.AbstractRabbitMQTest

class HeadersSpec extends AbstractRabbitMQTest {

    void "test publishing and receiving headers"() {
        when:
        startContext()
// tag::producer[]
        ProductClient productClient = applicationContext.getBean(ProductClient)
        productClient.send("body".bytes)
        productClient.send("medium", 20L, "body2".bytes)
        productClient.send(null, 30L, "body3".bytes)
        productClient.send([productSize: "large", "x-product-count": 40L], "body4".bytes)
// end::producer[]

        ProductListener productListener = applicationContext.getBean(ProductListener)

        then:
        waitFor {
            productListener.messageProperties.size() == 4
            productListener.messageProperties.contains("true|10|small")
            productListener.messageProperties.contains("true|20|medium")
            productListener.messageProperties.contains("true|30|null")
            productListener.messageProperties.contains("true|40|large")
        }
    }
}


<|start_filename|>docs-examples/example-java/src/test/java/io/micronaut/rabbitmq/docs/exchange/AnimalClient.java<|end_filename|>
package io.micronaut.rabbitmq.docs.exchange;

import io.micronaut.context.annotation.Requires;
// tag::imports[]
import io.micronaut.messaging.annotation.MessageHeader;
import io.micronaut.rabbitmq.annotation.RabbitClient;
// end::imports[]

@Requires(property = "spec.name", value = "CustomExchangeSpec")
// tag::clazz[]
@RabbitClient("animals") // <1>
public interface AnimalClient {

    void send(@MessageHeader String animalType, Animal animal); // <2>

    default void send(Animal animal) { // <3>
        send(animal.getClass().getSimpleName(), animal);
    }
}
// end::clazz[]


<|start_filename|>docs-examples/example-kotlin/src/test/kotlin/io/micronaut/rabbitmq/docs/quickstart/ProductListener.kt<|end_filename|>
package io.micronaut.rabbitmq.docs.quickstart

import io.micronaut.context.annotation.Requires
// tag::imports[]
import io.micronaut.rabbitmq.annotation.Queue
import io.micronaut.rabbitmq.annotation.RabbitListener
import java.util.Collections
// end::imports[]

@Requires(property = "spec.name", value = "QuickstartSpec")
// tag::clazz[]
@RabbitListener // <1>
class ProductListener {

    val messageLengths: MutableList<String> = Collections.synchronizedList(ArrayList())

    @Queue("product") // <2>
    fun receive(data: ByteArray) { // <3>
        val string = String(data)
        messageLengths.add(string)
        println("Kotlin received ${data.size} bytes from RabbitMQ: ${string}")
    }
}
// end::clazz[]


<|start_filename|>docs-examples/example-groovy/src/test/groovy/io/micronaut/rabbitmq/AbstractRabbitMQTest.groovy<|end_filename|>
package io.micronaut.rabbitmq

import io.micronaut.context.ApplicationContext
import org.testcontainers.containers.GenericContainer
import org.testcontainers.containers.wait.strategy.LogMessageWaitStrategy
import spock.lang.Specification
import spock.util.concurrent.PollingConditions

abstract class AbstractRabbitMQTest extends Specification {

    static GenericContainer rabbitContainer =
            new GenericContainer("library/rabbitmq:3.7")
                    .withExposedPorts(5672)
                    .waitingFor(new LogMessageWaitStrategy().withRegEx("(?s).*Server startup complete.*"))

    static {
        rabbitContainer.start()
    }

    protected ApplicationContext applicationContext
    protected PollingConditions conditions = new PollingConditions(timeout: 5)

    protected void startContext() {
        applicationContext = ApplicationContext.run(configuration, "test")
    }

    protected Map<String, Object> getConfiguration() {
        ["rabbitmq.port": rabbitContainer.getMappedPort(5672),
         "spec.name": getClass().simpleName] as Map
    }

    protected void waitFor(Closure<?> conditionEvaluator) {
        conditions.eventually conditionEvaluator
    }

    void cleanup() {
        applicationContext?.close()
    }
}


<|start_filename|>docs-examples/example-kotlin/src/test/kotlin/io/micronaut/rabbitmq/docs/exchange/Animal.kt<|end_filename|>
package io.micronaut.rabbitmq.docs.exchange

abstract class Animal {

    var name: String? = null

    constructor(name: String) {
        this.name = name
    }

    constructor() {}
}


<|start_filename|>rabbitmq/src/test/groovy/io/micronaut/rabbitmq/AbstractRabbitMQClusterTest.groovy<|end_filename|>
package io.micronaut.rabbitmq

import com.github.dockerjava.api.model.HealthCheck
import io.micronaut.context.ApplicationContext
import org.slf4j.Logger
import org.slf4j.LoggerFactory
import org.testcontainers.containers.BindMode
import org.testcontainers.containers.GenericContainer
import org.testcontainers.containers.InternetProtocol
import org.testcontainers.containers.Network
import org.testcontainers.containers.output.Slf4jLogConsumer
import org.testcontainers.containers.wait.strategy.AbstractWaitStrategy
import org.testcontainers.containers.wait.strategy.Wait
import org.testcontainers.utility.DockerImageName
import spock.lang.Specification
import spock.util.concurrent.PollingConditions

import java.time.Duration

abstract class AbstractRabbitMQClusterTest extends Specification {

    private static final Logger log = LoggerFactory.getLogger(AbstractRabbitMQClusterTest)

    private static final int AMQP_PORT = 5672
    private static final DockerImageName RABBIT_IMAGE = DockerImageName.parse("library/rabbitmq:3.8-management")
    private static final String CLUSTER_COOKIE = "test-cluster"
    private static final String RABBIT_CONFIG_PATH = ClassLoader.getSystemResource("rabbit/rabbitmq.conf").path
    private static final String RABBIT_DEFINITIONS_PATH = ClassLoader.getSystemResource("rabbit/definitions.json").path
    private static final Network mqClusterNet = Network.newNetwork()

    protected ApplicationContext applicationContext

    static final String EXCHANGE = "test-exchange"
    static final String QUEUE = "test-durable-queue"
    static final GenericContainer NODE1_CONT = new GenericContainer<>(RABBIT_IMAGE)
    static final GenericContainer NODE2_CONT = new GenericContainer<>(RABBIT_IMAGE)
    static final GenericContainer NODE3_CONT = new GenericContainer<>(RABBIT_IMAGE)
    static int node1Port
    static int node2Port
    static int node3Port

    static {
        getNodePorts()
        log.info("rabbit.conf path: {}", RABBIT_CONFIG_PATH)
        log.info("rabbit definitions path: {}", RABBIT_DEFINITIONS_PATH)
        log.info("rabbit node ports: {}, {}, {}", node1Port, node2Port, node3Port)

        configureContainer(NODE1_CONT, "rabbitmq1", node1Port)
        // first node must boot up completely so that the other nodes can join the new cluster
        NODE1_CONT.waitingFor(Wait.forHealthcheck().withStartupTimeout(Duration.ofMinutes(1)))
        NODE1_CONT.start()
        log.info("first node startup complete")

        configureContainer(NODE2_CONT, "rabbitmq2", node2Port)
        configureContainer(NODE3_CONT, "rabbitmq3", node3Port)
        // node 2 and 3 may start up in parallel as they can join the already existing cluster
        NODE2_CONT.waitingFor(new DoNotWaitStrategy())
        NODE3_CONT.waitingFor(new DoNotWaitStrategy())
        NODE2_CONT.start()
        NODE3_CONT.start()
        new PollingConditions(timeout: 60).eventually {
            assert NODE2_CONT.isHealthy()
            assert NODE3_CONT.isHealthy()
        }
        log.info("cluster startup complete")
    }

    protected void startContext(Map additionalConfig = [:]) {
        Map<String, Object> properties = [
                "spec.name" : getClass().simpleName,
                "rabbitmq.servers.node1.port": node1Port,
                "rabbitmq.servers.node2.port": node2Port,
                "rabbitmq.servers.node3.port": node3Port] << additionalConfig

        log.info("context properties: {}", properties)
        applicationContext = ApplicationContext.run(properties, "test")
    }

    void cleanup() {
        applicationContext?.close()
    }

    private static getNodePorts() {
        try (ServerSocket s1 = new ServerSocket(0)
             ServerSocket s2 = new ServerSocket(0)
             ServerSocket s3 = new ServerSocket(0)) {
            node1Port = s1.localPort
            node2Port = s2.localPort
            node3Port = s3.localPort
        }
    }

    private static configureContainer(GenericContainer mqContainer, String hostname, int nodePort) {
        mqContainer
                .withEnv("RABBITMQ_ERLANG_COOKIE", CLUSTER_COOKIE)
                .withFileSystemBind(RABBIT_CONFIG_PATH, "/etc/rabbitmq/rabbitmq.conf", BindMode.READ_ONLY)
                .withFileSystemBind(RABBIT_DEFINITIONS_PATH, "/etc/rabbitmq/definitions.json", BindMode.READ_ONLY)
                .withNetwork(mqClusterNet)
                .withLogConsumer(new Slf4jLogConsumer(log).withPrefix(hostname))
                .withCreateContainerCmdModifier(cmd -> cmd
                        .withHostName(hostname)
                        .withHealthcheck(new HealthCheck()
                                .withTest(["CMD-SHELL", "rabbitmqctl status"])
                                .withStartPeriod(Duration.ofMinutes(4).toNanos())
                                .withInterval(Duration.ofSeconds(5).toNanos())
                                .withRetries(10)
                                .withTimeout(Duration.ofSeconds(5).toNanos())))
        // Use fixed port binding, because the dynamic port binding would use different port on each container start.
        // These changing ports would make any reconnect attempt impossible, as the client assumes that the broker
        // address does not change.
        addPortBinding(mqContainer, nodePort, AMQP_PORT)
    }

    private static addPortBinding(GenericContainer cont, int hostPort, int contPort) {
        cont.portBindings << String.format("%d:%d/%s",
                hostPort, contPort, InternetProtocol.TCP.toDockerNotation())
    }

    private static class DoNotWaitStrategy extends AbstractWaitStrategy {
        @Override
        protected void waitUntilReady() {
            // NOOP - do not wait
        }
    }
}


<|start_filename|>docs-examples/example-kotlin/src/test/kotlin/io/micronaut/rabbitmq/docs/rpc/ProductListener.kt<|end_filename|>
package io.micronaut.rabbitmq.docs.rpc

import io.micronaut.context.annotation.Requires
// tag::imports[]
import io.micronaut.rabbitmq.annotation.Queue
import io.micronaut.rabbitmq.annotation.RabbitListener
// end::imports[]

@Requires(property = "spec.name", value = "RpcUppercaseSpec")
// tag::clazz[]
@RabbitListener
class ProductListener {

    @Queue("product")
    fun toUpperCase(data: String): String { // <1>
        return data.uppercase() // <2>
    }
}
// end::clazz[]


<|start_filename|>docs-examples/example-kotlin/src/test/kotlin/io/micronaut/rabbitmq/docs/consumer/executor/CustomExecutorSpec.kt<|end_filename|>
package io.micronaut.rabbitmq.docs.consumer.executor

import io.kotest.assertions.timing.eventually
import io.kotest.matchers.shouldBe
import io.micronaut.rabbitmq.AbstractRabbitMQTest
import kotlin.time.Duration
import kotlin.time.ExperimentalTime

@OptIn(ExperimentalTime::class)
class CustomExecutorSpec : AbstractRabbitMQTest({

    val specName = javaClass.simpleName

    given("A basic producer and consumer") {
        val config = AbstractRabbitMQTest.getDefaultConfig(specName)
        config["micronaut.executors.product-listener.type"] = "FIXED"

        val ctx = startContext(config)

        `when`("the message is published") {
            val productListener = ctx.getBean(ProductListener::class.java)

// tag::producer[]
            val productClient = ctx.getBean(ProductClient::class.java)
            productClient.send("custom-executor-test".toByteArray())
// end::producer[]

            then("the message is consumed") {
                eventually(Duration.seconds(10)) {
                    productListener.messageLengths.size shouldBe 1
                    productListener.messageLengths[0] shouldBe "custom-executor-test"
                }
            }
        }

        Thread.sleep(200)
        ctx.stop()
    }
})


<|start_filename|>docs-examples/example-groovy/src/test/groovy/io/micronaut/rabbitmq/docs/consumer/types/ProductListener.groovy<|end_filename|>
package io.micronaut.rabbitmq.docs.consumer.types

import io.micronaut.context.annotation.Requires
// tag::imports[]
import com.rabbitmq.client.BasicProperties
import com.rabbitmq.client.Channel
import com.rabbitmq.client.Envelope
import io.micronaut.rabbitmq.annotation.Queue
import io.micronaut.rabbitmq.annotation.RabbitListener

import java.util.concurrent.CopyOnWriteArrayList
// end::imports[]

@Requires(property = "spec.name", value = "TypeBindingSpec")
// tag::clazz[]
@RabbitListener
class ProductListener {

    CopyOnWriteArrayList<String> messages = []

    @Queue("product")
    void receive(byte[] data,
                 Envelope envelope, // <1>
                 BasicProperties basicProperties, // <2>
                 Channel channel) { // <3>
        messages << "exchange: [$envelope.exchange], routingKey: [$envelope.routingKey], contentType: [$basicProperties.contentType]".toString()
    }
}
// end::clazz[]


<|start_filename|>docs-examples/example-kotlin/src/test/kotlin/io/micronaut/rabbitmq/docs/exchange/CustomExchangeSpec.kt<|end_filename|>
package io.micronaut.rabbitmq.docs.exchange

import io.kotest.assertions.timing.eventually
import io.kotest.matchers.collections.shouldExist
import io.kotest.matchers.shouldBe
import io.micronaut.rabbitmq.AbstractRabbitMQTest
import kotlin.time.Duration
import kotlin.time.ExperimentalTime

@OptIn(ExperimentalTime::class)
class CustomExchangeSpec: AbstractRabbitMQTest({

    val specName = javaClass.simpleName

    given("Using a custom exchange") {
        val ctx = startContext(specName)

        val client = ctx.getBean(AnimalClient::class.java)
        val listener = ctx.getBean(AnimalListener::class.java)

        `when`("the messages are published") {
            client.send(Cat("Whiskers", 9))
            client.send(Cat("Mr. Bigglesworth", 8))
            client.send(Snake("Buttercup", false))
            client.send(Snake("Monty the Python", true))

            then("the messages are received") {
                val messages = listener.receivedAnimals
                eventually(Duration.seconds(10)) {
                    messages.size shouldBe  4
                    messages shouldExist({ animal: Animal ->
                        Cat::class.isInstance(animal) && animal.name == "Whiskers"
                    })
                    messages shouldExist({ animal: Animal ->
                        Cat::class.isInstance(animal) && animal.name == "Mr. Bigglesworth"
                    })
                    messages shouldExist({ animal: Animal ->
                        Snake::class.isInstance(animal) && animal.name == "Buttercup"
                    })
                    messages shouldExist({ animal: Animal ->
                        Snake::class.isInstance(animal) && animal.name == "<NAME> Python"
                    })
                }
            }
        }

        ctx.stop()
    }
})


<|start_filename|>docs-examples/example-groovy/src/test/groovy/io/micronaut/rabbitmq/docs/quickstart/QuickstartSpec.groovy<|end_filename|>
package io.micronaut.rabbitmq.docs.quickstart

import io.micronaut.rabbitmq.AbstractRabbitMQTest

class QuickstartSpec extends AbstractRabbitMQTest {

    void "test product client and listener"() {
        startContext()

        when:
// tag::producer[]
def productClient = applicationContext.getBean(ProductClient)
productClient.send("quickstart".bytes)
// end::producer[]

        ProductListener productListener = applicationContext.getBean(ProductListener)

        then:
        waitFor {
            productListener.messageLengths.size() == 1
            productListener.messageLengths[0] == "quickstart"
        }

        cleanup:
        // Finding that the context is closing the channel before ack is sent
        sleep 200
    }
}


<|start_filename|>docs-examples/example-java/src/test/java/io/micronaut/rabbitmq/docs/consumer/custom/annotation/DeliveryTag.java<|end_filename|>
package io.micronaut.rabbitmq.docs.consumer.custom.annotation;

// tag::imports[]
import io.micronaut.core.bind.annotation.Bindable;

import java.lang.annotation.Documented;
import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
// end::imports[]

// tag::clazz[]
@Documented
@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.PARAMETER})
@Bindable // <1>
public @interface DeliveryTag {
}
// end::clazz[]


<|start_filename|>rabbitmq/src/test/groovy/io/micronaut/rabbitmq/client/ClientErrorSpec.groovy<|end_filename|>
package io.micronaut.rabbitmq.client

import io.micronaut.context.annotation.Requires
import io.micronaut.messaging.annotation.MessageBody
import io.micronaut.messaging.annotation.MessageHeader
import io.micronaut.rabbitmq.AbstractRabbitMQTest
import io.micronaut.rabbitmq.annotation.RabbitClient
import io.micronaut.rabbitmq.annotation.RabbitProperty
import io.micronaut.rabbitmq.exception.RabbitClientException
import org.reactivestreams.Publisher
import org.reactivestreams.Subscriber
import org.reactivestreams.Subscription
import spock.util.concurrent.PollingConditions

class ClientErrorSpec extends AbstractRabbitMQTest {

    void "test no body argument"() {
        startContext()

        Sender sender = applicationContext.getBean(Sender)

        when:
        sender.noBody("abc")

        then:
        RabbitClientException ex = thrown()
        ex.message.startsWith("No valid message body argument found")

        when:
        sender.invalidProperty("abc")

        then:
        ex = thrown()
        ex.message == "Attempted to set property [xyz], but could not match the name to any of the com.rabbitmq.client.BasicProperties"

        when:
        sender.invalidBody(TimeZone.getDefault())

        then:
        ex = thrown()
        ex.message == "Could not find a serializer for the body argument of type [java.util.TimeZone]"
    }

    void "test publishing to an exchange that does not exist"() {
        startContext()

        SenderInvalidExchange sender = applicationContext.getBean(SenderInvalidExchange)
        PollingConditions conditions = new PollingConditions(delay: 3, timeout: 7)

        when:
        Throwable abc
        sender.invalidExchange("abc").subscribe(new Subscriber<Void>() {

            Subscription s

            @Override
            void onSubscribe(Subscription s) {
                this.s = s
                s.request(1)
            }

            @Override
            void onError(Throwable t) {
                abc = t
            }

            @Override void onNext(Void unused) {}
            @Override void onComplete() {}
        })

        then:
        conditions.eventually {
            abc instanceof RabbitClientException
            abc.message.contains("Timed out waiting for publisher confirm")
        }
    }

    @Requires(property = "spec.name", value = "ClientErrorSpec")
    @RabbitClient
    static interface Sender {
        void noBody(@MessageHeader String contentType)
        void invalidProperty(@RabbitProperty String xyz, byte[] body)
        void invalidBody(TimeZone body)
    }

    @Requires(property = "spec.name", value = "ClientErrorSpec")
    @RabbitClient("abc-xyz")
    static interface SenderInvalidExchange {
        Publisher<Void> invalidExchange(@MessageBody String body)
    }
}


<|start_filename|>docs-examples/example-groovy/src/test/groovy/io/micronaut/rabbitmq/docs/consumer/custom/annotation/DeliveryTag.groovy<|end_filename|>
package io.micronaut.rabbitmq.docs.consumer.custom.annotation

// tag::imports[]
import io.micronaut.core.bind.annotation.Bindable

import java.lang.annotation.Documented
import java.lang.annotation.ElementType
import java.lang.annotation.Retention
import java.lang.annotation.RetentionPolicy
import java.lang.annotation.Target
// end::imports[]

// tag::clazz[]
@Documented
@Retention(RetentionPolicy.RUNTIME)
@Target([ElementType.PARAMETER])
@Bindable // <1>
@interface DeliveryTag {
}
// end::clazz[]


<|start_filename|>docs-examples/example-groovy/src/test/groovy/io/micronaut/rabbitmq/docs/publisher/acknowledge/ProductClient.groovy<|end_filename|>
package io.micronaut.rabbitmq.docs.publisher.acknowledge

import io.micronaut.context.annotation.Requires
// tag::imports[]
import io.micronaut.rabbitmq.annotation.Binding
import io.micronaut.rabbitmq.annotation.RabbitClient
import org.reactivestreams.Publisher

import java.util.concurrent.CompletableFuture
// end::imports[]

@Requires(property = "spec.name", value = "PublisherAcknowledgeSpec")
// tag::clazz[]
@RabbitClient
interface ProductClient {

    @Binding("product")
    Publisher<Void> sendPublisher(byte[] data) // <1>

    @Binding("product")
    CompletableFuture<Void> sendFuture(byte[] data) // <2>
}
// end::clazz[]


<|start_filename|>rabbitmq/src/test/groovy/io/micronaut/rabbitmq/connect/RabbitConfigurationSpec.groovy<|end_filename|>
package io.micronaut.rabbitmq.connect

import com.rabbitmq.client.ConnectionFactory
import io.micronaut.context.ApplicationContext
import io.micronaut.context.DefaultApplicationContext
import spock.lang.Specification

class RabbitConfigurationSpec extends Specification {

    void "default rabbit configuration"() {
        given:
        ApplicationContext applicationContext = new DefaultApplicationContext("test").start()

        expect: "connection factory bean is available"
        applicationContext.containsBean(ConnectionFactory)

        when: "when the connection factory is returned"
        ConnectionFactory cf = applicationContext.getBean(ConnectionFactory)

        then: "default configuration is available"
        cf.username == "guest"
        cf.password == "<PASSWORD>"
        cf.virtualHost == "/"
        cf.host == "localhost"
        cf.port == 5672
        cf.requestedChannelMax == 2047
        cf.requestedFrameMax == 0
        cf.requestedHeartbeat == 60
        cf.connectionTimeout == 60_000
        cf.handshakeTimeout == 10_000
        cf.shutdownTimeout == 10_000

        cleanup:
        applicationContext.close()
    }

    void "default rabbit configuration is overridden when configuration properties are passed in"() {
        given:
        ApplicationContext applicationContext = ApplicationContext.run(
                ["rabbitmq.username"           : "guest1",
                 "rabbitmq.password"           : "<PASSWORD>",
                 "rabbitmq.virtualHost"        : "/guest1",
                 "rabbitmq.host"               : "guesthost",
                 "rabbitmq.port"               : 9999,
                 "rabbitmq.requestedChannelMax": 50,
                 "rabbitmq.requestedFrameMax"  : 50,
                 "rabbitmq.requestedHeartbeat" : 50,
                 "rabbitmq.connectionTimeout"  : 50,
                 "rabbitmq.handshakeTimeout"   : 50,
                 "rabbitmq.shutdownTimeout"    : 50],
                "test"
        )

        expect: "connection factory bean is available"
        applicationContext.containsBean(ConnectionFactory)
        applicationContext.containsBean(RabbitConnectionFactoryConfig)

        when: "when the connection factory is returned and values are overridden"
        ConnectionFactory cf = applicationContext.getBean(RabbitConnectionFactoryConfig)

        then: "default configuration is available"
        cf.username == "guest1"
        cf.password == "guest1"
        cf.virtualHost == "/guest1"
        cf.host == "guesthost"
        cf.port == 9999
        cf.requestedChannelMax == 50
        cf.requestedFrameMax == 50
        cf.requestedHeartbeat == 50
        cf.connectionTimeout == 50
        cf.handshakeTimeout == 50
        cf.shutdownTimeout == 50

        cleanup:
        applicationContext.close()
    }
}


<|start_filename|>rabbitmq/src/test/groovy/io/micronaut/rabbitmq/connect/FailOverConnectionSpec.groovy<|end_filename|>
package io.micronaut.rabbitmq.connect

import com.rabbitmq.client.Connection
import io.micronaut.rabbitmq.AbstractRabbitMQTest

class FailOverConnectionSpec extends AbstractRabbitMQTest {

    void "test multiple addresses"() {
        when:
        startContext('rabbitmq.addresses': ['localhost:62354', "localhost:${rabbitContainer.getMappedPort(5672)}"])

        then:
        applicationContext.getBean(Connection).port == rabbitContainer.getMappedPort(5672)
    }
}


<|start_filename|>docs-examples/example-kotlin/src/test/kotlin/io/micronaut/rabbitmq/docs/properties/ProductListener.kt<|end_filename|>
package io.micronaut.rabbitmq.docs.properties

import io.micronaut.context.annotation.Requires
// tag::imports[]
import io.micronaut.rabbitmq.annotation.Queue
import io.micronaut.rabbitmq.annotation.RabbitListener
import io.micronaut.rabbitmq.annotation.RabbitProperty
import java.util.Collections
// end::imports[]

@Requires(property = "spec.name", value = "PropertiesSpec")
// tag::clazz[]
@RabbitListener
class ProductListener {

    val messageProperties: MutableList<String> = Collections.synchronizedList(ArrayList())

    @Queue("product")
    @RabbitProperty(name = "x-priority", value = "10", type = Integer::class) // <1>
    fun receive(data: ByteArray,
                @RabbitProperty("userId") user: String, // <2>
                @RabbitProperty contentType: String?, // <3>
                appId: String) { // <4>
        messageProperties.add("$user|$contentType|$appId")
    }
}
// end::clazz[]


<|start_filename|>rabbitmq/src/test/groovy/io/micronaut/rabbitmq/annotation/HeaderBindingSpec.groovy<|end_filename|>
package io.micronaut.rabbitmq.annotation

import io.micronaut.context.annotation.Requires
import io.micronaut.messaging.annotation.MessageHeader
import io.micronaut.rabbitmq.AbstractRabbitMQTest

class HeaderBindingSpec extends AbstractRabbitMQTest {

    void "test simple producing and consuming with the header annotation"() {
        startContext()

        MyProducer producer = applicationContext.getBean(MyProducer)
        MyConsumer consumer = applicationContext.getBean(MyConsumer)

        when:
        producer.go(new Person(name: "abc"), "some header")

        then:
        waitFor {
            consumer.messages.size() == 1
            consumer.messages.keySet()[0].name == "abc"
            consumer.messages.values()[0] == "some header|static header"
        }
    }

    static class Person {
        String name
    }

    @Requires(property = "spec.name", value = "HeaderBindingSpec")
    @RabbitClient
    static interface MyProducer {
        @Binding("header")
        @MessageHeader(name = "static", value = "static header")
        void go(Person data, @MessageHeader String myHeader)
    }

    @Requires(property = "spec.name", value = "HeaderBindingSpec")
    @RabbitListener
    static class MyConsumer {

        static Map<Person, String> messages = [:]

        @Queue("header")
        void listen(Person data, @MessageHeader String myHeader,
                    @MessageHeader("static") String otherHeader) {
            messages[data] = myHeader + '|' + otherHeader
        }
    }
}


<|start_filename|>docs-examples/example-kotlin/src/test/kotlin/io/micronaut/rabbitmq/docs/parameters/ProductClient.kt<|end_filename|>
package io.micronaut.rabbitmq.docs.parameters

import io.micronaut.context.annotation.Requires
// tag::imports[]
import io.micronaut.rabbitmq.annotation.Binding
import io.micronaut.rabbitmq.annotation.RabbitClient
// end::imports[]

@Requires(property = "spec.name", value = "BindingSpec")
// tag::clazz[]
@RabbitClient
interface ProductClient {

    @Binding("product") // <1>
    fun send(data: ByteArray)

    fun send(@Binding binding: String, data: ByteArray)  // <2>
}
// end::clazz[]


<|start_filename|>rabbitmq/src/main/java/io/micronaut/rabbitmq/bind/RabbitPropertyBinder.java<|end_filename|>
/*
 * Copyright 2017-2020 original 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 io.micronaut.rabbitmq.bind;

import com.rabbitmq.client.AMQP;
import io.micronaut.core.convert.ArgumentConversionContext;
import io.micronaut.core.convert.ConversionService;
import io.micronaut.rabbitmq.annotation.RabbitProperty;
import jakarta.inject.Singleton;

import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
import java.util.function.Function;

/**
 * Binds an argument of with the {@link RabbitProperty} annotation from the {@link RabbitConsumerState}.
 *
 * @author <NAME>
 * @since 1.1.0
 */
@Singleton
public class RabbitPropertyBinder implements RabbitAnnotatedArgumentBinder<RabbitProperty> {

    private final Map<String, Function<AMQP.BasicProperties, Object>> properties = new HashMap<>();
    private final ConversionService conversionService;

    /**
     * Default constructor.
     *
     * @param conversionService The conversion service to convert the body
     */
    public RabbitPropertyBinder(ConversionService conversionService) {
        this.conversionService = conversionService;
        properties.put("contentType", AMQP.BasicProperties::getContentType);
        properties.put("contentEncoding", AMQP.BasicProperties::getContentEncoding);
        properties.put("deliveryMode", AMQP.BasicProperties::getDeliveryMode);
        properties.put("priority", AMQP.BasicProperties::getPriority);
        properties.put("correlationId", AMQP.BasicProperties::getCorrelationId);
        properties.put("replyTo", AMQP.BasicProperties::getReplyTo);
        properties.put("expiration", AMQP.BasicProperties::getExpiration);
        properties.put("messageId", AMQP.BasicProperties::getMessageId);
        properties.put("timestamp", AMQP.BasicProperties::getTimestamp);
        properties.put("type", AMQP.BasicProperties::getType);
        properties.put("userId", AMQP.BasicProperties::getUserId);
        properties.put("appId", AMQP.BasicProperties::getAppId);
        properties.put("clusterId", AMQP.BasicProperties::getClusterId);
    }

    @Override
    public Class<RabbitProperty> getAnnotationType() {
        return RabbitProperty.class;
    }

    @Override
    public BindingResult<Object> bind(ArgumentConversionContext<Object> context, RabbitConsumerState messageState) {
        Optional<Object> property = Optional.ofNullable(properties.get(getParameterName(context)))
                .map(f -> f.apply(messageState.getProperties()))
                .flatMap(prop -> conversionService.convert(prop, context));
        return () -> property;
    }

    /**
     * @param context The argument context
     * @return True if this binder can bind the argument
     */
    public boolean supports(ArgumentConversionContext<Object> context) {
        return properties.containsKey(getParameterName(context));
    }

    private String getParameterName(ArgumentConversionContext<Object> context) {
        return context.getAnnotationMetadata().getValue(RabbitProperty.class, String.class).orElse(context.getArgument().getName());
    }
}


<|start_filename|>rabbitmq/src/main/java/io/micronaut/rabbitmq/connect/ChannelPool.java<|end_filename|>
/*
 * Copyright 2017-2020 original 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 io.micronaut.rabbitmq.connect;

import com.rabbitmq.client.Channel;
import io.micronaut.core.naming.Named;

import java.io.IOException;

/**
 * A pool of {@link Channel}s to allow for channels to be shared across
 * threads but not at the same time. A channel retrieved from the pool should
 * not be accessible to any other threads until the channel is returned from
 * the pool.
 *
 * @author <NAME>
 * @since 1.1.0
 */
public interface ChannelPool extends Named {

    /**
     * The default delay to apply for recovery channel getter.
     */
    int DEFAULT_RECOVERY_DELAY = 5000;

    /**
     * Retrieves a channel from the pool. The channel must be returned to the
     * pool after it is no longer being used.
     *
     * @return The channel
     * @throws IOException If a channel needed to be created and encountered an error
     */
    Channel getChannel() throws IOException;

    /**
     * Retrieves a channel from the pool after blocking the thread for a delay period defined by the
     * {@link com.rabbitmq.client.ConnectionFactory#getRecoveryDelayHandler() RecoveryDelayHandler}
     * of the connection for this pool.
     *
     * @param recoveryAttempts the number of recovery attempts so far
     * @return a channel from the pool
     * @throws IOException if a channel needed to be created and encountered an error
     * @throws InterruptedException if the thread was interrupted during the delay period
     */
    default Channel getChannelWithRecoveringDelay(int recoveryAttempts) throws IOException, InterruptedException {
        Thread.sleep(DEFAULT_RECOVERY_DELAY);
        return getChannel();
    }

    /**
     * Returns whether {@link com.rabbitmq.client.ConnectionFactory#isTopologyRecoveryEnabled() topology recovery}
     * is enabled for the connection of this pool.
     *
     * @return true by default
     */
    default boolean isTopologyRecoveryEnabled() {
        return true;
    }

    /**
     * Returns a channel to the pool. No further use of the channel
     * is allowed by the returner.
     *
     * @param channel The channel
     */
    void returnChannel(Channel channel);
}


<|start_filename|>rabbitmq/src/test/groovy/io/micronaut/rabbitmq/connect/ChannelPoolListener.groovy<|end_filename|>
package io.micronaut.rabbitmq.connect

import com.rabbitmq.client.Channel
import jakarta.inject.Singleton

@Singleton
class ChannelPoolListener extends ChannelInitializer {

    @Override
    void initialize(Channel channel, String name) throws IOException {
        channel.queueDeclare("abc", false, false, false, [:])
        channel.queueDeclare("pojo", false, false, false, [:])
        channel.queueDeclare("pojo-list", false, false, false, [:])
        channel.queueDeclare("simple", false, false, false, [:])
        channel.queueDeclare("simple-list", false, false, false, [:])
        channel.queueDeclare("simple-header", false, false, false, [:])
        channel.queueDeclare("error", false, false, false, [:])
        channel.queueDeclare("header", false, false, false, [:])
        channel.queueDeclare("property", false, false, false, [:])
        channel.queueDeclare("type", false, false, false, [:])
        channel.queueDeclare("boolean", false, false, false, [:])
        channel.queueDeclare("product", false, false, false, [:])
        channel.queueDeclare("rpc", false, false, false, [:])

        channel.exchangeDeclare("animals", "headers", false)
        channel.queueDeclare("dogs", false, false, false, null)
        channel.queueDeclare("cats", false, false, false, null)
        channel.queueBind("cats", "animals", "", ["x-match": "all", animalType: "Cat"])

        channel.queueBind("dogs", "animals", "", ["x-match": "all", animalType: "Dog"])
    }
}


<|start_filename|>rabbitmq/src/test/groovy/io/micronaut/rabbitmq/listener/MissingHeaderSpec.groovy<|end_filename|>
package io.micronaut.rabbitmq.listener

import io.micronaut.context.annotation.Requires
import io.micronaut.core.annotation.Nullable
import io.micronaut.messaging.annotation.MessageHeader
import io.micronaut.rabbitmq.AbstractRabbitMQTest
import io.micronaut.rabbitmq.annotation.Binding
import io.micronaut.rabbitmq.annotation.Queue
import io.micronaut.rabbitmq.annotation.RabbitClient
import io.micronaut.rabbitmq.annotation.RabbitListener
import io.micronaut.rabbitmq.exception.RabbitListenerException
import io.micronaut.rabbitmq.exception.RabbitListenerExceptionHandler

import java.util.concurrent.CopyOnWriteArrayList

class MissingHeaderSpec extends AbstractRabbitMQTest {

    void "test a null header value"() {
        startContext()

        MyProducer producer = applicationContext.getBean(MyProducer)
        producer.go("abc")

        when:
        MyConsumer consumer = applicationContext.getBean(MyConsumer)

        then:
        waitFor {
            consumer.headers.size() == 0
            consumer.errors.size() == 0
            consumer.bodies.size() == 1
        }
    }

    @Requires(property = "spec.name", value = "MissingHeaderSpec")
    @RabbitClient
    static interface MyProducer {
        @Binding("simple-header")
        void go(String data)
    }

    @Requires(property = "spec.name", value = "MissingHeaderSpec")
    @RabbitListener
    static class MyConsumer implements RabbitListenerExceptionHandler {

        static CopyOnWriteArrayList<String> headers = []
        static CopyOnWriteArrayList<RabbitListenerException> errors = []
        static CopyOnWriteArrayList<String> bodies = []

        @Queue("simple-header")
        void listen(@Nullable @MessageHeader("X-Header") String header, String data) {
            if (header != null) headers << header
            bodies << data
        }

        @Override
        void handle(RabbitListenerException e) {
            errors << e
            e.printStackTrace()
        }
    }
}


<|start_filename|>docs-examples/example-java/src/test/java/io/micronaut/rabbitmq/docs/headers/ProductClient.java<|end_filename|>
package io.micronaut.rabbitmq.docs.headers;

import io.micronaut.context.annotation.Requires;
// tag::imports[]
import io.micronaut.messaging.annotation.MessageHeader;
import io.micronaut.rabbitmq.annotation.Binding;
import io.micronaut.rabbitmq.annotation.RabbitClient;
import io.micronaut.rabbitmq.annotation.RabbitHeaders;

import java.util.Map;
// end::imports[]

@Requires(property = "spec.name", value = "HeadersSpec")
// tag::clazz[]
@RabbitClient
@MessageHeader(name = "x-product-sealed", value = "true") // <1>
@MessageHeader(name = "productSize", value = "large")
public interface ProductClient {

    @Binding("product")
    @MessageHeader(name = "x-product-count", value = "10") // <2>
    @MessageHeader(name = "productSize", value = "small")
    void send(byte[] data);

    @Binding("product")
    void send(@MessageHeader String productSize, // <3>
              @MessageHeader("x-product-count") Long count,
              byte[] data);

    @Binding("product")
    void send(@RabbitHeaders Map<String, Object> headers, // <4>
              byte[] data);
}
// end::clazz[]


<|start_filename|>docs-examples/example-kotlin/src/test/kotlin/io/micronaut/rabbitmq/docs/publisher/acknowledge/PublisherAcknowledgeSpec.kt<|end_filename|>
package io.micronaut.rabbitmq.docs.publisher.acknowledge;

import io.kotest.assertions.timing.eventually
import io.kotest.matchers.shouldBe
import io.micronaut.rabbitmq.AbstractRabbitMQTest
import kotlinx.coroutines.async
import org.reactivestreams.Subscriber
import org.reactivestreams.Subscription
import java.util.concurrent.atomic.AtomicInteger
import kotlin.time.Duration
import kotlin.time.ExperimentalTime

@OptIn(ExperimentalTime::class)
class PublisherAcknowledgeSpec : AbstractRabbitMQTest({

    val specName = javaClass.simpleName

    given("Publisher acknowledgement") {
        val ctx = startContext(specName)
        val successCount = AtomicInteger(0)
        val errorCount = AtomicInteger(0)

        `when`("The messages are published") {
            // tag::producer[]
            val productClient = ctx.getBean(ProductClient::class.java)
            val publisher = productClient.sendPublisher("publisher body".toByteArray())
            val future = productClient.sendFuture("future body".toByteArray())
            val deferred = async {
                productClient.sendSuspend("suspend body".toByteArray())
            }

            val subscriber = (object: Subscriber<Void> {
                override fun onSubscribe(subscription: Subscription) { }

                override fun onNext(aVoid: Void) {
                    throw UnsupportedOperationException("Should never be called")
                }

                override fun onError(throwable: Throwable) {
                    // if an error occurs
                    errorCount.incrementAndGet()
                }

                override fun onComplete() {
                    // if the publish was acknowledged
                    successCount.incrementAndGet()
                }
            })
            publisher.subscribe(subscriber)
            future.handle { _, t ->
                if (t == null) {
                    successCount.incrementAndGet()
                } else {
                    errorCount.incrementAndGet()
                }
            }
            deferred.invokeOnCompletion {
                if (it == null) {
                    successCount.incrementAndGet()
                } else {
                    errorCount.incrementAndGet()
                }
            }
// end::producer[]

            then("The messages are published") {
                eventually(Duration.seconds(10)) {
                    errorCount.get() shouldBe 0
                    successCount.get() shouldBe 3
                }
            }
        }

        ctx.stop()
    }
})


<|start_filename|>rabbitmq/src/test/groovy/io/micronaut/rabbitmq/annotation/BasicAopSpec.groovy<|end_filename|>
package io.micronaut.rabbitmq.annotation

import io.micronaut.context.annotation.Requires
import io.micronaut.messaging.annotation.MessageBody
import io.micronaut.rabbitmq.AbstractRabbitMQTest
import org.reactivestreams.Publisher
import org.reactivestreams.Subscriber
import org.reactivestreams.Subscription

class BasicAopSpec extends AbstractRabbitMQTest {

    void "test simple producing and consuming"() {
        startContext()

        MyProducer producer = applicationContext.getBean(MyProducer)
        MyConsumer consumer = applicationContext.getBean(MyConsumer)

        when:
        producer.go("abc".bytes)
        boolean success
        producer.goConfirm("def".bytes)
                .subscribe(new Subscriber<Void>() {

                    Subscription s

                    @Override
                    void onSubscribe(Subscription s) {
                        this.s = s
                        s.request(1)
                    }

                    @Override
                    void onNext(Void unused) {
                        s.request(1)
                    }

                    @Override
                    void onError(Throwable t) {}

                    @Override
                    void onComplete() {
                        success = true
                    }
                })

        then:
        waitFor {
            success
            consumer.messages.size() == 2
            consumer.messages[0] == "abc".bytes
            consumer.messages[1] == "def".bytes
        }
    }

    @Requires(property = "spec.name", value = "BasicAopSpec")
    @RabbitClient
    static interface MyProducer {

        @Binding("abc")
        void go(@MessageBody byte[] data)

        @Binding("abc")
        Publisher<Void> goConfirm(byte[] data)
    }

    @Requires(property = "spec.name", value = "BasicAopSpec")
    @RabbitListener
    static class MyConsumer {

        static List<byte[]> messages = []

        @Queue("abc")
        void listen(byte[] data) {
            messages << data
        }
    }
}


<|start_filename|>docs-examples/example-java/src/test/java/io/micronaut/rabbitmq/docs/rpc/RpcUppercaseSpec.java<|end_filename|>
package io.micronaut.rabbitmq.docs.rpc;

import io.micronaut.rabbitmq.AbstractRabbitMQTest;
import org.junit.jupiter.api.Test;
import reactor.core.publisher.Mono;

import static org.junit.jupiter.api.Assertions.assertEquals;

public class RpcUppercaseSpec extends AbstractRabbitMQTest {

    @Test
    void testProductClientAndListener() {
        startContext();

// tag::producer[]
ProductClient productClient = applicationContext.getBean(ProductClient.class);
assertEquals("RPC", productClient.send("rpc"));
assertEquals("HELLO", Mono.from(productClient.sendReactive("hello")).block());
// end::producer[]
    }
}


<|start_filename|>docs-examples/example-java/src/test/java/io/micronaut/rabbitmq/docs/publisher/acknowledge/ProductClient.java<|end_filename|>
package io.micronaut.rabbitmq.docs.publisher.acknowledge;

import io.micronaut.context.annotation.Requires;
// tag::imports[]
import io.micronaut.rabbitmq.annotation.Binding;
import io.micronaut.rabbitmq.annotation.RabbitClient;
import org.reactivestreams.Publisher;

import java.util.concurrent.CompletableFuture;
// end::imports[]

@Requires(property = "spec.name", value = "PublisherAcknowledgeSpec")
// tag::clazz[]
@RabbitClient
public interface ProductClient {

    @Binding("product")
    Publisher<Void> sendPublisher(byte[] data); // <1>

    @Binding("product")
    CompletableFuture<Void> sendFuture(byte[] data); // <2>
}
// end::clazz[]


<|start_filename|>rabbitmq/src/main/java/io/micronaut/rabbitmq/intercept/RabbitMQConsumerAdvice.java<|end_filename|>
/*
 * Copyright 2017-2020 original 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 io.micronaut.rabbitmq.intercept;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.AlreadyClosedException;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Delivery;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.ShutdownSignalException;
import com.rabbitmq.client.RecoverableChannel;
import io.micronaut.context.BeanContext;
import io.micronaut.context.processor.ExecutableMethodProcessor;
import io.micronaut.core.annotation.AnnotationValue;
import io.micronaut.core.bind.BoundExecutable;
import io.micronaut.core.bind.DefaultExecutableBinder;
import io.micronaut.core.convert.ConversionService;
import io.micronaut.core.util.StringUtils;
import io.micronaut.inject.BeanDefinition;
import io.micronaut.inject.ExecutableMethod;
import io.micronaut.inject.qualifiers.Qualifiers;
import io.micronaut.messaging.Acknowledgement;
import io.micronaut.messaging.exceptions.MessageAcknowledgementException;
import io.micronaut.messaging.exceptions.MessageListenerException;
import io.micronaut.rabbitmq.annotation.Queue;
import io.micronaut.rabbitmq.annotation.RabbitConnection;
import io.micronaut.rabbitmq.annotation.RabbitProperty;
import io.micronaut.rabbitmq.bind.RabbitBinderRegistry;
import io.micronaut.rabbitmq.bind.RabbitConsumerState;
import io.micronaut.rabbitmq.bind.RabbitMessageCloseable;
import io.micronaut.rabbitmq.connect.ChannelPool;
import io.micronaut.rabbitmq.exception.RabbitListenerException;
import io.micronaut.rabbitmq.exception.RabbitListenerExceptionHandler;
import io.micronaut.rabbitmq.serdes.RabbitMessageSerDes;
import io.micronaut.rabbitmq.serdes.RabbitMessageSerDesRegistry;
import jakarta.inject.Singleton;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.PreDestroy;
import java.io.IOException;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * An {@link ExecutableMethodProcessor} that will process all beans annotated
 * with {@link io.micronaut.rabbitmq.annotation.RabbitListener} and create and subscribe the relevant methods
 * as consumers to RabbitMQ queues.
 *
 * @author <NAME>
 * @since 1.1.0
 */
@Singleton
public class RabbitMQConsumerAdvice implements ExecutableMethodProcessor<Queue>, AutoCloseable {

    private static final Logger LOG = LoggerFactory.getLogger(RabbitMQConsumerAdvice.class);

    private final BeanContext beanContext;
    private final RabbitBinderRegistry binderRegistry;
    private final RabbitListenerExceptionHandler exceptionHandler;
    private final RabbitMessageSerDesRegistry serDesRegistry;
    private final ConversionService<?> conversionService;
    private final Map<String, ChannelPool> channelPools;
    private final List<RecoverableConsumerWrapper> consumers = new CopyOnWriteArrayList<>();

    /**
     * Default constructor.
     *
     * @param beanContext       The bean context
     * @param binderRegistry    The registry to bind arguments to the method
     * @param exceptionHandler  The exception handler to use if the consumer isn't a handler
     * @param serDesRegistry    The serialization/deserialization registry
     * @param conversionService The service to convert consume argument values
     * @param channelPools      The channel pools to retrieve channels
     */
    public RabbitMQConsumerAdvice(BeanContext beanContext,
                                  RabbitBinderRegistry binderRegistry,
                                  RabbitListenerExceptionHandler exceptionHandler,
                                  RabbitMessageSerDesRegistry serDesRegistry,
                                  ConversionService<?> conversionService,
                                  List<ChannelPool> channelPools) {
        this.beanContext = beanContext;
        this.binderRegistry = binderRegistry;
        this.exceptionHandler = exceptionHandler;
        this.serDesRegistry = serDesRegistry;
        this.conversionService = conversionService;
        this.channelPools = new HashMap<>(channelPools.size());
        for (ChannelPool cp: channelPools) {
            this.channelPools.put(cp.getName(), cp);
        }
    }

    @Override
    public void process(BeanDefinition<?> beanDefinition, ExecutableMethod<?, ?> method) {

        AnnotationValue<Queue> queueAnn = method.getAnnotation(Queue.class);

        if (queueAnn != null) {
            String queue = queueAnn.getRequiredValue(String.class);

            String methodTag = method.getDeclaringType().getSimpleName() + '#' + method;

            boolean reQueue = queueAnn.getRequiredValue("reQueue", boolean.class);
            boolean exclusive = queueAnn.getRequiredValue("exclusive", boolean.class);

            boolean hasAckArg = Arrays.stream(method.getArguments())
                    .anyMatch(arg -> Acknowledgement.class.isAssignableFrom(arg.getType()));

            Integer prefetch = queueAnn.get("prefetch", Integer.class).orElse(null);
            int numberOfConsumers = queueAnn.intValue("numberOfConsumers").orElse(1);

            ChannelPool channelPool = getChannelPool(method);
            ExecutorService executorService = getExecutorService(method);

            Map<String, Object> arguments = retrieveArguments(method);
            Object bean = getExecutableMethodBean(beanDefinition, method);
            boolean isVoid = method.getReturnType().isVoid();

            DefaultExecutableBinder<RabbitConsumerState> binder = new DefaultExecutableBinder<>();

            DeliverCallback deliverCallback = (channel, message) -> {
                final RabbitConsumerState state = new RabbitConsumerState(message.getEnvelope(),
                        message.getProperties(), message.getBody(), channel);

                BoundExecutable boundExecutable = null;
                try {
                    boundExecutable = binder.bind(method, binderRegistry, state);
                } catch (Throwable e) {
                    handleException(new RabbitListenerException("An error occurred binding the message to the method",
                            e, bean, state));
                }
                try {
                    if (boundExecutable != null) {
                        try (RabbitMessageCloseable closeable = new RabbitMessageCloseable(state, false, reQueue)
                                .withAcknowledge(hasAckArg ? null : false)) {
                            Object returnedValue = boundExecutable.invoke(bean);

                            String replyTo = message.getProperties().getReplyTo();
                            if (!isVoid && StringUtils.isNotEmpty(replyTo)) {
                                MutableBasicProperties replyProps = new MutableBasicProperties();
                                replyProps.setCorrelationId(message.getProperties().getCorrelationId());

                                byte[] converted = null;
                                if (returnedValue != null) {
                                    RabbitMessageSerDes serDes = serDesRegistry
                                            .findSerdes(method.getReturnType().asArgument())
                                            .map(RabbitMessageSerDes.class::cast)
                                            .orElseThrow(() -> new RabbitListenerException(String.format(
                                                    "Could not find a serializer for the body argument of type [%s]",
                                                    returnedValue.getClass().getName()), bean, state));

                                    converted = serDes.serialize(returnedValue, replyProps);
                                }

                                channel.basicPublish("", replyTo, replyProps.toBasicProperties(), converted);
                            }

                            if (!hasAckArg) {
                                closeable.withAcknowledge(true);
                            }
                        } catch (MessageAcknowledgementException e) {
                            throw e;
                        } catch (RabbitListenerException e) {
                            handleException(e);
                        } catch (Throwable e) {
                            handleException(new RabbitListenerException("An error occurred executing the listener",
                                    e, bean, state));
                        }
                    } else {
                        new RabbitMessageCloseable(state, false, reQueue).withAcknowledge(false).close();
                    }
                } catch (MessageAcknowledgementException e) {
                    handleException(new RabbitListenerException(e.getMessage(), e, bean, state));
                }
            };

            try {
                for (int idx = 0; idx < numberOfConsumers; idx++) {
                    String consumerTag = methodTag + "[" + idx + "]";
                    LOG.debug("Registering a consumer to queue [{}] with client tag [{}]", queue, consumerTag);
                    consumers.add(new RecoverableConsumerWrapper(queue, consumerTag, executorService,
                            exclusive, arguments, channelPool, prefetch, deliverCallback));
                }
            } catch (Throwable e) {
                handleException(new RabbitListenerException("An error occurred subscribing to a queue", e, bean, null));
            }
        }
    }

    private ChannelPool getChannelPool(ExecutableMethod<?, ?> method) {
        String connection = method.stringValue(RabbitConnection.class, "connection")
                .orElse(RabbitConnection.DEFAULT_CONNECTION);

        return Optional.ofNullable(channelPools.get(connection))
                .orElseThrow(() -> new MessageListenerException(
                        String.format("Failed to find a channel pool named [%s] to register a listener", connection)));
    }

    private static void setChannelPrefetch(Integer prefetch, Channel channel) {
        try {
            if (prefetch != null) {
                channel.basicQos(prefetch);
            }
        } catch (IOException e) {
            throw new MessageListenerException(String.format("Failed to set a prefetch count of [%s] on the channel",
                    prefetch), e);
        }
    }

    private Map<String, Object> retrieveArguments(ExecutableMethod<?, ?> method) {
        Map<String, Object> arguments = new HashMap<>();

        List<AnnotationValue<RabbitProperty>> propertyAnnotations = method.getAnnotationValuesByType(RabbitProperty.class);
        Collections.reverse(propertyAnnotations); //set the values in the class first so methods can override
        propertyAnnotations.forEach((prop) -> {
            String name = prop.getRequiredValue("name", String.class);
            String value = prop.getValue(String.class).orElse(null);
            Class<?> type = prop.get("type", Class.class).orElse(null);

            if (StringUtils.isNotEmpty(name) && StringUtils.isNotEmpty(value)) {
                if (type != null && type != Void.class) {
                    Optional<?> converted = conversionService.convert(value, type);
                    if (converted.isPresent()) {
                        arguments.put(name, converted.get());
                    } else {
                        throw new MessageListenerException(String.format("Could not convert the argument [%s] to the required type [%s]", name, type));
                    }
                } else {
                    arguments.put(name, value);
                }
            }
        });
        return arguments;
    }

    private Object getExecutableMethodBean(BeanDefinition<?> beanDefinition, ExecutableMethod<?, ?> method) {
        io.micronaut.context.Qualifier<Object> qualifier = beanDefinition
                .getAnnotationNameByStereotype("javax.inject.Qualifier")
                .map(type -> Qualifiers.byAnnotation(beanDefinition, type))
                .orElse(null);

        Class<Object> beanType = (Class<Object>) beanDefinition.getBeanType();
        Object bean = beanContext.findBean(beanType, qualifier).orElseThrow(() -> new MessageListenerException("Could not find the bean to execute the method " + method));
        return bean;
    }

    private ExecutorService getExecutorService(ExecutableMethod<?, ?> method) {
        String executor = method.stringValue(RabbitConnection.class, "executor").orElse(null);
        if (executor != null) {
            return beanContext.findBean(ExecutorService.class, Qualifiers.byName(executor))
                    .orElseThrow(() -> new MessageListenerException(String.format("Could not find the executor service [%s] specified for the method [%s]", executor, method)));
        }
        return null;
    }

    @PreDestroy
    @Override
    public void close() throws Exception {
        consumers.forEach(RecoverableConsumerWrapper::cancel);
    }

    private void handleException(RabbitListenerException exception) {
        Object bean = exception.getListener();
        if (bean instanceof RabbitListenerExceptionHandler) {
            ((RabbitListenerExceptionHandler) bean).handle(exception);
        } else {
            exceptionHandler.handle(exception);
        }
    }

    /**
     * Callback interface to be notified when a message is delivered.
     */
    @FunctionalInterface
    private interface DeliverCallback {
        /**
         * Inspired by {@link com.rabbitmq.client.DeliverCallback#handle(String, Delivery)}.
         *
         * @param channel that is used to register the consumer for this callback
         * @param message the delivered message
         */
        void handle(Channel channel, Delivery message);
    }

    /**
     * This wrapper around a {@link com.rabbitmq.client.DefaultConsumer} allows to handle the different signals from
     * the underlying channel and to react accordingly.
     * <p>
     * If the consumer is canceled due to an external event (like an unavailable queue) we will try to recover from it.
     * Exceptions that are caused by the consumer itself will not trigger the recovery process. In such a case the
     * consumer will no longer receive any messages.
     *
     * @see com.rabbitmq.client.Consumer#handleShutdownSignal(String, ShutdownSignalException)
     * @see com.rabbitmq.client.Consumer#handleCancel(String)
     */
    private class RecoverableConsumerWrapper {
        final String consumerTag;
        private final ExecutorService executorService;
        private final String queue;
        private final boolean exclusive;
        private final Map<String, Object> arguments;
        private final ChannelPool channelPool;
        private final Integer prefetch;
        private final DeliverCallback deliverCallback;
        private com.rabbitmq.client.DefaultConsumer consumer;
        private boolean canceled = false;
        private final AtomicInteger handlingDeliveryCount = new AtomicInteger();

        /**
         * Create the consumer and register ({@code Channel.basicConsume}) it with a dedicated channel from the
         * provided pool.
         *
         * @throws IOException in case no channel is available or the registration of the consumer fails
         */
        RecoverableConsumerWrapper(String queue, String consumerTag, ExecutorService executorService, boolean exclusive,
                Map<String, Object> arguments, ChannelPool channelPool, Integer prefetch, DeliverCallback deliverCallback)
                throws IOException {
            this.queue = queue;
            this.consumerTag = consumerTag;
            this.executorService = executorService;
            this.exclusive = exclusive;
            this.arguments = arguments;
            this.channelPool = channelPool;
            this.prefetch = prefetch;
            this.deliverCallback = deliverCallback;

            Channel channel = null;
            try {
                channel = channelPool.getChannel();
                consumer = createConsumer(channel);
            } catch (IOException e) {
                if (channel != null) {
                    channelPool.returnChannel(channel);
                }
                throw e;
            }
        }

        /**
         * Cancel the consumer and return the associated channel to the pool.
         */
        public synchronized void cancel() {
            canceled = true;
            if (consumer == null) {
                return;
            }

            Channel channel = consumer.getChannel();
            try {
                channel.basicCancel(consumerTag);
            } catch (IOException | AlreadyClosedException e) {
                // ignore
            }
            try {
                while (handlingDeliveryCount.get() > 0) {
                    this.wait(500);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            } finally {
                consumer = null;
                consumers.remove(this);
                channelPool.returnChannel(channel);
            }
        }

        private void performConsumerRecovery() {
            com.rabbitmq.client.DefaultConsumer recoveredConsumer = null;
            int recoveryAttempts = 0;
            while (recoveredConsumer == null) {
                Channel channel = null;
                try {
                    synchronized (this) {
                        if (canceled) {
                            return;
                        }
                        LOG.debug("consumer [{}] recovery attempt: {}", consumerTag, recoveryAttempts + 1);
                        channel = channelPool.getChannelWithRecoveringDelay(recoveryAttempts++);
                        recoveredConsumer = createConsumer(channel);
                        consumer = recoveredConsumer;
                    }
                } catch (IOException e) {
                    if (channel != null) {
                        channelPool.returnChannel(channel);
                    }
                    LOG.warn("Recovery attempt {} for consumer [{}] failed, will retry.",
                            recoveryAttempts, consumerTag, e);
                } catch (InterruptedException e) {
                    LOG.warn("The consumer [{}] recovery was interrupted. The consumer will not recover.",
                            consumerTag, e);
                    Thread.currentThread().interrupt();
                    return;
                }
            }
            LOG.info("consumer [{}] recovered", consumerTag);
        }

        private com.rabbitmq.client.DefaultConsumer createConsumer(Channel channel) throws IOException {
            setChannelPrefetch(prefetch, channel);

            com.rabbitmq.client.DefaultConsumer consumer = new com.rabbitmq.client.DefaultConsumer(channel) {
                /**
                 * The consumer was irregular terminated. This may be caused by a deleted or temporary unavailable
                 * queue.
                 * <p>
                 * This kind of infrastructure failure may happen due to RabbitMQ cluster node restarts or other
                 * external actions. The client application will most likely be unable to restore the infrastructure,
                 * but it should return to normal operation as soon as the external infrastructure problem is solved.
                 * e.g. the RabbitMQ node restart is complete and the queue is available again.
                 */
                @Override
                public void handleCancel(String consumerTag) throws IOException {
                    synchronized (RecoverableConsumerWrapper.this) {
                        RecoverableConsumerWrapper.this.consumer = null;
                        channelPool.returnChannel(getChannel());
                    }

                    if (channelPool.isTopologyRecoveryEnabled() && getChannel() instanceof RecoverableChannel) {
                        LOG.warn("The consumer [{}] subscription was canceled, a recovery will be tried.",
                                consumerTag);
                        performConsumerRecovery();
                    } else {
                        LOG.warn("The RabbitMQ consumer [{}] was canceled. Recovery is not enabled. It will no longer receive messages",
                                consumerTag);
                        cancel();
                    }
                }

                /**
                 * A shutdown signal from the channel or the underlying connection does not always imply that the
                 * consumer is no longer usable. If the automatic topology recovery is active and the shutdown
                 * was not initiated by the application it will be recovered by the RabbitMQ client.
                 * <p>
                 * If the topology recovery is enabled we will also try to recover the consumer if (only) its channel
                 * fails. These events are not handled by the RabbitMQ client itself as they are most likely application
                 * specific. Also some edge cases like a delivery acknowledgement timeout may cause a channel shutdown.
                 * The registered exception handler of the consumer may handle these cases and it is possible to
                 * resume message handling by re-registering the consumer on a new channel.
                 */
                @Override
                public void handleShutdownSignal(String consumerTag, ShutdownSignalException sig) {
                    if (getChannel() instanceof RecoverableChannel && sig.isHardError()) {
                        // The RabbitMQ client automatic recovery is only triggered by connection errors.
                        // The consumer will be recovered by the client, so no additional handling here.
                        LOG.info("The underlying connection was terminated. Automatic recovery attempt is underway for consumer [{}]",
                                consumerTag);
                    } else if (channelPool.isTopologyRecoveryEnabled() && getChannel() instanceof RecoverableChannel) {
                        LOG.info("The channel of this consumer was terminated. Automatic recovery attempt is underway for consumer [{}]",
                                consumerTag, sig);
                        synchronized (RecoverableConsumerWrapper.this) {
                            RecoverableConsumerWrapper.this.consumer = null;
                            channelPool.returnChannel(getChannel());
                        }
                        performConsumerRecovery();
                    } else {
                        LOG.error("The channel was closed. Recovery is not enabled. The consumer [{}] will no longer receive messages",
                                consumerTag, sig);
                        cancel();
                    }
                }

                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
                        throws IOException {
                    // If the broker forces the channel to close, the client may already have prefetched messages in
                    // memory and will call handleDelivery for these messages, even if they are re-queued by the broker.
                    // The client will be unable to acknowledge these messages. So it is safe to silently discard
                    // them, without bothering the callback handler.
                    if (!getChannel().isOpen()) {
                        return;
                    }
                    handlingDeliveryCount.incrementAndGet();
                    if (executorService != null) {
                        executorService.submit(() -> callbackHandle(envelope, properties, body));
                    } else {
                        callbackHandle(envelope, properties, body);
                    }
                }

                private void callbackHandle(Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
                    try {
                        deliverCallback.handle(getChannel(), new Delivery(envelope, properties, body));
                    } finally {
                        handlingDeliveryCount.decrementAndGet();
                    }
                }
            };

            channel.basicConsume(queue, false, consumerTag, false, exclusive, arguments, consumer);
            return consumer;
        }
    }
}


<|start_filename|>rabbitmq/src/main/java/io/micronaut/rabbitmq/bind/AcknowledgementBinder.java<|end_filename|>
/*
 * Copyright 2017-2020 original 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 io.micronaut.rabbitmq.bind;

import io.micronaut.core.convert.ArgumentConversionContext;
import io.micronaut.core.type.Argument;
import io.micronaut.messaging.Acknowledgement;
import io.micronaut.messaging.exceptions.MessageAcknowledgementException;
import jakarta.inject.Singleton;

import java.util.Optional;

/**
 * Binds an argument of type {@link Acknowledgement} from the {@link RabbitConsumerState}.
 *
 * @param <T> Any type that extends {@link Acknowledgement}
 * @author <NAME>
 * @since 1.1.0
 */
@Singleton
public class AcknowledgementBinder<T extends Acknowledgement> implements RabbitTypeArgumentBinder<T> {

    @Override
    public Argument<T> argumentType() {
        return (Argument<T>) Argument.of(Acknowledgement.class);
    }

    @Override
    public BindingResult<T> bind(ArgumentConversionContext<T> context, RabbitConsumerState source) {
        Acknowledgement acknowledgement = new RabbitAcknowledgement() {
            @Override
            public void ack(boolean multiple) throws MessageAcknowledgementException {
                ackNack(true, multiple, false);
            }

            @Override
            public void nack(boolean multiple, boolean reQueue) throws MessageAcknowledgementException {
                ackNack(false, multiple, reQueue);
            }

            private void ackNack(boolean ack, boolean multiple, boolean requeue)  throws MessageAcknowledgementException {
                new RabbitMessageCloseable(source, multiple, requeue).withAcknowledge(ack).close();
            }
        };
        return () -> Optional.of((T) acknowledgement);
    }
}


<|start_filename|>docs-examples/example-groovy/src/test/groovy/io/micronaut/rabbitmq/docs/consumer/executor/CustomExecutorSpec.groovy<|end_filename|>
package io.micronaut.rabbitmq.docs.consumer.executor

import io.micronaut.rabbitmq.AbstractRabbitMQTest

class CustomExecutorSpec extends AbstractRabbitMQTest {

    void "test product client and listener"() {
        startContext()

        when:
// tag::producer[]
        def productClient = applicationContext.getBean(ProductClient)
        productClient.send("custom-executor-test".bytes)
// end::producer[]

        ProductListener productListener = applicationContext.getBean(ProductListener)

        then:
        waitFor {
            productListener.messageLengths.size() == 1
            productListener.messageLengths[0] == "custom-executor-test"
        }

        cleanup:
        // Finding that the context is closing the channel before ack is sent
        sleep 200
    }

    protected Map<String, Object> getConfiguration() {
        super.configuration + ["micronaut.executors.product-listener.type": "FIXED"]
    }
}


<|start_filename|>docs-examples/example-groovy/src/test/groovy/io/micronaut/rabbitmq/docs/headers/ProductListener.groovy<|end_filename|>
package io.micronaut.rabbitmq.docs.headers

import io.micronaut.context.annotation.Requires
// tag::imports[]
import io.micronaut.core.annotation.Nullable
import io.micronaut.messaging.annotation.MessageHeader
import io.micronaut.rabbitmq.annotation.Queue
import io.micronaut.rabbitmq.annotation.RabbitHeaders
import io.micronaut.rabbitmq.annotation.RabbitListener

import java.util.concurrent.CopyOnWriteArrayList
// end::imports[]

@Requires(property = "spec.name", value = "HeadersSpec")
// tag::clazz[]
@RabbitListener
class ProductListener {

    CopyOnWriteArrayList<String> messageProperties = []

    @Queue("product")
    void receive(byte[] data,
                 @MessageHeader("x-product-sealed") Boolean sealed, // <1>
                 @MessageHeader("x-product-count") Long count, // <2>
                 @Nullable @MessageHeader String productSize) { // <3>
        messageProperties << sealed.toString() + "|" + count + "|" + productSize
    }

    @Queue("product")
    void receive(byte[] data,
                 @RabbitHeaders Map<String, Object> headers) { // <4>
        messageProperties <<
                headers["x-product-sealed"].toString() + "|" +
                headers["x-product-count"] + "|" +
                headers["productSize"]?.toString()
    }
}
// end::clazz[]


<|start_filename|>docs-examples/example-groovy/src/test/groovy/io/micronaut/rabbitmq/docs/consumer/custom/annotation/ProductListener.groovy<|end_filename|>
package io.micronaut.rabbitmq.docs.consumer.custom.annotation

import io.micronaut.context.annotation.Requires
// tag::imports[]
import io.micronaut.rabbitmq.annotation.Queue
import io.micronaut.rabbitmq.annotation.RabbitListener

import java.util.concurrent.CopyOnWriteArraySet
// end::imports[]

@Requires(property = "spec.name", value = "DeliveryTagSpec")
// tag::clazz[]
@RabbitListener
class ProductListener {

    CopyOnWriteArraySet<Long> messages = []

    @Queue("product")
    void receive(byte[] data, @DeliveryTag Long tag) { // <1>
        messages << tag
    }
}
// end::clazz[]


<|start_filename|>docs-examples/example-groovy/src/test/groovy/io/micronaut/rabbitmq/docs/consumer/connection/ConnectionSpec.groovy<|end_filename|>
package io.micronaut.rabbitmq.docs.consumer.connection

import io.micronaut.rabbitmq.AbstractRabbitMQTest

class ConnectionSpec extends AbstractRabbitMQTest {

    void "test product client and listener"() {
        startContext()

        when:
// tag::producer[]
        def productClient = applicationContext.getBean(ProductClient)
        productClient.send("connection-test".bytes)
// end::producer[]

        ProductListener productListener = applicationContext.getBean(ProductListener)

        then:
        waitFor {
            productListener.messageLengths.size() == 1
            productListener.messageLengths[0] == "connection-test"
        }

        cleanup:
        // Finding that the context is closing the channel before ack is sent
        sleep 200
    }

    protected Map<String, Object> getConfiguration() {
        super.configuration + ["rabbitmq.servers.product-cluster.port": super.configuration.remove("rabbitmq.port")]
    }
}


<|start_filename|>rabbitmq/src/main/java/io/micronaut/rabbitmq/bind/RabbitMessageCloseable.java<|end_filename|>
/*
 * Copyright 2017-2020 original 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 io.micronaut.rabbitmq.bind;

import com.rabbitmq.client.Channel;
import io.micronaut.core.annotation.Nullable;
import io.micronaut.messaging.exceptions.MessageAcknowledgementException;

import java.io.IOException;

/**
 * Responsible for acknowledging or rejecting a message. Only
 * applies if the consuming method does not have an {@link io.micronaut.messaging.Acknowledgement}
 * argument.
 *
 * @author <NAME>
 * @since 1.1.0
 */
public class RabbitMessageCloseable implements AutoCloseable {

    private final Channel channel;
    private final long deliveryTag;
    private final boolean multiple;
    private final boolean reQueue;
    private Boolean acknowledge = null;

    /**
     * Default constructor.
     *
     * @param consumerState The message state
     * @param multiple Whether multiple messages should be responded to
     * @param reQueue Whether rejections should cause the messages to be re-queued
     */
    public RabbitMessageCloseable(RabbitConsumerState consumerState,
                           boolean multiple,
                           boolean reQueue) {
        this.channel = consumerState.getChannel();
        this.deliveryTag = consumerState.getEnvelope().getDeliveryTag();
        this.multiple = multiple;
        this.reQueue = reQueue;
    }

    @Override
    public void close() throws MessageAcknowledgementException {
        if (acknowledge != null) {
            if (acknowledge) {
                try {
                    channel.basicAck(deliveryTag, multiple);
                } catch (IOException e) {
                    throw new MessageAcknowledgementException("An error occurred acknowledging a message", e);
                }
            } else {
                try {
                    channel.basicNack(deliveryTag, multiple, reQueue);
                } catch (IOException e) {
                    throw new MessageAcknowledgementException("An error occurred rejecting a message", e);
                }
            }
        }
    }

    /**
     * Builder style sets whether the message should be acknowledged.
     *
     * Set to true if the message should be acknowledged.
     * Set to false if the message should be rejected.
     * Set to null if the message should not be acknowledged or rejected.
     *
     * @param acknowledge The acknowledge parameter.
     * @return The same instance
     */
    public RabbitMessageCloseable withAcknowledge(@Nullable Boolean acknowledge) {
        this.acknowledge = acknowledge;
        return this;
    }
}


<|start_filename|>rabbitmq/src/test/groovy/io/micronaut/rabbitmq/health/RabbitHealthIndicatorSpec.groovy<|end_filename|>
package io.micronaut.rabbitmq.health

import io.micronaut.context.ApplicationContext
import io.micronaut.management.health.indicator.HealthResult
import io.micronaut.rabbitmq.AbstractRabbitMQTest
import reactor.core.publisher.Mono

import static io.micronaut.health.HealthStatus.DOWN
import static io.micronaut.health.HealthStatus.UP

class RabbitHealthIndicatorSpec extends AbstractRabbitMQTest {

    void "test rabbitmq health indicator"() {
        given:
        startContext()

        when:
        RabbitMQHealthIndicator healthIndicator = applicationContext.getBean(RabbitMQHealthIndicator)
        HealthResult result = Mono.from(healthIndicator.result).block()

        then:
        result.status == UP
        ((Map<String, Object>) result.details).version.toString().startsWith("3.7")
    }

    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()

        then:
        result.status == UP
        Map<String, List> details = result.details
        details.get("connections")[0].get("version").toString().startsWith("3.7")
        details.get("connections")[1].get("version").toString().startsWith("3.7")
    }

    void "test rabbitmq health indicator shows down"() {
        given:
        startContext()

        when:
        RabbitMQHealthIndicator healthIndicator = applicationContext.getBean(RabbitMQHealthIndicator)
        rabbitContainer.stop()
        HealthResult result = Mono.from(healthIndicator.result).block()

        then:
        result.status == DOWN
        ((Map) result.details).get("error").toString().contains("RabbitMQ connection is not open")

        cleanup:
        rabbitContainer.start()
    }
}


<|start_filename|>docs-examples/example-kotlin/src/test/kotlin/io/micronaut/rabbitmq/docs/properties/ProductClient.kt<|end_filename|>
package io.micronaut.rabbitmq.docs.properties

import io.micronaut.context.annotation.Requires
// tag::imports[]
import io.micronaut.rabbitmq.annotation.Binding
import io.micronaut.rabbitmq.annotation.RabbitClient
import io.micronaut.rabbitmq.annotation.RabbitProperties
import io.micronaut.rabbitmq.annotation.RabbitProperty
// end::imports[]

@Requires(property = "spec.name", value = "PropertiesSpec")
// tag::clazz[]
@RabbitClient
@RabbitProperty(name = "appId", value = "myApp") // <1>
interface ProductClient {

    @Binding("product")
    @RabbitProperties( // <2>
            RabbitProperty(name = "contentType", value = "application/json"),
            RabbitProperty(name = "userId", value = "guest")
    )
    fun send(data: ByteArray)

    @Binding("product")
    fun send(@RabbitProperty("userId") user: String, @RabbitProperty contentType: String?, data: ByteArray)  // <3>
}
// end::clazz[]


<|start_filename|>docs-examples/example-java/src/test/java/io/micronaut/rabbitmq/AbstractRabbitMQTest.java<|end_filename|>
package io.micronaut.rabbitmq;

import io.micronaut.context.ApplicationContext;
import org.junit.jupiter.api.AfterEach;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.wait.strategy.LogMessageWaitStrategy;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Callable;

import static java.util.concurrent.TimeUnit.SECONDS;
import static org.awaitility.Awaitility.await;

public abstract class AbstractRabbitMQTest {

    protected static GenericContainer<?> rabbitContainer = new GenericContainer<>("library/rabbitmq:3.7")
            .withExposedPorts(5672)
            .waitingFor(new LogMessageWaitStrategy().withRegEx("(?s).*Server startup complete.*"));

    static {
        rabbitContainer.start();
    }

    protected ApplicationContext applicationContext;

    protected void startContext() {
        applicationContext = ApplicationContext.run(getConfiguration(), "test");
    }

    protected Map<String, Object> getConfiguration() {
        Map<String, Object> config = new HashMap<>();
        config.put("rabbitmq.port", rabbitContainer.getMappedPort(5672));
        config.put("spec.name", getClass().getSimpleName());
        return config;
    }

    protected void waitFor(Callable<Boolean> conditionEvaluator) {
        await().atMost(5, SECONDS).until(conditionEvaluator);
    }

    @AfterEach
    void cleanup() {
        if (applicationContext != null) {
            applicationContext.close();
        }
    }
}


<|start_filename|>docs-examples/example-groovy/src/test/groovy/io/micronaut/rabbitmq/docs/properties/PropertiesSpec.groovy<|end_filename|>
package io.micronaut.rabbitmq.docs.properties

import io.micronaut.rabbitmq.AbstractRabbitMQTest

class PropertiesSpec extends AbstractRabbitMQTest {

    void "test sending and receiving properties"() {
        startContext()

        when:
// tag::producer[]
        ProductClient productClient = applicationContext.getBean(ProductClient)
        productClient.send("body".bytes)
        productClient.send("guest", "text/html", "body2".bytes)
        productClient.send("guest", null, "body3".bytes)
// end::producer[]

        ProductListener productListener = applicationContext.getBean(ProductListener)

        then:
        waitFor {
            productListener.messageProperties.size() == 3
            productListener.messageProperties.contains("guest|application/json|myApp")
            productListener.messageProperties.contains("guest|text/html|myApp")
            productListener.messageProperties.contains("guest|null|myApp")
        }
    }
}


<|start_filename|>docs-examples/example-kotlin/src/test/kotlin/io/micronaut/rabbitmq/docs/parameters/BindingSpec.kt<|end_filename|>
package io.micronaut.rabbitmq.docs.parameters

import io.kotest.assertions.timing.eventually
import io.kotest.matchers.collections.shouldContain
import io.kotest.matchers.shouldBe
import io.micronaut.rabbitmq.AbstractRabbitMQTest
import kotlin.time.Duration
import kotlin.time.ExperimentalTime

@OptIn(ExperimentalTime::class)
class BindingSpec: AbstractRabbitMQTest({

    val specName = javaClass.simpleName

    given("A basic producer and consumer") {
        val ctx = startContext(specName)

        `when`("The messages are published") {
            val productListener = ctx.getBean(ProductListener::class.java)

            // tag::producer[]
            val productClient = ctx.getBean(ProductClient::class.java)
            productClient.send("message body".toByteArray())
            productClient.send("product", "message body2".toByteArray())
            // end::producer[]

            then("The messages are received") {
                eventually(Duration.seconds(10)) {
                    productListener.messageLengths.size shouldBe 2
                    productListener.messageLengths shouldContain 12
                    productListener.messageLengths shouldContain 13
                }
            }
        }

        ctx.stop()
    }
})


<|start_filename|>rabbitmq/src/test/groovy/io/micronaut/rabbitmq/annotation/HeaderExchangeSpec.groovy<|end_filename|>
package io.micronaut.rabbitmq.annotation

import io.micronaut.context.annotation.Requires
import io.micronaut.messaging.annotation.MessageHeader
import io.micronaut.rabbitmq.AbstractRabbitMQTest

class HeaderExchangeSpec extends AbstractRabbitMQTest {

    void "test publishing and consuming on a topic exchange"() {

        startContext()

        AnimalProducer producer = applicationContext.getBean(AnimalProducer)
        AnimalListener consumer = applicationContext.getBean(AnimalListener)

        when:
        producer.go("Cat", new Cat(lives: 9, name: "Whiskers"))
        producer.go("Cat", new Cat(lives: 8, name: "Mr. Bigglesworth"))
        producer.go("Dog", new Dog(size: "M", name: "Chloe"))
        producer.go("Dog", new Dog(size: "L", name: "Butch"))

        then:
        waitFor {
            consumer.messages.size() == 4
            consumer.messages.find({ it.name == "Whiskers" }).lives == 9
            consumer.messages.find({ it.name == "Chloe" }).size == "M"
            consumer.messages.find({ it.name == "Mr. Bigglesworth" }).lives == 8
            consumer.messages.find({ it.name == "Butch" }).size == "L"
        }
    }

    static class Cat extends Animal {
        int lives
    }
    static class Dog extends Animal {
        String size
    }
    static abstract class Animal {
        String name
    }

    @Requires(property = "spec.name", value = "HeaderExchangeSpec")
    @RabbitClient("animals")
    static interface AnimalProducer {
        void go(@MessageHeader String animalType, Animal animal)
    }

    @Requires(property = "spec.name", value = "HeaderExchangeSpec")
    @RabbitListener
    static class AnimalListener {

        static List<Animal> messages = []

        @Queue("dogs")
        void listen(Dog dog) {
            messages << dog
        }

        @Queue("cats")
        void listen(Cat cat) {
            messages << cat
        }
    }
}


<|start_filename|>rabbitmq/src/test/groovy/io/micronaut/rabbitmq/listener/ConsumerRecoverySpec.groovy<|end_filename|>
package io.micronaut.rabbitmq.listener

import com.github.dockerjava.api.DockerClient
import com.rabbitmq.client.AMQP
import com.rabbitmq.client.AlreadyClosedException
import com.rabbitmq.client.ConnectionFactory
import com.rabbitmq.client.Return
import com.rabbitmq.client.ReturnCallback
import io.micronaut.context.annotation.Requires
import io.micronaut.messaging.annotation.MessageBody
import io.micronaut.rabbitmq.AbstractRabbitMQClusterTest
import io.micronaut.rabbitmq.annotation.Queue
import io.micronaut.rabbitmq.annotation.RabbitListener
import io.micronaut.rabbitmq.bind.RabbitConsumerState
import io.micronaut.rabbitmq.exception.RabbitListenerException
import io.micronaut.rabbitmq.exception.RabbitListenerExceptionHandler
import org.slf4j.Logger
import org.slf4j.LoggerFactory
import org.testcontainers.DockerClientFactory
import org.testcontainers.containers.GenericContainer
import spock.lang.Shared
import spock.util.concurrent.PollingConditions

import java.util.concurrent.Executors
import java.util.concurrent.ScheduledExecutorService
import java.util.concurrent.TimeoutException

import static java.util.concurrent.TimeUnit.MILLISECONDS
import static org.hamcrest.MatcherAssert.assertThat
import static org.hamcrest.Matchers.equalTo

class ConsumerRecoverySpec extends AbstractRabbitMQClusterTest {

    private static final Logger log = LoggerFactory.getLogger(ConsumerRecoverySpec)

    private static final DockerClient DOCKER_CLIENT = DockerClientFactory.lazyClient()

    @Shared
    private ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor()

    @Shared
    private Set<String> publishedMessages = new LinkedHashSet<>()

    @Shared
    private boolean enablePublisher = false

    void setupSpec() {
        /*
         * The current Micronaut publisher implementation has a flaw in detecting unroutable drop/return messages
         * in a Rabbit cluster setup. It considers the messages as published even if the broker did not enqueue it.
         * So for this test a simple custom publisher is used that detects unpublished messages.
         */
        ConnectionFactory connectionFactory = new ConnectionFactory(port: node3Port)
        connectionFactory.newConnection().openChannel().map(ch -> {
                try {
                    ch.confirmSelect()
                    // returned messages must not count as published
                    ch.addReturnListener(new ReturnCallback() {
                        @Override
                        void handle(Return r) {
                            String returned = new String(r.body)
                            publishedMessages.remove(returned)
                            log.warn("{} publish message returned: {}", publishedMessages.size(), returned)
                        }
                    })
                } catch (IOException e) {
                    log.error("failed to set confirmSelect", e)
                }
                return ch
            })
            .ifPresent(ch -> executorService.scheduleWithFixedDelay(() -> {
                if (!enablePublisher) {
                    return
                }
                String msg = UUID.randomUUID()
                try {
                    publishedMessages << msg
                    ch.basicPublish(EXCHANGE, "", true,
                            new AMQP.BasicProperties.Builder().deliveryMode(2).build(),
                            msg.bytes)
                    if (ch.waitForConfirms(1000)) {
                        log.info("publish ack")
                    }
                } catch (IOException | RuntimeException | InterruptedException | TimeoutException e) {
                    publishedMessages.remove(msg)
                    log.error("failed to publish: {}", e.message)
                }
            }, 500, 500, MILLISECONDS))
    }

    void cleanupSpec() {
        executorService.shutdown()
    }

    void setup() {
        publishedMessages.clear()
        enablePublisher = true
    }

    void "test restart of Node1 with consumer connected to Node1"() {
        given:
        startContext(connectToNode: "node1")
        TestConsumer consumer = applicationContext.getBean(TestConsumer)
        awaitPublishConsumeOfMessages(consumer)

        when:
        restartContainer(NODE1_CONT)

        then:
        awaitPublishConsumeOfMessages(consumer)
        stopPublishAndAssertAllConsumed(consumer)
    }

    void "test restart of Node1 with consumer connected to Node2"() {
        given:
        startContext(connectToNode: "node2")
        TestConsumer consumer = applicationContext.getBean(TestConsumer)
        awaitPublishConsumeOfMessages(consumer)

        when:
        restartContainer(NODE1_CONT)

        then:
        awaitPublishConsumeOfMessages(consumer)
        stopPublishAndAssertAllConsumed(consumer)
    }

    void "test restart of Node2 with consumer connected to Node1"() {
        given:
        startContext(connectToNode: "node1")
        TestConsumer consumer = applicationContext.getBean(TestConsumer)
        awaitPublishConsumeOfMessages(consumer)

        when:
        restartContainer(NODE2_CONT)

        then:
        awaitPublishConsumeOfMessages(consumer)
        stopPublishAndAssertAllConsumed(consumer)
    }

    void "test restart of Node2 with consumer connected to Node2"() {
        given:
        startContext(connectToNode: "node2")
        TestConsumer consumer = applicationContext.getBean(TestConsumer)
        awaitPublishConsumeOfMessages(consumer)

        when:
        restartContainer(NODE2_CONT)

        then:
        awaitPublishConsumeOfMessages(consumer)
        stopPublishAndAssertAllConsumed(consumer)
    }

    void "test consumer recovery after delivery acknowledgement timeout"() {
        given:
        startContext(connectToNode: "node3")
        SlowTestConsumer slowConsumer = applicationContext.getBean(SlowTestConsumer)

        when:
        new PollingConditions(timeout: 120).eventually {
            RabbitListenerException e = slowConsumer.lastException
            e && e.cause instanceof AlreadyClosedException
        }
        slowConsumer.doSlowdown = false

        then:
        stopPublishAndAssertAllConsumed(slowConsumer)
    }

    @Requires(property = "spec.name", value = "ConsumerRecoverySpec")
    @Requires(property = "connectToNode", value = "node1")
    @RabbitListener(connection = "node1")
    static class Node1Consumer extends TestConsumer {}

    @Requires(property = "spec.name", value = "ConsumerRecoverySpec")
    @Requires(property = "connectToNode", value = "node2")
    @RabbitListener(connection = "node2")
    static class Node2Consumer extends TestConsumer {}

    @Requires(property = "spec.name", value = "ConsumerRecoverySpec")
    @Requires(property = "connectToNode", value = "node3")
    @RabbitListener(connection = "node3")
    static class SlowTestConsumer extends TestConsumer {

        boolean doSlowdown = true

        @Override
        @Queue(value = QUEUE, prefetch = 5)
        void handleMessage(@MessageBody String body) {
            super.handleMessage(body)

            if (doSlowdown && consumedMessages.size() % 10 == 0) {
                sleep 20_000 // simulate slow processing
                log.info("slow message processing complete")
            }
        }
    }

    private void awaitPublishConsumeOfMessages(TestConsumer consumer) {
        int targetPubCount = publishedMessages.size() + 10
        int targetConCount = consumer.consumedMessages.size() + 10

        new PollingConditions(timeout: 60).eventually {
            publishedMessages.size() > targetPubCount
            consumer.consumedMessages.size() > targetConCount
        }
    }

    private void stopPublishAndAssertAllConsumed(TestConsumer consumer) {
        enablePublisher = false

        new PollingConditions(timeout: 60).eventually {
            assertThat "all published messages must be consumed",
                    publishedMessages, equalTo(consumer.consumedMessages)
        }
    }

    private static restartContainer(GenericContainer container) throws InterruptedException {
        log.info("stopping container: {}", container.containerId)
        DOCKER_CLIENT.stopContainerCmd(container.containerId).exec()

        log.info("re-starting container: {}", container.containerId)
        DOCKER_CLIENT.startContainerCmd(container.containerId).exec()

        new PollingConditions(timeout: 60).eventually {
            container.isHealthy()
        }
        log.info("started container: {}", container.containerId)
    }
}

abstract class TestConsumer implements RabbitListenerExceptionHandler {

    static final Logger log = LoggerFactory.getLogger(TestConsumer)

    final Set<String> consumedMessages = new LinkedHashSet<>()

    RabbitListenerException lastException

    @Queue(AbstractRabbitMQClusterTest.QUEUE)
    void handleMessage(@MessageBody String body) {
        consumedMessages << body
        log.info("{} received: {}", consumedMessages.size(), body)
    }

    @Override
    void handle(RabbitListenerException e) {
        lastException = e
        String msg = e.messageState
                .map(RabbitConsumerState::getBody)
                .map(String::new)
                .orElse("<<no message>>")
        consumedMessages.remove(msg)
        log.warn("{} failed to consume: {}", consumedMessages.size(), msg, e)
    }
}


<|start_filename|>docs-examples/example-java/src/test/java/io/micronaut/rabbitmq/docs/consumer/executor/CustomExecutorSpec.java<|end_filename|>
package io.micronaut.rabbitmq.docs.consumer.executor;

import io.micronaut.rabbitmq.AbstractRabbitMQTest;
import org.junit.jupiter.api.Test;

import java.util.Map;

public class CustomExecutorSpec extends AbstractRabbitMQTest {

    @Test
    void testProductClientAndListener() {
        startContext();

// tag::producer[]
ProductClient productClient = applicationContext.getBean(ProductClient.class);
productClient.send("custom-executor-test".getBytes());
// end::producer[]

        ProductListener productListener = applicationContext.getBean(ProductListener.class);

        waitFor(() ->
                productListener.messageLengths.size() == 1 &&
                productListener.messageLengths.get(0).equals("custom-executor-test")
        );
    }

    protected Map<String, Object> getConfiguration() {
        Map<String, Object> config = super.getConfiguration();
        config.put("micronaut.executors.product-listener.type", "FIXED");
        return config;
    }
}


<|start_filename|>rabbitmq/src/test/groovy/io/micronaut/rabbitmq/annotation/SimpleBindingSpec.groovy<|end_filename|>
package io.micronaut.rabbitmq.annotation

import io.micronaut.context.annotation.Requires
import io.micronaut.core.annotation.Nullable
import io.micronaut.rabbitmq.AbstractRabbitMQTest

class SimpleBindingSpec extends AbstractRabbitMQTest {

    void "test simple producing and consuming with a boolean"() {
        startContext()
        MyProducer producer = applicationContext.getBean(MyProducer)
        MyConsumer consumer = applicationContext.getBean(MyConsumer)

        when:
        producer.go(true)
        producer.go(false)
        producer.go(null)

        then:
        waitFor {
            consumer.messages.size() == 3
            consumer.messages[0] == true
            consumer.messages[1] == false
            consumer.messages[2] == null
        }
    }

    void "test simple producing and consuming with a list of pojos"() {
        startContext()

        MyListProducer producer = applicationContext.getBean(MyListProducer)
        MyListConsumer consumer = applicationContext.getBean(MyListConsumer)

        when:
        producer.go([true, false])
        producer.go([null, true])

        then:
        waitFor {
            consumer.messages.size() == 4
            consumer.messages[0] == true
            consumer.messages[1] == false
            consumer.messages[2] == null
            consumer.messages[3] == true
        }
    }

    static class Person {
        String name
    }

    @Requires(property = "spec.name", value = "SimpleBindingSpec")
    @RabbitClient
    static interface MyProducer {
        @Binding("simple")
        void go(Boolean data)
    }

    @Requires(property = "spec.name", value = "SimpleBindingSpec")
    @RabbitListener
    static class MyConsumer {

        static List<Boolean> messages = []

        @Queue("simple")
        void listen(@Nullable Boolean data) {
            messages << data
        }
    }

    @Requires(property = "spec.name", value = "SimpleBindingSpec")
    @RabbitClient
    static interface MyListProducer {
        @Binding("simple-list")
        void go(List<Boolean> data)
    }

    @Requires(property = "spec.name", value = "SimpleBindingSpec")
    @RabbitListener
    static class MyListConsumer {

        static List<Boolean> messages = []

        @Queue("simple-list")
        void listen(List<Boolean> data) {
            messages.addAll(data)
        }
    }
}


<|start_filename|>docs-examples/example-kotlin/src/test/kotlin/io/micronaut/rabbitmq/docs/consumer/custom/type/ProductClient.kt<|end_filename|>
package io.micronaut.rabbitmq.docs.consumer.custom.type

import io.micronaut.context.annotation.Requires
// tag::imports[]
import io.micronaut.messaging.annotation.MessageHeader
import io.micronaut.messaging.annotation.MessageHeaders
import io.micronaut.rabbitmq.annotation.Binding
import io.micronaut.rabbitmq.annotation.RabbitClient
// end::imports[]

@Requires(property = "spec.name", value = "ProductInfoSpec")
// tag::clazz[]
@RabbitClient
@MessageHeaders( // <1>
        MessageHeader(name = "x-product-sealed", value = "true"),
        MessageHeader(name = "productSize", value = "large")
)
interface ProductClient {

    @Binding("product")
    @MessageHeaders( // <2>
        MessageHeader(name = "x-product-count", value = "10"),
        MessageHeader(name = "productSize", value = "small")
    )
    fun send(data: ByteArray)

    @Binding("product")
    fun send(@MessageHeader productSize: String?, // <3>
             @MessageHeader("x-product-count") count: Long,
             data: ByteArray)
}
// end::clazz[]


<|start_filename|>docs-examples/example-kotlin/src/test/kotlin/io/micronaut/rabbitmq/docs/consumer/concurrent/ConcurrentSpec.kt<|end_filename|>
package io.micronaut.rabbitmq.docs.consumer.concurrent

import io.kotest.framework.concurrency.eventually
import io.kotest.matchers.shouldBe
import io.micronaut.rabbitmq.AbstractRabbitMQTest
import kotlin.time.Duration
import kotlin.time.ExperimentalTime

@OptIn(ExperimentalTime::class)
class ConcurrentSpec : AbstractRabbitMQTest({

    val specName = javaClass.simpleName

    given("A basic producer and consumer") {
        val ctx = startContext(specName)

        `when`("The messages are published") {
            val productListener = ctx.getBean(ProductListener::class.java)
            val productClient = ctx.getBean(ProductClient::class.java)
            for (i in 0..3) {
                productClient.send("body".toByteArray())
            }

            then("The messages are received") {
                eventually(Duration.seconds(5)) {
                    productListener.threads.size shouldBe 4
                }
            }
        }

        ctx.close()
    }
})


<|start_filename|>docs-examples/example-groovy/src/test/groovy/io/micronaut/rabbitmq/docs/consumer/concurrent/ConcurrentSpec.groovy<|end_filename|>
package io.micronaut.rabbitmq.docs.consumer.concurrent

import io.micronaut.rabbitmq.AbstractRabbitMQTest

class ConcurrentSpec extends AbstractRabbitMQTest {

    void "test concurrent consumers"() {
        startContext()

        ProductClient productClient = applicationContext.getBean(ProductClient)
        4.times { productClient.send("body".bytes) }

        ProductListener productListener = applicationContext.getBean(ProductListener)

        waitFor {
            productListener.threads.size() == 4
        }
    }
}


<|start_filename|>docs-examples/example-java/src/test/java/io/micronaut/rabbitmq/docs/consumer/types/ProductListener.java<|end_filename|>
package io.micronaut.rabbitmq.docs.consumer.types;

import io.micronaut.context.annotation.Requires;
// tag::imports[]
import com.rabbitmq.client.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Envelope;
import io.micronaut.rabbitmq.annotation.Queue;
import io.micronaut.rabbitmq.annotation.RabbitListener;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
// end::imports[]

@Requires(property = "spec.name", value = "TypeBindingSpec")
// tag::clazz[]
@RabbitListener
public class ProductListener {

    List<String> messages = Collections.synchronizedList(new ArrayList<>());

    @Queue("product")
    public void receive(byte[] data,
                        Envelope envelope, // <1>
                        BasicProperties basicProperties, // <2>
                        Channel channel) { // <3>
        messages.add(String.format("exchange: [%s], routingKey: [%s], contentType: [%s]",
                envelope.getExchange(), envelope.getRoutingKey(), basicProperties.getContentType()));
    }
}
// end::clazz[]


<|start_filename|>docs-examples/example-java/src/test/java/io/micronaut/rabbitmq/docs/exchange/Snake.java<|end_filename|>
package io.micronaut.rabbitmq.docs.exchange;

public class Snake extends Animal {

    private boolean venomous;

    public Snake(String name, boolean venomous) {
        super(name);
        this.venomous = venomous;
    }

    public Snake() { }

    public boolean isVenomous() {
        return venomous;
    }

    public void setVenomous(boolean venomous) {
        this.venomous = venomous;
    }
}


<|start_filename|>rabbitmq/src/main/java/io/micronaut/rabbitmq/intercept/DefaultConsumer.java<|end_filename|>
/*
 * Copyright 2017-2020 original 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 io.micronaut.rabbitmq.intercept;

import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.ShutdownSignalException;

import java.io.IOException;

/**
 * A default {@link Consumer} that delegates cancelled or shutdown
 * with a single method {@link #handleTerminate(String)}.
 *
 * @author <NAME>
 * @since 1.1.0
 */
public interface DefaultConsumer extends Consumer {

    @Override
    default void handleConsumeOk(String consumerTag) { }

    @Override
    default void handleCancelOk(String consumerTag) {
        handleTerminate(consumerTag);
    }

    @Override
    default void handleCancel(String consumerTag) throws IOException {
        handleTerminate(consumerTag);
    }

    @Override
    default void handleShutdownSignal(String consumerTag, ShutdownSignalException sig) {
        handleTerminate(consumerTag);
    }

    @Override
    default void handleRecoverOk(String consumerTag) { }

    /**
     * Called when a consumer is cancelled or shut down.
     *
     * @param consumerTag The consumer tag
     */
    default void handleTerminate(String consumerTag) { }
}


<|start_filename|>docs-examples/example-groovy/src/test/groovy/io/micronaut/rabbitmq/docs/consumer/executor/ProductListener.groovy<|end_filename|>
package io.micronaut.rabbitmq.docs.consumer.executor

import io.micronaut.context.annotation.Requires
// tag::imports[]
import io.micronaut.rabbitmq.annotation.Queue
import io.micronaut.rabbitmq.annotation.RabbitListener

import java.util.concurrent.CopyOnWriteArrayList
// end::imports[]

@Requires(property = "spec.name", value = "CustomExecutorSpec")
// tag::clazz[]
@RabbitListener
class ProductListener {

    CopyOnWriteArrayList<String> messageLengths = []

    @Queue(value = "product", executor = "product-listener") // <1>
    void receive(byte[] data) {
        messageLengths << new String(data)
    }
}
// end::clazz[]


<|start_filename|>docs-examples/example-groovy/src/test/groovy/io/micronaut/rabbitmq/docs/consumer/custom/type/ProductInfoSpec.groovy<|end_filename|>
package io.micronaut.rabbitmq.docs.consumer.custom.type

import io.micronaut.rabbitmq.AbstractRabbitMQTest

class ProductInfoSpec extends AbstractRabbitMQTest {

    void "test using a custom type binder"() {
        startContext()

        when:
// tag::producer[]
        ProductClient productClient = applicationContext.getBean(ProductClient)
        productClient.send("body".bytes)
        productClient.send("medium", 20L, "body2".bytes)
        productClient.send(null, 30L, "body3".bytes)
// end::producer[]

        ProductListener productListener = applicationContext.getBean(ProductListener)

        then:
        waitFor {
            productListener.messages.size() == 3

            productListener.messages.find({ pi ->
                pi.size == "small" && pi.count == 10 && pi.sealed
            }) != null

            productListener.messages.find({ pi ->
                pi.size == "medium" && pi.count == 20 && pi.sealed
            }) != null

            productListener.messages.find({ pi ->
                pi.size == null && pi.count == 30 && pi.sealed
            }) != null
        }
    }
}


<|start_filename|>rabbitmq/src/test/groovy/io/micronaut/rabbitmq/annotation/QueueAnnotationSpec.groovy<|end_filename|>
package io.micronaut.rabbitmq.annotation

import io.micronaut.context.annotation.Requires
import io.micronaut.messaging.annotation.MessageMapping
import io.micronaut.rabbitmq.AbstractRabbitMQTest

class QueueAnnotationSpec extends AbstractRabbitMQTest {

    void 'test that @Queue value aliases to @MessageMapping'() {
        given:
        startContext()

        def definition = applicationContext.getBeanDefinition(MyConsumer)

        when:
        def method = definition.getRequiredMethod('receive', String)
        def annotationValue = method.getValue(MessageMapping, String[])

        then:
        annotationValue.isPresent()
        annotationValue.get().contains 'simple'
    }

    @Requires(property = 'spec.name', value = 'QueueAnnotationSpec')
    @RabbitListener
    static class MyConsumer {

        List<String> stuff = []

        @Queue('simple')
        void receive(String thing) {
            stuff << thing
        }
    }
}


<|start_filename|>docs-examples/example-groovy/src/test/groovy/io/micronaut/rabbitmq/docs/consumer/concurrent/ProductClient.groovy<|end_filename|>
package io.micronaut.rabbitmq.docs.consumer.concurrent

import io.micronaut.context.annotation.Requires
import io.micronaut.rabbitmq.annotation.Binding
import io.micronaut.rabbitmq.annotation.RabbitClient

@Requires(property = "spec.name", value = "ConcurrentSpec")
@RabbitClient
interface ProductClient {

    @Binding("product")
    void send(byte[] data)
}
