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

OPEN - issue BATCH-378: RepeatListener is confusing and too generic to use for 'intercepting' a step

Browse Source 
http://jira.springframework.org/browse/BATCH-378

Fix hibernate aware item writer - no need for listener interface now that ItemWRiter has flush/clear
This commit is contained in:
dsyer
2008-02-29 16:15:52 +00:00
parent b1ae2efaae
commit e5f4413101
10 changed files with 453 additions and 217 deletions
Download Patch File Download Diff File Expand all files Collapse all files

143
spring-batch-infrastructure/src/main/java/org/springframework/batch/io/support/HibernateAwareItemWriter.java
View File

@@ -22,11 +22,10 @@ import org.hibernate.SessionFactory;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.exception.ClearFailedException;
import org.springframework.batch.item.exception.FlushFailedException;
import org.springframework.batch.repeat.ExitStatus;
import org.springframework.batch.repeat.RepeatContext;
import org.springframework.batch.repeat.RepeatListener;
import org.springframework.batch.repeat.synch.RepeatSynchronizationManager;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.core.AttributeAccessor;
import org.springframework.orm.hibernate3.HibernateOperations;
import org.springframework.orm.hibernate3.HibernateTemplate;
import org.springframework.transaction.support.TransactionSynchronizationManager;
@@ -46,19 +45,13 @@ import org.springframework.util.Assert;
* @author Dave Syer
*
*/
public class HibernateAwareItemWriter implements ItemWriter, RepeatListener, InitializingBean {
public class HibernateAwareItemWriter implements ItemWriter, InitializingBean {
/**
* Key for items processed in the current transaction {@link RepeatContext}.
*/
private static final String ITEMS_PROCESSED = HibernateAwareItemWriter.class.getName() + ".ITEMS_PROCESSED";
/**
* Key for {@link RepeatContext} in transaction resource context.
*/
private static final String WRITER_REPEAT_CONTEXT = HibernateAwareItemWriter.class.getName()
+ ".WRITER_REPEAT_CONTEXT";
private Set failed = new HashSet();
private ItemWriter delegate;
@@ -112,99 +105,21 @@ public class HibernateAwareItemWriter implements ItemWriter, RepeatListener, Ini
* @see org.springframework.batch.io.OutputSource#write(java.lang.Object)
*/
public void write(Object output) throws Exception {
bindTransactionResources();
getProcessed().add(output);
delegate.write(output);
flushIfNecessary(output);
}
/**
* Does nothing unless the delegate is also a {@link RepeatListener} in
* which case pass on the call to him.
*
* @see org.springframework.batch.repeat.RepeatListener#before(org.springframework.batch.repeat.RepeatContext)
*/
public void before(RepeatContext context) {
if (delegate instanceof RepeatListener) {
RepeatListener interceptor = (RepeatListener) delegate;
interceptor.before(context);
}
}
/**
* Does nothing unless the delegate is also a {@link RepeatListener} in
* which case pass on the call to him.
*
* @see org.springframework.batch.repeat.RepeatListener#after(org.springframework.batch.repeat.RepeatContext,
* org.springframework.batch.repeat.ExitStatus)
*/
public void after(RepeatContext context, ExitStatus result) {
if (delegate instanceof RepeatListener) {
RepeatListener interceptor = (RepeatListener) delegate;
interceptor.after(context, result);
}
}
/**
* If the delegate is also a {@link RepeatListener} then it will be given
* the call before flushing.
*
*
* @see org.springframework.batch.repeat.RepeatListener#close(org.springframework.batch.repeat.RepeatContext)
*/
public void close(RepeatContext context) {
try {
flushInContext();
}
finally {
unsetContext();
if (delegate instanceof RepeatListener) {
RepeatListener interceptor = (RepeatListener) delegate;
interceptor.close(context);
}
}
}
/**
* Does nothing unless the delegate is also a {@link RepeatListener} in
* which case pass on the call to him.
*
* @see org.springframework.batch.repeat.RepeatListener#onError(org.springframework.batch.repeat.RepeatContext,
* java.lang.Throwable)
*/
public void onError(RepeatContext context, Throwable e) {
if (delegate instanceof RepeatListener) {
RepeatListener interceptor = (RepeatListener) delegate;
interceptor.onError(context, e);
}
}
/**
* Sets up the context as a transaction resource so that we can store state
* and refer back to it in the {@link #write(Object)} method. If the
* delegate is also a {@link RepeatListener} then it will be given the call
* afterwards.
*
* @see org.springframework.batch.repeat.RepeatListener#open(org.springframework.batch.repeat.RepeatContext)
*/
public void open(RepeatContext context) {
this.setContext(context);
getProcessed().clear();
if (delegate instanceof RepeatListener) {
RepeatListener interceptor = (RepeatListener) delegate;
interceptor.open(context);
}
}
/**
* Accessor for the list of processed items in this transaction.
*
* @return the processed
*/
private Set getProcessed() {
Assert.state(TransactionSynchronizationManager.hasResource(WRITER_REPEAT_CONTEXT),
"RepeatContext not bound to transaction.");
Set processed = (Set) ((AttributeAccessor) TransactionSynchronizationManager.getResource(WRITER_REPEAT_CONTEXT))
.getAttribute(ITEMS_PROCESSED);
Assert.state(TransactionSynchronizationManager.hasResource(ITEMS_PROCESSED),
"Processed items not bound to transaction.");
Set processed = (Set) TransactionSynchronizationManager.getResource(ITEMS_PROCESSED);
return processed;
}
@@ -213,22 +128,21 @@ public class HibernateAwareItemWriter implements ItemWriter, RepeatListener, Ini
*
* @param context the context to set
*/
private void setContext(RepeatContext context) {
if (TransactionSynchronizationManager.hasResource(WRITER_REPEAT_CONTEXT)) {
private void bindTransactionResources() {
if (TransactionSynchronizationManager.hasResource(ITEMS_PROCESSED)) {
return;
}
TransactionSynchronizationManager.bindResource(WRITER_REPEAT_CONTEXT, context);
context.setAttribute(ITEMS_PROCESSED, new HashSet());
TransactionSynchronizationManager.bindResource(ITEMS_PROCESSED, new HashSet());
}
/**
* Remove the transaction resource associated with this context.
*/
private void unsetContext() {
if (!TransactionSynchronizationManager.hasResource(WRITER_REPEAT_CONTEXT)) {
private void unbindTransactionResources() {
if (!TransactionSynchronizationManager.hasResource(ITEMS_PROCESSED)) {
return;
}
TransactionSynchronizationManager.unbindResource(WRITER_REPEAT_CONTEXT);
TransactionSynchronizationManager.unbindResource(ITEMS_PROCESSED);
}
/**
@@ -244,23 +158,21 @@ public class HibernateAwareItemWriter implements ItemWriter, RepeatListener, Ini
flush = failed.contains(output);
}
if (flush) {
RepeatContext context = (RepeatContext) TransactionSynchronizationManager
.getResource(WRITER_REPEAT_CONTEXT);
RepeatContext context = RepeatSynchronizationManager.getContext();
// Force early completion to commit aggressively if we encounter a
// failed item (from a failed chunk but we don't know which one was
// the problem).
context.setCompleteOnly();
// Flush now, so that if there is a failure this record can be
// skipped.
flushInContext();
doHibernateFlush();
}
}
/**
*
* Flush the hibernate session from within a repeat context.
*/
private void flushInContext() {
private void doHibernateFlush() {
try {
hibernateTemplate.flush();
// This should happen when the transaction commits anyway, but to be
@@ -277,25 +189,28 @@ public class HibernateAwareItemWriter implements ItemWriter, RepeatListener, Ini
}
}
/*
* (non-Javadoc)
/**
* Call the delegate clear() method, and then clear the hibernate session.
*
* @see org.springframework.batch.item.ItemWriter#clear()
*/
public void clear() throws ClearFailedException {
if (delegate != null) {
delegate.clear();
}
unbindTransactionResources();
hibernateTemplate.clear();
delegate.clear();
}
/**
* Flush the Hibernate session. The delegate flush will also be called
* before finishing.
* Flush the Hibernate session and record failures if there are any. The
* delegate flush will also be called.
*
* @see org.springframework.batch.item.ItemWriter#flush()
*/
public void flush() throws FlushFailedException {
if (delegate != null) {
delegate.flush();
}
bindTransactionResources();
doHibernateFlush();
unbindTransactionResources();
delegate.flush();
}
}

105
spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/interceptor/CompositeRepeatListener.java Normal file
View File

@@ -0,0 +1,105 @@
/*
* 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.interceptor;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import org.springframework.batch.repeat.ExitStatus;
import org.springframework.batch.repeat.RepeatContext;
import org.springframework.batch.repeat.RepeatListener;
/**
* @author Dave Syer
*
*/
public class CompositeRepeatListener implements RepeatListener {
private List listeners = new ArrayList();
/**
* Public setter for the listeners.
*
* @param listeners
*/
public void setListeners(RepeatListener[] listeners) {
this.listeners = Arrays.asList(listeners);
}
/**
* Register additional listener.
*
* @param itemReaderListener
*/
public void register(RepeatListener listener) {
if (!listeners.contains(listener)) {
listeners.add(listener);
}
}
/* (non-Javadoc)
* @see org.springframework.batch.repeat.RepeatListener#after(org.springframework.batch.repeat.RepeatContext, org.springframework.batch.repeat.ExitStatus)
*/
public void after(RepeatContext context, ExitStatus result) {
for (Iterator iterator = listeners.iterator(); iterator.hasNext();) {
RepeatListener listener = (RepeatListener) iterator.next();
listener.after(context, result);
}
}
/* (non-Javadoc)
* @see org.springframework.batch.repeat.RepeatListener#before(org.springframework.batch.repeat.RepeatContext)
*/
public void before(RepeatContext context) {
for (Iterator iterator = listeners.iterator(); iterator.hasNext();) {
RepeatListener listener = (RepeatListener) iterator.next();
listener.before(context);
}
}
/* (non-Javadoc)
* @see org.springframework.batch.repeat.RepeatListener#close(org.springframework.batch.repeat.RepeatContext)
*/
public void close(RepeatContext context) {
for (Iterator iterator = listeners.iterator(); iterator.hasNext();) {
RepeatListener listener = (RepeatListener) iterator.next();
listener.close(context);
}
}
/* (non-Javadoc)
* @see org.springframework.batch.repeat.RepeatListener#onError(org.springframework.batch.repeat.RepeatContext, java.lang.Throwable)
*/
public void onError(RepeatContext context, Throwable e) {
for (Iterator iterator = listeners.iterator(); iterator.hasNext();) {
RepeatListener listener = (RepeatListener) iterator.next();
listener.onError(context, e);
}
}
/* (non-Javadoc)
* @see org.springframework.batch.repeat.RepeatListener#open(org.springframework.batch.repeat.RepeatContext)
*/
public void open(RepeatContext context) {
for (Iterator iterator = listeners.iterator(); iterator.hasNext();) {
RepeatListener listener = (RepeatListener) iterator.next();
listener.open(context);
}
}
}