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

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 org.slf4j.Logger;
import org.slf4j.LoggerFactory;

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 {
                    logger.debug("[{}] waiting ...", getId());
                    final HarvestJobImpl job = queue.take();
                    logger.debug("[{}] got job {}", getId(), job.getId());
                    try {
                        while (true) {
                            logger.debug("[{}] processing job {} ...",
                                    getId(), job.getId());
                            worker.process(job, maxNetworkRetryCount);
                            if (job.isRunning()) {
                                if (job.getDelay(TimeUnit.MILLISECONDS) > 0) {
                                    queue.offer(job);
                                    logger.debug("[{}] requeuing job {} ...",
                                            getId(), job.getId());
                                } else {
                                    continue;
                                }
                            }
                            break;
                        } // while
                    } catch (HarvesterException e) {
                        e.printStackTrace();
                    }
                } catch (InterruptedException e) {
                    /* IGNORE */
                }
            } // while
            logger.debug("[{}] exiting ...", getId());
        }


        void shutdown() {
            loop.set(false);
            this.interrupt();
        }
    }
    private static final Logger logger =
            LoggerFactory.getLogger(ParallelHarvester.class);
    private final HashSet<WorkerThread> workers =
            new HashSet<WorkerThread>();
    private final DelayQueue<HarvestJobImpl> queue =
            new DelayQueue<HarvestJobImpl>();
    private final AtomicLong nextId = new AtomicLong(0);


    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) {
                logger.error("interrupted while shutting down");
            }
        }
    }


    @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(long id) throws HarvesterException {
        for (HarvestJobImpl job : queue) {
            if (id == job.getId()) {
                queue.remove(job);
                return job;
            }
        }
        return null;
    }

} // class ParallelHarvester