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

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

File:

• OAIHarvester/trunk/OAIHarvester/src/main/java/eu/clarin/cmdi/oai/harvester/impl/ParallelHarvester.java (modified) (9 diffs)

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