Spring/spring-batch
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

RESOLVED - issue BATCH-274: Support Callable<ExitStatus>

Browse Source 
Added CallablTaskletAdapter (also removed dependency on backport concurrent)
This commit is contained in:
dsyer
2008-07-17 12:15:54 +00:00
parent 609c7005e9
commit 06d781a950
13 changed files with 139 additions and 246 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
spring-batch-core/pom.xml
View File

@@ -106,10 +106,6 @@
<artifactId>spring-tx</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>backport-util-concurrent</groupId>
<artifactId>backport-util-concurrent</artifactId>
</dependency>
</dependencies>
<build>

40
spring-batch-core/src/main/java/org/springframework/batch/core/step/BackportConcurrentStepExecutionSynchronizer.java
View File

@@ -1,40 +0,0 @@
/*
* 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.core.step;
import org.springframework.batch.core.StepExecution;
import edu.emory.mathcs.backport.java.util.concurrent.Semaphore;
/**
* An implementation of the {@link StepExecutionSynchronizer} that uses Backport Concurrent Utilities.
*
* @author Dave Syer
* @author Ben Hale
*/
class BackportConcurrentStepExecutionSynchronizer implements StepExecutionSynchronizer {
private Semaphore semaphore = new Semaphore(1);
public void lock(StepExecution stepExecution) throws InterruptedException {
semaphore.acquire();
}
public void release(StepExecution stepExecution) {
semaphore.release();
}
}

22
spring-batch-core/src/main/java/org/springframework/batch/core/step/StepExecutionSynchronizerFactory.java
View File

@@ -16,33 +16,19 @@
package org.springframework.batch.core.step;
import org.springframework.core.JdkVersion;
import org.springframework.util.ClassUtils;
/**
* A factory that properly determines which version of the {@link StepExecutionSynchronizer} to return based on the
* availabilty of Java 5 or Backport Concurrent.
* A factory for {@link StepExecutionSynchronizer} which simply creates one from
* java.util.concurrent components.
*
* @author Ben Hale
* @author Dave Syer
*/
public class StepExecutionSynchronizerFactory {
/** Whether the backport-concurrent library is present on the classpath */
private static final boolean backportConcurrentAvailable = ClassUtils.isPresent(
"edu.emory.mathcs.backport.java.util.concurrent.Semaphore", StepExecutionSynchronizerFactory.class
.getClassLoader());
private final StepExecutionSynchronizer synchronizer;
public StepExecutionSynchronizerFactory() {
if (JdkVersion.isAtLeastJava15()) {
synchronizer = new JdkConcurrentStepExecutionSynchronizer();
} else if (backportConcurrentAvailable) {
synchronizer = new BackportConcurrentStepExecutionSynchronizer();
} else {
throw new IllegalStateException("Cannot create StepExecutionSynchronizer - "
+ "neither JDK 1.5 nor backport-concurrent available on the classpath");
}
synchronizer = new JdkConcurrentStepExecutionSynchronizer();
}
public StepExecutionSynchronizer getStepExecutionSynchronizer() {

60
spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/CallableTaskletAdapter.java Normal file
View File

@@ -0,0 +1,60 @@
/*
* 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.core.step.tasklet;
import java.util.concurrent.Callable;
import org.springframework.batch.repeat.ExitStatus;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.Assert;
/**
* Adapts a {@link Callable}&lt;{@link ExitStatus}&gt; to the
* {@link Tasklet} interface.
*
* @author Dave Syer
*
*/
public class CallableTaskletAdapter implements Tasklet, InitializingBean {
private Callable<ExitStatus> callable;
/**
* Public setter for the {@link Callable}.
* @param callable the {@link Callable} to set
*/
public void setCallable(Callable<ExitStatus> callable) {
this.callable = callable;
}
/**
* Assert that the callable is set.
*
* @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
*/
public void afterPropertiesSet() throws Exception {
Assert.notNull(callable);
}
/**
* Execute the provided Callable and return its {@link ExitStatus}.
* @see org.springframework.batch.core.step.tasklet.Tasklet#execute()
*/
public ExitStatus execute() throws Exception {
return callable.call();
}
}

54
spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/CallableTaskletAdapterTests.java Normal file
View File

@@ -0,0 +1,54 @@
/*
* 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.core.step.tasklet;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.fail;
import java.util.concurrent.Callable;
import org.junit.Test;
import org.springframework.batch.repeat.ExitStatus;
public class CallableTaskletAdapterTests {
private CallableTaskletAdapter adapter = new CallableTaskletAdapter();
/**
* Test method for {@link org.springframework.batch.core.step.tasklet.CallableTaskletAdapter#execute()}.
* @throws Exception
*/
@Test
public void testExecute() throws Exception {
adapter.setCallable(new Callable<ExitStatus>() {
public ExitStatus call() throws Exception {
return ExitStatus.FINISHED;
}
});
assertEquals(ExitStatus.FINISHED, adapter.execute());
}
@Test
public void testAfterPropertiesSet() throws Exception {
try {
adapter.afterPropertiesSet();
fail("Expected IllegalArgumentException");
}
catch (IllegalArgumentException e) {
// expected
}
}
}

17
spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/TaskletAdapterTests.java
View File

@@ -15,16 +15,17 @@
*/
package org.springframework.batch.core.step.tasklet;
import org.springframework.batch.core.step.tasklet.TaskletAdapter;
import org.springframework.batch.repeat.ExitStatus;
import static org.junit.Assert.assertEquals;
import junit.framework.TestCase;
import org.junit.Before;
import org.junit.Test;
import org.springframework.batch.repeat.ExitStatus;
/**
* @author Dave Syer
*
*/
public class TaskletAdapterTests extends TestCase {
public class TaskletAdapterTests {
private TaskletAdapter tasklet = new TaskletAdapter();
private Object result = null;
@@ -40,7 +41,8 @@ public class TaskletAdapterTests extends TestCase {
/* (non-Javadoc)
* @see junit.framework.TestCase#setUp()
*/
protected void setUp() throws Exception {
@Before
public void setUp() throws Exception {
tasklet.setTargetObject(this);
tasklet.setTargetMethod("execute");
}
@@ -49,6 +51,7 @@ public class TaskletAdapterTests extends TestCase {
* Test method for {@link org.springframework.batch.core.step.tasklet.TaskletAdapter#execute()}.
* @throws Exception
*/
@Test
public void testExecuteWithExitStatus() throws Exception {
assertEquals(ExitStatus.NOOP, tasklet.execute());
}
@@ -56,6 +59,7 @@ public class TaskletAdapterTests extends TestCase {
/**
* Test method for {@link org.springframework.batch.core.step.tasklet.TaskletAdapter#mapResult(java.lang.Object)}.
*/
@Test
public void testMapResultWithNull() throws Exception {
tasklet.setTargetMethod("process");
assertEquals(ExitStatus.FINISHED, tasklet.execute());
@@ -63,7 +67,8 @@ public class TaskletAdapterTests extends TestCase {
/**
* Test method for {@link org.springframework.batch.core.step.tasklet.TaskletAdapter#mapResult(java.lang.Object)}.
*/
*/
@Test
public void testMapResultWithNonNull() throws Exception {
tasklet.setTargetMethod("process");
this.result = "foo";

5
spring-batch-infrastructure/pom.xml
View File

@@ -61,7 +61,6 @@
<instructions>
<Import-Package><![CDATA[
com.ibatis.sqlmap.*;version="[2.3.0, 3.0.0)";resolution:=optional,
edu.emory.mathcs.backport.*;version="[3.0.0, 4.0.0)";resolution:=optional,
javax.jms.*;version=1.1.0;resolution:=optional,
javax.xml.stream.*;version=1.2.0;resolution:=optional,
org.aopalliance.*;version=1.0.0;resolution:=optional,
@@ -110,10 +109,6 @@
<artifactId>cglib-nodep</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>backport-util-concurrent</groupId>
<artifactId>backport-util-concurrent</artifactId>
</dependency>
<dependency>
<!-- This is just for the JMS spec - could use javax.jms -->
<groupId>org.apache.geronimo.specs</groupId>

52
spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/context/AtomicCounterFactory.java
View File

@@ -1,52 +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.batch.repeat.context;
import org.springframework.core.JdkVersion;
import org.springframework.util.ClassUtils;
/**
* A factory that properly determines which version of the {@link AtomicCounter}
* to return based on the availability of Java 5 or Backport Concurrent.
*
* @author Dave Syer
*/
class AtomicCounterFactory {
/** Whether the backport-concurrent library is present on the classpath */
private static final boolean backportConcurrentAvailable = ClassUtils.isPresent(
"edu.emory.mathcs.backport.java.util.concurrent.Semaphore", AtomicCounterFactory.class.getClassLoader());
private final AtomicCounter counter;
public AtomicCounterFactory() {
if (JdkVersion.isAtLeastJava15()) {
counter = new JdkConcurrentAtomicCounter();
}
else if (backportConcurrentAvailable) {
counter = new BackportConcurrentAtomicCounter();
}
else {
throw new IllegalStateException("Cannot create AtomicCounter - "
+ "neither JDK 1.5 nor backport-concurrent available on the classpath");
}
}
public AtomicCounter getAtomicCounter() {
return counter;
}
}

19
spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/context/BackportConcurrentAtomicCounter.java
View File

@@ -1,19 +0,0 @@
package org.springframework.batch.repeat.context;
/**
* @author Dave Syer
*
*/
class BackportConcurrentAtomicCounter implements AtomicCounter {
private edu.emory.mathcs.backport.java.util.concurrent.atomic.AtomicInteger counter = new edu.emory.mathcs.backport.java.util.concurrent.atomic.AtomicInteger();
public void addAndGet(int delta) {
counter.addAndGet(delta);
}
public int intValue() {
return counter.intValue();
}
}

19
spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/context/JdkConcurrentAtomicCounter.java
View File

@@ -1,19 +0,0 @@
package org.springframework.batch.repeat.context;
/**
* @author Dave Syer
*
*/
class JdkConcurrentAtomicCounter implements AtomicCounter {
private java.util.concurrent.atomic.AtomicInteger counter = new java.util.concurrent.atomic.AtomicInteger();
public void addAndGet(int delta) {
counter.addAndGet(delta);
}
public int intValue() {
return counter.intValue();
}
}

11
spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/context/RepeatContextCounter.java
View File

@@ -16,6 +16,8 @@
package org.springframework.batch.repeat.context;
import java.util.concurrent.atomic.AtomicInteger;
import org.springframework.batch.repeat.RepeatContext;
import org.springframework.util.Assert;
@@ -47,7 +49,7 @@ public class RepeatContextCounter {
* @param delta the amount by which to increment the counter.
*/
final public void increment(int delta) {
AtomicCounter count = getCounter();
AtomicInteger count = getCounter();
count.addAndGet(delta);
}
@@ -94,8 +96,7 @@ public class RepeatContextCounter {
this.context = context;
}
if (!this.context.hasAttribute(countKey)) {
AtomicCounterFactory factory = new AtomicCounterFactory();
this.context.setAttribute(countKey, factory.getAtomicCounter());
this.context.setAttribute(countKey, new AtomicInteger());
}
}
@@ -107,8 +108,8 @@ public class RepeatContextCounter {
return getCounter().intValue();
}
private AtomicCounter getCounter() {
return ((AtomicCounter) context.getAttribute(countKey));
private AtomicInteger getCounter() {
return ((AtomicInteger) context.getAttribute(countKey));
}
}

61
spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/BackportConcurrentResultQueue.java
View File

@@ -1,61 +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.batch.repeat.support;
import edu.emory.mathcs.backport.java.util.concurrent.BlockingQueue;
import edu.emory.mathcs.backport.java.util.concurrent.LinkedBlockingQueue;
import edu.emory.mathcs.backport.java.util.concurrent.Semaphore;
/**
* An implementation of the {@link ResultQueue} that uses Backport Concurrent Utilities.
*
* @author Ben Hale
*/
class BackportConcurrentResultQueue extends AbstractResultQueue implements RepeatInternalState {
// Accumulation of result objects as they finish.
private final BlockingQueue results;
// Accumulation of dummy objects flagging expected results in the future.
private final Semaphore waits;
BackportConcurrentResultQueue(int throttleLimit) {
results = new LinkedBlockingQueue();
waits = new Semaphore(throttleLimit);
}
protected void addResult(ResultHolder resultHolder) {
results.add(resultHolder);
}
protected void aquireWait() throws InterruptedException {
waits.acquire();
}
protected void releaseWait() {
waits.release();
}
protected ResultHolder takeResult() throws InterruptedException {
return (ResultHolder) results.take();
}
public boolean isEmpty() {
return results.isEmpty();
}
}

21
spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/ResultQueueFactory.java
View File

@@ -16,30 +16,17 @@
package org.springframework.batch.repeat.support;
import org.springframework.core.JdkVersion;
import org.springframework.util.ClassUtils;
/**
* A factory that properly determines which version of the {@link ResultQueue} to return based on the availability of
* Java 5 or Backport Concurrent.
* A factory for {@link ResultQueue} which simply creates one from
* java.util.concurrent components.
*
* @author Ben Hale
* @author Dave Syer
*/
class ResultQueueFactory {
/** Whether the backport-concurrent library is present on the classpath */
private static final boolean backportConcurrentAvailable = ClassUtils.isPresent(
"edu.emory.mathcs.backport.java.util.concurrent.Semaphore", ResultQueueFactory.class.getClassLoader());
public RepeatInternalState getResultQueue(int throttleLimit) {
if (JdkVersion.isAtLeastJava15()) {
return new JdkConcurrentResultQueue(throttleLimit);
} else if (backportConcurrentAvailable) {
return new BackportConcurrentResultQueue(throttleLimit);
} else {
throw new IllegalStateException("Cannot create ResultQueue - "
+ "neither JDK 1.5 nor backport-concurrent available on the classpath");
}
return new JdkConcurrentResultQueue(throttleLimit);
}
}