/**
 * Copyright 2005 The Apache Software Foundation
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.nutch.crawl;

import java.util.Iterator;
import java.io.IOException;

import org.apache.hadoop.io.*;
import org.apache.hadoop.mapred.*;

/** Merge new page entries with existing entries. */
public class CrawlDbReducer implements Reducer {
  private int retryMax;

  public void configure(JobConf job) {
    retryMax = job.getInt("db.fetch.retry.max", 3);
  }

  public void close() {}

  public void reduce(WritableComparable key, Iterator values,
                     OutputCollector output, Reporter reporter)
    throws IOException {

    CrawlDatum highest = null;
    CrawlDatum old = null;
    byte[] signature = null;
    float scoreIncrement = 0.0f;

    while (values.hasNext()) {
      CrawlDatum datum = (CrawlDatum)values.next();

      if (highest == null || datum.getStatus() > highest.getStatus()) {
        highest = datum;                          // find highest status
      }

      switch (datum.getStatus()) {                // find old entry, if any
      case CrawlDatum.STATUS_DB_UNFETCHED:
      case CrawlDatum.STATUS_DB_FETCHED:
        old = datum;
        break;
      case CrawlDatum.STATUS_LINKED:
        scoreIncrement += datum.getScore();
        break;
      case CrawlDatum.STATUS_SIGNATURE:
        signature = datum.getSignature();
      }
    }

    CrawlDatum result = null;

    switch (highest.getStatus()) {                // determine new status

    case CrawlDatum.STATUS_DB_UNFETCHED:          // no new entry
    case CrawlDatum.STATUS_DB_FETCHED:
    case CrawlDatum.STATUS_DB_GONE:
      result = old;                               // use old
      break;

    case CrawlDatum.STATUS_LINKED:                // highest was link
      if (old != null) {                          // if old exists
        result = old;                             // use it
      } else {
        result = highest;                         // use new entry
        result.setStatus(CrawlDatum.STATUS_DB_UNFETCHED);
        result.setScore(1.0f);                    // initial score is 1.0f
      }
      result.setSignature(null);                  // reset the signature
      break;
      
    case CrawlDatum.STATUS_FETCH_SUCCESS:         // succesful fetch
      result = highest;                           // use new entry
      if (highest.getSignature() == null) highest.setSignature(signature);
      result.setStatus(CrawlDatum.STATUS_DB_FETCHED);
      result.setNextFetchTime();
      break;

    case CrawlDatum.STATUS_FETCH_RETRY:           // temporary failure
      result = highest;                           // use new entry
      if (old != null)
        result.setSignature(old.getSignature());  // use old signature
      if (highest.getRetriesSinceFetch() < retryMax) {
        result.setStatus(CrawlDatum.STATUS_DB_UNFETCHED);
      } else {
        result.setStatus(CrawlDatum.STATUS_DB_GONE);
      }
      break;

    case CrawlDatum.STATUS_FETCH_GONE:            // permanent failure
      result = highest;                           // use new entry
      if (old != null)
        result.setSignature(old.getSignature());  // use old signature
      result.setStatus(CrawlDatum.STATUS_DB_GONE);
      break;

    default:
      throw new RuntimeException("Unknown status: "+highest.getStatus());
    }
    
    if (result != null) {
      result.setScore(result.getScore() + scoreIncrement);
      output.collect(key, result);
    }
  }

}

@@ -890,16 +890,64 @@
 etryMax;
+%0A  private CrawlDatum result = new CrawlDatum();
 %0A%0A  publ
@@ -1970,33 +1970,579 @@
     
-CrawlDatum result = null;
+// initialize with the latest version%0A    result.set(highest);%0A    if (old != null) %7B%0A      // copy metadata from old, if exists%0A      if (old.getMetaData() != null) %7B%0A        result.getMetaData().putAll(old.getMetaData());%0A        // overlay with new, if any%0A        if (highest.getMetaData() != null)%0A          result.getMetaData().putAll(highest.getMetaData());%0A      %7D%0A      // set the most recent valid value of modifiedTime%0A      if (old.getModifiedTime() %3E 0 && highest.getModifiedTime() == 0) %7B%0A        result.setModifiedTime(old.getModifiedTime());%0A      %7D%0A    %7D
 %0A%0A  
@@ -2763,34 +2763,34 @@
       result
- = old;   
+.set(old);
             
@@ -2985,18 +2985,18 @@
 sult
- = old;   
+.set(old);
     
@@ -3060,274 +3060,59 @@
 sult
- = highest;                         // use new entry%0A        result.setStatus(CrawlDatum.STATUS_DB_UNFETCHED);%0A        result.setScore(1.0f);                    // initial score is 1.0f%0A      %7D%0A      result.setSignature(null);                  // reset the signature
+.setStatus(CrawlDatum.STATUS_DB_UNFETCHED);%0A      %7D
 %0A   
@@ -3201,75 +3201,8 @@
 tch%0A
-      result = highest;                           // use new entry%0A
     
@@ -3239,22 +3239,21 @@
 = null) 
-highes
+resul
 t.setSig
@@ -3447,75 +3447,8 @@
 ure%0A
-      result = highest;                           // use new entry%0A
     
@@ -3815,75 +3815,8 @@
 ure%0A
-      result = highest;                           // use new entry%0A
     
@@ -4072,36 +4072,8 @@
    %0A
-    if (result != null) %7B%0A  
     
@@ -4129,18 +4129,16 @@
 t);%0A    
-  
 output.c
@@ -4158,21 +4158,15 @@
 esult);%0A
-    %7D%0A
   %7D%0A%0A%7D%0A

Make sure we use new values for score, metadata, fetch interval and fetch time.