Changeset 3011 for OAIHarvester

Timestamp:

06/10/13 19:50:23 (11 years ago)

Author:

oschonef

Message:

• update Woodstox, SLF4J and Apache Commons HTTP client dependencies
• several interoperability fixes (including relaxed behavior)
• several bug fixes

Location:

OAIHarvester/trunk/OAIHarvester

Files:

• pom.xml (modified) (4 diffs)
• src/main/java/eu/clarin/cmdi/oai/harvester/ProtocolError.java (modified) (3 diffs)
• src/main/java/eu/clarin/cmdi/oai/harvester/impl/AbstractHarvester.java (modified) (10 diffs)
• src/main/java/eu/clarin/cmdi/oai/harvester/impl/HarvestJobImpl.java (modified) (46 diffs)
• src/main/java/eu/clarin/cmdi/oai/harvester/impl/HarvestWorker.java (modified) (26 diffs)
• src/main/java/eu/clarin/cmdi/oai/harvester/impl/ParallelHarvester.java (modified) (9 diffs)
• src/main/java/eu/clarin/cmdi/oai/harvester/impl/Response.java (modified) (21 diffs)
• src/main/java/eu/clarin/cmdi/oai/harvester/impl/SimpleHarvester.java (modified) (7 diffs)
• src/main/java/eu/clarin/cmdi/oai/harvester/impl/XmlStreamReaderProxy.java (modified) (19 diffs)
• src/test/java/eu/clarin/cmdi/oai/harvester/HarvesterTest.java (modified) (5 diffs)

OAIHarvester/trunk/OAIHarvester/pom.xml

@@ -61 +61 @@
         <groupId>org.apache.httpcomponents</groupId>
         <artifactId>httpclient</artifactId>
-        <version>4.1.3</version>
+        <version>4.2.5</version>
     </dependency>
 
@@ -67 +67 @@
         <groupId>org.codehaus.woodstox</groupId>
         <artifactId>woodstox-core-lgpl</artifactId>
-        <version>4.1.3</version>
+        <version>4.2.0</version>
     </dependency>
 
@@ -85 +85 @@
     <dependency>
       <groupId>org.slf4j</groupId>
-      <artifactId>slf4j-jdk14</artifactId>
+      <artifactId>slf4j-log4j12</artifactId>
       <version>${slf4j.version}</version>
       <scope>test</scope>
@@ -98 +98 @@
     <maven.compiler.target>1.6</maven.compiler.target>
     <!-- versions of common dependencies -->
-    <slf4j.version>1.6.4</slf4j.version>
+    <slf4j.version>1.7.2</slf4j.version>
   </properties>
 

OAIHarvester/trunk/OAIHarvester/src/main/java/eu/clarin/cmdi/oai/harvester/ProtocolError.java

@@ -15 +15 @@
         public static Code fromString(String s) {
             if ("cannotDisseminateFormat".equals(s)) {
-               return CANNOT_DISSERMINATE_FORMAT;
+                return CANNOT_DISSERMINATE_FORMAT;
             } else if ("idDoesNotExist".equals(s)) {
                 return ID_DOES_NOR_EXIST;
@@ -34 +34 @@
             }
         }
-    }
+    } // enum Code
 
     private final Code code;
     private final String message;
-    
+
+
     public ProtocolError(String code, String message) {
-        this.code = Code.fromString(code);
+        this.code    = Code.fromString(code);
         this.message = message;
     }
+
 
     public Code getCode() {
         return code;
     }
-    
+
+
     public String getMessage() {
         return message;
     }
+
 
     public String toString() {
@@ -56 +60 @@
     }
 
+    public boolean isCode(Code code) {
+        if (code == null) {
+            throw new NullPointerException("code == null");
+        }
+        return code.equals(this.code);
+    }
+
 } // class ProtocolError

OAIHarvester/trunk/OAIHarvester/src/main/java/eu/clarin/cmdi/oai/harvester/impl/AbstractHarvester.java

@@ -7 +7 @@
 import java.util.List;
 
+import javax.xml.namespace.QName;
 import javax.xml.stream.XMLInputFactory;
 import javax.xml.stream.XMLStreamException;
-import javax.xml.stream.XMLStreamReader;
 
 import org.codehaus.stax2.XMLInputFactory2;
@@ -22 +22 @@
 import eu.clarin.cmdi.oai.harvester.ext.DescriptionParser;
 
+
 abstract class AbstractHarvester implements Harvester {
     private static final int DEFAULT_MAX_NETWORK_RETRY_COUNT = 5;
     protected int maxNetworkRetryCount = DEFAULT_MAX_NETWORK_RETRY_COUNT;
-    private final XMLInputFactory factory;
-//    private final XMLValidationSchema schema;
+    private final XMLInputFactory2 factory;
+    // private final XMLValidationSchema schema;
     private List<DescriptionParser> descriptionParsers =
-        new ArrayList<DescriptionParser>();
+            new ArrayList<DescriptionParser>();
+
 
     protected AbstractHarvester() {
-        factory = XMLInputFactory2.newInstance();
+        factory = (XMLInputFactory2) XMLInputFactory.newInstance();
 
         // Stax settings
@@ -54 +56 @@
     }
 
+
+    @Override
     public final void registerDescriptionParser(DescriptionParser parser)
             throws HarvesterException {
@@ -61 +65 @@
         if (findDescriptionParser(parser.getNamespaceURI(),
                 parser.getLocalName()) != null) {
-            throw new HarvesterException("description parser for '{" +
-                    parser.getNamespaceURI() + "}" + parser.getLocalName() +
-                    "' was already registered");
+            throw new HarvesterException("description parser for '" +
+                    new QName(parser.getNamespaceURI(), parser.getLocalName()) +
+                            "' was already registered");
         }
         if (descriptionParsers == null) {
@@ -70 +74 @@
         descriptionParsers.add(parser);
     }
-    
+
+
+    @Override
     public abstract HarvestJob createJob(URI repositoryURI,
             HarvestHandler handler) throws HarvesterException;
 
+
+    @Override
     public final HarvestJob createJob(URI repositoryURI)
             throws HarvesterException {
         return createJob(repositoryURI, null);
     }
-    
+
+
     @Override
     public final HarvestJob createJob(String repositoryURI,
@@ -89 +98 @@
     }
 
+
     @Override
     public final HarvestJob createJob(String repositoryURI)
@@ -95 +105 @@
     }
 
+
     @Override
     public final HarvestJob findJob(long id) throws HarvesterException {
         return doFindJob(id);
     }
+
 
     @Override
@@ -109 +121 @@
     }
 
+
     @Override
     public final int getMaxNetworkRetryCount() {
         return maxNetworkRetryCount;
     }
-    
+
+
     @Override
     public final void setMaxNetworkRetryCount(int maxNetworkRetryCount) {
@@ -123 +137 @@
         this.maxNetworkRetryCount = maxNetworkRetryCount;
     }
+
 
     DescriptionParser findDescriptionParser(String namespaceURI,
@@ -137 +152 @@
     }
 
+
     abstract void doRunJob(HarvestJobImpl job) throws HarvesterException;
+
 
     abstract void doCancelJob(HarvestJobImpl job) throws HarvesterException;
 
-    protected abstract HarvestJob doFindJob(long id)
-        throws HarvesterException;
 
-    protected final XMLStreamReader createReader(InputStream in)
+    protected abstract HarvestJob doFindJob(long id) throws HarvesterException;
+
+
+    protected final XMLStreamReader2 createReader(InputStream in)
             throws XMLStreamException {
         XMLStreamReader2 reader =
-            (XMLStreamReader2) factory.createXMLStreamReader(in);
+                (XMLStreamReader2) factory.createXMLStreamReader(in);
         // reader.validateAgainst(schema);
         return reader;

OAIHarvester/trunk/OAIHarvester/src/main/java/eu/clarin/cmdi/oai/harvester/impl/HarvestJobImpl.java

@@ -23 +23 @@
 import eu.clarin.cmdi.oai.harvester.Repository.Granularity;
 
+
 final class HarvestJobImpl implements HarvestJob, Delayed {
     public enum Task {
-        IDENTIFY_REPOSITORY,
-        ENUMERATE_METADATA_FORMATS,
-        HARVEST_RECORDS
-    }
+        IDENTIFY_REPOSITORY, ENUMERATE_METADATA_FORMATS, HARVEST_RECORDS
+    }
+
     private static final class StatisticsImpl implements Statistics {
         private final String prefix;
@@ -36 +36 @@
         private long bytedTransferred = 0;
         private Date latestDatestamp = null;
-        
+
+
         private StatisticsImpl(String prefix) {
             this.prefix = prefix;
         }
+
 
         @Override
@@ -46 +48 @@
         }
 
+
         @Override
         public long getRecordCount() {
@@ -51 +54 @@
         }
 
+
         private void incRecordCount() {
             this.recordCount++;
         }
+
 
         @Override
@@ -60 +65 @@
         }
 
+
         private void incRequestCount() {
             this.requestCount++;
         }
+
 
         @Override
@@ -69 +76 @@
         }
 
+
         private void incResumptionCount() {
             this.resumptionCount++;
         }
+
 
         @Override
@@ -77 +86 @@
             return bytedTransferred;
         }
-    
+
+
         private void incBytesTransferred(long bytes) {
             this.bytedTransferred += bytes;
         }
+
 
         @Override
@@ -86 +97 @@
             return latestDatestamp;
         }
-        
+
+
         private void updateLatestDatestmp(Date latestDatestamp) {
             if (this.latestDatestamp == null) {
@@ -97 +109 @@
         }
     } // inner class Statistics
+
     private final AbstractHarvester harvester;
     private final long id;
@@ -131 +144 @@
     private String resumptionToken = null;
 
+
     HarvestJobImpl(AbstractHarvester harvester, long id, URI repositoryURI,
             HarvestHandler handler) {
-        this.harvester = harvester;
-        this.id = id;
+        this.harvester     = harvester;
+        this.id            = id;
         this.repositoryURI = repositoryURI;
-        this.handler = handler;
-    }
+        this.handler       = handler;
+    }
+
 
     @Override
@@ -144 +159 @@
     }
 
+
     @Override
     public State getState() {
@@ -149 +165 @@
     }
 
+
     @Override
     public boolean isRunning() {
@@ -154 +171 @@
     }
 
+
     @Override
     public long getTotelRecordCount() {
@@ -159 +177 @@
     }
 
+
     @Override
     public int getTotelRequestCount() {
@@ -164 +183 @@
     }
 
+
     @Override
     public long getTotalBytesTransferred() {
@@ -169 +189 @@
     }
 
+
     @Override
     public long getTotalTime() {
@@ -174 +195 @@
     }
 
+
     @Override
     public long getNetworkTime() {
         return timeNetwork;
     }
-    
+
+
     @Override
     public long getWaitTime() {
@@ -184 +207 @@
     }
 
+
     @Override
     public long getProcessingTime() {
@@ -189 +213 @@
     }
 
-    
+
     @Override
     public HarvestHandler getHarvestHandler() {
@@ -195 +219 @@
     }
 
+
     @Override
     public Date getFrom() {
         return from;
     }
-    
+
+
     @Override
     public void setFrom(Date from) {
@@ -207 +233 @@
         this.from = from;
     }
-    
+
+
     @Override
     public Date getUntil() {
         return until;
     }
-    
+
+
     @Override
     public void setUntil(Date until) {
@@ -221 +249 @@
     }
 
+
     @Override
     public String getSet() {
         return set;
     }
+
 
     @Override
@@ -238 +268 @@
     }
 
+
     @Override
     public List<String> getMetadataPrefixes() {
         return metadataPrefixes;
     }
-    
+
+
     @Override
     public void setMetadataPrefixes(List<String> metadataPrefixes) {
@@ -254 +286 @@
         }
     }
-    
+
+
     @Override
     public Statistics getStatistics(String prefix) {
@@ -271 +304 @@
     }
 
+
     @Override
     public List<Statistics> getStatistics() {
@@ -283 +317 @@
     }
 
+
     @Override
     public void run() throws HarvesterException {
         harvester.doRunJob(this);
     }
+
 
     @Override
@@ -294 +330 @@
     }
 
+
     @Override
     public long getDelay(TimeUnit unit) {
@@ -299 +336 @@
                 TimeUnit.MILLISECONDS);
     }
+
+
+    @Override
+    public boolean equals(Object obj) {
+        if (obj == null) {
+            return false;
+        }
+        if (obj == this) {
+            return true;
+        }
+        if (!(obj instanceof HarvestJobImpl)) {
+            return false;
+        }
+        final HarvestJobImpl rhs = (HarvestJobImpl) obj;
+        return (this.id == rhs.id) && this.harvester.equals(rhs.harvester);
+    }
+
+    @Override
+    public int hashCode() {
+        return super.hashCode();
+    }
+    
 
     @Override
@@ -312 +371 @@
         }
     }
+
 
     @Override
@@ -327 +387 @@
     }
 
+
     public URI getRepositoryURI() {
         return repositoryURI;
     }
+
 
     void setState(State state) {
@@ -341 +403 @@
     }
 
+
     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(Date 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 setDescriptions(List<Description> descriptions) {
         this.descriptions = descriptions;
     }
+
 
     void incRequestCount() {
@@ -393 +467 @@
     }
 
+
     void incResumptionCount() {
         final StatisticsImpl stats = getCurrentStatistics();
@@ -400 +475 @@
     }
 
+
     void addToNetworkTime(long delta) {
         timeNetwork += delta;
     }
+
 
     void finishRequest(long bytesTransferred, long timeProcessingDelta) {
@@ -413 +490 @@
     }
 
+
     boolean isState(State state) {
         return this.state == state;
     }
 
+
     Task getTask() {
         return task;
     }
 
+
     void setTask(Task task) {
         this.task = task;
     }
+
 
     String getCurrentPrefix() {
@@ -441 +522 @@
     }
 
+
     boolean removeCurrentPrefix() {
         prefixWorklist.remove(0);
@@ -446 +528 @@
     }
 
+
     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() {
@@ -471 +559 @@
     }
 
+
     void setNetworkRequestDelay(long delay) {
         this.timeDelayed += delay;
         this.delayUntil = System.currentTimeMillis() + delay;
     }
+
 
     void onStartListingRecords() {
@@ -488 +578 @@
     }
 
+
     void onFinishListingRecords() {
         Date latestDatestamp = null;
@@ -501 +592 @@
         resumptionToken = null;
     }
+
 
     void onRecord(Header header) {
@@ -514 +606 @@
     }
 
+
     void onRecordMetadata(Header header, XMLStreamReader reader) {
         if (handler != null) {
@@ -520 +613 @@
     }
 
+
     void onRecordAbout(Header header, XMLStreamReader reader) {
         if (handler != null) {
@@ -526 +620 @@
     }
 
+
     void onIdentify() {
         if (handler != null) {
@@ -533 +628 @@
                     return repositoryName;
                 }
-                
+
+
                 @Override
                 public String getProtocolVersion() {
                     return protocolVersion;
                 }
-                
+
+
                 @Override
                 public Granularity getGranularity() {
                     return granularity;
                 }
-                
+
+
                 @Override
                 public Date 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 != null) ?
-                            Collections.unmodifiableList(adminEmail) : null;
-                }
+                    return (adminEmail != null) ? Collections
+                            .unmodifiableList(adminEmail) : null;
+                }
+
 
                 @Override
                 public List<Description> getDescriptions() {
-                    return (descriptions != null) ?
-                            Collections.unmodifiableList(descriptions) : null;
+                    return (descriptions != null) ? Collections
+                            .unmodifiableList(descriptions) : null;
                 }
             };
@@ -579 +682 @@
         }
     }
+
 
     void onListMetadataFormats(List<MetadataFormat> metadataFormats) {
@@ -588 +692 @@
     }
 
+
     InputStream wrap(InputStream stream) throws IOException {
         if (handler != null) {
@@ -596 +701 @@
     }
 
+
     private StatisticsImpl getCurrentStatistics() {
         if ((statistics != null) && !statistics.isEmpty()) {

OAIHarvester/trunk/OAIHarvester/src/main/java/eu/clarin/cmdi/oai/harvester/impl/HarvestWorker.java

@@ -7 +7 @@
 import java.util.List;
 
+import javax.xml.namespace.QName;
 import javax.xml.stream.XMLStreamException;
 
@@ -12 +13 @@
 import org.apache.http.HttpResponse;
 import org.apache.http.HttpStatus;
+import org.apache.http.client.HttpClient;
 import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.utils.HttpClientUtils;
 import org.apache.http.impl.client.DefaultHttpClient;
 import org.apache.http.impl.cookie.DateParseException;
@@ -19 +22 @@
 import org.apache.http.params.CoreProtocolPNames;
 import org.apache.http.params.HttpParams;
+import org.apache.http.util.EntityUtils;
 import org.joda.time.DateTime;
 import org.joda.time.DateTimeZone;
@@ -24 +28 @@
 import org.joda.time.format.DateTimeFormat;
 import org.joda.time.format.DateTimeFormatter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import eu.clarin.cmdi.oai.harvester.Description;
 import eu.clarin.cmdi.oai.harvester.HarvestJob.State;
 import eu.clarin.cmdi.oai.harvester.HarvesterException;
+import eu.clarin.cmdi.oai.harvester.HarvesterProtocolErrorException;
 import eu.clarin.cmdi.oai.harvester.MetadataFormat;
+import eu.clarin.cmdi.oai.harvester.ProtocolError;
 import eu.clarin.cmdi.oai.harvester.Repository;
 import eu.clarin.cmdi.oai.harvester.Repository.DeletedNotion;
@@ -34 +42 @@
 import eu.clarin.cmdi.oai.harvester.ext.DescriptionParser;
 import eu.clarin.cmdi.oai.harvester.impl.HarvestJobImpl.Task;
+
 
 public class HarvestWorker {
@@ -48 +57 @@
     private static final String COMPRESSION_GZIP = "gzip";
     private static final String VERB_IDENTIFY = "Identify";
-    private static final String VERB_LIST_METADATAFORMATS =
-        "ListMetadataFormats";
+    private static final String VERB_LIST_METADATAFORMATS = "ListMetadataFormats";
     private static final String VERB_LIST_RECORDS = "ListRecords";
     private static final String HEADER_ACCEPT_ENCODING = "Accept-Encoding";
     private static final String HEADER_CONNECTION = "Connection";
-    private static final String[] DATEFORMATS_DAYS =
-        { "yyyy'-'MM'-'dd" };
-    private static final String[] DATEFORMATS_FULL =
-        { "yyyy'-'MM'-'dd'T'HH':'mm':'ss'Z'", "yyyy'-'MM'-'dd" };
+    private static final String DATEFORMAT_SECONDS = "yyyy'-'MM'-'dd'T'HH':'mm':'ss'Z'";
+    private static final String DATEFORMAT_DAYS = "yyyy'-'MM'-'dd";
+    private static final String[] DATEFORMATS =
+        { DATEFORMAT_SECONDS, DATEFORMAT_DAYS };
+    private static final Logger logger =
+            LoggerFactory.getLogger(HarvestWorker.class);
     private final AbstractHarvester harvester;
-    private final DefaultHttpClient client;
+    private final HttpClient client;
     private final RequestUriBuilder uriBuilder;
     private final HeaderImpl header;
     private final XmlStreamReaderProxy readerWrapper;
     private long delay;
+
 
     HarvestWorker(AbstractHarvester harvester) {
@@ -69 +80 @@
         client = new DefaultHttpClient();
         client.getParams().setParameter(CoreProtocolPNames.USER_AGENT,
-                    "eu.clarin.cmdi.oai.Harvester/0.0.1");
+                "eu.clarin.cmdi.oai.Harvester/0.0.1");
         // request stuff
         uriBuilder = new RequestUriBuilder();
         // other stuff
-        this.header = new HeaderImpl();
+        this.header        = new HeaderImpl();
         this.readerWrapper = new XmlStreamReaderProxy();
     }
+
 
     long process(HarvestJobImpl job, int maxNetworkTries)
@@ -115 +127 @@
                 } else if ((job.getResumptionToken() != null) && (delay == 0)) {
                     /*
-                     * special case: request successful, but list was split
-                     * die to OAI flow control
+                     * special case: request successful, but list was split due
+                     * to OAI flow control
                      */
                     ok = true;
@@ -151 +163 @@
     }
 
+
     private boolean doIdentifyTask(HarvestJobImpl job) throws IOException,
             XMLStreamException, HarvesterException {
@@ -171 +184 @@
             s = response.readContent(OAI_NS, "protocolVersion", true);
             if (!PROTOCOL_VERSION.equals(s)) {
-                throw new HarvesterException("unsupported protocol version: " +
-                        s);
+                throw new HarvesterException(
+                        "unsupported protocol version: " + s);
             }
             job.setProtocolVersion(PROTOCOL_VERSION);
@@ -178 +191 @@
             List<String> adminEmail = null;
             do {
-                s = response.readContent(OAI_NS, "adminEmail",
-                        (adminEmail == null));
+                s = response.readContent(OAI_NS,
+                        "adminEmail", (adminEmail == null));
                 if (s != null) {
                     if (adminEmail == null) {
@@ -190 +203 @@
 
             /*
-             *  defer setting of earliestDatestamp in job until we know
-             *  the datestamp granularity of the repository 
+             * defer setting of earliestDatestamp in job until we know the
+             * datestamp granularity of the repository
              */
             final String earliestDatestamp =
-                response.readContent(OAI_NS, "earliestDatestamp", true);
+                    response.readContent(OAI_NS, "earliestDatestamp", true);
 
             s = response.readContent(OAI_NS, "deletedRecord", true);
@@ -242 +255 @@
                         namespaceURI, localName);
                 if (parser != null) {
-                    Description desc =
-                        parser.parseDescription(response.getXMLStreamReader());
+                    Description desc = parser.parseDescription(
+                            response.getXMLStreamReader());
                     if (desc != null) {
                         if (descriptions == null) {
@@ -251 +264 @@
                     }
                 } else {
-                    System.err.println("skipping DESC " + namespaceURI + ", " +
-                            localName);
+                    logger.info("skipping <description> for {}", new QName(
+                            namespaceURI, localName));
                 }
                 response.readEnd(OAI_NS, "description", true);
@@ -259 +272 @@
             response.readEnd(OAI_NS, "Identify");
             response.close();
-            
+
             job.onIdentify();
 
@@ -269 +282 @@
         }
     }
+
 
     private boolean doListMetadataFormatsTask(HarvestJobImpl job)
@@ -282 +296 @@
             response.readStart(OAI_NS, "ListMetadataFormats", true);
             List<MetadataFormat> metadataFormats =
-                new ArrayList<MetadataFormat>();
+                    new ArrayList<MetadataFormat>();
             while (response.readStart(OAI_NS, "metadataFormat",
                     metadataFormats.isEmpty())) {
                 final String prefix =
-                    response.readContent(OAI_NS, "metadataPrefix", true);
+                        response.readContent(OAI_NS, "metadataPrefix", true);
                 final String schema =
-                    response.readContent(OAI_NS, "schema", true);
+                        response.readContent(OAI_NS, "schema", true);
                 final String namespace =
-                    response.readContent(OAI_NS, "metadataNamespace", true);
+                        response.readContent(OAI_NS, "metadataNamespace", true);
                 metadataFormats.add(
                         new MetadataFormatImpl(prefix, schema, namespace));
@@ -299 +313 @@
 
             job.onListMetadataFormats(metadataFormats);
-            
+
             return true;
         } finally {
@@ -307 +321 @@
         }
     }
+
 
     private boolean doListRecordsTask(HarvestJobImpl job) throws IOException,
@@ -344 +359 @@
             boolean first = true;
             while (response.readStart(OAI_NS, "record", first)) {
+                boolean deleted = false;
+
                 response.readStart(OAI_NS, "header", true, true);
                 String s = response.readAttributeValue(null, "status");
@@ -349 +366 @@
                     if ("deleted".equals(s)) {
                         header.setDeleted(true);
+                        deleted = true;
                     } else {
-                        throw new XMLStreamException("attribute 'status' of " +
-                                "element 'record' must contain 'deleted'");
+                        throw new XMLStreamException("attribute 'status' of "
+                                + "element 'record' must contain 'deleted'",
+                                response.getLocation());
                     }
                 }
@@ -365 +384 @@
                 response.readEnd(OAI_NS, "header");
                 job.onRecord(header);
-                if (response.readStart(OAI_NS, "metadata", true)) {
-                    response.consumeWhitespace();
-                    readerWrapper.reset(response.getXMLStreamReader());
-                    job.onRecordMetadata(header, readerWrapper);
-                    response.consumeWhitespace();
-                    response.readEnd(OAI_NS, "metadata");
+                /*
+                 * only parse metadata from non-deleted records
+                 */
+                if (!deleted) {
+                    if (response.readStart(OAI_NS, "metadata", true)) {
+                        response.consumeWhitespace();
+                        readerWrapper.reset(response.getXMLStreamReader());
+                        job.onRecordMetadata(header, readerWrapper);
+                        response.consumeWhitespace();
+                        response.readEnd(OAI_NS, "metadata");
+                    }
                 }
                 while (response.readStart(OAI_NS, "about", false)) {
@@ -398 +422 @@
             }
             response.readEnd(OAI_NS, "ListRecords");
-            
+
             response.close();
 
             return (nextToken == null);
+        } catch (HarvesterProtocolErrorException e) {
+            /*
+             * XXX: handle NO_RECORDS_MATCH protocol error
+             */
+            List<ProtocolError> errors = e.getErrors();
+            if ((errors != null) && !errors.isEmpty()) {
+                for (ProtocolError error : errors) {
+                    if (error.isCode(ProtocolError.Code.NO_RECORDS_MATCH)) {
+                        logger.debug("no records match prefix '{}'",
+                                job.getCurrentPrefix());
+                        return true;
+                    }
+                }
+            }
+            throw e;
         } finally {
             if (response != null) {
@@ -409 +448 @@
     }
 
+
     private Response execute(HarvestJobImpl job, boolean compress)
             throws IOException, XMLStreamException, HarvesterException {
-        long now = System.currentTimeMillis();
+        final long now = System.currentTimeMillis();
         job.incRequestCount();
         final URI uri = uriBuilder.createURI(job.getRepositoryURI());
-        System.err.println("HTTP GET " + uri);
+        logger.debug("Performing HTTP GET request to {}", uri);
         final HttpGet request = new HttpGet(uri);
         if (compress) {
-            if ((job.getCompressionMask() &
-                    Repository.COMPRESSION_METHOD_GZIP) > 0) {
+            final int mask = job.getCompressionMask();
+            if ((mask & Repository.COMPRESSION_METHOD_GZIP) > 0) {
                 request.addHeader(HEADER_ACCEPT_ENCODING, "gzip");
             }
-            if ((job.getCompressionMask() &
-                    Repository.COMPRESSION_METHOD_DEFLATE) > 0) {
+            if ((mask & Repository.COMPRESSION_METHOD_DEFLATE) > 0) {
                 request.addHeader(HEADER_ACCEPT_ENCODING, "deflate");
             }
@@ -434 +473 @@
 
         final HttpResponse response = client.execute(request);
-        job.addToNetworkTime(System.currentTimeMillis() - now);
-        int status = response.getStatusLine().getStatusCode();
-        if (status == HttpStatus.SC_OK) {
-            delay = 0;
-            return new Response(job, response.getEntity(), harvester);
-        } else {
-            request.abort();
-            if (status == HttpStatus.SC_SERVICE_UNAVAILABLE) {
-                System.err.println("-> " + status);
-                delay = parseTryAfter(response);
+        /*
+         * FIXME: the following code need serious re-factoring
+         */
+        boolean close = false;
+        try {
+            job.addToNetworkTime(System.currentTimeMillis() - now);
+            final int status = response.getStatusLine().getStatusCode();
+            if (status == HttpStatus.SC_OK) {
+                delay = 0;
+                return new Response(job, response, harvester);
             } else {
-                throw new HarvesterException(
-                        "provider returned unexpected HTTP status: " + status);
-            }
-        }
-        return null;
-    }
+                if (status == HttpStatus.SC_SERVICE_UNAVAILABLE) {
+                    delay = parseTryAfter(response);
+                    logger.debug("got service unavailable status, retrying " +
+                            "after {} seconds", delay);
+                    close = true;
+                    return null;
+                } else {
+                    throw new HarvesterException(
+                            "provider returned unexpected HTTP status: " +
+                                    status);
+                }
+            }
+        } catch (IOException e) {
+            close = true;
+            throw e;
+        } catch (XMLStreamException e) {
+            close = true;
+            throw e;
+        } catch (HarvesterException e) {
+            close = true;
+            throw e;
+        } finally {
+            if (close) {
+                /*
+                 * try hard to release HTTP client resources ...
+                 */
+                try {
+                    EntityUtils.consume(response.getEntity());
+                } catch (IOException ex) {
+                    /* IGNORE */
+                }
+
+                /* make sure to release allocated resources */
+                HttpClientUtils.closeQuietly(response);
+
+                if (request != null) {
+                    request.abort();
+                }
+            }
+        }
+    }
+
 
     private long parseTryAfter(HttpResponse response) {
@@ -474 +549 @@
     }
 
-    private Date parseDate(Repository.Granularity granularity, String s)
+
+    private static Date parseDate(Repository.Granularity granularity, String s)
             throws HarvesterException {
-        String[] patterns = null;
+        final MutableDateTime date = new MutableDateTime(0);
+        for (int i = 0; i < DATEFORMATS.length; i++) {
+            DateTimeFormatter fmt = DateTimeFormat
+                    .forPattern(DATEFORMATS[i])
+                    .withZone(DateTimeZone.UTC);
+            if (fmt.parseInto(date, s, 0) == s.length()) {
+                if (DATEFORMATS[i].equals(DATEFORMAT_SECONDS) &&
+                        (granularity == Repository.Granularity.DAYS)) {
+                    logger.warn("repository announced DAYS granularity but " +
+                            "provided timestamp with SECONDS granularity");
+                }
+                if (DATEFORMATS[i].equals(DATEFORMAT_DAYS)) {
+                    date.setTime(0, 0, 0, 0);
+                }
+                return date.toDate();
+            }
+        }
+        throw new HarvesterException("invalid date: " + s);
+    }
+
+
+    private static String formatDate(Repository.Granularity granularity,
+            Date date) throws HarvesterException {
+        DateTimeFormatter fmt = null;
         switch (granularity) {
         case DAYS:
-            patterns = DATEFORMATS_DAYS;
+            fmt = DateTimeFormat.forPattern(DATEFORMAT_DAYS)
+                .withZone(DateTimeZone.UTC);
             break;
         case SECONDS:
-            patterns = DATEFORMATS_FULL;
-        }
-        MutableDateTime date = new MutableDateTime(0);
-        for (int i = 0; i < patterns.length; i++) {
-            DateTimeFormatter fmt = DateTimeFormat
-                .forPattern(patterns[i])
+            fmt = DateTimeFormat.forPattern(DATEFORMAT_SECONDS)
                 .withZone(DateTimeZone.UTC);
-            if (fmt.parseInto(date, s, 0) == s.length()) {
-                if (patterns[i].equals(DATEFORMATS_DAYS[0])) {
-                    date.setTime(0, 0, 0, 0);
-                }
-                return date.toDate();
-            }
-        }
-        throw new HarvesterException("invalid date: " + s);
-    }
-
-    private String formatDate(Repository.Granularity granularity, Date date) {
-        String pattern = null;
-        switch (granularity) {
-        case DAYS:
-            pattern = DATEFORMATS_DAYS[0];
             break;
-        case SECONDS:
-            pattern = DATEFORMATS_FULL[0];
-        }
-        DateTimeFormatter fmt =
-            DateTimeFormat.forPattern(pattern).withZone(DateTimeZone.UTC);
+        default:
+            /* cannot happen, but silence FindBugs warning */
+            throw new HarvesterException("invalid granularity: " + granularity);
+        }
         return fmt.print(new DateTime(date));
     }

OAIHarvester/trunk/OAIHarvester/src/main/java/eu/clarin/cmdi/oai/harvester/impl/ParallelHarvester.java

@@ -8 +8 @@
 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;
@@ -13 +16 @@
 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 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());
+                    logger.debug("[{}] waiting ...", getId());
+                    final HarvestJobImpl job = queue.take();
+                    logger.debug("[{}] got job {}", getId(), job.getId());
                     try {
                         while (true) {
-                            System.err.println(getId() + " process job " +
-                                    job.getId());
+                            logger.debug("[{}] processing job {} ...",
+                                    getId(), 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());
+                                    logger.debug("[{}] requeuing job {} ...",
+                                            getId(), job.getId());
                                 } else {
                                     continue;
@@ -49 +54 @@
                 }
             } // while
-            System.err.println(getId() + " ... exit");
+            logger.debug("[{}] exiting ...", getId());
         }
+
 
         void shutdown() {
@@ -57 +63 @@
         }
     }
+    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 final HashSet<WorkerThread> workers = new HashSet<WorkerThread>();
-    private final DelayQueue<HarvestJobImpl> queue = new DelayQueue<HarvestJobImpl>();
-    private AtomicLong nextId = new AtomicLong();
 
     private ParallelHarvester() {
@@ -68 +78 @@
     }
 
+
     @Override
     public HarvestJob createJob(URI repositoryURI, HarvestHandler handler)
@@ -74 +85 @@
                 repositoryURI, handler);
     }
+
 
     public static Harvester newInstance() {
@@ -83 +95 @@
     }
 
+
     @Override
     public void shutdown() {
@@ -90 +103 @@
                 w.join();
             } catch (InterruptedException e) {
-                System.err.println("join fail!");
+                logger.error("interrupted while shutting down");
             }
         }
@@ -102 +115 @@
     }
 
+
     @Override
     void doCancelJob(HarvestJobImpl job) throws HarvesterException {
         queue.remove(job);
     }
+
 
     @Override

OAIHarvester/trunk/OAIHarvester/src/main/java/eu/clarin/cmdi/oai/harvester/impl/Response.java

@@ -10 +10 @@
 import java.util.zip.InflaterInputStream;
 
+import javax.xml.namespace.NamespaceContext;
 import javax.xml.namespace.QName;
+import javax.xml.stream.Location;
 import javax.xml.stream.XMLStreamConstants;
 import javax.xml.stream.XMLStreamException;
 import javax.xml.stream.XMLStreamReader;
+import javax.xml.stream.XMLStreamWriter;
 
 import org.apache.http.HttpEntity;
+import org.apache.http.HttpResponse;
+import org.apache.http.client.utils.HttpClientUtils;
 import org.apache.http.util.EntityUtils;
+import org.codehaus.stax2.XMLStreamReader2;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import eu.clarin.cmdi.oai.harvester.HarvesterException;
@@ -22 +30 @@
 import eu.clarin.cmdi.oai.harvester.ProtocolError;
 
+
 final class Response {
     private static final class CountingInputStream extends FilterInputStream {
         private long count = 0;
 
+
         private CountingInputStream(final InputStream stream) {
             super(stream);
         }
+
 
         @Override
@@ -36 +47 @@
             return result;
         }
+
 
         @Override
@@ -45 +57 @@
         }
 
+
         @Override
         public int read(byte[] buffer) throws IOException {
@@ -52 +65 @@
         }
 
+
         @Override
         public long skip(long n) throws IOException {
@@ -58 +72 @@
             return result;
         }
-        
+
+
         public long getByteCount() {
             return count;
@@ -64 +79 @@
 
     } // class CountingInputStream
-
-    private static final String OAI_NS =
-        "http://www.openarchives.org/OAI/2.0/";
-    private final HttpEntity entity;
+    private static final String OAI_NS = "http://www.openarchives.org/OAI/2.0/";
+    private static final String ENCODING_GZIP = "gzip";
+    private static final String ENCODING_DEFLATE = "defalte";
+    private static final Logger logger =
+            LoggerFactory.getLogger(Response.class);
+    private final HttpResponse response;
     private final long now = System.currentTimeMillis();
-    private final CountingInputStream stream; 
-    private final XMLStreamReader reader;
-
-    Response(HarvestJobImpl job, HttpEntity entity, AbstractHarvester harvester)
+    private final CountingInputStream stream;
+    private final XMLStreamReader2 reader;
+
+
+    Response(HarvestJobImpl job, HttpResponse response, AbstractHarvester harvester)
             throws IOException, XMLStreamException, HarvesterException {
-        this.entity = entity;
-
+        if (job == null) {
+           throw new NullPointerException("job == null");
+        }
+        if (response == null) {
+            throw new NullPointerException("response == null");
+        }
+        if (harvester == null) {
+            throw new NullPointerException("harvester == null");
+        }
+        this.response = response;
+
+        final HttpEntity entity = response.getEntity();
         if (entity.getContentType() != null) {
-            System.err.println("Content-Type: " +
-                    entity.getContentType().getValue());
-            System.err.println("Content-Length: " + entity.getContentLength());
-        }
-        InputStream in = entity.getContent();
+            logger.debug("Content-Type: {}", entity.getContentType().getValue());
+            logger.debug("Content-Length: {}", entity.getContentLength());
+        }
+        InputStream in             = entity.getContent();
         org.apache.http.Header enc = entity.getContentEncoding();
         if (enc != null) {
-            System.err.println("Content-Encoding: " +
-                    entity.getContentEncoding().getValue());
-            if ("gzip".equalsIgnoreCase(enc.getValue())) {
+            final String encoding = enc.getValue();
+            logger.debug("Content-Encoding: {}", encoding);
+            if (ENCODING_GZIP.equalsIgnoreCase(encoding)) {
                 in = new GZIPInputStream(in);
-            } else if ("deflate".equalsIgnoreCase(enc.getValue())) {
+            } else if (ENCODING_DEFLATE.equalsIgnoreCase(encoding)) {
                 in = new InflaterInputStream(in, new Inflater(true));
+            } else {
+                throw new HarvesterException("Unsupported content encoding " +
+                        "in HTTP response: " + encoding);
             }
         }
@@ -97 +127 @@
         readContent(OAI_NS, "responseDate", true);
         readStart(OAI_NS, "request", true, true);
-        // System.err.println("YYY XXX = " + toReadable(reader) + ", " +
-        // reader.getAttributeValue(null, "verb"));
         readEnd(OAI_NS, "request", true);
 
@@ -116 +144 @@
         }
     }
-    
+
+
     public void close() throws XMLStreamException {
         readEnd(OAI_NS, "OAI-PMH");
     }
+
 
     public void release(HarvestJobImpl job) {
@@ -134 +164 @@
             /* IGNORE */
         }
+
+        /*
+         * try hard to release HTTP client resources ...
+         */
         try {
-            EntityUtils.consume(entity);
-        } catch (IOException e) {
+            EntityUtils.consume(response.getEntity());
+        } catch (IOException ex) {
             /* IGNORE */
         }
-    }
-
-    public boolean readStart(String namespaceURI, String localName,
-            boolean required) throws XMLStreamException {
+
+        /* make sure to release allocated resources */
+        HttpClientUtils.closeQuietly(response);
+    }
+
+
+    XMLStreamReader getXMLStreamReader() {
+        return reader;
+    }
+
+
+    Location getLocation() {
+        return reader.getLocation();
+    }
+
+
+    boolean readStart(String namespaceURI, String localName, boolean required)
+            throws XMLStreamException {
         return readStart(namespaceURI, localName, required, false);
     }
 
-    public boolean readStart(String namespaceURI, String localName,
-            boolean required, boolean attributes) throws XMLStreamException {
-//        System.err.println("readStart (" + localName + ", required = " +
-//                required + ") @ " + toReadable(reader));
+
+    boolean readStart(String namespaceURI, String localName, boolean required,
+            boolean attributes) throws XMLStreamException {
+        // System.err.println("readStart (" + localName + ", required = " +
+        // required + ") @ " + toReadable(reader));
         if (!reader.isEndElement()) {
             while (reader.hasNext()) {
 //                System.err.println("  LOOP: " + dumpState());
-                if (reader.isWhiteSpace()) {
-                    reader.next();
-                    continue;
-                }
+                consumeWhitespace();
                 if (reader.isStartElement()) {
                     if (namespaceURI.equals(reader.getNamespaceURI()) &&
                             localName.equals(reader.getLocalName())) {
-//                        System.err.print("--> found ");
+                        // System.err.print("--> found ");
                         if (!attributes) {
-//                            System.err.print("and consumed ");
+                            // System.err.print("and consumed ");
                             reader.next(); // skip to next event
                         }
-//                        System.err.println("@ " + toReadable(reader));
+                        // System.err.println("@ " + toReadable(reader));
                         return true;
                     }
@@ -177 +223 @@
         }
         if (required) {
-//            System.err.println("--> error, not found @ " + toReadable(reader));
-            throw new XMLStreamException("expected element '" +
-                    new QName(namespaceURI, localName) + "', but found '" +
-                    reader.getName() + "'", reader.getLocation());
-        }
-//        System.err.println("--> not found @ " + toReadable(reader));
+            logger.error("fail: {}", dumpState());
+
+            // System.err.println("--> error, not found @ " +
+            // toReadable(reader));
+            if (reader.hasName()) {
+                throw new XMLStreamException("expected element '" +
+                        new QName(namespaceURI, localName) + "', but found '" +
+                        reader.getName() + "'", reader.getLocation());
+            } else {
+                throw new XMLStreamException("unexpacted state while scanning for start tag for element '" + new QName(namespaceURI, localName) + "'", reader.getLocation());
+            }
+        }
+        // System.err.println("--> not found @ " + toReadable(reader));
         return false;
     }
 
-    public void readEnd(String namespaceURI, String localName)
+
+    void readEnd(String namespaceURI, String localName)
             throws XMLStreamException {
         readEnd(namespaceURI, localName, false);
     }
 
-    public void readEnd(String namespaceURI, String localName,
-            boolean skipContent) throws XMLStreamException {
-//        System.err.println("readEnd (" + localName + ") @ " + dumpState() +
-//                ", skipContent = " + skipContent);
+
+    void readEnd(String namespaceURI, String localName, boolean skipContent)
+            throws XMLStreamException {
+        // System.err.println("readEnd (" + localName + ") @ " + dumpState() +
+        // ", skipContent = " + skipContent);
         int level = 1;
         while (reader.hasNext()) {
-//            System.err.println("  LOOP " + dumpState() + " [" +
-//                    level + "]");
-            if (reader.isWhiteSpace()) {
-                reader.next();
-                continue;
-            }
+            // System.err.println("  LOOP " + dumpState() + " [" +
+            // level + "]");
+            consumeWhitespace();
             if (skipContent) {
                 if (reader.isCharacters()) {
@@ -219 +271 @@
             if (reader.isEndElement()) {
                 level--;
-//                System.err.println("   @END-TAG: " + dumpState() + " [" +
-//                        level + "]");
+                // System.err.println("   @END-TAG: " + dumpState() + " [" +
+                // level + "]");
                 if (level == 0) {
                     if (namespaceURI.equals(reader.getNamespaceURI()) &&
-                        localName.equals(reader.getLocalName())) {
+                            localName.equals(reader.getLocalName())) {
                         reader.next(); // consume tag
                         break;
@@ -236 +288 @@
             reader.next();
         }
-//        System.err.println("--> ok @ " + dumpState());
-    }
-
-    public String readContent(String namespaceURI, String localName,
-            boolean required) throws XMLStreamException {
+        // System.err.println("--> ok @ " + dumpState());
+    }
+
+
+    boolean peekStart(String namespaceURI, String localName)
+            throws XMLStreamException {
+        if (!reader.isEndElement()) {
+            while (reader.hasNext()) {
+                // System.err.println("  LOOP: " + dumpState());
+                if (reader.isWhiteSpace()) {
+                    reader.next();
+                    continue;
+                }
+                if (reader.isStartElement()) {
+                    if (namespaceURI.equals(reader.getNamespaceURI()) &&
+                            localName.equals(reader.getLocalName())) {
+                        return true;
+                    } else {
+                        return false;
+                    }
+                }
+                if (reader.isCharacters() || reader.isEndElement()) {
+                    break;
+                }
+                reader.next();
+            } // while
+        }
+        return false;
+    }
+
+
+    String readContent(String namespaceURI, String localName, boolean required)
+            throws XMLStreamException {
+        return readContent(namespaceURI, localName, required, true);
+    }
+
+
+    String readContent(String namespaceURI, String localName, boolean required,
+            boolean contentRequired) throws XMLStreamException {
         String result = null;
         if (readStart(namespaceURI, localName, required)) {
-            result = readString(true);
+            try {
+                result = readString(contentRequired);
+                if (!contentRequired && (result == null)) {
+                    result = "";
+                }
+            } catch (XMLStreamException e) {
+                StringBuilder sb = new StringBuilder();
+                sb.append("element '");
+                sb.append(new QName(namespaceURI, localName));
+                sb.append(localName).append("' may not be empty");
+                throw new XMLStreamException(sb.toString(), e.getLocation());
+            }
             readEnd(namespaceURI, localName);
         }
@@ -249 +346 @@
     }
 
-    public String readString(boolean required)
-            throws XMLStreamException {
-//        System.err.println("readString @ " + toReadable(reader));
+
+    int readContent(String namespaceURI, String localName, boolean required,
+            int defaultValue) throws XMLStreamException {
+        if (readStart(namespaceURI, localName, required)) {
+            String s = readString(true);
+            try {
+                readEnd(namespaceURI, localName);
+            } catch (XMLStreamException e) {
+                StringBuilder sb = new StringBuilder();
+                sb.append("element '");
+                sb.append(new QName(namespaceURI, localName));
+                sb.append(localName).append("' may not be empty");
+                throw new XMLStreamException(sb.toString(), e.getLocation());
+            }
+            try {
+                return Integer.parseInt(s);
+            } catch (NumberFormatException e) {
+                StringBuilder sb = new StringBuilder();
+                sb.append("element '");
+                sb.append(new QName(namespaceURI, localName));
+                sb.append("' was expected to be of type xs:integer; ");
+                sb.append("incompatible value was: ");
+                sb.append(s);
+                throw new XMLStreamException(sb.toString(),
+                        reader.getLocation(), e);
+            }
+        }
+        return defaultValue;
+    }
+
+
+    String readString(boolean required) throws XMLStreamException {
+        // System.err.println("readString @ " + toReadable(reader));
+        StringBuilder sb = new StringBuilder();
+        while (reader.isCharacters()) {
+            String s = reader.getText();
+            if (s != null) {
+                sb.append(s);
+            }
+            reader.next();
+        } // while
         String s = null;
-        if (reader.isCharacters()) {
-            s = reader.getText();
-            if (s != null) {
-                s = s.trim();
-            }
-            reader.next();
+        if (sb.length() > 0) {
+            s = sb.toString().trim();
         }
         if (required && ((s == null) || s.isEmpty())) {
             throw new XMLStreamException("expected character content "
-                    + "at position", reader.getLocation());
-        }
-//      System.err.println("--> ok @ " + toReadable(reader));
+                    + "at position ", reader.getLocation());
+        }
+        // System.err.println("--> ok @ " + toReadable(reader));
         return s;
     }
 
-    public String readAttributeValue(String namespaceURI, String localName)
+
+    String readAttributeValue(String namespaceURI, String localName)
             throws XMLStreamException {
         if (!reader.isStartElement()) {
@@ -281 +413 @@
     }
 
-    public String readNamespaceURI() throws XMLStreamException {
+
+    String readNamespaceURI() throws XMLStreamException {
         if (!reader.isStartElement()) {
             throw new XMLStreamException("not at a start elment event",
@@ -289 +422 @@
     }
 
-    public String peekElementLocalName() throws XMLStreamException {
+
+    String peekElementLocalName() throws XMLStreamException {
         if (!reader.isStartElement()) {
             throw new XMLStreamException("not at a start elment event",
@@ -297 +431 @@
     }
 
-    public void consumeStart() throws XMLStreamException {
+
+    void consumeStart() throws XMLStreamException {
         if (!reader.isStartElement()) {
             throw new XMLStreamException("not at a start elment event",
@@ -305 +440 @@
     }
 
-    public void consumeWhitespace() throws XMLStreamException {
-        while (reader.isWhiteSpace() && reader.hasNext()) {
+
+    void consumeWhitespace() throws XMLStreamException {
+        outer:
+        while (reader.hasNext()) {
+            /*
+             * some hackish method to also skip stray UTF-8 BOMs; only works
+             * where whitespace is expected
+             */
+            if (reader.isCharacters()) {
+                final char buffer[] = reader.getTextCharacters();
+                final int end = reader.getTextStart() + reader.getTextLength();
+                for (int i = reader.getTextStart(); i < end; i++) {
+                    if (!(Character.isWhitespace(buffer[i]) ||
+                            (buffer[i] == 0x0FEFF))) {
+                        break outer;
+                    }
+                }
+            } else if (!reader.isWhiteSpace()) {
+                break outer;
+            }
             reader.next();
-            continue;
-        }
-    }
-
-    public XMLStreamReader getXMLStreamReader() {
-        return reader;
-    }
-
-    public String dumpState() {
+        } // while
+    }
+
+
+    void copyTo(XMLStreamWriter writer) throws XMLStreamException {
+        final int depth = reader.getDepth();
+        do {
+            copyEvent(reader, writer);
+            reader.next();
+        } while (reader.getDepth() >= depth);
+    }
+
+
+    private static void copyEvent(XMLStreamReader from, XMLStreamWriter to)
+            throws XMLStreamException {
+        switch (from.getEventType()) {
+        case XMLStreamConstants.START_DOCUMENT:
+            {
+                String version = from.getVersion();
+                if (version == null || version.length() == 0) {
+                    to.writeStartDocument();
+                } else {
+                    to.writeStartDocument(from.getCharacterEncodingScheme(),
+                            from.getVersion());
+                }
+                to.writeCharacters("\n");
+            }
+            return;
+
+        case XMLStreamConstants.END_DOCUMENT:
+            to.writeCharacters("\n");
+            to.writeEndDocument();
+            return;
+
+        case XMLStreamConstants.START_ELEMENT:
+            copyStartElement(from, to);
+            return;
+
+        case XMLStreamConstants.END_ELEMENT:
+            to.writeEndElement();
+            return;
+
+        case XMLStreamConstants.SPACE:
+            to.writeCharacters(from.getTextCharacters(), from.getTextStart(),
+                    from.getTextLength());
+            return;
+
+        case XMLStreamConstants.CDATA:
+            to.writeCData(from.getText());
+            return;
+
+        case XMLStreamConstants.CHARACTERS:
+            to.writeCharacters(from.getTextCharacters(), from.getTextStart(),
+                    from.getTextLength());
+            return;
+
+        case XMLStreamConstants.COMMENT:
+            to.writeComment(from.getText());
+            return;
+
+        case XMLStreamConstants.PROCESSING_INSTRUCTION:
+            to.writeProcessingInstruction(from.getPITarget(), from.getPIData());
+            return;
+
+        case XMLStreamConstants.DTD:
+        case XMLStreamConstants.ENTITY_REFERENCE:
+        case XMLStreamConstants.ATTRIBUTE:
+        case XMLStreamConstants.NAMESPACE:
+        case XMLStreamConstants.ENTITY_DECLARATION:
+        case XMLStreamConstants.NOTATION_DECLARATION:
+            /* FALL_TROUGH */
+        }
+        throw new XMLStreamException("unsupported event type: " +
+                from.getEventType());
+    }
+
+
+    private static void copyStartElement(XMLStreamReader from,
+            XMLStreamWriter to) throws XMLStreamException {
+        final int nsCount = from.getNamespaceCount();
+        if (nsCount > 0) { // yup, got some...
+            for (int i = 0; i < nsCount; ++i) {
+                String pfx = from.getNamespacePrefix(i);
+                String uri = from.getNamespaceURI(i);
+                if ((pfx == null) || pfx.isEmpty()) { // default NS
+                    to.setDefaultNamespace(uri);
+                } else {
+                    to.setPrefix(pfx, uri);
+                }
+            }
+        }
+
+        final String prefix = from.getPrefix();
+        final NamespaceContext from_ctx = from.getNamespaceContext();
+        final NamespaceContext to_ctx = to.getNamespaceContext();
+        boolean repair_prefix_namespace = false;
+        if ((prefix != null) && (to_ctx.getNamespaceURI(prefix) == null)) {
+            repair_prefix_namespace = true;
+            to.setPrefix(prefix, from_ctx.getNamespaceURI(prefix));
+        }
+
+        to.writeStartElement(prefix, from.getLocalName(),
+                from.getNamespaceURI());
+
+        if (nsCount > 0) {
+            // write namespace declarations
+            for (int i = 0; i < nsCount; ++i) {
+                String pfx = from.getNamespacePrefix(i);
+                String uri = from.getNamespaceURI(i);
+
+                if ((pfx == null) || pfx.isEmpty()) { // default NS
+                    to.writeDefaultNamespace(uri);
+                } else {
+                    to.writeNamespace(pfx, uri);
+                }
+            }
+        }
+        if (repair_prefix_namespace) {
+            to.writeNamespace(prefix, from_ctx.getNamespaceURI(prefix));
+        }
+
+        int attrCount = from.getAttributeCount();
+        if (attrCount > 0) {
+            for (int i = 0; i < attrCount; ++i) {
+                to.writeAttribute(from.getAttributePrefix(i),
+                        from.getAttributeNamespace(i),
+                        from.getAttributeLocalName(i),
+                        from.getAttributeValue(i));
+            }
+        }
+    }
+
+
+    String dumpState() {
         StringBuilder sb = new StringBuilder();
         switch (reader.getEventType()) {
         case XMLStreamConstants.START_DOCUMENT:
-            return "START_DOC";
+            return "START_DOC[";
         case XMLStreamConstants.END_DOCUMENT:
-            return "END_DOC";
+            return "END_DOC[";
         case XMLStreamConstants.START_ELEMENT:
             sb.append("START[");
@@ -328 +606 @@
             sb.append(",");
             sb.append(reader.getLocalName());
-            sb.append("]");
             break;
         case XMLStreamConstants.END_ELEMENT:
@@ -335 +612 @@
             sb.append(",");
             sb.append(reader.getLocalName());
-            sb.append("]");
             break;
         case XMLStreamConstants.CHARACTERS:
             sb.append("CHARACTERS[\"");
-            sb.append(reader.getText().replace("\n", "\\n")
-                    .replace("\r", "\\r").replace("\t", "\\t"));
+            sb.append(reader.getText()
+                    .replace("\n", "\\n")
+                    .replace("\r", "\\r")
+                    .replace("\t", "\\t"));
             sb.append("\", isWhitespace = ");
             sb.append(reader.isWhiteSpace());
-            sb.append("]");
             break;
         case XMLStreamConstants.CDATA:
@@ -351 +628 @@
             sb.append("\", isWhitespace = ");
             sb.append(reader.isWhiteSpace());
-            sb.append("]");
             break;
         default:
             sb.append(Integer.toString(reader.getEventType()));
         }
+        Location location = reader.getLocation();
+        sb.append(", row = ").append(location.getLineNumber());
+        sb.append(", col = ").append(location.getColumnNumber());
+        sb.append(", offset = ").append(location.getCharacterOffset());
+        sb.append("]");
         return sb.toString();
     }
-    
+
 } // class XmlStreamReaderWrapper

OAIHarvester/trunk/OAIHarvester/src/main/java/eu/clarin/cmdi/oai/harvester/impl/SimpleHarvester.java

@@ -3 +3 @@
 import java.net.URI;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import eu.clarin.cmdi.oai.harvester.HarvestHandler;
@@ -9 +13 @@
 import eu.clarin.cmdi.oai.harvester.HarvesterException;
 
+
 public final class SimpleHarvester extends AbstractHarvester {
-    private HarvestWorker worker;
-    private long nextId = 0;
+    private static final Logger logger =
+            LoggerFactory.getLogger(SimpleHarvester.class);
+    private final AtomicLong nextId = new AtomicLong(0);
+    private final HarvestWorker worker;
+
 
     private SimpleHarvester() {
         this.worker = new HarvestWorker(this);
     }
+
 
     @Override
@@ -22 +31 @@
             throw new IllegalArgumentException("repositoryURI == null");
         }
-        long jobId;
-        synchronized (this) {
-            jobId = ++nextId;
-        } // synchronized
-        return new HarvestJobImpl(this, jobId, repositoryURI, handler);
+        return new HarvestJobImpl(this, nextId.incrementAndGet(),
+                repositoryURI, handler);
     }
+
 
     public static Harvester newInstance() {
@@ -33 +40 @@
     }
 
+
     @Override
     public void shutdown() {
     }
+
 
     @Override
@@ -52 +61 @@
                     ((delay = job.getDelay(TimeUnit.MILLISECONDS)) > 0)) {
                 try {
-                    System.err.println(job.getId() + ": (" +
-                            job.getNetworkRetryCount() + ") waiting " +
-                            delay + " ...");
-                    /* add a few milliseconds, so we do not loop  */
+                    logger.debug("{}: ({}) waiting {} ...",
+                            job.getId(), job.getNetworkRetryCount(), delay); 
+                    /* add a few milliseconds, so we do not loop */
                     Thread.sleep(delay + 25);
                 } catch (InterruptedException e) {
@@ -64 +72 @@
     }
 
+
     @Override
     void doCancelJob(HarvestJobImpl job) throws HarvesterException {
@@ -69 +78 @@
     }
 
+
     @Override
     protected HarvestJob doFindJob(long id) throws HarvesterException {
-        throw new HarvesterException("findJob() is not supported"); 
+        throw new HarvesterException("findJob() is not supported");
     }
 

OAIHarvester/trunk/OAIHarvester/src/main/java/eu/clarin/cmdi/oai/harvester/impl/XmlStreamReaderProxy.java

@@ -15 +15 @@
     private State state;
     private int depth;
-    
+
     XmlStreamReaderProxy() {
         super();
@@ -35 +35 @@
                 XMLStreamConstants.ATTRIBUTE)) {
             throw new IllegalStateException(
-                    "Current event is not START_ELEMENT or ATTRIBUTE");
+                    "Current event is not START_ELEMENT or ATTRIBUTE: " + reader.getLocation());
         }
         return reader.getAttributeCount();
@@ -45 +45 @@
                 XMLStreamConstants.ATTRIBUTE)) {
             throw new IllegalStateException(
-                    "Current event is not START_ELEMENT or ATTRIBUTE");
+                    "Current event is not START_ELEMENT or ATTRIBUTE: " + reader.getLocation());
         }
         return reader.getAttributeLocalName(index);
@@ -55 +55 @@
                 XMLStreamConstants.ATTRIBUTE)) {
             throw new IllegalStateException(
-                    "Current event is not START_ELEMENT or ATTRIBUTE");
+                    "Current event is not START_ELEMENT or ATTRIBUTE: " + reader.getLocation());
         }
         return reader.getAttributeName(index);
@@ -65 +65 @@
                 XMLStreamConstants.ATTRIBUTE)) {
             throw new IllegalStateException(
-                    "Current event is not START_ELEMENT or ATTRIBUTE");
+                    "Current event is not START_ELEMENT or ATTRIBUTE: " + reader.getLocation());
         }
         return reader.getAttributeNamespace(index);
@@ -75 +75 @@
                 XMLStreamConstants.ATTRIBUTE)) {
             throw new IllegalStateException(
-                    "Current event is not START_ELEMENT or ATTRIBUTE");
+                    "Current event is not START_ELEMENT or ATTRIBUTE: " + reader.getLocation());
         }
         return reader.getAttributePrefix(index);
@@ -85 +85 @@
                 XMLStreamConstants.ATTRIBUTE)) {
             throw new IllegalStateException(
-                    "Current event is not START_ELEMENT or ATTRIBUTE");
+                    "Current event is not START_ELEMENT or ATTRIBUTE: " + reader.getLocation());
         }
         return reader.getAttributeType(index);
@@ -95 +95 @@
                 XMLStreamConstants.ATTRIBUTE)) {
             throw new IllegalStateException(
-                    "Current event is not START_ELEMENT or ATTRIBUTE");
+                    "Current event is not START_ELEMENT or ATTRIBUTE: " + reader.getLocation());
         }
         return reader.getAttributeValue(index);
@@ -105 +105 @@
                 XMLStreamConstants.ATTRIBUTE)) {
             throw new IllegalStateException(
-                    "Current event is not START_ELEMENT or ATTRIBUTE");
+                    "Current event is not START_ELEMENT or ATTRIBUTE: " + reader.getLocation());
         }
         return reader.getAttributeValue(namespaceURI, localName);
@@ -119 +119 @@
         if (!checkEventType(XMLStreamConstants.START_ELEMENT)) {
             throw new IllegalStateException(
-                    "Current event is not START_ELEMENT");
+                    "Current event is not START_ELEMENT: " + reader.getLocation());
         }
         return reader.getElementText();
@@ -166 +166 @@
                 XMLStreamConstants.END_ELEMENT)) {
             throw new IllegalStateException("Current event is not " +
-                    "START_ELEMENT or END_ELEMENT");
+                    "START_ELEMENT or END_ELEMENT: " + reader.getLocation());
         }
         return reader.getName();
@@ -176 +176 @@
                 XMLStreamConstants.END_ELEMENT, XMLStreamConstants.NAMESPACE)) {
             throw new IllegalStateException("Current event is not " +
-                    "START_ELEMENT, END_ELEMENT or NAMESPACE");
+                    "START_ELEMENT, END_ELEMENT or NAMESPACE: " + reader.getLocation());
         }
         return reader.getNamespaceContext();
@@ -186 +186 @@
                 XMLStreamConstants.END_ELEMENT, XMLStreamConstants.NAMESPACE)) {
             throw new IllegalStateException("Current event is not " +
-                    "START_ELEMENT, END_ELEMENT or NAMESPACE");
+                    "START_ELEMENT, END_ELEMENT or NAMESPACE: " + reader.getLocation());
         }
         return reader.getNamespaceCount();
@@ -196 +196 @@
                 XMLStreamConstants.END_ELEMENT, XMLStreamConstants.NAMESPACE)) {
             throw new IllegalStateException("Current event is not " +
-                    "START_ELEMENT, END_ELEMENT or NAMESPACE");
+                    "START_ELEMENT, END_ELEMENT or NAMESPACE: " + reader.getLocation());
         }
         return reader.getNamespacePrefix(index);
@@ -206 +206 @@
                 XMLStreamConstants.END_ELEMENT, XMLStreamConstants.NAMESPACE)) {
             throw new IllegalStateException("Current event is not " +
-                    "START_ELEMENT, END_ELEMENT or NAMESPACE");
+                    "START_ELEMENT, END_ELEMENT or NAMESPACE: " + reader.getLocation());
         }
         return reader.getNamespaceURI();
@@ -221 +221 @@
                 XMLStreamConstants.END_ELEMENT, XMLStreamConstants.NAMESPACE)) {
             throw new IllegalStateException("Current event is not " +
-                    "START_ELEMENT, END_ELEMENT or NAMESPACE");
+                    "START_ELEMENT, END_ELEMENT or NAMESPACE: " + reader.getLocation());
         }
         return reader.getNamespaceURI(index);
@@ -230 +230 @@
         if (!checkEventType(XMLStreamConstants.PROCESSING_INSTRUCTION)) {
             throw new IllegalStateException(
-                    "Current event is not PROCESSING_INSTRUCTION");
+                    "Current event is not PROCESSING_INSTRUCTION: " + reader.getLocation());
         }
         return reader.getPIData();
@@ -239 +239 @@
         if (!checkEventType(XMLStreamConstants.PROCESSING_INSTRUCTION)) {
             throw new IllegalStateException(
-                    "Current event is not PROCESSING_INSTRUCTION");
+                    "Current event is not PROCESSING_INSTRUCTION: " + reader.getLocation());
         }
         return reader.getPITarget();
@@ -445 +445 @@
                 break;
             default:
-                ;
+                break;
             } // switch (t)
             if (depth < 0) {

OAIHarvester/trunk/OAIHarvester/src/test/java/eu/clarin/cmdi/oai/harvester/HarvesterTest.java

@@ -15 +15 @@
 import javax.xml.stream.XMLOutputFactory;
 import javax.xml.stream.XMLStreamConstants;
+import javax.xml.stream.XMLStreamException;
 import javax.xml.stream.XMLStreamReader;
 import javax.xml.stream.XMLStreamWriter;
@@ -105 +106 @@
             }
         }
-        
+
         @Override
         public void onListMetadataFormats(List<MetadataFormat> metadataFormats) {
@@ -182 +183 @@
                     output.flush();
                 }
-            } catch (Exception e) {
+            } catch (IOException e) {
+                e.printStackTrace();
+                throw new RuntimeException("record failed");
+            } catch (XMLStreamException e) {
                 e.printStackTrace();
                 throw new RuntimeException("record failed");
@@ -275 +279 @@
 //            cal.set(2010, Calendar.DECEMBER, 15);
 //            job.setFrom(cal.getTime());
-//            job.setMetadataPrefixes(Arrays.asList("oai_dc", "cmdi"));
+//            job.setMetadataPrefixes(Arrays.asList("cmdi"));
             job.run();
 
+            /*
+             * Bad example of (busy) wait loop; should be done better, e.g.
+             * with CountDownLatch ...
+             */
             while (job.isRunning()) {
                 try {
@@ -372 +380 @@
     }
 
+
+    static {
+        /*
+         * quick and dirty setup for Log4J; don't copy blindly!
+         */
+        org.apache.log4j.BasicConfigurator
+                .configure(new org.apache.log4j.ConsoleAppender(
+                        new org.apache.log4j.PatternLayout("%-5p [%t] %m%n"),
+                        org.apache.log4j.ConsoleAppender.SYSTEM_ERR));
+        org.apache.log4j.Logger logger =
+                org.apache.log4j.Logger.getRootLogger();
+        logger.setLevel(org.apache.log4j.Level.INFO);
+        logger.getLoggerRepository().getLogger("eu.clarin")
+                .setLevel(org.apache.log4j.Level.DEBUG);
+    }
+
 } // class HarvesterTest