From 03ce9b436acb31623656a91e03d920a27417d052 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Fri, 28 Dec 2007 01:50:57 +0000 Subject: [PATCH] Added SourceAdapter, PollableSource, and PollingSourceAdapter. --- .../integration/adapter/PollableSource.java | 30 +++++ .../adapter/PollingSourceAdapter.java | 109 ++++++++++++++++++ .../integration/adapter/SourceAdapter.java | 30 +++++ 3 files changed, 169 insertions(+) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/adapter/PollableSource.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/adapter/SourceAdapter.java 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 new file mode 100644 index 0000000000..65d240a64f --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollableSource.java @@ -0,0 +1,30 @@ +/* + * Copyright 2002-2007 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.adapter; + +import java.util.Collection; + +/** + * Interface for any external data source that can be polled. + * + * @author Mark Fisher + */ +public interface PollableSource { + + Collection poll(int limit); + +} 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 new file mode 100644 index 0000000000..83af4f6738 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java @@ -0,0 +1,109 @@ +/* + * Copyright 2002-2007 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.adapter; + +import java.util.Collection; + +import org.springframework.integration.MessageHandlingException; +import org.springframework.integration.bus.ConsumerPolicy; +import org.springframework.integration.bus.MessageDispatcher; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageMapper; +import org.springframework.integration.message.SimplePayloadMessageMapper; +import org.springframework.util.Assert; + +/** + * A channel adapter that retrieves objects from a {@link PollableSource}, + * delegates to a {@link MessageMapper} to create messages from those objects, + * and then sends the resulting messages to the provided {@link MessageChannel}. + * + * @author Mark Fisher + */ +public class PollingSourceAdapter implements SourceAdapter, MessageDispatcher { + + private PollableSource source; + + private MessageChannel channel; + + private MessageMapper mapper = new SimplePayloadMessageMapper(); + + private ConsumerPolicy policy; + + private long sendTimeout = -1; + + + public PollingSourceAdapter(PollableSource source, MessageChannel channel, int pollInterval) { + this.source = source; + this.channel = channel; + this.initConsumerPolicy(pollInterval); + } + + public void setMessageMapper(MessageMapper mapper) { + Assert.notNull(mapper, "'mapper' must not be null"); + this.mapper = mapper; + } + + protected MessageMapper getMessageMapper() { + return this.mapper; + } + + public void setLimit(int limit) { + Assert.isTrue(limit > 0, "'limit' must be a positive value"); + this.policy.setMaxMessagesPerTask(limit); + } + + public void setSendTimeout(long sendTimeout) { + this.sendTimeout = sendTimeout; + } + + public ConsumerPolicy getConsumerPolicy() { + return this.policy; + } + + private void initConsumerPolicy(int pollInterval) { + ConsumerPolicy policy = new ConsumerPolicy(); + policy.setPeriod(pollInterval); + this.policy = policy; + } + + public int receiveAndDispatch() { + int messagesProcessed = 0; + int limit = this.policy.getMaxMessagesPerTask(); + Collection results = this.source.poll(limit); + if (results != null) { + if (results.size() > limit) { + throw new MessageHandlingException("source returned too many results, the limit is " + limit); + } + for (T next : results) { + Message message = this.mapper.toMessage(next); + if (this.sendTimeout < 0) { + if (this.channel.send(message)) { + messagesProcessed++; + } + } + else { + if (this.channel.send(message, this.sendTimeout)) { + messagesProcessed++; + } + } + } + } + return messagesProcessed; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/SourceAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/SourceAdapter.java new file mode 100644 index 0000000000..f968ceb264 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/SourceAdapter.java @@ -0,0 +1,30 @@ +/* + * Copyright 2002-2007 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.adapter; + +import org.springframework.integration.bus.ConsumerPolicy; + +/** + * Base interface for source adapters. + * + * @author Mark Fisher + */ +public interface SourceAdapter { + + ConsumerPolicy getConsumerPolicy(); + +}