From 09bcd7fc38473d81a640d4be248ad9600e3e30c1 Mon Sep 17 00:00:00 2001 From: Arjen Poutsma Date: Wed, 25 Jan 2012 10:57:39 +0000 Subject: [PATCH] SWS-742 - JMSCorrelationID together with Oracle AQ --- .../ws/transport/jms/JmsSenderConnection.java | 22 ++++----- .../jms/JmsMessageSenderIntegrationTest.java | 48 +++++++++++++++++-- .../jms/jms-sender-applicationContext.xml | 4 ++ 3 files changed, 57 insertions(+), 17 deletions(-) diff --git a/support/src/main/java/org/springframework/ws/transport/jms/JmsSenderConnection.java b/support/src/main/java/org/springframework/ws/transport/jms/JmsSenderConnection.java index 54f60824..fdf8b933 100644 --- a/support/src/main/java/org/springframework/ws/transport/jms/JmsSenderConnection.java +++ b/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.MessageConsumer; 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 @@ public class JmsSenderConnection extends AbstractSenderConnection { messageProducer.setPriority(priority); if (responseDestination == null) { responseDestination = session.createTemporaryQueue(); + temporaryResponseQueueCreated = true; } requestMessage.setJMSReplyTo(responseDestination); if (postProcessor != null) { @@ -243,7 +245,7 @@ public class JmsSenderConnection extends AbstractSenderConnection { 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 @@ public class JmsSenderConnection extends AbstractSenderConnection { } 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 } } diff --git a/support/src/test/java/org/springframework/ws/transport/jms/JmsMessageSenderIntegrationTest.java b/support/src/test/java/org/springframework/ws/transport/jms/JmsMessageSenderIntegrationTest.java index 96b40046..2798f7c9 100644 --- a/support/src/test/java/org/springframework/ws/transport/jms/JmsMessageSenderIntegrationTest.java +++ b/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 class JmsMessageSenderIntegrationTest { } @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 class JmsMessageSenderIntegrationTest { } } + @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; diff --git a/support/src/test/resources/org/springframework/ws/transport/jms/jms-sender-applicationContext.xml b/support/src/test/resources/org/springframework/ws/transport/jms/jms-sender-applicationContext.xml index 24d59f16..f9c085c5 100644 --- a/support/src/test/resources/org/springframework/ws/transport/jms/jms-sender-applicationContext.xml +++ b/support/src/test/resources/org/springframework/ws/transport/jms/jms-sender-applicationContext.xml @@ -10,6 +10,10 @@ + + + +