syncrdfrepository.java

这是外国一个开源推理机

/*  Sesame - Storage and Querying architecture for RDF and RDF Schema
 *  Copyright (C) 2001-2005 Aduna
 *
 *  Contact: 
 *  	Aduna
 *  	Prinses Julianaplein 14 b
 *  	3817 CS Amersfoort
 *  	The Netherlands
 *  	tel. [phone]
 *  	fax. [phone]
 *
 *  	http://aduna.biz/
 *  	http://www.openrdf.org/
 *  
 *  This library is free software; you can redistribute it and/or
 *  modify it under the terms of the GNU Lesser General Public
 *  License as published by the Free Software Foundation; either
 *  version 2.1 of the License, or (at your option) any later version.
 *
 *  This library 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 General Public License for more details.
 *
 *  You should have received a copy of the GNU Lesser General Public
 *  License along with this library; if not, write to the Free Software
 *  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 */

package org.openrdf.sesame.sailimpl.sync;

import java.util.Map;

import org.openrdf.util.log.ThreadLog;

import org.openrdf.model.Resource;
import org.openrdf.model.URI;
import org.openrdf.model.Value;
import org.openrdf.model.ValueFactory;

import org.openrdf.sesame.sail.NamespaceIterator;
import org.openrdf.sesame.sail.RdfRepository;
import org.openrdf.sesame.sail.Sail;
import org.openrdf.sesame.sail.SailChangedListener;
import org.openrdf.sesame.sail.SailInitializationException;
import org.openrdf.sesame.sail.SailInternalException;
import org.openrdf.sesame.sail.SailUpdateException;
import org.openrdf.sesame.sail.StackedSail;
import org.openrdf.sesame.sail.StatementIterator;
import org.openrdf.sesame.sail.query.Query;

/**
 * @author Arjohn Kampman
 * @version $Revision: 1.10 $
 */
public class SyncRdfRepository implements RdfRepository, StackedSail {

	private RdfRepository _rdfRepository;

	private boolean _transactionStarted = false;

	public SyncRdfRepository() {
		ThreadLog.trace("new SyncRdfRepository created");
	}

	public void setBaseSail(Sail sail) {
		if (sail instanceof RdfRepository) {
			_rdfRepository = (RdfRepository)sail;
			ThreadLog.trace("SyncRdfRepository: base sail was set");
		}
		else {
			throw new SailInternalException("base Sail should be an RdfRepository");
		}
	}

	public Sail getBaseSail() {
		return _rdfRepository;
	}

	public void initialize(Map configParams)
		throws SailInitializationException
	{
		if (_rdfRepository == null) {
			throw new SailInitializationException("Base sail was not set");
		}
		ThreadLog.trace("SyncRdfRepository initialized");
	}

	public void shutDown() {
		_getWriteLock();
		try {
			_rdfRepository.shutDown();
		}
		finally {
			_releaseWriteLock();
		}
	}

	public void startTransaction() {
		_getWriteLock();
		try {
			_rdfRepository.startTransaction();
			_transactionStarted = true;
		}
		catch (RuntimeException e) {
			_releaseWriteLock();
			throw e;
		}
	}

	public void commitTransaction() {
		if (!_transactionStarted) {
			throw new SailInternalException("No transaction started");
		}

		try {
			_rdfRepository.commitTransaction();
		}
		finally {
			_transactionStarted = false;
			_releaseWriteLock();
		}
	}

	public boolean transactionStarted() {
		return _transactionStarted;
	}

	public void addStatement(Resource subj, URI pred, Value obj)
		throws SailUpdateException
	{
		if (!_transactionStarted) {
			throw new SailUpdateException("No transaction started");
		}

		_rdfRepository.addStatement(subj, pred, obj);
	}

	public int removeStatements(Resource subj, URI pred, Value obj) 
		throws SailUpdateException
	{
		if (!_transactionStarted) {
			throw new SailUpdateException("No transaction started");
		}
		return _rdfRepository.removeStatements(subj, pred, obj);
	}

	public void clearRepository() 
		throws SailUpdateException
	{
		if (!_transactionStarted) {
			throw new SailUpdateException("No transaction started");
		}
		_rdfRepository.clearRepository();
	}

	public void changeNamespacePrefix(String namespace, String prefix)
		throws SailUpdateException
	{
		if (!_transactionStarted) {
			throw new SailUpdateException("No transaction started");
		}
		_rdfRepository.changeNamespacePrefix(namespace, prefix);
	}

	public ValueFactory getValueFactory() {
		return _rdfRepository.getValueFactory();
	}

	public StatementIterator getStatements(Resource subj, URI pred, Value obj) {
		StatementIterator result = null;

		_getReadLock();
		try {
			result = _rdfRepository.getStatements(subj, pred, obj);
		}
		finally {
			_releaseReadLock();
		}

		return result;
	}

	public boolean hasStatement(Resource subj, URI pred, Value obj) {
		boolean result = false;

		_getReadLock();
		try {
			result = _rdfRepository.hasStatement(subj, pred, obj);
		}
		finally {
			_releaseReadLock();
		}

		return result;
	}

	public Query optimizeQuery(Query qc) {
		
		Query result = qc;
		
		_getReadLock();
		try {
			result = _rdfRepository.optimizeQuery(qc);
		}
		finally {
			_releaseReadLock();
		}
		return result;
	}

	public NamespaceIterator getNamespaces() {
		NamespaceIterator result = null;

		_getReadLock();
		try {
			result = _rdfRepository.getNamespaces();
		}
		finally {
			_releaseReadLock();
		}

		return result;
	}
	
	/* (non-Javadoc)
	 * @see org.openrdf.sesame.sail.RdfRepository#addListener(org.openrdf.sesame.sail.SailChangedListener)
	 */
	public void addListener(SailChangedListener listener) {
		_rdfRepository.addListener(listener);
	}

	/* (non-Javadoc)
	 * @see org.openrdf.sesame.sail.RdfRepository#removeListener(org.openrdf.sesame.sail.SailChangedListener)
	 */
	public void removeListener(SailChangedListener listener) {
		_rdfRepository.removeListener(listener);
	}


/*-----------------------------------------+
| Synchronization methods                  |
+-----------------------------------------*/

	private boolean _writeRequested = false;
	private int _readingThreads = 0;

	protected void _getWriteLock() {
		synchronized (this) {
//ThreadLog.trace("entering _getWriteLock()");
			while (_writeRequested) {
				// Someone else wants to write
				try {
ThreadLog.trace("waiting for other writer to finish");
					wait();
ThreadLog.trace("waking up");
				} catch (InterruptedException ignore) {
				}
			}

			_writeRequested = true;
//ThreadLog.trace("Got the lock");

			// Wait for the readingThreads to finish
			while (_readingThreads > 0) {
				try {
ThreadLog.trace("Waiting for readers to finish");
					wait();
ThreadLog.trace("waking up");
				} catch (InterruptedException ignore) {
				}
			}
//ThreadLog.trace("We've got a go!!!");
		}
	}

	protected void _releaseWriteLock() {
		synchronized (this) {
//ThreadLog.trace("Releasing write lock");
			_writeRequested = false;
			notifyAll();
		}
	}

	protected void _getReadLock() {
		synchronized (this) {
//ThreadLog.trace("entering _getReadLock()");
			// Wait for any writing threads to finish
			while (_writeRequested) {
				try {
ThreadLog.trace("Waiting for writer to finish");
					wait();
ThreadLog.trace("waking up");
				} catch (InterruptedException ignore) {
				}
			}

			// No candidates for writing anymore, go ahead
			_readingThreads++;
//ThreadLog.trace("We've got a go!!! (" + _readingThreads + ")");
		}
	}

	protected void _releaseReadLock() {
		synchronized (this) {
			_readingThreads--;
//ThreadLog.trace("Releasing read lock (" + _readingThreads + ")");
			if (_readingThreads == 0) {
				// Maybe someone wants to write?
				notifyAll();
			}
		}
	}
}