Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Removed the PollableSource interface, and migrated the 'receive()' method to MessageSource.

Browse Source
This commit is contained in:
Mark Fisher
2008-09-26 16:31:26 +00:00
parent 78c49157bd
commit 9ee921ed7c
18 changed files with 35 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SourcePoller.java
View File

@@ -21,8 +21,8 @@ import org.springframework.integration.message.BlockingSource;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageDeliveryAware;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.MessageSource;
import org.springframework.integration.message.MessagingException;
import org.springframework.integration.message.PollableSource;
import org.springframework.integration.scheduling.Trigger;
import org.springframework.util.Assert;
@@ -31,14 +31,14 @@ import org.springframework.util.Assert;
*/
public class SourcePoller extends AbstractPoller {
private final PollableSource<?> source;
private final MessageSource<?> source;
private final MessageChannel channel;
private volatile long receiveTimeout = 1000;
public SourcePoller(PollableSource<?> source, MessageChannel channel, Trigger trigger) {
public SourcePoller(MessageSource<?> source, MessageChannel channel, Trigger trigger) {
super(trigger);
Assert.notNull(source, "source must not be null");
Assert.notNull(channel, "channel must not be null");

6
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java
View File

@@ -20,8 +20,8 @@ import java.util.concurrent.ScheduledFuture;
import org.springframework.context.Lifecycle;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.message.MessageSource;
import org.springframework.integration.message.MethodInvokingSource;
import org.springframework.integration.message.PollableSource;
import org.springframework.integration.scheduling.TaskScheduler;
import org.springframework.integration.scheduling.Trigger;
@@ -34,7 +34,7 @@ import org.springframework.integration.scheduling.Trigger;
*/
public class SourcePollingChannelAdapter extends AbstractMessageProducingEndpoint implements Lifecycle {
private volatile PollableSource<?> source;
private volatile MessageSource<?> source;
private volatile Trigger trigger;
@@ -49,7 +49,7 @@ public class SourcePollingChannelAdapter extends AbstractMessageProducingEndpoin
private final Object lifecycleMonitor = new Object();
public void setSource(PollableSource<?> source) {
public void setSource(MessageSource<?> source) {
this.source = source;
}

2
org.springframework.integration/src/main/java/org/springframework/integration/message/BlockingSource.java
View File

@@ -21,7 +21,7 @@ package org.springframework.integration.message;
*
* @author Mark Fisher
*/
public interface BlockingSource<T> extends PollableSource<T> {
public interface BlockingSource<T> extends MessageSource<T> {
/**
* Receive a message, blocking indefinitely if necessary.

8
org.springframework.integration/src/main/java/org/springframework/integration/message/MessageSource.java
View File

@@ -17,10 +17,16 @@
package org.springframework.integration.message;
/**
* Base interface for any source of {@link Message Messages}.
* Base interface for any source of {@link Message Messages} that can be polled.
*
* @author Mark Fisher
*/
public interface MessageSource<T> {
/**
* Retrieve the next available message from this source.
* Returns <code>null</code> if no message is available.
*/
Message<T> receive();
}

2
org.springframework.integration/src/main/java/org/springframework/integration/message/MethodInvokingSource.java
View File

@@ -33,7 +33,7 @@ import org.springframework.util.Assert;
*
* @author Mark Fisher
*/
public class MethodInvokingSource implements PollableSource<Object>, InitializingBean {
public class MethodInvokingSource implements MessageSource<Object>, InitializingBean {
private volatile Object object;

31
org.springframework.integration/src/main/java/org/springframework/integration/message/PollableSource.java
View File

@@ -1,31 +0,0 @@
/*
* 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.message;
/**
* Base interface for any source of {@link Message Messages} that can be polled.
*
* @author Mark Fisher
*/
public interface PollableSource<T> extends MessageSource<T> {
/**
* Retrieve a message from this source or <code>null</code> if no message is available.
*/
Message<T> receive();
}

4
org.springframework.integration/src/test/java/org/springframework/integration/bus/DefaultMessageBusTests.java
View File

@@ -39,7 +39,7 @@ import org.springframework.integration.message.ErrorMessage;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.message.PollableSource;
import org.springframework.integration.message.MessageSource;
import org.springframework.integration.message.StringMessage;
import org.springframework.integration.scheduling.IntervalTrigger;
@@ -269,7 +269,7 @@ public class DefaultMessageBusTests {
}
private static class FailingSource implements PollableSource<Object> {
private static class FailingSource implements MessageSource<Object> {
private CountDownLatch latch;

32
org.springframework.integration/src/test/java/org/springframework/integration/channel/config/TestPollableSource.java
View File

@@ -1,32 +0,0 @@
/*
* 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.channel.config;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.PollableSource;
import org.springframework.integration.message.StringMessage;
/**
* @author Mark Fisher
*/
public class TestPollableSource implements PollableSource<String> {
public Message<String> receive() {
return new StringMessage("test");
}
}

4
org.springframework.integration/src/test/java/org/springframework/integration/channel/config/TestSource.java
View File

@@ -17,13 +17,13 @@
package org.springframework.integration.channel.config;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.PollableSource;
import org.springframework.integration.message.MessageSource;
import org.springframework.integration.message.StringMessage;
/**
* @author Mark Fisher
*/
public class TestSource implements PollableSource<String> {
public class TestSource implements MessageSource<String> {
private final String text;

48
org.springframework.integration/src/test/java/org/springframework/integration/channel/config/TestSubscribableSource.java
View File

@@ -1,48 +0,0 @@
/*
* 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.channel.config;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageConsumer;
import org.springframework.integration.message.Subscribable;
/**
* @author Mark Fisher
*/
public class TestSubscribableSource implements Subscribable {
private final List<MessageConsumer> subscibers = new CopyOnWriteArrayList<MessageConsumer>();
public boolean subscribe(MessageConsumer subsciber) {
return this.subscibers.add(subsciber);
}
public boolean unsubscribe(MessageConsumer subsciber) {
return this.subscibers.remove(subsciber);
}
public void publishMessage(Message<?> message) {
for (MessageConsumer subsciber : this.subscibers) {
subsciber.onMessage(message);
}
}
}

4
org.springframework.integration/src/test/java/org/springframework/integration/config/TestSource.java
View File

@@ -17,14 +17,14 @@
package org.springframework.integration.config;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.PollableSource;
import org.springframework.integration.message.MessageSource;
import org.springframework.integration.message.StringMessage;
/**
* @author Mark Fisher
*/
@SuppressWarnings("unchecked")
public class TestSource implements PollableSource {
public class TestSource implements MessageSource {
public Message<?> receive() {
return new StringMessage("test");