Skip to content

Commit b884681

Browse files
snicollsnicoll
committed
Use JmsListener in JmsMessageSenderIntegrationTests
This commit replaces the use of a synchronous receive operation using JmsTemplate by the use of JmsListener. The latter makes sure that the handling of the request and sending the response happens in the same session and that should hopefully fix the flakiness of this test. Closes gh-1539
1 parent 185a637 commit b884681

File tree

1 file changed

+128
-106
lines changed

1 file changed

+128
-106
lines changed

‎spring-ws-support/src/test/java/org/springframework/ws/transport/jms/JmsMessageSenderIntegrationTests.java‎

Lines changed: 128 additions & 106 deletions
Original file line numberDiff line numberDiff line change
@@ -17,22 +17,34 @@
1717
package org.springframework.ws.transport.jms;
1818

1919
import java.io.ByteArrayOutputStream;
20+
import java.io.IOException;
2021
import java.net.URI;
2122
import java.nio.charset.StandardCharsets;
23+
import java.time.Duration;
2224

2325
import jakarta.jms.BytesMessage;
26+
import jakarta.jms.ConnectionFactory;
27+
import jakarta.jms.JMSException;
28+
import jakarta.jms.Message;
2429
import jakarta.jms.TextMessage;
2530
import jakarta.xml.soap.MessageFactory;
2631
import jakarta.xml.soap.SOAPConstants;
27-
import org.junit.jupiter.api.BeforeEach;
32+
import jakarta.xml.soap.SOAPException;
33+
import org.apache.activemq.ActiveMQConnectionFactory;
2834
import org.junit.jupiter.api.Test;
29-
import org.junit.jupiter.api.extension.ExtendWith;
3035

3136
import org.springframework.beans.factory.annotation.Autowired;
32-
import org.springframework.jms.core.JmsTemplate;
37+
import org.springframework.context.annotation.Bean;
38+
import org.springframework.context.annotation.Configuration;
39+
import org.springframework.context.annotation.Import;
40+
import org.springframework.jms.annotation.EnableJms;
41+
import org.springframework.jms.annotation.JmsListener;
42+
import org.springframework.jms.config.DefaultJmsListenerContainerFactory;
3343
import org.springframework.jms.core.MessagePostProcessor;
34-
import org.springframework.test.context.ContextConfiguration;
35-
import org.springframework.test.context.junit.jupiter.SpringExtension;
44+
import org.springframework.messaging.handler.annotation.SendTo;
45+
import org.springframework.messaging.support.MessageBuilder;
46+
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
47+
import org.springframework.util.function.ThrowingFunction;
3648
import org.springframework.ws.soap.SoapMessage;
3749
import org.springframework.ws.soap.SoapVersion;
3850
import org.springframework.ws.soap.saaj.SaajSoapMessage;
@@ -41,57 +53,33 @@
4153

4254
import static org.assertj.core.api.Assertions.assertThat;
4355

44-
@ExtendWith(SpringExtension.class)
45-
@ContextConfiguration("jms-sender-applicationContext.xml")
56+
@SpringJUnitConfig
4657
class JmsMessageSenderIntegrationTests {
4758

59+
private static final String SOAP_ACTION = "\"http://springframework.org/DoIt\"";
60+
61+
private static final MessageFactory messageFactory = createMessageFactory();
62+
4863
@Autowired
4964
private JmsMessageSender messageSender;
5065

5166
@Autowired
52-
private JmsTemplate jmsTemplate;
53-
54-
private MessageFactory messageFactory;
55-
56-
private static final String SOAP_ACTION = "\"http://springframework.org/DoIt\"";
57-
58-
@BeforeEach
59-
void createMessageFactory() throws Exception {
60-
this.messageFactory = MessageFactory.newInstance(SOAPConstants.SOAP_1_1_PROTOCOL);
61-
}
67+
private TestJmsListener testJmsListener;
6268

6369
@Test
6470
void testSendAndReceiveQueueBytesMessageTemporaryQueue() throws Exception {
65-
6671
URI uri = new URI("jms:SenderRequestQueue?deliveryMode=NON_PERSISTENT");
67-
6872
try (WebServiceConnection connection = this.messageSender.createConnection(uri)) {
69-
70-
SoapMessage soapRequest = new SaajSoapMessage(this.messageFactory.createMessage());
73+
SoapMessage soapRequest = new SaajSoapMessage(messageFactory.createMessage());
7174
soapRequest.setSoapAction(SOAP_ACTION);
7275
connection.send(soapRequest);
7376

74-
BytesMessage request = (BytesMessage) this.jmsTemplate.receive();
75-
76-
assertThat(request).isNotNull();
77-
assertThat(request.readByte()).isNotEqualTo(-1);
78-
79-
ByteArrayOutputStream bos = new ByteArrayOutputStream();
80-
this.messageFactory.createMessage().writeTo(bos);
81-
final byte[] buf = bos.toByteArray();
82-
83-
this.jmsTemplate.send(request.getJMSReplyTo(), session -> {
84-
85-
BytesMessage response = session.createBytesMessage();
86-
response.setStringProperty(JmsTransportConstants.PROPERTY_SOAP_ACTION, SOAP_ACTION);
87-
response.setStringProperty(JmsTransportConstants.PROPERTY_CONTENT_TYPE,
88-
SoapVersion.SOAP_11.getContentType());
89-
response.writeBytes(buf);
90-
return response;
77+
this.testJmsListener.handleMessage((message) -> {
78+
assertNonEmptyByteMessage(message);
79+
return createEmptySoapMessage();
9180
});
9281

93-
SoapMessage response = (SoapMessage) connection.receive(new SaajSoapMessageFactory(this.messageFactory));
94-
82+
SoapMessage response = (SoapMessage) connection.receive(new SaajSoapMessageFactory(messageFactory));
9583
assertThat(response).isNotNull();
9684
assertThat(response.getSoapAction()).isEqualTo(SOAP_ACTION);
9785
assertThat(response.hasFault()).isFalse();
@@ -100,38 +88,20 @@ void testSendAndReceiveQueueBytesMessageTemporaryQueue() throws Exception {
10088

10189
@Test
10290
void testSendAndReceiveQueueBytesMessagePermanentQueue() throws Exception {
103-
10491
String responseQueueName = "SenderResponseQueue";
10592
URI uri = new URI("jms:SenderRequestQueue?replyToName=" + responseQueueName + "&deliveryMode=NON_PERSISTENT");
10693

10794
try (WebServiceConnection connection = this.messageSender.createConnection(uri)) {
108-
109-
SoapMessage soapRequest = new SaajSoapMessage(this.messageFactory.createMessage());
95+
SoapMessage soapRequest = new SaajSoapMessage(messageFactory.createMessage());
11096
soapRequest.setSoapAction(SOAP_ACTION);
11197
connection.send(soapRequest);
11298

113-
final BytesMessage request = (BytesMessage) this.jmsTemplate.receive();
114-
115-
assertThat(request).isNotNull();
116-
assertThat(request.readByte()).isNotEqualTo(-1);
117-
118-
ByteArrayOutputStream bos = new ByteArrayOutputStream();
119-
this.messageFactory.createMessage().writeTo(bos);
120-
final byte[] buf = bos.toByteArray();
121-
122-
this.jmsTemplate.send(responseQueueName, session -> {
123-
124-
BytesMessage response = session.createBytesMessage();
125-
response.setJMSCorrelationID(request.getJMSMessageID());
126-
response.setStringProperty(JmsTransportConstants.PROPERTY_SOAP_ACTION, SOAP_ACTION);
127-
response.setStringProperty(JmsTransportConstants.PROPERTY_CONTENT_TYPE,
128-
SoapVersion.SOAP_11.getContentType());
129-
response.writeBytes(buf);
130-
return response;
99+
this.testJmsListener.handleMessage((message) -> {
100+
assertNonEmptyByteMessage(message);
101+
return createEmptySoapMessage();
131102
});
132103

133-
SoapMessage response = (SoapMessage) connection.receive(new SaajSoapMessageFactory(this.messageFactory));
134-
104+
SoapMessage response = (SoapMessage) connection.receive(new SaajSoapMessageFactory(messageFactory));
135105
assertThat(response).isNotNull();
136106
assertThat(response.getSoapAction()).isEqualTo(SOAP_ACTION);
137107
assertThat(response.hasFault()).isFalse();
@@ -140,36 +110,18 @@ void testSendAndReceiveQueueBytesMessagePermanentQueue() throws Exception {
140110

141111
@Test
142112
void testSendAndReceiveQueueTextMessage() throws Exception {
143-
144113
URI uri = new URI("jms:SenderRequestQueue?deliveryMode=NON_PERSISTENT&messageType=TEXT_MESSAGE");
145-
146114
try (WebServiceConnection connection = this.messageSender.createConnection(uri)) {
147-
148-
SoapMessage soapRequest = new SaajSoapMessage(this.messageFactory.createMessage());
115+
SoapMessage soapRequest = new SaajSoapMessage(messageFactory.createMessage());
149116
soapRequest.setSoapAction(SOAP_ACTION);
150117
connection.send(soapRequest);
151118

152-
TextMessage request = (TextMessage) this.jmsTemplate.receive();
153-
154-
assertThat(request).isNotNull();
155-
assertThat(request.getText()).isNotNull();
156-
157-
ByteArrayOutputStream bos = new ByteArrayOutputStream();
158-
this.messageFactory.createMessage().writeTo(bos);
159-
final String text = bos.toString(StandardCharsets.UTF_8);
160-
161-
this.jmsTemplate.send(request.getJMSReplyTo(), session -> {
162-
163-
TextMessage response = session.createTextMessage();
164-
response.setStringProperty(JmsTransportConstants.PROPERTY_SOAP_ACTION, SOAP_ACTION);
165-
response.setStringProperty(JmsTransportConstants.PROPERTY_CONTENT_TYPE,
166-
SoapVersion.SOAP_11.getContentType());
167-
response.setText(text);
168-
return response;
119+
this.testJmsListener.handleMessage((message) -> {
120+
assertNonEmptyTextMessage(message);
121+
return createEmptySoapMessage();
169122
});
170123

171-
SoapMessage response = (SoapMessage) connection.receive(new SaajSoapMessageFactory(this.messageFactory));
172-
124+
SoapMessage response = (SoapMessage) connection.receive(new SaajSoapMessageFactory(messageFactory));
173125
assertThat(response).isNotNull();
174126
assertThat(response.getSoapAction()).isEqualTo(SOAP_ACTION);
175127
assertThat(response.hasFault()).isFalse();
@@ -178,49 +130,119 @@ void testSendAndReceiveQueueTextMessage() throws Exception {
178130

179131
@Test
180132
void testSendNoResponse() throws Exception {
181-
182133
URI uri = new URI("jms:SenderRequestQueue?deliveryMode=NON_PERSISTENT");
183-
184134
try (WebServiceConnection connection = this.messageSender.createConnection(uri)) {
185-
186-
SoapMessage soapRequest = new SaajSoapMessage(this.messageFactory.createMessage());
135+
SoapMessage soapRequest = new SaajSoapMessage(messageFactory.createMessage());
187136
soapRequest.setSoapAction(SOAP_ACTION);
188137
connection.send(soapRequest);
189138

190-
BytesMessage request = (BytesMessage) this.jmsTemplate.receive();
191-
192-
assertThat(request).isNotNull();
193-
194-
ByteArrayOutputStream bos = new ByteArrayOutputStream();
195-
this.messageFactory.createMessage().writeTo(bos);
196-
SoapMessage response = (SoapMessage) connection.receive(new SaajSoapMessageFactory(this.messageFactory));
139+
this.testJmsListener.handleMessage((message) -> {
140+
assertNonEmptyByteMessage(message);
141+
return null;
142+
});
197143

144+
SoapMessage response = (SoapMessage) connection.receive(new SaajSoapMessageFactory(messageFactory));
198145
assertThat(response).isNull();
199146
}
200147
}
201148

202149
@Test
203150
void testPostProcessor() throws Exception {
204-
205151
MessagePostProcessor processor = message -> {
206-
207152
message.setBooleanProperty("processed", true);
208153
return message;
209154
};
210-
211155
URI uri = new URI("jms:SenderRequestQueue?deliveryMode=NON_PERSISTENT");
212-
213156
try (JmsSenderConnection connection = (JmsSenderConnection) this.messageSender.createConnection(uri)) {
214-
215157
connection.setPostProcessor(processor);
216-
SoapMessage soapRequest = new SaajSoapMessage(this.messageFactory.createMessage());
158+
SoapMessage soapRequest = new SaajSoapMessage(messageFactory.createMessage());
217159
connection.send(soapRequest);
218160

219-
BytesMessage request = (BytesMessage) this.jmsTemplate.receive();
161+
this.testJmsListener.handleMessage((message) -> {
162+
assertNonEmptyByteMessage(message);
163+
assertThat(message.getBooleanProperty("processed")).isTrue();
164+
return null;
165+
});
166+
}
167+
}
168+
169+
private Object createEmptySoapMessage() throws SOAPException, IOException {
170+
ByteArrayOutputStream bos = new ByteArrayOutputStream();
171+
messageFactory.createMessage().writeTo(bos);
172+
String text = bos.toString(StandardCharsets.UTF_8);
173+
return MessageBuilder.withPayload(text)
174+
.setHeader(JmsTransportConstants.PROPERTY_SOAP_ACTION, SOAP_ACTION)
175+
.setHeader(JmsTransportConstants.PROPERTY_CONTENT_TYPE, SoapVersion.SOAP_11.getContentType())
176+
.build();
177+
}
178+
179+
private static void assertNonEmptyByteMessage(jakarta.jms.Message message) throws JMSException {
180+
if (message instanceof BytesMessage bytesMessage) {
181+
assertThat(bytesMessage.readByte()).isNotEqualTo(-1);
182+
}
183+
else {
184+
throw new IllegalStateException("Unexpected message type: " + message.getClass().getName());
185+
}
186+
}
187+
188+
private static void assertNonEmptyTextMessage(jakarta.jms.Message message) throws JMSException {
189+
if (message instanceof TextMessage textMessage) {
190+
assertThat(textMessage.getText()).isNotEmpty();
191+
}
192+
else {
193+
throw new IllegalStateException("Unexpected message type: " + message.getClass().getName());
194+
}
195+
}
196+
197+
private static MessageFactory createMessageFactory() {
198+
try {
199+
return MessageFactory.newInstance(SOAPConstants.SOAP_1_1_PROTOCOL);
200+
}
201+
catch (SOAPException ex) {
202+
throw new IllegalStateException(ex);
203+
}
204+
}
205+
206+
static class TestJmsListener {
207+
208+
private ThrowingFunction<jakarta.jms.Message, Object> messageHandler;
209+
210+
public void handleMessage(ThrowingFunction<Message, Object> messageHandler) {
211+
this.messageHandler = messageHandler;
212+
}
213+
214+
@JmsListener(destination = "SenderRequestQueue")
215+
@SendTo("SenderResponseQueue")
216+
Object handleRequest(jakarta.jms.Message message) {
217+
return this.messageHandler.apply(message);
218+
}
219+
220+
}
220221

221-
assertThat(request).isNotNull();
222-
assertThat(request.getBooleanProperty("processed")).isTrue();
222+
@Configuration(proxyBeanMethods = false)
223+
@EnableJms
224+
@Import(TestJmsListener.class)
225+
static class JmsBrokerConfig {
226+
227+
@Bean
228+
ActiveMQConnectionFactory connectionFactory() {
229+
return new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false");
223230
}
231+
232+
@Bean
233+
public DefaultJmsListenerContainerFactory jmsListenerContainerFactory(ConnectionFactory connectionFactory) {
234+
DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
235+
factory.setConnectionFactory(connectionFactory);
236+
return factory;
237+
}
238+
239+
@Bean
240+
JmsMessageSender messageSender(ConnectionFactory connectionFactory) {
241+
JmsMessageSender messageSender = new JmsMessageSender(connectionFactory);
242+
messageSender.setReceiveTimeout(Duration.ofSeconds(2).toMillis());
243+
return messageSender;
244+
}
245+
224246
}
225247

226248
}

0 commit comments

Comments
 (0)