Lambdas for Remaining Modules JPA -> ZK

Polishing - PR Comments and Closeable Warnings

Eclipse emits bogus warnings with exceptions in lambdas.
Even though the lambda might run on another thread, elipse thinks it could
cause the context to not be closed.

SPR-14854: MessageChannel is now a @FunctionalInterface

* Additional Lambda polishing and some code style fixes
This commit is contained in:
Gary Russell
2016-10-28 12:07:06 -04:00
committed by Artem Bilan
parent 16be9fc47d
commit 67d6cd0c89
83 changed files with 870 additions and 1324 deletions

View File

@@ -96,7 +96,6 @@ public class AmqpInboundGatewayParserTests {
assertEquals(expected, defaultReplyTo);
}
@SuppressWarnings("rawtypes")
@Test
public void verifyUsageWithHeaderMapper() throws Exception {
DirectChannel requestChannel = context.getBean("requestChannel", DirectChannel.class);

View File

@@ -97,7 +97,7 @@ public class AmqpOutboundGatewayParserTests {
.getExpressionString());
}
@SuppressWarnings("rawtypes")
@SuppressWarnings({ "resource" })
@Test
public void withHeaderMapperCustomRequestResponse() {
ConfigurableApplicationContext context = new ClassPathXmlApplicationContext(
@@ -166,7 +166,7 @@ public class AmqpOutboundGatewayParserTests {
context.close();
}
@SuppressWarnings("rawtypes")
@SuppressWarnings({ "resource" })
@Test
public void withHeaderMapperCustomAndStandardResponse() {
ConfigurableApplicationContext context = new ClassPathXmlApplicationContext(
@@ -220,7 +220,7 @@ public class AmqpOutboundGatewayParserTests {
context.close();
}
@SuppressWarnings("rawtypes")
@SuppressWarnings({ "resource" })
@Test
public void withHeaderMapperNothingToMap() {
ConfigurableApplicationContext context = new ClassPathXmlApplicationContext(
@@ -272,6 +272,7 @@ public class AmqpOutboundGatewayParserTests {
context.close();
}
@SuppressWarnings("resource")
@Test //INT-1029
public void amqpOutboundGatewayWithinChain() {
ConfigurableApplicationContext context = new ClassPathXmlApplicationContext(

View File

@@ -118,18 +118,13 @@ public class MessageHandlerChain extends AbstractMessageProducingHandler impleme
if (i < this.handlers.size() - 1) { // not the last handler
Assert.isInstanceOf(MessageProducer.class, handler, "All handlers except for " +
"the last one in the chain must implement the MessageProducer interface.");
final MessageHandler nextHandler = this.handlers.get(i + 1);
final MessageChannel nextChannel = new MessageChannel() {
@Override
public boolean send(Message<?> message, long timeout) {
return this.send(message);
}
@Override
public boolean send(Message<?> message) {
nextHandler.handleMessage(message);
return true;
}
MessageHandler nextHandler = this.handlers.get(i + 1);
MessageChannel nextChannel = (message, timeout) -> {
nextHandler.handleMessage(message);
return true;
};
((MessageProducer) handler).setOutputChannel(nextChannel);
// If this 'handler' is a nested non-last &lt;chain&gt;, it is necessary
@@ -146,7 +141,7 @@ public class MessageHandlerChain extends AbstractMessageProducingHandler impleme
else {
Assert.isNull(getOutputChannel(),
"An output channel was provided, but the final handler in " +
"the chain does not implement the MessageProducer interface.");
"the chain does not implement the MessageProducer interface.");
}
}
}
@@ -239,16 +234,11 @@ public class MessageHandlerChain extends AbstractMessageProducingHandler impleme
}
@Override
public boolean send(Message<?> message) {
public boolean send(Message<?> message, long timeout) {
produceOutput(message, message);
return true;
}
@Override
public boolean send(Message<?> message, long timeout) {
return send(message);
}
}
}

View File

@@ -46,7 +46,6 @@ import org.springframework.integration.store.SimpleMessageStore;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.GenericMessage;
/**
@@ -94,33 +93,24 @@ public class AbstractCorrelatingMessageHandlerTests {
});
final List<Message<?>> outputMessages = new ArrayList<Message<?>>();
handler.setOutputChannel(new MessageChannel() {
handler.setOutputChannel((message, timeout) -> {
/*
* Executes when group 'bar' completes normally
*/
@Override
public boolean send(Message<?> message, long timeout) {
outputMessages.add(message);
// wake reaper
waitReapStartLatch.countDown();
try {
waitForSendLatch.await(10, TimeUnit.SECONDS);
// wait a little longer for reaper to grab groups
Thread.sleep(2000);
// simulate tx commit
groupStore.removeMessageGroup("bar");
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
return true;
outputMessages.add(message);
// wake reaper
waitReapStartLatch.countDown();
try {
waitForSendLatch.await(10, TimeUnit.SECONDS);
// wait a little longer for reaper to grab groups
Thread.sleep(2000);
// simulate tx commit
groupStore.removeMessageGroup("bar");
}
@Override
public boolean send(Message<?> message) {
return this.send(message, 0);
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
return true;
});
handler.setReleaseStrategy(group -> group.size() == 2);
@@ -160,21 +150,12 @@ public class AbstractCorrelatingMessageHandlerTests {
AggregatingMessageHandler handler = new AggregatingMessageHandler(group -> group, groupStore);
final List<Message<?>> outputMessages = new ArrayList<Message<?>>();
handler.setOutputChannel(new MessageChannel() {
handler.setOutputChannel((message, timeout) -> {
/*
* Executes when group 'bar' completes normally
*/
@Override
public boolean send(Message<?> message, long timeout) {
outputMessages.add(message);
return true;
}
@Override
public boolean send(Message<?> message) {
return this.send(message, 0);
}
outputMessages.add(message);
return true;
});
handler.setReleaseStrategy(group -> group.size() == 1);
@@ -196,21 +177,12 @@ public class AbstractCorrelatingMessageHandlerTests {
AggregatingMessageHandler handler = new AggregatingMessageHandler(group -> group, groupStore);
final List<Message<?>> outputMessages = new ArrayList<Message<?>>();
handler.setOutputChannel(new MessageChannel() {
handler.setOutputChannel((message, timeout) -> {
/*
* Executes when group 'bar' completes normally
*/
@Override
public boolean send(Message<?> message, long timeout) {
outputMessages.add(message);
return true;
}
@Override
public boolean send(Message<?> message) {
return this.send(message, 0);
}
outputMessages.add(message);
return true;
});
handler.setReleaseStrategy(group -> group.size() == 1);

View File

@@ -130,6 +130,7 @@ public class EventOutboundChannelAdapterParserTests {
new ClassPathXmlApplicationContext("EventOutboundChannelAdapterParserTestsWithPollable-context.xml",
EventOutboundChannelAdapterParserTests.class);
final CyclicBarrier barrier = new CyclicBarrier(2);
@SuppressWarnings("resource")
ApplicationListener<?> listener = event -> {
Object source = event.getSource();
if (source instanceof Message) {

View File

@@ -399,6 +399,7 @@ public class RemoteFileTemplate<F> implements RemoteFileOperations<F>, Initializ
return this.sessionFactory.getSession();
}
@SuppressWarnings("rawtypes")
@Override
public <T> T execute(SessionCallback<F, T> callback) {
Session<F> session = null;

View File

@@ -37,6 +37,7 @@ import org.springframework.jms.core.MessageCreator;
*/
public class ExceptionHandlingSiConsumerTests {
@SuppressWarnings("resource")
@Test
public void nonSiProducer_siConsumer_sync_withReturn() throws Exception {
ActiveMqTestUtils.prepare();
@@ -55,6 +56,7 @@ public class ExceptionHandlingSiConsumerTests {
applicationContext.close();
}
@SuppressWarnings("resource")
@Test
public void nonSiProducer_siConsumer_sync_withReturnNoException() throws Exception {
ActiveMqTestUtils.prepare();

View File

@@ -67,6 +67,7 @@ public class RequestReplyScenariosWithTempReplyQueuesTests extends ActiveMQMulti
@Rule
public LongRunningIntegrationTest longTests = new LongRunningIntegrationTest();
@SuppressWarnings("resource")
@Test
public void messageCorrelationBasedOnRequestMessageId() throws Exception {
ActiveMqTestUtils.prepare();

View File

@@ -58,7 +58,7 @@ public class BeanPropertyParameterSourceFactory implements ParameterSourceFactor
private final Map<String, Object> staticParameters;
private StaticBeanPropertyParameterSource(Object input, Map<String, Object> staticParameters) {
StaticBeanPropertyParameterSource(Object input, Map<String, Object> staticParameters) {
this.input = new BeanPropertyParameterSource(input);
this.staticParameters = staticParameters;
}

View File

@@ -39,7 +39,6 @@ import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.test.annotation.DirtiesContext;
@@ -108,12 +107,9 @@ public class JpaOutboundGatewayIntegrationTests {
*/
@Test
public void retrieveFromSecondRecordAndMaximumOneRecord() throws Exception {
this.handler = new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
assertEquals(2, ((List<?>) message.getPayload()).size());
assertEquals(1, entityManager.createQuery("from Student").getResultList().size());
}
this.handler = message -> {
assertEquals(2, ((List<?>) message.getPayload()).size());
assertEquals(1, entityManager.createQuery("from Student").getResultList().size());
};
this.responseChannel.subscribe(this.handler);
@@ -126,13 +122,10 @@ public class JpaOutboundGatewayIntegrationTests {
@Test
public void testFindByEntityClass() throws Exception {
this.handler = new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
assertThat(message.getPayload(), Matchers.instanceOf(StudentDomain.class));
StudentDomain student = (StudentDomain) message.getPayload();
assertEquals("First One", student.getFirstName());
}
this.handler = message -> {
assertThat(message.getPayload(), Matchers.instanceOf(StudentDomain.class));
StudentDomain student = (StudentDomain) message.getPayload();
assertEquals("First One", student.getFirstName());
};
this.responseChannel.subscribe(this.handler);
@@ -142,13 +135,10 @@ public class JpaOutboundGatewayIntegrationTests {
@Test
public void testFindByPayloadType() throws Exception {
this.handler = new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
assertThat(message.getPayload(), Matchers.instanceOf(StudentDomain.class));
StudentDomain student = (StudentDomain) message.getPayload();
assertEquals("First Two", student.getFirstName());
}
this.handler = message -> {
assertThat(message.getPayload(), Matchers.instanceOf(StudentDomain.class));
StudentDomain student = (StudentDomain) message.getPayload();
assertEquals("First Two", student.getFirstName());
};
this.responseChannel.subscribe(this.handler);

View File

@@ -597,7 +597,7 @@ public abstract class AbstractMailReceiver extends IntegrationObjectSupport impl
private final Object content;
private IntegrationMimeMessage(MimeMessage source) throws MessagingException {
IntegrationMimeMessage(MimeMessage source) throws MessagingException {
super(source);
this.source = source;
if (AbstractMailReceiver.this.simpleContent) {

View File

@@ -183,30 +183,27 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport implements Be
}
private Runnable createMessageSendingTask(final Object mailMessage) {
Runnable sendingTask = new Runnable() {
@Override
public void run() {
@SuppressWarnings("unchecked")
org.springframework.messaging.Message<?> message =
mailMessage instanceof Message
? ImapIdleChannelAdapter.this.getMessageBuilderFactory().withPayload(mailMessage).build()
: (org.springframework.messaging.Message<Object>) mailMessage;
Runnable sendingTask = () -> {
@SuppressWarnings("unchecked")
org.springframework.messaging.Message<?> message =
mailMessage instanceof Message
? ImapIdleChannelAdapter.this.getMessageBuilderFactory().withPayload(mailMessage).build()
: (org.springframework.messaging.Message<Object>) mailMessage;
if (TransactionSynchronizationManager.isActualTransactionActive()) {
if (ImapIdleChannelAdapter.this.transactionSynchronizationFactory != null) {
TransactionSynchronization synchronization =
ImapIdleChannelAdapter.this.transactionSynchronizationFactory
.create(ImapIdleChannelAdapter.this);
TransactionSynchronizationManager.registerSynchronization(synchronization);
if (synchronization instanceof IntegrationResourceHolderSynchronization) {
IntegrationResourceHolder holder =
((IntegrationResourceHolderSynchronization) synchronization).getResourceHolder();
holder.setMessage(message);
}
if (TransactionSynchronizationManager.isActualTransactionActive()) {
if (ImapIdleChannelAdapter.this.transactionSynchronizationFactory != null) {
TransactionSynchronization synchronization =
ImapIdleChannelAdapter.this.transactionSynchronizationFactory
.create(ImapIdleChannelAdapter.this);
TransactionSynchronizationManager.registerSynchronization(synchronization);
if (synchronization instanceof IntegrationResourceHolderSynchronization) {
IntegrationResourceHolder holder =
((IntegrationResourceHolderSynchronization) synchronization).getResourceHolder();
holder.setMessage(message);
}
}
sendMessage(message);
}
sendMessage(message);
};
// wrap in the TX proxy if necessary
@@ -235,6 +232,11 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport implements Be
private class ReceivingTask implements Runnable {
ReceivingTask() {
super();
}
@Override
public void run() {
try {
@@ -254,6 +256,10 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport implements Be
private class IdleTask implements Runnable {
IdleTask() {
super();
}
@Override
public void run() {
final TaskScheduler scheduler = getTaskScheduler();
@@ -303,6 +309,10 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport implements Be
private volatile boolean delayNextExecution;
ExceptionAwarePeriodicTrigger() {
super();
}
@Override
public Date nextExecutionTime(TriggerContext triggerContext) {
if (this.delayNextExecution) {

View File

@@ -240,6 +240,11 @@ public class ImapMailReceiver extends AbstractMailReceiver {
}
private class IdleCanceler implements Runnable {
IdleCanceler() {
super();
}
@Override
public void run() {
try {
@@ -259,6 +264,10 @@ public class ImapMailReceiver extends AbstractMailReceiver {
*/
private class SimpleMessageCountListener extends MessageCountAdapter {
SimpleMessageCountListener() {
super();
}
@Override
public void messagesAdded(MessageCountEvent event) {
Message[] messages = event.getMessages();
@@ -271,6 +280,10 @@ public class ImapMailReceiver extends AbstractMailReceiver {
private class DefaultSearchTermStrategy implements SearchTermStrategy {
DefaultSearchTermStrategy() {
super();
}
@Override
public SearchTerm generateSearchTerm(Flags supportedFlags, Folder folder) {
SearchTerm searchTerm = null;

View File

@@ -64,13 +64,9 @@ import org.junit.Ignore;
import org.junit.Rule;
import org.junit.Test;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.core.io.ClassPathResource;
@@ -127,17 +123,13 @@ public class ImapMailReceiverTests {
public void testIdleWithServerCustomSearch() throws Exception {
ImapMailReceiver receiver = new ImapMailReceiver("imap://user:pw@localhost:" + imapIdleServer.getPort()
+ "/INBOX");
receiver.setSearchTermStrategy(new SearchTermStrategy() {
@Override
public SearchTerm generateSearchTerm(Flags supportedFlags, Folder folder) {
try {
FromTerm fromTerm = new FromTerm(new InternetAddress("bar@baz"));
return new AndTerm(fromTerm, new FlagTerm(new Flags(Flag.SEEN), false));
}
catch (AddressException e) {
throw new RuntimeException(e);
}
receiver.setSearchTermStrategy((supportedFlags, folder) -> {
try {
FromTerm fromTerm = new FromTerm(new InternetAddress("bar@baz"));
return new AndTerm(fromTerm, new FlagTerm(new Flags(Flag.SEEN), false));
}
catch (AddressException e) {
throw new RuntimeException(e);
}
});
testIdleWithServerGuts(receiver, false);
@@ -268,35 +260,19 @@ public class ImapMailReceiverTests {
final Message[] messages = new Message[] { msg1, msg2 };
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
DirectFieldAccessor accessor = new DirectFieldAccessor(invocation.getMock());
int folderOpenMode = (Integer) accessor.getPropertyValue("folderOpenMode");
if (folderOpenMode != Folder.READ_WRITE) {
throw new IllegalArgumentException("Folder had to be open in READ_WRITE mode");
}
return null;
doAnswer(invocation -> {
DirectFieldAccessor accessor = new DirectFieldAccessor(invocation.getMock());
int folderOpenMode = (Integer) accessor.getPropertyValue("folderOpenMode");
if (folderOpenMode != Folder.READ_WRITE) {
throw new IllegalArgumentException("Folder had to be open in READ_WRITE mode");
}
return null;
}).when(receiver).openFolder();
doAnswer(new Answer<Object>() {
doAnswer(invocation -> messages).when(receiver).searchForNewMessages();
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
return messages;
}
}).when(receiver).searchForNewMessages();
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
return null;
}
}).when(receiver).fetchMessages(messages);
doAnswer(invocation -> null).when(receiver).fetchMessages(messages);
receiver.receive();
return receiver;
}
@@ -348,34 +324,18 @@ public class ImapMailReceiverTests {
Message msg1 = mock(MimeMessage.class);
Message msg2 = mock(MimeMessage.class);
final Message[] messages = new Message[] { msg1, msg2 };
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
DirectFieldAccessor accessor = new DirectFieldAccessor(invocation.getMock());
int folderOpenMode = (Integer) accessor.getPropertyValue("folderOpenMode");
if (folderOpenMode != Folder.READ_WRITE) {
throw new IllegalArgumentException("Folder had to be open in READ_WRITE mode");
}
return null;
doAnswer(invocation -> {
DirectFieldAccessor accessor = new DirectFieldAccessor(invocation.getMock());
int folderOpenMode = (Integer) accessor.getPropertyValue("folderOpenMode");
if (folderOpenMode != Folder.READ_WRITE) {
throw new IllegalArgumentException("Folder had to be open in READ_WRITE mode");
}
return null;
}).when(receiver).openFolder();
doAnswer(new Answer<Object>() {
doAnswer(invocation -> messages).when(receiver).searchForNewMessages();
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
return messages;
}
}).when(receiver).searchForNewMessages();
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
return null;
}
}).when(receiver).fetchMessages(messages);
doAnswer(invocation -> null).when(receiver).fetchMessages(messages);
receiver.receive();
verify(msg1, times(1)).setFlag(Flag.SEEN, true);
verify(msg2, times(1)).setFlag(Flag.SEEN, true);
@@ -400,29 +360,11 @@ public class ImapMailReceiverTests {
Message msg1 = mock(MimeMessage.class);
Message msg2 = mock(MimeMessage.class);
final Message[] messages = new Message[] { msg1, msg2 };
doAnswer(new Answer<Object>() {
doAnswer(invocation -> null).when(receiver).openFolder();
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
return null;
}
}).when(receiver).openFolder();
doAnswer(invocation -> messages).when(receiver).searchForNewMessages();
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
return messages;
}
}).when(receiver).searchForNewMessages();
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
return null;
}
}).when(receiver).fetchMessages(messages);
doAnswer(invocation -> null).when(receiver).fetchMessages(messages);
receiver.afterPropertiesSet();
receiver.receive();
verify(msg1, times(0)).setFlag(Flag.SEEN, true);
@@ -432,7 +374,7 @@ public class ImapMailReceiverTests {
@Test
public void receiveAndDontMarkAsReadButDelete() throws Exception {
AbstractMailReceiver receiver = new ImapMailReceiver();
((ImapMailReceiver) receiver).setShouldDeleteMessages(true);
receiver.setShouldDeleteMessages(true);
((ImapMailReceiver) receiver).setShouldMarkMessagesAsRead(false);
receiver = spy(receiver);
receiver.setBeanFactory(mock(BeanFactory.class));
@@ -447,34 +389,18 @@ public class ImapMailReceiverTests {
Message msg1 = mock(MimeMessage.class);
Message msg2 = mock(MimeMessage.class);
final Message[] messages = new Message[] { msg1, msg2 };
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
DirectFieldAccessor accessor = new DirectFieldAccessor(invocation.getMock());
int folderOpenMode = (Integer) accessor.getPropertyValue("folderOpenMode");
if (folderOpenMode != Folder.READ_WRITE) {
throw new IllegalArgumentException("Folder had to be open in READ_WRITE mode");
}
return null;
doAnswer(invocation -> {
DirectFieldAccessor accessor = new DirectFieldAccessor(invocation.getMock());
int folderOpenMode = (Integer) accessor.getPropertyValue("folderOpenMode");
if (folderOpenMode != Folder.READ_WRITE) {
throw new IllegalArgumentException("Folder had to be open in READ_WRITE mode");
}
return null;
}).when(receiver).openFolder();
doAnswer(new Answer<Object>() {
doAnswer(invocation -> messages).when(receiver).searchForNewMessages();
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
return messages;
}
}).when(receiver).searchForNewMessages();
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
return null;
}
}).when(receiver).fetchMessages(messages);
doAnswer(invocation -> null).when(receiver).fetchMessages(messages);
receiver.afterPropertiesSet();
receiver.receive();
verify(msg1, times(0)).setFlag(Flag.SEEN, true);
@@ -499,40 +425,25 @@ public class ImapMailReceiverTests {
Message msg1 = mock(MimeMessage.class);
Message msg2 = mock(MimeMessage.class);
final Message[] messages = new Message[] { msg1, msg2 };
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
DirectFieldAccessor accessor = new DirectFieldAccessor(invocation.getMock());
int folderOpenMode = (Integer) accessor.getPropertyValue("folderOpenMode");
if (folderOpenMode != Folder.READ_WRITE) {
throw new IllegalArgumentException("Folder had to be open in READ_WRITE mode");
}
return null;
doAnswer(invocation -> {
DirectFieldAccessor accessor = new DirectFieldAccessor(invocation.getMock());
int folderOpenMode = (Integer) accessor.getPropertyValue("folderOpenMode");
if (folderOpenMode != Folder.READ_WRITE) {
throw new IllegalArgumentException("Folder had to be open in READ_WRITE mode");
}
return null;
}).when(receiver).openFolder();
doAnswer(new Answer<Object>() {
doAnswer(invocation -> messages).when(receiver).searchForNewMessages();
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
return messages;
}
}).when(receiver).searchForNewMessages();
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
return null;
}
}).when(receiver).fetchMessages(messages);
doAnswer(invocation -> null).when(receiver).fetchMessages(messages);
receiver.receive();
verify(msg1, times(1)).setFlag(Flag.SEEN, true);
verify(msg2, times(1)).setFlag(Flag.SEEN, true);
verify(receiver, times(0)).deleteMessages((Message[]) Mockito.any());
}
@SuppressWarnings("resource")
@Test
@Ignore
public void testMessageHistory() throws Exception {
@@ -553,33 +464,17 @@ public class ImapMailReceiverTests {
when(mailMessage.getFlags()).thenReturn(flags);
final Message[] messages = new Message[] { mailMessage };
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
DirectFieldAccessor accessor = new DirectFieldAccessor((invocation.getMock()));
IMAPFolder folder = mock(IMAPFolder.class);
accessor.setPropertyValue("folder", folder);
when(folder.hasNewMessages()).thenReturn(true);
return null;
}
doAnswer(invocation -> {
DirectFieldAccessor accessor = new DirectFieldAccessor((invocation.getMock()));
IMAPFolder folder = mock(IMAPFolder.class);
accessor.setPropertyValue("folder", folder);
when(folder.hasNewMessages()).thenReturn(true);
return null;
}).when(receiver).openFolder();
doAnswer(new Answer<Object>() {
doAnswer(invocation -> messages).when(receiver).searchForNewMessages();
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
return messages;
}
}).when(receiver).searchForNewMessages();
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
return null;
}
}).when(receiver).fetchMessages(messages);
doAnswer(invocation -> null).when(receiver).fetchMessages(messages);
PollableChannel channel = context.getBean("channel", PollableChannel.class);
@@ -626,21 +521,9 @@ public class ImapMailReceiverTests {
when(folder.getPermanentFlags()).thenReturn(new Flags(Flags.Flag.USER));
folderField.set(receiver, folder);
doAnswer(new Answer<Object>() {
doAnswer(invocation -> true).when(folder).isOpen();
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
return true;
}
}).when(folder).isOpen();
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
return null;
}
}).when(receiver).openFolder();
doAnswer(invocation -> null).when(receiver).openFolder();
DirectFieldAccessor adapterAccessor = new DirectFieldAccessor(adapter);
adapterAccessor.setPropertyValue("mailReceiver", receiver);
@@ -650,21 +533,9 @@ public class ImapMailReceiverTests {
when(mailMessage.getFlags()).thenReturn(flags);
final Message[] messages = new Message[] { mailMessage };
doAnswer(new Answer<Object>() {
doAnswer(invocation -> messages).when(receiver).searchForNewMessages();
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
return messages;
}
}).when(receiver).searchForNewMessages();
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
return null;
}
}).when(receiver).fetchMessages(messages);
doAnswer(invocation -> null).when(receiver).fetchMessages(messages);
adapter.start();
org.springframework.messaging.Message<?> replMessage = errorChannel.receive(10000);
@@ -674,6 +545,7 @@ public class ImapMailReceiverTests {
context.close();
}
@SuppressWarnings("resource")
@Test
public void testNoInitialIdleDelayWhenRecentNotSupported() throws Exception {
ConfigurableApplicationContext context =
@@ -703,13 +575,7 @@ public class ImapMailReceiverTests {
when(store.getFolder(Mockito.any(URLName.class))).thenReturn(folder);
storeField.set(receiver, store);
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
return folder;
}
}).when(receiver).getFolder();
doAnswer(invocation -> folder).when(receiver).getFolder();
MimeMessage mailMessage = mock(MimeMessage.class);
Flags flags = mock(Flags.class);
@@ -717,41 +583,27 @@ public class ImapMailReceiverTests {
final Message[] messages = new Message[] { mailMessage };
final AtomicInteger shouldFindMessagesCounter = new AtomicInteger(2);
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
/*
* Return the message from first invocation of waitForMessages()
* and in receive(); then return false in the next call to
* waitForMessages() so we enter idle(); counter will be reset
* to 1 in the mocked idle().
*/
if (shouldFindMessagesCounter.decrementAndGet() >= 0) {
return messages;
}
else {
return new Message[0];
}
doAnswer(invocation -> {
/*
* Return the message from first invocation of waitForMessages()
* and in receive(); then return false in the next call to
* waitForMessages() so we enter idle(); counter will be reset
* to 1 in the mocked idle().
*/
if (shouldFindMessagesCounter.decrementAndGet() >= 0) {
return messages;
}
else {
return new Message[0];
}
}).when(receiver).searchForNewMessages();
doAnswer(new Answer<Object>() {
doAnswer(invocation -> null).when(receiver).fetchMessages(messages);
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
return null;
}
}).when(receiver).fetchMessages(messages);
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
Thread.sleep(5000);
shouldFindMessagesCounter.set(1);
return null;
}
doAnswer(invocation -> {
Thread.sleep(5000);
shouldFindMessagesCounter.set(1);
return null;
}).when(folder).idle();
adapter.start();
@@ -768,6 +620,7 @@ public class ImapMailReceiverTests {
context.close();
}
@SuppressWarnings("resource")
@Test
public void testInitialIdleDelayWhenRecentIsSupported() throws Exception {
ConfigurableApplicationContext context =
@@ -797,44 +650,22 @@ public class ImapMailReceiverTests {
when(store.getFolder(Mockito.any(URLName.class))).thenReturn(folder);
storeField.set(receiver, store);
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
return folder;
}
}).when(receiver).getFolder();
doAnswer(invocation -> folder).when(receiver).getFolder();
MimeMessage mailMessage = mock(MimeMessage.class);
Flags flags = mock(Flags.class);
when(mailMessage.getFlags()).thenReturn(flags);
final Message[] messages = new Message[] { mailMessage };
doAnswer(new Answer<Object>() {
doAnswer(invocation -> messages).when(receiver).searchForNewMessages();
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
return messages;
}
}).when(receiver).searchForNewMessages();
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
return null;
}
}).when(receiver).fetchMessages(messages);
doAnswer(invocation -> null).when(receiver).fetchMessages(messages);
final CountDownLatch idles = new CountDownLatch(2);
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
idles.countDown();
Thread.sleep(5000);
return null;
}
doAnswer(invocation -> {
idles.countDown();
Thread.sleep(5000);
return null;
}).when(folder).idle();
adapter.start();
@@ -856,20 +687,10 @@ public class ImapMailReceiverTests {
ImapIdleChannelAdapter adapter = new ImapIdleChannelAdapter(mailReceiver);
final AtomicReference<ImapIdleExceptionEvent> theEvent = new AtomicReference<ImapIdleExceptionEvent>();
final CountDownLatch latch = new CountDownLatch(1);
adapter.setApplicationEventPublisher(new ApplicationEventPublisher() {
@Override
public void publishEvent(ApplicationEvent event) {
assertNull("only one event expected", theEvent.get());
theEvent.set((ImapIdleExceptionEvent) event);
latch.countDown();
}
@Override
public void publishEvent(Object event) {
}
adapter.setApplicationEventPublisher(event -> {
assertNull("only one event expected", theEvent.get());
theEvent.set((ImapIdleExceptionEvent) event);
latch.countDown();
});
ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
taskScheduler.initialize();
@@ -898,34 +719,24 @@ public class ImapMailReceiverTests {
receiver.setBeanFactory(mock(BeanFactory.class));
receiver.afterPropertiesSet();
new Thread(new Runnable() {
@Override
public void run() {
try {
receiver.receive();
}
catch (javax.mail.MessagingException e) {
if (e.getCause() instanceof NullPointerException) {
e.printStackTrace();
failed.getAndIncrement();
}
}
new Thread(() -> {
try {
receiver.receive();
}
catch (javax.mail.MessagingException e) {
if (e.getCause() instanceof NullPointerException) {
failed.getAndIncrement();
}
}
}).start();
new Thread(new Runnable() {
@Override
public void run() {
try {
receiver.destroy();
}
catch (Exception ignore) {
// ignore
ignore.printStackTrace();
}
new Thread(() -> {
try {
receiver.destroy();
}
catch (Exception ignore) {
// ignore
}
}).start();
}

View File

@@ -60,6 +60,10 @@ public class MailReceivingMessageSourceTests {
private final ConcurrentLinkedQueue<javax.mail.Message[]> messages = new ConcurrentLinkedQueue<javax.mail.Message[]>();
StubMailReceiver() {
super();
}
@Override
public javax.mail.Message[] receive() {
return messages.poll();

View File

@@ -34,8 +34,6 @@ import javax.mail.Folder;
import javax.mail.Message;
import org.junit.Test;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.mail.ImapIdleChannelAdapter;
@@ -49,6 +47,7 @@ import org.springframework.util.ReflectionUtils;
*/
public class ImapIdleIntegrationTests {
@SuppressWarnings("resource")
@Test
public void testWithTransactionSynchronization() throws Exception {
final AtomicBoolean block = new AtomicBoolean(false);
@@ -62,17 +61,14 @@ public class ImapIdleIntegrationTests {
// setup mock scenario
receiver = spy(receiver);
doAnswer(new Answer<Object>() { // ensures that waitFornewMessages call blocks after a first execution
doAnswer(invocation -> {
// ensures that waitFornewMessages call blocks after a first execution
// to emulate the behavior of IDLE
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
if (block.get()) {
Thread.sleep(5000);
}
block.set(true);
return null;
if (block.get()) {
Thread.sleep(5000);
}
block.set(true);
return null;
}).when(receiver).waitForNewMessages();
Message m1 = mock(Message.class);
@@ -90,14 +86,9 @@ public class ImapIdleIntegrationTests {
// end mock setup
final CountDownLatch txProcessorLatch = new CountDownLatch(1);
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
txProcessorLatch.countDown();
return null;
}
doAnswer(invocation -> {
txProcessorLatch.countDown();
return null;
}).when(processor).process(any(Message.class));
adapter.start();

View File

@@ -212,6 +212,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore
return (messageWrapper != null) ? messageWrapper.getMessage() : null;
}
@Override
public MessageMetadata getMessageMetadata(UUID id) {
Assert.notNull(id, "'id' must not be null");
MessageWrapper messageWrapper =
@@ -474,7 +475,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore
*/
private final class MessageReadingMongoConverter extends MappingMongoConverter {
private MessageReadingMongoConverter(MongoDbFactory mongoDbFactory,
MessageReadingMongoConverter(MongoDbFactory mongoDbFactory,
MappingContext<? extends MongoPersistentEntity<?>, MongoPersistentProperty> mappingContext) {
super(new DefaultDbRefResolver(mongoDbFactory), mappingContext);
}
@@ -603,6 +604,10 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore
private static class UuidToDBObjectConverter implements Converter<UUID, DBObject> {
UuidToDBObjectConverter() {
super();
}
@Override
public DBObject convert(UUID source) {
BasicDBObject dbObject = new BasicDBObject();
@@ -614,6 +619,10 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore
private static class DBObjectToUUIDConverter implements Converter<DBObject, UUID> {
DBObjectToUUIDConverter() {
super();
}
@Override
public UUID convert(DBObject source) {
return UUID.fromString((String) source.get("_value"));
@@ -623,6 +632,10 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore
private static class MessageHistoryToDBObjectConverter implements Converter<MessageHistory, DBObject> {
MessageHistoryToDBObjectConverter() {
super();
}
@Override
public DBObject convert(MessageHistory source) {
BasicDBObject obj = new BasicDBObject();
@@ -642,8 +655,11 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore
private class DBObjectToGenericMessageConverter implements Converter<DBObject, GenericMessage<?>> {
@Override
DBObjectToGenericMessageConverter() {
super();
}
@Override
public GenericMessage<?> convert(DBObject source) {
@SuppressWarnings("unchecked")
Map<String, Object> headers =
@@ -660,6 +676,10 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore
private final class DBObjectToMutableMessageConverter implements Converter<DBObject, MutableMessage<?>> {
DBObjectToMutableMessageConverter() {
super();
}
@Override
public MutableMessage<?> convert(DBObject source) {
@SuppressWarnings("unchecked")
@@ -676,6 +696,10 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore
private class DBObjectToAdviceMessageConverter implements Converter<DBObject, AdviceMessage<?>> {
DBObjectToAdviceMessageConverter() {
super();
}
@Override
public AdviceMessage<?> convert(DBObject source) {
@SuppressWarnings("unchecked")
@@ -711,6 +735,10 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore
private final Converter<byte[], Object> deserializingConverter = new DeserializingConverter();
DBObjectToErrorMessageConverter() {
super();
}
@Override
public ErrorMessage convert(DBObject source) {
@SuppressWarnings("unchecked")
@@ -731,6 +759,10 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore
private final Converter<Object, byte[]> serializingConverter = new SerializingConverter();
ThrowableToBytesConverter() {
super();
}
@Override
public byte[] convert(Throwable source) {
return this.serializingConverter.convert(source);
@@ -783,7 +815,7 @@ public class MongoDbMessageStore extends AbstractMessageGroupStore
@SuppressWarnings("unused")
private int sequence;
private MessageWrapper(Message<?> message) {
MessageWrapper(Message<?> message) {
Assert.notNull(message, "'message' must not be null");
this.message = message;
this._messageType = message.getClass().getName();

View File

@@ -326,11 +326,12 @@ public abstract class AbstractMongoDbMessageStoreTests extends MongoDbAvailableT
private final String name = "abx";
private Abc() { }
Abc() { }
public String getName() {
return name;
}
}
public static class Xyz implements Serializable {
@@ -342,7 +343,8 @@ public abstract class AbstractMongoDbMessageStoreTests extends MongoDbAvailableT
@SuppressWarnings("unused")
private final String name = "xyz";
private Xyz() { }
Xyz() { }
}

View File

@@ -258,7 +258,7 @@ public abstract class AbstractMqttMessageDrivenChannelAdapter extends MessagePro
private volatile int qos;
private Topic(String topic, int qos) {
Topic(String topic, int qos) {
this.topic = topic;
this.qos = qos;
}

View File

@@ -275,23 +275,18 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
private void scheduleReconnect() {
try {
this.reconnectFuture = this.getTaskScheduler().scheduleWithFixedDelay(new Runnable() {
@Override
public void run() {
try {
if (logger.isDebugEnabled()) {
logger.debug("Attempting reconnect");
}
if (!MqttPahoMessageDrivenChannelAdapter.this.connected) {
connectAndSubscribe();
}
this.reconnectFuture = this.getTaskScheduler().scheduleWithFixedDelay(() -> {
try {
if (logger.isDebugEnabled()) {
logger.debug("Attempting reconnect");
}
catch (MqttException e) {
logger.error("Exception while connecting and subscribing", e);
if (!MqttPahoMessageDrivenChannelAdapter.this.connected) {
connectAndSubscribe();
}
}
catch (MqttException e) {
logger.error("Exception while connecting and subscribing", e);
}
}, this.recoveryInterval);
}
catch (Exception e) {

View File

@@ -335,6 +335,10 @@ public class BackToBackAdapterTests {
private final CountDownLatch latch = new CountDownLatch(2);
EventPublisher() {
super();
}
@Override
public void publishEvent(ApplicationEvent event) {
if (event instanceof MqttMessageSentEvent) {

View File

@@ -35,8 +35,6 @@ import org.apache.commons.logging.Log;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.BeanFactory;
@@ -80,15 +78,11 @@ public class DownstreamExceptionTests {
service.n = 0;
Log logger = spy(TestUtils.getPropertyValue(noErrorChannel, "logger", Log.class));
final CountDownLatch latch = new CountDownLatch(1);
doAnswer(new Answer<Void>() {
@Override
public Void answer(InvocationOnMock invocation) throws Throwable {
if ((invocation.getArgumentAt(0, String.class)).contains("Unhandled")) {
latch.countDown();
}
return null;
doAnswer(invocation -> {
if ((invocation.getArgumentAt(0, String.class)).contains("Unhandled")) {
latch.countDown();
}
return null;
}).when(logger).error(anyString(), any(Throwable.class));
new DirectFieldAccessor(noErrorChannel).setPropertyValue("logger", logger);
MqttPahoMessageHandler adapter = new MqttPahoMessageHandler("tcp://localhost:1883", "si-test-out");

View File

@@ -44,7 +44,6 @@ import java.util.concurrent.atomic.AtomicReference;
import javax.net.SocketFactory;
import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
import org.eclipse.paho.client.mqttv3.IMqttToken;
import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
import org.eclipse.paho.client.mqttv3.MqttCallback;
@@ -56,8 +55,6 @@ import org.eclipse.paho.client.mqttv3.MqttSecurityException;
import org.eclipse.paho.client.mqttv3.MqttToken;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.junit.Test;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.springframework.aop.framework.ProxyFactoryBean;
import org.springframework.beans.factory.BeanFactory;
@@ -87,14 +84,7 @@ public class MqttAdapterTests {
{
ProxyFactoryBean pfb = new ProxyFactoryBean();
pfb.addAdvice(new MethodInterceptor() {
@Override
public Object invoke(MethodInvocation invocation) throws Throwable {
return null;
}
});
pfb.addAdvice((MethodInterceptor) invocation -> null);
pfb.setInterfaces(IMqttToken.class);
this.alwaysComplete = (IMqttToken) pfb.getObject();
}
@@ -147,13 +137,7 @@ public class MqttAdapterTests {
factory = spy(factory);
final MqttAsyncClient client = mock(MqttAsyncClient.class);
doAnswer(new Answer<MqttAsyncClient>() {
@Override
public MqttAsyncClient answer(InvocationOnMock invocation) throws Throwable {
return client;
}
}).when(factory).getAsyncClientInstance(anyString(), anyString());
doAnswer(invocation -> client).when(factory).getAsyncClientInstance(anyString(), anyString());
MqttPahoMessageHandler handler = new MqttPahoMessageHandler("foo", "bar", factory);
handler.setDefaultTopic("mqtt-foo");
@@ -163,38 +147,30 @@ public class MqttAdapterTests {
final MqttToken token = mock(MqttToken.class);
final AtomicBoolean connectCalled = new AtomicBoolean();
doAnswer(new Answer<MqttToken>() {
@Override
public MqttToken answer(InvocationOnMock invocation) throws Throwable {
MqttConnectOptions options = invocation.getArgumentAt(0, MqttConnectOptions.class);
assertEquals(23, options.getConnectionTimeout());
assertEquals(45, options.getKeepAliveInterval());
assertEquals("pass", new String(options.getPassword()));
assertSame(socketFactory, options.getSocketFactory());
assertSame(props, options.getSSLProperties());
assertEquals("user", options.getUserName());
assertEquals("foo", options.getWillDestination());
assertEquals("bar", new String(options.getWillMessage().getPayload()));
assertEquals(2, options.getWillMessage().getQos());
connectCalled.set(true);
return token;
}
doAnswer(invocation -> {
MqttConnectOptions options = invocation.getArgumentAt(0, MqttConnectOptions.class);
assertEquals(23, options.getConnectionTimeout());
assertEquals(45, options.getKeepAliveInterval());
assertEquals("pass", new String(options.getPassword()));
assertSame(socketFactory, options.getSocketFactory());
assertSame(props, options.getSSLProperties());
assertEquals("user", options.getUserName());
assertEquals("foo", options.getWillDestination());
assertEquals("bar", new String(options.getWillMessage().getPayload()));
assertEquals(2, options.getWillMessage().getQos());
connectCalled.set(true);
return token;
}).when(client).connect(any(MqttConnectOptions.class));
doReturn(token).when(client).subscribe(any(String[].class), any(int[].class));
final MqttDeliveryToken deliveryToken = mock(MqttDeliveryToken.class);
final AtomicBoolean publishCalled = new AtomicBoolean();
doAnswer(new Answer<MqttDeliveryToken>() {
@Override
public MqttDeliveryToken answer(InvocationOnMock invocation) throws Throwable {
assertEquals("mqtt-foo", invocation.getArguments()[0]);
MqttMessage message = invocation.getArgumentAt(1, MqttMessage.class);
assertEquals("Hello, world!", new String(message.getPayload()));
publishCalled.set(true);
return deliveryToken;
}
doAnswer(invocation -> {
assertEquals("mqtt-foo", invocation.getArguments()[0]);
MqttMessage message = invocation.getArgumentAt(1, MqttMessage.class);
assertEquals("Hello, world!", new String(message.getPayload()));
publishCalled.set(true);
return deliveryToken;
}).when(client).publish(anyString(), any(MqttMessage.class));
handler.handleMessage(new GenericMessage<String>("Hello, world!"));
@@ -222,13 +198,7 @@ public class MqttAdapterTests {
factory = spy(factory);
final MqttAsyncClient client = mock(MqttAsyncClient.class);
doAnswer(new Answer<MqttAsyncClient>() {
@Override
public MqttAsyncClient answer(InvocationOnMock invocation) throws Throwable {
return client;
}
}).when(factory).getAsyncClientInstance(anyString(), anyString());
doAnswer(invocation -> client).when(factory).getAsyncClientInstance(anyString(), anyString());
final MqttToken token = mock(MqttToken.class);
final AtomicBoolean connectCalled = new AtomicBoolean();
@@ -237,41 +207,33 @@ public class MqttAdapterTests {
final CountDownLatch failInProcess = new CountDownLatch(1);
final CountDownLatch goodConnection = new CountDownLatch(2);
final MqttException reconnectException = new MqttException(MqttException.REASON_CODE_SERVER_CONNECT_ERROR);
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
if (failConnection.get()) {
failInProcess.countDown();
waitToFail.await(10, TimeUnit.SECONDS);
throw reconnectException;
}
MqttConnectOptions options = invocation.getArgumentAt(0, MqttConnectOptions.class);
assertEquals(23, options.getConnectionTimeout());
assertEquals(45, options.getKeepAliveInterval());
assertEquals("pass", new String(options.getPassword()));
assertSame(socketFactory, options.getSocketFactory());
assertSame(props, options.getSSLProperties());
assertEquals("user", options.getUserName());
assertEquals("foo", options.getWillDestination());
assertEquals("bar", new String(options.getWillMessage().getPayload()));
assertEquals(2, options.getWillMessage().getQos());
connectCalled.set(true);
goodConnection.countDown();
return token;
doAnswer(invocation -> {
if (failConnection.get()) {
failInProcess.countDown();
waitToFail.await(10, TimeUnit.SECONDS);
throw reconnectException;
}
MqttConnectOptions options = invocation.getArgumentAt(0, MqttConnectOptions.class);
assertEquals(23, options.getConnectionTimeout());
assertEquals(45, options.getKeepAliveInterval());
assertEquals("pass", new String(options.getPassword()));
assertSame(socketFactory, options.getSocketFactory());
assertSame(props, options.getSSLProperties());
assertEquals("user", options.getUserName());
assertEquals("foo", options.getWillDestination());
assertEquals("bar", new String(options.getWillMessage().getPayload()));
assertEquals(2, options.getWillMessage().getQos());
connectCalled.set(true);
goodConnection.countDown();
return token;
}).when(client).connect(any(MqttConnectOptions.class));
doReturn(token).when(client).subscribe(any(String[].class), any(int[].class));
doReturn(token).when(client).disconnect();
final AtomicReference<MqttCallback> callback = new AtomicReference<MqttCallback>();
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
callback.set(invocation.getArgumentAt(0, MqttCallback.class));
return null;
}
doAnswer(invocation -> {
callback.set(invocation.getArgumentAt(0, MqttCallback.class));
return null;
}).when(client).setCallback(any(MqttCallback.class));
when(client.isConnected()).thenReturn(true);
@@ -286,13 +248,9 @@ public class MqttAdapterTests {
adapter.setBeanFactory(mock(BeanFactory.class));
ApplicationEventPublisher applicationEventPublisher = mock(ApplicationEventPublisher.class);
final BlockingQueue<MqttIntegrationEvent> events = new LinkedBlockingQueue<MqttIntegrationEvent>();
doAnswer(new Answer<Void>() {
@Override
public Void answer(InvocationOnMock invocation) throws Throwable {
events.add(invocation.getArgumentAt(0, MqttIntegrationEvent.class));
return null;
}
doAnswer(invocation -> {
events.add(invocation.getArgumentAt(0, MqttIntegrationEvent.class));
return null;
}).when(applicationEventPublisher).publishEvent(any(MqttIntegrationEvent.class));
adapter.setApplicationEventPublisher(applicationEventPublisher);
adapter.setRecoveryInterval(500);

View File

@@ -123,7 +123,6 @@ public class SubscribableRedisChannel extends AbstractMessageChannel
return this.dispatcher.removeHandler(handler);
}
@SuppressWarnings("unchecked")
@Override
protected boolean doSend(Message<?> message, long arg1) {
this.redisTemplate.convertAndSend(this.topicName, this.messageConverter.fromMessage(message, Object.class));
@@ -212,7 +211,11 @@ public class SubscribableRedisChannel extends AbstractMessageChannel
private class MessageListenerDelegate {
@SuppressWarnings({ "unused", "unchecked" })
MessageListenerDelegate() {
super();
}
@SuppressWarnings({ "unused" })
public void handleMessage(Object payload) {
Message<?> siMessage = SubscribableRedisChannel.this.messageConverter.toMessage(payload, null);
try {

View File

@@ -138,6 +138,10 @@ public class RedisInboundChannelAdapter extends MessageProducerSupport {
private class MessageListenerDelegate {
MessageListenerDelegate() {
super();
}
@SuppressWarnings("unused")
public void handleMessage(Object object) {
sendMessage(convertMessage(object));

View File

@@ -332,6 +332,10 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport implements
private class ListenerTask implements SchedulingAwareRunnable {
ListenerTask() {
super();
}
@Override
public boolean isLongLived() {
return true;

View File

@@ -321,6 +321,10 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport impl
private class ListenerTask implements SchedulingAwareRunnable {
ListenerTask() {
super();
}
@Override
public boolean isLongLived() {
return true;

View File

@@ -16,8 +16,6 @@
package org.springframework.integration.redis.outbound;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
@@ -135,18 +133,16 @@ public class RedisOutboundGateway extends AbstractReplyProducingMessageHandler {
final byte[][] actualArgs = args;
return this.redisTemplate.execute(new RedisCallback<Object>() {
@Override
public Object doInRedis(RedisConnection connection) throws DataAccessException {
return connection.execute(command, actualArgs);
}
});
return this.redisTemplate.execute(
(RedisCallback<Object>) connection -> connection.execute(command, actualArgs));
}
private class PayloadArgumentsStrategy implements ArgumentsStrategy {
PayloadArgumentsStrategy() {
super();
}
@Override
public Object[] resolve(String command, Message<?> message) {
Object payload = message.getPayload();

View File

@@ -288,26 +288,20 @@ public class RedisStoreWritingMessageHandler extends AbstractMessageHandler {
if (this.extractPayloadElements) {
if ((payload instanceof Map<?, ?> && this.verifyAllMapValuesOfTypeNumber((Map<?, ?>) payload))) {
final Map<Object, Number> payloadAsMap = (Map<Object, Number>) payload;
this.processInPipeline(new PipelineCallback() {
@Override
public void process() {
for (Entry<Object, Number> entry : payloadAsMap.entrySet()) {
Number d = entry.getValue();
incrementOrOverwrite(ops, entry.getKey(), d == null ?
determineScore(message) :
NumberUtils.convertNumberToTargetClass(d, Double.class),
zsetIncrementHeader);
}
this.processInPipeline(() -> {
for (Entry<Object, Number> entry : payloadAsMap.entrySet()) {
Number d = entry.getValue();
incrementOrOverwrite(ops, entry.getKey(), d == null ?
determineScore(message) :
NumberUtils.convertNumberToTargetClass(d, Double.class),
zsetIncrementHeader);
}
});
}
else if (payload instanceof Collection<?>) {
this.processInPipeline(new PipelineCallback() {
@Override
public void process() {
for (Object object : ((Collection<?>) payload)) {
incrementOrOverwrite(ops, object, determineScore(message), zsetIncrementHeader);
}
this.processInPipeline(() -> {
for (Object object : ((Collection<?>) payload)) {
incrementOrOverwrite(ops, object, determineScore(message), zsetIncrementHeader);
}
});
}
@@ -350,12 +344,9 @@ public class RedisStoreWritingMessageHandler extends AbstractMessageHandler {
final BoundSetOperations<String, Object> ops =
(BoundSetOperations<String, Object>) this.redisTemplate.boundSetOps(set.getKey());
this.processInPipeline(new PipelineCallback() {
@Override
public void process() {
for (Object object : ((Collection<?>) payload)) {
ops.add(object);
}
this.processInPipeline(() -> {
for (Object object : ((Collection<?>) payload)) {
ops.add(object);
}
});
}
@@ -368,12 +359,7 @@ public class RedisStoreWritingMessageHandler extends AbstractMessageHandler {
private void writeToMap(final RedisMap<Object, Object> map, Message<?> message) {
final Object payload = message.getPayload();
if (this.extractPayloadElements && payload instanceof Map<?, ?>) {
this.processInPipeline(new PipelineCallback() {
@Override
public void process() {
map.putAll((Map<? extends Object, ? extends Object>) payload);
}
});
this.processInPipeline(() -> map.putAll((Map<? extends Object, ? extends Object>) payload));
}
else {
Object key = this.determineMapKey(message, false);
@@ -384,12 +370,7 @@ public class RedisStoreWritingMessageHandler extends AbstractMessageHandler {
private void writeToProperties(final RedisProperties properties, Message<?> message) {
final Object payload = message.getPayload();
if (this.extractPayloadElements && payload instanceof Properties) {
this.processInPipeline(new PipelineCallback() {
@Override
public void process() {
properties.putAll((Properties) payload);
}
});
this.processInPipeline(() -> properties.putAll((Properties) payload));
}
else {
Assert.isInstanceOf(String.class, payload, "For property, payload must be a String.");
@@ -482,4 +463,5 @@ public class RedisStoreWritingMessageHandler extends AbstractMessageHandler {
private interface PipelineCallback {
void process();
}
}

View File

@@ -47,14 +47,7 @@ import org.springframework.util.Assert;
public class RedisChannelPriorityMessageStore extends RedisChannelMessageStore
implements PriorityCapableChannelMessageStore {
private final Comparator<String> keysComparator = new Comparator<String>() {
@Override
public int compare(String s1, String s2) {
return s2.compareTo(s1);
}
};
private final Comparator<String> keysComparator = (s1, s2) -> s2.compareTo(s1);
public RedisChannelPriorityMessageStore(RedisConnectionFactory connectionFactory) {
super(connectionFactory);

View File

@@ -20,7 +20,6 @@ import java.net.InetAddress;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
@@ -32,13 +31,12 @@ import java.util.WeakHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.stream.Collectors;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.dao.CannotAcquireLockException;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
@@ -153,7 +151,7 @@ public final class RedisLockRegistry implements LockRegistry {
Assert.notNull(connectionFactory, "'connectionFactory' cannot be null");
Assert.notNull(registryKey, "'registryKey' cannot be null");
Assert.notNull(localRegistry, "'localRegistry' cannot be null");
this.redisTemplate = new RedisTemplate<String, RedisLockRegistry.RedisLock>();
this.redisTemplate = new RedisTemplate<>();
this.redisTemplate.setConnectionFactory(connectionFactory);
this.redisTemplate.setKeySerializer(new StringRedisSerializer());
this.redisTemplate.setValueSerializer(this.lockSerializer);
@@ -198,7 +196,7 @@ public final class RedisLockRegistry implements LockRegistry {
private Collection<RedisLock> getHardThreadLocks() {
List<RedisLock> locks = this.hardThreadLocks.get();
if (locks == null) {
locks = new LinkedList<RedisLock>();
locks = new LinkedList<>();
this.hardThreadLocks.set(locks);
}
return locks;
@@ -293,20 +291,15 @@ public final class RedisLockRegistry implements LockRegistry {
}
public Collection<Lock> listLocks() {
return this.redisTemplate.execute(new RedisCallback<Collection<Lock>>() {
@Override
public Collection<Lock> doInRedis(RedisConnection connection) throws DataAccessException {
Set<byte[]> keys = connection.keys((RedisLockRegistry.this.registryKey + ":*").getBytes());
ArrayList<Lock> list = new ArrayList<Lock>(keys.size());
if (keys.size() > 0) {
List<byte[]> locks = connection.mGet(keys.toArray(new byte[keys.size()][]));
for (byte[] lock : locks) {
list.add(RedisLockRegistry.this.lockSerializer.deserialize(lock));
}
}
return list;
return this.redisTemplate.execute((RedisCallback<Collection<Lock>>) connection -> {
Set<byte[]> keys = connection.keys((RedisLockRegistry.this.registryKey + ":*").getBytes());
if (keys.size() > 0) {
List<byte[]> locks = connection.mGet(keys.toArray(new byte[keys.size()][]));
return locks.stream()
.map(RedisLockRegistry.this.lockSerializer::deserialize)
.collect(Collectors.toList());
}
return Collections.emptyList();
});
}
@@ -324,7 +317,7 @@ public final class RedisLockRegistry implements LockRegistry {
private int reLock;
private RedisLock(String lockKey) {
RedisLock(String lockKey) {
this.lockKey = lockKey;
this.lockHost = RedisLockRegistry.hostName;
}
@@ -420,30 +413,25 @@ public final class RedisLockRegistry implements LockRegistry {
Boolean success = false;
try {
success = RedisLockRegistry.this.redisTemplate.execute(new RedisCallback<Boolean>() {
success = RedisLockRegistry.this.redisTemplate.execute((RedisCallback<Boolean>) connection -> {
@Override
public Boolean doInRedis(RedisConnection connection) throws DataAccessException {
/*
Perform Redis command 'SET resource-name anystring NX EX max-lock-time' directly.
As it is recommended by Redis: http://redis.io/commands/set.
This command isn't supported directly by RedisTemplate.
*/
long expireAfter = TimeoutUtils.toSeconds(RedisLockRegistry.this.expireAfter,
TimeUnit.MILLISECONDS);
RedisSerializer<String> serializer = RedisLockRegistry.this.redisTemplate.getStringSerializer();
byte[][] actualArgs = new byte[][] {
serializer.serialize(constructLockKey()),
RedisLockRegistry.this.lockSerializer.serialize(RedisLock.this),
serializer.serialize("NX"),
serializer.serialize("EX"),
serializer.serialize(String.valueOf(expireAfter))
};
return connection.execute("SET", actualArgs) != null;
}
/*
Perform Redis command 'SET resource-name anystring NX EX max-lock-time' directly.
As it is recommended by Redis: http://redis.io/commands/set.
This command isn't supported directly by RedisTemplate.
*/
long expireAfter = TimeoutUtils.toSeconds(RedisLockRegistry.this.expireAfter,
TimeUnit.MILLISECONDS);
RedisSerializer<String> serializer = RedisLockRegistry.this.redisTemplate.getStringSerializer();
byte[][] actualArgs = new byte[][] {
serializer.serialize(constructLockKey()),
RedisLockRegistry.this.lockSerializer.serialize(RedisLock.this),
serializer.serialize("NX"),
serializer.serialize("EX"),
serializer.serialize(String.valueOf(expireAfter))
};
return connection.execute("SET", actualArgs) != null;
});
}
finally {
@@ -603,6 +591,10 @@ public final class RedisLockRegistry implements LockRegistry {
private class LockSerializer implements RedisSerializer<RedisLock> {
LockSerializer() {
super();
}
@Override
public byte[] serialize(RedisLock t) throws SerializationException {
int hostLength = t.lockHost.length;

View File

@@ -38,9 +38,7 @@ import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.integration.redis.rules.RedisAvailable;
import org.springframework.integration.redis.rules.RedisAvailableTests;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.util.ReflectionUtils;
/**
@@ -66,13 +64,7 @@ public class SubscribableRedisChannelTests extends RedisAvailableTests {
RedisMessageListenerContainer.class));
final CountDownLatch latch = new CountDownLatch(3);
MessageHandler handler = new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
latch.countDown();
}
};
MessageHandler handler = message -> latch.countDown();
channel.subscribe(handler);
channel.send(new GenericMessage<String>("1"));

View File

@@ -34,8 +34,6 @@ import org.springframework.integration.redis.rules.RedisAvailableTests;
import org.springframework.integration.support.utils.IntegrationUtils;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.messaging.support.GenericMessage;
@@ -81,12 +79,9 @@ public class RedisChannelParserTests extends RedisAvailableTests {
final Message<?> m = new GenericMessage<String>("Hello Redis");
final CountDownLatch latch = new CountDownLatch(1);
redisChannel.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
assertEquals(m.getPayload(), message.getPayload());
latch.countDown();
}
redisChannel.subscribe(message -> {
assertEquals(m.getPayload(), message.getPayload());
latch.countDown();
});
redisChannel.send(m);

View File

@@ -45,7 +45,6 @@ import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.data.redis.RedisConnectionFailureException;
import org.springframework.data.redis.RedisSystemException;
import org.springframework.data.redis.connection.RedisConnectionFactory;
@@ -262,14 +261,7 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
final CountDownLatch stopLatch = new CountDownLatch(1);
endpoint.stop(new Runnable() {
@Override
public void run() {
stopLatch.countDown();
}
});
endpoint.stop(() -> stopLatch.countDown());
executorService.shutdown();
assertTrue(executorService.awaitTermination(10, TimeUnit.SECONDS));
@@ -282,7 +274,6 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
@Test
@RedisAvailable
@SuppressWarnings("unchecked")
@Ignore("JedisConnectionFactory doesn't support proper 'destroy()' and allows to create new fresh Redis connection")
public void testInt3196Recovery() throws Exception {
String queueName = "test.si.Int3196Recovery";
@@ -294,19 +285,9 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
RedisQueueMessageDrivenEndpoint endpoint = new RedisQueueMessageDrivenEndpoint(queueName, this.connectionFactory);
endpoint.setBeanFactory(Mockito.mock(BeanFactory.class));
endpoint.setApplicationEventPublisher(new ApplicationEventPublisher() {
@Override
public void publishEvent(ApplicationEvent event) {
exceptionEvents.add(event);
exceptionsLatch.countDown();
}
@Override
public void publishEvent(Object event) {
}
endpoint.setApplicationEventPublisher(event -> {
exceptionEvents.add((ApplicationEvent) event);
exceptionsLatch.countDown();
});
endpoint.setOutputChannel(channel);
endpoint.setReceiveTimeout(100);

View File

@@ -35,8 +35,6 @@ import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
import org.springframework.integration.redis.rules.RedisAvailable;
import org.springframework.integration.redis.rules.RedisAvailableTests;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.SubscribableChannel;
/**
@@ -108,6 +106,7 @@ public class RedisStoreInboundChannelAdapterIntegrationTests extends RedisAvaila
context.close();
}
@SuppressWarnings("resource")
@Test
@RedisAvailable
// synchronization rollback renames the list
@@ -120,14 +119,9 @@ public class RedisStoreInboundChannelAdapterIntegrationTests extends RedisAvaila
this.getClass());
SubscribableChannel fail = context.getBean("redisFailChannel", SubscribableChannel.class);
final CountDownLatch latch = new CountDownLatch(1);
fail.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
latch.countDown();
throw new RuntimeException("Test Rollback");
}
fail.subscribe(message -> {
latch.countDown();
throw new RuntimeException("Test Rollback");
});
SourcePollingChannelAdapter spca = context.getBean("listAdapterWithSynchronizationAndRollback",
SourcePollingChannelAdapter.class);

View File

@@ -83,7 +83,7 @@ public class RedisPublishingMessageHandlerTests extends RedisAvailableTests {
private final CountDownLatch latch;
private Listener(CountDownLatch latch) {
Listener(CountDownLatch latch) {
this.latch = latch;
}

View File

@@ -327,27 +327,19 @@ public class RedisMessageGroupStoreTests extends RedisAvailableTests {
for (int i = 0; i < 100; i++) {
executor = Executors.newCachedThreadPool();
executor.execute(new Runnable() {
@Override
public void run() {
MessageGroup group = store1.addMessageToGroup(1, message);
if (group.getMessages().size() != 1) {
failures.add("ADD");
throw new AssertionFailedError("Failed on ADD");
}
executor.execute(() -> {
MessageGroup group = store1.addMessageToGroup(1, message);
if (group.getMessages().size() != 1) {
failures.add("ADD");
throw new AssertionFailedError("Failed on ADD");
}
});
executor.execute(new Runnable() {
@Override
public void run() {
store2.removeMessagesFromGroup(1, message);
MessageGroup group = store2.getMessageGroup(1);
if (group.getMessages().size() != 0) {
failures.add("REMOVE");
throw new AssertionFailedError("Failed on Remove");
}
executor.execute(() -> {
store2.removeMessagesFromGroup(1, message);
MessageGroup group = store2.getMessageGroup(1);
if (group.getMessages().size() != 0) {
failures.add("REMOVE");
throw new AssertionFailedError("Failed on Remove");
}
});

View File

@@ -131,17 +131,12 @@ public class AggregatorWithRedisLocksTests extends RedisAvailableTests {
public void testDistributedAggregator() throws Exception {
this.releaseStrategy.reset(1);
Executors.newSingleThreadExecutor().execute(asyncSend("foo", 1, 1));
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
try {
in2.send(new GenericMessage<String>("bar", stubHeaders(2, 2, 1)));
}
catch (Exception e) {
e.printStackTrace();
exception = e;
}
Executors.newSingleThreadExecutor().execute(() -> {
try {
in2.send(new GenericMessage<String>("bar", stubHeaders(2, 2, 1)));
}
catch (Exception e) {
exception = e;
}
});
assertTrue(this.releaseStrategy.latch2.await(10, TimeUnit.SECONDS));
@@ -162,17 +157,12 @@ public class AggregatorWithRedisLocksTests extends RedisAvailableTests {
}
private Runnable asyncSend(final String payload, final int sequence, final int correlation) {
return new Runnable() {
@Override
public void run() {
try {
in.send(new GenericMessage<String>(payload, stubHeaders(sequence, 2, correlation)));
}
catch (Exception e) {
e.printStackTrace();
exception = e;
}
return () -> {
try {
in.send(new GenericMessage<String>(payload, stubHeaders(sequence, 2, correlation)));
}
catch (Exception e) {
exception = e;
}
};
}

View File

@@ -31,7 +31,6 @@ import static org.junit.Assert.fail;
import java.util.Collection;
import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
@@ -82,7 +81,7 @@ public class RedisLockRegistryTests extends RedisAvailableTests {
}
private RedisTemplate<String, ?> createTemplate() {
RedisTemplate<String, ?> template = new RedisTemplate<String, Object>();
RedisTemplate<String, ?> template = new RedisTemplate<>();
template.setConnectionFactory(this.getConnectionFactoryForTest());
template.setKeySerializer(new StringRedisSerializer());
template.afterPropertiesSet();
@@ -206,21 +205,17 @@ public class RedisLockRegistryTests extends RedisAvailableTests {
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 = registry.obtain("foo");
locked.set(lock2.tryLock(200, TimeUnit.MILLISECONDS));
latch.countDown();
try {
lock2.unlock();
}
catch (IllegalStateException ise) {
return ise;
}
return null;
Future<Object> result = Executors.newSingleThreadExecutor().submit(() -> {
Lock lock2 = registry.obtain("foo");
locked.set(lock2.tryLock(200, TimeUnit.MILLISECONDS));
latch.countDown();
try {
lock2.unlock();
}
catch (IllegalStateException ise) {
return ise;
}
return null;
});
assertTrue(latch.await(10, TimeUnit.SECONDS));
assertFalse(locked.get());
@@ -242,25 +237,21 @@ public class RedisLockRegistryTests extends RedisAvailableTests {
final CountDownLatch latch3 = new CountDownLatch(1);
lock1.lockInterruptibly();
assertNotNull(TestUtils.getPropertyValue(registry, "hardThreadLocks", ThreadLocal.class).get());
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
Lock lock2 = registry.obtain("foo");
try {
latch1.countDown();
lock2.lockInterruptibly();
assertNotNull(TestUtils.getPropertyValue(registry, "hardThreadLocks", ThreadLocal.class).get());
latch2.await(10, TimeUnit.SECONDS);
locked.set(true);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
finally {
lock2.unlock();
latch3.countDown();
}
Executors.newSingleThreadExecutor().execute(() -> {
Lock lock2 = registry.obtain("foo");
try {
latch1.countDown();
lock2.lockInterruptibly();
assertNotNull(TestUtils.getPropertyValue(registry, "hardThreadLocks", ThreadLocal.class).get());
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));
@@ -284,31 +275,26 @@ public class RedisLockRegistryTests extends RedisAvailableTests {
final CountDownLatch latch3 = new CountDownLatch(1);
lock1.lockInterruptibly();
assertNotNull(TestUtils.getPropertyValue(registry1, "hardThreadLocks", ThreadLocal.class).get());
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
Lock lock2 = registry2.obtain("foo");
Executors.newSingleThreadExecutor().execute(() -> {
Lock lock2 = registry2.obtain("foo");
try {
latch1.countDown();
lock2.lockInterruptibly();
assertNotNull(TestUtils.getPropertyValue(registry2, "hardThreadLocks", ThreadLocal.class).get());
latch2.await(10, TimeUnit.SECONDS);
locked.set(true);
}
catch (InterruptedException e1) {
Thread.currentThread().interrupt();
logger.error("Interrupted while locking: " + lock2, e1);
}
finally {
try {
latch1.countDown();
lock2.lockInterruptibly();
assertNotNull(TestUtils.getPropertyValue(registry2, "hardThreadLocks", ThreadLocal.class).get());
latch2.await(10, TimeUnit.SECONDS);
locked.set(true);
logger.debug("Locks in store: " + registry2.listLocks());
lock2.unlock();
latch3.countDown();
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
logger.error("Interrupted while locking: " + lock2, e);
}
finally {
try {
lock2.unlock();
latch3.countDown();
}
catch (IllegalStateException e) {
logger.error("Failed to unlock: " + lock2, e);
}
catch (IllegalStateException e2) {
logger.error("Failed to unlock: " + lock2, e2);
}
}
});
@@ -330,19 +316,15 @@ public class RedisLockRegistryTests extends RedisAvailableTests {
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 (IllegalStateException ise) {
latch.countDown();
return ise;
}
return null;
Future<Object> result = Executors.newSingleThreadExecutor().submit(() -> {
try {
lock.unlock();
}
catch (IllegalStateException ise) {
latch.countDown();
return ise;
}
return null;
});
assertTrue(latch.await(10, TimeUnit.SECONDS));
assertFalse(locked.get());
@@ -500,15 +482,10 @@ public class RedisLockRegistryTests extends RedisAvailableTests {
Long expire = getExpire(registry, "foo");
Future<Object> result = Executors.newSingleThreadExecutor().submit(new Callable<Object>() {
@Override
public Object call() throws Exception {
Lock lock2 = registry.obtain("foo");
assertFalse(lock2.tryLock());
return null;
}
Future<Object> result = Executors.newSingleThreadExecutor().submit(() -> {
Lock lock2 = registry.obtain("foo");
assertFalse(lock2.tryLock());
return null;
});
result.get();
assertEquals(expire, getExpire(registry, "foo"));

View File

@@ -27,15 +27,15 @@ import java.rmi.RemoteException;
import org.junit.Before;
import org.junit.Test;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.MessagingException;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.gateway.RequestReplyExchanger;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.test.util.SocketUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.remoting.RemoteLookupFailureException;
import org.springframework.remoting.rmi.RmiServiceExporter;
@@ -160,6 +160,11 @@ public class RmiOutboundGatewayTests {
private static class TestExchanger implements RequestReplyExchanger {
TestExchanger() {
super();
}
@Override
public Message<?> exchange(Message<?> message) {
if (message.getPayload().equals("fail")) {
new AbstractReplyProducingMessageHandler() {
@@ -176,6 +181,11 @@ public class RmiOutboundGatewayTests {
private static class NonSerializableTestObject {
NonSerializableTestObject() {
super();
}
}
}

View File

@@ -26,11 +26,12 @@ import java.io.InputStream;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.io.AbstractResource;
import org.springframework.messaging.MessageChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.MessageChannel;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -67,8 +68,13 @@ public class Jsr223RefreshTests {
private static class CycleResource extends AbstractResource {
private int count = -1;
private String[] scripts = {"\"ruby-#{payload}-0\"", "\"ruby-#{payload}-1\""};
private final String[] scripts = {"\"ruby-#{payload}-0\"", "\"ruby-#{payload}-1\""};
CycleResource() {
super();
}
@Override
public String getDescription() {
return "CycleResource";
}
@@ -83,6 +89,7 @@ public class Jsr223RefreshTests {
return -1;
}
@Override
public InputStream getInputStream() throws IOException {
if (++count > scripts.length - 1) {
count = 0;

View File

@@ -469,6 +469,10 @@ public class DefaultSftpSessionFactory implements SessionFactory<LsEntry>, Share
*/
private class UserInfoWrapper implements UserInfo, UIKeyboardInteractive {
UserInfoWrapper() {
super();
}
/**
* Convenience to check whether enclosing factory's UserInfo is configured.
* @return true if there's a delegate.

View File

@@ -261,14 +261,7 @@ public abstract class AbstractStompSessionManager implements StompSessionManager
if (this.stompClient.getTaskScheduler() != null) {
this.reconnectFuture = this.stompClient.getTaskScheduler()
.schedule(new Runnable() {
@Override
public void run() {
connect();
}
}, new Date(System.currentTimeMillis() + this.recoveryInterval));
.schedule((Runnable) () -> connect(), new Date(System.currentTimeMillis() + this.recoveryInterval));
}
else {
this.logger.info("For automatic reconnection the 'stompClient' should be configured with a TaskScheduler.");
@@ -376,6 +369,10 @@ public abstract class AbstractStompSessionManager implements StompSessionManager
private volatile StompSession session;
CompositeStompSessionHandler() {
super();
}
void addHandler(StompSessionHandler delegate) {
if (this.session != null) {
delegate.afterConnected(this.session, getConnectHeaders());

View File

@@ -240,32 +240,22 @@ public class StompInboundChannelAdapter extends MessageProducerSupport implement
if (this.stompSessionManager.isAutoReceiptEnabled()) {
final ApplicationEventPublisher applicationEventPublisher = this.applicationEventPublisher;
if (applicationEventPublisher != null) {
subscription.addReceiptTask(new Runnable() {
@Override
public void run() {
StompReceiptEvent event = new StompReceiptEvent(StompInboundChannelAdapter.this,
destination, subscription.getReceiptId(), StompCommand.SUBSCRIBE, false);
applicationEventPublisher.publishEvent(event);
}
subscription.addReceiptTask(() -> {
StompReceiptEvent event = new StompReceiptEvent(StompInboundChannelAdapter.this,
destination, subscription.getReceiptId(), StompCommand.SUBSCRIBE, false);
applicationEventPublisher.publishEvent(event);
});
}
subscription.addReceiptLostTask(new Runnable() {
@Override
public void run() {
if (applicationEventPublisher != null) {
StompReceiptEvent event = new StompReceiptEvent(StompInboundChannelAdapter.this,
destination, subscription.getReceiptId(), StompCommand.SUBSCRIBE, true);
applicationEventPublisher.publishEvent(event);
}
else {
logger.error("The receipt [" + subscription.getReceiptId() + "] is lost for [" +
subscription.getSubscriptionId() + "] on destination [" + destination + "]");
}
subscription.addReceiptLostTask(() -> {
if (applicationEventPublisher != null) {
StompReceiptEvent event = new StompReceiptEvent(StompInboundChannelAdapter.this,
destination, subscription.getReceiptId(), StompCommand.SUBSCRIBE, true);
applicationEventPublisher.publishEvent(event);
}
else {
logger.error("The receipt [" + subscription.getReceiptId() + "] is lost for [" +
subscription.getSubscriptionId() + "] on destination [" + destination + "]");
}
});
}
this.subscriptions.put(destination, subscription);

View File

@@ -149,34 +149,24 @@ public class StompMessageHandler extends AbstractMessageHandler implements Appli
final String destination = stompHeaders.getDestination();
final ApplicationEventPublisher applicationEventPublisher = this.applicationEventPublisher;
if (applicationEventPublisher != null) {
receiptable.addReceiptTask(new Runnable() {
@Override
public void run() {
StompReceiptEvent event = new StompReceiptEvent(StompMessageHandler.this,
destination, receiptable.getReceiptId(), StompCommand.SEND, false);
event.setMessage(message);
applicationEventPublisher.publishEvent(event);
}
receiptable.addReceiptTask(() -> {
StompReceiptEvent event = new StompReceiptEvent(StompMessageHandler.this,
destination, receiptable.getReceiptId(), StompCommand.SEND, false);
event.setMessage(message);
applicationEventPublisher.publishEvent(event);
});
}
receiptable.addReceiptLostTask(new Runnable() {
@Override
public void run() {
if (applicationEventPublisher != null) {
StompReceiptEvent event = new StompReceiptEvent(StompMessageHandler.this,
destination, receiptable.getReceiptId(), StompCommand.SEND, true);
event.setMessage(message);
applicationEventPublisher.publishEvent(event);
}
else {
logger.error("The receipt [" + receiptable.getReceiptId() + "] is lost for [" +
message + "] on destination [" + destination + "]");
}
receiptable.addReceiptLostTask(() -> {
if (applicationEventPublisher != null) {
StompReceiptEvent event = new StompReceiptEvent(StompMessageHandler.this,
destination, receiptable.getReceiptId(), StompCommand.SEND, true);
event.setMessage(message);
applicationEventPublisher.publishEvent(event);
}
else {
logger.error("The receipt [" + receiptable.getReceiptId() + "] is lost for [" +
message + "] on destination [" + destination + "]");
}
});
}
}
@@ -224,6 +214,10 @@ public class StompMessageHandler extends AbstractMessageHandler implements Appli
private class IntegrationOutboundStompSessionHandler extends StompSessionHandlerAdapter {
IntegrationOutboundStompSessionHandler() {
super();
}
@Override
public void afterConnected(StompSession session, StompHeaders connectedHeaders) {
StompMessageHandler.this.transportError = null;

View File

@@ -303,7 +303,6 @@ public class StompInboundChannelAdapterWebSocketIntegrationTests extends LogAdju
}
@Bean
@SuppressWarnings("unchecked")
public ApplicationListener<ApplicationEvent> stompEventListener() {
ApplicationEventListeningMessageProducer producer = new ApplicationEventListeningMessageProducer();
producer.setEventTypes(StompIntegrationEvent.class);
@@ -355,23 +354,17 @@ public class StompInboundChannelAdapterWebSocketIntegrationTests extends LogAdju
//TODO SimpleBrokerMessageHandler doesn't support RECEIPT frame, hence we emulate it this way
@Bean
@SuppressWarnings("unchecked")
public ApplicationListener<SessionSubscribeEvent> webSocketEventListener(
final AbstractSubscribableChannel clientOutboundChannel) {
return new ApplicationListener<SessionSubscribeEvent>() {
@Override
public void onApplicationEvent(SessionSubscribeEvent event) {
Message<byte[]> message = event.getMessage();
StompHeaderAccessor stompHeaderAccessor = StompHeaderAccessor.wrap(message);
if (stompHeaderAccessor.getReceipt() != null) {
stompHeaderAccessor.setHeader("stompCommand", StompCommand.RECEIPT);
stompHeaderAccessor.setReceiptId(stompHeaderAccessor.getReceipt());
clientOutboundChannel.send(
MessageBuilder.createMessage(new byte[0], stompHeaderAccessor.getMessageHeaders()));
}
return event -> {
Message<byte[]> message = event.getMessage();
StompHeaderAccessor stompHeaderAccessor = StompHeaderAccessor.wrap(message);
if (stompHeaderAccessor.getReceipt() != null) {
stompHeaderAccessor.setHeader("stompCommand", StompCommand.RECEIPT);
stompHeaderAccessor.setReceiptId(stompHeaderAccessor.getReceipt());
clientOutboundChannel.send(
MessageBuilder.createMessage(new byte[0], stompHeaderAccessor.getMessageHeaders()));
}
};
}

View File

@@ -51,7 +51,7 @@ public class ByteStreamWritingMessageHandlerTests {
private PollingConsumer endpoint;
private TestTrigger trigger = new TestTrigger();
private final TestTrigger trigger = new TestTrigger();
private ThreadPoolTaskScheduler scheduler;
@@ -244,6 +244,11 @@ public class ByteStreamWritingMessageHandlerTests {
private volatile CountDownLatch latch = new CountDownLatch(1);
TestTrigger() {
super();
}
@Override
public Date nextExecutionTime(TriggerContext triggerContext) {
if (!hasRun.getAndSet(true)) {
return new Date();

View File

@@ -50,7 +50,7 @@ public class CharacterStreamWritingMessageHandlerTests {
private PollingConsumer endpoint;
private TestTrigger trigger = new TestTrigger();
private final TestTrigger trigger = new TestTrigger();
private ThreadPoolTaskScheduler scheduler;
@@ -192,12 +192,13 @@ public class CharacterStreamWritingMessageHandlerTests {
private static class TestObject {
private String text;
private final String text;
TestObject(String text) {
this.text = text;
}
@Override
public String toString() {
return this.text;
}
@@ -211,6 +212,11 @@ public class CharacterStreamWritingMessageHandlerTests {
private volatile CountDownLatch latch = new CountDownLatch(1);
TestTrigger() {
super();
}
@Override
public Date nextExecutionTime(TriggerContext triggerContext) {
if (!hasRun.getAndSet(true)) {
return new Date();

View File

@@ -18,9 +18,6 @@ package org.springframework.integration.syslog.inbound;
import org.springframework.integration.channel.FixedSubscriberChannel;
import org.springframework.integration.ip.udp.UnicastReceivingChannelAdapter;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
/**
* UDP implementation of a syslog inbound channel adapter.
@@ -58,14 +55,7 @@ public class UdpSyslogReceivingChannelAdapter extends SyslogReceivingChannelAdap
"of the provided 'UnicastReceivingChannelAdapter' to support Syslog conversion " +
"for the incoming UDP packets");
}
this.udpAdapter.setOutputChannel(new FixedSubscriberChannel(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
convertAndSend(message);
}
}));
this.udpAdapter.setOutputChannel(new FixedSubscriberChannel(message -> convertAndSend(message)));
if (!this.udpAdapterSet) {
this.udpAdapter.afterPropertiesSet();
}

View File

@@ -38,8 +38,6 @@ import javax.net.SocketFactory;
import org.apache.commons.logging.Log;
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;
@@ -76,7 +74,6 @@ public class SyslogReceivingChannelAdapterTests {
factory.afterPropertiesSet();
factory.start();
UdpSyslogReceivingChannelAdapter adapter = (UdpSyslogReceivingChannelAdapter) factory.getObject();
Thread.sleep(1000);
byte[] buf = "<157>JUL 26 22:08:35 WEBERN TESTING[70729]: TEST SYSLOG MESSAGE".getBytes("UTF-8");
DatagramPacket packet = new DatagramPacket(buf, buf.length, new InetSocketAddress("localhost", port));
DatagramSocket socket = new DatagramSocket();
@@ -98,13 +95,9 @@ public class SyslogReceivingChannelAdapterTests {
factory.setOutputChannel(outputChannel);
ApplicationEventPublisher publisher = mock(ApplicationEventPublisher.class);
final CountDownLatch latch = new CountDownLatch(2);
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
latch.countDown();
return null;
}
doAnswer(invocation -> {
latch.countDown();
return null;
}).when(publisher).publishEvent(any(ApplicationEvent.class));
factory.setApplicationEventPublisher(publisher);
factory.setBeanFactory(mock(BeanFactory.class));
@@ -114,19 +107,14 @@ public class SyslogReceivingChannelAdapterTests {
Log logger = spy(TestUtils.getPropertyValue(adapter, "logger", Log.class));
doReturn(true).when(logger).isDebugEnabled();
final CountDownLatch sawLog = new CountDownLatch(1);
doAnswer(new Answer<Void>() {
@Override
public Void answer(InvocationOnMock invocation) throws Throwable {
if ((invocation.getArgumentAt(0, String.class)).contains("Error on syslog socket")) {
sawLog.countDown();
}
invocation.callRealMethod();
return null;
doAnswer(invocation -> {
if ((invocation.getArgumentAt(0, String.class)).contains("Error on syslog socket")) {
sawLog.countDown();
}
invocation.callRealMethod();
return null;
}).when(logger).debug(anyString());
new DirectFieldAccessor(adapter).setPropertyValue("logger", logger);
Thread.sleep(1000);
byte[] buf = "<157>JUL 26 22:08:35 WEBERN TESTING[70729]: TEST SYSLOG MESSAGE\n".getBytes("UTF-8");
Socket socket = SocketFactory.getDefault().createSocket("localhost", port);
socket.getOutputStream().write(buf);
@@ -154,7 +142,6 @@ public class SyslogReceivingChannelAdapterTests {
DefaultMessageConverter defaultMessageConverter = new DefaultMessageConverter();
defaultMessageConverter.setAsMap(false);
adapter.setConverter(defaultMessageConverter);
Thread.sleep(1000);
byte[] buf = "<157>JUL 26 22:08:35 WEBERN TESTING[70729]: TEST SYSLOG MESSAGE".getBytes("UTF-8");
DatagramPacket packet = new DatagramPacket(buf, buf.length, new InetSocketAddress("localhost", port));
DatagramSocket socket = new DatagramSocket();
@@ -177,13 +164,9 @@ public class SyslogReceivingChannelAdapterTests {
factory.setOutputChannel(outputChannel);
ApplicationEventPublisher publisher = mock(ApplicationEventPublisher.class);
final CountDownLatch latch = new CountDownLatch(2);
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
latch.countDown();
return null;
}
doAnswer(invocation -> {
latch.countDown();
return null;
}).when(publisher).publishEvent(any(ApplicationEvent.class));
factory.setBeanFactory(mock(BeanFactory.class));
AbstractServerConnectionFactory connectionFactory = new TcpNioServerConnectionFactory(port);
@@ -197,19 +180,14 @@ public class SyslogReceivingChannelAdapterTests {
Log logger = spy(TestUtils.getPropertyValue(adapter, "logger", Log.class));
doReturn(true).when(logger).isDebugEnabled();
final CountDownLatch sawLog = new CountDownLatch(1);
doAnswer(new Answer<Void>() {
@Override
public Void answer(InvocationOnMock invocation) throws Throwable {
if ((invocation.getArgumentAt(0, String.class)).contains("Error on syslog socket")) {
sawLog.countDown();
}
invocation.callRealMethod();
return null;
doAnswer(invocation -> {
if ((invocation.getArgumentAt(0, String.class)).contains("Error on syslog socket")) {
sawLog.countDown();
}
invocation.callRealMethod();
return null;
}).when(logger).debug(anyString());
new DirectFieldAccessor(adapter).setPropertyValue("logger", logger);
Thread.sleep(1000);
byte[] buf = ("253 <14>1 2014-06-20T09:14:07+00:00 loggregator d0602076-b14a-4c55-852a-981e7afeed38 DEA - " +
"[exampleSDID@32473 iut=\\\"3\\\" eventSource=\\\"Application\\\" eventID=\\\"1011\\\"]" +
"[exampleSDID@32473 iut=\\\"3\\\" eventSource=\\\"Application\\\" eventID=\\\"1011\\\"] Removing instance")
@@ -239,7 +217,6 @@ public class SyslogReceivingChannelAdapterTests {
factory.afterPropertiesSet();
factory.start();
UdpSyslogReceivingChannelAdapter adapter = (UdpSyslogReceivingChannelAdapter) factory.getObject();
Thread.sleep(1000);
byte[] buf = ("<14>1 2014-06-20T09:14:07+00:00 loggregator d0602076-b14a-4c55-852a-981e7afeed38 DEA - " +
"[exampleSDID@32473 iut=\\\"3\\\" eventSource=\\\"Application\\\" eventID=\\\"1011\\\"]" +
"[exampleSDID@32473 iut=\\\"3\\\" eventSource=\\\"Application\\\" eventID=\\\"1011\\\"] Removing instance")

View File

@@ -194,13 +194,7 @@ public class TestSubscriber<T>
public static void await(Duration timeout,
final String errorMessage,
BooleanSupplier conditionSupplier) {
await(timeout, new Supplier<String>() {
@Override
public String get() {
return errorMessage;
}
}, conditionSupplier);
await(timeout, () -> errorMessage, conditionSupplier);
}
/**

View File

@@ -18,7 +18,6 @@ package org.springframework.integration.test.util;
import java.io.File;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.List;
import java.util.Properties;
import java.util.Set;
@@ -45,8 +44,6 @@ import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.ErrorHandler;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.ReflectionUtils.MethodCallback;
import org.springframework.util.ReflectionUtils.MethodFilter;
import org.springframework.util.StringUtils;
/**
@@ -133,7 +130,7 @@ public abstract class TestUtils {
public static class TestApplicationContext extends GenericApplicationContext {
private TestApplicationContext() {
TestApplicationContext() {
super();
}
@@ -156,26 +153,14 @@ public abstract class TestUtils {
final AtomicReference<String> componentName = new AtomicReference<String>();
for (Class<?> intface : interfaces) {
if ("org.springframework.integration.support.context.NamedComponent".equals(intface.getName())) {
ReflectionUtils.doWithMethods(channel.getClass(), new MethodCallback() {
@Override
public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
try {
componentName.set((String) method.invoke(channel, new Object[0]));
}
catch (InvocationTargetException e) {
throw new IllegalArgumentException(e);
}
ReflectionUtils.doWithMethods(channel.getClass(), method -> {
try {
componentName.set((String) method.invoke(channel, new Object[0]));
}
}, new MethodFilter() {
@Override
public boolean matches(Method method) {
return method.getName().equals("getComponentName");
catch (InvocationTargetException e) {
throw new IllegalArgumentException(e);
}
});
}, method -> method.getName().equals("getComponentName"));
break;
}
}
@@ -229,7 +214,7 @@ public abstract class TestUtils {
private final TestApplicationContext context;
private MessagePublishingErrorHandler(TestApplicationContext ctx) {
MessagePublishingErrorHandler(TestApplicationContext ctx) {
this.context = ctx;
}

View File

@@ -240,6 +240,10 @@ abstract class AbstractTwitterMessageSource<T> extends IntegrationObjectSupport
private class TweetComparator implements Comparator<T> {
TweetComparator() {
super();
}
@Override
public int compare(T tweet1, T tweet2) {
// hopefully temporary logic. Will suggest that SpringSocial use a common base class for DM and Tweet

View File

@@ -30,8 +30,6 @@ import java.util.List;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Matchers;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
@@ -80,15 +78,11 @@ public class TwitterSearchOutboundGatewayTests {
Tweet tweet = mock(Tweet.class);
SearchMetadata searchMetadata = mock(SearchMetadata.class);
final SearchResults searchResults = new SearchResults(Collections.singletonList(tweet), searchMetadata);
doAnswer(new Answer<SearchResults>() {
@Override
public SearchResults answer(InvocationOnMock invocation) throws Throwable {
SearchParameters searchParameters = invocation.getArgumentAt(0, SearchParameters.class);
assertEquals("foo", searchParameters.getQuery());
assertEquals(Integer.valueOf(20), searchParameters.getCount());
return searchResults;
}
doAnswer(invocation -> {
SearchParameters searchParameters = invocation.getArgumentAt(0, SearchParameters.class);
assertEquals("foo", searchParameters.getQuery());
assertEquals(Integer.valueOf(20), searchParameters.getCount());
return searchResults;
}).when(this.searchOps).search(Matchers.any(SearchParameters.class));
this.gateway.handleMessage(new GenericMessage<String>("foo"));
Message<?> reply = this.outputChannel.receive(0);
@@ -107,15 +101,11 @@ public class TwitterSearchOutboundGatewayTests {
Tweet tweet = mock(Tweet.class);
SearchMetadata searchMetadata = mock(SearchMetadata.class);
final SearchResults searchResults = new SearchResults(Collections.singletonList(tweet), searchMetadata);
doAnswer(new Answer<SearchResults>() {
@Override
public SearchResults answer(InvocationOnMock invocation) throws Throwable {
SearchParameters searchParameters = invocation.getArgumentAt(0, SearchParameters.class);
assertEquals("foo", searchParameters.getQuery());
assertEquals(Integer.valueOf(30), searchParameters.getCount());
return searchResults;
}
doAnswer(invocation -> {
SearchParameters searchParameters = invocation.getArgumentAt(0, SearchParameters.class);
assertEquals("foo", searchParameters.getQuery());
assertEquals(Integer.valueOf(30), searchParameters.getCount());
return searchResults;
}).when(this.searchOps).search(Matchers.any(SearchParameters.class));
this.gateway.handleMessage(new GenericMessage<String>("foo"));
Message<?> reply = this.outputChannel.receive(0);
@@ -134,17 +124,13 @@ public class TwitterSearchOutboundGatewayTests {
Tweet tweet = mock(Tweet.class);
SearchMetadata searchMetadata = mock(SearchMetadata.class);
final SearchResults searchResults = new SearchResults(Collections.singletonList(tweet), searchMetadata);
doAnswer(new Answer<SearchResults>() {
@Override
public SearchResults answer(InvocationOnMock invocation) throws Throwable {
SearchParameters searchParameters = invocation.getArgumentAt(0, SearchParameters.class);
assertEquals("bar", searchParameters.getQuery());
assertEquals(Integer.valueOf(1), searchParameters.getCount());
assertEquals(Long.valueOf(2), searchParameters.getSinceId());
assertEquals(Long.valueOf(3), searchParameters.getMaxId());
return searchResults;
}
doAnswer(invocation -> {
SearchParameters searchParameters = invocation.getArgumentAt(0, SearchParameters.class);
assertEquals("bar", searchParameters.getQuery());
assertEquals(Integer.valueOf(1), searchParameters.getCount());
assertEquals(Long.valueOf(2), searchParameters.getSinceId());
assertEquals(Long.valueOf(3), searchParameters.getMaxId());
return searchResults;
}).when(this.searchOps).search(Matchers.any(SearchParameters.class));
this.gateway.handleMessage(new GenericMessage<String>("foo"));
Message<?> reply = this.outputChannel.receive(0);
@@ -162,14 +148,10 @@ public class TwitterSearchOutboundGatewayTests {
SearchMetadata searchMetadata = mock(SearchMetadata.class);
final SearchResults searchResults = new SearchResults(Collections.singletonList(tweet), searchMetadata);
final SearchParameters parameters = new SearchParameters("bar");
doAnswer(new Answer<SearchResults>() {
@Override
public SearchResults answer(InvocationOnMock invocation) throws Throwable {
SearchParameters searchParameters = invocation.getArgumentAt(0, SearchParameters.class);
assertSame(parameters, searchParameters);
return searchResults;
}
doAnswer(invocation -> {
SearchParameters searchParameters = invocation.getArgumentAt(0, SearchParameters.class);
assertSame(parameters, searchParameters);
return searchResults;
}).when(this.searchOps).search(Matchers.any(SearchParameters.class));
this.gateway.handleMessage(new GenericMessage<SearchParameters>(parameters));
Message<?> reply = this.outputChannel.receive(0);
@@ -188,16 +170,12 @@ public class TwitterSearchOutboundGatewayTests {
Tweet tweet = mock(Tweet.class);
SearchMetadata searchMetadata = mock(SearchMetadata.class);
final SearchResults searchResults = new SearchResults(Collections.singletonList(tweet), searchMetadata);
doAnswer(new Answer<SearchResults>() {
@Override
public SearchResults answer(InvocationOnMock invocation) throws Throwable {
SearchParameters searchParameters = invocation.getArgumentAt(0, SearchParameters.class);
assertEquals("foobar", searchParameters.getQuery());
assertEquals(Integer.valueOf(5), searchParameters.getCount());
assertEquals(Long.valueOf(11), searchParameters.getSinceId());
return searchResults;
}
doAnswer(invocation -> {
SearchParameters searchParameters = invocation.getArgumentAt(0, SearchParameters.class);
assertEquals("foobar", searchParameters.getQuery());
assertEquals(Integer.valueOf(5), searchParameters.getCount());
assertEquals(Long.valueOf(11), searchParameters.getSinceId());
return searchResults;
}).when(this.searchOps).search(Matchers.any(SearchParameters.class));
this.gateway.handleMessage(new GenericMessage<String>("bar"));
Message<?> reply = this.outputChannel.receive(0);
@@ -214,15 +192,11 @@ public class TwitterSearchOutboundGatewayTests {
SearchMetadata searchMetadata = mock(SearchMetadata.class);
List<Tweet> empty = new ArrayList<Tweet>(0);
final SearchResults searchResults = new SearchResults(empty, searchMetadata);
doAnswer(new Answer<SearchResults>() {
@Override
public SearchResults answer(InvocationOnMock invocation) throws Throwable {
SearchParameters searchParameters = invocation.getArgumentAt(0, SearchParameters.class);
assertEquals("foo", searchParameters.getQuery());
assertEquals(Integer.valueOf(20), searchParameters.getCount());
return searchResults;
}
doAnswer(invocation -> {
SearchParameters searchParameters = invocation.getArgumentAt(0, SearchParameters.class);
assertEquals("foo", searchParameters.getQuery());
assertEquals(Integer.valueOf(20), searchParameters.getCount());
return searchResults;
}).when(this.searchOps).search(Matchers.any(SearchParameters.class));
this.gateway.handleMessage(new GenericMessage<String>("foo"));
Message<?> reply = this.outputChannel.receive(0);

View File

@@ -204,7 +204,7 @@ public final class ClientWebSocketContainer extends IntegrationWebSocketContaine
private final boolean syncClientLifecycle;
private IntegrationWebSocketConnectionManager(WebSocketClient client, String uriTemplate,
IntegrationWebSocketConnectionManager(WebSocketClient client, String uriTemplate,
Object... uriVariables) {
super(uriTemplate, uriVariables);
this.client = client;

View File

@@ -148,6 +148,10 @@ public abstract class IntegrationWebSocketContainer implements DisposableBean {
*/
private class IntegrationWebSocketHandler implements WebSocketHandler, SubProtocolCapable {
IntegrationWebSocketHandler() {
super();
}
@Override
public List<String> getSubProtocols() {
return IntegrationWebSocketContainer.this.getSubProtocols();

View File

@@ -37,9 +37,7 @@ import org.springframework.integration.websocket.support.PassThruSubProtocolHand
import org.springframework.integration.websocket.support.SubProtocolHandlerRegistry;
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.converter.ByteArrayMessageConverter;
import org.springframework.messaging.converter.CompositeMessageConverter;
import org.springframework.messaging.converter.DefaultContentTypeResolver;
@@ -119,18 +117,13 @@ public class WebSocketInboundChannelAdapter extends MessageProducerSupport
this.webSocketContainer = webSocketContainer;
this.server = this.webSocketContainer instanceof ServerWebSocketContainer;
this.subProtocolHandlerRegistry = protocolHandlerRegistry;
this.subProtocolHandlerChannel = new FixedSubscriberChannel(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
try {
handleMessageAndSend(message);
}
catch (Exception e) {
throw new MessageHandlingException(message, e);
}
this.subProtocolHandlerChannel = new FixedSubscriberChannel(message -> {
try {
handleMessageAndSend(message);
}
catch (Exception e) {
throw new MessageHandlingException(message, e);
}
});
}

View File

@@ -489,23 +489,17 @@ public class StompIntegrationTests extends LogAdjustingTestSupport {
//TODO SimpleBrokerMessageHandler doesn't support RECEIPT frame, hence we emulate it this way
@Bean
@SuppressWarnings("unchecked")
public ApplicationListener<SessionSubscribeEvent> webSocketEventListener(
final AbstractSubscribableChannel clientOutboundChannel) {
return new ApplicationListener<SessionSubscribeEvent>() {
@Override
public void onApplicationEvent(SessionSubscribeEvent event) {
Message<byte[]> message = event.getMessage();
StompHeaderAccessor stompHeaderAccessor = StompHeaderAccessor.wrap(message);
if (stompHeaderAccessor.getReceipt() != null) {
stompHeaderAccessor.setHeader("stompCommand", StompCommand.RECEIPT);
stompHeaderAccessor.setReceiptId(stompHeaderAccessor.getReceipt());
clientOutboundChannel.send(
MessageBuilder.createMessage(new byte[0], stompHeaderAccessor.getMessageHeaders()));
}
return event -> {
Message<byte[]> message = event.getMessage();
StompHeaderAccessor stompHeaderAccessor = StompHeaderAccessor.wrap(message);
if (stompHeaderAccessor.getReceipt() != null) {
stompHeaderAccessor.setHeader("stompCommand", StompCommand.RECEIPT);
stompHeaderAccessor.setReceiptId(stompHeaderAccessor.getReceipt());
clientOutboundChannel.send(
MessageBuilder.createMessage(new byte[0], stompHeaderAccessor.getMessageHeaders()));
}
};
}

View File

@@ -318,6 +318,10 @@ public class WebSocketServerTests {
private ApplicationEventPublisher applicationEventPublisher;
TestWebSocketHandlerDecoratorFactory() {
super();
}
@Override
public WebSocketHandler decorate(WebSocketHandler handler) {
return new TestWebSocketHandler(handler);

View File

@@ -119,7 +119,7 @@ public class MarshallingWebServiceOutboundGateway extends AbstractWebServiceOutb
private final class MarshallingRequestMessageCallback extends RequestMessageCallback {
private MarshallingRequestMessageCallback(WebServiceMessageCallback requestCallback,
MarshallingRequestMessageCallback(WebServiceMessageCallback requestCallback,
Message<?> requestMessage) {
super(requestCallback, requestMessage);
}
@@ -132,9 +132,14 @@ public class MarshallingWebServiceOutboundGateway extends AbstractWebServiceOutb
private class MarshallingResponseMessageExtractor extends ResponseMessageExtractor {
MarshallingResponseMessageExtractor() {
super();
}
@Override
public Object doExtractData(WebServiceMessage message) throws IOException {
return MarshallingUtils.unmarshal(MarshallingWebServiceOutboundGateway.this.unmarshaller, message);
}
}
}

View File

@@ -87,6 +87,11 @@ public class SimpleWebServiceInboundGateway extends AbstractWebServiceInboundGat
private static class TransformerSupportDelegate extends TransformerObjectSupport {
TransformerSupportDelegate() {
super();
}
void transformSourceToResult(Source source, Result result) throws TransformerException {
this.transform(source, result);
}

View File

@@ -102,7 +102,7 @@ public class SimpleWebServiceOutboundGateway extends AbstractWebServiceOutboundG
private final class SimpleRequestMessageCallback extends RequestMessageCallback {
private SimpleRequestMessageCallback(WebServiceMessageCallback requestCallback, Message<?> requestMessage) {
SimpleRequestMessageCallback(WebServiceMessageCallback requestCallback, Message<?> requestMessage) {
super(requestCallback, requestMessage);
}
@@ -145,7 +145,7 @@ public class SimpleWebServiceOutboundGateway extends AbstractWebServiceOutboundG
private final Result result;
private SimpleResponseMessageExtractor(Result result) {
SimpleResponseMessageExtractor(Result result) {
super();
this.result = result;
}
@@ -175,6 +175,10 @@ public class SimpleWebServiceOutboundGateway extends AbstractWebServiceOutboundG
private static class DefaultSourceExtractor extends TransformerObjectSupport implements SourceExtractor<DOMSource> {
DefaultSourceExtractor() {
super();
}
@Override
public DOMSource extractData(Source source) throws IOException, TransformerException {
if (source instanceof DOMSource) {

View File

@@ -39,7 +39,6 @@ import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mock;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.runners.MockitoJUnitRunner;
import org.mockito.stubbing.Answer;
@@ -131,14 +130,9 @@ public class SimpleWebServiceInboundGatewayTests {
}
private Answer<Boolean> withReplyTo(final MessageChannel replyChannel) {
return new Answer<Boolean>() {
@Override
public Boolean answer(InvocationOnMock invocation) throws Throwable {
replyChannel.send((Message<?>) invocation.getArguments()[0]);
return true;
}
return invocation -> {
replyChannel.send((Message<?>) invocation.getArguments()[0]);
return true;
};
}

View File

@@ -22,17 +22,12 @@ import static org.junit.Assert.assertThat;
import static org.mockito.Mockito.mock;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.net.URI;
import java.util.concurrent.atomic.AtomicReference;
import javax.xml.transform.TransformerException;
import org.hamcrest.Matchers;
import org.junit.Test;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.context.support.ClassPathXmlApplicationContext;
@@ -42,9 +37,7 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.ws.WebServiceMessage;
import org.springframework.ws.WebServiceMessageFactory;
import org.springframework.ws.client.core.WebServiceMessageCallback;
import org.springframework.ws.client.support.destination.DestinationProvider;
import org.springframework.ws.soap.SoapMessage;
import org.springframework.ws.transport.WebServiceConnection;
@@ -76,12 +69,9 @@ public class SimpleWebServiceOutboundGatewayTests {
String uri = "http://www.example.org";
SimpleWebServiceOutboundGateway gateway = new SimpleWebServiceOutboundGateway(new TestDestinationProvider(uri));
final AtomicReference<String> soapActionFromCallback = new AtomicReference<String>();
gateway.setRequestCallback(new WebServiceMessageCallback() {
@Override
public void doWithMessage(WebServiceMessage message) throws IOException, TransformerException {
SoapMessage soapMessage = (SoapMessage) message;
soapActionFromCallback.set(soapMessage.getSoapAction());
}
gateway.setRequestCallback(message -> {
SoapMessage soapMessage = (SoapMessage) message;
soapActionFromCallback.set(soapMessage.getSoapAction());
});
gateway.setBeanFactory(mock(BeanFactory.class));
gateway.afterPropertiesSet();
@@ -127,13 +117,10 @@ public class SimpleWebServiceOutboundGatewayTests {
Mockito.when(messageSender.createConnection(Mockito.any(URI.class))).thenReturn(wsConnection);
Mockito.when(messageSender.supports(Mockito.any(URI.class))).thenReturn(true);
Mockito.doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Exception {
Object[] args = invocation.getArguments();
WebServiceMessageFactory factory = (WebServiceMessageFactory) args[0];
return factory.createWebServiceMessage(new ByteArrayInputStream(mockResponseMessage.getBytes()));
}
Mockito.doAnswer(invocation -> {
Object[] args = invocation.getArguments();
WebServiceMessageFactory factory = (WebServiceMessageFactory) args[0];
return factory.createWebServiceMessage(new ByteArrayInputStream(mockResponseMessage.getBytes()));
}).when(wsConnection).receive(Mockito.any(WebServiceMessageFactory.class));
return messageSender;

View File

@@ -19,6 +19,7 @@ package org.springframework.integration.ws.config;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.doAnswer;
import java.io.IOException;
import java.net.URI;
@@ -42,8 +43,6 @@ import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.ArgumentCaptor;
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;
@@ -116,12 +115,9 @@ public class UriVariableTests {
WebServiceTemplate webServiceTemplate = TestUtils.getPropertyValue(this.httpOutboundGateway, "webServiceTemplate", WebServiceTemplate.class);
webServiceTemplate = Mockito.spy(webServiceTemplate);
final AtomicReference<String> uri = new AtomicReference<String>();
Mockito.doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
uri.set(invocation.getArgumentAt(0, String.class));
throw new WebServiceIOException("intentional");
}
doAnswer(invocation -> {
uri.set(invocation.getArgumentAt(0, String.class));
throw new WebServiceIOException("intentional");
}).when(webServiceTemplate)
.sendAndReceive(Mockito.anyString(),
Mockito.any(WebServiceMessageCallback.class),
@@ -235,10 +231,15 @@ public class UriVariableTests {
private volatile URI lastUri;
TestClientInterceptor() {
super();
}
public URI getLastUri() {
return this.lastUri;
}
@Override
public boolean handleRequest(MessageContext messageContext) throws WebServiceClientException {
TransportContext tc = TransportContextHolder.getTransportContext();
if (tc != null) {
@@ -255,14 +256,17 @@ public class UriVariableTests {
return true;
}
@Override
public boolean handleResponse(MessageContext messageContext) throws WebServiceClientException {
return false;
}
@Override
public boolean handleFault(MessageContext messageContext) throws WebServiceClientException {
return false;
}
@Override
public void afterCompletion(MessageContext messageContext, Exception ex) throws WebServiceClientException {
}

View File

@@ -35,8 +35,6 @@ import javax.xml.transform.dom.DOMSource;
import org.junit.Test;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.w3c.dom.Document;
import org.w3c.dom.Element;
@@ -83,7 +81,6 @@ public class WebServiceOutboundGatewayWithHeaderMapperTests {
String responseNonSoapMessage = "<?xml version=\"1.0\" encoding=\"UTF-8\" ?> " +
"<person><name>oleg</name></person>";
@SuppressWarnings("unchecked")
@Test
public void headerMapperParserTest() throws Exception {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
@@ -184,7 +181,7 @@ public class WebServiceOutboundGatewayWithHeaderMapperTests {
assertNull(replyMessage.getHeaders().get("baz"));
}
@SuppressWarnings("rawtypes")
@SuppressWarnings({ "resource" })
public Message<?> process(Object payload, String gatewayName, String channelName, final boolean soap) throws Exception {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"ws-outbound-gateway-with-headermappers.xml", this.getClass());
@@ -201,12 +198,9 @@ public class WebServiceOutboundGatewayWithHeaderMapperTests {
Mockito.when(messageSender.createConnection(Mockito.any(URI.class))).thenReturn(wsConnection);
Mockito.when(messageSender.supports(Mockito.any(URI.class))).thenReturn(true);
Mockito.doAnswer(new Answer() {
@Override
public Object answer(InvocationOnMock invocation) {
Object[] args = invocation.getArguments();
WebServiceMessage wsMessage = (WebServiceMessage) args[0];
Mockito.doAnswer(invocation -> {
Object[] args = invocation.getArguments();
WebServiceMessage wsMessage = (WebServiceMessage) args[0];
// try { // uncomment if you want to see a pretty-print of SOAP message
// Transformer transformer = TransformerFactory.newInstance().newTransformer();
// transformer.setOutputProperty(OutputKeys.INDENT, "yes");
@@ -215,32 +209,28 @@ public class WebServiceOutboundGatewayWithHeaderMapperTests {
// catch (Exception e) {
// // ignore
// }
if (soap) {
SoapHeader soapHeader = ((SoapMessage) wsMessage).getSoapHeader();
assertNotNull(soapHeader.getAttributeValue(QNameUtils.parseQNameString("foo")));
assertNotNull(soapHeader.getAttributeValue(QNameUtils.parseQNameString("foobar")));
assertNotNull(soapHeader.getAttributeValue(QNameUtils.parseQNameString("abaz")));
assertNull(soapHeader.getAttributeValue(QNameUtils.parseQNameString("bar")));
}
return null;
if (soap) {
SoapHeader soapHeader = ((SoapMessage) wsMessage).getSoapHeader();
assertNotNull(soapHeader.getAttributeValue(QNameUtils.parseQNameString("foo")));
assertNotNull(soapHeader.getAttributeValue(QNameUtils.parseQNameString("foobar")));
assertNotNull(soapHeader.getAttributeValue(QNameUtils.parseQNameString("abaz")));
assertNull(soapHeader.getAttributeValue(QNameUtils.parseQNameString("bar")));
}
return null;
}).when(wsConnection).send(Mockito.any(WebServiceMessage.class));
Mockito.doAnswer(new Answer() {
Mockito.doAnswer(invocation -> {
Object[] args = invocation.getArguments();
WebServiceMessageFactory factory = (WebServiceMessageFactory) args[0];
String responseMessage = factory instanceof SoapMessageFactory ? responseSoapMessage
: responseNonSoapMessage;
WebServiceMessage wsMessage = factory
.createWebServiceMessage(new ByteArrayInputStream(responseMessage.getBytes()));
if (soap) {
@Override
public Object answer(InvocationOnMock invocation) throws Exception {
Object[] args = invocation.getArguments();
WebServiceMessageFactory factory = (WebServiceMessageFactory) args[0];
String responseMessage = factory instanceof SoapMessageFactory ? responseSoapMessage
: responseNonSoapMessage;
WebServiceMessage wsMessage = factory
.createWebServiceMessage(new ByteArrayInputStream(responseMessage.getBytes()));
if (soap) {
((SoapMessage) wsMessage).getSoapHeader().addAttribute(QNameUtils.parseQNameString("bar"), "bar");
((SoapMessage) wsMessage).getSoapHeader().addAttribute(QNameUtils.parseQNameString("baz"), "baz");
}
((SoapMessage) wsMessage).getSoapHeader().addAttribute(QNameUtils.parseQNameString("bar"), "bar");
((SoapMessage) wsMessage).getSoapHeader().addAttribute(QNameUtils.parseQNameString("baz"), "baz");
}
// try { // uncomment if you want to see a pretty-print of SOAP message
// Transformer transformer = TransformerFactory.newInstance().newTransformer();
@@ -250,8 +240,7 @@ public class WebServiceOutboundGatewayWithHeaderMapperTests {
// catch (Exception e) {
// // ignore
// }
return wsMessage;
}
return wsMessage;
}).when(wsConnection).receive(Mockito.any(WebServiceMessageFactory.class));
gateway.setMessageSender(messageSender);

View File

@@ -130,6 +130,10 @@ public class XPathRouter extends AbstractMappingMessageRouter {
private static class TextContentNodeMapper implements NodeMapper<Object> {
TextContentNodeMapper() {
super();
}
@Override
public Object mapNode(Node node, int nodeNum) throws DOMException {
return node.getTextContent();

View File

@@ -271,7 +271,7 @@ public class XPathMessageSplitter extends AbstractMessageSplitter {
private int index;
private NodeListIterator(NodeList nodeList) throws ParserConfigurationException {
NodeListIterator(NodeList nodeList) throws ParserConfigurationException {
this.nodeList = nodeList;
if (XPathMessageSplitter.this.createDocuments) {
this.documentBuilder = getNewDocumentBuilder();

View File

@@ -185,6 +185,10 @@ public class DefaultXmlPayloadConverterTests {
private static class MySource implements Source {
MySource() {
super();
}
@Override
public void setSystemId(String systemId) {
}

View File

@@ -28,9 +28,9 @@ import javax.xml.transform.dom.DOMResult;
import org.junit.Test;
import org.springframework.integration.xml.result.StringResultFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.integration.xml.result.StringResultFactory;
import org.springframework.oxm.Marshaller;
import org.springframework.oxm.XmlMappingException;
import org.springframework.xml.transform.StringResult;
@@ -81,10 +81,16 @@ public class MarshallingTransformerTests {
private final List<Object> payloads = new ArrayList<Object>();
TestMarshaller() {
super();
}
@Override
public boolean supports(Class<?> clazz) {
return true;
}
@Override
public void marshal(Object source, Result result) throws XmlMappingException, IOException {
if (source instanceof Message) {
this.messages.add((Message<?>) source);

View File

@@ -32,10 +32,10 @@ import org.w3c.dom.Document;
import org.w3c.dom.Node;
import org.xml.sax.InputSource;
import org.springframework.messaging.Message;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.xml.XmlPayloadConverter;
import org.springframework.integration.xml.xpath.XPathEvaluationType;
import org.springframework.messaging.Message;
import org.springframework.xml.xpath.NodeMapper;
import org.springframework.xml.xpath.XPathExpression;
import org.springframework.xml.xpath.XPathExpressionFactory;
@@ -139,6 +139,11 @@ public class XPathTransformerTests {
private static class TestNodeMapper implements NodeMapper<Object> {
TestNodeMapper() {
super();
}
@Override
public Object mapNode(Node node, int nodeNum) throws DOMException {
return node.getTextContent() + "-mapped";
}
@@ -147,10 +152,16 @@ public class XPathTransformerTests {
private static class TestXmlPayloadConverter implements XmlPayloadConverter {
TestXmlPayloadConverter() {
super();
}
@Override
public Source convertToSource(Object object) {
throw new UnsupportedOperationException();
}
@Override
public Node convertToNode(Object object) {
try {
return DocumentBuilderFactory.newInstance().newDocumentBuilder().parse(
@@ -161,6 +172,7 @@ public class XPathTransformerTests {
}
}
@Override
public Document convertToDocument(Object object) {
throw new UnsupportedOperationException();
}

View File

@@ -212,6 +212,10 @@ public class XmppConnectionFactoryBean extends AbstractFactoryBean<XMPPConnectio
private class LoggingConnectionListener implements ConnectionListener {
LoggingConnectionListener() {
super();
}
@Override
public void reconnectionSuccessful() {
logger.debug("Reconnection successful");

View File

@@ -116,6 +116,10 @@ public class ChatMessageListeningEndpoint extends AbstractXmppConnectionAwareEnd
private class ChatMessagePublishingStanzaListener implements StanzaListener {
ChatMessagePublishingStanzaListener() {
super();
}
@Override
public void processPacket(final Stanza packet) {
if (packet instanceof org.jivesoftware.smack.packet.Message) {

View File

@@ -20,10 +20,10 @@ import java.util.Collection;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.jivesoftware.smack.roster.Roster;
import org.jivesoftware.smack.roster.RosterListener;
import org.jivesoftware.smack.XMPPConnection;
import org.jivesoftware.smack.packet.Presence;
import org.jivesoftware.smack.roster.Roster;
import org.jivesoftware.smack.roster.RosterListener;
import org.springframework.integration.xmpp.core.AbstractXmppConnectionAwareEndpoint;
import org.springframework.messaging.Message;
@@ -85,24 +85,32 @@ public class PresenceListeningEndpoint extends AbstractXmppConnectionAwareEndpoi
*/
private class PresencePublishingRosterListener implements RosterListener {
PresencePublishingRosterListener() {
super();
}
@Override
public void entriesAdded(Collection<String> entries) {
if (logger.isDebugEnabled()) {
logger.debug("entries added: " + StringUtils.collectionToCommaDelimitedString(entries));
}
}
@Override
public void entriesUpdated(Collection<String> entries) {
if (logger.isDebugEnabled()) {
logger.debug("entries updated: " + StringUtils.collectionToCommaDelimitedString(entries));
}
}
@Override
public void entriesDeleted(Collection<String> entries) {
if (logger.isDebugEnabled()) {
logger.debug("entries deleted: " + StringUtils.collectionToCommaDelimitedString(entries));
}
}
@Override
public void presenceChanged(Presence presence) {
if (presence != null) {
if (logger.isDebugEnabled()) {

View File

@@ -20,6 +20,7 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
@@ -29,8 +30,6 @@ import org.jivesoftware.smackx.jiveproperties.JivePropertiesManager;
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.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
@@ -96,7 +95,6 @@ public class ChatMessageOutboundChannelAdapterParserTests {
assertEquals(1, adviceCalled);
}
@SuppressWarnings("unchecked")
@Test
public void testEventConsumer() {
Object eventConsumer = context.getBean("outboundEventAdapter");
@@ -130,17 +128,13 @@ public class ChatMessageOutboundChannelAdapterParserTests {
setHeader("foobar", "foobar").build();
XMPPConnection connection = context.getBean("testConnection", XMPPConnection.class);
Mockito.doAnswer(new Answer() {
public Object answer(InvocationOnMock invocation) {
Object[] args = invocation.getArguments();
org.jivesoftware.smack.packet.Message xmppMessage = (org.jivesoftware.smack.packet.Message) args[0];
assertEquals("oleg", xmppMessage.getTo());
assertEquals("foobar", JivePropertiesManager.getProperty(xmppMessage, "foobar"));
return null;
}
})
.when(connection).sendStanza(Mockito.any(org.jivesoftware.smack.packet.Message.class));
doAnswer(invocation -> {
Object[] args = invocation.getArguments();
org.jivesoftware.smack.packet.Message xmppMessage = (org.jivesoftware.smack.packet.Message) args[0];
assertEquals("oleg", xmppMessage.getTo());
assertEquals("foobar", JivePropertiesManager.getProperty(xmppMessage, "foobar"));
return null;
}).when(connection).sendStanza(Mockito.any(org.jivesoftware.smack.packet.Message.class));
channel.send(message);
@@ -148,22 +142,17 @@ public class ChatMessageOutboundChannelAdapterParserTests {
Mockito.reset(connection);
}
@SuppressWarnings("rawtypes")
@Test //INT-2275
public void testOutboundChannelAdapterInsideChain() throws Exception {
MessageChannel channel = context.getBean("outboundChainChannel", MessageChannel.class);
Message<?> message = MessageBuilder.withPayload("hello").setHeader(XmppHeaders.TO, "artem").build();
XMPPConnection connection = context.getBean("testConnection", XMPPConnection.class);
Mockito.doAnswer(new Answer() {
public Object answer(InvocationOnMock invocation) {
Object[] args = invocation.getArguments();
org.jivesoftware.smack.packet.Message xmppMessage = (org.jivesoftware.smack.packet.Message) args[0];
assertEquals("artem", xmppMessage.getTo());
assertEquals("hello", xmppMessage.getBody());
return null;
}
doAnswer(invocation -> {
Object[] args = invocation.getArguments();
org.jivesoftware.smack.packet.Message xmppMessage = (org.jivesoftware.smack.packet.Message) args[0];
assertEquals("artem", xmppMessage.getTo());
assertEquals("hello", xmppMessage.getBody());
return null;
}).when(connection).sendStanza(Mockito.any(org.jivesoftware.smack.packet.Message.class));
channel.send(message);

View File

@@ -26,15 +26,14 @@ import static org.mockito.Mockito.verify;
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.factory.annotation.Value;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.xmpp.XmppHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -59,14 +58,12 @@ public class XmppHeaderEnricherParserTests {
public void to() {
MessagingTemplate messagingTemplate = new MessagingTemplate();
MessageHandler handler = mock(MessageHandler.class);
doAnswer(new Answer() {
public Object answer(InvocationOnMock invocation) throws Throwable {
Message message = invocation.getArgumentAt(0, Message.class);
String chatToUser = (String) message.getHeaders().get(XmppHeaders.TO);
assertNotNull(chatToUser);
assertEquals("test1@example.org", chatToUser);
return null;
}
doAnswer(invocation -> {
Message message = invocation.getArgumentAt(0, Message.class);
String chatToUser = (String) message.getHeaders().get(XmppHeaders.TO);
assertNotNull(chatToUser);
assertEquals("test1@example.org", chatToUser);
return null;
}).when(handler).handleMessage(Mockito.any(Message.class));
output.subscribe(handler);
messagingTemplate.send(input, MessageBuilder.withPayload("foo").build());

View File

@@ -45,8 +45,6 @@ import org.jivesoftware.smackx.gcm.packet.GcmPacketExtension;
import org.junit.Test;
import org.mockito.ArgumentCaptor;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.xmlpull.v1.XmlPullParser;
import org.springframework.beans.DirectFieldAccessor;
@@ -58,7 +56,6 @@ import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.integration.xmpp.core.XmppContextUtils;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.ErrorMessage;
@@ -81,25 +78,15 @@ public class ChatMessageListeningEndpointTests {
XMPPConnection connection = mock(XMPPConnection.class);
ChatMessageListeningEndpoint endpoint = new ChatMessageListeningEndpoint(connection);
willAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
packetListSet.add(invocation.getArgumentAt(0, StanzaListener.class));
return null;
}
willAnswer(invocation -> {
packetListSet.add(invocation.getArgumentAt(0, StanzaListener.class));
return null;
}).given(connection)
.addAsyncStanzaListener(Mockito.any(StanzaListener.class), Mockito.any(StanzaFilter.class));
willAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
packetListSet.remove(invocation.getArguments()[0]);
return null;
}
willAnswer(invocation -> {
packetListSet.remove(invocation.getArguments()[0]);
return null;
}).given(connection)
.removeAsyncStanzaListener(Mockito.any(StanzaListener.class));
@@ -146,14 +133,8 @@ public class ChatMessageListeningEndpointTests {
ChatMessageListeningEndpoint endpoint = new ChatMessageListeningEndpoint();
DirectChannel outChannel = new DirectChannel();
outChannel.subscribe(new MessageHandler() {
@Override
public void handleMessage(org.springframework.messaging.Message<?> message)
throws MessagingException {
throw new RuntimeException("ooops");
}
outChannel.subscribe(message -> {
throw new RuntimeException("ooops");
});
PollableChannel errorChannel = new QueueChannel();
endpoint.setBeanFactory(bf);
@@ -203,15 +184,10 @@ public class ChatMessageListeningEndpointTests {
Log logger = Mockito.spy(TestUtils.getPropertyValue(endpoint, "logger", Log.class));
given(logger.isInfoEnabled()).willReturn(true);
final CountDownLatch logLatch = new CountDownLatch(1);
willAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
Object result = invocation.callRealMethod();
logLatch.countDown();
return result;
}
willAnswer(invocation -> {
Object result = invocation.callRealMethod();
logLatch.countDown();
return result;
}).given(logger).info(anyString());
new DirectFieldAccessor(endpoint).setPropertyValue("logger", logger);
@@ -276,7 +252,7 @@ public class ChatMessageListeningEndpointTests {
private static class TestXMPPConnection extends XMPPTCPConnection {
private TestXMPPConnection() {
TestXMPPConnection() {
super(null);
}

View File

@@ -19,6 +19,7 @@ package org.springframework.integration.xmpp.inbound;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertSame;
import static org.mockito.Matchers.any;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
@@ -33,9 +34,6 @@ import org.jivesoftware.smack.packet.Presence.Type;
import org.jivesoftware.smack.roster.Roster;
import org.jivesoftware.smack.roster.RosterListener;
import org.junit.Test;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
@@ -44,7 +42,6 @@ import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.integration.xmpp.core.XmppContextUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.ErrorMessage;
@@ -66,25 +63,15 @@ public class PresenceListeningEndpointTests {
Map<XMPPConnection, Roster> instances = TestUtils.getPropertyValue(roster, "INSTANCES", Map.class);
instances.put(connection, roster);
doAnswer(new Answer<Object>() {
doAnswer(invocation -> {
rosterSet.add(invocation.getArgumentAt(0, RosterListener.class));
return null;
}).when(roster).addRosterListener(any(RosterListener.class));
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
rosterSet.add(invocation.getArgumentAt(0, RosterListener.class));
return null;
}
}).when(roster).addRosterListener(Mockito.any(RosterListener.class));
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
rosterSet.remove(invocation.getArgumentAt(0, RosterListener.class));
return null;
}
}).when(roster).removeRosterListener(Mockito.any(RosterListener.class));
doAnswer(invocation -> {
rosterSet.remove(invocation.getArgumentAt(0, RosterListener.class));
return null;
}).when(roster).removeRosterListener(any(RosterListener.class));
PresenceListeningEndpoint rosterEndpoint = new PresenceListeningEndpoint(connection);
rosterEndpoint.setOutputChannel(new QueueChannel());
rosterEndpoint.setBeanFactory(mock(BeanFactory.class));
@@ -149,14 +136,8 @@ public class PresenceListeningEndpointTests {
PresenceListeningEndpoint endpoint = new PresenceListeningEndpoint();
DirectChannel outChannel = new DirectChannel();
outChannel.subscribe(new MessageHandler() {
@Override
public void handleMessage(org.springframework.messaging.Message<?> message)
throws MessagingException {
throw new RuntimeException("ooops");
}
outChannel.subscribe(message -> {
throw new RuntimeException("ooops");
});
PollableChannel errorChannel = new QueueChannel();
endpoint.setBeanFactory(bf);

View File

@@ -252,6 +252,10 @@ public class LeaderInitiator implements SmartLifecycle {
*/
private class CuratorContext implements Context {
CuratorContext() {
super();
}
@Override
public boolean isLeader() {
return LeaderInitiator.this.leaderSelector.hasLeadership();

View File

@@ -178,6 +178,7 @@ public class ZookeeperLockRegistry implements ExpirableLockRegistry, DisposableB
* Zookeeper path.
*
*/
@FunctionalInterface
public interface KeyToPathStrategy {
/**
@@ -191,7 +192,9 @@ public class ZookeeperLockRegistry implements ExpirableLockRegistry, DisposableB
* @return true if this strategy returns a bounded number of locks, removing
* the need for removing LRU locks.
*/
boolean bounded();
default boolean bounded() {
return true;
}
}
@@ -199,7 +202,7 @@ public class ZookeeperLockRegistry implements ExpirableLockRegistry, DisposableB
private final String root;
private DefaultKeyToPathStrategy(String rootPath) {
DefaultKeyToPathStrategy(String rootPath) {
Assert.notNull(rootPath, "'rootPath' cannot be null");
if (!rootPath.endsWith("/")) {
this.root = rootPath + "/";
@@ -233,7 +236,7 @@ public class ZookeeperLockRegistry implements ExpirableLockRegistry, DisposableB
private long lastUsed;
private ZkLock(CuratorFramework client, AsyncTaskExecutor mutexTaskExecutor, String path) {
ZkLock(CuratorFramework client, AsyncTaskExecutor mutexTaskExecutor, String path) {
this.client = client;
this.mutex = new InterProcessMutex(client, path);
this.mutexTaskExecutor = mutexTaskExecutor;

View File

@@ -334,7 +334,7 @@ public class ZookeeperMetadataStore implements ListenableMetadataStore, SmartLif
private final int version;
private LocalChildData(String value, int version) {
LocalChildData(String value, int version) {
this.value = value;
this.version = version;
}
@@ -351,6 +351,10 @@ public class ZookeeperMetadataStore implements ListenableMetadataStore, SmartLif
private class MetadataStoreListenerInvokingPathChildrenCacheListener implements PathChildrenCacheListener {
MetadataStoreListenerInvokingPathChildrenCacheListener() {
super();
}
@Override
public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
synchronized (ZookeeperMetadataStore.this.updateMap) {

View File

@@ -108,15 +108,10 @@ public class LeaderInitiatorFactoryBeanTests extends ZookeeperTestSupport {
@Bean
public ApplicationListener<?> listener() {
return new ApplicationListener<AbstractLeaderEvent>() {
@Override
public void onApplicationEvent(AbstractLeaderEvent event) {
events.add(event);
latch1.countDown();
latch2.countDown();
}
return event -> {
events.add((AbstractLeaderEvent) event);
latch1.countDown();
latch2.countDown();
};
}

View File

@@ -27,7 +27,6 @@ import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
@@ -39,7 +38,6 @@ import org.junit.Test;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.integration.zookeeper.ZookeeperTestSupport;
import org.springframework.integration.zookeeper.lock.ZookeeperLockRegistry.KeyToPathStrategy;
import org.springframework.messaging.MessagingException;
/**
@@ -150,21 +148,17 @@ public class ZkLockRegistryTests extends ZookeeperTestSupport {
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 = registry.obtain("foo");
locked.set(lock2.tryLock(200, TimeUnit.MILLISECONDS));
latch.countDown();
try {
lock2.unlock();
}
catch (MessagingException e) {
return e.getCause();
}
return null;
Future<Object> result = Executors.newSingleThreadExecutor().submit(() -> {
Lock lock2 = registry.obtain("foo");
locked.set(lock2.tryLock(200, TimeUnit.MILLISECONDS));
latch.countDown();
try {
lock2.unlock();
}
catch (MessagingException e) {
return e.getCause();
}
return null;
});
assertTrue(latch.await(10, TimeUnit.SECONDS));
assertFalse(locked.get());
@@ -184,24 +178,20 @@ public class ZkLockRegistryTests extends ZookeeperTestSupport {
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 = 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 = 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));
@@ -223,24 +213,20 @@ public class ZkLockRegistryTests extends ZookeeperTestSupport {
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));
@@ -260,19 +246,15 @@ public class ZkLockRegistryTests extends ZookeeperTestSupport {
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.getCause();
}
return null;
Future<Object> result = Executors.newSingleThreadExecutor().submit(() -> {
try {
lock.unlock();
}
catch (Exception e) {
latch.countDown();
return e.getCause();
}
return null;
});
assertTrue(latch.await(10, TimeUnit.SECONDS));
assertFalse(locked.get());
@@ -285,18 +267,8 @@ public class ZkLockRegistryTests extends ZookeeperTestSupport {
@Test
public void testLockWithBoundedStrategy() throws Exception {
ZookeeperLockRegistry registry = new ZookeeperLockRegistry(this.client, new KeyToPathStrategy() {
@Override
public String pathFor(String key) {
return "/SpringIntegration-LockRegistry/singleLock";
}
@Override
public boolean bounded() {
return true;
}
});
ZookeeperLockRegistry registry = new ZookeeperLockRegistry(this.client,
key -> "/SpringIntegration-LockRegistry/singleLock");
for (int i = 0; i < 10; i++) {
Lock lock = registry.obtain("foo");
lock.lock();