From 10d47903ef73658283325b41775d0e705a97afc8 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Fri, 5 Sep 2008 17:05:41 +0000 Subject: [PATCH] Removed AbstractRequestReplyEndpoint. MessagingBridge now extends AbstractInOutEndpoint. --- .../AbstractRequestReplyEndpoint.java | 65 ------------- .../endpoint/ConcurrencyPolicy.java | 97 ------------------- .../integration/endpoint/MessagingBridge.java | 18 +--- 3 files changed, 4 insertions(+), 176 deletions(-) delete mode 100644 org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractRequestReplyEndpoint.java delete mode 100644 org.springframework.integration/src/main/java/org/springframework/integration/endpoint/ConcurrencyPolicy.java diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractRequestReplyEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractRequestReplyEndpoint.java deleted file mode 100644 index 33d493be72..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractRequestReplyEndpoint.java +++ /dev/null @@ -1,65 +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.endpoint; - -import org.springframework.integration.message.CompositeMessage; -import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageHandlingException; - -/** - * @author Mark Fisher - */ -public abstract class AbstractRequestReplyEndpoint extends AbstractEndpoint { - - private volatile boolean requiresReply = false; - - - /** - * Specify whether this endpoint should throw an Exception when - * it returns an invalid reply Message after handling the request. - */ - public void setRequiresReply(boolean requiresReply) { - this.requiresReply = requiresReply; - } - - - protected boolean sendInternal(Message requestMessage) { - Message replyMessage = this.handleRequestMessage(requestMessage); - if (!this.isValidReplyMessage(replyMessage)) { - if (this.requiresReply) { - throw new MessageHandlingException(requestMessage, - "endpoint requires reply but none was received"); - } - } - else if (replyMessage instanceof CompositeMessage) { - for (Message nextReply : (CompositeMessage) replyMessage) { - this.sendReplyMessage(nextReply, requestMessage); - } - } - else { - this.sendReplyMessage(replyMessage, requestMessage); - } - return true; - } - - protected abstract Message handleRequestMessage(Message requestMessage); - - protected abstract boolean isValidReplyMessage(Message replyMessage); - - protected abstract void sendReplyMessage(Message replyMessage, Message requestMessage); - -} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/ConcurrencyPolicy.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/ConcurrencyPolicy.java deleted file mode 100644 index d294713ce2..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/ConcurrencyPolicy.java +++ /dev/null @@ -1,97 +0,0 @@ -/* - * 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.endpoint; - -import org.springframework.util.Assert; - -/** - * Metadata for configuring a pool of concurrent threads. - * - * @author Mark Fisher - */ -public class ConcurrencyPolicy { - - public static final int DEFAULT_CORE_SIZE = 1; - - public static final int DEFAULT_MAX_SIZE = 10; - - public static final int DEFAULT_QUEUE_CAPACITY = 0; - - public static final int DEFAULT_KEEP_ALIVE_SECONDS = 60; - - - private int coreSize = DEFAULT_CORE_SIZE; - - private int maxSize = DEFAULT_MAX_SIZE; - - private int queueCapacity = DEFAULT_QUEUE_CAPACITY; - - private int keepAliveSeconds = DEFAULT_KEEP_ALIVE_SECONDS; - - - public ConcurrencyPolicy() { - } - - public ConcurrencyPolicy(int coreSize, int maxSize) { - Assert.isTrue(maxSize >= coreSize, "'coreSize' must not exceed 'maxSize'"); - this.setCoreSize(coreSize); - this.setMaxSize(maxSize); - } - - - public int getCoreSize() { - return this.coreSize; - } - - public void setCoreSize(int coreSize) { - Assert.isTrue(coreSize > 0, "'coreSize' must be at least 1"); - this.coreSize = coreSize; - } - - public int getMaxSize() { - return this.maxSize; - } - - public void setMaxSize(int maxSize) { - Assert.isTrue(maxSize > 0, "'maxSize' must be at least 1"); - this.maxSize = maxSize; - } - - public int getQueueCapacity() { - return this.queueCapacity; - } - - public void setQueueCapacity(int queueCapacity) { - Assert.isTrue(queueCapacity >= 0, "'queueCapacity' must not be negative"); - this.queueCapacity = queueCapacity; - } - - public int getKeepAliveSeconds() { - return this.keepAliveSeconds; - } - - public void setKeepAliveSeconds(int keepAliveSeconds) { - Assert.isTrue(keepAliveSeconds >= 0, "'keepAliveSeconds' must not be negative"); - this.keepAliveSeconds = keepAliveSeconds; - } - - public String toString() { - return "[coreSize=" + this.coreSize + ", maxSize=" + this.maxSize + - ", queueCapacity=" + this.queueCapacity + ", keepAliveSeconds=" + this.keepAliveSeconds + "]"; - } - -} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessagingBridge.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessagingBridge.java index a5d813c3b9..956bd948bd 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessagingBridge.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessagingBridge.java @@ -23,7 +23,7 @@ import org.springframework.util.Assert; /** * @author Mark Fisher */ -public class MessagingBridge extends AbstractRequestReplyEndpoint { +public class MessagingBridge extends AbstractInOutEndpoint { private final MessageTarget target; @@ -33,20 +33,10 @@ public class MessagingBridge extends AbstractRequestReplyEndpoint { this.target = target; } - @Override - protected Message handleRequestMessage(Message requestMessage) { - return requestMessage; - } - - @Override - protected boolean isValidReplyMessage(Message replyMessage) { - return replyMessage != null && replyMessage.getPayload() != null; - } - - @Override - protected void sendReplyMessage(Message replyMessage, Message requestMessage) { - this.target.send(replyMessage); + protected Object handle(Message message) { + this.target.send(message); + return null; } }