More Lambdas

Also, make inner ctors package rather than private to avoid the synthetic class and method
the compiler has to create.

See: http://stackoverflow.com/questions/921025/eclipse-warning-about-synthetic-accessor-for-private-static-nested-classes-in-jav

JMX Lambdas

Polishing - clean up

More
This commit is contained in:
Gary Russell
2016-10-25 16:17:21 -04:00
committed by Artem Bilan
parent 7b1d43a6dc
commit c865d38576
55 changed files with 1015 additions and 1746 deletions

View File

@@ -117,7 +117,7 @@ public class ResourceInboundChannelAdapterParserTests {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"ResourcePatternResolver-config-usage.xml", this.getClass());
QueueChannel resultChannel = context.getBean("resultChannel", QueueChannel.class);
Message<Resource[]> message = (Message<Resource[]>) resultChannel.receive(3000);
Message<Resource[]> message = (Message<Resource[]>) resultChannel.receive(10000);
assertNotNull(message);
Resource[] resources = message.getPayload();
for (Resource resource : resources) {
@@ -148,7 +148,7 @@ public class ResourceInboundChannelAdapterParserTests {
assertFalse(customFilter.invoked);
resourceAdapter.start();
QueueChannel resultChannel = context.getBean("resultChannel", QueueChannel.class);
Message<Resource[]> message = (Message<Resource[]>) resultChannel.receive(1000);
Message<Resource[]> message = (Message<Resource[]>) resultChannel.receive(10000);
assertNotNull(message);
assertTrue(customFilter.invoked);
context.close();
@@ -183,6 +183,7 @@ public class ResourceInboundChannelAdapterParserTests {
this.invoked = true;
return unfilteredResources;
}
}
}

View File

@@ -32,8 +32,6 @@ import java.util.TimeZone;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.PropertyAccessor;
@@ -121,24 +119,19 @@ public class HttpProxyScenarioTests {
final String contentDispositionValue = "attachment; filename=\"test.txt\"";
Mockito.doAnswer(new Answer<ResponseEntity<?>>() {
@Override
public ResponseEntity<?> answer(InvocationOnMock invocation) throws Throwable {
URI uri = (URI) invocation.getArguments()[0];
assertEquals(new URI("http://testServer/test?foo=bar&FOO=BAR"), uri);
HttpEntity<?> httpEntity = (HttpEntity<?>) invocation.getArguments()[2];
HttpHeaders httpHeaders = httpEntity.getHeaders();
assertEquals(ifModifiedSince, httpHeaders.getIfModifiedSince());
assertEquals(ifUnmodifiedSinceValue, httpHeaders.getFirst("If-Unmodified-Since"));
assertEquals("Keep-Alive", httpHeaders.getFirst("Connection"));
MultiValueMap<String, String> responseHeaders = new LinkedMultiValueMap<String, String>(httpHeaders);
responseHeaders.set("Connection", "close");
responseHeaders.set("Content-Disposition", contentDispositionValue);
return new ResponseEntity<Object>(responseHeaders, HttpStatus.OK);
}
Mockito.doAnswer(invocation -> {
URI uri = (URI) invocation.getArguments()[0];
assertEquals(new URI("http://testServer/test?foo=bar&FOO=BAR"), uri);
HttpEntity<?> httpEntity = (HttpEntity<?>) invocation.getArguments()[2];
HttpHeaders httpHeaders = httpEntity.getHeaders();
assertEquals(ifModifiedSince, httpHeaders.getIfModifiedSince());
assertEquals(ifUnmodifiedSinceValue, httpHeaders.getFirst("If-Unmodified-Since"));
assertEquals("Keep-Alive", httpHeaders.getFirst("Connection"));
MultiValueMap<String, String> responseHeaders = new LinkedMultiValueMap<String, String>(httpHeaders);
responseHeaders.set("Connection", "close");
responseHeaders.set("Content-Disposition", contentDispositionValue);
return new ResponseEntity<Object>(responseHeaders, HttpStatus.OK);
}).when(template).exchange(Mockito.any(URI.class), Mockito.any(HttpMethod.class),
Mockito.any(HttpEntity.class), (Class<?>) Mockito.any(Class.class));
@@ -179,28 +172,23 @@ public class HttpProxyScenarioTests {
MockHttpServletResponse response = new MockHttpServletResponse();
RestTemplate template = Mockito.spy(new RestTemplate());
Mockito.doAnswer(new Answer<ResponseEntity<?>>() {
Mockito.doAnswer(invocation -> {
URI uri = (URI) invocation.getArguments()[0];
assertEquals(new URI("http://testServer/testmp"), uri);
HttpEntity<?> httpEntity = (HttpEntity<?>) invocation.getArguments()[2];
HttpHeaders httpHeaders = httpEntity.getHeaders();
assertEquals("Keep-Alive", httpHeaders.getFirst("Connection"));
assertEquals("multipart/form-data;boundary=----WebKitFormBoundarywABD2xqC1FLBijlQ",
httpHeaders.getContentType().toString());
@Override
public ResponseEntity<?> answer(InvocationOnMock invocation) throws Throwable {
URI uri = (URI) invocation.getArguments()[0];
assertEquals(new URI("http://testServer/testmp"), uri);
HttpEntity<?> httpEntity = (HttpEntity<?>) invocation.getArguments()[2];
HttpHeaders httpHeaders = httpEntity.getHeaders();
assertEquals("Keep-Alive", httpHeaders.getFirst("Connection"));
assertEquals("multipart/form-data;boundary=----WebKitFormBoundarywABD2xqC1FLBijlQ",
httpHeaders.getContentType().toString());
HttpEntity<?> entity = (HttpEntity<?>) invocation.getArguments()[2];
assertThat(entity.getBody(), instanceOf(byte[].class));
assertEquals("foo", new String((byte[]) entity.getBody()));
MultiValueMap<String, String> responseHeaders = new LinkedMultiValueMap<String, String>(httpHeaders);
responseHeaders.set("Connection", "close");
responseHeaders.set("Content-Type", "text/plain");
return new ResponseEntity<Object>(responseHeaders, HttpStatus.OK);
}
HttpEntity<?> entity = (HttpEntity<?>) invocation.getArguments()[2];
assertThat(entity.getBody(), instanceOf(byte[].class));
assertEquals("foo", new String((byte[]) entity.getBody()));
MultiValueMap<String, String> responseHeaders = new LinkedMultiValueMap<String, String>(httpHeaders);
responseHeaders.set("Connection", "close");
responseHeaders.set("Content-Type", "text/plain");
return new ResponseEntity<Object>(responseHeaders, HttpStatus.OK);
}).when(template).exchange(Mockito.any(URI.class), Mockito.any(HttpMethod.class),
Mockito.any(HttpEntity.class), (Class<?>) Mockito.any(Class.class));

View File

@@ -46,7 +46,6 @@ import org.springframework.expression.common.LiteralExpression;
import org.springframework.expression.spel.standard.SpelExpression;
import org.springframework.http.HttpHeaders;
import org.springframework.http.converter.HttpMessageConverter;
import org.springframework.integration.MessageRejectedException;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.http.converter.SerializingHttpMessageConverter;
import org.springframework.integration.http.inbound.HttpRequestHandlingController;
@@ -54,9 +53,7 @@ import org.springframework.integration.http.inbound.HttpRequestHandlingMessaging
import org.springframework.integration.http.support.DefaultHttpHeaderMapper;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageDeliveryException;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.SubscribableChannel;
@@ -273,12 +270,7 @@ public class HttpInboundGatewayParserTests {
@SuppressWarnings("rawtypes")
private MessageHandler handlerExpecting(final Matcher<Message> messageMatcher) {
return new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessageRejectedException, MessageHandlingException, MessageDeliveryException {
assertThat(message, is(messageMatcher));
}
};
return message -> assertThat(message, is(messageMatcher));
}

View File

@@ -29,18 +29,19 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.commons.logging.LogFactory;
import org.junit.Test;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.expression.Expression;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.http.HttpStatus;
import org.springframework.messaging.Message;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.http.AbstractHttpInboundTests;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.mock.web.MockHttpServletRequest;
import org.springframework.mock.web.MockHttpServletResponse;
import org.springframework.validation.Errors;
@@ -349,26 +350,24 @@ public class HttpRequestHandlingControllerTests extends AbstractHttpInboundTests
MockHttpServletResponse response = new MockHttpServletResponse();
final AtomicInteger active = new AtomicInteger();
final AtomicBoolean expected503 = new AtomicBoolean();
Executors.newSingleThreadExecutor().execute(new Runnable() {
public void run() {
try {
// wait for the active thread
latch2.await(10, TimeUnit.SECONDS);
}
catch (InterruptedException e1) {
Thread.currentThread().interrupt();
}
// start the shutdown
active.set(controller.beforeShutdown());
try {
MockHttpServletResponse response = new MockHttpServletResponse();
controller.handleRequest(request, response);
expected503.set(response.getStatus() == HttpStatus.SERVICE_UNAVAILABLE.value());
latch1.countDown();
}
catch (Exception e) {
e.printStackTrace();
}
Executors.newSingleThreadExecutor().execute(() -> {
try {
// wait for the active thread
latch2.await(10, TimeUnit.SECONDS);
}
catch (InterruptedException e1) {
Thread.currentThread().interrupt();
}
// start the shutdown
active.set(controller.beforeShutdown());
try {
MockHttpServletResponse response1 = new MockHttpServletResponse();
controller.handleRequest(request, response1);
expected503.set(response1.getStatus() == HttpStatus.SERVICE_UNAVAILABLE.value());
latch1.countDown();
}
catch (Exception e) {
LogFactory.getLog(getClass()).error("Async handleRequest failed", e);
}
});
ModelAndView modelAndView = controller.handleRequest(request, response);

View File

@@ -36,10 +36,7 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.integration.http.AbstractHttpInboundTests;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.mock.web.MockHttpServletRequest;
import org.springframework.mock.web.MockHttpServletResponse;
@@ -118,35 +115,31 @@ public class Int2312RequestMappingIntegrationTests extends AbstractHttpInboundTe
final RequestAttributes attributes = new ServletRequestAttributes(request);
RequestContextHolder.setRequestAttributes(attributes);
this.toLowerCaseChannel.subscribe(new MessageHandler() {
this.toLowerCaseChannel.subscribe(message -> {
MessageHeaders headers = message.getHeaders();
@Override
public void handleMessage(Message<?> message) throws MessagingException {
MessageHeaders headers = message.getHeaders();
assertEquals(attributes, headers.get("requestAttributes"));
assertEquals(attributes, headers.get("requestAttributes"));
Object requestParams = headers.get("requestParams");
assertNotNull(requestParams);
assertEquals(params, ((MultiValueMap<String, String>) requestParams).toSingleValueMap());
Object requestParams = headers.get("requestParams");
assertNotNull(requestParams);
assertEquals(params, ((MultiValueMap<String, String>) requestParams).toSingleValueMap());
Object matrixVariables = headers.get("matrixVariables");
assertThat(matrixVariables, Matchers.instanceOf(Map.class));
Object value = ((Map<?, ?>) matrixVariables).get("value");
assertThat(value, Matchers.instanceOf(MultiValueMap.class));
assertEquals("1", ((MultiValueMap<String, ?>) value).getFirst("q1"));
assertEquals("2", ((MultiValueMap<String, ?>) value).getFirst("q2"));
Object matrixVariables = headers.get("matrixVariables");
assertThat(matrixVariables, Matchers.instanceOf(Map.class));
Object value = ((Map<?, ?>) matrixVariables).get("value");
assertThat(value, Matchers.instanceOf(MultiValueMap.class));
assertEquals("1", ((MultiValueMap<String, ?>) value).getFirst("q1"));
assertEquals("2", ((MultiValueMap<String, ?>) value).getFirst("q2"));
Object requestHeaders = headers.get("requestHeaders");
assertNotNull(requestParams);
assertEquals(MediaType.TEXT_PLAIN, ((HttpHeaders) requestHeaders).getContentType());
Object requestHeaders = headers.get("requestHeaders");
assertNotNull(requestParams);
assertEquals(MediaType.TEXT_PLAIN, ((HttpHeaders) requestHeaders).getContentType());
Map<String, Cookie> cookies = (Map<String, Cookie>) headers.get("cookies");
assertEquals(1, cookies.size());
Cookie foo = cookies.get("foo");
assertNotNull(foo);
assertEquals(cookie, foo);
}
Map<String, Cookie> cookies = (Map<String, Cookie>) headers.get("cookies");
assertEquals(1, cookies.size());
Cookie foo = cookies.get("foo");
assertNotNull(foo);
assertEquals(cookie, foo);
});
MockHttpServletResponse response = new MockHttpServletResponse();

View File

@@ -62,7 +62,7 @@ public class BeanPropertySqlParameterSourceFactory implements SqlParameterSource
private final Map<String, Object> staticParameters;
private StaticBeanPropertySqlParameterSource(Object input, Map<String, Object> staticParameters) {
StaticBeanPropertySqlParameterSource(Object input, Map<String, Object> staticParameters) {
this.input = new BeanPropertySqlParameterSource(input);
this.staticParameters = staticParameters;
}

View File

@@ -158,7 +158,7 @@ public class ExpressionEvaluatingSqlParameterSourceFactory extends AbstractExpre
private final boolean cache;
private ExpressionEvaluatingSqlParameterSource(Object input, Map<String, ?> staticParameters,
ExpressionEvaluatingSqlParameterSource(Object input, Map<String, ?> staticParameters,
Map<String, Expression[]> parameterExpressions, boolean cache) {
this.input = input;
this.parameterExpressions = parameterExpressions;

View File

@@ -16,10 +16,8 @@
package org.springframework.integration.jdbc;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.Collections;
import java.util.LinkedList;
@@ -33,7 +31,6 @@ import org.springframework.jdbc.core.ColumnMapRowMapper;
import org.springframework.jdbc.core.JdbcOperations;
import org.springframework.jdbc.core.PreparedStatementCallback;
import org.springframework.jdbc.core.PreparedStatementCreator;
import org.springframework.jdbc.core.PreparedStatementSetter;
import org.springframework.jdbc.core.ResultSetExtractor;
import org.springframework.jdbc.core.RowMapperResultSetExtractor;
import org.springframework.jdbc.core.namedparam.EmptySqlParameterSource;
@@ -70,14 +67,8 @@ public class JdbcMessageHandler extends AbstractMessageHandler {
private final NamedParameterJdbcOperations jdbcOperations;
private final PreparedStatementCreator generatedKeysStatementCreator = new PreparedStatementCreator() {
@Override
public PreparedStatement createPreparedStatement(Connection con) throws SQLException {
return con.prepareStatement(JdbcMessageHandler.this.updateSql, Statement.RETURN_GENERATED_KEYS);
}
};
private final PreparedStatementCreator generatedKeysStatementCreator = con ->
con.prepareStatement(JdbcMessageHandler.this.updateSql, Statement.RETURN_GENERATED_KEYS);
private volatile String updateSql;
@@ -175,26 +166,20 @@ public class JdbcMessageHandler extends AbstractMessageHandler {
if (keysGenerated) {
if (this.preparedStatementSetter != null) {
return this.jdbcOperations.getJdbcOperations().execute(this.generatedKeysStatementCreator,
new PreparedStatementCallback<List<Map<String, Object>>>() {
(PreparedStatementCallback<List<Map<String, Object>>>) ps -> {
JdbcMessageHandler.this.preparedStatementSetter.setValues(ps, message);
ps.executeUpdate();
ResultSet keys = ps.getGeneratedKeys();
if (keys != null) {
try {
@Override
public List<Map<String, Object>> doInPreparedStatement(PreparedStatement ps)
throws SQLException {
JdbcMessageHandler.this.preparedStatementSetter.setValues(ps, message);
ps.executeUpdate();
ResultSet keys = ps.getGeneratedKeys();
if (keys != null) {
try {
return JdbcMessageHandler.this.generatedKeysResultSetExtractor.extractData(keys);
}
finally {
JdbcUtils.closeResultSet(keys);
}
return JdbcMessageHandler.this.generatedKeysResultSetExtractor.extractData(keys);
}
finally {
JdbcUtils.closeResultSet(keys);
}
return new LinkedList<Map<String, Object>>();
}
return new LinkedList<Map<String, Object>>();
});
}
else {
@@ -207,14 +192,7 @@ public class JdbcMessageHandler extends AbstractMessageHandler {
int updated;
if (this.preparedStatementSetter != null) {
updated = this.jdbcOperations.getJdbcOperations().update(this.updateSql,
new PreparedStatementSetter() {
@Override
public void setValues(PreparedStatement ps) throws SQLException {
JdbcMessageHandler.this.preparedStatementSetter.setValues(ps, message);
}
});
ps -> JdbcMessageHandler.this.preparedStatementSetter.setValues(ps, message));
}
else {
updated = this.jdbcOperations.update(this.updateSql, updateParameterSource);

View File

@@ -16,7 +16,6 @@
package org.springframework.integration.jdbc;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Timestamp;
@@ -51,8 +50,6 @@ import org.springframework.integration.store.SimpleMessageGroup;
import org.springframework.integration.util.UUIDConverter;
import org.springframework.jdbc.core.JdbcOperations;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.ParameterizedPreparedStatementSetter;
import org.springframework.jdbc.core.PreparedStatementSetter;
import org.springframework.jdbc.core.RowCallbackHandler;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.jdbc.core.SingleColumnRowMapper;
@@ -406,20 +403,15 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
this.jdbcTemplate.batchUpdate(getQuery(Query.CREATE_GROUP_TO_MESSAGE),
Arrays.asList(messages),
100,
new ParameterizedPreparedStatementSetter<Message<?>>() {
@Override
public void setValues(PreparedStatement ps, Message<?> messageToAdd) throws SQLException {
String messageId = getKey(messageToAdd.getHeaders().getId());
if (logger.isDebugEnabled()) {
logger.debug("Inserting message with id key=" + messageId +
" and created date=" + createdDate);
}
ps.setString(1, groupKey);
ps.setString(2, messageId);
ps.setString(3, JdbcMessageStore.this.region);
(ps, messageToAdd) -> {
String messageId = getKey(messageToAdd.getHeaders().getId());
if (logger.isDebugEnabled()) {
logger.debug("Inserting message with id key=" + messageId +
" and created date=" + createdDate);
}
ps.setString(1, groupKey);
ps.setString(2, messageId);
ps.setString(3, JdbcMessageStore.this.region);
});
}
@@ -486,27 +478,17 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
this.jdbcTemplate.batchUpdate(getQuery(Query.REMOVE_MESSAGE_FROM_GROUP),
messages,
getRemoveBatchSize(),
new ParameterizedPreparedStatementSetter<Message<?>>() {
@Override
public void setValues(PreparedStatement ps, Message<?> messageToRemove) throws SQLException {
ps.setString(1, groupKey);
ps.setString(2, getKey(messageToRemove.getHeaders().getId()));
ps.setString(3, JdbcMessageStore.this.region);
}
(ps, messageToRemove) -> {
ps.setString(1, groupKey);
ps.setString(2, getKey(messageToRemove.getHeaders().getId()));
ps.setString(3, JdbcMessageStore.this.region);
});
this.jdbcTemplate.batchUpdate(getQuery(Query.DELETE_MESSAGE),
messages,
getRemoveBatchSize(),
new ParameterizedPreparedStatementSetter<Message<?>>() {
@Override
public void setValues(PreparedStatement ps, Message<?> messageToRemove) throws SQLException {
ps.setString(1, getKey(messageToRemove.getHeaders().getId()));
ps.setString(2, JdbcMessageStore.this.region);
}
(ps, messageToRemove) -> {
ps.setString(1, getKey(messageToRemove.getHeaders().getId()));
ps.setString(2, JdbcMessageStore.this.region);
});
this.updateMessageGroup(groupKey);
}
@@ -520,28 +502,20 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
this.removeMessage(messageIds);
}
this.jdbcTemplate.update(getQuery(Query.REMOVE_GROUP_TO_MESSAGE_JOIN), new PreparedStatementSetter() {
@Override
public void setValues(PreparedStatement ps) throws SQLException {
if (logger.isDebugEnabled()) {
logger.debug("Removing relationships for the group with group key=" + groupKey);
}
ps.setString(1, groupKey);
ps.setString(2, JdbcMessageStore.this.region);
this.jdbcTemplate.update(getQuery(Query.REMOVE_GROUP_TO_MESSAGE_JOIN), ps -> {
if (logger.isDebugEnabled()) {
logger.debug("Removing relationships for the group with group key=" + groupKey);
}
ps.setString(1, groupKey);
ps.setString(2, JdbcMessageStore.this.region);
});
this.jdbcTemplate.update(getQuery(Query.DELETE_MESSAGE_GROUP), new PreparedStatementSetter() {
@Override
public void setValues(PreparedStatement ps) throws SQLException {
if (logger.isDebugEnabled()) {
logger.debug("Marking messages with group key=" + groupKey);
}
ps.setString(1, groupKey);
ps.setString(2, JdbcMessageStore.this.region);
this.jdbcTemplate.update(getQuery(Query.DELETE_MESSAGE_GROUP), ps -> {
if (logger.isDebugEnabled()) {
logger.debug("Marking messages with group key=" + groupKey);
}
ps.setString(1, groupKey);
ps.setString(2, JdbcMessageStore.this.region);
});
}
@@ -550,17 +524,13 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
final long updatedDate = System.currentTimeMillis();
final String groupKey = getKey(groupId);
this.jdbcTemplate.update(getQuery(Query.COMPLETE_GROUP), new PreparedStatementSetter() {
@Override
public void setValues(PreparedStatement ps) throws SQLException {
if (logger.isDebugEnabled()) {
logger.debug("Completing MessageGroup: " + groupKey);
}
ps.setTimestamp(1, new Timestamp(updatedDate));
ps.setString(2, groupKey);
ps.setString(3, JdbcMessageStore.this.region);
this.jdbcTemplate.update(getQuery(Query.COMPLETE_GROUP), ps -> {
if (logger.isDebugEnabled()) {
logger.debug("Completing MessageGroup: " + groupKey);
}
ps.setTimestamp(1, new Timestamp(updatedDate));
ps.setString(2, groupKey);
ps.setString(3, JdbcMessageStore.this.region);
});
}
@@ -570,19 +540,15 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
final long updatedDate = System.currentTimeMillis();
final String groupKey = getKey(groupId);
this.jdbcTemplate.update(getQuery(Query.UPDATE_LAST_RELEASED_SEQUENCE), new PreparedStatementSetter() {
@Override
public void setValues(PreparedStatement ps) throws SQLException {
if (logger.isDebugEnabled()) {
logger.debug("Updating the sequence number of the last released Message in the MessageGroup: " +
groupKey);
}
ps.setTimestamp(1, new Timestamp(updatedDate));
ps.setInt(2, sequenceNumber);
ps.setString(3, groupKey);
ps.setString(4, JdbcMessageStore.this.region);
this.jdbcTemplate.update(getQuery(Query.UPDATE_LAST_RELEASED_SEQUENCE), ps -> {
if (logger.isDebugEnabled()) {
logger.debug("Updating the sequence number of the last released Message in the MessageGroup: " +
groupKey);
}
ps.setTimestamp(1, new Timestamp(updatedDate));
ps.setInt(2, sequenceNumber);
ps.setString(3, groupKey);
ps.setString(4, JdbcMessageStore.this.region);
});
this.updateMessageGroup(groupKey);
}
@@ -681,48 +647,36 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
}
private void doCreateMessageGroup(final String groupKey, final Timestamp createdDate) {
this.jdbcTemplate.update(getQuery(Query.CREATE_MESSAGE_GROUP), new PreparedStatementSetter() {
@Override
public void setValues(PreparedStatement ps) throws SQLException {
if (logger.isDebugEnabled()) {
logger.debug("Creating message group with id key=" + groupKey + " and created date=" + createdDate);
}
ps.setString(1, groupKey);
ps.setString(2, JdbcMessageStore.this.region);
ps.setTimestamp(3, createdDate);
ps.setTimestamp(4, createdDate);
this.jdbcTemplate.update(getQuery(Query.CREATE_MESSAGE_GROUP), ps -> {
if (logger.isDebugEnabled()) {
logger.debug("Creating message group with id key=" + groupKey + " and created date=" + createdDate);
}
ps.setString(1, groupKey);
ps.setString(2, JdbcMessageStore.this.region);
ps.setTimestamp(3, createdDate);
ps.setTimestamp(4, createdDate);
});
}
private void doUpdateMessageGroup(final String groupKey, final Timestamp updatedDate) {
this.jdbcTemplate.update(getQuery(Query.UPDATE_MESSAGE_GROUP), new PreparedStatementSetter() {
@Override
public void setValues(PreparedStatement ps) throws SQLException {
if (logger.isDebugEnabled()) {
logger.debug("Updating message group with id key=" + groupKey + " and updated date=" + updatedDate);
}
ps.setTimestamp(1, updatedDate);
ps.setString(2, groupKey);
ps.setString(3, JdbcMessageStore.this.region);
this.jdbcTemplate.update(getQuery(Query.UPDATE_MESSAGE_GROUP), ps -> {
if (logger.isDebugEnabled()) {
logger.debug("Updating message group with id key=" + groupKey + " and updated date=" + updatedDate);
}
ps.setTimestamp(1, updatedDate);
ps.setString(2, groupKey);
ps.setString(3, JdbcMessageStore.this.region);
});
}
private void updateMessageGroup(final String groupId) {
this.jdbcTemplate.update(getQuery(Query.UPDATE_GROUP), new PreparedStatementSetter() {
@Override
public void setValues(PreparedStatement ps) throws SQLException {
if (logger.isDebugEnabled()) {
logger.debug("Updating MessageGroup: " + groupId);
}
ps.setTimestamp(1, new Timestamp(System.currentTimeMillis()));
ps.setString(2, groupId);
ps.setString(3, JdbcMessageStore.this.region);
this.jdbcTemplate.update(getQuery(Query.UPDATE_GROUP), ps -> {
if (logger.isDebugEnabled()) {
logger.debug("Updating MessageGroup: " + groupId);
}
ps.setTimestamp(1, new Timestamp(System.currentTimeMillis()));
ps.setString(2, groupId);
ps.setString(3, JdbcMessageStore.this.region);
});
}
@@ -732,13 +686,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
final List<UUID> messageIds = new ArrayList<UUID>();
this.jdbcTemplate.query(getQuery(Query.LIST_MESSAGEIDS_BY_GROUP_KEY),
new RowCallbackHandler() {
@Override
public void processRow(ResultSet rs) throws SQLException {
messageIds.add(UUID.fromString(rs.getString(1)));
}
}, key, this.region);
(RowCallbackHandler) rs -> messageIds.add(UUID.fromString(rs.getString(1))), key, this.region);
return messageIds;
}

View File

@@ -16,14 +16,11 @@
package org.springframework.integration.jdbc;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import javax.sql.DataSource;
import org.springframework.dao.DataAccessException;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.core.MessageSource;
import org.springframework.jdbc.core.ColumnMapRowMapper;
@@ -145,6 +142,7 @@ public class JdbcPollingChannelAdapter extends IntegrationObjectSupport implemen
* query returns no rows, this method will return <code>null</code>.
* #return the {@link Message} or {@code null} as a result of query.
*/
@Override
public Message<Object> receive() {
Object payload = poll();
if (payload == null) {
@@ -186,16 +184,13 @@ public class JdbcPollingChannelAdapter extends IntegrationObjectSupport implemen
ResultSetExtractor<List<Object>> resultSetExtractor;
if (this.maxRowsPerPoll > 0) {
resultSetExtractor = new ResultSetExtractor<List<Object>>() {
public List<Object> extractData(ResultSet rs) throws SQLException, DataAccessException {
List<Object> results = new ArrayList<Object>(JdbcPollingChannelAdapter.this.maxRowsPerPoll);
int rowNum = 0;
while (rs.next() && rowNum < JdbcPollingChannelAdapter.this.maxRowsPerPoll) {
results.add(rowMapper.mapRow(rs, rowNum++));
}
return results;
resultSetExtractor = rs -> {
List<Object> results = new ArrayList<Object>(JdbcPollingChannelAdapter.this.maxRowsPerPoll);
int rowNum = 0;
while (rs.next() && rowNum < JdbcPollingChannelAdapter.this.maxRowsPerPoll) {
results.add(rowMapper.mapRow(rs, rowNum++));
}
return results;
};
}
else {

View File

@@ -49,7 +49,7 @@ public class JdbcLockRegistry implements ExpirableLockRegistry {
private final Map<String, JdbcLock> locks = new HashMap<String, JdbcLock>();
private LockRepository client;
private final LockRepository client;
public JdbcLockRegistry(LockRepository client) {
this.client = client;
@@ -99,9 +99,9 @@ public class JdbcLockRegistry implements ExpirableLockRegistry {
private volatile long lastUsed = System.currentTimeMillis();
private ReentrantLock delegate = new ReentrantLock();
private final ReentrantLock delegate = new ReentrantLock();
private JdbcLock(LockRepository client, String path) {
JdbcLock(LockRepository client, String path) {
this.mutex = client;
this.path = path;
}

View File

@@ -90,6 +90,10 @@ public class AggregatorIntegrationTests {
public static CountDownLatch latch = new CountDownLatch(1);
RollbackTxSync() {
super();
}
@Override
public void afterCompletion(int status) {
if (TransactionSynchronization.STATUS_ROLLED_BACK == status) {

View File

@@ -175,8 +175,7 @@ public class DelayerHandlerRescheduleIntegrationTests {
@SuppressWarnings("unused")
private static class TestJdbcMessageStore extends JdbcMessageStore {
private TestJdbcMessageStore() {
super();
TestJdbcMessageStore() {
this.setDataSource(dataSource);
}
@@ -197,6 +196,10 @@ public class DelayerHandlerRescheduleIntegrationTests {
public static CountDownLatch latch = new CountDownLatch(2);
RollbackTxSync() {
super();
}
@Override
public void afterCompletion(int status) {
if (TransactionSynchronization.STATUS_ROLLED_BACK == status) {

View File

@@ -19,8 +19,6 @@ package org.springframework.integration.jdbc;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -86,14 +84,9 @@ public class JdbcMessageHandlerIntegrationTests {
public void testInsertWithMessagePreparedStatementSetter() {
JdbcMessageHandler handler = new JdbcMessageHandler(jdbcTemplate, "insert into foos (id, status, name) values (1, 0, ?)");
final AtomicBoolean setterInvoked = new AtomicBoolean();
handler.setPreparedStatementSetter(new MessagePreparedStatementSetter() {
@Override
public void setValues(PreparedStatement ps, Message<?> requestMessage) throws SQLException {
ps.setObject(1, requestMessage.getPayload());
setterInvoked.set(true);
}
handler.setPreparedStatementSetter((ps, requestMessage) -> {
ps.setObject(1, requestMessage.getPayload());
setterInvoked.set(true);
});
handler.afterPropertiesSet();
Message<String> message = new GenericMessage<String>("foo");

View File

@@ -48,9 +48,7 @@ import org.springframework.test.annotation.Repeat;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.DefaultTransactionDefinition;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;
import org.springframework.util.StopWatch;
@@ -107,20 +105,16 @@ public class JdbcMessageStoreChannelIntegrationTests {
// After a rollback in the poller the message is still waiting to be delivered
// but unless we use a transaction here there is a chance that the queue will
// appear empty....
new TransactionTemplate(transactionManager).execute(new TransactionCallback<Void>() {
new TransactionTemplate(transactionManager).execute(status -> {
@Override
public Void doInTransaction(TransactionStatus status) {
synchronized (storeLock) {
synchronized (storeLock) {
assertEquals(1, input.getQueueSize());
assertNotNull(input.receive(100L));
}
return null;
assertEquals(1, input.getQueueSize());
assertNotNull(input.receive(100L));
}
return null;
});
}
@@ -128,28 +122,24 @@ public class JdbcMessageStoreChannelIntegrationTests {
@Repeat(2)
public void testTransactionalSendAndReceive() throws Exception {
boolean result = new TransactionTemplate(transactionManager).execute(new TransactionCallback<Boolean>() {
boolean result = new TransactionTemplate(transactionManager).execute(status -> {
@Override
public Boolean doInTransaction(TransactionStatus status) {
synchronized (storeLock) {
boolean result = input.send(new GenericMessage<String>("foo"), 500L);
// This will time out because the transaction has not committed yet
try {
Service.await(3000);
fail("Expected timeout");
}
catch (Exception e) {
// expected
}
return result;
synchronized (storeLock) {
boolean result1 = input.send(new GenericMessage<String>("foo"), 500L);
// This will time out because the transaction has not committed yet
try {
Service.await(3000);
fail("Expected timeout");
}
catch (Exception e) {
// expected
}
return result1;
}
});
assertTrue("Could not send message", result);
@@ -194,36 +184,32 @@ public class JdbcMessageStoreChannelIntegrationTests {
transactionDefinition.setTimeout(200);
boolean result = new TransactionTemplate(transactionManager, transactionDefinition)
.execute(new TransactionCallback<Boolean>() {
.execute(status -> {
@Override
public Boolean doInTransaction(TransactionStatus status) {
synchronized (storeLock) {
boolean result = input.send(new GenericMessage<String>("foo"), 500L);
// This will time out because the transaction has not committed yet
try {
Service.await(1000);
fail("Expected timeout");
}
catch (Exception e) {
// expected
}
try {
stopWatch.start();
assertNotNull(input.receive(100L));
}
finally {
stopWatch.stop();
}
return result;
synchronized (storeLock) {
boolean result1 = input.send(new GenericMessage<String>("foo"), 500L);
// This will time out because the transaction has not committed yet
try {
Service.await(1000);
fail("Expected timeout");
}
catch (Exception e) {
// expected
}
try {
stopWatch.start();
assertNotNull(input.receive(100L));
}
finally {
stopWatch.stop();
}
return result1;
}
});
assertTrue("Could not send message", result);

View File

@@ -39,8 +39,6 @@ import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;
import org.springframework.util.StopWatch;
@@ -81,37 +79,33 @@ public class JdbcMessageStoreChannelOnePollerIntegrationTests {
assertNull(relay.receive(100L));
final StopWatch stopWatch = new StopWatch();
boolean result = new TransactionTemplate(transactionManager).execute(new TransactionCallback<Boolean>() {
boolean result = new TransactionTemplate(transactionManager).execute(status -> {
@Override
public Boolean doInTransaction(TransactionStatus status) {
synchronized (storeLock) {
boolean result = relay.send(new GenericMessage<String>("foo"), 500L);
// This will time out because the transaction has not committed yet
try {
Service.await(1000);
fail("Expected timeout");
}
catch (Exception e) {
// expected
}
try {
stopWatch.start();
// It hasn't arrive yet because we are still in the sending transaction
assertNull(durable.receive(100L));
}
finally {
stopWatch.stop();
}
return result;
synchronized (storeLock) {
boolean result1 = relay.send(new GenericMessage<String>("foo"), 500L);
// This will time out because the transaction has not committed yet
try {
Service.await(1000);
fail("Expected timeout");
}
catch (Exception e) {
// expected
}
try {
stopWatch.start();
// It hasn't arrive yet because we are still in the sending transaction
assertNull(durable.receive(100L));
}
finally {
stopWatch.stop();
}
return result1;
}
});
assertTrue("Could not send message", result);
@@ -130,24 +124,19 @@ public class JdbcMessageStoreChannelOnePollerIntegrationTests {
*
* With the storeLock: It doesn't deadlock as long as the lock is injected into the poller as well.
*/
new TransactionTemplate(transactionManager).execute(new TransactionCallback<Void>() {
@Override
public Void doInTransaction(TransactionStatus status) {
synchronized (storeLock) {
try {
stopWatch.start();
durable.receive(100L);
return null;
}
finally {
stopWatch.stop();
}
new TransactionTemplate(transactionManager).execute(status -> {
synchronized (storeLock) {
try {
stopWatch.start();
durable.receive(100L);
return null;
}
finally {
stopWatch.stop();
}
}
}
});
// If the poll blocks in the RDBMS there is no way for the queue to respect the timeout

View File

@@ -18,8 +18,6 @@ package org.springframework.integration.jdbc;
import static org.junit.Assert.assertEquals;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.List;
import org.junit.After;
@@ -28,6 +26,7 @@ import org.junit.Assert;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.RowMapper;
@@ -128,26 +127,16 @@ public class JdbcMessageStoreRegionTests {
messageStore1.addMessageToGroup("group1", MessageBuilder.withPayload("payload1").build());
List<String> regions = jdbcTemplate.query("Select * from INT_MESSAGE_GROUP where REGION = 'region1'", new RowMapper<String>() {
public String mapRow(ResultSet rs, int rowNum) throws SQLException {
return rs.getString("REGION");
}
});
List<String> regions = jdbcTemplate.query("Select * from INT_MESSAGE_GROUP where REGION = 'region1'",
(RowMapper<String>) (rs, rowNum) -> rs.getString("REGION"));
assertEquals(1, regions.size());
assertEquals("region1", regions.get(0));
messageStore2.addMessageToGroup("group1", MessageBuilder.withPayload("payload1").build());
List<String> regions2 = jdbcTemplate.query("Select * from INT_MESSAGE_GROUP where REGION = 'region2'", new RowMapper<String>() {
public String mapRow(ResultSet rs, int rowNum) throws SQLException {
return rs.getString("REGION");
}
});
List<String> regions2 = jdbcTemplate.query("Select * from INT_MESSAGE_GROUP where REGION = 'region2'",
(RowMapper<String>) (rs, rowNum) -> rs.getString("REGION"));
assertEquals(1, regions2.size());
assertEquals("region2", regions2.get(0));

View File

@@ -25,12 +25,7 @@ import static org.junit.Assert.assertTrue;
import static org.springframework.integration.test.matcher.PayloadAndHeaderMatcher.sameExceptIgnorableHeaders;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.List;
@@ -46,14 +41,10 @@ import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.serializer.Deserializer;
import org.springframework.core.serializer.Serializer;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.history.MessageHistory;
import org.springframework.integration.store.MessageGroup;
import org.springframework.integration.store.MessageGroupStore;
import org.springframework.integration.store.MessageGroupStore.MessageGroupCallback;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.util.UUIDConverter;
import org.springframework.jdbc.core.JdbcTemplate;
@@ -136,19 +127,13 @@ public class JdbcMessageStoreTests {
@Test
public void testSerializer() throws Exception {
// N.B. these serializers are not realistic (just for test purposes)
messageStore.setSerializer(new Serializer<Message<?>>() {
@Override
public void serialize(Message<?> object, OutputStream outputStream) throws IOException {
outputStream.write(((Message<?>) object).getPayload().toString().getBytes());
outputStream.flush();
}
messageStore.setSerializer((object, outputStream) -> {
outputStream.write(((Message<?>) object).getPayload().toString().getBytes());
outputStream.flush();
});
messageStore.setDeserializer(new Deserializer<GenericMessage<String>>() {
@Override
public GenericMessage<String> deserialize(InputStream inputStream) throws IOException {
BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream));
return new GenericMessage<String>(reader.readLine());
}
messageStore.setDeserializer(inputStream -> {
BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream));
return new GenericMessage<String>(reader.readLine());
});
Message<String> message = MessageBuilder.withPayload("foo").build();
Message<String> saved = messageStore.addMessage(message);
@@ -327,14 +312,9 @@ public class JdbcMessageStoreTests {
Message<String> message = MessageBuilder.withPayload("foo").setCorrelationId(groupId).build();
messageStore.addMessagesToGroup(groupId, message);
final CountDownLatch groupRemovalLatch = new CountDownLatch(1);
messageStore.registerMessageGroupExpiryCallback(new MessageGroupCallback() {
@Override
public void execute(MessageGroupStore messageGroupStore, MessageGroup group) {
messageGroupStore.removeMessageGroup(group.getGroupId());
groupRemovalLatch.countDown();
}
messageStore.registerMessageGroupExpiryCallback((messageGroupStore, group) -> {
messageGroupStore.removeMessageGroup(group.getGroupId());
groupRemovalLatch.countDown();
});
messageStore.expireMessageGroups(2000);
@@ -347,15 +327,10 @@ public class JdbcMessageStoreTests {
template.afterPropertiesSet();
template.update("UPDATE INT_MESSAGE_GROUP set CREATED_DATE=? where GROUP_KEY=? and REGION=?",
new PreparedStatementSetter() {
@Override
public void setValues(PreparedStatement ps) throws SQLException {
ps.setTimestamp(1, new Timestamp(System.currentTimeMillis() - 10000));
ps.setString(2, UUIDConverter.getUUID(groupId).toString());
ps.setString(3, "DEFAULT");
}
(PreparedStatementSetter) ps -> {
ps.setTimestamp(1, new Timestamp(System.currentTimeMillis() - 10000));
ps.setString(2, UUIDConverter.getUUID(groupId).toString());
ps.setString(3, "DEFAULT");
});
messageStore.expireMessageGroups(2000);
@@ -371,14 +346,7 @@ public class JdbcMessageStoreTests {
Message<String> message = MessageBuilder.withPayload("foo").setCorrelationId(groupId).build();
messageStore.setTimeoutOnIdle(true);
messageStore.addMessagesToGroup(groupId, message);
messageStore.registerMessageGroupExpiryCallback(new MessageGroupCallback() {
@Override
public void execute(MessageGroupStore messageGroupStore, MessageGroup group) {
messageGroupStore.removeMessageGroup(group.getGroupId());
}
});
messageStore.registerMessageGroupExpiryCallback((messageGroupStore, group) -> messageGroupStore.removeMessageGroup(group.getGroupId()));
JdbcTemplate template = new JdbcTemplate(this.dataSource);
template.afterPropertiesSet();
@@ -404,15 +372,10 @@ public class JdbcMessageStoreTests {
private void updateMessageGroup(JdbcTemplate template, final String groupId, final long timeout) {
template.update("UPDATE INT_MESSAGE_GROUP set UPDATED_DATE=? where GROUP_KEY=? and REGION=?",
new PreparedStatementSetter() {
@Override
public void setValues(PreparedStatement ps) throws SQLException {
ps.setTimestamp(1, new Timestamp(System.currentTimeMillis() - timeout));
ps.setString(2, UUIDConverter.getUUID(groupId).toString());
ps.setString(3, "DEFAULT");
}
(PreparedStatementSetter) ps -> {
ps.setTimestamp(1, new Timestamp(System.currentTimeMillis() - timeout));
ps.setString(2, UUIDConverter.getUUID(groupId).toString());
ps.setString(3, "DEFAULT");
});
}

View File

@@ -25,11 +25,10 @@ import static org.junit.Assert.assertTrue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.commons.logging.LogFactory;
import org.junit.ClassRule;
import org.junit.Test;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.springframework.integration.store.MessageGroup;
import org.springframework.integration.store.MessageGroupQueue;
@@ -55,19 +54,14 @@ public class MessageGroupQueueTests {
final AtomicReference<InterruptedException> exceptionHolder = new AtomicReference<InterruptedException>();
Thread t = new Thread(new Runnable() {
@Override
public void run() {
queue.offer(new GenericMessage<String>("hello"));
try {
queue.offer(new GenericMessage<String>("hello"), 100, TimeUnit.SECONDS);
}
catch (InterruptedException e) {
exceptionHolder.set(e);
}
Thread t = new Thread(() -> {
queue.offer(new GenericMessage<String>("hello"));
try {
queue.offer(new GenericMessage<String>("hello"), 100, TimeUnit.SECONDS);
}
catch (InterruptedException e) {
exceptionHolder.set(e);
}
});
t.start();
Thread.sleep(1000);
@@ -81,31 +75,21 @@ public class MessageGroupQueueTests {
final MessageGroupQueue queue = new MessageGroupQueue(new SimpleMessageStore(), 1, 1);
final AtomicReference<Message<?>> messageHolder = new AtomicReference<Message<?>>();
Thread t1 = new Thread(new Runnable() {
@Override
public void run() {
try {
messageHolder.set(queue.poll(1000, TimeUnit.SECONDS));
}
catch (Exception e) {
e.printStackTrace();
}
Thread t1 = new Thread(() -> {
try {
messageHolder.set(queue.poll(1000, TimeUnit.SECONDS));
}
catch (Exception e) {
LogFactory.getLog(getClass()).error("queue poll failed", e);
}
});
Thread t2 = new Thread(new Runnable() {
@Override
public void run() {
try {
queue.offer(new GenericMessage<String>("hello"), 1000, TimeUnit.SECONDS);
}
catch (Exception e) {
e.printStackTrace();
}
Thread t2 = new Thread(() -> {
try {
queue.offer(new GenericMessage<String>("hello"), 1000, TimeUnit.SECONDS);
}
catch (Exception e) {
LogFactory.getLog(getClass()).error("queue offer failed", e);
}
});
t1.start();
t2.start();
@@ -120,32 +104,22 @@ public class MessageGroupQueueTests {
queue.offer(new GenericMessage<String>("hello"), 1000, TimeUnit.SECONDS);
Thread t1 = new Thread(new Runnable() {
@Override
public void run() {
try {
queue.offer(new GenericMessage<String>("Hi"), 1000, TimeUnit.SECONDS);
}
catch (Exception e) {
e.printStackTrace();
}
Thread t1 = new Thread(() -> {
try {
queue.offer(new GenericMessage<String>("Hi"), 1000, TimeUnit.SECONDS);
}
catch (Exception e) {
LogFactory.getLog(getClass()).error("queue offer failed", e);
}
});
Thread t2 = new Thread(new Runnable() {
@Override
public void run() {
try {
queue.poll(1000, TimeUnit.SECONDS);
messageHolder.set(queue.poll(1000, TimeUnit.SECONDS));
}
catch (Exception e) {
e.printStackTrace();
}
Thread t2 = new Thread(() -> {
try {
queue.poll(1000, TimeUnit.SECONDS);
messageHolder.set(queue.poll(1000, TimeUnit.SECONDS));
}
catch (Exception e) {
LogFactory.getLog(getClass()).error("queue poll failed", e);
}
});
t1.start();
@@ -162,57 +136,37 @@ public class MessageGroupQueueTests {
final AtomicReference<Message<?>> messageHolder2 = new AtomicReference<Message<?>>();
final AtomicReference<Message<?>> messageHolder3 = new AtomicReference<Message<?>>();
Thread t1 = new Thread(new Runnable() {
@Override
public void run() {
try {
messageHolder1.set(queue.poll(10, TimeUnit.SECONDS));
}
catch (Exception e) {
e.printStackTrace();
}
Thread t1 = new Thread(() -> {
try {
messageHolder1.set(queue.poll(10, TimeUnit.SECONDS));
}
catch (Exception e) {
LogFactory.getLog(getClass()).error("queue poll failed", e);
}
});
Thread t2 = new Thread(new Runnable() {
@Override
public void run() {
try {
messageHolder2.set(queue.poll(10, TimeUnit.SECONDS));
}
catch (Exception e) {
e.printStackTrace();
}
Thread t2 = new Thread(() -> {
try {
messageHolder2.set(queue.poll(10, TimeUnit.SECONDS));
}
catch (Exception e) {
LogFactory.getLog(getClass()).error("queue poll failed", e);
}
});
Thread t3 = new Thread(new Runnable() {
@Override
public void run() {
try {
messageHolder3.set(queue.poll(10, TimeUnit.SECONDS));
}
catch (Exception e) {
e.printStackTrace();
}
Thread t3 = new Thread(() -> {
try {
messageHolder3.set(queue.poll(10, TimeUnit.SECONDS));
}
catch (Exception e) {
LogFactory.getLog(getClass()).error("queue poll failed", e);
}
});
Thread t4 = new Thread(new Runnable() {
@Override
public void run() {
try {
queue.offer(new GenericMessage<String>("Hi"), 10, TimeUnit.SECONDS);
}
catch (Exception e) {
e.printStackTrace();
}
Thread t4 = new Thread(() -> {
try {
queue.offer(new GenericMessage<String>("Hi"), 10, TimeUnit.SECONDS);
}
catch (Exception e) {
LogFactory.getLog(getClass()).error("queue offer failed", e);
}
});
t1.start();
Thread.sleep(1000);
@@ -235,46 +189,31 @@ public class MessageGroupQueueTests {
final AtomicReference<Boolean> booleanHolder2 = new AtomicReference<Boolean>(true);
final AtomicReference<Boolean> booleanHolder3 = new AtomicReference<Boolean>(true);
Thread t1 = new Thread(new Runnable() {
@Override
public void run() {
try {
booleanHolder1.set(queue.offer(new GenericMessage<String>("Hi-1"), 2, TimeUnit.SECONDS));
}
catch (Exception e) {
e.printStackTrace();
}
Thread t1 = new Thread(() -> {
try {
booleanHolder1.set(queue.offer(new GenericMessage<String>("Hi-1"), 2, TimeUnit.SECONDS));
}
catch (Exception e) {
LogFactory.getLog(getClass()).error("queue offer failed", e);
}
});
Thread t2 = new Thread(new Runnable() {
@Override
public void run() {
try {
boolean offered = queue.offer(new GenericMessage<String>("Hi-2"), 2, TimeUnit.SECONDS);
booleanHolder2.set(offered);
}
catch (Exception e) {
e.printStackTrace();
}
Thread t2 = new Thread(() -> {
try {
boolean offered = queue.offer(new GenericMessage<String>("Hi-2"), 2, TimeUnit.SECONDS);
booleanHolder2.set(offered);
}
catch (Exception e) {
LogFactory.getLog(getClass()).error("queue offer failed", e);
}
});
Thread t3 = new Thread(new Runnable() {
@Override
public void run() {
try {
boolean offered = queue.offer(new GenericMessage<String>("Hi-3"), 2, TimeUnit.SECONDS);
booleanHolder3.set(offered);
}
catch (Exception e) {
e.printStackTrace();
}
Thread t3 = new Thread(() -> {
try {
boolean offered = queue.offer(new GenericMessage<String>("Hi-3"), 2, TimeUnit.SECONDS);
booleanHolder3.set(offered);
}
catch (Exception e) {
LogFactory.getLog(getClass()).error("queue offer failed", e);
}
});
t1.start();
Thread.sleep(1000);
@@ -294,36 +233,26 @@ public class MessageGroupQueueTests {
queue.offer(new GenericMessage<String>("hello"), 1000, TimeUnit.SECONDS);
Thread t1 = new Thread(new Runnable() {
@Override
public void run() {
try {
queue.offer(new GenericMessage<String>("Hi"), 1000, TimeUnit.SECONDS);
queue.offer(new GenericMessage<String>("Hi"), 1000, TimeUnit.SECONDS);
queue.offer(new GenericMessage<String>("Hi"), 1000, TimeUnit.SECONDS);
}
catch (Exception e) {
e.printStackTrace();
}
Thread t1 = new Thread(() -> {
try {
queue.offer(new GenericMessage<String>("Hi"), 1000, TimeUnit.SECONDS);
queue.offer(new GenericMessage<String>("Hi"), 1000, TimeUnit.SECONDS);
queue.offer(new GenericMessage<String>("Hi"), 1000, TimeUnit.SECONDS);
}
catch (Exception e) {
LogFactory.getLog(getClass()).error("queue offer failed", e);
}
});
Thread t2 = new Thread(new Runnable() {
@Override
public void run() {
try {
queue.poll(1000, TimeUnit.SECONDS);
messageHolder.set(queue.poll(1000, TimeUnit.SECONDS));
queue.poll(1000, TimeUnit.SECONDS);
queue.poll(1000, TimeUnit.SECONDS);
}
catch (Exception e) {
e.printStackTrace();
}
Thread t2 = new Thread(() -> {
try {
queue.poll(1000, TimeUnit.SECONDS);
messageHolder.set(queue.poll(1000, TimeUnit.SECONDS));
queue.poll(1000, TimeUnit.SECONDS);
queue.poll(1000, TimeUnit.SECONDS);
}
catch (Exception e) {
LogFactory.getLog(getClass()).error("queue poll failed", e);
}
});
t1.start();
@@ -338,14 +267,9 @@ public class MessageGroupQueueTests {
public void validateMgqInterruptionStoreLock() throws Exception {
MessageGroupStore mgs = Mockito.mock(MessageGroupStore.class);
Mockito.doAnswer(new Answer<MessageGroup>() {
@Override
public MessageGroup answer(InvocationOnMock invocation) throws Throwable {
Thread.sleep(5000);
return null;
}
Mockito.doAnswer(invocation -> {
Thread.sleep(5000);
return null;
}).when(mgs).addMessageToGroup(Mockito.any(Integer.class), Mockito.any(Message.class));
MessageGroup mg = Mockito.mock(MessageGroup.class);
@@ -356,29 +280,17 @@ public class MessageGroupQueueTests {
final AtomicReference<InterruptedException> exceptionHolder = new AtomicReference<InterruptedException>();
Thread t1 = new Thread(new Runnable() {
@Override
public void run() {
queue.offer(new GenericMessage<String>("hello"));
}
});
Thread t1 = new Thread(() -> queue.offer(new GenericMessage<String>("hello")));
t1.start();
Thread.sleep(500);
Thread t2 = new Thread(new Runnable() {
@Override
public void run() {
queue.offer(new GenericMessage<String>("hello"));
try {
queue.offer(new GenericMessage<String>("hello"), 100, TimeUnit.SECONDS);
}
catch (InterruptedException e) {
exceptionHolder.set(e);
}
Thread t2 = new Thread(() -> {
queue.offer(new GenericMessage<String>("hello"));
try {
queue.offer(new GenericMessage<String>("hello"), 100, TimeUnit.SECONDS);
}
catch (InterruptedException e) {
exceptionHolder.set(e);
}
});
t2.start();
Thread.sleep(1000);

View File

@@ -43,8 +43,6 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.PollableChannel;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;
/**
@@ -152,15 +150,12 @@ public class JdbcPollingChannelAdapterParserTests {
@Test
public void testMaxRowsInboundChannelAdapter() {
setUp("pollingWithMaxRowsJdbcInboundChannelAdapterTest.xml", getClass());
new TransactionTemplate(transactionManager).execute(new TransactionCallback<Void>() {
@Override
public Void doInTransaction(TransactionStatus status) {
jdbcTemplate.update("insert into item values(1,'',2)");
jdbcTemplate.update("insert into item values(2,'',2)");
jdbcTemplate.update("insert into item values(3,'',2)");
jdbcTemplate.update("insert into item values(4,'',2)");
return null;
}
new TransactionTemplate(transactionManager).execute(status -> {
jdbcTemplate.update("insert into item values(1,'',2)");
jdbcTemplate.update("insert into item values(2,'',2)");
jdbcTemplate.update("insert into item values(3,'',2)");
jdbcTemplate.update("insert into item values(4,'',2)");
return null;
});
int count = 0;
while (count < 4) {

View File

@@ -106,24 +106,20 @@ public class JdbcLockRegistryDifferentClientTests {
final CountDownLatch latch2 = new CountDownLatch(1);
final CountDownLatch latch3 = new CountDownLatch(1);
lock1.lockInterruptibly();
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
Lock lock2 = registry2.obtain("foo");
try {
latch1.countDown();
lock2.lockInterruptibly();
latch2.await(10, TimeUnit.SECONDS);
locked.set(true);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
finally {
lock2.unlock();
latch3.countDown();
}
Executors.newSingleThreadExecutor().execute(() -> {
Lock lock2 = registry2.obtain("foo");
try {
latch1.countDown();
lock2.lockInterruptibly();
latch2.await(10, TimeUnit.SECONDS);
locked.set(true);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
finally {
lock2.unlock();
latch3.countDown();
}
});
assertTrue(latch1.await(10, TimeUnit.SECONDS));
@@ -147,50 +143,42 @@ public class JdbcLockRegistryDifferentClientTests {
final List<String> locked = new ArrayList<String>();
final CountDownLatch latch = new CountDownLatch(2);
ExecutorService pool = Executors.newFixedThreadPool(2);
pool.execute(new Runnable() {
@Override
public void run() {
Lock lock = registry1.obtain("foo");
pool.execute(() -> {
Lock lock = registry1.obtain("foo");
try {
lock.lockInterruptibly();
locked.add("1");
latch.countDown();
}
catch (InterruptedException e1) {
Thread.currentThread().interrupt();
}
finally {
try {
lock.lockInterruptibly();
locked.add("1");
latch.countDown();
lock.unlock();
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
finally {
try {
lock.unlock();
}
catch (Exception e) {
// ignore
}
catch (Exception e2) {
// ignore
}
}
});
pool.execute(new Runnable() {
@Override
public void run() {
Lock lock = registry2.obtain("foo");
pool.execute(() -> {
Lock lock = registry2.obtain("foo");
try {
lock.lockInterruptibly();
locked.add("2");
latch.countDown();
}
catch (InterruptedException e1) {
Thread.currentThread().interrupt();
}
finally {
try {
lock.lockInterruptibly();
locked.add("2");
latch.countDown();
lock.unlock();
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
finally {
try {
lock.unlock();
}
catch (Exception e) {
// ignore
}
catch (Exception e2) {
// ignore
}
}
});
@@ -217,30 +205,26 @@ public class JdbcLockRegistryDifferentClientTests {
final DefaultLockRepository client = new DefaultLockRepository(this.dataSource);
client.afterPropertiesSet();
this.context.getAutowireCapableBeanFactory().autowireBean(client);
Callable<Boolean> task = new Callable<Boolean>() {
@Override
public Boolean call() {
Lock lock = new JdbcLockRegistry(client).obtain("foo");
try {
if (locked.isEmpty() && lock.tryLock()) {
if (locked.isEmpty()) {
locked.add("done");
return true;
}
Callable<Boolean> task = () -> {
Lock lock = new JdbcLockRegistry(client).obtain("foo");
try {
if (locked.isEmpty() && lock.tryLock()) {
if (locked.isEmpty()) {
locked.add("done");
return true;
}
}
finally {
try {
lock.unlock();
}
catch (Exception e) {
// ignore
}
latch.countDown();
}
return false;
}
finally {
try {
lock.unlock();
}
catch (Exception e) {
// ignore
}
latch.countDown();
}
return false;
};
tasks.add(task);
}
@@ -266,29 +250,25 @@ public class JdbcLockRegistryDifferentClientTests {
final BlockingQueue<Integer> data = new LinkedBlockingQueue<Integer>();
final CountDownLatch latch1 = new CountDownLatch(1);
lock1.lockInterruptibly();
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
Lock lock2 = new JdbcLockRegistry(client2).obtain("foo");
try {
latch1.countDown();
StopWatch stopWatch = new StopWatch();
stopWatch.start();
lock2.lockInterruptibly();
stopWatch.stop();
data.add(4);
Thread.sleep(10);
data.add(5);
Thread.sleep(10);
data.add(6);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
finally {
lock2.unlock();
}
Executors.newSingleThreadExecutor().execute(() -> {
Lock lock2 = new JdbcLockRegistry(client2).obtain("foo");
try {
latch1.countDown();
StopWatch stopWatch = new StopWatch();
stopWatch.start();
lock2.lockInterruptibly();
stopWatch.stop();
data.add(4);
Thread.sleep(10);
data.add(5);
Thread.sleep(10);
data.add(6);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
finally {
lock2.unlock();
}
});
assertTrue(latch1.await(10, TimeUnit.SECONDS));

View File

@@ -26,7 +26,6 @@ import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
@@ -154,21 +153,17 @@ public class JdbcLockRegistryTests {
lock1.lockInterruptibly();
final AtomicBoolean locked = new AtomicBoolean();
final CountDownLatch latch = new CountDownLatch(1);
Future<Object> result = Executors.newSingleThreadExecutor().submit(new Callable<Object>() {
@Override
public Object call() throws Exception {
Lock lock2 = JdbcLockRegistryTests.this.registry.obtain("foo");
locked.set(lock2.tryLock(200, TimeUnit.MILLISECONDS));
latch.countDown();
try {
lock2.unlock();
}
catch (Exception e) {
return e;
}
return null;
Future<Object> result = Executors.newSingleThreadExecutor().submit(() -> {
Lock lock2 = JdbcLockRegistryTests.this.registry.obtain("foo");
locked.set(lock2.tryLock(200, TimeUnit.MILLISECONDS));
latch.countDown();
try {
lock2.unlock();
}
catch (Exception e) {
return e;
}
return null;
});
assertTrue(latch.await(10, TimeUnit.SECONDS));
assertFalse(locked.get());
@@ -186,24 +181,20 @@ public class JdbcLockRegistryTests {
final CountDownLatch latch2 = new CountDownLatch(1);
final CountDownLatch latch3 = new CountDownLatch(1);
lock1.lockInterruptibly();
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
Lock lock2 = JdbcLockRegistryTests.this.registry.obtain("foo");
try {
latch1.countDown();
lock2.lockInterruptibly();
latch2.await(10, TimeUnit.SECONDS);
locked.set(true);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
finally {
lock2.unlock();
latch3.countDown();
}
Executors.newSingleThreadExecutor().execute(() -> {
Lock lock2 = JdbcLockRegistryTests.this.registry.obtain("foo");
try {
latch1.countDown();
lock2.lockInterruptibly();
latch2.await(10, TimeUnit.SECONDS);
locked.set(true);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
finally {
lock2.unlock();
latch3.countDown();
}
});
assertTrue(latch1.await(10, TimeUnit.SECONDS));
@@ -226,24 +217,20 @@ public class JdbcLockRegistryTests {
final CountDownLatch latch2 = new CountDownLatch(1);
final CountDownLatch latch3 = new CountDownLatch(1);
lock1.lockInterruptibly();
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
Lock lock2 = registry2.obtain("foo");
try {
latch1.countDown();
lock2.lockInterruptibly();
latch2.await(10, TimeUnit.SECONDS);
locked.set(true);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
finally {
lock2.unlock();
latch3.countDown();
}
Executors.newSingleThreadExecutor().execute(() -> {
Lock lock2 = registry2.obtain("foo");
try {
latch1.countDown();
lock2.lockInterruptibly();
latch2.await(10, TimeUnit.SECONDS);
locked.set(true);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
finally {
lock2.unlock();
latch3.countDown();
}
});
assertTrue(latch1.await(10, TimeUnit.SECONDS));
@@ -262,19 +249,15 @@ public class JdbcLockRegistryTests {
lock.lockInterruptibly();
final AtomicBoolean locked = new AtomicBoolean();
final CountDownLatch latch = new CountDownLatch(1);
Future<Object> result = Executors.newSingleThreadExecutor().submit(new Callable<Object>() {
@Override
public Object call() throws Exception {
try {
lock.unlock();
}
catch (Exception e) {
latch.countDown();
return e;
}
return null;
Future<Object> result = Executors.newSingleThreadExecutor().submit(() -> {
try {
lock.unlock();
}
catch (Exception e) {
latch.countDown();
return e;
}
return null;
});
assertTrue(latch.await(10, TimeUnit.SECONDS));
assertFalse(locked.get());

View File

@@ -48,8 +48,6 @@ import org.springframework.test.annotation.DirtiesContext.ClassMode;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;
/**
@@ -94,31 +92,20 @@ public class MySqlJdbcMessageStoreMultipleChannelTests {
@After
public void afterTest() {
new TransactionTemplate(this.transactionManager).execute(new TransactionCallback<Void>() {
@Override
public Void doInTransaction(TransactionStatus status) {
final int deletedGroupToMessageRows = jdbcTemplate.update("delete from INT_GROUP_TO_MESSAGE");
final int deletedMessages = jdbcTemplate.update("delete from INT_MESSAGE");
final int deletedMessageGroups = jdbcTemplate.update("delete from INT_MESSAGE_GROUP");
LOG.info(String.format("Cleaning Database - Deleted Messages: %s, " +
"Deleted GroupToMessage Rows: %s, Deleted Message Groups: %s",
deletedMessages, deletedGroupToMessageRows, deletedMessageGroups));
return null;
}
new TransactionTemplate(this.transactionManager).execute(status -> {
this.jdbcTemplate.update("delete from INT_GROUP_TO_MESSAGE");
this.jdbcTemplate.update("delete from INT_MESSAGE");
this.jdbcTemplate.update("delete from INT_MESSAGE_GROUP");
return null;
});
}
@Test
public void testSendAndActivateTransactionalSend() throws Exception {
new TransactionTemplate(this.transactionManager).execute(new TransactionCallback<Void>() {
@Override
public Void doInTransaction(TransactionStatus status) {
requestChannel.send(MessageBuilder.withPayload("Hello ").build());
return null;
}
new TransactionTemplate(this.transactionManager).execute(status -> {
requestChannel.send(MessageBuilder.withPayload("Hello ").build());
return null;
});
assertTrue("countDownLatch1 was " + countDownLatch1.getCount(), countDownLatch1.await(10000, TimeUnit.MILLISECONDS));

View File

@@ -26,10 +26,7 @@ import static org.junit.Assert.assertTrue;
import static org.springframework.integration.test.matcher.PayloadAndHeaderMatcher.sameExceptIgnorableHeaders;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.util.Properties;
import java.util.UUID;
@@ -44,20 +41,16 @@ import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.serializer.Deserializer;
import org.springframework.core.serializer.Serializer;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.history.MessageHistory;
import org.springframework.integration.jdbc.JdbcMessageStore;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.integration.store.MessageGroup;
import org.springframework.integration.store.MessageGroupStore;
import org.springframework.integration.store.MessageGroupStore.MessageGroupCallback;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.util.UUIDConverter;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.annotation.DirtiesContext.ClassMode;
import org.springframework.test.annotation.Repeat;
@@ -65,9 +58,7 @@ import org.springframework.test.annotation.Rollback;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;
/**
@@ -112,18 +103,15 @@ public class MySqlJdbcMessageStoreTests {
@After
public void afterTest() {
final JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource);
new TransactionTemplate(this.transactionManager).execute(new TransactionCallback<Void>() {
new TransactionTemplate(this.transactionManager).execute(status -> {
final int deletedGroupToMessageRows = jdbcTemplate.update("delete from INT_GROUP_TO_MESSAGE");
final int deletedMessages = jdbcTemplate.update("delete from INT_MESSAGE");
final int deletedMessageGroups = jdbcTemplate.update("delete from INT_MESSAGE_GROUP");
public Void doInTransaction(TransactionStatus status) {
final int deletedGroupToMessageRows = jdbcTemplate.update("delete from INT_GROUP_TO_MESSAGE");
final int deletedMessages = jdbcTemplate.update("delete from INT_MESSAGE");
final int deletedMessageGroups = jdbcTemplate.update("delete from INT_MESSAGE_GROUP");
LOG.info(String.format("Cleaning Database - Deleted Messages: %s, " +
"Deleted GroupToMessage Rows: %s, Deleted Message Groups: %s",
deletedMessages, deletedGroupToMessageRows, deletedMessageGroups));
return null;
}
LOG.info(String.format("Cleaning Database - Deleted Messages: %s, " +
"Deleted GroupToMessage Rows: %s, Deleted Message Groups: %s",
deletedMessages, deletedGroupToMessageRows, deletedMessageGroups));
return null;
});
}
@@ -178,19 +166,13 @@ public class MySqlJdbcMessageStoreTests {
@Transactional
public void testSerializer() throws Exception {
// N.B. these serializers are not realistic (just for test purposes)
messageStore.setSerializer(new Serializer<Message<?>>() {
public void serialize(Message<?> object, OutputStream outputStream) throws IOException {
outputStream.write(((Message<?>) object).getPayload().toString().getBytes());
outputStream.flush();
}
messageStore.setSerializer((object, outputStream) -> {
outputStream.write(((Message<?>) object).getPayload().toString().getBytes());
outputStream.flush();
});
messageStore.setDeserializer(new Deserializer<GenericMessage<String>>() {
public GenericMessage<String> deserialize(InputStream inputStream) throws IOException {
BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream));
return new GenericMessage<String>(reader.readLine());
}
messageStore.setDeserializer(inputStream -> {
BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream));
return new GenericMessage<String>(reader.readLine());
});
Message<String> message = MessageBuilder.withPayload("foo").build();
Message<String> saved = messageStore.addMessage(message);
@@ -363,12 +345,8 @@ public class MySqlJdbcMessageStoreTests {
String groupId = "X";
Message<String> message = MessageBuilder.withPayload("foo").setCorrelationId(groupId).build();
messageStore.addMessageToGroup(groupId, message);
messageStore.registerMessageGroupExpiryCallback(new MessageGroupCallback() {
public void execute(MessageGroupStore messageGroupStore, MessageGroup group) {
messageGroupStore.removeMessageGroup(group.getGroupId());
}
});
messageStore.registerMessageGroupExpiryCallback(
(messageGroupStore, group) -> messageGroupStore.removeMessageGroup(group.getGroupId()));
Thread.sleep(1000);
messageStore.expireMessageGroups(2000);
MessageGroup group = messageStore.getMessageGroup(groupId);
@@ -387,12 +365,8 @@ public class MySqlJdbcMessageStoreTests {
Message<String> message = MessageBuilder.withPayload("foo").setCorrelationId(groupId).build();
messageStore.setTimeoutOnIdle(true);
messageStore.addMessageToGroup(groupId, message);
messageStore.registerMessageGroupExpiryCallback(new MessageGroupCallback() {
public void execute(MessageGroupStore messageGroupStore, MessageGroup group) {
messageGroupStore.removeMessageGroup(group.getGroupId());
}
});
messageStore.registerMessageGroupExpiryCallback(
(messageGroupStore, group) -> messageGroupStore.removeMessageGroup(group.getGroupId()));
Thread.sleep(1000);
messageStore.expireMessageGroups(2000);
MessageGroup group = messageStore.getMessageGroup(groupId);
@@ -473,8 +447,8 @@ public class MySqlJdbcMessageStoreTests {
LOG.info("messageFromGroup1: " + messageFromGroup1.getHeaders().getId() + "; Sequence #: " + new IntegrationMessageHeaderAccessor(messageFromGroup1).getSequenceNumber());
LOG.info("messageFromGroup2: " + messageFromGroup2.getHeaders().getId() + "; Sequence #: " + new IntegrationMessageHeaderAccessor(messageFromGroup2).getSequenceNumber());
assertEquals(Integer.valueOf(1), (Integer) messageFromGroup1.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER));
assertEquals(Integer.valueOf(2), (Integer) messageFromGroup2.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER));
assertEquals(Integer.valueOf(1), messageFromGroup1.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER));
assertEquals(Integer.valueOf(2), messageFromGroup2.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER));
}
@@ -517,8 +491,8 @@ public class MySqlJdbcMessageStoreTests {
LOG.info("messageFromRegion1: " + messageFromRegion1.getHeaders().getId() + "; Sequence #: " + new IntegrationMessageHeaderAccessor(messageFromRegion1).getSequenceNumber());
LOG.info("messageFromRegion2: " + messageFromRegion2.getHeaders().getId() + "; Sequence #: " + new IntegrationMessageHeaderAccessor(messageFromRegion2).getSequenceNumber());
assertEquals(Integer.valueOf(1), (Integer) messageFromRegion1.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER));
assertEquals(Integer.valueOf(2), (Integer) messageFromRegion2.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER));
assertEquals(Integer.valueOf(1), messageFromRegion1.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER));
assertEquals(Integer.valueOf(2), messageFromRegion2.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER));
}

View File

@@ -23,7 +23,6 @@ import static org.junit.Assert.assertTrue;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
@@ -59,7 +58,6 @@ import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.annotation.Isolation;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionCallbackWithoutResult;
import org.springframework.transaction.support.TransactionTemplate;
@@ -153,40 +151,34 @@ public abstract class AbstractTxTimeoutMessageStoreTests {
final TransactionTemplate transactionTemplate = new TransactionTemplate(transactionManager);
for (int i = 0; i < concurrency; i++) {
completionService.submit(new Callable<Boolean>() {
@Override
public Boolean call() throws Exception {
for (int i = 0; i < 100; i++) {
boolean result = transactionTemplate.execute(new TransactionCallback<Boolean>() {
@Override
public Boolean doInTransaction(TransactionStatus status) {
Message<?> message = null;
try {
message = jdbcChannelMessageStore.pollMessageFromGroup(groupId);
}
catch (Exception e) {
log.error("IdCache race condition.", e);
return false;
}
try {
Thread.sleep(10);
}
catch (InterruptedException e) {
log.error(e);
}
if (message != null) {
jdbcChannelMessageStore.removeFromIdCache(message.getHeaders().getId().toString());
}
return true;
}
});
if (!result) {
completionService.submit(() -> {
for (int i1 = 0; i1 < 100; i1++) {
boolean result = transactionTemplate.execute(status -> {
Message<?> message = null;
try {
message = jdbcChannelMessageStore.pollMessageFromGroup(groupId);
}
catch (Exception e1) {
log.error("IdCache race condition.", e1);
return false;
}
try {
Thread.sleep(10);
}
catch (InterruptedException e2) {
log.error(e2);
}
if (message != null) {
jdbcChannelMessageStore.removeFromIdCache(message.getHeaders().getId().toString());
}
return true;
});
if (!result) {
return false;
}
return true;
}
return true;
});
}

View File

@@ -485,7 +485,7 @@ public class ChannelPublishingJmsMessageListener
private final boolean isTopic;
private DestinationNameHolder(String name, boolean isTopic) {
DestinationNameHolder(String name, boolean isTopic) {
this.name = name;
this.isTopic = isTopic;
}
@@ -494,6 +494,10 @@ public class ChannelPublishingJmsMessageListener
private class GatewayDelegate extends MessagingGatewaySupport {
GatewayDelegate() {
super();
}
@Override
public MessageChannel getErrorChannel() {
return super.getErrorChannel();

View File

@@ -1172,14 +1172,8 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
new SettableListenableFuture<AbstractIntegrationMessageBuilder<?>>();
this.futures.put(correlationId, future);
if (this.receiveTimeout > 0) {
getTaskScheduler().schedule(new Runnable() {
@Override
public void run() {
expire(correlationId);
}
}, new Date(System.currentTimeMillis() + this.receiveTimeout));
getTaskScheduler().schedule((Runnable) () -> expire(correlationId),
new Date(System.currentTimeMillis() + this.receiveTimeout));
}
return future;
}
@@ -1339,6 +1333,10 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
private volatile Destination replyDestination;
GatewayReplyListenerContainer() {
super();
}
@Override
protected Destination resolveDestinationName(Session session, String destinationName) throws JMSException {
if (!StringUtils.hasText(destinationName)) {
@@ -1429,7 +1427,7 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
private final javax.jms.Message reply;
private TimedReply(javax.jms.Message reply) {
TimedReply(javax.jms.Message reply) {
this.reply = reply;
}
@@ -1444,6 +1442,10 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
private class LateReplyReaper implements Runnable {
LateReplyReaper() {
super();
}
@Override
public void run() {
if (logger.isTraceEnabled()) {
@@ -1472,6 +1474,10 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
private class IdleContainerStopper implements Runnable {
IdleContainerStopper() {
super();
}
@Override
public void run() {
synchronized (JmsOutboundGateway.this.lifeCycleMonitor) {
@@ -1518,6 +1524,10 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
private volatile Executor taskExecutor;
ReplyContainerProperties() {
super();
}
public String getSessionAcknowledgeModeName() {
return this.sessionAcknowledgeModeName;
}

View File

@@ -157,7 +157,7 @@ public class JmsSendingMessageHandler extends AbstractMessageHandler {
private final JmsHeaderMapper headerMapper;
private HeaderMappingMessagePostProcessor(Message<?> integrationMessage, JmsHeaderMapper headerMapper) {
HeaderMappingMessagePostProcessor(Message<?> integrationMessage, JmsHeaderMapper headerMapper) {
this.integrationMessage = integrationMessage;
this.headerMapper = headerMapper;
}

View File

@@ -183,7 +183,7 @@ public class SubscribableJmsChannel extends AbstractJmsChannel implements Subscr
private final MessageBuilderFactory messageBuilderFactory;
private DispatchingMessageListener(JmsTemplate jmsTemplate,
DispatchingMessageListener(JmsTemplate jmsTemplate,
MessageDispatcher dispatcher, SubscribableJmsChannel channel, boolean isPubSub,
MessageBuilderFactory messageBuilderFactory) {
this.jmsTemplate = jmsTemplate;

View File

@@ -97,13 +97,10 @@ public class ChannelPublishingJmsMessageListenerTests {
}
private void startBackgroundReplier(final PollableChannel channel) {
new SimpleAsyncTaskExecutor().execute(new Runnable() {
@Override
public void run() {
Message<?> request = channel.receive(50000);
Message<?> reply = new GenericMessage<String>(((String) request.getPayload()).toUpperCase());
((MessageChannel) request.getHeaders().getReplyChannel()).send(reply, 5000);
}
new SimpleAsyncTaskExecutor().execute(() -> {
Message<?> request = channel.receive(50000);
Message<?> reply = new GenericMessage<String>(((String) request.getPayload()).toUpperCase());
((MessageChannel) request.getHeaders().getReplyChannel()).send(reply, 5000);
});
}

View File

@@ -29,12 +29,9 @@ import java.util.Map;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.junit.Test;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.support.MessageBuilder;
@@ -536,14 +533,8 @@ public class DefaultJmsHeaderMapperTests {
Session session = Mockito.mock(Session.class);
Mockito.doAnswer(new Answer<TextMessage>() {
@Override
public TextMessage answer(InvocationOnMock invocation) throws Throwable {
return new StubTextMessage((String) invocation.getArguments()[0]);
}
}).when(session).createTextMessage(Mockito.anyString());
Mockito.doAnswer(invocation -> new StubTextMessage((String) invocation.getArguments()[0])).when(session)
.createTextMessage(Mockito.anyString());
javax.jms.Message request = converter.toMessage(new Foo(), session);

View File

@@ -34,7 +34,6 @@ import java.util.concurrent.atomic.AtomicInteger;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.Session;
@@ -45,8 +44,6 @@ import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.Test;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
@@ -61,7 +58,6 @@ import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.util.ErrorHandler;
import org.springframework.util.ObjectUtils;
/**
@@ -91,6 +87,7 @@ public class JmsOutboundGatewayTests extends LogAdjustingTestSupport {
TestUtils.getPropertyValue(gateway, "replyContainer.beanName"));
}
@SuppressWarnings("serial")
@Test
public void testReplyContainerRecovery() throws Exception {
JmsOutboundGateway gateway = new JmsOutboundGateway();
@@ -101,34 +98,23 @@ public class JmsOutboundGatewayTests extends LogAdjustingTestSupport {
ReplyContainerProperties replyContainerProperties = new ReplyContainerProperties();
final List<Throwable> errors = new ArrayList<Throwable>();
ErrorHandlingTaskExecutor errorHandlingTaskExecutor =
new ErrorHandlingTaskExecutor(Executors.newFixedThreadPool(10), new ErrorHandler() {
@Override
public void handleError(Throwable t) {
logger.info("Error:", t);
errors.add(t);
throw new RuntimeException(t);
}
new ErrorHandlingTaskExecutor(Executors.newFixedThreadPool(10), t -> {
errors.add(t);
throw new RuntimeException(t);
});
replyContainerProperties.setTaskExecutor(errorHandlingTaskExecutor);
replyContainerProperties.setRecoveryInterval(100L);
gateway.setReplyContainerProperties(replyContainerProperties);
final Connection connection = mock(Connection.class);
final AtomicInteger connectionAttempts = new AtomicInteger();
doAnswer(new Answer<Connection>() {
doAnswer(invocation -> {
int theCount = connectionAttempts.incrementAndGet();
if (theCount > 1 && theCount < 4) {
throw new JmsException("bar") {
@SuppressWarnings("serial")
@Override
public Connection answer(InvocationOnMock invocation) throws Throwable {
int theCount = connectionAttempts.incrementAndGet();
if (theCount > 1 && theCount < 4) {
throw new JmsException("bar") {
};
}
return connection;
};
}
return connection;
}).when(connectionFactory).createConnection();
Session session = mock(Session.class);
when(connection.createSession(false, 1)).thenReturn(session);
@@ -137,23 +123,18 @@ public class JmsOutboundGatewayTests extends LogAdjustingTestSupport {
when(session.createTemporaryQueue()).thenReturn(mock(TemporaryQueue.class));
final Message message = mock(Message.class);
final AtomicInteger count = new AtomicInteger();
doAnswer(new Answer<Message>() {
doAnswer(invocation -> {
int theCount = count.incrementAndGet();
if (theCount > 1 && theCount < 4) {
throw new JmsException("foo") {
@SuppressWarnings("serial")
@Override
public Message answer(InvocationOnMock invocation) throws Throwable {
int theCount = count.incrementAndGet();
if (theCount > 1 && theCount < 4) {
throw new JmsException("foo") {
};
}
if (theCount > 4) {
Thread.sleep(100);
return null;
}
return message;
};
}
if (theCount > 4) {
Thread.sleep(100);
return null;
}
return message;
}).when(consumer).receive(anyLong());
when(message.getJMSCorrelationID()).thenReturn("foo");
DefaultListableBeanFactory beanFactory = new DefaultListableBeanFactory();
@@ -192,13 +173,7 @@ public class JmsOutboundGatewayTests extends LogAdjustingTestSupport {
gateway.setReceiveTimeout(60000);
gateway.afterPropertiesSet();
gateway.start();
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
gateway.handleMessage(new GenericMessage<String>("foo"));
}
});
Executors.newSingleThreadExecutor().execute(() -> gateway.handleMessage(new GenericMessage<String>("foo")));
CachingConnectionFactory connectionFactory2 = new CachingConnectionFactory(
new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false"));
JmsTemplate template = new JmsTemplate(connectionFactory2);
@@ -207,15 +182,10 @@ public class JmsOutboundGatewayTests extends LogAdjustingTestSupport {
final Message request = template.receive(requestQ);
assertNotNull(request);
connectionFactory1.resetConnection();
MessageCreator reply = new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
TextMessage reply = session.createTextMessage("bar");
reply.setJMSCorrelationID(request.getJMSMessageID());
logger.debug("Sent reply: " + reply);
return reply;
}
MessageCreator reply = session -> {
TextMessage reply1 = session.createTextMessage("bar");
reply1.setJMSCorrelationID(request.getJMSMessageID());
return reply1;
};
template.send(replyQ, reply);
org.springframework.messaging.Message<?> received = queueChannel.receive(20000);
@@ -243,13 +213,7 @@ public class JmsOutboundGatewayTests extends LogAdjustingTestSupport {
gateway.setCorrelationKey("JMSCorrelationID");
gateway.afterPropertiesSet();
gateway.start();
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
gateway.handleMessage(new GenericMessage<String>("foo"));
}
});
Executors.newSingleThreadExecutor().execute(() -> gateway.handleMessage(new GenericMessage<String>("foo")));
CachingConnectionFactory connectionFactory2 = new CachingConnectionFactory(
new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false"));
JmsTemplate template = new JmsTemplate(connectionFactory2);
@@ -258,14 +222,10 @@ public class JmsOutboundGatewayTests extends LogAdjustingTestSupport {
final Message request = template.receive(requestQ);
assertNotNull(request);
connectionFactory1.resetConnection();
MessageCreator reply = new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
TextMessage reply = session.createTextMessage("bar");
reply.setJMSCorrelationID(request.getJMSCorrelationID());
return reply;
}
MessageCreator reply = session -> {
TextMessage reply1 = session.createTextMessage("bar");
reply1.setJMSCorrelationID(request.getJMSCorrelationID());
return reply1;
};
logger.debug("Sending reply to: " + replyQ);
template.send(replyQ, reply);

View File

@@ -28,21 +28,19 @@ import java.util.concurrent.atomic.AtomicReference;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.command.ActiveMQQueue;
import org.junit.Ignore;
import org.junit.Test;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.jms.connection.CachingConnectionFactory;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
@@ -53,9 +51,9 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
*/
public class OutboundGatewayConnectionTests {
private Destination requestQueue1 = new ActiveMQQueue("request1");
private final Destination requestQueue1 = new ActiveMQQueue("request1");
private Destination replyQueue1 = new ActiveMQQueue("reply1");
private final Destination replyQueue1 = new ActiveMQQueue("reply1");
@Test @Ignore // need a more reliable stop/start for AMQ
public void testContainerWithDestBrokenConnection() throws Exception {
@@ -82,15 +80,13 @@ public class OutboundGatewayConnectionTests {
final AtomicReference<Object> reply = new AtomicReference<Object>();
final CountDownLatch latch1 = new CountDownLatch(1);
final CountDownLatch latch2 = new CountDownLatch(1);
Executors.newSingleThreadExecutor().execute(new Runnable() {
public void run() {
latch1.countDown();
try {
reply.set(gateway.handleRequestMessage(new GenericMessage<String>("foo")));
}
finally {
latch2.countDown();
}
Executors.newSingleThreadExecutor().execute(() -> {
latch1.countDown();
try {
reply.set(gateway.handleRequestMessage(new GenericMessage<String>("foo")));
}
finally {
latch2.countDown();
}
});
assertTrue(latch1.await(10, TimeUnit.SECONDS));
@@ -100,12 +96,7 @@ public class OutboundGatewayConnectionTests {
javax.jms.Message request = template.receive(requestQueue1);
assertNotNull(request);
final javax.jms.Message jmsReply = request;
template.send(request.getJMSReplyTo(), new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
return jmsReply;
}
});
template.send(request.getJMSReplyTo(), (MessageCreator) session -> jmsReply);
assertTrue(latch2.await(10, TimeUnit.SECONDS));
assertNotNull(reply.get());
@@ -116,15 +107,13 @@ public class OutboundGatewayConnectionTests {
final CountDownLatch latch3 = new CountDownLatch(1);
final CountDownLatch latch4 = new CountDownLatch(1);
Executors.newSingleThreadExecutor().execute(new Runnable() {
public void run() {
latch3.countDown();
try {
reply.set(gateway.handleRequestMessage(new GenericMessage<String>("foo")));
}
finally {
latch4.countDown();
}
Executors.newSingleThreadExecutor().execute(() -> {
latch3.countDown();
try {
reply.set(gateway.handleRequestMessage(new GenericMessage<String>("foo")));
}
finally {
latch4.countDown();
}
});
assertTrue(latch3.await(10, TimeUnit.SECONDS));
@@ -134,12 +123,7 @@ public class OutboundGatewayConnectionTests {
request = template.receive(requestQueue1);
assertNotNull(request);
final javax.jms.Message jmsReply2 = request;
template.send(request.getJMSReplyTo(), new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
return jmsReply2;
}
});
template.send(request.getJMSReplyTo(), (MessageCreator) session -> jmsReply2);
assertTrue(latch4.await(10, TimeUnit.SECONDS));
assertNotNull(reply.get());

View File

@@ -30,8 +30,6 @@ import java.util.concurrent.atomic.AtomicReference;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQQueue;
@@ -99,16 +97,13 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport {
final AtomicReference<Object> reply = new AtomicReference<Object>();
final CountDownLatch latch1 = new CountDownLatch(1);
final CountDownLatch latch2 = new CountDownLatch(1);
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
latch1.countDown();
try {
reply.set(gateway.handleRequestMessage(new GenericMessage<String>("foo")));
}
finally {
latch2.countDown();
}
Executors.newSingleThreadExecutor().execute(() -> {
latch1.countDown();
try {
reply.set(gateway.handleRequestMessage(new GenericMessage<String>("foo")));
}
finally {
latch2.countDown();
}
});
assertTrue(latch1.await(10, TimeUnit.SECONDS));
@@ -118,13 +113,7 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport {
javax.jms.Message request = template.receive(requestQueue1);
assertNotNull(request);
final javax.jms.Message jmsReply = request;
template.send(request.getJMSReplyTo(), new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
return jmsReply;
}
});
template.send(request.getJMSReplyTo(), (MessageCreator) session -> jmsReply);
assertTrue(latch2.await(10, TimeUnit.SECONDS));
assertNotNull(reply.get());
@@ -151,16 +140,13 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport {
final AtomicReference<Object> reply = new AtomicReference<Object>();
final CountDownLatch latch1 = new CountDownLatch(1);
final CountDownLatch latch2 = new CountDownLatch(1);
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
latch1.countDown();
try {
reply.set(gateway.handleRequestMessage(new GenericMessage<String>("foo")));
}
finally {
latch2.countDown();
}
Executors.newSingleThreadExecutor().execute(() -> {
latch1.countDown();
try {
reply.set(gateway.handleRequestMessage(new GenericMessage<String>("foo")));
}
finally {
latch2.countDown();
}
});
assertTrue(latch1.await(10, TimeUnit.SECONDS));
@@ -170,13 +156,9 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport {
javax.jms.Message request = template.receive(requestQueue2);
assertNotNull(request);
final javax.jms.Message jmsReply = request;
template.send(request.getJMSReplyTo(), new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
jmsReply.setJMSCorrelationID(jmsReply.getJMSMessageID());
return jmsReply;
}
template.send(request.getJMSReplyTo(), (MessageCreator) session -> {
jmsReply.setJMSCorrelationID(jmsReply.getJMSMessageID());
return jmsReply;
});
assertTrue(latch2.await(20, TimeUnit.SECONDS));
assertNotNull(reply.get());
@@ -205,16 +187,13 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport {
final AtomicReference<Object> reply = new AtomicReference<Object>();
final CountDownLatch latch1 = new CountDownLatch(1);
final CountDownLatch latch2 = new CountDownLatch(1);
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
latch1.countDown();
try {
reply.set(gateway.handleRequestMessage(new GenericMessage<String>("foo")));
}
finally {
latch2.countDown();
}
Executors.newSingleThreadExecutor().execute(() -> {
latch1.countDown();
try {
reply.set(gateway.handleRequestMessage(new GenericMessage<String>("foo")));
}
finally {
latch2.countDown();
}
});
assertTrue(latch1.await(10, TimeUnit.SECONDS));
@@ -224,13 +203,7 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport {
javax.jms.Message request = template.receive(requestQueue3);
assertNotNull(request);
final javax.jms.Message jmsReply = request;
template.send(request.getJMSReplyTo(), new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
return jmsReply;
}
});
template.send(request.getJMSReplyTo(), (MessageCreator) session -> jmsReply);
assertTrue(latch2.await(10, TimeUnit.SECONDS));
assertNotNull(reply.get());
@@ -257,16 +230,13 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport {
final AtomicReference<Object> reply = new AtomicReference<Object>();
final CountDownLatch latch1 = new CountDownLatch(1);
final CountDownLatch latch2 = new CountDownLatch(1);
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
latch1.countDown();
try {
reply.set(gateway.handleRequestMessage(new GenericMessage<String>("foo")));
}
finally {
latch2.countDown();
}
Executors.newSingleThreadExecutor().execute(() -> {
latch1.countDown();
try {
reply.set(gateway.handleRequestMessage(new GenericMessage<String>("foo")));
}
finally {
latch2.countDown();
}
});
assertTrue(latch1.await(10, TimeUnit.SECONDS));
@@ -276,13 +246,9 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport {
javax.jms.Message request = template.receive(requestQueue4);
assertNotNull(request);
final javax.jms.Message jmsReply = request;
template.send(request.getJMSReplyTo(), new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
jmsReply.setJMSCorrelationID(jmsReply.getJMSMessageID());
return jmsReply;
}
template.send(request.getJMSReplyTo(), (MessageCreator) session -> {
jmsReply.setJMSCorrelationID(jmsReply.getJMSMessageID());
return jmsReply;
});
assertTrue(latch2.await(10, TimeUnit.SECONDS));
assertNotNull(reply.get());
@@ -311,16 +277,13 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport {
final AtomicReference<Object> reply = new AtomicReference<Object>();
final CountDownLatch latch1 = new CountDownLatch(1);
final CountDownLatch latch2 = new CountDownLatch(1);
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
latch1.countDown();
try {
reply.set(gateway.handleRequestMessage(new GenericMessage<String>("foo")));
}
finally {
latch2.countDown();
}
Executors.newSingleThreadExecutor().execute(() -> {
latch1.countDown();
try {
reply.set(gateway.handleRequestMessage(new GenericMessage<String>("foo")));
}
finally {
latch2.countDown();
}
});
assertTrue(latch1.await(10, TimeUnit.SECONDS));
@@ -330,13 +293,7 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport {
javax.jms.Message request = template.receive(requestQueue5);
assertNotNull(request);
final javax.jms.Message jmsReply = request;
template.send(request.getJMSReplyTo(), new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
return jmsReply;
}
});
template.send(request.getJMSReplyTo(), (MessageCreator) session -> jmsReply);
assertTrue(latch2.await(10, TimeUnit.SECONDS));
assertNotNull(reply.get());
@@ -363,16 +320,13 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport {
final AtomicReference<Object> reply = new AtomicReference<Object>();
final CountDownLatch latch1 = new CountDownLatch(1);
final CountDownLatch latch2 = new CountDownLatch(1);
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
latch1.countDown();
try {
reply.set(gateway.handleRequestMessage(new GenericMessage<String>("foo")));
}
finally {
latch2.countDown();
}
Executors.newSingleThreadExecutor().execute(() -> {
latch1.countDown();
try {
reply.set(gateway.handleRequestMessage(new GenericMessage<String>("foo")));
}
finally {
latch2.countDown();
}
});
assertTrue(latch1.await(10, TimeUnit.SECONDS));
@@ -382,13 +336,9 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport {
javax.jms.Message request = template.receive(requestQueue6);
assertNotNull(request);
final javax.jms.Message jmsReply = request;
template.send(request.getJMSReplyTo(), new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
jmsReply.setJMSCorrelationID(jmsReply.getJMSMessageID());
return jmsReply;
}
template.send(request.getJMSReplyTo(), (MessageCreator) session -> {
jmsReply.setJMSCorrelationID(jmsReply.getJMSMessageID());
return jmsReply;
});
assertTrue(latch2.await(10, TimeUnit.SECONDS));
assertNotNull(reply.get());
@@ -417,33 +367,12 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport {
gateway.setReceiveTimeout(20000);
gateway.afterPropertiesSet();
gateway.start();
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
JmsTemplate template = new JmsTemplate();
template.setConnectionFactory(getConnectionFactory());
template.setReceiveTimeout(20000);
receiveAndSend(template);
receiveAndSend(template);
}
private void receiveAndSend(JmsTemplate template) {
javax.jms.Message request = template.receive(requestQueue7);
final javax.jms.Message jmsReply = request;
try {
template.send(request.getJMSReplyTo(), new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
return jmsReply;
}
});
}
catch (JmsException e) {
}
catch (JMSException e) {
}
}
Executors.newSingleThreadExecutor().execute(() -> {
JmsTemplate template = new JmsTemplate();
template.setConnectionFactory(getConnectionFactory());
template.setReceiveTimeout(20000);
receiveAndSend(template);
receiveAndSend(template);
});
assertNotNull(gateway.handleRequestMessage(new GenericMessage<String>("foo")));
@@ -462,6 +391,16 @@ public class OutboundGatewayFunctionTests extends LogAdjustingTestSupport {
scheduler.destroy();
}
private void receiveAndSend(JmsTemplate template) {
javax.jms.Message request = template.receive(requestQueue7);
final javax.jms.Message jmsReply = request;
try {
template.send(request.getJMSReplyTo(), (MessageCreator) session -> jmsReply);
}
catch (JmsException | JMSException e) {
}
}
private ConnectionFactory getConnectionFactory() {
ActiveMQConnectionFactory activeMQConnectionFactory =
new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false");

View File

@@ -31,9 +31,8 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnectionFactory;
@@ -48,7 +47,6 @@ import org.springframework.integration.jms.config.JmsChannelFactoryBean;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.jms.connection.CachingConnectionFactory;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.ChannelInterceptor;
@@ -190,14 +188,9 @@ public class PollableJmsChannelTests {
assertTrue(sent1);
final AtomicReference<javax.jms.Message> message = new AtomicReference<javax.jms.Message>();
final CountDownLatch latch1 = new CountDownLatch(1);
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
message.set(receiver.receive(queue));
latch1.countDown();
}
Executors.newSingleThreadExecutor().execute(() -> {
message.set(receiver.receive(queue));
latch1.countDown();
});
assertTrue(latch1.await(10, TimeUnit.SECONDS));
assertNotNull(message.get());
@@ -208,14 +201,9 @@ public class PollableJmsChannelTests {
final CountDownLatch latch2 = new CountDownLatch(1);
boolean sent2 = channel.send(MessageBuilder.withPayload("foo").setPriority(6).build());
assertTrue(sent2);
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
message.set(receiver.receive(queue));
latch2.countDown();
}
Executors.newSingleThreadExecutor().execute(() -> {
message.set(receiver.receive(queue));
latch2.countDown();
});
assertTrue(latch2.await(10, TimeUnit.SECONDS));
assertNotNull(message.get());
@@ -246,15 +234,10 @@ public class PollableJmsChannelTests {
JmsTemplate jmsTemplate = new JmsTemplate(this.connectionFactory);
jmsTemplate.setDefaultDestinationName("pollableJmsChannelSelectorTestQueue");
jmsTemplate.send(new MessageCreator() {
@Override
public javax.jms.Message createMessage(Session session) throws JMSException {
TextMessage message = session.createTextMessage("bar");
message.setStringProperty("baz", "qux");
return message;
}
jmsTemplate.send(session -> {
TextMessage message = session.createTextMessage("bar");
message.setStringProperty("baz", "qux");
return message;
});
Message<?> result2 = channel.receive(10000);

View File

@@ -45,8 +45,6 @@ import org.apache.commons.logging.Log;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.BeanFactory;
@@ -100,20 +98,14 @@ public class SubscribableJmsChannelTests {
public void queueReference() throws Exception {
final CountDownLatch latch = new CountDownLatch(2);
final List<Message<?>> receivedList1 = Collections.synchronizedList(new ArrayList<Message<?>>());
MessageHandler handler1 = new MessageHandler() {
@Override
public void handleMessage(Message<?> message) {
receivedList1.add(message);
latch.countDown();
}
MessageHandler handler1 = message -> {
receivedList1.add(message);
latch.countDown();
};
final List<Message<?>> receivedList2 = Collections.synchronizedList(new ArrayList<Message<?>>());
MessageHandler handler2 = new MessageHandler() {
@Override
public void handleMessage(Message<?> message) {
receivedList2.add(message);
latch.countDown();
}
MessageHandler handler2 = message -> {
receivedList2.add(message);
latch.countDown();
};
JmsChannelFactoryBean factoryBean = new JmsChannelFactoryBean(true);
factoryBean.setConnectionFactory(this.connectionFactory);
@@ -141,20 +133,14 @@ public class SubscribableJmsChannelTests {
public void topicReference() throws Exception {
final CountDownLatch latch = new CountDownLatch(4);
final List<Message<?>> receivedList1 = Collections.synchronizedList(new ArrayList<Message<?>>());
MessageHandler handler1 = new MessageHandler() {
@Override
public void handleMessage(Message<?> message) {
receivedList1.add(message);
latch.countDown();
}
MessageHandler handler1 = message -> {
receivedList1.add(message);
latch.countDown();
};
final List<Message<?>> receivedList2 = Collections.synchronizedList(new ArrayList<Message<?>>());
MessageHandler handler2 = new MessageHandler() {
@Override
public void handleMessage(Message<?> message) {
receivedList2.add(message);
latch.countDown();
}
MessageHandler handler2 = message -> {
receivedList2.add(message);
latch.countDown();
};
JmsChannelFactoryBean factoryBean = new JmsChannelFactoryBean(true);
factoryBean.setConnectionFactory(this.connectionFactory);
@@ -185,22 +171,14 @@ public class SubscribableJmsChannelTests {
public void queueName() throws Exception {
final CountDownLatch latch = new CountDownLatch(2);
final List<Message<?>> receivedList1 = Collections.synchronizedList(new ArrayList<Message<?>>());
MessageHandler handler1 = new MessageHandler() {
@Override
public void handleMessage(Message<?> message) {
receivedList1.add(message);
latch.countDown();
}
MessageHandler handler1 = message -> {
receivedList1.add(message);
latch.countDown();
};
final List<Message<?>> receivedList2 = Collections.synchronizedList(new ArrayList<Message<?>>());
MessageHandler handler2 = new MessageHandler() {
@Override
public void handleMessage(Message<?> message) {
receivedList2.add(message);
latch.countDown();
}
MessageHandler handler2 = message -> {
receivedList2.add(message);
latch.countDown();
};
JmsChannelFactoryBean factoryBean = new JmsChannelFactoryBean(true);
factoryBean.setConnectionFactory(this.connectionFactory);
@@ -233,20 +211,14 @@ public class SubscribableJmsChannelTests {
public void topicName() throws Exception {
final CountDownLatch latch = new CountDownLatch(4);
final List<Message<?>> receivedList1 = Collections.synchronizedList(new ArrayList<Message<?>>());
MessageHandler handler1 = new MessageHandler() {
@Override
public void handleMessage(Message<?> message) {
receivedList1.add(message);
latch.countDown();
}
MessageHandler handler1 = message -> {
receivedList1.add(message);
latch.countDown();
};
final List<Message<?>> receivedList2 = Collections.synchronizedList(new ArrayList<Message<?>>());
MessageHandler handler2 = new MessageHandler() {
@Override
public void handleMessage(Message<?> message) {
receivedList2.add(message);
latch.countDown();
}
MessageHandler handler2 = message -> {
receivedList2.add(message);
latch.countDown();
};
JmsChannelFactoryBean factoryBean = new JmsChannelFactoryBean(true);
@@ -345,16 +317,12 @@ public class SubscribableJmsChannelTests {
channel, "container", AbstractMessageListenerContainer.class);
Log logger = mock(Log.class);
final ArrayList<String> logList = new ArrayList<String>();
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation)
throws Throwable {
String message = (String) invocation.getArguments()[0];
if (message.startsWith("Dispatcher has no subscribers")) {
logList.add(message);
}
return null;
doAnswer(invocation -> {
String message = (String) invocation.getArguments()[0];
if (message.startsWith("Dispatcher has no subscribers")) {
logList.add(message);
}
return null;
}).when(logger).warn(anyString(), any(Exception.class));
when(logger.isWarnEnabled()).thenReturn(true);
Object listener = container.getMessageListener();

View File

@@ -21,9 +21,7 @@ import static org.junit.Assert.assertNotNull;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.junit.Test;
@@ -46,15 +44,11 @@ public class ExceptionHandlingSiConsumerTests {
JmsTemplate jmsTemplate = new JmsTemplate(applicationContext.getBean("jmsConnectionFactory", ConnectionFactory.class));
Destination request = applicationContext.getBean("requestQueueA", Destination.class);
final Destination reply = applicationContext.getBean("replyQueueA", Destination.class);
jmsTemplate.send(request, new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
TextMessage message = session.createTextMessage();
message.setText("echoChannel");
message.setJMSReplyTo(reply);
return message;
}
jmsTemplate.send(request, (MessageCreator) session -> {
TextMessage message = session.createTextMessage();
message.setText("echoChannel");
message.setJMSReplyTo(reply);
return message;
});
Message message = jmsTemplate.receive(reply);
assertNotNull(message);
@@ -68,15 +62,11 @@ public class ExceptionHandlingSiConsumerTests {
JmsTemplate jmsTemplate = new JmsTemplate(applicationContext.getBean("jmsConnectionFactory", ConnectionFactory.class));
Destination request = applicationContext.getBean("requestQueueB", Destination.class);
final Destination reply = applicationContext.getBean("replyQueueB", Destination.class);
jmsTemplate.send(request, new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
TextMessage message = session.createTextMessage();
message.setText("echoWithExceptionChannel");
message.setJMSReplyTo(reply);
return message;
}
jmsTemplate.send(request, (MessageCreator) session -> {
TextMessage message = session.createTextMessage();
message.setText("echoWithExceptionChannel");
message.setJMSReplyTo(reply);
return message;
});
Message message = jmsTemplate.receive(reply);
assertNotNull(message);

View File

@@ -38,7 +38,6 @@ import org.springframework.integration.jms.JmsOutboundGateway;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.messaging.support.GenericMessage;
@@ -103,19 +102,14 @@ public class ExtractRequestReplyPayloadTests {
this.inboundGateway.setExtractRequestPayload(true);
final AtomicBoolean failOnce = new AtomicBoolean();
MessageHandler handler = new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
assertTrue(message.getPayload() instanceof String);
if (failOnce.compareAndSet(false, true)) {
throw new RuntimeException("test tx");
}
MessagingTemplate template = new MessagingTemplate();
template.setDefaultDestination((MessageChannel) message.getHeaders().getReplyChannel());
template.send(message);
MessageHandler handler = message -> {
assertTrue(message.getPayload() instanceof String);
if (failOnce.compareAndSet(false, true)) {
throw new RuntimeException("test tx");
}
MessagingTemplate template = new MessagingTemplate();
template.setDefaultDestination((MessageChannel) message.getHeaders().getReplyChannel());
template.send(message);
};
this.jmsInputChannel.subscribe(handler);
this.outboundChannel.send(new GenericMessage<String>("Hello " + this.testName.getMethodName()));
@@ -239,58 +233,43 @@ public class ExtractRequestReplyPayloadTests {
}
private MessageHandler echoInboundStringHandler() {
return new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
assertTrue(message.getPayload() instanceof String);
MessagingTemplate template = new MessagingTemplate();
template.setDefaultDestination((MessageChannel) message.getHeaders().getReplyChannel());
template.send(message);
}
return message -> {
assertTrue(message.getPayload() instanceof String);
MessagingTemplate template = new MessagingTemplate();
template.setDefaultDestination((MessageChannel) message.getHeaders().getReplyChannel());
template.send(message);
};
}
private MessageHandler unwrapObjectMessageAndEchoHandler() {
return new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
assertThat(message.getPayload(), instanceOf(javax.jms.ObjectMessage.class));
MessagingTemplate template = new MessagingTemplate();
template.setDefaultDestination((MessageChannel) message.getHeaders().getReplyChannel());
Message<?> origMessage = null;
try {
origMessage = (Message<?>) ((javax.jms.ObjectMessage) message.getPayload()).getObject();
}
catch (JMSException e) {
fail("failed to deserialize message");
}
template.send(origMessage);
return message -> {
assertThat(message.getPayload(), instanceOf(javax.jms.ObjectMessage.class));
MessagingTemplate template = new MessagingTemplate();
template.setDefaultDestination((MessageChannel) message.getHeaders().getReplyChannel());
Message<?> origMessage = null;
try {
origMessage = (Message<?>) ((javax.jms.ObjectMessage) message.getPayload()).getObject();
}
catch (JMSException e) {
fail("failed to deserialize message");
}
template.send(origMessage);
};
}
private MessageHandler unwrapTextMessageAndEchoHandler() {
return new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
assertThat(message.getPayload(), instanceOf(javax.jms.TextMessage.class));
MessagingTemplate template = new MessagingTemplate();
template.setDefaultDestination((MessageChannel) message.getHeaders().getReplyChannel());
String payload = null;
try {
payload = ((javax.jms.TextMessage) message.getPayload()).getText();
}
catch (JMSException e) {
fail("failed to deserialize message");
}
template.send(new GenericMessage<String>(payload));
return message -> {
assertThat(message.getPayload(), instanceOf(javax.jms.TextMessage.class));
MessagingTemplate template = new MessagingTemplate();
template.setDefaultDestination((MessageChannel) message.getHeaders().getReplyChannel());
String payload = null;
try {
payload = ((javax.jms.TextMessage) message.getPayload()).getText();
}
catch (JMSException e) {
fail("failed to deserialize message");
}
template.send(new GenericMessage<String>(payload));
};
}

View File

@@ -25,8 +25,6 @@ import java.util.concurrent.TimeUnit;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Session;
import org.junit.Test;
@@ -40,6 +38,7 @@ import org.springframework.util.ErrorHandler;
/**
* @author Mark Fisher
* @author Gary Russell
*/
public class InboundOneWayErrorTests {
@@ -48,12 +47,7 @@ public class InboundOneWayErrorTests {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("InboundOneWayErrorTests-context.xml", getClass());
JmsTemplate jmsTemplate = new JmsTemplate(context.getBean("jmsConnectionFactory", ConnectionFactory.class));
Destination queue = context.getBean("queueA", Destination.class);
jmsTemplate.send(queue, new MessageCreator() {
@Override
public javax.jms.Message createMessage(Session session) throws JMSException {
return session.createTextMessage("test-A");
}
});
jmsTemplate.send(queue, (MessageCreator) session -> session.createTextMessage("test-A"));
TestErrorHandler errorHandler = context.getBean("testErrorHandler", TestErrorHandler.class);
errorHandler.latch.await(3000, TimeUnit.MILLISECONDS);
assertNotNull(errorHandler.lastError);
@@ -69,12 +63,7 @@ public class InboundOneWayErrorTests {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("InboundOneWayErrorTests-context.xml", getClass());
JmsTemplate jmsTemplate = new JmsTemplate(context.getBean("jmsConnectionFactory", ConnectionFactory.class));
Destination queue = context.getBean("queueB", Destination.class);
jmsTemplate.send(queue, new MessageCreator() {
@Override
public javax.jms.Message createMessage(Session session) throws JMSException {
return session.createTextMessage("test-B");
}
});
jmsTemplate.send(queue, (MessageCreator) session -> session.createTextMessage("test-B"));
PollableChannel errorChannel = context.getBean("testErrorChannel", PollableChannel.class);
Message<?> errorMessage = errorChannel.receive(3000);
assertNotNull(errorMessage);

View File

@@ -24,8 +24,6 @@ import static org.mockito.Mockito.verify;
import org.junit.Test;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
@@ -41,11 +39,11 @@ import org.springframework.messaging.support.GenericMessage;
/**
* @author Oleg Zhurakousky
* @author Gunnar Hillert
* @author Gary Russell
*
*/
public class JmsChannelHistoryTests {
@SuppressWarnings("rawtypes")
@Test
public void testMessageHistory() throws Exception {
AbstractMessageListenerContainer mlContainer = mock(AbstractMessageListenerContainer.class);
@@ -55,16 +53,12 @@ public class JmsChannelHistoryTests {
channel.setBeanName("jmsChannel");
Message<String> message = new GenericMessage<String>("hello");
doAnswer(new Answer() {
@Override
doAnswer(invocation -> {
@SuppressWarnings("unchecked")
public Object answer(InvocationOnMock invocation) {
Message<String> msg = (Message<String>) invocation.getArguments()[0];
MessageHistory history = MessageHistory.read(msg);
assertTrue(history.get(0).contains("jmsChannel"));
return null;
}
Message<String> msg = invocation.getArgumentAt(0, Message.class);
MessageHistory history = MessageHistory.read(msg);
assertTrue(history.get(0).contains("jmsChannel"));
return null;
}).when(template).convertAndSend(Mockito.any(Message.class));
channel.send(message);
verify(template, times(1)).convertAndSend(Mockito.any(Message.class));

View File

@@ -26,9 +26,6 @@ import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.lang.reflect.Method;
@@ -40,7 +37,6 @@ import javax.jms.Queue;
import javax.jms.Session;
import org.junit.Test;
import org.mockito.Mockito;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.NotReadablePropertyException;
@@ -62,7 +58,6 @@ import org.springframework.jms.support.converter.MessageConverter;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.messaging.support.GenericMessage;
@@ -236,23 +231,18 @@ public class JmsOutboundGatewayParserTests {
"gatewayMaintainsReplyChannel.xml", this.getClass());
SampleGateway gateway = context.getBean("gateway", SampleGateway.class);
SubscribableChannel jmsInput = context.getBean("jmsInput", SubscribableChannel.class);
MessageHandler handler = new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
MessageHistory history = MessageHistory.read(message);
assertNotNull(history);
Properties componentHistoryRecord = TestUtils.locateComponentInHistory(history, "inboundGateway", 0);
assertNotNull(componentHistoryRecord);
assertEquals("jms:inbound-gateway", componentHistoryRecord.get("type"));
MessagingTemplate messagingTemplate = new MessagingTemplate();
messagingTemplate.setDefaultDestination((MessageChannel) message.getHeaders().getReplyChannel());
messagingTemplate.send(message);
}
MessageHandler handler = message -> {
MessageHistory history = MessageHistory.read(message);
assertNotNull(history);
Properties componentHistoryRecord = TestUtils.locateComponentInHistory(history, "inboundGateway", 0);
assertNotNull(componentHistoryRecord);
assertEquals("jms:inbound-gateway", componentHistoryRecord.get("type"));
MessagingTemplate messagingTemplate = new MessagingTemplate();
messagingTemplate.setDefaultDestination((MessageChannel) message.getHeaders().getReplyChannel());
messagingTemplate.send(message);
};
handler = spy(handler);
jmsInput.subscribe(handler);
String result = gateway.echo("hello");
verify(handler, times(1)).handleMessage(Mockito.any(Message.class));
assertEquals("hello", result);
JmsOutboundGateway gw1 = context.getBean("chain1$child.gateway.handler", JmsOutboundGateway.class);
MessageChannel out = TestUtils.getPropertyValue(gw1, "outputChannel", MessageChannel.class);

View File

@@ -22,9 +22,7 @@ import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertThat;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnectionFactory;
@@ -80,15 +78,10 @@ public class AsyncGatewayTests {
template.setReceiveTimeout(10000);
final Message received = template.receive("asyncTest1");
assertNotNull(received);
template.send(received.getJMSReplyTo(), new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
TextMessage textMessage = session.createTextMessage("bar");
textMessage.setJMSCorrelationID(received.getJMSCorrelationID());
return textMessage;
}
template.send(received.getJMSReplyTo(), (MessageCreator) session -> {
TextMessage textMessage = session.createTextMessage("bar");
textMessage.setJMSCorrelationID(received.getJMSCorrelationID());
return textMessage;
});
org.springframework.messaging.Message<?> reply = replies.receive(10000);
assertNotNull(reply);

View File

@@ -66,18 +66,15 @@ public class MiscellaneousTests {
private void exchange(final CountDownLatch latch, final RequestReplyExchanger gateway, final AtomicInteger replies) {
new Thread(new Runnable() {
@Override
public void run() {
try {
gateway.exchange(new GenericMessage<String>(""));
replies.incrementAndGet();
}
catch (Exception e) {
//ignore
}
latch.countDown();
new Thread(() -> {
try {
gateway.exchange(new GenericMessage<String>(""));
replies.incrementAndGet();
}
catch (Exception e) {
//ignore
}
latch.countDown();
}).start();
}
}

View File

@@ -161,22 +161,19 @@ public class PipelineJmsTests extends ActiveMQMultiContextTests {
try {
for (int i = 0; i < requests; i++) {
final int y = i;
executor.execute(new Runnable() {
@Override
public void run() {
try {
assertEquals(y, gateway.exchange(new GenericMessage<Integer>(y)).getPayload());
successCounter.incrementAndGet();
}
catch (MessageTimeoutException e) {
timeoutCounter.incrementAndGet();
}
catch (Throwable t) {
failureCounter.incrementAndGet();
}
finally {
latch.countDown();
}
executor.execute(() -> {
try {
assertEquals(y, gateway.exchange(new GenericMessage<Integer>(y)).getPayload());
successCounter.incrementAndGet();
}
catch (MessageTimeoutException e) {
timeoutCounter.incrementAndGet();
}
catch (Throwable t) {
failureCounter.incrementAndGet();
}
finally {
latch.countDown();
}
});
}

View File

@@ -174,23 +174,20 @@ public class PipelineNamedReplyQueuesJmsTests extends ActiveMQMultiContextTests
for (int i = 1000000; i < 1000000 + requests * 100000; i += 100000) {
final int y = i;
executor.execute(new Runnable() {
@Override
public void run() {
try {
assertEquals(y + offset, gateway.exchange(new GenericMessage<Integer>(y)).getPayload());
successCounter.incrementAndGet();
}
catch (MessageTimeoutException e) {
timeoutCounter.incrementAndGet();
}
catch (Throwable t) {
t.printStackTrace();
failureCounter.incrementAndGet();
}
finally {
latch.countDown();
}
executor.execute(() -> {
try {
assertEquals(y + offset, gateway.exchange(new GenericMessage<Integer>(y)).getPayload());
successCounter.incrementAndGet();
}
catch (MessageTimeoutException e) {
timeoutCounter.incrementAndGet();
}
catch (Throwable t) {
logger.error("gateway invocation failed", t);
failureCounter.incrementAndGet();
}
finally {
latch.countDown();
}
});
}

View File

@@ -23,10 +23,8 @@ import java.util.concurrent.CountDownLatch;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.junit.Rule;
@@ -71,22 +69,14 @@ public class RequestReplyScenariosWithCachedConsumersTests extends ActiveMQMulti
final Destination requestDestination = context.getBean("siOutQueueOptimizedA", Destination.class);
final Destination replyDestination = context.getBean("siInQueueOptimizedA", Destination.class);
new Thread(new Runnable() {
@Override
public void run() {
final Message requestMessage = jmsTemplate.receive(requestDestination);
jmsTemplate.send(replyDestination, new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
TextMessage message = session.createTextMessage();
message.setText("bar");
message.setJMSCorrelationID(requestMessage.getJMSMessageID());
return message;
}
});
}
new Thread(() -> {
final Message requestMessage = jmsTemplate.receive(requestDestination);
jmsTemplate.send(replyDestination, (MessageCreator) session -> {
TextMessage message = session.createTextMessage();
message.setText("bar");
message.setJMSCorrelationID(requestMessage.getJMSMessageID());
return message;
});
}).start();
gateway.exchange(new GenericMessage<String>("foo"));
}
@@ -109,22 +99,14 @@ public class RequestReplyScenariosWithCachedConsumersTests extends ActiveMQMulti
final Destination requestDestination = context.getBean("siOutQueueNonOptimizedB", Destination.class);
final Destination replyDestination = context.getBean("siInQueueNonOptimizedB", Destination.class);
new Thread(new Runnable() {
@Override
public void run() {
final Message requestMessage = jmsTemplate.receive(requestDestination);
jmsTemplate.send(replyDestination, new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
TextMessage message = session.createTextMessage();
message.setText("bar");
message.setJMSCorrelationID(requestMessage.getJMSMessageID());
return message;
}
});
}
new Thread(() -> {
final Message requestMessage = jmsTemplate.receive(requestDestination);
jmsTemplate.send(replyDestination, (MessageCreator) session -> {
TextMessage message = session.createTextMessage();
message.setText("bar");
message.setJMSCorrelationID(requestMessage.getJMSMessageID());
return message;
});
}).start();
org.springframework.messaging.Message<?> siReplyMessage = gateway.exchange(new GenericMessage<String>("foo"));
assertEquals("bar", siReplyMessage.getPayload());
@@ -146,22 +128,14 @@ public class RequestReplyScenariosWithCachedConsumersTests extends ActiveMQMulti
final Destination requestDestination = context.getBean("siOutQueueOptimizedC", Destination.class);
final Destination replyDestination = context.getBean("siInQueueOptimizedC", Destination.class);
new Thread(new Runnable() {
@Override
public void run() {
final Message requestMessage = jmsTemplate.receive(requestDestination);
jmsTemplate.send(replyDestination, new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
TextMessage message = session.createTextMessage();
message.setText("bar");
message.setJMSCorrelationID(requestMessage.getJMSCorrelationID());
return message;
}
});
}
new Thread(() -> {
final Message requestMessage = jmsTemplate.receive(requestDestination);
jmsTemplate.send(replyDestination, (MessageCreator) session -> {
TextMessage message = session.createTextMessage();
message.setText("bar");
message.setJMSCorrelationID(requestMessage.getJMSCorrelationID());
return message;
});
}).start();
org.springframework.messaging.Message<?> siReplyMessage = gateway.exchange(new GenericMessage<String>("foo"));
assertEquals("bar", siReplyMessage.getPayload());
@@ -183,22 +157,14 @@ public class RequestReplyScenariosWithCachedConsumersTests extends ActiveMQMulti
final Destination requestDestination = context.getBean("siOutQueueNonOptimizedD", Destination.class);
final Destination replyDestination = context.getBean("siInQueueNonOptimizedD", Destination.class);
new Thread(new Runnable() {
@Override
public void run() {
final Message requestMessage = jmsTemplate.receive(requestDestination);
jmsTemplate.send(replyDestination, new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
TextMessage message = session.createTextMessage();
message.setText("bar");
message.setJMSCorrelationID(requestMessage.getJMSCorrelationID());
return message;
}
});
}
new Thread(() -> {
final Message requestMessage = jmsTemplate.receive(requestDestination);
jmsTemplate.send(replyDestination, (MessageCreator) session -> {
TextMessage message = session.createTextMessage();
message.setText("bar");
message.setJMSCorrelationID(requestMessage.getJMSCorrelationID());
return message;
});
}).start();
org.springframework.messaging.Message<?> siReplyMessage = gateway.exchange(new GenericMessage<String>("foo"));
assertEquals("bar", siReplyMessage.getPayload());
@@ -230,34 +196,26 @@ public class RequestReplyScenariosWithCachedConsumersTests extends ActiveMQMulti
}
final CountDownLatch latch = new CountDownLatch(1);
new Thread(new Runnable() {
@Override
public void run() {
DefaultMessageListenerContainer dmlc = new DefaultMessageListenerContainer();
dmlc.setConnectionFactory(connectionFactory);
dmlc.setDestination(requestDestination);
dmlc.setMessageListener(new SessionAwareMessageListener<Message>() {
@Override
public void onMessage(Message message, Session session) {
String requestPayload = (String) extractPayload(message);
try {
TextMessage replyMessage = session.createTextMessage();
replyMessage.setText(requestPayload);
replyMessage.setJMSCorrelationID(message.getJMSCorrelationID());
MessageProducer producer = session.createProducer(replyDestination);
producer.send(replyMessage);
}
catch (Exception e) {
// ignore. the test will fail
}
}
});
dmlc.afterPropertiesSet();
dmlc.start();
latch.countDown();
}
new Thread(() -> {
DefaultMessageListenerContainer dmlc = new DefaultMessageListenerContainer();
dmlc.setConnectionFactory(connectionFactory);
dmlc.setDestination(requestDestination);
dmlc.setMessageListener((SessionAwareMessageListener<Message>) (message, session) -> {
String requestPayload = (String) extractPayload(message);
try {
TextMessage replyMessage = session.createTextMessage();
replyMessage.setText(requestPayload);
replyMessage.setJMSCorrelationID(message.getJMSCorrelationID());
MessageProducer producer = session.createProducer(replyDestination);
producer.send(replyMessage);
}
catch (Exception e) {
// ignore. the test will fail
}
});
dmlc.afterPropertiesSet();
dmlc.start();
latch.countDown();
}).start();
latch.await();

View File

@@ -20,9 +20,7 @@ import static org.junit.Assert.assertEquals;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.junit.Rule;
@@ -58,22 +56,14 @@ public class RequestReplyScenariosWithNonCachedConsumersTests extends ActiveMQMu
final Destination requestDestination = context.getBean("siOutQueueC", Destination.class);
final Destination replyDestination = context.getBean("siInQueueC", Destination.class);
new Thread(new Runnable() {
@Override
public void run() {
final Message requestMessage = jmsTemplate.receive(requestDestination);
jmsTemplate.send(replyDestination, new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
TextMessage message = session.createTextMessage();
message.setText("bar");
message.setJMSCorrelationID(requestMessage.getJMSMessageID());
return message;
}
});
}
new Thread(() -> {
final Message requestMessage = jmsTemplate.receive(requestDestination);
jmsTemplate.send(replyDestination, (MessageCreator) session -> {
TextMessage message = session.createTextMessage();
message.setText("bar");
message.setJMSCorrelationID(requestMessage.getJMSMessageID());
return message;
});
}).start();
org.springframework.messaging.Message<?> siReplyMessage = gateway.exchange(new GenericMessage<String>("foo"));
assertEquals("bar", siReplyMessage.getPayload());
@@ -94,21 +84,14 @@ public class RequestReplyScenariosWithNonCachedConsumersTests extends ActiveMQMu
final Destination requestDestination = context.getBean("siOutQueueD", Destination.class);
final Destination replyDestination = context.getBean("siInQueueD", Destination.class);
new Thread(new Runnable() {
@Override
public void run() {
final Message requestMessage = jmsTemplate.receive(requestDestination);
jmsTemplate.send(replyDestination, new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
TextMessage message = session.createTextMessage();
message.setText("bar");
message.setJMSCorrelationID(requestMessage.getJMSMessageID());
return message;
}
});
}
new Thread(() -> {
final Message requestMessage = jmsTemplate.receive(requestDestination);
jmsTemplate.send(replyDestination, (MessageCreator) session -> {
TextMessage message = session.createTextMessage();
message.setText("bar");
message.setJMSCorrelationID(requestMessage.getJMSMessageID());
return message;
});
}).start();
org.springframework.messaging.Message<?> siReplyMessage = gateway.exchange(new GenericMessage<String>("foo"));
assertEquals("bar", siReplyMessage.getPayload());
@@ -129,22 +112,14 @@ public class RequestReplyScenariosWithNonCachedConsumersTests extends ActiveMQMu
final Destination requestDestination = context.getBean("siOutQueueA", Destination.class);
final Destination replyDestination = context.getBean("siInQueueA", Destination.class);
new Thread(new Runnable() {
@Override
public void run() {
final Message requestMessage = jmsTemplate.receive(requestDestination);
jmsTemplate.send(replyDestination, new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
TextMessage message = session.createTextMessage();
message.setText("bar");
message.setJMSCorrelationID(requestMessage.getJMSCorrelationID());
return message;
}
});
}
new Thread(() -> {
final Message requestMessage = jmsTemplate.receive(requestDestination);
jmsTemplate.send(replyDestination, (MessageCreator) session -> {
TextMessage message = session.createTextMessage();
message.setText("bar");
message.setJMSCorrelationID(requestMessage.getJMSCorrelationID());
return message;
});
}).start();
org.springframework.messaging.Message<?> siReplyMessage = gateway.exchange(new GenericMessage<String>("foo"));
assertEquals("bar", siReplyMessage.getPayload());
@@ -165,22 +140,14 @@ public class RequestReplyScenariosWithNonCachedConsumersTests extends ActiveMQMu
final Destination requestDestination = context.getBean("siOutQueueB", Destination.class);
final Destination replyDestination = context.getBean("siInQueueB", Destination.class);
new Thread(new Runnable() {
@Override
public void run() {
final Message requestMessage = jmsTemplate.receive(requestDestination);
jmsTemplate.send(replyDestination, new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
TextMessage message = session.createTextMessage();
message.setText("bar");
message.setJMSCorrelationID(requestMessage.getJMSCorrelationID());
return message;
}
});
}
new Thread(() -> {
final Message requestMessage = jmsTemplate.receive(requestDestination);
jmsTemplate.send(replyDestination, (MessageCreator) session -> {
TextMessage message = session.createTextMessage();
message.setText("bar");
message.setJMSCorrelationID(requestMessage.getJMSCorrelationID());
return message;
});
}).start();
org.springframework.messaging.Message<?> siReplyMessage = gateway.exchange(new GenericMessage<String>("foo"));
assertEquals("bar", siReplyMessage.getPayload());

View File

@@ -29,10 +29,8 @@ import java.util.concurrent.atomic.AtomicInteger;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.broker.BrokerService;
@@ -80,35 +78,27 @@ public class RequestReplyScenariosWithTempReplyQueuesTests extends ActiveMQMulti
final Destination requestDestination = context.getBean("siOutQueue", Destination.class);
new Thread(new Runnable() {
@Override
public void run() {
final Message requestMessage = jmsTemplate.receive(requestDestination);
Destination replyTo = null;
new Thread(() -> {
final Message requestMessage = jmsTemplate.receive(requestDestination);
Destination replyTo = null;
try {
replyTo = requestMessage.getJMSReplyTo();
}
catch (Exception e) {
fail();
}
jmsTemplate.send(replyTo, (MessageCreator) session -> {
try {
replyTo = requestMessage.getJMSReplyTo();
TextMessage message = session.createTextMessage();
message.setText("bar");
message.setJMSCorrelationID(requestMessage.getJMSMessageID());
return message;
}
catch (Exception e) {
fail();
// ignore
}
jmsTemplate.send(replyTo, new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
try {
TextMessage message = session.createTextMessage();
message.setText("bar");
message.setJMSCorrelationID(requestMessage.getJMSMessageID());
return message;
}
catch (Exception e) {
// ignore
}
return null;
}
});
}
return null;
});
}).start();
gateway.exchange(new GenericMessage<String>("foo"));
context.close();
@@ -127,34 +117,30 @@ public class RequestReplyScenariosWithTempReplyQueuesTests extends ActiveMQMulti
DefaultMessageListenerContainer dmlc = new DefaultMessageListenerContainer();
dmlc.setConnectionFactory(connectionFactory);
dmlc.setDestination(requestDestination);
dmlc.setMessageListener(new SessionAwareMessageListener<Message>() {
@Override
public void onMessage(Message message, Session session) {
Destination replyTo = null;
dmlc.setMessageListener((SessionAwareMessageListener<Message>) (message, session) -> {
Destination replyTo = null;
try {
replyTo = message.getJMSReplyTo();
}
catch (Exception e1) {
fail();
}
String requestPayload = (String) extractPayload(message);
if (requestPayload.equals("foo")) {
try {
replyTo = message.getJMSReplyTo();
}
catch (Exception e) {
fail();
}
String requestPayload = (String) extractPayload(message);
if (requestPayload.equals("foo")) {
try {
Thread.sleep(6000);
}
catch (Exception e) { /*ignore*/ }
}
try {
TextMessage replyMessage = session.createTextMessage();
replyMessage.setText(requestPayload);
replyMessage.setJMSCorrelationID(message.getJMSMessageID());
MessageProducer producer = session.createProducer(replyTo);
producer.send(replyMessage);
}
catch (Exception e) {
// ignore. the test will fail
Thread.sleep(6000);
}
catch (Exception e2) { /*ignore*/ }
}
try {
TextMessage replyMessage = session.createTextMessage();
replyMessage.setText(requestPayload);
replyMessage.setJMSCorrelationID(message.getJMSMessageID());
MessageProducer producer = session.createProducer(replyTo);
producer.send(replyMessage);
}
catch (Exception e3) {
// ignore. the test will fail
}
});
dmlc.afterPropertiesSet();
@@ -231,32 +217,29 @@ public class RequestReplyScenariosWithTempReplyQueuesTests extends ActiveMQMulti
final AtomicInteger missmatches = new AtomicInteger();
for (int i = 0; i < testNumbers; i++) {
final int y = i;
executor.execute(new Runnable() {
@Override
public void run() {
try {
executor.execute(() -> {
try {
String reply = (String) gateway.exchange(new GenericMessage<String>(String.valueOf(y))).getPayload();
if (!String.valueOf(y).equals(reply)) {
missmatches.incrementAndGet();
}
String reply = (String) gateway.exchange(new GenericMessage<String>(String.valueOf(y))).getPayload();
if (!String.valueOf(y).equals(reply)) {
missmatches.incrementAndGet();
}
catch (Exception e) {
if (e instanceof MessageDeliveryException) {
timeouts.incrementAndGet();
}
else {
failures.incrementAndGet();
}
}
catch (Exception e) {
if (e instanceof MessageDeliveryException) {
timeouts.incrementAndGet();
}
else {
failures.incrementAndGet();
}
}
// if (latch.getCount()%100 == 0){
// long count = testNumbers-latch.getCount();
// if (count > 0){
// print(failures, timeouts, missmatches, testNumbers-latch.getCount());
// }
// }
latch.countDown();
}
latch.countDown();
});
}
latch.await();

View File

@@ -82,8 +82,6 @@ import org.springframework.messaging.MessageHandler;
import org.springframework.util.Assert;
import org.springframework.util.PatternMatchUtils;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.ReflectionUtils.FieldCallback;
import org.springframework.util.ReflectionUtils.FieldFilter;
import org.springframework.util.StringValueResolver;
/**
@@ -304,23 +302,13 @@ public class IntegrationMBeanExporter extends MBeanExporter implements Applicati
private MessageHandler handlerInAnonymousWrapper(final Object bean) {
if (bean != null && bean.getClass().isAnonymousClass()) {
final AtomicReference<MessageHandler> wrapped = new AtomicReference<MessageHandler>();
ReflectionUtils.doWithFields(bean.getClass(), new FieldCallback() {
@Override
public void doWith(Field field) throws IllegalArgumentException, IllegalAccessException {
field.setAccessible(true);
Object handler = field.get(bean);
if (handler instanceof MessageHandler) {
wrapped.set((MessageHandler) handler);
}
ReflectionUtils.doWithFields(bean.getClass(), field -> {
field.setAccessible(true);
Object handler = field.get(bean);
if (handler instanceof MessageHandler) {
wrapped.set((MessageHandler) handler);
}
}, new FieldFilter() {
@Override
public boolean matches(Field field) {
return wrapped.get() == null && field.getName().startsWith("val$");
}
});
}, field -> wrapped.get() == null && field.getName().startsWith("val$"));
return wrapped.get();
}
else {

View File

@@ -26,7 +26,6 @@ import java.util.concurrent.atomic.AtomicInteger;
import javax.management.MBeanServer;
import javax.management.Notification;
import javax.management.NotificationFilter;
import javax.management.ObjectName;
import org.junit.After;
@@ -129,12 +128,7 @@ public class NotificationListeningMessageProducerTests {
adapter.setServer(this.server);
adapter.setObjectName(this.objectName);
adapter.setOutputChannel(outputChannel);
adapter.setFilter(new NotificationFilter() {
@Override
public boolean isNotificationEnabled(Notification notification) {
return !notification.getMessage().equals("bad");
}
});
adapter.setFilter(notification -> !notification.getMessage().equals("bad"));
adapter.setBeanFactory(mock(BeanFactory.class));
adapter.afterPropertiesSet();
adapter.start();

View File

@@ -28,10 +28,7 @@ import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
@@ -62,12 +59,7 @@ public class MethodInvokerTests {
// System . err.println(names);
// the router and the error handler...
assertEquals(2, names.size());
underscores.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
assertEquals("foo", message.getPayload());
}
});
underscores.subscribe(message -> assertEquals("foo", message.getPayload()));
echos.send(MessageBuilder.withPayload("foo").setHeader("entity-type", "underscore").build());
}

View File

@@ -174,6 +174,10 @@ public class NotificationPublishingChannelAdapterParserTests {
private static class TestData {
TestData() {
super();
}
}
public static class FooAdvice extends AbstractRequestHandlerAdvice {

View File

@@ -61,7 +61,6 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.stereotype.Component;
@@ -203,21 +202,10 @@ public class IdempotentReceiverIntegrationTests {
@Bean
public IdempotentReceiverInterceptor idempotentReceiverInterceptor() {
return new IdempotentReceiverInterceptor(new MetadataStoreSelector(new MessageProcessor<String>() {
@Override
public String processMessage(Message<?> message) {
return message.getPayload().toString();
}
}, new MessageProcessor<String>() {
@Override
public String processMessage(Message<?> message) {
return message.getPayload().toString().toUpperCase();
}
}, store()));
return new IdempotentReceiverInterceptor(
new MetadataStoreSelector(
message -> message.getPayload().toString(),
message -> message.getPayload().toString().toUpperCase(), store()));
}
@Bean
@@ -234,14 +222,7 @@ public class IdempotentReceiverIntegrationTests {
@org.springframework.integration.annotation.Transformer(inputChannel = "input",
outputChannel = "output", adviceChain = {"fooAdvice", "idempotentReceiverInterceptor"})
public Transformer transformer() {
return new Transformer() {
@Override
public Message<?> transform(Message<?> message) {
return message;
}
};
return message -> message;
}
@Bean
@@ -291,15 +272,10 @@ public class IdempotentReceiverIntegrationTests {
@ServiceActivator(inputChannel = "annotatedBeanMessageHandlerChannel2")
@IdempotentReceiver("idempotentReceiverInterceptor")
public MessageHandler messageHandler2() {
return new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
if (message.getHeaders().containsKey(IntegrationMessageHeaderAccessor.DUPLICATE_MESSAGE)) {
throw new MessageHandlingException(message, "duplicate message has been received");
}
return message -> {
if (message.getHeaders().containsKey(IntegrationMessageHeaderAccessor.DUPLICATE_MESSAGE)) {
throw new MessageHandlingException(message, "duplicate message has been received");
}
};
}

View File

@@ -108,6 +108,10 @@ public class MessageMetricsAdviceTests {
@SuppressWarnings("unused")
boolean invoked = false;
DummyHandler() {
super();
}
@Override
public void handleMessage(Message<?> message) throws MessagingException {
invoked = true;
@@ -119,6 +123,10 @@ public class MessageMetricsAdviceTests {
boolean invoked = false;
DummyInterceptor() {
super();
}
@Override
public Object invoke(MethodInvocation invocation) throws Throwable {
invoked = true;

View File

@@ -30,8 +30,6 @@ import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.annotation.Autowired;
@@ -91,15 +89,10 @@ public class MonitorTests {
TestUtils.getPropertyValue(this.next, "channelMetrics", DefaultMessageChannelMetrics.class);
channelMetrics = Mockito.spy(channelMetrics);
Mockito.doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
Object result = invocation.callRealMethod();
afterSendLatch.countDown();
return result;
}
Mockito.doAnswer(invocation -> {
Object result = invocation.callRealMethod();
afterSendLatch.countDown();
return result;
}).when(channelMetrics).afterSend(Mockito.any(MetricsContext.class), Mockito.eq(Boolean.TRUE));
new DirectFieldAccessor(this.next).setPropertyValue("channelMetrics", channelMetrics);