JMS - ActiveMQ works, Oracle AQ JMS does not

[Bear Giles]

We're prototyping open source replacements for a legacy application.

When we're running a prototype again ActiveMQ, we don't have any problems. Well, we ended up running spring on the applet side(!) because of an initialization issue with the naked activemq client libraries, but that was a mere annoyance.

When we're running a prototype against the Oracle JMS wrapper on ActiveQueues, we run into a brick wall. (The legacy app uses AQ, we're prototyping on new AQ queues with JMS support enabled.) With the help of some online resources we've been able to get a test case to the point where the JMS calls throw a generic "corrupt data" exception, and the corresponding AQ-specific calls throw a NPE when writing to a HashTable in a 'convertPropertyListToHashtable' call. We're pretty confident that the JMS exception is just a wrapper for this NPE.

This suggests that we haven't specified some message property, but we're coming up blank since it works with ActiveMQ and there's no mention of this in the AQ JMS resources.

BTW this happens with test payloads of Date, String and a String containing XML text.

Ideas?

P.S., we're running 11.1g and saw a similar problem reported about 6 months ago, but no resolution was discussed.

[Bear Giles]

We've made a little progress -- using the native API it looks like the underlying message payload is null. That could reasonably result in an NPE when it hits the JMS wrapper logic.

The test code is:

java.sql.Connection aqconn = oracleDataSource.getConnection();
aqconn.setAutoCommit(false);
AQSession aqsession = null;
 
try {
    AQEnqueueOption enqueueOption = new AQEnqueueOption();
 
    AQDequeueOption dequeueOption = new AQDequeueOption();
    dequeueOption.setConsumerName("SUBSCRIPTION1");
    dequeueOption.setWaitTime(AQDequeueOption.WAIT_NONE);
 
    // -------------------------------------------
    // send a message via native interface
    // -------------------------------------------
    aqsession = AQDriverManager.createAQSession(aqconn);
    AQQueue queue = aqsession.getQueue("casoemaster", "QU_ALERTS");
 
    AQMessage m = ((AQOracleQueue) queue).createMessage();
 
    AQObjectPayload payload = m.getObjectPayload();
    Object payloadData = new XMLType(aqconn, "<?xml version=\"1.0\" ?>\r\n\r\n<foo>bar</foo>");
    payload.setPayloadData(payloadData);
 
    queue.enqueue(enqueueOption, m);
    aqconn.commit();
 
    // -------------------------------------------
    // receive a message via native interface
    // -------------------------------------------
    ORADataFactory factory = new XMLTypeFactory();
 
    AQMessage message = ((AQOracleQueue) queue).dequeue(dequeueOption, factory);
    // same problems if we commit immediately after dequeueing message.
    //aqconn.commit();
 
    AQRawPayload rp = message.getRawPayload();
    System.out.println("raw: " + rp);  // is null
 
    payload = message.getObjectPayload();
 
    XMLType messageData = (XMLType) payload.getPayloadData();
    System.out.println("length: " + messageData.getLength());  // is zero
 
    System.out.println("schema: " + messageData.getSchemaURL()); // throws exception
} finally {
    // clean up
    aqsession.close();
    aqconn.close();
}

We can verify that the XML -is- properly written into oracle tables. We can also verify that we can see messages written into oracle tables via other mechanisms, but we still get a null payload.

[Bear Giles]

A group of us finally figured out how to send/receive native messages. We don't know if it's the -right- approach, but it does successfully send messages.

Enqueue:

String owner = "owner", queue = "queue";
java.sql.Connection aqconn = oracleDataSource.getConnection();
aqconn.setAutoCommit(false);
 
AQSession aqsession = null;
 
try {
    // fluff that doesn't affect operation.
    AQEnqueueOption enqueueOption = new AQEnqueueOption();
 
    aqsession = AQDriverManager.createAQSession(aqconn);
    AQQueue queue = aqsession.getQueue(owner, queue);
    AQMessage msg = ((AQOracleQueue) queue).createMessage();
 
    AQObjectPayload payload = msg.getObjectPayload();
    [B]XMLType payloadData = XMLType.createXML(aqconn, "<foo>bar</foo>");[/B]
    payload.setPayloadData(payloadData);
 
    queue.enqueue(enqueueOption, msg);
    aqconn.commit();
} finally {
    aqsession.close();
    aqconn.close();
}

Dequeue:

String owner = "owner", queue = "queue";
 
java.sql.Connection aqconn = oracleDataSource.getConnection();
aqconn.setAutoCommit(false);
AQSession aqsession = null;
 
try {
    AQDequeueOption dequeueOption = new AQDequeueOption();
    dequeueOption.setConsumerName("SUBSCRIPTION1");
    dequeueOption.setWaitTime(AQDequeueOption.WAIT_NONE);
 
    aqsession = AQDriverManager.createAQSession(aqconn);
 
    AQQueue queue = aqsession.getQueue(owner, queue);
 
    // receive a message via native interface
    ORADataFactory factory = XMLType.getORADataFactory();
 
    AQMessage message = ((AQOracleQueue) queue).dequeue(dequeueOption, factory);
 
    if (message == null) {
       log.info("no messages");
    } else {
       log.info("message id: " + new String(Hex.encodeHex(message.getMessageId())));
 
       [B]XMLType messageData = (XMLType) message.getObjectPayload().getPayloadData();
       String xmlString = messageData.getStringVal();[/B]
       log.info("content: " + xmlString);
    }
 
    aqconn.commit();
} finally {
    aqsession.close();
    aqconn.close();
}

[Bear Giles]

And here is the code that uses AQ JMS. Still not quite pure-JMS, and we still haven't figured out how to get XMLType out of the received message. But it doesn't blow chunks any longer.

Sent:

try {
    conn = factory.createTopicConnection();
    conn.start();
 
    session = conn.createTopicSession(true, Session.CLIENT_ACKNOWLEDGE);
    topic = ((AQjmsSession) session).getTopic(OWNER, TOPIC_NAME);
 
    // create explicit recipients list
    AQjmsAgent[] recipients = new AQjmsAgent[]{
            new AQjmsAgent(SUBSCRIBER_NAME, null)
        };
 
    producer = ((AQjmsSession) session).createPublisher(topic);
    producer.setTimeToLive(900000L); // give us 15 minutes.
 
    // our legacy app writes XMLTypes, so we need to as well.
    [B]java.sql.Connection c = oracleDataSource.getConnection();
    XMLType payload = XMLType.createXML(c, TEST_PAYLOAD);
    byte[] content = payload.toBytes();
    c.close();[/B]
 
    Message msg = session.createObjectMessage(content);
 
    ((AQjmsTopicPublisher) producer).publish(msg, recipients);
    //publisher.publish(msg);
    session.commit();
 
    log.info("successfully sent message");
} finally {
    producer.close();
    session.close();
    conn.stop();
    conn.close();
}

Receiver:

try {
    conn = factory.createTopicConnection();
    conn.start();
 
    session = conn.createTopicSession(true, Session.CLIENT_ACKNOWLEDGE);
    topic = ((AQjmsSession) session).getTopic(OWNER, TOPIC_NAME);
    consumer = ((AQjmsSession) session).createTopicReceiver(topic, SUBSCRIBER_NAME, null);
 
    Message message = consumer.receiveNoWait();
 
    if (message == null) {
       log.info("no messages");
    } else if (!(message instanceof ObjectMessage)) {
       fail("expected ObjectMessage, got " + message.getClass().getName());
    } else {
       Object obj = ((ObjectMessage) message).getObject();
       assertTrue(obj instanceof byte[]);
 
       byte[] content = (byte[]) obj;
       log.info("content: " + new String(content));
 
       [B]// we now have encoded XMLType, but still need to get actual XMLType
       // object from it.
 
       //XMLType messageData = (XMLType) message.getObjectPayload().getPayloadData();
       //String xmlString = messageData.getStringVal();
       //assertEquals(TEST_PAYLOAD, xmlString);[/B]
   }
 
   session.commit();
} finally {
    producer.close();
    session.close();
    conn.stop();
    conn.close();
}