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

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

import java.io.IOException;
import java.io.InputStream;
import java.net.URI;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.List;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;

import javax.xml.stream.XMLStreamReader;

import eu.clarin.cmdi.oai.harvester.HarvestHandler;
import eu.clarin.cmdi.oai.harvester.HarvestJob;
import eu.clarin.cmdi.oai.harvester.HarvesterException;
import eu.clarin.cmdi.oai.harvester.Header;
import eu.clarin.cmdi.oai.harvester.MetadataFormat;
import eu.clarin.cmdi.oai.harvester.Repository;
import eu.clarin.cmdi.oai.harvester.Repository.DeletedNotion;
import eu.clarin.cmdi.oai.harvester.Repository.Granularity;

final class HarvestJobImpl implements HarvestJob, Delayed {
    public enum Task {
        IDENTIFY_REPOSITORY,
        ENUMERATE_METADATA_FORMATS,
        HARVEST_RECORDS
    }
    private static final class StatisticsImpl implements Statistics {
        private final String prefix;
        private long recordCount = 0;
        private long requestCount = 0;
        private long resumptionCount = 0;
        private long bytedTransferred = 0;
        private Date latestDatestamp = null;
        
        private StatisticsImpl(String prefix) {
            this.prefix = prefix;
        }

        @Override
        public String getPrefix() {
            return prefix;
        }

        @Override
        public long getRecordCount() {
            return recordCount;
        }

        private void incRecordCount() {
            this.recordCount++;
        }

        @Override
        public long getRequestCount() {
            return requestCount;
        }

        private void incRequestCount() {
            this.requestCount++;
        }

        @Override
        public long getResumptionCount() {
            return resumptionCount;
        }

        private void incResumptionCount() {
            this.resumptionCount++;
        }

        @Override
        public long getBytesTransferred() {
            return bytedTransferred;
        }
    
        private void incBytesTransferred(long bytes) {
            this.bytedTransferred += bytes;
        }

        @Override
        public Date getLatestDatestamp() {
            return latestDatestamp;
        }
        
        private void updateLatestDatestmp(Date latestDatestamp) {
            if (this.latestDatestamp == null) {
                this.latestDatestamp = latestDatestamp;
            } else {
                if (latestDatestamp.compareTo(this.latestDatestamp) > 0) {
                    this.latestDatestamp = latestDatestamp;
                }
            }
        }
    } // inner class Statistics
    private final AbstractHarvester harvester;
    private final long id;
    private final URI repositoryURI;
    private final HarvestHandler handler;
    private State state = State.NEW;
    private Date from;
    private Date until;
    private String set;
    private List<String> metadataPrefixes;
    private String repositoryName;
    private String baseURL;
    private String protocolVersion;
    private List<String> adminEmail = Collections.emptyList();
    private String earliestTimestamp;
    private DeletedNotion deletedNotion;
    private Granularity granularity;
    private int compressionMask;
    private List<MetadataFormat> metadataFormats;
    private long totalRecordCount = 0;
    private long totalRequestCount = 0;
    private long totalBytesTransferred = 0;
    private long timeTotal = 0;
    private long timeNetwork = 0;
    private long timeDelayed = 0;
    private long timeProcessing = 0;
    private long startTimestamp;
    private List<Statistics> statistics = null;
    private Task task = Task.IDENTIFY_REPOSITORY;
    private long delayUntil = 0;
    private int networkRetryCount = 0;
    private List<String> prefixWorklist = null;
    private String resumptionToken = null;

    HarvestJobImpl(AbstractHarvester harvester, long id, URI repositoryURI,
            HarvestHandler handler) {
        this.harvester = harvester;
        this.id = id;
        this.repositoryURI = repositoryURI;
        this.handler = handler;
    }

    @Override
    public long getId() {
        return id;
    }

    @Override
    public State getState() {
        return state;
    }

    @Override
    public boolean isRunning() {
        return state == HarvestJob.State.RUNNING;
    }

    @Override
    public long getTotelRecordCount() {
        return totalRecordCount;
    }

    @Override
    public long getTotelRequestCount() {
        return totalRequestCount;
    }

    @Override
    public long getTotalBytesTransferred() {
        return totalBytesTransferred;
    }

    @Override
    public long getTotalTime() {
        return timeTotal;
    }

    @Override
    public long getNetworkTime() {
        return timeNetwork;
    }
    
    @Override
    public long getWaitTime() {
        return timeDelayed;
    }

    @Override
    public long getProcessingTime() {
        return timeProcessing;
    }

    
    @Override
    public HarvestHandler getHarvestHandler() {
        return handler;
    }

    @Override
    public Date getFrom() {
        return from;
    }
    
    @Override
    public void setFrom(Date from) {
        if (state != State.NEW) {
            throw new IllegalStateException("state != new");
        }
        this.from = from;
    }
    
    @Override
    public Date getUntil() {
        return until;
    }
    
    @Override
    public void setUntil(Date until) {
        if (state != State.NEW) {
            throw new IllegalStateException("state != new");
        }
        this.until = until;
    }

    @Override
    public String getSet() {
        return set;
    }

    @Override
    public void setSet(String set) {
        if (state != State.NEW) {
            throw new IllegalStateException("state != new");
        }
        if ((set != null) && !set.isEmpty()) {
            this.set = set;
        } else {
            this.set = null;
        }
    }

    @Override
    public List<String> getMetadataPrefixes() {
        return metadataPrefixes;
    }
    
    @Override
    public void setMetadataPrefixes(List<String> metadataPrefixes) {
        if (state != State.NEW) {
            throw new IllegalStateException("state != new");
        }
        if ((metadataPrefixes != null) && !metadataPrefixes.isEmpty()) {
            this.metadataPrefixes = metadataPrefixes;
        } else {
            this.metadataPrefixes = null;
        }
    }
    
    @Override
    public Statistics getStatistics(String prefix) {
        if (prefix == null) {
            throw new NullPointerException("prefix == null");
        }
        if ((state == State.NEW) || (state == State.RUNNING)) {
            throw new IllegalStateException("invalid state");
        }
        for (Statistics stats : statistics) {
            if (prefix.equals(stats.getPrefix())) {
                return stats;
            }
        }
        return null;
    }

    @Override
    public List<Statistics> getStatistics() {
        if ((state == State.NEW) || (state == State.RUNNING)) {
            throw new IllegalStateException("invalid state");
        }
        if ((statistics != null) && !statistics.isEmpty()) {
            return Collections.unmodifiableList(statistics);
        } else {
            return null;
        }
    }

    @Override
    public void run() throws HarvesterException {
        harvester.doRunJob(this);
    }

    @Override
    public void cancel() throws HarvesterException {
        setState(State.CANCELED);
        harvester.doCancelJob(this);
    }

    @Override
    public long getDelay(TimeUnit unit) {
        return unit.convert((delayUntil - System.currentTimeMillis()),
                TimeUnit.MILLISECONDS);
    }

    @Override
    public int compareTo(Delayed o) {
        final long me = getDelay(TimeUnit.MICROSECONDS);
        final long they = o.getDelay(TimeUnit.MILLISECONDS);
        if (me > they) {
            return 1;
        } else if (me < they) {
            return -1;
        } else {
            return 0;
        }
    }

    @Override
    public String toString() {
        StringBuilder sb = new StringBuilder();
        sb.append(getClass().getCanonicalName());
        sb.append("[id=");
        sb.append(id);
        sb.append(", state=");
        sb.append(state);
        sb.append(", task=");
        sb.append(task);
        sb.append("]");
        return sb.toString();
    }

    public URI getRepositoryURI() {
        return repositoryURI;
    }

    void setState(State state) {
        if ((this.state != State.RUNNING) && (state == State.RUNNING)) {
            startTimestamp = System.currentTimeMillis();
        }
        if ((this.state == State.RUNNING) && (state != State.RUNNING)) {
            timeTotal = System.currentTimeMillis() - startTimestamp;
        }
        this.state = state;
    }

    void setRepositoryName(String repositoryName) {
        this.repositoryName = repositoryName;
    }

    void setBaseURL(String baseURL) {
        this.baseURL = baseURL;
    }

    void setProtocolVersion(String protocolVersion) {
        this.protocolVersion = protocolVersion;
    }
    
    void setAdminEmail(List<String> adminEmail) {
        this.adminEmail = adminEmail;
    }

    void setEarliestTimestamp(String earliestTimestamp) {
        this.earliestTimestamp = earliestTimestamp;
    }

    void setDeletedNotion(DeletedNotion deletedNotion) {
        this.deletedNotion = deletedNotion;
    }

    void setGranularity(Granularity granularity) {
        this.granularity = granularity;
    }

    Granularity getGranularity() {
        return granularity;
    }

    void setCompressionMask(int compressionMask) {
        this.compressionMask = compressionMask;
    }
    
    int getCompressionMask() {
        return compressionMask;
    }

    void incRequestCount() {
        totalRequestCount++;
        final StatisticsImpl stats = getCurrentStatistics();
        if (stats != null) {
            stats.incRequestCount();
        }
    }

    void incResumptionCount() {
        final StatisticsImpl stats = getCurrentStatistics();
        if (stats != null) {
            stats.incResumptionCount();
        }
    }

    void addToNetworkTime(long delta) {
        timeNetwork += delta;
    }

    void finishRequest(long bytesTransferred, long timeProcessingDelta) {
        timeProcessing += timeProcessingDelta;
        totalBytesTransferred += bytesTransferred;
        final StatisticsImpl stats = getCurrentStatistics();
        if (stats != null) {
            stats.incBytesTransferred(bytesTransferred);
        }
    }

    boolean isState(State state) {
        return this.state == state;
    }

    Task getTask() {
        return task;
    }

    void setTask(Task task) {
        this.task = task;
    }

    String getCurrentPrefix() {
        if (prefixWorklist == null) {
            prefixWorklist = new ArrayList<String>();
            if (metadataPrefixes != null) {
                prefixWorklist.addAll(metadataPrefixes);
            } else if (metadataFormats != null) {
                for (MetadataFormat format : metadataFormats) {
                    prefixWorklist.add(format.getPrefix());
                }
            } else {
                prefixWorklist.add("oai_dc");
            }
        }
        return prefixWorklist.isEmpty() ? null : prefixWorklist.get(0);
    }

    boolean removeCurrentPrefix() {
        prefixWorklist.remove(0);
        return !prefixWorklist.isEmpty();
    }

    String getResumptionToken() {
        return resumptionToken;
    }
    
    void setResumptionToken(String resumptionToken) {
        this.resumptionToken = resumptionToken;
    }

    boolean isHarvestingNewPrefix() {
        return (resumptionToken == null) && (networkRetryCount == 0);
    }

    int incNetworkRetryCount() {
        return ++networkRetryCount;
    }
    
    int getNetworkRetryCount() {
        return networkRetryCount;
    }

    void resetNetworkRetryCount() {
        delayUntil = 0;
        networkRetryCount = 0;
    }

    void setNetworkRequestDelay(long delay) {
        this.timeDelayed += delay;
        this.delayUntil = System.currentTimeMillis() + delay;
    }

    void onStartListingRecords() {
        final String prefix = getCurrentPrefix();
        if (statistics == null) {
            statistics = new ArrayList<Statistics>();
        }
        statistics.add(0, new StatisticsImpl(prefix));

        if (handler != null) {
            handler.onStartListingRecords(prefix);
        }
    }

    void onFinishListingRecords() {
        Date latestDatestamp = null;

        final StatisticsImpl stats = getCurrentStatistics();
        if (stats != null) {
            latestDatestamp = stats.getLatestDatestamp();
        }
        if (handler != null) {
            handler.onFinishListingRecords(getCurrentPrefix(), latestDatestamp);
        }
        // reset ...
        resumptionToken = null;
    }

    void onRecord(Header header) {
        totalRecordCount++;
        StatisticsImpl stats = getCurrentStatistics();
        if (stats != null) {
            stats.incRecordCount();
            stats.updateLatestDatestmp(header.getDatestamp());
        }
        if (handler != null) {
            handler.onRecord(header);
        }
    }

    void onRecordMetadata(Header header, XMLStreamReader reader) {
        if (handler != null) {
            handler.onRecordMetadata(header, reader);
        }
    }

    void onRecordAbout(Header header, XMLStreamReader reader) {
        if (handler != null) {
            handler.onRecordAbout(header, reader);
        }
    }

    void onIdentify() {
        if (handler != null) {
            final Repository repository = new Repository() {
                @Override
                public String getRepositoryName() {
                    return repositoryName;
                }
                
                @Override
                public String getProtocolVersion() {
                    return protocolVersion;
                }
                
                @Override
                public Granularity getGranularity() {
                    return granularity;
                }
                
                @Override
                public String getEarliestTimestamp() {
                    return earliestTimestamp;
                }
                
                @Override
                public DeletedNotion getDeletedNotion() {
                    return deletedNotion;
                }
                
                @Override
                public int getCompressionMask() {
                    return compressionMask;
                }
                
                @Override
                public String getBaseURL() {
                    return baseURL;
                }
                
                @Override
                public List<String> getAdminEmail() {
                    return adminEmail;
                }
            };
            handler.onIdentify(repository);
        }
    }

    void onListMetadataFormats(List<MetadataFormat> metadataFormats) {
        this.metadataFormats = metadataFormats;
        if (handler != null) {
            handler.onListMetadataFormats(this.metadataFormats);
        }
    }

    InputStream wrap(InputStream stream) throws IOException {
        if (handler != null) {
            return handler.wrap(stream);
        } else {
            return stream;
        }
    }

    private StatisticsImpl getCurrentStatistics() {
        if ((statistics != null) && !statistics.isEmpty()) {
            return (StatisticsImpl) statistics.get(0);
        }
        return null;
    }

} // class HarvestJobImpl