SWS-742 - JMSCorrelationID together with Oracle AQ

Author: poutsma

support/src/main/java/org/springframework/ws/transport/jms/JmsSenderConnection.java

@@ -1,11 +1,11 @@
 /*
- * Copyright 2005-2010 the original author or authors.
+ * Copyright 2005-2012 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
- *      http://www.apache.org/licenses/LICENSE-2.0
+ *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
@@ -32,7 +32,6 @@
 import javax.jms.MessageProducer;
 import javax.jms.Session;
 import javax.jms.TemporaryQueue;
-import javax.jms.TemporaryTopic;
 import javax.jms.TextMessage;
 
 import org.springframework.jms.connection.ConnectionFactoryUtils;
@@ -81,6 +80,8 @@ public class JmsSenderConnection extends AbstractSenderConnection {
 
     private boolean sessionTransacted = false;
 
+    private boolean temporaryResponseQueueCreated = false;
+
     /** Constructs a new JMS connection with the given parameters. */
     protected JmsSenderConnection(ConnectionFactory connectionFactory,
                                   Connection connection,
@@ -211,6 +212,7 @@ protected void onSendAfterWrite(WebServiceMessage message) throws IOException {
             messageProducer.setPriority(priority);
             if (responseDestination == null) {
                 responseDestination = session.createTemporaryQueue();
+                temporaryResponseQueueCreated = true;
             }
             requestMessage.setJMSReplyTo(responseDestination);
             if (postProcessor != null) {
@@ -243,7 +245,7 @@ private boolean isSessionLocallyTransacted(Session session) {
     protected void onReceiveBeforeRead() throws IOException {
         MessageConsumer messageConsumer = null;
         try {
-            if (responseDestination instanceof TemporaryQueue || responseDestination instanceof TemporaryTopic) {
+            if (temporaryResponseQueueCreated) {
                 messageConsumer = session.createConsumer(responseDestination);
             }
             else {
@@ -266,19 +268,11 @@ else if (message != null) {
         }
         finally {
             JmsUtils.closeMessageConsumer(messageConsumer);
-            if (responseDestination instanceof TemporaryQueue) {
+            if (temporaryResponseQueueCreated) {
                 try {
                     ((TemporaryQueue) responseDestination).delete();
                 }
-                catch (JMSException e) {
-                    // ignore
-                }
-            }
-            else if (responseDestination instanceof TemporaryTopic) {
-                try {
-                    ((TemporaryTopic) responseDestination).delete();
-                }
-                catch (JMSException e) {
+                catch (JMSException ex) {
                     // ignore
                 }
             }

support/src/test/java/org/springframework/ws/transport/jms/JmsMessageSenderIntegrationTest.java

@@ -1,11 +1,11 @@
 /*
- * Copyright 2005-2010 the original author or authors.
+ * Copyright 2005-2012 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
- *      http://www.apache.org/licenses/LICENSE-2.0
+ *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
@@ -64,7 +64,7 @@ public void createMessageFactory() throws Exception {
     }
 
     @Test
-    public void testSendAndReceiveQueueBytesMessage() throws Exception {
+    public void testSendAndReceiveQueueBytesMessageTemporaryQueue() throws Exception {
         WebServiceConnection connection = null;
         try {
             URI uri = new URI("jms:SenderRequestQueue?deliveryMode=NON_PERSISTENT");
@@ -102,6 +102,48 @@ public Message createMessage(Session session) throws JMSException {
         }
     }
 
+    @Test
+    public void testSendAndReceiveQueueBytesMessagePermanentQueue() throws Exception {
+        WebServiceConnection connection = null;
+        try {
+            String responseQueueName = "SenderResponseQueue";
+            URI uri = new URI(
+                    "jms:SenderRequestQueue?replyToName=" + responseQueueName + "&deliveryMode=NON_PERSISTENT");
+            connection = messageSender.createConnection(uri);
+            SoapMessage soapRequest = new SaajSoapMessage(messageFactory.createMessage());
+            soapRequest.setSoapAction(SOAP_ACTION);
+            connection.send(soapRequest);
+
+            final BytesMessage request = (BytesMessage) jmsTemplate.receive();
+            assertNotNull("No message received", request);
+            assertTrue("No message content received", request.readByte() != -1);
+            ByteArrayOutputStream bos = new ByteArrayOutputStream();
+            messageFactory.createMessage().writeTo(bos);
+            final byte[] buf = bos.toByteArray();
+            jmsTemplate.send(responseQueueName, new MessageCreator() {
+
+                public Message createMessage(Session session) throws JMSException {
+                    BytesMessage response = session.createBytesMessage();
+                    response.setJMSCorrelationID(request.getJMSMessageID());
+                    response.setStringProperty(JmsTransportConstants.PROPERTY_SOAP_ACTION, SOAP_ACTION);
+                    response.setStringProperty(JmsTransportConstants.PROPERTY_CONTENT_TYPE,
+                            SoapVersion.SOAP_11.getContentType());
+                    response.writeBytes(buf);
+                    return response;
+                }
+            });
+            SoapMessage response = (SoapMessage) connection.receive(new SaajSoapMessageFactory(messageFactory));
+            assertNotNull("No response received", response);
+            assertEquals("Invalid SOAPAction", SOAP_ACTION, response.getSoapAction());
+            assertFalse("Message is fault", response.hasFault());
+        }
+        finally {
+            if (connection != null) {
+                connection.close();
+            }
+        }
+    }
+
     @Test
     public void testSendAndReceiveQueueTextMessage() throws Exception {
         WebServiceConnection connection = null;

support/src/test/resources/org/springframework/ws/transport/jms/jms-sender-applicationContext.xml

@@ -10,6 +10,10 @@
         <property name="physicalName" value="SenderRequestQueue"/>
     </bean>
 
+    <bean id="responseQueue" class="org.apache.activemq.command.ActiveMQQueue">
+        <property name="physicalName" value="SenderResponseQueue"/>
+    </bean>
+
     <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
         <property name="connectionFactory" ref="connectionFactory"/>
         <property name="defaultDestination" ref="requestQueue"/>