Rabbitmq JMS client get TextMessage

Mudit Mishra

unread,

Mar 25, 2018, 7:36:19 AM3/25/18

to rabbitmq-users

We have an application engine that supports only JMS interface and hence we have to use JMS client of Rabbitmq to read messages.

Our application expects JMS message to be type TextMessage. Rabbitmq message receive function returns message of type RMQBytesMessage.

How can message type be changed to TextMessage?

Error:

03-25 22:35:01,685 [main][ERROR SendReceiveJMSExample] - Error

java.lang.ClassCastException: com.rabbitmq.jms.client.message.RMQBytesMessage cannot be cast to javax.jms.TextMessage

at SendReceiveJMSExample.runTest(SendReceiveJMSExample.java:54)

at SendReceiveJMSExample.main(SendReceiveJMSExample.java:24)

Exception in thread "main" java.lang.NullPointerException

at SendReceiveJMSExample.runTest(SendReceiveJMSExample.java:72)

at SendReceiveJMSExample.main(SendReceiveJMSExample.java:24)

export CLASSPATH=.:/tmp/rabbitmq/jars/jdk17/rabbitmq-jms-1.7.0.jar:\

/tmp/rabbitmq/jars/jdk17/amqp-client-4.2.0.jar:\

/tmp/rabbitmq/jars/slf4j-log4j12.jar:\

/tmp/rabbitmq/jars/jdk17/rabbitmq-jms-1.7.0.jar:\

/tmp/rabbitmq/jars/providerutil-1.2.1.jar:\

/tmp/rabbitmq/jars/geronimo-jms_1.1_spec-1.1.1.jar:\

/tmp/rabbitmq/jars/slf4j-api.jar:\

/tmp/rabbitmq/jars/fscontext-4.6-b01.jar:\

/tmp/rabbitmq/jars/log4j.jar

$ javac SendReceiveJMSExample.java

$ java -Dlog4j.configuration=file:"/opt/sas/vicpol/sashome/SASFoundation/9.4/misc/tkjava/sas.log4j.properties" SendReceiveJMSExample

Java code sample:

import javax.jms.*;

//import com.rabbitmq.*;

import javax.naming.Context;

import javax.naming.InitialContext;

import java.util.Properties;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

public class SendReceiveJMSExample {

protected final Logger logger = LoggerFactory.getLogger(SendReceiveJMSExample.class);

private Context ctx;

private Connection connection;

private MessageConsumer messageConsumer;

private MessageProducer messageProducer;

private Session session;

public SendReceiveJMSExample() {

}

public static void main(String[] args) throws Exception {

SendReceiveJMSExample example = new SendReceiveJMSExample();

example.runTest();

}

private void runTest() throws Exception {

Properties env = new Properties();

env.put(Context.INITIAL_CONTEXT_FACTORY, "com.sun.jndi.fscontext.RefFSContextFactory");

env.put(Context.PROVIDER_URL, "file:///tmp/rabbitmq/jndi");

Properties properties = new Properties();

try {

ctx = new InitialContext(env);

ConnectionFactory connectionFactory

= (ConnectionFactory) ctx.lookup("ConnectionFactory");

connection = connectionFactory.createConnection();

connection.start();

session = connection.createSession(true, Session.SESSION_TRANSACTED);

Queue queue = (Queue) ctx.lookup("sas.audit.queue");

messageConsumer = session.createConsumer(queue);

MessageProducer messageProducer = session.createProducer(queue);

TextMessage message = session.createTextMessage("Hello world!");

messageProducer.send(message);

session.commit();

//RMQBytesMessage message_bt;

//Class cls = messageConsumer.receive().getClass();

//Class cls = message_bt.getClass();

//System.out.println("The type of the object is: " + cls.getName());

message = (TextMessage)messageConsumer.receive(1000);

//Message message_cls = messageConsumer.receive(1000);

//System.out.println("JMSType = " + message_cls.getJMSType());

session.commit();

System.out.println(message.getText());

session.close();

messageConsumer.close();

messageProducer.close();

connection.close();

ctx.close();

}

catch (Exception e) {

//System.out.println(e);

//e.printStackTrace();

logger.error("Error", e);

session.close();

messageConsumer.close();

messageProducer.close();

connection.close();

ctx.close();

}

finally {

connection.close();

ctx.close();

}

//properties.load(this.getClass().getResourceAsStream("/tmp/rabbitmq/jndi/helloworld.properties"));

//Context context = new InitialContext(properties);

}

Mudit Mishra

unread,

Mar 25, 2018, 11:45:54 PM3/25/18

to rabbitmq-users

Added JMSType header to the message and then message could be read by the below code. We cannot set JMSType header for every message.

Arnaud Cogoluègnes

unread,

Mar 26, 2018, 8:58:41 AM3/26/18

to rabbitm...@googlegroups.com

Can you share the definition of the queue? I think you defined it as an AMQP resource, this may be not what you want.

--
You received this message because you are subscribed to the Google Groups "rabbitmq-users" group.
To unsubscribe from this group and stop receiving emails from it, send an email to rabbitmq-users+unsubscribe@googlegroups.com.
To post to this group, send email to rabbitmq-users@googlegroups.com.
For more options, visit https://groups.google.com/d/optout.

Mudit Mishra

unread,

Mar 26, 2018, 7:52:33 PM3/26/18

to rabbitm...@googlegroups.com

[root@rabbitmq1 rabbitmq]# rabbitmqctl report

Reporting server status of node rabbit@rabbitmq1 ...

Status of node rabbit@rabbitmq1 ...

[{pid,2052},

{running_applications,

[{rabbitmq_management,"RabbitMQ Management Console","3.7.4"},

{amqp_client,"RabbitMQ AMQP Client","3.7.4"},

{rabbitmq_management_agent,"RabbitMQ Management Agent","3.7.4"},

{rabbitmq_web_dispatch,"RabbitMQ Web Dispatcher","3.7.4"},

{rabbitmq_jms_topic_exchange,

"RabbitMQ JMS topic selector exchange plugin","3.7.4"},

{rabbit,"RabbitMQ","3.7.4"},

{rabbit_common,

"Modules shared by rabbitmq-server and rabbitmq-erlang-client",

"3.7.4"},

{xmerl,"XML parser","1.3.16"},

{ranch_proxy_protocol,"Ranch Proxy Protocol Transport","1.4.4"},

{cowboy,"Small, fast, modern HTTP server.","2.2.2"},

{ranch,"Socket acceptor pool for TCP protocols.","1.4.0"},

{ssl,"Erlang/OTP SSL application","8.2.4"},

{public_key,"Public key infrastructure","1.5.2"},

{asn1,"The Erlang ASN1 compiler version 5.0.5","5.0.5"},

{recon,"Diagnostic tools for production use","2.3.2"},

{cowlib,"Support library for manipulating Web protocols.","2.1.0"},

{crypto,"CRYPTO","4.2.1"},

{os_mon,"CPO CXC 138 46","2.4.4"},

{jsx,"a streaming, evented json parsing toolkit","2.8.2"},

{inets,"INETS CXC 138 49","6.5"},

{mnesia,"MNESIA CXC 138 12","4.15.3"},

{lager,"Erlang logging framework","3.5.1"},

{goldrush,"Erlang event stream processor","0.1.9"},

{compiler,"ERTS CXC 138 10","7.1.5"},

{syntax_tools,"Syntax tools","2.1.4"},

{sasl,"SASL CXC 138 11","3.1.1"},

{stdlib,"ERTS CXC 138 10","3.4.4"},

{kernel,"ERTS CXC 138 10","5.4.3"}]},

{os,{unix,linux}},

{erlang_version,

"Erlang/OTP 20 [erts-9.3] [source] [64-bit] [smp:1:1] [ds:1:1:10] [async-threads:64] [hipe] [kernel-poll:true]\n"},

{memory,

[{connection_readers,0},

{connection_writers,0},

{connection_channels,0},

{connection_other,24672},

{queue_procs,22576},

{queue_slave_procs,0},

{plugins,1794256},

{other_proc,25587848},

{metrics,197664},

{mgmt_db,390184},

{mnesia,81704},

{other_ets,2194168},

{binary,560752},

{msg_index,28912},

{code,28471743},

{atom,1123529},

{other_system,9654848},

{allocated_unused,17914760},

{reserved_unallocated,0},

{strategy,rss},

{total,[{erlang,70132856},{rss,81543168},{allocated,88047616}]}]},

{alarms,[]},

{listeners,[{clustering,25672,"::"},{amqp,5672,"::"},{http,15672,"::"}]},

{vm_memory_calculation_strategy,rss},

{vm_memory_high_watermark,0.4},

{vm_memory_limit,205558579},

{disk_free_limit,50000000},

{disk_free,33685524480},

{file_descriptors,

[{total_limit,924},{total_used,3},{sockets_limit,829},{sockets_used,0}]},

{processes,[{limit,1048576},{used,369}]},

{run_queue,0},

{uptime,60649},

{kernel,{net_ticktime,60}}]

Cluster status of node rabbit@rabbitmq1 ...

[{nodes,[{disc,[rabbit@rabbitmq1]}]},

{running_nodes,[rabbit@rabbitmq1]},

{cluster_name,<<"rab...@rabbitmq1.intranet.local">>},

{partitions,[]},

{alarms,[{rabbit@rabbitmq1,[]}]}]

Application environment of node rabbit@rabbitmq1 ...

[{amqp_client,[{prefer_ipv6,false},{ssl_options,[]}]},

{asn1,[]},

{compiler,[]},

{cowboy,[]},

{cowlib,[]},

{crypto,[{fips_mode,false}]},

{goldrush,[]},

{inets,[]},

{jsx,[]},

{kernel,

[{error_logger,tty},

{inet_default_connect_options,[{nodelay,true}]},

{inet_dist_listen_max,25672},

{inet_dist_listen_min,25672}]},

{lager,

[{async_threshold,20},

{async_threshold_window,5},

{colored,false},

{colors,

[{debug,"\e[0;38m"},

{info,"\e[1;37m"},

{notice,"\e[1;36m"},

{warning,"\e[1;33m"},

{error,"\e[1;31m"},

{critical,"\e[1;35m"},

{alert,"\e[1;44m"},

{emergency,"\e[1;41m"}]},

{crash_log,"log/crash.log"},

{crash_log_count,5},

{crash_log_date,"$D0"},

{crash_log_msg_size,65536},

{crash_log_size,10485760},

{error_logger_format_raw,true},

{error_logger_hwm,50},

{error_logger_hwm_original,50},

{error_logger_redirect,true},

{extra_sinks,

[{error_logger_lager_event,

[{handlers,[{lager_forwarder_backend,[lager_event,inherit]}]},

{rabbit_handlers,