From 4a1d5fc4d7c13ac2ea20ba42a0deb065769506b4 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Wed, 26 Dec 2007 21:29:39 +0000 Subject: [PATCH] Added null check for EndpointExecutor 'getActiveCount()' and adjusted MessageSelector definition for generic Message. --- .../integration/bus/EndpointExecutor.java | 3 +++ .../integration/message/MessageSelector.java | 2 +- .../channel/PointToPointChannelTests.java | 22 +++++++++---------- 3 files changed, 15 insertions(+), 12 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/EndpointExecutor.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/EndpointExecutor.java index 1594a77478..0fa4cf019d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/EndpointExecutor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/EndpointExecutor.java @@ -127,6 +127,9 @@ public class EndpointExecutor implements Lifecycle { } public int getActiveCount() { + if (this.threadPoolExecutor == null) { + return 0; + } return this.threadPoolExecutor.getActiveCount(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/message/MessageSelector.java b/spring-integration-core/src/main/java/org/springframework/integration/message/MessageSelector.java index c0a17fe2a5..a6278dc223 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/message/MessageSelector.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/message/MessageSelector.java @@ -23,6 +23,6 @@ package org.springframework.integration.message; */ public interface MessageSelector { - boolean accept(Message message); + boolean accept(Message message); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/PointToPointChannelTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/PointToPointChannelTests.java index 30150b8303..6a252eb4d1 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/PointToPointChannelTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/PointToPointChannelTests.java @@ -45,7 +45,7 @@ public class PointToPointChannelTests { final PointToPointChannel channel = new PointToPointChannel(); new Thread(new Runnable() { public void run() { - Message message = channel.receive(); + Message message = channel.receive(); if (message != null) { messageReceived.set(true); latch.countDown(); @@ -67,7 +67,7 @@ public class PointToPointChannelTests { Executor singleThreadExecutor = Executors.newSingleThreadExecutor(); Runnable receiveTask1 = new Runnable() { public void run() { - Message message = channel.receive(0); + Message message = channel.receive(0); if (message != null) { messageReceived.set(true); } @@ -85,7 +85,7 @@ public class PointToPointChannelTests { assertFalse(messageReceived.get()); Runnable receiveTask2 = new Runnable() { public void run() { - Message message = channel.receive(0); + Message message = channel.receive(0); if (message != null) { messageReceived.set(true); } @@ -104,7 +104,7 @@ public class PointToPointChannelTests { final CountDownLatch latch = new CountDownLatch(1); Thread t = new Thread(new Runnable() { public void run() { - Message message = channel.receive(); + Message message = channel.receive(); receiveInterrupted.set(true); assertTrue(message == null); latch.countDown(); @@ -124,7 +124,7 @@ public class PointToPointChannelTests { final CountDownLatch latch = new CountDownLatch(1); Thread t = new Thread(new Runnable() { public void run() { - Message message = channel.receive(10000); + Message message = channel.receive(10000); receiveInterrupted.set(true); assertTrue(message == null); latch.countDown(); @@ -196,11 +196,11 @@ public class PointToPointChannelTests { public void testSelectorMatchesWithinTimeout() throws Exception { final PointToPointChannel channel = new PointToPointChannel(); final CountDownLatch latch = new CountDownLatch(1); - final AtomicReference> messageRef = new AtomicReference>(); + final AtomicReference> messageRef = new AtomicReference>(); Thread receiver = new Thread(new Runnable() { public void run() { - Message message = channel.receive(new MessageSelector() { - public boolean accept(Message message) { + Message message = channel.receive(new MessageSelector() { + public boolean accept(Message message) { return (((Integer)message.getId()).intValue() == 3); } }, 50); @@ -229,11 +229,11 @@ public class PointToPointChannelTests { public void testSelectorDoesNotMatchWithinTimeout() throws Exception { final PointToPointChannel channel = new PointToPointChannel(); final CountDownLatch latch = new CountDownLatch(1); - final AtomicReference> messageRef = new AtomicReference>(); + final AtomicReference> messageRef = new AtomicReference>(); Thread receiver = new Thread(new Runnable() { public void run() { - Message message = channel.receive(new MessageSelector() { - public boolean accept(Message message) { + Message message = channel.receive(new MessageSelector() { + public boolean accept(Message message) { return (((Integer)message.getId()).intValue() == 3); } }, 7);