Всем привет! Проблема в следующем:
У меня есть приложение, которое слушает определенную очередь в RabbitMQ. Когда в эту очередь поступает сообщение, то оно определенным образом обрабатывается. Проблема в том, что в эту очередь нельзя загружать обычные сообщения без заголовков и кодирования (например, в base64).
Если я помещу в очередь обычное сообщение (ниже есть пример), то в программе будет ошибка в виде Setup of JMS message listener invoker failed for destination 'queue' - trying to recover. Cause: invalid stream header: 304A2F52
.
Причем само сообщение он получает: есть метод RMQMessage#fromMessage(byte[] b, List<String> trustedPackages) throws RMQJMSException
и если там поставить точку остановки, то видно, что в переменную b
приходит именно то, что я отправлял.
Опытным путём я выяснил, что его не устраивает вид, в котором приходит сообщение и понял, что вот такие сообщения программа разбирает правильно:
И как следствие вывод приложения нормальный:
2021-10-15 16:10:12.286 INFO 14264 --- [enerContainer-1] .m.m.w.CustomMessageListener : [x] Received <<< queue Привет
Вопрос в том, как можно перенастроить RabbitMQ Client, чтобы он мог принимать обычные текстовые сообщения, а не закодированные с кучей заголовков.
Зависимость я использую следующую:
<dependency>
<groupId>com.rabbitmq.jms</groupId>
<artifactId>rabbitmq-jms</artifactId>
<version>2.3.0</version>
</dependency>
Назначение слушателей происходит следующим образом (код немного видоизменен и в реальности происходит динамическое создания очередей + назначение им слушателей, но в любом случае это делается через JmsListenerEndpointRegistrar
):
public class JmsListenerConfiguration implements JmsListenerConfigurer {
private final CustomMessageListener messageListener;
@Override
public void configureJmsListeners(JmsListenerEndpointRegistrar registrar) {
final String queue = "queue";
final SimpleJmsListenerEndpoint endpoint = new SimpleJmsListenerEndpoint();
endpoint.setId(queue);
endpoint.setDestination(queue);
endpoint.setMessageListener((message) -> messageListener.onMessage(message, queue));
registrar.registerEndpoint(endpoint);
}
}