[Archive-access-cvs] archive-access/projects/wayback/src/java/org/archive/wayback/cdx/indexer BDBResourceIndexWriter.java,NONE,1.1 ArcIndexer.java,NONE,1.1 PipelineFilter.java,NONE,1.1 PipelineStatus.java,NONE,1.1 IndexPipeline.java,NONE,1.1

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

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

Added Files:
	BDBResourceIndexWriter.java ArcIndexer.java 
	PipelineFilter.java PipelineStatus.java IndexPipeline.java 
Log Message:
Massive overhaul decomposing into three main categories of changes:

  1) All internal datatypes are now extensible (currently Properties, but should be Maps) including:
	a) WaybackRequest(was WBRequest)
	b) SearchResults (was ResourceResults)
	c) SearchResult (was ResourceResult)
	d) Resource

	so that there is no longer an assumption of Archival URL queries, or "CDX-style" index results. This will put more responsiblility on the UI components to interrogate SearchResults to decide how to render, but should enable extension to data returned from Indexes, as well as allow far more flexibility in queries, predominantly geared towards free-text searching. This is still somewhat clunky, as there are no convenience accessor methods, so all users refer to constants when interacting with them.

  2) Major cleanup of servlet and filter interaction with servlet container. ReplayUI and QueryUI are now just plain old servlets, and filters can be optionally added to allow non-CGI argument requests to be coerced into standard WaybackRequest objects.

  3) Alternate "Proxy" Replay mode is now functional, and some work has been done towards an alternate Nutch ResourceIndex. Currently the web.xml contains example configurations for both Proxy and Archival Url replay modes, but the Proxy related configurations are commented out. Proxy mode *requires* changing the servlet context to ROOT. ArchivalUrl replay mode works as ROOT context and as any (I think) other context. There are some cosmetic double-slashe issues to work out.


--- 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.cdx.indexer;

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

import org.archive.wayback.cdx.BDBResourceIndex;
import org.archive.wayback.cdx.CDXRecord;
import org.archive.wayback.core.SearchResult;
import org.archive.wayback.core.SearchResults;

import com.sleepycat.je.DatabaseException;

/**
 * Implements updates to a BDBResourceIndex
 * 
 * @author Brad Tofel
 * @version $Date: 2005/11/16 03:11:29 $, $Revision: 1.1 $
 */
public class BDBResourceIndexWriter {
	// TODO: move to somewhere better...
	private final static String CDX_HEADER_MAGIC = " CDX ";
	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 {
		SearchResults results = readFile(indexFile);
		db.addResults(results);
	}

	private SearchResults readFile(File indexFile) throws Exception {
		RandomAccessFile raFile = new RandomAccessFile(indexFile, "r");
		SearchResults results = new SearchResults();
		int lineNumber = 0;
		CDXRecord cdxRecord = new CDXRecord();
		while (true) {
			String line = raFile.readLine();
			if (line == null) {
				break;
			}
			lineNumber++;
			if ((lineNumber == 1) && (line.contains(CDX_HEADER_MAGIC))) {
				continue;
			}
			cdxRecord.parseLine(line, lineNumber);
			SearchResult result = cdxRecord.toSearchResult();
			results.addSearchResult(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();
		}

	}

}

--- NEW FILE: ArcIndexer.java ---
/* ArcIndexer
 *
 * 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.cdx.indexer;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.text.ParseException;
import java.util.Iterator;

import org.archive.io.arc.ARCReader;
import org.archive.io.arc.ARCReaderFactory;
import org.archive.io.arc.ARCRecord;
import org.archive.io.arc.ARCRecordMetaData;
import org.archive.net.UURI;
import org.archive.wayback.WaybackConstants;
import org.archive.wayback.core.SearchResult;
import org.archive.wayback.core.SearchResults;
import org.apache.commons.httpclient.Header;

/**
 * Transforms an ARC file into ResourceResults, or a serialized ResourceResults
 * file(CDX).
 * 
 * @author Brad Tofel
 * @version $Date: 2005/11/16 03:11:29 $, $Revision: 1.1 $
 */
public class ArcIndexer {
	private final static String LOCATION_HTTP_HEADER = "Location";
    private final static String CDX_HEADER_STRING = " CDX N b h m s k r V g";


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

	/**
	 * Create a ResourceResults representing the records in ARC file at arcPath.
	 * 
	 * @param arc
	 * @return ResourceResults in arcPath.
	 * @throws IOException
	 */
	public SearchResults indexArc(File arc) throws IOException {
		SearchResults results = new SearchResults();
		ARCReader arcReader = ARCReaderFactory.get(arc);
		arcReader.setParseHttpHeaders(true);
		// doh. this does not generate quite the columns we need:
		// arcReader.createCDXIndexFile(arcPath);
		Iterator itr = arcReader.iterator();
		while (itr.hasNext()) {
			ARCRecord rec = (ARCRecord) itr.next();
			SearchResult result;
			try {
				result = arcRecordToSearchResult(rec, arc);
			} catch (NullPointerException e) {
				e.printStackTrace();
				continue;
			} catch (ParseException e) {
				e.printStackTrace();
				continue;
			}
			if(result != null) {
				results.addSearchResult(result);
			}
		}
		return results;
	}

	private SearchResult arcRecordToSearchResult(final ARCRecord rec,
			File arc) throws NullPointerException, IOException, ParseException {
		rec.close();
		ARCRecordMetaData meta = rec.getMetaData();

		SearchResult result = new SearchResult();
		result.put(WaybackConstants.RESULT_ARC_FILE,arc.getName());
		result.put(WaybackConstants.RESULT_OFFSET,""+meta.getOffset());

		String statusCode = (meta.getStatusCode() == null) ? "-" : meta
				.getStatusCode();
		result.put(WaybackConstants.RESULT_HTTP_CODE,statusCode);

		result.put(WaybackConstants.RESULT_MD5_DIGEST,meta.getDigest());
		result.put(WaybackConstants.RESULT_MIME_TYPE,meta.getMimetype());
		
		String uriStr = meta.getUrl();
		if(uriStr.startsWith(ARCRecord.ARC_MAGIC_NUMBER)) {
			// skip filedesc record...
			return null;
		}
		UURI uri = new UURI(uriStr, false);
		result.put(WaybackConstants.RESULT_ORIG_HOST,uri.getHost());

		String redirectUrl = "-";
		Header[] headers = rec.getHttpHeaders();
		if (headers != null) {
			for (int i = 0; i < headers.length; i++) {
				if (headers[i].getName().equals(LOCATION_HTTP_HEADER)) {
					redirectUrl = headers[i].getValue();
					break;
				}
			}
		}
		result.put(WaybackConstants.RESULT_REDIRECT_URL,redirectUrl);
		result.put(WaybackConstants.RESULT_CAPTURE_DATE,meta.getDate());
		UURI uriCap = new UURI(meta.getUrl(), false);
		String searchHost = uriCap.getHostBasename();
		String searchPath = uriCap.getEscapedPathQuery();

		String indexUrl = searchHost + searchPath;
		result.put(WaybackConstants.RESULT_URL,indexUrl);

		return result;
	}

	/**
	 * Write out ResourceResults into CDX file at cdxPath
	 * 
	 * @param results
	 * @param target
	 * @throws IOException
	 */
	public void serializeResults(final SearchResults results,
			File target) throws IOException {
		
		// TODO will this automatically close when it falls out of scope?
		FileOutputStream output = new FileOutputStream(target);
		output.write((CDX_HEADER_STRING + "\n").getBytes());

		Iterator itr = results.iterator();
		while (itr.hasNext()) {
			SearchResult result = (SearchResult) itr.next();
			output.write((result.toString() + "\n").getBytes());
		}
	}

	/**
	 * @param args
	 */
	public static void main(String[] args) {
		ArcIndexer indexer = new ArcIndexer();
		File arc = new File(args[0]);
		File cdx = new File(args[1]);
		try {
			SearchResults results = indexer.indexArc(arc);
			indexer.serializeResults(results, cdx);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

}

--- 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.cdx.indexer;

/**
 * Data bag for handing off status of Pipeline to PipelineStatus.jsp.
 * 
 * @author brad
 * @version $Date: 2005/11/16 03:11:29 $, $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

	}

}

--- NEW FILE: IndexPipeline.java ---
/* IndexPipeline
 *
 * 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.cdx.indexer;

import java.io.File;
import java.io.IOException;
import java.net.MalformedURLException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.Properties;

import org.archive.wayback.PropertyConfigurable;
import org.archive.wayback.cdx.BDBResourceIndex;
import org.archive.wayback.core.SearchResults;
import org.archive.wayback.exception.ConfigurationException;

import com.sleepycat.je.DatabaseException;
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
 * @version $Date: 2005/11/16 03:11:29 $, $Revision: 1.1 $
 */
public class IndexPipeline implements PropertyConfigurable{
	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;

	private File toBeIndexedDir = null;

	private File indexingDir = null;

	private File toBeMergedDir = null;

	private BDBResourceIndex db = null;

	private static Thread indexUpdateThread = null;
	

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

	private void ensureDir(File dir) throws IOException {
		if (!dir.isDirectory() && !dir.mkdirs()) {
			throw new IOException("FAILED to create " + dir.getAbsolutePath());
		}
	}

	/**
	 * Initialize this object, creating directories if needed, and starting
	 * thread if configured.
	 * 
	 * @param p configuration 
	 * @throws IOException
	 */

	public void init(Properties p) throws ConfigurationException {
		
		// 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);
		}

		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);

		try {
			ensureDir(workDir);
			ensureDir(queuedDir);
			ensureDir(toBeIndexedDir);
			ensureDir(indexingDir);
			ensureDir(toBeMergedDir);
			File dbFile = new File(dbPath);
			ensureDir(dbFile);
		} catch (IOException e) {
			e.printStackTrace();
			throw new ConfigurationException(e.getMessage());
		}

		String runPipeline = (String) p.get(RUN_PIPELINE);
		try {
			db = new BDBResourceIndex(dbPath, dbName);
		} catch (DatabaseException e) {
			e.printStackTrace();
			throw new ConfigurationException(e.getMessage());
		}

		if ((runPipeline != null) && (runPipeline.equals("1"))) {

			// TODO: Logger!
			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");
		}
		return hash;
	}

	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();
	}

	/**
	 * 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);

			SearchResults 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
	 */
	public static void main(String[] args) {

	}

	/**
	 * Thread that repeatedly runs processing of an IndexPipeline and merges new
	 * data into a BDBResourceIndex
	 * 
	 * @author Brad Tofel
	 * @version $Date: 2005/11/16 03:11:29 $, $Revision: 1.1 $
	 */
	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();
				} catch (IOException e) {
					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.cdx.indexer;

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;

import org.archive.wayback.exception.ConfigurationException;

/**
 * @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 (ConfigurationException e) {
			e.printStackTrace();
			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() {

	}
}