archive-access-cvs

[Brad <bra...@us...>]

Update of /cvsroot/archive-access/archive-access/projects/wayback/src/java/org/archive/wayback/arcindexer
In directory sc8-pr-cvs1.sourceforge.net:/tmp/cvs-serv14129/src/java/org/archive/wayback/arcindexer

Modified Files:
	IndexPipeline.java ArcIndexer.java 
Added Files:
	PipelineFilter.java PipelineStatus.java 
	BDBResourceIndexWriter.java 
Log Message:
Heavy modification of configuration to be Context-level, instead of Servlet-level, which dramatically reduces configuration redundancy.

Cleaned up IndexPipeline, moved a few classes around, added a really simple JSP to view the Index and Merge queue sizes, and a filter, which both allows the index thread to start with the context, and allows access to the jsp.

--- NEW FILE: BDBResourceIndexWriter.java ---
/* BDBResourceIndexWriter
 *
 * Created on 2005/10/18 14:00:00
 *
 * Copyright (C) 2005 Internet Archive.
 *
 * This file is part of the Wayback Machine (crawler.archive.org).
 *
 * Wayback Machine is free software; you can redistribute it and/or modify
 * it under the terms of the GNU Lesser Public License as published by
 * the Free Software Foundation; either version 2.1 of the License, or
 * any later version.
 *
 * Wayback Machine is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Lesser Public License for more details.
 *
 * You should have received a copy of the GNU Lesser Public License
 * along with Wayback Machine; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 */

package org.archive.wayback.arcindexer;

import java.io.File;
import java.io.RandomAccessFile;

import org.archive.wayback.core.ResourceResult;
import org.archive.wayback.core.ResourceResults;
import org.archive.wayback.localbdbresourceindex.BDBResourceIndex;

import com.sleepycat.je.DatabaseException;

/**
 * Implements updates to a BDBResourceIndex
 * 
 * @author Brad Tofel
 * @version $Date: 2005/10/21 03:24:40 $, $Revision: 1.1 $
 */
public class BDBResourceIndexWriter {
	private BDBResourceIndex db = null;

	/**
	 * Constructor
	 */
	public BDBResourceIndexWriter() {
		super();
	}

	protected void init(final String thePath, final String theDbName)
			throws Exception {
		db = new BDBResourceIndex(thePath, theDbName);
	}

	protected void init(BDBResourceIndex db) {
		this.db = db;
	}

	protected void shutdown() throws DatabaseException {
		db.shutdownDB();
	}

	/**
	 * reads all ResourceResult objects from CDX at filePath, and merges them
	 * into the BDBResourceIndex.
	 * 
	 * @param indexFile
	 *            to CDX file
	 * @throws Exception
	 */
	public void importFile(File indexFile) throws Exception {
		ResourceResults results = readFile(indexFile);
		db.addResults(results);
	}

	private ResourceResults readFile(File indexFile) throws Exception {
		RandomAccessFile raFile = new RandomAccessFile(indexFile, "r");
		ResourceResults results = new ResourceResults();
		int lineNumber = 0;
		while (true) {
			String line = raFile.readLine();
			if (line == null) {
				break;
			}
			lineNumber++;
			if ((lineNumber == 1) && (line.contains(" CDX "))) {
				continue;
			}
			ResourceResult result = new ResourceResult();
			result.parseLine(line, lineNumber);

			results.addResourceResult(result);
		}
		return results;
	}

	/**
	 * @param args
	 */
	public static void main(String[] args) {
		try {
			BDBResourceIndexWriter idx = new BDBResourceIndexWriter();
			idx.init(args[0], args[1]);
			idx.importFile(new File(args[2]));

			idx.shutdown();
		} catch (Exception e) {
			e.printStackTrace();
		}

	}

}

Index: ArcIndexer.java
===================================================================
RCS file: /cvsroot/archive-access/archive-access/projects/wayback/src/java/org/archive/wayback/arcindexer/ArcIndexer.java,v
retrieving revision 1.2
retrieving revision 1.3
diff -C2 -d -r1.2 -r1.3
*** ArcIndexer.java	19 Oct 2005 01:22:36 -0000	1.2
--- ArcIndexer.java	21 Oct 2005 03:24:40 -0000	1.3
***************
*** 60,70 ****
  	 * Create a ResourceResults representing the records in ARC file at arcPath.
  	 * 
! 	 * @param arcPath
  	 * @return ResourceResults in arcPath.
  	 * @throws IOException
  	 */
! 	public ResourceResults indexArc(final String arcPath) throws IOException {
  		ResourceResults results = new ResourceResults();
- 		File arc = new File(arcPath);
  		ARCReader arcReader = ARCReaderFactory.get(arc);
  		arcReader.setParseHttpHeaders(true);
--- 60,69 ----
  	 * Create a ResourceResults representing the records in ARC file at arcPath.
  	 * 
! 	 * @param arc
  	 * @return ResourceResults in arcPath.
  	 * @throws IOException
  	 */
! 	public ResourceResults indexArc(File arc) throws IOException {
  		ResourceResults results = new ResourceResults();
  		ARCReader arcReader = ARCReaderFactory.get(arc);
  		arcReader.setParseHttpHeaders(true);
***************
*** 133,145 ****
  	 * 
  	 * @param results
! 	 * @param cdxPath
  	 * @throws IOException
  	 */
  	public void serializeResults(final ResourceResults results,
! 			final String cdxPath) throws IOException {
! 		Iterator itr = results.iterator();
! 		File cdx = new File(cdxPath);
! 		FileOutputStream output = new FileOutputStream(cdx);
  		output.write((ResourceResult.getCDXHeaderString() + "\n").getBytes());
  		while (itr.hasNext()) {
  			ResourceResult result = (ResourceResult) itr.next();
--- 132,146 ----
  	 * 
  	 * @param results
! 	 * @param target
  	 * @throws IOException
  	 */
  	public void serializeResults(final ResourceResults results,
! 			File target) throws IOException {
! 		
! 		// TODO will this automatically close when it falls out of scope?
! 		FileOutputStream output = new FileOutputStream(target);
  		output.write((ResourceResult.getCDXHeaderString() + "\n").getBytes());
+ 
+ 		Iterator itr = results.iterator();
  		while (itr.hasNext()) {
  			ResourceResult result = (ResourceResult) itr.next();
***************
*** 153,158 ****
  	public static void main(String[] args) {
  		ArcIndexer indexer = new ArcIndexer();
! 		String arc = args[0];
! 		String cdx = args[1];
  		try {
  			ResourceResults results = indexer.indexArc(arc);
--- 154,159 ----
  	public static void main(String[] args) {
  		ArcIndexer indexer = new ArcIndexer();
! 		File arc = new File(args[0]);
! 		File cdx = new File(args[1]);
  		try {
  			ResourceResults results = indexer.indexArc(arc);

--- NEW FILE: PipelineStatus.java ---
/* PipelineStatus
 *
 * Created on Oct 20, 2005
 *
 * Copyright (C) 2005 Internet Archive.
 *
 * This file is part of the wayback (crawler.archive.org).
 *
 * wayback is free software; you can redistribute it and/or modify
 * it under the terms of the GNU Lesser Public License as published by
 * the Free Software Foundation; either version 2.1 of the License, or
 * any later version.
 *
 * wayback is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Lesser Public License for more details.
 *
 * You should have received a copy of the GNU Lesser Public License
 * along with wayback; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 */

package org.archive.wayback.arcindexer;

/**
 * Data bag for handing off status of Pipeline to PipelineStatus.jsp.
 * 
 * @author brad
 * @version $Date: 2005/10/21 03:24:40 $, $Revision: 1.1 $
 */
public class PipelineStatus {

	private String numQueuedForIndex;

	private String numQueuedForMerge;

	/**
	 * Constructor
	 */
	public PipelineStatus() {
		super();
		// TODO Auto-generated constructor stub
	}

	/**
	 * @return Returns the numQueuedForIndex.
	 */
	public String getNumQueuedForIndex() {
		return numQueuedForIndex;
	}

	/**
	 * @param numQueuedForIndex
	 *            The numQueuedForIndex to set.
	 */
	public void setNumQueuedForIndex(String numQueuedForIndex) {
		this.numQueuedForIndex = numQueuedForIndex;
	}

	/**
	 * @return Returns the numQueuedForMerge.
	 */
	public String getNumQueuedForMerge() {
		return numQueuedForMerge;
	}

	/**
	 * @param numQueuedForMerge
	 *            The numQueuedForMerge to set.
	 */
	public void setNumQueuedForMerge(String numQueuedForMerge) {
		this.numQueuedForMerge = numQueuedForMerge;
	}

	/**
	 * @param args
	 */
	public static void main(String[] args) {
		// TODO Auto-generated method stub

	}

}

Index: IndexPipeline.java
===================================================================
RCS file: /cvsroot/archive-access/archive-access/projects/wayback/src/java/org/archive/wayback/arcindexer/IndexPipeline.java,v
retrieving revision 1.2
retrieving revision 1.3
diff -C2 -d -r1.2 -r1.3
*** IndexPipeline.java	19 Oct 2005 01:22:36 -0000	1.2
--- IndexPipeline.java	21 Oct 2005 03:24:40 -0000	1.3
***************
*** 29,40 ****
  import java.util.ArrayList;
  import java.util.Iterator;
  
  import org.archive.wayback.core.ResourceResults;
  
  import com.sun.org.apache.xml.internal.utils.StringToStringTable;
  
  /**
!  * Implements updating of a BDBResourceIndex using several directories with data
!  * files or flag files.
   * 
   * @author Brad Tofel
--- 29,49 ----
  import java.util.ArrayList;
  import java.util.Iterator;
+ import java.util.Properties;
  
  import org.archive.wayback.core.ResourceResults;
+ import org.archive.wayback.localbdbresourceindex.BDBResourceIndex;
  
  import com.sun.org.apache.xml.internal.utils.StringToStringTable;
  
  /**
!  * Implements indexing of new ARC files, and merging with a BDBResourceIndex.
!  * Assumes LocalBDBResourceIndex and LocalARCResourceStore for now.
!  * Maintains state using directories and files for now.
!  * 
!  * There are 3 primary components, each could be a thread, but the steps are
!  * run in serial for the moment:
!  * 1) watch for new ARC files, and queue them for indexing
!  * 2) index queued ARC files into CDX format, queue the CDX files for merging.
!  * 3) merge queued CDX files with the ResourceIndex.
   * 
   * @author Brad Tofel
***************
*** 42,48 ****
   */
  public class IndexPipeline {
! 	private File arcDir = null;
  
! 	private File mergeDir = null;
  
  	private File queuedDir = null;
--- 51,76 ----
   */
  public class IndexPipeline {
! 	private final static String RUN_PIPELINE = "indexpipeline.runpipeline";
  
! 	private final static String INDEX_PATH = "resourceindex.indexpath";
! 
! 	private final static String DB_NAME = "resourceindex.dbname";
! 
! 	private final static String ARC_PATH = "arcpath";
! 
! 	private final static String WORK_PATH = "indexpipeline.workpath";
! 
! 	private final static String QUEUED_DIR = "queued";
! 
! 	private final static String TO_BE_INDEXED_DIR = "toBeIndexed";
! 	
! 	private final static String INDEXING_DIR = "indexing";
! 	
! 	private final static String TO_BE_MERGED_DIR = "toBeMerged";
! 	
! 	
! 	private File arcDir = null;
! 	
! 	private File workDir = null;
  
  	private File queuedDir = null;
***************
*** 52,56 ****
  	private File indexingDir = null;
  
! 	private ArcIndexer indexer = null;
  
  	/**
--- 80,89 ----
  	private File indexingDir = null;
  
! 	private File toBeMergedDir = null;
! 
! 	private BDBResourceIndex db = null;
! 
! 	private static Thread indexUpdateThread = null;
! 	
  
  	/**
***************
*** 68,95 ****
  
  	/**
! 	 * Initialize this object from several path arguments.
  	 * 
! 	 * @param arcDir
! 	 * @param mergeDir
! 	 * @param workDir
  	 * @throws IOException
  	 */
! 	public void init(final String arcDir, final String mergeDir,
! 			final String workDir) throws IOException {
! 		this.arcDir = new File(arcDir);
! 		this.mergeDir = new File(mergeDir);
! 		this.queuedDir = new File(workDir + "/queued");
! 		this.toBeIndexedDir = new File(workDir + "/to-be-indexed");
! 		this.indexingDir = new File(workDir + "/indexing");
! 		ensureDir(new File(workDir));
! 		ensureDir(this.queuedDir);
! 		ensureDir(this.toBeIndexedDir);
! 		ensureDir(this.indexingDir);
! 		indexer = new ArcIndexer();
  	}
  
! 	private StringToStringTable dirToSTST(File dir) {
  		StringToStringTable hash = new StringToStringTable();
! 		String entries[] = dir.list();
  		for (int i = 0; i < entries.length; i++) {
  			hash.put(entries[i], "i");
--- 101,178 ----
  
  	/**
! 	 * Initialize this object, creating directories if needed, and starting
! 	 * thread if configured.
  	 * 
! 	 * @param p configuration 
  	 * @throws IOException
  	 */
! 
! 	public void init(Properties p) throws IOException {
! 		
! 		// where do we find ARC files?
! 		String arcPath = (String) p.get(ARC_PATH);
! 		if (arcPath == null || (arcPath.length() <= 0)) {
! 			throw new IllegalArgumentException("Failed to find " + ARC_PATH);
! 		}
! 
! 		// where is the BDB? (and what is it named?)
! 		String dbPath = (String) p.get(INDEX_PATH);
! 		if (dbPath == null || (dbPath.length() <= 0)) {
! 			throw new IllegalArgumentException("Failed to find " + INDEX_PATH);
! 		}
! 
! 		String dbName = (String) p.get(DB_NAME);
! 		if (dbName == null || (dbName.length() <= 0)) {
! 			throw new IllegalArgumentException("Failed to find " + DB_NAME);
! 		}
! 		
! 		// where do we keep working files?
! 		String workPath = (String) p.get(WORK_PATH);
! 		if (workPath == null || (workPath.length() <= 0)) {
! 			throw new IllegalArgumentException("Failed to find " + WORK_PATH);
! 		}
! 
! 		String runPipeline = (String) p.get(RUN_PIPELINE);
! 		try {
! 			db = new BDBResourceIndex(dbPath, dbName);
! 		} catch (Exception e) {
! 			// TODO is this the right choice? was already obfuscated from BDBException...
! 			throw new IOException(e.getMessage());
! 		}
! 		arcDir = new File(arcPath);
! 		workDir = new File(workPath);
! 		queuedDir = new File(workDir,QUEUED_DIR);
! 		toBeIndexedDir = new File(workDir,TO_BE_INDEXED_DIR);
! 		indexingDir = new File(workDir,INDEXING_DIR);
! 		toBeMergedDir = new File(workDir,TO_BE_MERGED_DIR);
! 
! 		ensureDir(workDir);
! 		ensureDir(queuedDir);
! 		ensureDir(toBeIndexedDir);
! 		ensureDir(indexingDir);
! 		ensureDir(toBeMergedDir);
! 
! 		if ((runPipeline != null) && (runPipeline.equals("1"))) {
! 
! 			System.out
! 					.println("LocalDBDResourceIndex starting pipeline thread...");
! 			if (indexUpdateThread == null) {
! 				startIndexPipelineThread(db);
! 			}
! 		}
! 	}
! 	
! 	private synchronized void startIndexPipelineThread(
! 			final BDBResourceIndex bdb) {
! 		if (indexUpdateThread != null) {
! 			return;
! 		}
! 		indexUpdateThread = new IndexPipelineThread(bdb, this);
! 		indexUpdateThread.start();
  	}
  
! 	private StringToStringTable getQueuedFiles() {
  		StringToStringTable hash = new StringToStringTable();
! 		String entries[] = queuedDir.list();
  		for (int i = 0; i < entries.length; i++) {
  			hash.put(entries[i], "i");
***************
*** 98,124 ****
  	}
  
! 	private StringToStringTable getQueuedFiles() {
! 		return dirToSTST(this.queuedDir);
  	}
  
! 	private ArrayList getNewArcs() {
  		StringToStringTable queued = getQueuedFiles();
  		ArrayList newArcs = new ArrayList();
  
! 		String arcs[] = this.arcDir.list();
! 		for (int i = 0; i < arcs.length; i++) {
! 			if (!queued.contains(arcs[i])) {
! 				newArcs.add(arcs[i]);
  			}
  		}
! 		return newArcs;
  	}
  
! 	private void queueArc(final String newArc) throws IOException {
! 		File newQueuedFile = new File(this.queuedDir.getAbsolutePath() + "/"
! 				+ newArc);
! 		File newToBeIndexedFile = new File(this.toBeIndexedDir
! 				.getAbsolutePath()
! 				+ "/" + newArc);
  		newToBeIndexedFile.createNewFile();
  		newQueuedFile.createNewFile();
--- 181,220 ----
  	}
  
! 	private Iterator getDirFilesIterator(File dir) {
! 		String files[] = dir.list();
! 		ArrayList list = new ArrayList();
! 		if (files != null) {
! 			for (int i = 0; i < files.length; i++) {
! 				File file = new File(dir, files[i]);
! 				if (file.isFile()) {
! 					list.add(files[i]);
! 				}
! 			}
! 		}
! 		return list.iterator();
  	}
  
! 	// this should be a method call into ResourceStore...
! 	private Iterator getNewArcs() {
  		StringToStringTable queued = getQueuedFiles();
  		ArrayList newArcs = new ArrayList();
  
! 		String arcs[] = arcDir.list();
! 		if (arcs != null) {
! 			for (int i = 0; i < arcs.length; i++) {
! 				File arc = new File(arcDir,arcs[i]);
! 				if(arc.isFile() && arcs[i].endsWith(".arc.gz")) {
! 					if (!queued.contains(arcs[i])) {
! 						newArcs.add(arcs[i]);
! 					}
! 				}
  			}
  		}
! 		return newArcs.iterator();
  	}
  
! 	private void queueArcForIndex(final String newArc) throws IOException {
! 		File newQueuedFile = new File(queuedDir,newArc);
! 		File newToBeIndexedFile = new File(toBeIndexedDir,newArc);
  		newToBeIndexedFile.createNewFile();
  		newQueuedFile.createNewFile();
***************
*** 126,182 ****
  
  	/**
! 	 * Find all new ARC files, and queue them for indexing.
! 	 * 
  	 * @throws IOException
  	 */
! 	public void queueNewArcs() throws IOException {
! 		ArrayList newArcs = getNewArcs();
! 		if (!newArcs.isEmpty()) {
! 			Iterator itr = newArcs.iterator();
! 			while (itr.hasNext()) {
! 				String newArc = (String) itr.next();
! 				queueArc(newArc);
! 			}
  		}
  	}
! 
  	/**
  	 * Index any ARC files queued for indexing, queueing the resulting CDX files
  	 * for merging with the BDBResourceIndex.
  	 * 
  	 * @throws MalformedURLException
  	 * @throws IOException
  	 */
! 	public void indexArcs() throws MalformedURLException, IOException {
! 		queueNewArcs();
! 		String toBeIndexed[] = this.toBeIndexedDir.list();
! 		for (int i = 0; i < toBeIndexed.length; i++) {
! 
! 			String base = toBeIndexed[i];
  
! 			File arcFile = new File(this.arcDir.getAbsolutePath().concat(
! 					"/" + base));
! 			File tmpFile = new File(this.indexingDir.getAbsolutePath().concat(
! 					"/" + base));
! 			File flagFile = new File(this.toBeIndexedDir.getAbsolutePath()
! 					.concat("/" + base));
! 			File finalFile = new File(this.mergeDir.getAbsolutePath().concat(
! 					"/" + base));
  
! 			ResourceResults res = indexer.indexArc(arcFile.getAbsolutePath());
! 			indexer.serializeResults(res, tmpFile.getAbsolutePath());
! 			if (!tmpFile.renameTo(finalFile)) {
  				throw new IOException("Unable to move "
! 						+ tmpFile.getAbsolutePath() + " to "
! 						+ finalFile.getAbsolutePath());
  			}
! 			if (!flagFile.delete()) {
  				throw new IOException("Unable to delete "
! 						+ flagFile.getAbsolutePath());
  			}
  		}
  	}
  
  	/**
  	 * @param args
  	 */
--- 222,312 ----
  
  	/**
! 	 * Find any new ARC files and queue them for indexing. 
  	 * @throws IOException
  	 */
! 	public void queueNewArcsForIndex() throws IOException {
! 		Iterator newArcs = getNewArcs();
! 		while(newArcs.hasNext()) {
! 			String newArc = (String) newArcs.next();
! 			queueArcForIndex(newArc);
  		}
  	}
! 	
  	/**
  	 * Index any ARC files queued for indexing, queueing the resulting CDX files
  	 * for merging with the BDBResourceIndex.
  	 * 
+ 	 * @param indexer
  	 * @throws MalformedURLException
  	 * @throws IOException
  	 */
! 	public void indexArcs(ArcIndexer indexer) throws MalformedURLException, IOException {
! 		Iterator toBeIndexed = getDirFilesIterator(toBeIndexedDir);
! 		while(toBeIndexed.hasNext()) {
! 			String base = (String) toBeIndexed.next();
  
! 			File arcFile = new File(arcDir,base);
! 			File toBeIndexedFlagFile = new File(toBeIndexedDir,base);
! 			File indexFile = new File(indexingDir,base);
! 			File toBeMergedFile = new File(toBeMergedDir,base);
  
! 			ResourceResults res = indexer.indexArc(arcFile);
! 			indexer.serializeResults(res, indexFile);
! 			if (!indexFile.renameTo(toBeMergedFile)) {
  				throw new IOException("Unable to move "
! 						+ indexFile.getAbsolutePath() + " to "
! 						+ toBeMergedFile.getAbsolutePath());
  			}
! 			if (!toBeIndexedFlagFile.delete()) {
  				throw new IOException("Unable to delete "
! 						+ toBeIndexedFlagFile.getAbsolutePath());
! 			}
! 		}
! 	}
! 
! 	/**
! 	 * Add any new CDX files in toBeMergedDir to the BDB, deleting the CDX
! 	 * files as they are merged
! 	 * @param dbWriter
! 	 */
! 	public void mergeIndex(BDBResourceIndexWriter dbWriter) {
! 		int numMerged = 0;
! 		Iterator toBeMerged = getDirFilesIterator(toBeMergedDir);
! 		while(toBeMerged.hasNext()) {
! 			
! 			File indexFile = new File(toBeMergedDir,(String) toBeMerged.next());
! 			
! 			try {
! 				dbWriter.importFile(indexFile);
! 				if (!indexFile.delete()) {
! 					throw new IOException("Unable to unlink "
! 							+ indexFile.getAbsolutePath());
! 				}
! 				numMerged++;
! 			} catch (Exception e) {
! 				e.printStackTrace();
  			}
  		}
+ 		if (numMerged > 0) {
+ 			System.out.println("Merged " + numMerged + " files.");
+ 		}
  	}
  
  	/**
+ 	 * Gather a snapshot of the pipeline in a PipelineStatus object.
+ 	 * @return PipelineStatus
+ 	 */
+ 	public PipelineStatus getStatus() {
+ 		PipelineStatus status = new PipelineStatus();
+ 		String index[] = toBeIndexedDir.list();
+ 		String merge[] = toBeMergedDir.list();
+ 		String numQueuedForIndex = (index == null) ? "0" : "" + index.length;
+ 		String numQueuedForMerge = (merge == null) ? "0" : "" + merge.length;
+ 		status.setNumQueuedForIndex(numQueuedForIndex);
+ 		status.setNumQueuedForMerge(numQueuedForMerge);
+ 		return status;
+ 	}
+ 	
+ 	/**
  	 * @param args
  	 */
***************
*** 186,194 ****
  
  	/**
! 	 * @return Returns the mergeDir.
  	 */
! 	public File getMergeDir() {
! 		return mergeDir;
! 	}
  
  }
--- 316,369 ----
  
  	/**
! 	 * Thread that repeatedly runs processing of an IndexPipeline and merges new
! 	 * data into a BDBResourceIndex
! 	 * 
! 	 * @author Brad Tofel
! 	 * @version $Date$, $Revision$
  	 */
! 	private class IndexPipelineThread extends Thread {
! 		private final static int SLEEP_MILLISECONDS = 10000;
! 
! 		private BDBResourceIndexWriter merger = null;
! 		private ArcIndexer indexer = new ArcIndexer();
! 		IndexPipeline pipeline = null;
! 
! 		/**
! 		 * Constructor
! 		 * 
! 		 * @param bdb
! 		 *            initialized BDBResourceIndex
! 		 * @param pipeline
! 		 *            initialized IndexPipeline
! 		 */
! 		public IndexPipelineThread(final BDBResourceIndex bdb,
! 				IndexPipeline pipeline) {
! 			super("IndexPipelineThread");
! 			super.setDaemon(true);
! 			merger = new BDBResourceIndexWriter();
! 			merger.init(bdb);
! 			this.pipeline = pipeline;
! 			System.out.print("Pipeline Thread is ALIVE!");
! 		}
  
+ 		public void run() {
+ 
+ 			while (true) {
+ 				try {
+ 					pipeline.queueNewArcsForIndex();
+ 					pipeline.indexArcs(indexer);
+ 					pipeline.mergeIndex(merger);
+ 					sleep(SLEEP_MILLISECONDS);
+ 				} catch (InterruptedException e) {
+ 					e.printStackTrace();
+ 				// System.out.println("I'm running!"); catch (MalformedURLException e) {
+ 					// TODO Auto-generated catch block
+ 					e.printStackTrace();
+ 				} catch (IOException e) {
+ 					// TODO Auto-generated catch block
+ 					e.printStackTrace();
+ 				}
+ 			}
+ 		}
+ 	}
  }

--- NEW FILE: PipelineFilter.java ---
/* PipeLineServletFilter
 *
 * Created on Oct 20, 2005
 *
 * Copyright (C) 2005 Internet Archive.
 *
 * This file is part of the wayback (crawler.archive.org).
 *
 * wayback is free software; you can redistribute it and/or modify
 * it under the terms of the GNU Lesser Public License as published by
 * the Free Software Foundation; either version 2.1 of the License, or
 * any later version.
 *
 * wayback is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Lesser Public License for more details.
 *
 * You should have received a copy of the GNU Lesser Public License
 * along with wayback; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 */

package org.archive.wayback.arcindexer;

import java.io.IOException;
import java.util.Enumeration;
import java.util.Properties;

import javax.servlet.Filter;
import javax.servlet.FilterChain;
import javax.servlet.FilterConfig;
import javax.servlet.RequestDispatcher;
import javax.servlet.ServletContext;
import javax.servlet.ServletException;
import javax.servlet.ServletRequest;
import javax.servlet.ServletResponse;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;

/**
 * @author brad
 * 
 */
public class PipelineFilter implements Filter {

	private final String PIPELINE_STATUS_JSP = "pipeline.statusjsp";

	private IndexPipeline pipeline = null;

	private String pipelineStatusJsp = null;

	/**
	 * Constructor
	 */
	public PipelineFilter() {
		super();
	}

	public void init(FilterConfig c) throws ServletException {

		Properties p = new Properties();

		pipelineStatusJsp = c.getInitParameter(PIPELINE_STATUS_JSP);
		if ((pipelineStatusJsp == null) || (pipelineStatusJsp.length() <= 0)) {
			throw new ServletException("No config (" + PIPELINE_STATUS_JSP
					+ ")");
		}

		ServletContext sc = c.getServletContext();
		for (Enumeration e = sc.getInitParameterNames(); e.hasMoreElements();) {
			String key = (String) e.nextElement();
			p.put(key, sc.getInitParameter(key));
		}

		pipeline = new IndexPipeline();
		try {
			pipeline.init(p);
		} catch (IOException e) {
			throw new ServletException(e.getMessage());
		}
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see javax.servlet.Filter#doFilter(javax.servlet.ServletRequest,
	 *      javax.servlet.ServletResponse, javax.servlet.FilterChain)
	 */
	public void doFilter(ServletRequest request, ServletResponse response,
			FilterChain chain) throws IOException, ServletException {
		if (!handle(request, response)) {
			chain.doFilter(request, response);
		}
	}

	protected boolean handle(final ServletRequest request,
			final ServletResponse response) throws IOException,
			ServletException {
		if (!(request instanceof HttpServletRequest)) {
			return false;
		}
		if (!(response instanceof HttpServletResponse)) {
			return false;
		}
		HttpServletRequest httpRequest = (HttpServletRequest) request;

		PipelineStatus status = pipeline.getStatus();

		request.setAttribute("pipelinestatus", status);
		RequestDispatcher dispatcher = httpRequest
				.getRequestDispatcher(pipelineStatusJsp);

		dispatcher.forward(request, response);
		return true;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see javax.servlet.Filter#destroy()
	 */
	public void destroy() {

	}
}