rdfrepository.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.memory;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;
import java.util.zip.GZIPOutputStream;

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

import org.openrdf.rio.RdfDocumentWriter;
import org.openrdf.rio.ntriples.NTriplesWriter;
import org.openrdf.rio.rdfxml.RdfXmlWriter;
import org.openrdf.rio.turtle.TurtleWriter;

import org.openrdf.sesame.constants.RDFFormat;
import org.openrdf.sesame.export.RdfExport;
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.util.SailChangedEventImpl;

/**
 * An implementation of the RdfRepository interface extending the class
 * org.openrdf.sesame.sail.memory.RdfSource with write-methods.
 *
 * @author Arjohn Kampman
 * @version $Revision: 1.22 $
 **/
public class RdfRepository extends RdfSource
	implements org.openrdf.sesame.sail.RdfRepository
{

/*--------------+
| Constants     |
+--------------*/

	/** Key used to specify a file for persistent storage. **/
	public static final String SYNC_DELAY_KEY = "syncDelay";

/*----------+
| Variables |
+----------*/

	/**
	 * Flag indicating whether a transaction has been started.
	 **/
	protected boolean _transactionStarted;

	/**
	 * Flag indicating whether the contents of this repository have changed.
	 **/
	protected boolean _contentsChanged;
	
	protected SailChangedEventImpl _sailChangedEvent;

	/**
	 * The sync delay.
	 * @see #setSyncDelay
	 **/
	protected long _syncDelay;

	/**
	 * The timer used to trigger file synchronization.
	 **/
	protected Timer _syncTimer;

	protected List _sailChangedListeners;
	
/*-------------+
| Constructors |
+-------------*/

	/**
	 * Creates a new RdfRepository.
	 **/
	public RdfRepository() {
		super();

		_transactionStarted = false;
		_file = null;
		_syncDelay = 0L;
		
		_sailChangedListeners = new ArrayList(0);
	}

/*------------------------------------------+
| Methods from org.openrdf.sesame.sail.Sail |
+------------------------------------------*/

	// Overrides RdfSource.initialize(Map)
	public void initialize(Map configParams)
		throws SailInitializationException
	{
		super.initialize(configParams);

		// Check for presence of a syncDelay property
		String syncDelay = (String)configParams.get(SYNC_DELAY_KEY);
		if (syncDelay != null) {
			try {
				setSyncDelay(Long.parseLong(syncDelay));
			}
			catch (NumberFormatException e) {
				throw new SailInitializationException(
						"Illegal value for syncDelay parameter: " + syncDelay);
			}
		}
	}

	// Overrides RdfSource.initialize(File, RDFFormat, boolean)
	public void initialize(File file, RDFFormat dataFormat, boolean compressFile)
		throws SailInitializationException
	{
		super.initialize(file, dataFormat, compressFile);

		// Nothing changed yet
		_contentsChanged = false;

		if (_file != null) {
			// A file for persistent storage has been specified

			if (_file.exists()) {
				// File already exists, check write access
				if (!_file.canWrite()) {
					throw new SailInitializationException("File is not writeable: " + _file.getPath());
				}
			}
			else {
				// File does not yet exist, try to create it
				try {
					File parentDir = _file.getParentFile();
					if (parentDir != null && !parentDir.exists()) {
						boolean created = parentDir.mkdirs();
						if (!created) {
							throw new SailInitializationException(
									"Unable to create directory for file " + _file.getPath());
						}
					}

					_file.createNewFile();

					if (_compressFile) {
						// An empty gzipped file isn't really empty. We need
						// to write at least the gzip header to the file, or
						// we will not be able to read it the next time (an
						// empty file is not a legal gzip-file).
						_contentsChanged = true;
					}
				}
				catch (IOException e) {
					throw new SailInitializationException(
							"Unable to create file " + _file.getPath(), e);
				}
			}
		}
	}

	public void shutDown() {
		sync();
		super.shutDown();
	}

/*---------------------------------------------------+
| Methods from org.openrdf.sesame.sail.RdfRepository |
+---------------------------------------------------*/

	public void startTransaction() {
		_stopSyncTimer();

		_transactionStarted = true;
		_contentsChanged = true;
		
		_sailChangedEvent = new SailChangedEventImpl();
	}

	public void commitTransaction() {
		_exportStatusUpToDate = false;
		_transactionStarted = false;

		_notifySailChanged(_sailChangedEvent);
		_sailChangedEvent = null;
		_startSyncTimer();
	}

	public boolean transactionStarted() {
		return _transactionStarted;
	}

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

		_addStatement(subj, pred, obj);
		_sailChangedEvent.setStatementsAdded(true);
	}

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

		if (subj == null && pred == null && obj == null) {
			// Everything should be removed
			int result = _statements.size();
			clearRepository();
			return result;
		}

		// Get the ValueNodes
		ResourceNode subjNode = (subj == null) ? null : _getResourceNode(subj);
		URINode predNode      = (pred == null) ? null : _getURINode(pred);
		ValueNode objNode     = (obj  == null) ? null : _getValueNode(obj);

		if (subjNode == null && subj != null ||
			predNode == null && pred != null ||
			objNode  == null && obj  != null)
		{
			// One of the specified parameters did not have an associated
			// ValueNode, no result will be found.
			return 0;
		}

		boolean patternHasWildcards =
				subjNode == null || predNode == null || objNode == null;

		int oldSize = _statements.size();

		for (int i = oldSize - 1; i >= 0; --i) {
			Statement st = (Statement)_statements.get(i);

			if (_matchesForRemoval(st, subjNode, predNode, objNode)) {
				// Remove statement from nodes
				ResourceNode rn = (ResourceNode)st.getSubject();
				rn.removeSubjectStatement(st);

				URINode un = (URINode)st.getPredicate();
				un.removePredicateStatement(st);

				ValueNode vn = (ValueNode)st.getObject();
				vn.removeObjectStatement(st);

				// Remove this statement from the list
				_statements.remove(i);
				_sailChangedEvent.setStatementsRemoved(true);

				if (!patternHasWildcards) {
					// No wildcards, we'll never remove more than one statement
					// as the statements in the list are unique
					return 1;
				}
			}
		}

		return oldSize - _statements.size();
	}

	public void clearRepository()
		throws SailUpdateException
	{
		if (!transactionStarted()) {
			throw new SailUpdateException("no transaction started.");
		}
		// New containers are created, instead of clearing the old ones.
		// Clearing a collection doesn't make it smaller.

		_statements = new StatementList(256);

		_namespacesTable = new HashMap();
		_namespacesList = new ArrayList();
		_nextNsPrefixId = 1;

		_addDefaultNamespaces();

		_uriNodesMap = new HashMap();
		_bNodeNodesMap = new HashMap();
		_literalNodesMap = new HashMap();

		_updateBNodePrefix();
		_sailChangedEvent.setStatementsRemoved(true);
	}

	public void changeNamespacePrefix(String namespace, String prefix)
		throws SailUpdateException
	{
		// Check if prefix is already used for another namespace
		Iterator iter = _namespacesList.iterator();
		while (iter.hasNext()) {
			Namespace ns = (Namespace)iter.next();

			if (ns.getPrefix().equals(prefix) &&
				!ns.getName().equals(namespace))
			{
				throw new SailUpdateException("Prefix '" + prefix +
						"' is already used for another namespace");
			}
		}

		// Update the prefix
		Namespace n = (Namespace)_namespacesTable.get(namespace);
		if (n != null) {
			n.setPrefix(prefix);
		}
	}

/*----------------------------+
| Matching statement patterns |
+----------------------------*/

	/**
	 * Checks whether a Statement matches a pattern of subject, predicate
	 * and object that has been specified for removal of the concerning
	 * statements. The pattern can contain null values to indicate wild cards.
	 **/
	protected boolean _matchesForRemoval(Statement st, Resource subj, URI pred, Value obj) {
		return
			(subj == null || subj.equals(st.getSubject())) &&
			(pred == null || pred.equals(st.getPredicate())) &&
			( obj == null ||  obj.equals(st.getObject()));
	}

/*-----------------+
| Writing to files |
+-----------------*/

	/**
	 * Sets the time (in milliseconds) to wait after a transaction was commited
	 * before writing the changed data to file. Setting this variable to 0 will
	 * force a file sync immediately after each commit. A negative value will
	 * deactivate file synchronization until the Sail is shut down. A positive
	 * value will postpone the synchronization for at least that amount of
	 * milliseconds. If in the meantime a new transaction is started, the file
	 * synchronization will be rescheduled to wait for another
	 * <tt>syncDelay</tt> ms. This way, bursts of transaction events can
	 * be combined in one file sync.
	 * <p>
	 * The default value for this parameter is <tt>0</tt> (immediate
	 * synchronization).
	 *
	 * @param syncDelay The sync delay in milliseconds.
	 **/
	public void setSyncDelay(long syncDelay) {
		_syncDelay = syncDelay;
	}

	/**
	 * Gets the currently configured sync delay.
	 *
	 * @return syncDelay The sync delay in milliseconds.
	 * @see #setSyncDelay
	 **/
	public long getSyncDelay() {
		return _syncDelay;
	}

	protected synchronized void _startSyncTimer() {
		if (_syncDelay == 0L) {
			// Sync immediately
			sync();
		}
		else if (_syncDelay > 0L) {
			// Sync in _syncDelay milliseconds
			_syncTimer = new Timer();
			TimerTask tt = new TimerTask() {
					public void run() {
						sync();
					}
				};
			_syncTimer.schedule(tt, _syncDelay);
		}
	}

	protected synchronized void _stopSyncTimer() {
		if (_syncTimer != null) {
			_syncTimer.cancel();
			_syncTimer = null;
		}
	}

	/**
	 * Synchronizes the contents of this repository with the data that is stored
	 * on disk. Data will only be written when the contents of the repository
	 * and data in the file are out of sync.
	 **/
	public void sync() {
		if (_contentsChanged && _file != null) {
			_writeToFile();
			_contentsChanged = false;
		}
	}

	protected void _writeToFile() {
		try {
			OutputStream out = new FileOutputStream(_file);
			if (_compressFile) {
				out = new GZIPOutputStream(out, 1024);
			}

			RdfDocumentWriter docWriter = null;
			if (_dataFormat == RDFFormat.RDFXML) {
				docWriter = new RdfXmlWriter(out);
			}
			else if (_dataFormat == RDFFormat.NTRIPLES) {
				docWriter = new NTriplesWriter(out);
			}
			else if (_dataFormat == RDFFormat.TURTLE) {
				docWriter = new TurtleWriter(out);
			}
			else {
				throw new SailInternalException("Illegal value for data format: " + _dataFormat.toString());
			}

			_exportData(docWriter);

			out.flush();
			out.close();
		}
		catch (IOException e) {
			throw new SailInternalException(e);
		}
	}

	protected void _exportData(RdfDocumentWriter docWriter)
		throws IOException
	{
		RdfExport rdfExport = new RdfExport();
		rdfExport.exportRdf(this, docWriter, false);
	}

	/* (non-Javadoc)
	 * @see org.openrdf.sesame.sail.RdfRepository#addListener(org.openrdf.sesame.sail.SailChangedListener)
	 */
	public void addListener(SailChangedListener listener) {
		synchronized(_sailChangedListeners) {
			_sailChangedListeners.add(listener);
		} // end synchronized block
	}

	/* (non-Javadoc)
	 * @see org.openrdf.sesame.sail.RdfRepository#removeListener(org.openrdf.sesame.sail.SailChangedListener)
	 */
	public void removeListener(SailChangedListener listener) {
		synchronized(_sailChangedListeners) {
			_sailChangedListeners.remove(listener);
		} // end synchronized block
	}
	
	
	protected void _notifySailChanged(SailChangedEventImpl event) {
		synchronized(_sailChangedListeners) {
			if (_sailChangedListeners != null) {
				if (event.sailChanged()) { // only notify if something actually changed
					Iterator listeners = _sailChangedListeners.iterator();
					
					while (listeners.hasNext()) {
						SailChangedListener listener = (SailChangedListener)listeners.next();
						listener.sailChanged(event);
					}
				}
			}
		} // end synchronized block
	}
}