class AbstractItemCountingItemStreamItemReader<T> extends AbstractItemStreamItemReader<T> {

	private static final String READ_COUNT = "read.count";

	private static final String READ_COUNT_MAX = "read.count.max";

	private int currentItemCount = 0;

	private int maxItemCount = Integer.MAX_VALUE;

	private boolean saveState = true;

	/**
	 * Read next item from input.
	 * @return an item or {@code null} if the data source is exhausted
	 * @throws Exception Allows subclasses to throw checked exceptions for interpretation
	 * by the framework
	 */
	@Nullable
	protected abstract T doRead() throws Exception;

	/**
	 * Open resources necessary to start reading input.
	 * @throws Exception Allows subclasses to throw checked exceptions for interpretation
	 * by the framework
	 */
	protected abstract void doOpen() throws Exception;

	/**
	 * Close the resources opened in {@link #doOpen()}.
	 * @throws Exception Allows subclasses to throw checked exceptions for interpretation
	 * by the framework
	 */
	protected abstract void doClose() throws Exception;

	/**
	 * Move to the given item index. Subclasses should override this method if there is a
	 * more efficient way of moving to given index than re-reading the input using
	 * {@link #doRead()}.
	 * @param itemIndex index of item (0 based) to jump to.
	 * @throws Exception Allows subclasses to throw checked exceptions for interpretation
	 * by the framework
	 */
	protected void jumpToItem(int itemIndex) throws Exception {
		for (int i = 0; i < itemIndex; i++) {
			read();
		}
	}

	@Nullable
	@Override
	public T read() throws Exception {
		if (currentItemCount >= maxItemCount) {
			return null;
		}
		currentItemCount++;
		T item = doRead();
		if (item instanceof ItemCountAware) {
			((ItemCountAware) item).setItemCount(currentItemCount);
		}
		return item;
	}

	/**
	 * Returns the current item count.
	 * @return the current item count
	 * @since 5.1
	 */
	public int getCurrentItemCount() {
		return this.currentItemCount;
	}

	/**
	 * The index of the item to start reading from. If the {@link ExecutionContext}
	 * contains a key <code>[name].read.count</code> (where <code>[name]</code> is the
	 * name of this component) the value from the {@link ExecutionContext} will be used in
	 * preference.
	 *
	 * @see #setName(String)
	 * @param count the value of the current item count
	 */
	public void setCurrentItemCount(int count) {
		this.currentItemCount = count;
	}

	/**
	 * The maximum index of the items to be read. If the {@link ExecutionContext} contains
	 * a key <code>[name].read.count.max</code> (where <code>[name]</code> is the name of
	 * this component) the value from the {@link ExecutionContext} will be used in
	 * preference.
	 *
	 * @see #setName(String)
	 * @param count the value of the maximum item count. count must be greater than zero.
	 */
	public void setMaxItemCount(int count) {
		Assert.isTrue(count > 0, "count must be greater than zero");
		this.maxItemCount = count;
	}

	@Override
	public void close() throws ItemStreamException {
		super.close();
		currentItemCount = 0;
		try {
			doClose();
		}
		catch (Exception e) {
			throw new ItemStreamException("Error while closing item reader", e);
		}
	}

	@Override
	public void open(ExecutionContext executionContext) throws ItemStreamException {<FILL_FUNCTION_BODY>}

	@Override
	public void update(ExecutionContext executionContext) throws ItemStreamException {
		super.update(executionContext);
		if (saveState) {
			Assert.notNull(executionContext, "ExecutionContext must not be null");
			executionContext.putInt(getExecutionContextKey(READ_COUNT), currentItemCount);
			if (maxItemCount < Integer.MAX_VALUE) {
				executionContext.putInt(getExecutionContextKey(READ_COUNT_MAX), maxItemCount);
			}
		}

	}

	/**
	 * Set the flag that determines whether to save internal data for
	 * {@link ExecutionContext}. Only switch this to false if you don't want to save any
	 * state from this stream, and you don't need it to be restartable. Always set it to
	 * false if the reader is being used in a concurrent environment.
	 * @param saveState flag value (default true).
	 */
	public void setSaveState(boolean saveState) {
		this.saveState = saveState;
	}

	/**
	 * The flag that determines whether to save internal state for restarts.
	 * @return true if the flag was set
	 */
	public boolean isSaveState() {
		return saveState;
	}

}

		super.open(executionContext);
		try {
			doOpen();
		}
		catch (Exception e) {
			throw new ItemStreamException("Failed to initialize the reader", e);
		}
		if (!isSaveState()) {
			return;
		}

		if (executionContext.containsKey(getExecutionContextKey(READ_COUNT_MAX))) {
			maxItemCount = executionContext.getInt(getExecutionContextKey(READ_COUNT_MAX));
		}

		int itemCount = 0;
		if (executionContext.containsKey(getExecutionContextKey(READ_COUNT))) {
			itemCount = executionContext.getInt(getExecutionContextKey(READ_COUNT));
		}
		else if (currentItemCount > 0) {
			itemCount = currentItemCount;
		}

		if (itemCount > 0 && itemCount < maxItemCount) {
			try {
				jumpToItem(itemCount);
			}
			catch (Exception e) {
				throw new ItemStreamException("Could not move to stored position on restart", e);
			}
		}

		currentItemCount = itemCount;

	
<methods>public non-sealed void <init>() <variables>