ERXBatchProcessor simplifies operations that change a large number of EOs

ERXBatchProcessor.java

package er.extensions.eof;

import java.util.function.BiConsumer;
import java.util.function.Consumer;

import com.webobjects.eocontrol.EOEditingContext;
import com.webobjects.eocontrol.EOEnterpriseObject;
import com.webobjects.eocontrol.EOFetchSpecification;
import com.webobjects.foundation.NSArray;

/**
 * The {@code ERXBatchProcessor} class implements a mechanism to simplify operations that change a large number of
 * {@code EOEnterpriseObject}s without consuming lots of memory. It does the heavy lifting of fetching objects in
 * batches, locking, unlocking, and recycling the {@code EOEditingContext} to reduce memory usage.
 * <p>
 * Considerations about this class:
 * <ul>
 * <li>It persists changes using one transaction per batch. For this reason, partial changes may end up saved to the
 * database if the code throws an error in the middle of the process.</li>
 * <li>All batches are processed in the current thread. It optimizes for memory usage, not CPU usage.</li>
 * <li>An error interrupts the process, and the remaining batches won't process.</li>
 * </ul>
 * Usage:
 *
 * <pre>
 * EOFetchSpecification fs = ...;
 * int batchSize = 250;
 * int disposalCycleCount = 8; // recycles the editing context after processing 2000 (8 * 250) objects
 * ERXBatchProcessor&lt;Foo&gt; processor = new ERXBatchProcessor&lt;&gt;(fs, batchSize, disposalCycleCount);
 * int count = processor.processObjects(eo -> eo.doSomething());
 * </pre>
 *
 * @author <a href="mailto:[email protected]">Henrique Prange</a>
 *
 * @param <T> the type of {@code EOEnterpriseObject} being processed.
 */
public class ERXBatchProcessor<T extends EOEnterpriseObject> {
    private final EOFetchSpecification fetchSpecification;
    private final int batchSize;
    private final int disposalCycleCount;

    /**
     * Creates a batch processor.
     *
     * @param fetchSpecification the specification used to fetch objects
     * @param batchSize          number of objects to fetch in a given batch
     * @param disposalCycleCount the number of iterations after which the editing context must be disposed and recycled
     */
    public ERXBatchProcessor(EOFetchSpecification fetchSpecification, int batchSize, int disposalCycleCount) {
        this.fetchSpecification = fetchSpecification;
        this.batchSize = batchSize;
        this.disposalCycleCount = disposalCycleCount;
    }

    /**
     * Processes changes to {@code EOEnterpriseObject}s in batches.
     *
     * @param eosConsumer a function to handle a batch of objects
     * @return Returns the number of processed objects after each batch has been processed.
     */
    public int processBatches(Consumer<NSArray<T>> eosConsumer) {
        return processBatches((ec, eos) -> eosConsumer.accept(eos));
    }

    /**
     * Processes changes to {@code EOEnterpriseObject}s in batches.
     *
     * @param eosConsumer a function to handle a batch of objects
     * @return Returns the number of processed objects after each batch has been processed.
     */
    public int processBatches(BiConsumer<EOEditingContext, NSArray<T>> eosConsumer) {
        int processedObjectsCount = 0;

        EOEditingContext ec = ERXEC.newEditingContext();
        ec.lock();

        try {
            ERXFetchSpecificationBatchIterator<T> batchFetchSpecification = new ERXFetchSpecificationBatchIterator<>(fetchSpecification, ec);
            batchFetchSpecification.setBatchSize(batchSize);

            while (batchFetchSpecification.hasNextBatch()) {
                NSArray<T> batch = batchFetchSpecification.nextBatch();

                eosConsumer.accept(ec, batch);

                ec.saveChanges();

                processedObjectsCount += batch.size();

                if (shouldRecycleEditingContext(batchFetchSpecification.currentBatchIndex())) {
                    ec.unlock();
                    ec.dispose();
                    ec = ERXEC.newEditingContext();
                    ec.lock();

                    batchFetchSpecification.setEditingContext(ec);
                }
            }

            return processedObjectsCount;
        } finally {
            ec.unlock();
            ec.dispose();
        }
    }

    /**
     * Processes changes to {@code EOEnterpriseObject}s in batches.
     *
     * @param eoConsumer a function to handle one object
     * @return Returns the number of processed objects after each batch has been processed.
     */
    public int processObjects(Consumer<T> eoConsumer) {
        return processBatches(eos -> eos.forEach(eoConsumer::accept));
    }

    /**
     * Processes changes to {@code EOEnterpriseObject}s in batches.
     *
     * @param eoConsumer a function to handle one object
     * @return Returns the number of processed objects after each batch has been processed.
     */
    public int processObjects(BiConsumer<EOEditingContext, T> eoConsumer) {
        return processBatches((ec, eos) -> eos.forEach(eo -> eoConsumer.accept(ec, eo)));
    }

    private boolean shouldRecycleEditingContext(int index) {
        return index % disposalCycleCount == 0;
    }
}