byte[] received when using @RabbitListener with Batching

[arths31]

Hello,
I'm having an issue when using @RabbitListener with batching enabled.

Batching works fine but I want to receive a list of org.springframework.amqp.core.Message and I am receiving a list of byte[] instead.

Here is the bean I use to configure the containerFactory

@Bean
    fun myFactory(configurer: SimpleRabbitListenerContainerFactoryConfigurer): SimpleRabbitListenerContainerFactory {
        val factory = SimpleRabbitListenerContainerFactory()
        factory.setConnectionFactory(connectionFactory)
        factory.setBatchListener(true)
        factory.setBatchSize(10)
        factory.setConsumerBatchEnabled(true)
        factory.setDeBatchingEnabled(false)
        factory.setReceiveTimeout(5000)
        return factory
    }

and here is where I expect to receive my messages

@RabbitListener(bindings = [
        QueueBinding(
            value = org.springframework.amqp.rabbit.annotation.Queue(name = "test", durable = "true"),
            key = ["xxx.#"],
            exchange = Exchange(name = Constants.EVENT_EXCHANGE_NAME, type = ExchangeTypes.TOPIC)
        )
    ], containerFactory = "myFactory")
    fun listen(messages: List<Message>) {
       ...
     }

messages returns a list of byte[] which contains my payloads.

I have tried switching to List<org.springframework.messaging.Message> and even a list of String just like in the following exemple but to no avail : https://docs.spring.io/spring-amqp/reference/amqp/receiving-messages/batch.html

Thank you in advance for your help

[artembilan]

The SimpleMessageConverter is used by default in the AbstractAdaptableMessageListener and logic of that is like this:

	@Override
	public Object fromMessage(Message message) throws MessageConversionException {
		Object content = null;
		MessageProperties properties = message.getMessageProperties();
		if (properties != null) {
			String contentType = properties.getContentType();
			if (contentType != null && contentType.startsWith("text")) {
				String encoding = properties.getContentEncoding();
				if (encoding == null) {
					encoding = this.defaultCharset;
				}
				try {
					content = new String(message.getBody(), encoding);
				}
				catch (UnsupportedEncodingException e) {
					throw new MessageConversionException("failed to convert text-based Message content", e);
				}
			}
			else if (contentType != null &&
					contentType.equals(MessageProperties.CONTENT_TYPE_SERIALIZED_OBJECT)) {
				try {
					content = SerializationUtils.deserialize(
							createObjectInputStream(new ByteArrayInputStream(message.getBody())));
				}
				catch (IOException | IllegalArgumentException | IllegalStateException e) {
					throw new MessageConversionException("failed to convert serialized Message content", e);
				}
			}
		}
		if (content == null) {
			content = message.getBody();
		}
		return content;
	}

So, if contentType is not text/plain and not application/x-java-serialized-object the message.getBody() is returned, which is, essentially, byte[].

We do have logic in the MessagingMessageListenerAdapter like this:

					this.isMessageList = paramType.equals(Message.class) || messageHasGeneric;
					this.isAmqpMessageList = paramType.equals(org.springframework.amqp.core.Message.class);

So, List<(org.springframework.amqp.core.Message> should work.

Perhaps you can share with us a simple project where we can reproduce the problem?
Might be some special can for Kotlin...

[arths31]

Hello,
Thank you for your fast reply !
I've enabled breakpoints in MessagingMessageListenerAdapter.java on the following lines :

boolean messageHasGeneric = paramType instanceof ParameterizedType pType
							&& pType.getRawType().equals(Message.class);
this.isMessageList = paramType.equals(Message.class) || messageHasGeneric;
this.isAmqpMessageList = paramType.equals(org.springframework.amqp.core.Message.class);

It seems paramType is wrong with Kotlin, it is seen as WildcardTypeImpl<? extends org.springframework.amqp.core.Message> instead of org.springframework.amqp.core.Message thus not setting isAmqpMessageList.

[arths31]

I've created a Java listener in my Kotlin project to make a quick test and I can confirm that it works using Java.

public class SyncRabbitMQListenerJava {

    @RabbitListener(bindings =
    @QueueBinding(
            value = @Queue(name = "test", durable = "true"),
            key = "xxx.#",
            exchange = @Exchange(name = Constants.EVENT_EXCHANGE_NAME, type = ExchangeTypes.TOPIC)
    )
            , containerFactory = "myFactory")
    public void listen(List<Message> messages) {
        messages.size();
    }
}

My breakpoint on messages.size() returns a list of Message.

[arths31]

On further investigation it works in Kotlin with
fun listen(messages: java.util.List<Message>)
It doesn't seem to like kotlin.collections.List or kotlin.collections.Collection

[artembilan]

So, our parameterizedType.getRawType().equals(List.class) cannot be passed because of that kotlin.collections.List.
But at the same time our java.util.List<org.springframework.amqp.core.Message> is able to be passed into that argument.
So, there is some conversion on the Kotlin side from java.util.List into a kotlin.collections.List.

I'm not sure what to do, but it has to be fixed some way indeed.

Raising an issue from this discussion.

[arths31]

Great thank you, I'm using the java.util.List<Message> workaround in the meantime
Have a nice day