RESOLVED - issue BATCH-1347: Restartable look-ahead (peekable) ItemReader
This commit is contained in:
dsyer
@@ -0,0 +1,49 @@
|
||||
/*
|
||||
* Copyright 2006-2010 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.item;
|
||||
|
||||
/**
|
||||
* <p>
|
||||
* A specialisation of {@link ItemReader} that allows the user to look ahead
|
||||
* into the stream of items. This is useful, for instance, when reading flat
|
||||
* file data that contains record separator lines which are actually part of the
|
||||
* next record.
|
||||
* </p>
|
||||
*
|
||||
* <p>
|
||||
* The detailed contract for {@link #peek()} has to be defined by the
|
||||
* implementation because there is no general way to define it in a concurrent
|
||||
* environment. The definition of "the next read()" operation is tenuous if
|
||||
* multiple clients are reading concurrently, and the ability to peek implies
|
||||
* that some state is likely to be stored, so implementations of
|
||||
* {@link PeekableItemReader} may well be restricted to single threaded use.
|
||||
* </p>
|
||||
*
|
||||
* @author Dave Syer
|
||||
*
|
||||
*/
|
||||
public interface PeekableItemReader<T> extends ItemReader<T> {
|
||||
|
||||
/**
|
||||
* Get the next item that would be returned by {@link #read()}, without
|
||||
* affecting the result of {@link #read()}.
|
||||
*
|
||||
* @return the next item
|
||||
* @throws Exception if there is a problem
|
||||
*/
|
||||
T peek() throws Exception, UnexpectedInputException, ParseException;
|
||||
|
||||
}
|
||||
@@ -0,0 +1,155 @@
|
||||
/*
|
||||
* Copyright 2006-2010 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.item.support;
|
||||
|
||||
import java.util.Map.Entry;
|
||||
|
||||
import org.springframework.batch.item.ExecutionContext;
|
||||
import org.springframework.batch.item.ItemReader;
|
||||
import org.springframework.batch.item.ItemStream;
|
||||
import org.springframework.batch.item.ItemStreamException;
|
||||
import org.springframework.batch.item.ItemStreamReader;
|
||||
import org.springframework.batch.item.ParseException;
|
||||
import org.springframework.batch.item.PeekableItemReader;
|
||||
import org.springframework.batch.item.UnexpectedInputException;
|
||||
|
||||
/**
|
||||
* <p>
|
||||
* A {@link PeekableItemReader} that allows the user to peek one item ahead.
|
||||
* Repeated calls to {@link #peek()} will return the same item, and this will be
|
||||
* the next item returned from {@link #read()}.
|
||||
* </p>
|
||||
*
|
||||
* <p>
|
||||
* Intentionally not thread safe: it wouldn't be possible to honour the peek in
|
||||
* multiple threads because only one of the threads that peeked would get that
|
||||
* item in the next call to read.
|
||||
* </p>
|
||||
*
|
||||
* @author Dave Syer
|
||||
*
|
||||
*/
|
||||
public class SingleItemPeekableItemReader<T> implements ItemStreamReader<T>, PeekableItemReader<T> {
|
||||
|
||||
private ItemReader<T> delegate;
|
||||
|
||||
private T next;
|
||||
|
||||
private ExecutionContext executionContext = new ExecutionContext();
|
||||
|
||||
/**
|
||||
* The item reader to use as a delegate. Items are read from the delegate
|
||||
* and passed to the caller in {@link #read()}.
|
||||
*
|
||||
* @param delegate the delegate to set
|
||||
*/
|
||||
public void setDelegate(ItemReader<T> delegate) {
|
||||
this.delegate = delegate;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the next item from the delegate (whether or not it has already been
|
||||
* peeked at).
|
||||
*
|
||||
* @see ItemReader#read()
|
||||
*/
|
||||
public T read() throws Exception, UnexpectedInputException, ParseException {
|
||||
if (next != null) {
|
||||
T item = next;
|
||||
next = null;
|
||||
// executionContext = new ExecutionContext();
|
||||
return item;
|
||||
}
|
||||
return delegate.read();
|
||||
}
|
||||
|
||||
/**
|
||||
* Peek at the next item, ensuring that if the delegate is an
|
||||
* {@link ItemStream} the state is stored for the next call to
|
||||
* {@link #update(ExecutionContext)}.
|
||||
*
|
||||
* @return the next item (or null if there is none).
|
||||
*
|
||||
* @see PeekableItemReader#peek()
|
||||
*/
|
||||
@Override
|
||||
public T peek() throws Exception, UnexpectedInputException, ParseException {
|
||||
if (next == null) {
|
||||
updateDelegate(executionContext);
|
||||
next = delegate.read();
|
||||
}
|
||||
return next;
|
||||
}
|
||||
|
||||
/**
|
||||
* If the delegate is an {@link ItemStream}, just pass the call on,
|
||||
* otherwise reset the peek cache.
|
||||
*
|
||||
* @throws ItemStreamException if there is a problem
|
||||
* @see ItemStream#close()
|
||||
*/
|
||||
public void close() throws ItemStreamException {
|
||||
next = null;
|
||||
if (delegate instanceof ItemStream) {
|
||||
((ItemStream) delegate).close();
|
||||
}
|
||||
executionContext = new ExecutionContext();
|
||||
}
|
||||
|
||||
/**
|
||||
* If the delegate is an {@link ItemStream}, just pass the call on,
|
||||
* otherwise reset the peek cache.
|
||||
*
|
||||
* @param executionContext the current context
|
||||
* @throws ItemStreamException if there is a problem
|
||||
* @see ItemStream#open(ExecutionContext)
|
||||
*/
|
||||
public void open(ExecutionContext executionContext) throws ItemStreamException {
|
||||
next = null;
|
||||
if (delegate instanceof ItemStream) {
|
||||
((ItemStream) delegate).open(executionContext);
|
||||
}
|
||||
executionContext = new ExecutionContext();
|
||||
}
|
||||
|
||||
/**
|
||||
* If there is a cached peek, then retrieve the execution context state from
|
||||
* that point. If there is no peek cached, then call directly to the
|
||||
* delegate.
|
||||
*
|
||||
* @param executionContext the current context
|
||||
* @throws ItemStreamException if there is a problem
|
||||
* @see ItemStream#update(ExecutionContext)
|
||||
*/
|
||||
public void update(ExecutionContext executionContext) throws ItemStreamException {
|
||||
if (next != null) {
|
||||
// Get the last state from the delegate instead of using
|
||||
// current value.
|
||||
for (Entry<String, Object> entry : this.executionContext.entrySet()) {
|
||||
executionContext.put(entry.getKey(), entry.getValue());
|
||||
}
|
||||
return;
|
||||
}
|
||||
updateDelegate(executionContext);
|
||||
}
|
||||
|
||||
private void updateDelegate(ExecutionContext executionContext) {
|
||||
if (delegate instanceof ItemStream) {
|
||||
((ItemStream) delegate).update(executionContext);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user