From 6e30173509793ed32af1462b4d10f430d5575603 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Wed, 9 Apr 2008 16:33:16 +0000 Subject: [PATCH] PollableSource no longer accepts a "limit" argument in its poll() method (INT-181). --- .../integration/adapter/file/FileSource.java | 17 ++-- .../adapter/ftp/DirectoryContentManager.java | 18 ++--- .../adapter/ftp/FtpSourceAdapter.java | 77 ++++++++++--------- .../adapter/jms/JmsPollableSource.java | 7 +- .../adapter/stream/ByteStreamSource.java | 51 ++++++------ .../adapter/stream/CharacterStreamSource.java | 31 +++----- .../adapter/MethodInvokingSource.java | 6 +- .../integration/adapter/PollableSource.java | 6 +- .../adapter/PollingSourceAdapter.java | 18 ++--- .../dispatcher/SynchronousChannel.java | 7 +- .../adapter/MethodInvokingSourceTests.java | 12 ++- .../adapter/PollingSourceAdapterTests.java | 38 ++++----- .../integration/bus/MessageBusTests.java | 2 +- .../dispatcher/SynchronousChannelTests.java | 10 +-- 14 files changed, 123 insertions(+), 177 deletions(-) diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/FileSource.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/FileSource.java index bfbc3b357e..95add15912 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/FileSource.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/FileSource.java @@ -19,9 +19,6 @@ package org.springframework.integration.adapter.file; import java.io.File; import java.io.FileFilter; import java.io.FilenameFilter; -import java.util.ArrayList; -import java.util.Collection; -import java.util.List; import org.springframework.integration.adapter.PollableSource; import org.springframework.integration.message.MessagingException; @@ -54,13 +51,13 @@ public class FileSource implements PollableSource { this.filenameFilter = filenameFilter; } - public Collection poll(int limit) { + public File poll() { File[] files = null; if (this.fileFilter != null) { - files = this.directory.listFiles(fileFilter); + files = this.directory.listFiles(this.fileFilter); } else if (this.filenameFilter != null) { - files = this.directory.listFiles(filenameFilter); + files = this.directory.listFiles(this.filenameFilter); } else { files = this.directory.listFiles(); @@ -69,14 +66,12 @@ public class FileSource implements PollableSource { throw new MessagingException("Problem occurred while polling for files. " + "Is '" + directory.getAbsolutePath() + "' a directory?"); } - List results = new ArrayList(); - int size = Math.min(limit, files.length); - for (int i = 0; i < size; i++) { + for (int i = 0; i < files.length; i++) { if (files[i].isFile()) { - results.add(files[i]); + return files[i]; } } - return results; + return null; } } diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/DirectoryContentManager.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/DirectoryContentManager.java index 52e30c951a..cf0419d3ba 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/DirectoryContentManager.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/DirectoryContentManager.java @@ -22,7 +22,7 @@ import java.util.Iterator; import java.util.Map; /** - * Tracks changes in the context. This implementation is thread-safe as it + * Tracks changes in a directory. This implementation is thread-safe as it * allows to synchronously process a new directory structure. * * @author Marius Bogoevici @@ -30,26 +30,26 @@ import java.util.Map; */ public class DirectoryContentManager { - private Map snapshot = new HashMap(); + private Map previousSnapshot = new HashMap(); private final Map backlog = new HashMap(); - public synchronized void processSnapshot(Map remoteSnapshot) { + public synchronized void processSnapshot(Map currentSnapshot) { Iterator> iter = this.backlog.entrySet().iterator(); while (iter.hasNext()) { String fileName = iter.next().getKey(); - if (!remoteSnapshot.containsKey(fileName)) { + if (!currentSnapshot.containsKey(fileName)) { iter.remove(); } } - for (String fileName : remoteSnapshot.keySet()) { - if (!this.snapshot.containsKey(fileName) - || (!this.snapshot.get(fileName).equals(remoteSnapshot.get(fileName)))) { - this.backlog.put(fileName, remoteSnapshot.get(fileName)); + for (String fileName : currentSnapshot.keySet()) { + if (!this.previousSnapshot.containsKey(fileName) + || (!this.previousSnapshot.get(fileName).equals(currentSnapshot.get(fileName)))) { + this.backlog.put(fileName, currentSnapshot.get(fileName)); } } - this.snapshot = new HashMap(remoteSnapshot); + this.previousSnapshot = new HashMap(currentSnapshot); } public synchronized void fileProcessed(String fileName) { diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/FtpSourceAdapter.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/FtpSourceAdapter.java index 25088a1cf0..a731b43b41 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/FtpSourceAdapter.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/FtpSourceAdapter.java @@ -19,9 +19,8 @@ package org.springframework.integration.adapter.ftp; import java.io.File; import java.io.FileOutputStream; import java.io.IOException; -import java.util.Collection; import java.util.HashMap; -import java.util.LinkedList; +import java.util.Map; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -124,50 +123,30 @@ public class FtpSourceAdapter extends PollingSourceAdapter implements Poll } - public final Collection poll(int limit) { + public final File poll() { try { - LinkedList localFileList = new LinkedList(); - this.client.connect(this.host, this.port); - if (!StringUtils.hasText(this.username)) { - throw new MessagingException("username is required"); - } - if (!this.client.login(this.username, this.password)) { - throw new MessagingException("Login failed. Please check the username and password."); - } - if (logger.isDebugEnabled()) { - logger.debug("login successful"); - } - this.client.setFileType(FTP.IMAGE_FILE_TYPE); - if (!this.remoteWorkingDirectory.equals(this.client.printWorkingDirectory()) - && !this.client.changeWorkingDirectory(this.remoteWorkingDirectory)) { - throw new MessagingException("Could not change directory to '" + - remoteWorkingDirectory + "'. Please check the path."); - } - if (logger.isDebugEnabled()) { - logger.debug("working directory is: " + this.client.printWorkingDirectory()); - } + this.establishConnection(); FTPFile[] fileList = this.client.listFiles(); HashMap snapshot = new HashMap(); for (FTPFile ftpFile : fileList) { - FileInfo fileInfo = new FileInfo(ftpFile.getName(), ftpFile.getTimestamp().getTimeInMillis(), - ftpFile.getSize()); + FileInfo fileInfo = new FileInfo( + ftpFile.getName(), ftpFile.getTimestamp().getTimeInMillis(), ftpFile.getSize()); snapshot.put(ftpFile.getName(), fileInfo); } this.directoryContentManager.processSnapshot(snapshot); - for (String fileName : this.directoryContentManager.getBacklog().keySet()) { - File file = new File(this.localWorkingDirectory, fileName); - if (file.exists()) { - file.delete(); - } - FileOutputStream fileOutputStream = new FileOutputStream(file); - this.client.retrieveFile(fileName, fileOutputStream); - fileOutputStream.close(); - localFileList.add(file); - if (limit >= localFileList.size()) { - break; - } + Map backlog = this.directoryContentManager.getBacklog(); + if (backlog.isEmpty()) { + return null; } - return localFileList; + String fileName = backlog.keySet().iterator().next(); + File file = new File(this.localWorkingDirectory, fileName); + if (file.exists()) { + file.delete(); + } + FileOutputStream fileOutputStream = new FileOutputStream(file); + this.client.retrieveFile(fileName, fileOutputStream); + fileOutputStream.close(); + return file; } catch (Exception e) { try { @@ -182,4 +161,26 @@ public class FtpSourceAdapter extends PollingSourceAdapter implements Poll } } + private void establishConnection() throws IOException { + if (!StringUtils.hasText(this.username)) { + throw new MessagingException("username is required"); + } + this.client.connect(this.host, this.port); + if (!this.client.login(this.username, this.password)) { + throw new MessagingException("Login failed. Please check the username and password."); + } + if (logger.isDebugEnabled()) { + logger.debug("login successful"); + } + this.client.setFileType(FTP.IMAGE_FILE_TYPE); + if (!this.remoteWorkingDirectory.equals(this.client.printWorkingDirectory()) + && !this.client.changeWorkingDirectory(this.remoteWorkingDirectory)) { + throw new MessagingException("Could not change directory to '" + + remoteWorkingDirectory + "'. Please check the path."); + } + if (logger.isDebugEnabled()) { + logger.debug("working directory is: " + this.client.printWorkingDirectory()); + } + } + } diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsPollableSource.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsPollableSource.java index e5fde50ddd..6f8cc4d83b 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsPollableSource.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsPollableSource.java @@ -16,9 +16,6 @@ package org.springframework.integration.adapter.jms; -import java.util.Arrays; -import java.util.Collection; - import javax.jms.ConnectionFactory; import javax.jms.Destination; @@ -52,8 +49,8 @@ public class JmsPollableSource extends AbstractJmsTemplateBasedAdapter implement } - public Collection poll(int limit) { - return Arrays.asList(this.getJmsTemplate().receiveAndConvert()); + public Object poll() { + return this.getJmsTemplate().receiveAndConvert(); } } diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/ByteStreamSource.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/ByteStreamSource.java index 29c9681cb6..e15ad84d99 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/ByteStreamSource.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/ByteStreamSource.java @@ -19,9 +19,6 @@ package org.springframework.integration.adapter.stream; import java.io.BufferedInputStream; import java.io.IOException; import java.io.InputStream; -import java.util.ArrayList; -import java.util.Collection; -import java.util.List; import org.springframework.integration.adapter.PollableSource; import org.springframework.integration.message.MessagingException; @@ -68,36 +65,32 @@ public class ByteStreamSource implements PollableSource { this.shouldTruncate = shouldTruncate; } - public Collection poll(int limit) { - List results = new ArrayList(); - while (results.size() < limit) { - try { - byte[] bytes; - int bytesRead = 0; - synchronized (this.streamMonitor) { - if (stream.available() == 0) { - return results; - } - bytes = new byte[bytesPerMessage]; - bytesRead = stream.read(bytes, 0, bytes.length); - } - if (bytesRead <= 0) { - return results; - } - if (!this.shouldTruncate) { - results.add(bytes); - } - else { - byte[] result = new byte[bytesRead]; - System.arraycopy(bytes, 0, result, 0, result.length); - results.add(result); + public byte[] poll() { + try { + byte[] bytes; + int bytesRead = 0; + synchronized (this.streamMonitor) { + if (stream.available() == 0) { + return null; } + bytes = new byte[bytesPerMessage]; + bytesRead = stream.read(bytes, 0, bytes.length); } - catch (IOException e) { - throw new MessagingException("IO failure occurred in adapter", e); + if (bytesRead <= 0) { + return null; + } + if (!this.shouldTruncate) { + return bytes; + } + else { + byte[] result = new byte[bytesRead]; + System.arraycopy(bytes, 0, result, 0, result.length); + return result; } } - return results; + catch (IOException e) { + throw new MessagingException("IO failure occurred in adapter", e); + } } } diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSource.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSource.java index 5c66130cf8..8f38e9d6f4 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSource.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSource.java @@ -19,9 +19,6 @@ package org.springframework.integration.adapter.stream; import java.io.BufferedReader; import java.io.IOException; import java.io.Reader; -import java.util.ArrayList; -import java.util.Collection; -import java.util.List; import org.springframework.integration.adapter.PollableSource; import org.springframework.integration.message.MessagingException; @@ -58,28 +55,18 @@ public class CharacterStreamSource implements PollableSource { } - public Collection poll(int limit) { - List results = new ArrayList(); - while (results.size() < limit) { - try { - String line = null; - synchronized (this.monitor) { - boolean isReady = this.reader.ready(); - if (!isReady) { - return results; - } - line = this.reader.readLine(); + public String poll() { + try { + synchronized (this.monitor) { + if (!this.reader.ready()) { + return null; } - if (line == null) { - return results; - } - results.add(line); - } - catch (IOException e) { - throw new MessagingException("IO failure occurred in adapter", e); + return this.reader.readLine(); } } - return results; + catch (IOException e) { + throw new MessagingException("IO failure occurred in adapter", e); + } } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/MethodInvokingSource.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/MethodInvokingSource.java index 660bdfde36..c7571b89ca 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/MethodInvokingSource.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/MethodInvokingSource.java @@ -17,8 +17,6 @@ package org.springframework.integration.adapter; import java.lang.reflect.Method; -import java.util.Arrays; -import java.util.Collection; import org.springframework.beans.factory.InitializingBean; import org.springframework.integration.ConfigurationException; @@ -56,11 +54,11 @@ public class MethodInvokingSource implements PollableSource, Initiali this.invoker.setMethodValidator(new MessageReceivingMethodValidator()); } - public Collection poll(int limit) { + public Object poll() { if (this.invoker == null) { this.afterPropertiesSet(); } - return Arrays.asList(this.invoker.invokeMethod(new Object[] {})); + return this.invoker.invokeMethod(new Object[] {}); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollableSource.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollableSource.java index 65d240a64f..492e0e4023 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollableSource.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollableSource.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2007 the original author or authors. + * Copyright 2002-2008 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -16,8 +16,6 @@ package org.springframework.integration.adapter; -import java.util.Collection; - /** * Interface for any external data source that can be polled. * @@ -25,6 +23,6 @@ import java.util.Collection; */ public interface PollableSource { - Collection poll(int limit); + T poll(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java index 3eb2bff121..c1bdf03320 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java @@ -16,14 +16,12 @@ package org.springframework.integration.adapter; -import java.util.Collection; import java.util.concurrent.Executors; import org.springframework.context.Lifecycle; import org.springframework.integration.ConfigurationException; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.message.MessageMapper; -import org.springframework.integration.message.MessagingException; import org.springframework.integration.scheduling.MessagingTask; import org.springframework.integration.scheduling.MessagingTaskScheduler; import org.springframework.integration.scheduling.MessagingTaskSchedulerAware; @@ -142,16 +140,14 @@ public class PollingSourceAdapter extends AbstractSourceAdapter implements } int messagesProcessed = 0; int limit = this.maxMessagesPerTask; - Collection results = this.source.poll(limit); - if (results != null) { - if (results.size() > limit) { - throw new MessagingException("source returned too many results, the limit is " + limit); + while (messagesProcessed < limit) { + T result = this.source.poll(); + if (result != null && this.sendToChannel(result)) { + messagesProcessed++; + this.onSend(result); } - for (T next : results) { - if (this.sendToChannel(next)) { - messagesProcessed++; - this.onSend(next); - } + else { + return messagesProcessed; } } return messagesProcessed; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SynchronousChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SynchronousChannel.java index c9b1278ab2..668f000f5e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SynchronousChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SynchronousChannel.java @@ -17,7 +17,6 @@ package org.springframework.integration.dispatcher; import java.util.ArrayList; -import java.util.Collection; import java.util.List; import java.util.Queue; import java.util.concurrent.LinkedBlockingQueue; @@ -30,7 +29,6 @@ import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.Message; import org.springframework.integration.message.SimplePayloadMessageMapper; import org.springframework.integration.message.selector.MessageSelector; -import org.springframework.util.CollectionUtils; /** * A channel that invokes the subscribed {@link MessageHandler handler(s)} in a @@ -86,9 +84,8 @@ public class SynchronousChannel extends AbstractMessageChannel { @Override protected Message doReceive(long timeout) { if (this.source != null) { - Collection results = this.source.poll(1); - if (!CollectionUtils.isEmpty(results)) { - Object result = results.iterator().next(); + Object result = this.source.poll(); + if (result != null) { return (result instanceof Message) ? (Message) result : new SimplePayloadMessageMapper().toMessage(result); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/MethodInvokingSourceTests.java b/spring-integration-core/src/test/java/org/springframework/integration/adapter/MethodInvokingSourceTests.java index c7240dadde..aa09aebe6e 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/adapter/MethodInvokingSourceTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/MethodInvokingSourceTests.java @@ -19,8 +19,6 @@ package org.springframework.integration.adapter; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; -import java.util.Collection; - import org.junit.Test; import org.springframework.integration.message.MessagingException; @@ -35,9 +33,9 @@ public class MethodInvokingSourceTests { MethodInvokingSource source = new MethodInvokingSource(); source.setObject(new TestBean()); source.setMethod("validMethod"); - Collection result = source.poll(5); + Object result = source.poll(); assertNotNull(result); - assertEquals("valid", result.iterator().next()); + assertEquals("valid", result); } @Test(expected=MessagingException.class) @@ -45,7 +43,7 @@ public class MethodInvokingSourceTests { MethodInvokingSource source = new MethodInvokingSource(); source.setObject(new TestBean()); source.setMethod("noMatchingMethod"); - source.poll(5); + source.poll(); } @Test(expected=MessagingException.class) @@ -53,7 +51,7 @@ public class MethodInvokingSourceTests { MethodInvokingSource source = new MethodInvokingSource(); source.setObject(new TestBean()); source.setMethod("invalidMethodWithArg"); - source.poll(5); + source.poll(); } @Test(expected=MessagingException.class) @@ -61,7 +59,7 @@ public class MethodInvokingSourceTests { MethodInvokingSource source = new MethodInvokingSource(); source.setObject(new TestBean()); source.setMethod("invalidMethodWithNoReturnValue"); - source.poll(5); + source.poll(); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/PollingSourceAdapterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/adapter/PollingSourceAdapterTests.java index 16ed7bce0c..819ba1b135 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/adapter/PollingSourceAdapterTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/PollingSourceAdapterTests.java @@ -20,16 +20,12 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; -import java.util.ArrayList; -import java.util.Collection; -import java.util.List; import java.util.concurrent.atomic.AtomicInteger; import org.junit.Test; import org.springframework.integration.channel.SimpleChannel; import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessagingException; /** * @author Mark Fisher @@ -66,11 +62,12 @@ public class PollingSourceAdapterTests { assertEquals("testing.1", message1.getPayload()); Message message2 = channel.receive(0); assertNull("second message should be null", message2); + source.resetCounter(); adapter.start(); adapter.processMessages(); Message message3 = channel.receive(100); assertNotNull("third message should not be null", message3); - assertEquals("testing.3", message3.getPayload()); + assertEquals("testing.1", message3.getPayload()); } @Test @@ -96,38 +93,29 @@ public class PollingSourceAdapterTests { assertNull("message should be null", message4); } - @Test(expected=MessagingException.class) - public void testResultSizeExceedsLimit() { - TestSource source = new TestSource("testing", 3); - SimpleChannel channel = new SimpleChannel(); - PollingSourceAdapter adapter = new PollingSourceAdapter(source); - adapter.setChannel(channel); - adapter.setPeriod(1000); - adapter.setMaxMessagesPerTask(2); - adapter.start(); - adapter.processMessages(); - } - private static class TestSource implements PollableSource { private String message; - private int messagesPerPoll; + private int limit; private AtomicInteger count = new AtomicInteger(); - public TestSource(String message, int messagesPerPoll) { + public TestSource(String message, int limit) { this.message = message; - this.messagesPerPoll = messagesPerPoll; + this.limit = limit; } - public Collection poll(int limit) { - List results = new ArrayList(this.messagesPerPoll); - for (int i = 0; i < this.messagesPerPoll; i++) { - results.add(message + "." + count.incrementAndGet()); + public void resetCounter() { + this.count.set(0); + } + + public String poll() { + if (count.get() >= limit) { + return null; } - return results; + return message + "." + count.incrementAndGet(); } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java b/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java index 525fcfbe1b..670d755081 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java @@ -269,7 +269,7 @@ public class MessageBusTests { this.latch = latch; } - public Collection poll(int limit) { + public Collection poll() { latch.countDown(); throw new RuntimeException("intentional test failure"); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SynchronousChannelTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SynchronousChannelTests.java index 6f0239feea..feb3065db8 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SynchronousChannelTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SynchronousChannelTests.java @@ -21,8 +21,6 @@ import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; -import java.util.Collection; -import java.util.Collections; import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.SynchronousQueue; @@ -99,8 +97,8 @@ public class SynchronousChannelTests { @Test public void testReceive() { SynchronousChannel channel = new SynchronousChannel(new PollableSource() { - public Collection poll(int limit) { - return Collections.singleton("foo"); + public String poll() { + return "foo"; } }); Message message = channel.receive(); @@ -186,10 +184,10 @@ public class SynchronousChannelTests { this.messageText = messageText; } - public Collection poll(int limit) { + public StringMessage poll() { StringMessage message = new StringMessage(messageText); message.getHeader().setProperty(HANDLER_THREAD, Thread.currentThread().getName()); - return Collections.singleton(message); + return message; } }