From a1614c4e3a7c4ea1bd959afd609ced0eb3e1cab2 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Thu, 17 Jan 2008 17:49:51 +0000 Subject: [PATCH] Simplifications and counting changes for DefaultMessageDispatcherTests. --- .../dispatcher/DispatcherPolicy.java | 3 +- .../integration/router/Aggregator.java | 30 ----------- .../integration/router/Resequencer.java | 31 ----------- .../integration/router/RoutingBarrier.java | 30 ----------- .../transformer/MessageFilter.java | 2 +- .../bus/DefaultMessageDispatcherTests.java | 52 ++++++++++++------- 6 files changed, 34 insertions(+), 114 deletions(-) delete mode 100644 spring-integration-core/src/main/java/org/springframework/integration/router/Aggregator.java delete mode 100644 spring-integration-core/src/main/java/org/springframework/integration/router/Resequencer.java delete mode 100644 spring-integration-core/src/main/java/org/springframework/integration/router/RoutingBarrier.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DispatcherPolicy.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DispatcherPolicy.java index 24d455720c..9cf0604465 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DispatcherPolicy.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DispatcherPolicy.java @@ -16,7 +16,6 @@ package org.springframework.integration.dispatcher; -import org.springframework.integration.endpoint.EndpointPolicy; import org.springframework.util.Assert; /** @@ -24,7 +23,7 @@ import org.springframework.util.Assert; * * @author Mark Fisher */ -public class DispatcherPolicy implements EndpointPolicy { +public class DispatcherPolicy { public final static int DEFAULT_MAX_MESSAGES_PER_TASK = 1; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/Aggregator.java b/spring-integration-core/src/main/java/org/springframework/integration/router/Aggregator.java deleted file mode 100644 index ab4e882e8a..0000000000 --- a/spring-integration-core/src/main/java/org/springframework/integration/router/Aggregator.java +++ /dev/null @@ -1,30 +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.router; - -import java.util.Collection; - -/** - * Base interface for aggregators. - * - * @author Mark Fisher - */ -public interface Aggregator { - - T aggregate(Collection t); - -} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/Resequencer.java b/spring-integration-core/src/main/java/org/springframework/integration/router/Resequencer.java deleted file mode 100644 index 72dba380c4..0000000000 --- a/spring-integration-core/src/main/java/org/springframework/integration/router/Resequencer.java +++ /dev/null @@ -1,31 +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.router; - -import java.util.Collection; -import java.util.List; - -/** - * Base interface for resequencers. - * - * @author Mark Fisher - */ -public interface Resequencer { - - List resequence(Collection t); - -} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/RoutingBarrier.java b/spring-integration-core/src/main/java/org/springframework/integration/router/RoutingBarrier.java deleted file mode 100644 index 8941bd1184..0000000000 --- a/spring-integration-core/src/main/java/org/springframework/integration/router/RoutingBarrier.java +++ /dev/null @@ -1,30 +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.router; - -/** - * Base strategy interface for barriers. Defines the common requirement of - * routing scenarios that involve waiting for a particular condition to be - * satisfied, such as an {@link Aggregator} or {@link Resequencer}. - * - * @author Mark Fisher - */ -public interface RoutingBarrier { - - boolean checkCondition(T t); - -} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transformer/MessageFilter.java b/spring-integration-core/src/main/java/org/springframework/integration/transformer/MessageFilter.java index ec6da51596..8e08474086 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/transformer/MessageFilter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/transformer/MessageFilter.java @@ -35,7 +35,7 @@ public class MessageFilter implements MessageHandler { this.selector = selector; } - public Message handle(Message message) { + public Message handle(Message message) { if (this.selector.accept(message)) { return message; } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/DefaultMessageDispatcherTests.java b/spring-integration-core/src/test/java/org/springframework/integration/bus/DefaultMessageDispatcherTests.java index 7453d2424a..5fd7d0eadd 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/DefaultMessageDispatcherTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/DefaultMessageDispatcherTests.java @@ -60,7 +60,8 @@ public class DefaultMessageDispatcherTests { scheduler.start(); dispatcher.setMessagingTaskScheduler(scheduler); dispatcher.start(); - latch.await(100, TimeUnit.MILLISECONDS); + latch.await(500, TimeUnit.MILLISECONDS); + assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); assertEquals("exactly one endpoint should have received message", 1, counter1.get() + counter2.get()); } @@ -78,7 +79,8 @@ public class DefaultMessageDispatcherTests { dispatcher.addHandler(new PooledMessageHandler(endpoint1, 1, 1)); dispatcher.addHandler(new PooledMessageHandler(endpoint2, 1, 1)); dispatcher.start(); - latch.await(100, TimeUnit.MILLISECONDS); + latch.await(500, TimeUnit.MILLISECONDS); + assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); assertEquals("both endpoints should have received message", 2, counter1.get() + counter2.get()); } @@ -102,7 +104,8 @@ public class DefaultMessageDispatcherTests { dispatcher.addHandler(new PooledMessageHandler(endpoint2, 1, 1)); dispatcher.addHandler(new PooledMessageHandler(endpoint3, 1, 1)); dispatcher.start(); - latch.await(100, TimeUnit.MILLISECONDS); + latch.await(500, TimeUnit.MILLISECONDS); + assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); assertEquals("inactive endpoint should not have received message", 0, counter1.get()); assertEquals("exactly one endpoint should have received message", 1, counter2.get() + counter3.get()); } @@ -128,7 +131,8 @@ public class DefaultMessageDispatcherTests { }); dispatcher.addHandler(new PooledMessageHandler(endpoint3, 1, 1)); dispatcher.start(); - latch.await(100, TimeUnit.MILLISECONDS); + latch.await(500, TimeUnit.MILLISECONDS); + assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); assertEquals("inactive endpoint should not have received message", 0, counter2.get()); assertEquals("both active endpoints should have received message", 2, counter1.get() + counter3.get()); } @@ -170,6 +174,7 @@ public class DefaultMessageDispatcherTests { dispatcher.setMessagingTaskScheduler(scheduler); dispatcher.start(); latch.await(500, TimeUnit.MILLISECONDS); + assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); Message errorMessage = errorChannel.receive(100); assertNotNull(errorMessage); assertTrue(errorMessage instanceof ErrorMessage); @@ -201,16 +206,17 @@ public class DefaultMessageDispatcherTests { }); dispatcher.addHandler(new PooledMessageHandler(endpoint3, 1, 1)); dispatcher.start(); - latch.await(1000, TimeUnit.MILLISECONDS); + latch.await(500, TimeUnit.MILLISECONDS); + assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); assertEquals("rejecting endpoint should not have received message", 0, counter2.get()); assertEquals("both non-rejecting endpoints should have received message", 2, counter1.get() + counter3.get()); } @Test - public void testNonBroadcastingDispatcherReachesRejectionLimitAndShouldFail() { + public void testNonBroadcastingDispatcherReachesRejectionLimitAndShouldFail() throws InterruptedException { final AtomicInteger counter1 = new AtomicInteger(); final AtomicInteger counter2 = new AtomicInteger(); - final CountDownLatch latch = new CountDownLatch(2); + final CountDownLatch latch = new CountDownLatch(4); TestEndpoint endpoint1 = new TestEndpoint(counter1, latch); TestEndpoint endpoint2 = new TestEndpoint(counter2, latch); SimpleChannel channel = new SimpleChannel(); @@ -237,7 +243,9 @@ public class DefaultMessageDispatcherTests { scheduler.setErrorHandler(new MessagePublishingErrorHandler(errorChannel)); dispatcher.setMessagingTaskScheduler(scheduler); dispatcher.start(); - Message errorMessage = errorChannel.receive(100); + latch.await(500, TimeUnit.MILLISECONDS); + assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); + Message errorMessage = errorChannel.receive(200); assertNotNull(errorMessage); assertTrue(errorMessage instanceof ErrorMessage); assertEquals(MessageDeliveryException.class, ((ErrorMessage) errorMessage).getPayload().getClass()); @@ -275,7 +283,8 @@ public class DefaultMessageDispatcherTests { } }); dispatcher.start(); - latch.await(300, TimeUnit.MILLISECONDS); + latch.await(500, TimeUnit.MILLISECONDS); + assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); assertEquals("latch should have counted down within allotted time", 0, latch.getCount()); assertEquals("rejecting endpoints should not have received message", 0, counter1.get() + counter2.get()); assertEquals("endpoint1 should have rejected two times", 2, rejectedCounter1.get()); @@ -290,7 +299,7 @@ public class DefaultMessageDispatcherTests { final AtomicInteger rejectedCounter1 = new AtomicInteger(); final AtomicInteger rejectedCounter2 = new AtomicInteger(); final AtomicInteger rejectedCounter3 = new AtomicInteger(); - final CountDownLatch latch = new CountDownLatch(1); + final CountDownLatch latch = new CountDownLatch(5); TestEndpoint endpoint1 = new TestEndpoint(counter1, latch); TestEndpoint endpoint2 = new TestEndpoint(counter2, latch); TestEndpoint endpoint3 = new TestEndpoint(counter3, latch); @@ -304,6 +313,7 @@ public class DefaultMessageDispatcherTests { @Override public Message handle(Message message) { rejectedCounter1.incrementAndGet(); + latch.countDown(); throw new MessageHandlerRejectedExecutionException(); } }); @@ -314,6 +324,7 @@ public class DefaultMessageDispatcherTests { return super.handle(message); } rejectedCounter2.incrementAndGet(); + latch.countDown(); throw new MessageHandlerRejectedExecutionException(); } }); @@ -321,12 +332,13 @@ public class DefaultMessageDispatcherTests { @Override public Message handle(Message message) { rejectedCounter3.incrementAndGet(); + latch.countDown(); throw new MessageHandlerRejectedExecutionException(); } }); dispatcher.start(); - latch.await(300, TimeUnit.MILLISECONDS); - assertEquals("latch should have counted down within allotted time", 0, latch.getCount()); + latch.await(500, TimeUnit.MILLISECONDS); + assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); assertEquals("endpoint1 should not have received message", 0, counter1.get()); assertEquals("endpoint2 should have received message the second time", 1, counter2.get()); assertEquals("endpoint3 should not have received message", 0, counter3.get()); @@ -374,8 +386,8 @@ public class DefaultMessageDispatcherTests { } }); dispatcher.start(); - latch.await(300, TimeUnit.MILLISECONDS); - assertEquals("latch should have counted down within allotted time", 0, latch.getCount()); + latch.await(500, TimeUnit.MILLISECONDS); + assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); assertEquals("endpoint1 should have received one message", 1, counter1.get()); assertEquals("endpoint2 should have received one message", 1, counter2.get()); assertEquals("endpoint1 should have rejected two times", 2, rejectedCounter1.get()); @@ -399,8 +411,8 @@ public class DefaultMessageDispatcherTests { dispatcher.addHandler(executor1); dispatcher.addHandler(executor2); dispatcher.start(); - latch.await(300, TimeUnit.MILLISECONDS); - assertEquals("latch should have counted down within allotted time", 0, latch.getCount()); + latch.await(500, TimeUnit.MILLISECONDS); + assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); assertEquals("endpoint1 should not have accepted the message", 0, counter1.get()); assertEquals("endpoint2 should have accepted the message", 1, counter2.get()); } @@ -439,8 +451,8 @@ public class DefaultMessageDispatcherTests { dispatcher.addHandler(executor1); dispatcher.addHandler(executor2); dispatcher.start(); - attemptedLatch.await(300, TimeUnit.MILLISECONDS); - assertEquals("attemptedLatch should have counted down within allotted time", 0, attemptedLatch.getCount()); + attemptedLatch.await(500, TimeUnit.MILLISECONDS); + assertEquals("messages should have been dispatched within allotted time", 0, attemptedLatch.getCount()); assertEquals("endpoint1 should not have accepted the message", 0, counter1.get()); assertEquals("endpoint2 should not have accepted the message", 0, counter2.get()); assertEquals("executor1 should have had exactly one attempt", 1, attemptedCounter1.get()); @@ -467,8 +479,8 @@ public class DefaultMessageDispatcherTests { dispatcher.addHandler(executor1); dispatcher.addHandler(executor2); dispatcher.start(); - latch.await(300, TimeUnit.MILLISECONDS); - assertEquals("latch should have counted down within allotted time", 0, latch.getCount()); + latch.await(500, TimeUnit.MILLISECONDS); + assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount()); assertEquals("endpoint1 should not have accepted the message", 0, counter1.get()); assertEquals("endpoint2 should have accepted the message", 1, counter2.get()); }