Reputation: 7070
I am reading from RabbitMQ like this:
connection = factory.newConnection();
ch = connection.createChannel() ;
String queueName = managerProps.getProperty("rmq.queue.name");
ch.queueDeclare(queueName ,true,false,false, null) ;
while (true) {
GetResponse chResponse = ch.basicGet(queueName, false);
logger.info("----" + new String(chResponse.getBody(), "UTF-8") + " ---\n");
}
And here is what I see in the log:
[Thread-5] INFO com.mycompany.RmqReader - ----?? ♣wx .com.rabbitmq.jms.client.message.RMQTextMessage $b1213c86-10f4-4113-bd2f-45aaabce083f ♠ ←rmq.jms.meamqpQueueNameq ~ ☺L ♫amqpRoutingKeyq ~ ☺L ☼destinationNameq ~ ☺xp ☺ t ↕jms.durable.queuest !MY.Queue.Name ~ ♦q ~ ♦z ☻O ↔rmq.jms.message.delivery.mode♦ ☻ ↓rmq.jms.message.timestamp♣ ☺j?∟ ↑rmq.jms.message.priority♦ ♦ →rmq.jms.message.expiration♣ ↕rmq.jms.message.i 'ID:b1213c86-10f4-4113-bd2f-45aaabce083f ♂ ◄objectTransaction☺☺ ►templateEndpoin -jtemplate://JSONDeliveryTemplateParallel.java ►deliveryLocatio +jms:queue:My.Queue.Name ►destinationIndex♦ ☺ ♫subsCutOffTime♦ ♀breadcrumbI ♀1149808347.0 ◄globalDeliveryUID♦ ?]? ►subscriptionNam §option_session_pubsub ◄originalMessageI ♀1149808347.0 ↨subscriptionDeliveryUID♦ ??8 ¶transactionTimestamp♣ ☺j??(z ♥R ♥M[
{"OptSession": {.... the actual body is here....}}
] ---
Why do I see the headers here? And how do I actually extract the body?
Upvotes: 0
Views: 536
Reputation: 26
Looks like you have used RMQConnectionFactory to publish the data which is jms compliant and now you are using non jms ConnectionFactory
Please see sample consumer using RMQConnectionFactory https://github.com/kunhaj/samples/blob/master/rabbitmq/src/main/java/RabbitMqConsumer.java
import com.rabbitmq.jms.admin.RMQConnectionFactory;
import javax.jms.*;
/**
* docker run -d --hostname my-rabbit --name
* some-rabbit -p 5672:5672 -p 15672:15672 rabbitmq:3-management
*/
public class RabbitMqConsumer {
public static void main(String[] args) throws Exception {
RMQConnectionFactory connectionFactory = new RMQConnectionFactory();
connectionFactory.setUsername("guest");
connectionFactory.setPassword("guest");
connectionFactory.setVirtualHost("/");
connectionFactory.setHost("localhost");
connectionFactory.setPort(5672);
connectionFactory.setDeclareReplyToDestination(false);
Connection connection = connectionFactory.createConnection();
connection.start();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Queue queue = session.createQueue("TEST-QUEUE");
drainQueue(session, queue);
}
protected static void drainQueue(Session session, Queue queue) throws Exception {
MessageConsumer receiver = session.createConsumer(queue);
Message msg = receiver.receiveNoWait();
while (msg != null) {
String msgBody = ((TextMessage) msg).getText();
System.out.println("recieved" + msgBody);
msg = receiver.receiveNoWait();
}
}
}
Also please see JMS and AMQP 0-9-1 Destination Interoperability https://www.rabbitmq.com/jms-client.html#destination-interoperability
Upvotes: 1