@@ -0,0 +1,2805 @@
+package com.marklogic.batch.item.reader;%0A%0Aimport com.marklogic.client.DatabaseClient;%0Aimport com.marklogic.client.datamovement.DataMovementManager;%0Aimport com.marklogic.client.datamovement.QueryBatch;%0Aimport com.marklogic.client.datamovement.QueryBatchListener;%0Aimport com.marklogic.client.datamovement.QueryBatcher;%0Aimport com.marklogic.client.document.DocumentPage;%0Aimport com.marklogic.client.document.DocumentRecord;%0Aimport com.marklogic.client.document.GenericDocumentManager;%0Aimport com.marklogic.client.document.ServerTransform;%0Aimport com.marklogic.client.query.StructuredQueryDefinition;%0Aimport org.springframework.batch.item.*;%0A%0Aimport java.util.Queue;%0Aimport java.util.concurrent.ConcurrentLinkedQueue;%0A%0Apublic class MarkLogicItemReader implements QueryBatchListener, ItemReader%3CDocumentRecord%3E, ItemStream %7B%0A%0A    protected DatabaseClient databaseClient;%0A    protected QueryBatcher batcher;%0A    protected StructuredQueryDefinition query;%0A    protected Queue%3CDocumentRecord%3E documentQueue;%0A    protected GenericDocumentManager docMgr;%0A    protected DataMovementManager dataMovementManger;%0A    protected ServerTransform serverTransform;%0A%0A    public MarkLogicItemReader(DatabaseClient databaseClient, StructuredQueryDefinition sqd, ServerTransform serverTransform) %7B%0A        this.databaseClient = databaseClient;%0A        this.query = query;%0A        this.serverTransform = serverTransform;%0A        documentQueue = new ConcurrentLinkedQueue%3CDocumentRecord%3E();%0A        docMgr = databaseClient.newDocumentManager();%0A    %7D%0A%0A    @Override%0A    public void open(ExecutionContext executionContext) throws ItemStreamException %7B%0A        dataMovementManger = databaseClient.newDataMovementManager();%0A        batcher = dataMovementManger.newQueryBatcher(query)%0A                    .withConsistentSnapshot()%0A                    .withBatchSize(10)%0A                    .withThreadCount(1)%0A                    .onUrisReady(this);%0A    %7D%0A%0A    @Override%0A    public void update(ExecutionContext executionContext) throws ItemStreamException %7B%0A    %7D%0A%0A    @Override%0A    public void close() throws ItemStreamException %7B%0A        batcher.awaitCompletion();%0A        dataMovementManger.release();%0A    %7D%0A%0A    @Override%0A    public DocumentRecord read() throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException %7B%0A        if (documentQueue.isEmpty() && batcher.isStopped()) %7B%0A            return null;%0A        %7D else if (documentQueue.isEmpty() && batcher.isStarted()) %7B%0A            batcher.awaitCompletion();%0A        %7D%0A        return documentQueue.poll();%0A    %7D%0A%0A    @Override%0A    public void processEvent(QueryBatch batch) %7B%0A        DocumentPage page = docMgr.read(serverTransform, batch.getItems());%0A        while (page.hasNext()) %7B%0A            documentQueue.add(page.next());%0A        %7D%0A    %7D%0A%7D%0A

Check in MarkLogicItemReader