source: OAIHarvester/trunk/OAIHarvester/src/main/java/eu/clarin/cmdi/oai/harvester/impl/ParallelHarvester.java @ 1140

package eu.clarin.cmdi.oai.harvester.impl;

import java.net.URI;
import java.util.HashSet;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;

import eu.clarin.cmdi.oai.harvester.HarvestHandler;
import eu.clarin.cmdi.oai.harvester.HarvestJob;
import eu.clarin.cmdi.oai.harvester.Harvester;
import eu.clarin.cmdi.oai.harvester.HarvesterException;

public class ParallelHarvester extends AbstractHarvester {
    private final class WorkerThread extends Thread {
        private final HarvestWorker worker =
            new HarvestWorker(ParallelHarvester.this);
        private AtomicBoolean loop = new AtomicBoolean(true);
        
        @Override
        public void run() {
            while (loop.get()) {
                try {
                    System.err.println(getId() + " waiting ...");
                    HarvestJobImpl job = queue.take();
                    System.err.println(getId() + " got job " + job.getId());
                    try {
                        while (true) {
                            System.err.println(getId() + " process job " +
                                    job.getId());
                            worker.process(job, maxNetworkRetryCount);
                            if (job.isRunning()) {
                                if (job.getDelay(TimeUnit.MILLISECONDS) > 0) {
                                    queue.offer(job);
                                    System.err.println(getId() +
                                            " requeuing job " + job.getId());
                                } else {
                                    continue;
                                }
                            }
                            break;
                        } // while
                    } catch (HarvesterException e) {
                        e.printStackTrace();
                    }
                } catch (InterruptedException e) {
                    /* IGNORE */
                }
            } // while
            System.err.println(getId() + " ... exit");
        }

        void shutdown() {
            loop.set(false);
            this.interrupt();
        }
    }

    private final HashSet<WorkerThread> workers = new HashSet<WorkerThread>();
    private final DelayQueue<HarvestJobImpl> queue = new DelayQueue<HarvestJobImpl>();
    private AtomicLong nextId = new AtomicLong();

    private ParallelHarvester() {
        for (int i = 0; i < 4; i++) {
            workers.add(new WorkerThread());
        }
    }

    @Override
    public HarvestJob createJob(URI repositoryURI, HarvestHandler handler)
            throws HarvesterException {
        return new HarvestJobImpl(this, nextId.incrementAndGet(),
                repositoryURI, handler);
    }

    public static Harvester newInstance() {
        ParallelHarvester harvester = new ParallelHarvester();
        for (WorkerThread w : harvester.workers) {
            w.start();
        }
        return harvester;
    }

    @Override
    public void shutdown() {
        for (WorkerThread w : workers) {
            w.shutdown();
            try {
                w.join();
            } catch (InterruptedException e) {
                System.err.println("join fail!");
            }
        }
    }


    @Override
    void doRunJob(HarvestJobImpl job) throws HarvesterException {
        job.setState(HarvestJob.State.RUNNING);
        queue.offer(job);
    }

    @Override
    void doCancelJob(HarvestJobImpl job) throws HarvesterException {
        queue.remove(job);
    }

    @Override
    protected HarvestJob doFindJob(String id) throws HarvesterException {
        for (HarvestJobImpl job : queue) {
            if (id.equals(job.getId())) {
                queue.remove(job);
                return job;
            }
        }
        return null;
    }

} // class ParallelHarvester