From e304b1bdb7d067d2c13fcd1d188557bcc01a07ce Mon Sep 17 00:00:00 2001 From: dsyer Date: Tue, 7 Oct 2008 07:29:58 +0000 Subject: [PATCH] Refine and document the ResultQueue a bit better. --- .../batch/repeat/support/ResultQueue.java | 26 ++++-- .../support/ThrottleLimitResultQueue.java | 28 +++--- .../ThrottleLimitResultQueueTests.java | 89 +++++++++++++++++++ 3 files changed, 125 insertions(+), 18 deletions(-) create mode 100644 spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/support/ThrottleLimitResultQueueTests.java diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/ResultQueue.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/ResultQueue.java index 3ab47a630..c3c83f292 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/ResultQueue.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/ResultQueue.java @@ -16,12 +16,15 @@ package org.springframework.batch.repeat.support; +import java.util.NoSuchElementException; +import java.util.concurrent.BlockingQueue; + import org.springframework.core.task.TaskExecutor; /** - * Abstraction for queue of {@link ResultHolder} objects. Acts as a - * BlockingQueue with the ability to count the number of items it expects to - * ever hold. When clients schedule an item to be added they call + * Abstraction for queue of {@link ResultHolder} objects. Acts a bit likeT a + * {@link BlockingQueue} with the ability to count the number of items it + * expects to ever hold. When clients schedule an item to be added they call * {@link #expect()}, and then collect the result later with {@link #take()}. * Result providers in another thread call {@link #put(Object)} to notify the * expecting client of a new result. @@ -40,26 +43,33 @@ interface ResultQueue { * usually managed elsewhere, e.g. by a {@link TaskExecutor}.

* Implementations may choose to block here, if they need to limit the * number or rate of tasks being submitted. + * * @throws InterruptedException if the call blocks and is then interrupted. */ void expect() throws InterruptedException; /** - * In a master-worker pattern, the workers call this method to deposit the - * result of a finished task on the queue for collection. + * Once it is expecting a result, clients call this method to satisfy the + * expectation. In a master-worker pattern, the workers call this method to + * deposit the result of a finished task on the queue for collection. * * @param result the result for later collection. + * + * @throws IllegalArgumentException if the queue is not expecting a new + * result */ - void put(T result); + void put(T result) throws IllegalArgumentException; /** - * Gets the next available result, blocking if there are none. + * Gets the next available result, blocking if there are none yet available. * * @return a result previously deposited + * + * @throws NoSuchElementException if there is no result expected * @throws InterruptedException if the operation is interrupted while * waiting */ - T take() throws InterruptedException; + T take() throws NoSuchElementException, InterruptedException; /** * Used by master thread to verify that there are results available from diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/ThrottleLimitResultQueue.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/ThrottleLimitResultQueue.java index 7309c67be..b391e456e 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/ThrottleLimitResultQueue.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/ThrottleLimitResultQueue.java @@ -16,17 +16,18 @@ package org.springframework.batch.repeat.support; +import java.util.NoSuchElementException; import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.Semaphore; /** * An implementation of the {@link ResultQueue} that throttles the number of - * expected results, limiting it ti a maximum at any given time. + * expected results, limiting it to a maximum at any given time. * * @author Dave Syer */ -class ThrottleLimitResultQueue implements ResultQueue { +public class ThrottleLimitResultQueue implements ResultQueue { // Accumulation of result objects as they finish. private final BlockingQueue results; @@ -51,15 +52,16 @@ class ThrottleLimitResultQueue implements ResultQueue { return results.isEmpty(); } - /* (non-Javadoc) + /* + * (non-Javadoc) + * * @see org.springframework.batch.repeat.support.ResultQueue#isExpecting() */ public boolean isExpecting() { - synchronized (lock) { - // Base the decision about whether we expect more results on a - // counter of the number of expected results actually collected. - return count > 0; - } + // Base the decision about whether we expect more results on a + // counter of the number of expected results actually collected. + // Do not synchronize! Otherwise put and expect can deadlock. + return count > 0; } /** @@ -76,7 +78,10 @@ class ThrottleLimitResultQueue implements ResultQueue { } } - public void put(T holder) { + public void put(T holder) throws IllegalArgumentException { + if (!isExpecting()) { + throw new IllegalArgumentException("Not expecting a result. Call expect() before put()."); + } // There should be no need to block here, or to use offer() results.add(holder); // Take from the waits queue now to allow another result to @@ -84,7 +89,10 @@ class ThrottleLimitResultQueue implements ResultQueue { waits.release(); } - public T take() throws InterruptedException { + public T take() throws NoSuchElementException, InterruptedException { + if (!isExpecting()) { + throw new NoSuchElementException("Not expecting a result. Call expect() before take()."); + } T value; synchronized (lock) { value = results.take(); diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/support/ThrottleLimitResultQueueTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/support/ThrottleLimitResultQueueTests.java new file mode 100644 index 000000000..8f542dcd0 --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/support/ThrottleLimitResultQueueTests.java @@ -0,0 +1,89 @@ +/* + * Copyright 2006-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.batch.repeat.support; + +import static org.junit.Assert.*; + +import java.util.NoSuchElementException; + +import org.junit.Test; + + +/** + * @author Dave Syer + * + */ +public class ThrottleLimitResultQueueTests { + + private ThrottleLimitResultQueue queue = new ThrottleLimitResultQueue(1); + + @Test + public void testPutTake() throws Exception { + queue.expect(); + assertTrue(queue.isExpecting()); + assertTrue(queue.isEmpty()); + queue.put("foo"); + assertFalse(queue.isEmpty()); + assertEquals("foo", queue.take()); + assertFalse(queue.isExpecting()); + } + + @Test + public void testPutWithoutExpecting() throws Exception { + assertFalse(queue.isExpecting()); + try { + queue.put("foo"); + fail("Expected IllegalArgumentException"); + } catch (IllegalArgumentException e) { + // expected + } + } + + @Test + public void testTakeWithoutExpecting() throws Exception { + assertFalse(queue.isExpecting()); + try { + queue.take(); + fail("Expected NoSuchElementException"); + } catch (NoSuchElementException e) { + // expected + } + } + + @Test + public void testThrottleLimit() throws Exception { + queue.expect(); + new Thread(new Runnable() { + public void run() { + try { + Thread.sleep(100L); + } + catch (InterruptedException e) { + e.printStackTrace(); + Thread.currentThread().interrupt(); + } + queue.put("foo"); + } + }).start(); + long t0 = System.currentTimeMillis(); + queue.expect(); + long t1 = System.currentTimeMillis(); + assertEquals("foo", queue.take()); + assertTrue(queue.isExpecting()); + assertTrue("Did not block on expect (throttle limit should have been hit): time taken="+(t1-t0), t1-t0>50); + } + +}