litvinovg/vitro
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

change to listener interface to simplify things and avoid sharing the reasoner's individual queue between concurrent writing threads

Browse source
This commit is contained in:
brianjlowe 2015-12-12 21:51:50 +02:00
parent 295ddbfad1
commit 539bffd6bf
22 changed files with 476 additions and 629 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

40
webapp/src/edu/cornell/mannlib/vitro/webapp/dao/jena/ABoxJenaChangeListener.java
View file

@ -1,40 +0,0 @@
/* $This file is distributed under the terms of the license in /doc/license.txt$ */
package edu.cornell.mannlib.vitro.webapp.dao.jena;
import com.hp.hpl.jena.rdf.model.ModelChangedListener;
import edu.cornell.mannlib.vitro.webapp.modelaccess.ModelNames;
public class ABoxJenaChangeListener extends JenaChangeListener {
public ABoxJenaChangeListener(ModelChangedListener listener) {
super(listener);
ignoredGraphs.add(ModelNames.ABOX_INFERENCES);
ignoredGraphs.add(ModelNames.TBOX_ASSERTIONS);
ignoredGraphs.add(ModelNames.TBOX_INFERENCES);
}
@Override
public void addedStatement(String serializedTriple, String graphURI) {
if (isABoxGraph(graphURI)) {
super.addedStatement(serializedTriple, graphURI);
}
}
@Override
public void removedStatement(String serializedTriple, String graphURI) {
if (isABoxGraph(graphURI)) {
super.removedStatement(serializedTriple, graphURI);
}
}
private boolean isABoxGraph(String graphURI) {
return (graphURI == null ||
ModelNames.ABOX_ASSERTIONS.equals(graphURI)
|| (!ignoredGraphs.contains(graphURI)
&& !graphURI.contains("filegraph")
&& !graphURI.contains("tbox")));
}
}

98
webapp/src/edu/cornell/mannlib/vitro/webapp/dao/jena/JenaChangeListener.java
View file

@ -1,98 +0,0 @@
/* $This file is distributed under the terms of the license in /doc/license.txt$ */
package edu.cornell.mannlib.vitro.webapp.dao.jena;
import java.io.ByteArrayInputStream;
import java.io.UnsupportedEncodingException;
import java.util.HashSet;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import com.hp.hpl.jena.rdf.model.Model;
import com.hp.hpl.jena.rdf.model.ModelChangedListener;
import com.hp.hpl.jena.rdf.model.ModelFactory;
import com.hp.hpl.jena.rdf.model.ResourceFactory;
import com.hp.hpl.jena.rdf.model.Statement;
import com.hp.hpl.jena.rdf.model.StmtIterator;
import com.hp.hpl.jena.shared.Lock;
import com.hp.hpl.jena.vocabulary.OWL;
import com.hp.hpl.jena.vocabulary.RDF;
import edu.cornell.mannlib.vitro.webapp.rdfservice.ChangeListener;
/**
* A ChangeListener that forwards events to a Jena ModelChangedListener
* @author bjl23
*
*/
public class JenaChangeListener implements ChangeListener {
private static final Log log = LogFactory.getLog(JenaChangeListener.class);
protected HashSet<String> ignoredGraphs = new HashSet<String>();
private ModelChangedListener listener;
private Model m = ModelFactory.createDefaultModel();
public JenaChangeListener(ModelChangedListener listener) {
this.listener = listener;
m.register(listener);
// these graphs no longer used
// ignoredGraphs.add(SimpleReasonerSetup.JENA_INF_MODEL_REBUILD);
// ignoredGraphs.add(SimpleReasonerSetup.JENA_INF_MODEL_SCRATCHPAD);
}
@Override
public void addedStatement(String serializedTriple, String graphURI) {
if (isRelevantGraph(graphURI)) {
listener.addedStatement(parseTriple(serializedTriple));
}
}
@Override
public void removedStatement(String serializedTriple, String graphURI) {
if (isRelevantGraph(graphURI)) {
listener.removedStatement(parseTriple(serializedTriple));
}
}
private boolean isRelevantGraph(String graphURI) {
return (graphURI == null || !ignoredGraphs.contains(graphURI));
}
@Override
public void notifyEvent(String graphURI, Object event) {
log.debug("event: " + event.getClass());
listener.notifyEvent(m, event);
}
// TODO avoid overhead of Model
private Statement parseTriple(String serializedTriple) {
try {
m.enterCriticalSection(Lock.WRITE);
m.removeAll();
// Model m = ModelFactory.createDefaultModel();
m.read(new ByteArrayInputStream(
serializedTriple.getBytes("UTF-8")), null, "N-TRIPLE");
StmtIterator sit = m.listStatements();
if (!sit.hasNext()) {
throw new RuntimeException("no triple parsed from change event");
} else {
Statement s = sit.nextStatement();
if (sit.hasNext()) {
log.warn("More than one triple parsed from change event");
}
return s;
}
} catch (RuntimeException riot) {
log.error("Unable to parse triple " + serializedTriple, riot);
throw riot;
} catch (UnsupportedEncodingException uee) {
throw new RuntimeException(uee);
} finally {
m.leaveCriticalSection();
}
}
}

3
webapp/src/edu/cornell/mannlib/vitro/webapp/dao/jena/ModelContext.java
View file

@ -35,8 +35,7 @@ public class ModelContext {
public static void registerListenerForChanges(ServletContext ctx, ModelChangedListener ml){
try {
RDFServiceUtils.getRDFServiceFactory(ctx).registerListener(
new JenaChangeListener(ml));
RDFServiceUtils.getRDFServiceFactory(ctx).registerJenaModelChangedListener(ml);
} catch (RDFServiceException e) {
log.error(e,e);
}

17
webapp/src/edu/cornell/mannlib/vitro/webapp/rdfservice/ChangeListener.java
View file

@ -9,21 +9,12 @@ package edu.cornell.mannlib.vitro.webapp.rdfservice;
public interface ChangeListener {
/**
* Override this to listen to all statements added to the RDF store.
* Override this to listen to each model change
*
* @param serializedTriple - the added statement in n3 format
* @param graphURI - the graph to which the statement was added
* @param modelChange - the object representing the model change
*/
public void addedStatement(String serializedTriple, String graphURI);
/**
* Override this to listen to all statements removed from the RDF store.
*
* @param serializedTriple - the removed statement in n3 format
* @param graphURI - the graph from which the statement was removed
*/
public void removedStatement(String serializedTriple, String graphURI);
public void notifyModelChange(ModelChange modelChange);
/**
* Override this to listen to events pertaining to the given graphURI.
*

31
webapp/src/edu/cornell/mannlib/vitro/webapp/rdfservice/RDFService.java
View file

@ -2,12 +2,13 @@
package edu.cornell.mannlib.vitro.webapp.rdfservice;
import com.hp.hpl.jena.rdf.model.Model;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.List;
import com.hp.hpl.jena.rdf.model.Model;
import com.hp.hpl.jena.rdf.model.ModelChangedListener;
/**
* Interface for API to write, read, and update Vitro's RDF store, with support
* to allow listening, logging and auditing.
@ -197,20 +198,38 @@ public interface RDFService {
ModelSerializationFormat serializationFormat) throws RDFServiceException;
/**
* Registers a listener to listen to changes in any graph in
* Registers a Jena listener to listen to changes in any graph in
* the RDF store.
*
* @param changeListener - the change listener
*/
public void registerListener(ChangeListener changeListener) throws RDFServiceException;
public void registerJenaModelChangedListener(ModelChangedListener changeListener)
throws RDFServiceException;
/**
* Unregisters a listener from listening to changes in
* Unregisters a Jena listener from listening to changes in
* any graph in the RDF store
*
* @param changeListener - the change listener
*/
public void unregisterListener(ChangeListener changeListener) throws RDFServiceException;
public void unregisterJenaModelChangedListener(ModelChangedListener changeListener)
throws RDFServiceException;
/**
* Registers a listener to listen to changes in any graph in
* the RDF store.
*
* @param changeListener - the change listener
*/
public void registerListener(ChangeListener changeListener) throws RDFServiceException;
/**
* Unregisters a listener from listening to changes in
* any graph in the RDF store
*
* @param changeListener - the change listener
*/
public void unregisterListener(ChangeListener changeListener) throws RDFServiceException;
/**
* Creates a ChangeSet object

22
webapp/src/edu/cornell/mannlib/vitro/webapp/rdfservice/RDFServiceFactory.java
View file

@ -2,6 +2,8 @@
package edu.cornell.mannlib.vitro.webapp.rdfservice;
import com.hp.hpl.jena.rdf.model.ModelChangedListener;
public interface RDFServiceFactory {
/**
@ -37,11 +39,29 @@ public interface RDFServiceFactory {
/**
* Unregisters a listener from listening to changes in the RDF store.
* Any RDFService objects returned by this factory should notify
* Any RDFService objects returned by this factory should no longer notify
* this listener of changes.
*
* @param changeListener - the change listener
*/
public void unregisterListener(ChangeListener changeListener) throws RDFServiceException;
/**
* Registers a Jena ModelChangedListener to listen to changes in any graph in
* the RDF store. Any RDFService objects returned by this factory
* should notify this listener of changes.
*
* @param changeListener - the change listener
*/
public void registerJenaModelChangedListener(ModelChangedListener changeListener) throws RDFServiceException;
/**
* Unregisters a Jena ModelChangedListener from listening to changes in the RDF store.
* Any RDFService objects returned by this factory should no longer notify
* this listener of changes.
*
* @param changeListener - the change listener
*/
public void unregisterJenaModelChangedListener(ModelChangedListener changeListener) throws RDFServiceException;
}

17
webapp/src/edu/cornell/mannlib/vitro/webapp/rdfservice/filter/LanguageFilteringRDFService.java
View file

@ -12,7 +12,6 @@ import java.util.Comparator;
import java.util.Iterator;
import java.util.List;
import edu.cornell.mannlib.vitro.webapp.rdfservice.ResultSetConsumer;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@ -22,6 +21,7 @@ import com.hp.hpl.jena.query.ResultSetFactory;
import com.hp.hpl.jena.query.ResultSetFormatter;
import com.hp.hpl.jena.rdf.model.Literal;
import com.hp.hpl.jena.rdf.model.Model;
import com.hp.hpl.jena.rdf.model.ModelChangedListener;
import com.hp.hpl.jena.rdf.model.RDFNode;
import com.hp.hpl.jena.rdf.model.Statement;
import com.hp.hpl.jena.rdf.model.StmtIterator;
@ -30,6 +30,7 @@ import edu.cornell.mannlib.vitro.webapp.rdfservice.ChangeListener;
import edu.cornell.mannlib.vitro.webapp.rdfservice.ChangeSet;
import edu.cornell.mannlib.vitro.webapp.rdfservice.RDFService;
import edu.cornell.mannlib.vitro.webapp.rdfservice.RDFServiceException;
import edu.cornell.mannlib.vitro.webapp.rdfservice.ResultSetConsumer;
import edu.cornell.mannlib.vitro.webapp.rdfservice.impl.RDFServiceUtils;
public class LanguageFilteringRDFService implements RDFService {
@ -434,7 +435,6 @@ public class LanguageFilteringRDFService implements RDFService {
@Override
public void registerListener(ChangeListener changeListener)
throws RDFServiceException {
// TODO Auto-generated method stub
s.registerListener(changeListener);
}
@ -443,6 +443,19 @@ public class LanguageFilteringRDFService implements RDFService {
throws RDFServiceException {
s.unregisterListener(changeListener);
}
@Override
public void registerJenaModelChangedListener(ModelChangedListener changeListener)
throws RDFServiceException {
s.registerJenaModelChangedListener(changeListener);
}
@Override
public void unregisterJenaModelChangedListener(ModelChangedListener changeListener)
throws RDFServiceException {
s.unregisterJenaModelChangedListener(changeListener);
}
@Override
public ChangeSet manufactureChangeSet() {

13
webapp/src/edu/cornell/mannlib/vitro/webapp/rdfservice/filter/SameAsFilteringRDFServiceFactory.java
View file

@ -10,7 +10,6 @@ import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import edu.cornell.mannlib.vitro.webapp.rdfservice.ResultSetConsumer;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@ -23,6 +22,7 @@ import com.hp.hpl.jena.query.ResultSet;
import com.hp.hpl.jena.query.ResultSetFactory;
import com.hp.hpl.jena.query.ResultSetFormatter;
import com.hp.hpl.jena.rdf.model.Model;
import com.hp.hpl.jena.rdf.model.ModelChangedListener;
import com.hp.hpl.jena.rdf.model.ModelFactory;
import com.hp.hpl.jena.rdf.model.RDFNode;
import com.hp.hpl.jena.rdf.model.Resource;
@ -36,6 +36,7 @@ import edu.cornell.mannlib.vitro.webapp.rdfservice.RDFService;
import edu.cornell.mannlib.vitro.webapp.rdfservice.RDFService.ModelSerializationFormat;
import edu.cornell.mannlib.vitro.webapp.rdfservice.RDFServiceException;
import edu.cornell.mannlib.vitro.webapp.rdfservice.RDFServiceFactory;
import edu.cornell.mannlib.vitro.webapp.rdfservice.ResultSetConsumer;
import edu.cornell.mannlib.vitro.webapp.rdfservice.impl.RDFServiceImpl;
import edu.cornell.mannlib.vitro.webapp.rdfservice.impl.RDFServiceUtils;
@ -76,6 +77,16 @@ public class SameAsFilteringRDFServiceFactory implements RDFServiceFactory {
f.registerListener(changeListener);
}
@Override
public void registerJenaModelChangedListener(ModelChangedListener changeListener) throws RDFServiceException {
f.registerJenaModelChangedListener(changeListener);
}
@Override
public void unregisterJenaModelChangedListener(ModelChangedListener changeListener) throws RDFServiceException {
f.registerJenaModelChangedListener(changeListener);
}
public class SameAsFilteringRDFService extends RDFServiceImpl implements RDFService {
private final Log log = LogFactory.getLog(SameAsFilteringRDFService.class);

24
webapp/src/edu/cornell/mannlib/vitro/webapp/rdfservice/impl/RDFServiceFactorySingle.java
View file

@ -7,6 +7,8 @@ import java.io.OutputStream;
import java.util.List;
import com.hp.hpl.jena.rdf.model.Model;
import com.hp.hpl.jena.rdf.model.ModelChangedListener;
import edu.cornell.mannlib.vitro.webapp.rdfservice.ChangeListener;
import edu.cornell.mannlib.vitro.webapp.rdfservice.ChangeSet;
import edu.cornell.mannlib.vitro.webapp.rdfservice.RDFService;
@ -48,6 +50,16 @@ public class RDFServiceFactorySingle implements RDFServiceFactory {
this.rdfService.unregisterListener(listener);
}
@Override
public void registerJenaModelChangedListener(ModelChangedListener listener) throws RDFServiceException {
this.rdfService.registerJenaModelChangedListener(listener);
}
@Override
public void unregisterJenaModelChangedListener(ModelChangedListener listener) throws RDFServiceException {
this.rdfService.unregisterJenaModelChangedListener(listener);
}
public class UnclosableRDFService implements RDFService {
private RDFService s;
@ -156,6 +168,18 @@ public class RDFServiceFactorySingle implements RDFServiceFactory {
throws RDFServiceException {
s.unregisterListener(changeListener);
}
@Override
public void registerJenaModelChangedListener(ModelChangedListener changeListener)
throws RDFServiceException {
s.registerJenaModelChangedListener(changeListener);
}
@Override
public void unregisterJenaModelChangedListener(ModelChangedListener changeListener)
throws RDFServiceException {
s.unregisterJenaModelChangedListener(changeListener);
}
@Override
public ChangeSet manufactureChangeSet() {

77
webapp/src/edu/cornell/mannlib/vitro/webapp/rdfservice/impl/RDFServiceImpl.java
View file

@ -3,12 +3,12 @@
package edu.cornell.mannlib.vitro.webapp.rdfservice.impl;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import edu.cornell.mannlib.vitro.webapp.rdfservice.ResultSetConsumer;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@ -18,20 +18,22 @@ import com.hp.hpl.jena.graph.Triple;
import com.hp.hpl.jena.query.Query;
import com.hp.hpl.jena.query.QueryFactory;
import com.hp.hpl.jena.query.QueryParseException;
import com.hp.hpl.jena.query.ResultSet;
import com.hp.hpl.jena.query.Syntax;
import com.hp.hpl.jena.rdf.model.Model;
import com.hp.hpl.jena.rdf.model.ModelChangedListener;
import com.hp.hpl.jena.rdf.model.ModelFactory;
import com.hp.hpl.jena.rdf.model.Statement;
import com.hp.hpl.jena.rdf.model.StmtIterator;
import com.hp.hpl.jena.sparql.resultset.XMLInput;
import com.hp.hpl.jena.vocabulary.RDF;
import edu.cornell.mannlib.vitro.webapp.rdfservice.ChangeListener;
import edu.cornell.mannlib.vitro.webapp.rdfservice.ChangeSet;
import edu.cornell.mannlib.vitro.webapp.rdfservice.ModelChange;
import edu.cornell.mannlib.vitro.webapp.rdfservice.ModelChange.Operation;
import edu.cornell.mannlib.vitro.webapp.rdfservice.impl.jena.ListeningGraph;
import edu.cornell.mannlib.vitro.webapp.rdfservice.RDFService;
import edu.cornell.mannlib.vitro.webapp.rdfservice.RDFServiceException;
import edu.cornell.mannlib.vitro.webapp.rdfservice.ResultSetConsumer;
import edu.cornell.mannlib.vitro.webapp.utils.logging.ToString;
public abstract class RDFServiceImpl implements RDFService {
@ -42,6 +44,7 @@ public abstract class RDFServiceImpl implements RDFService {
protected String defaultWriteGraphURI;
protected List<ChangeListener> registeredListeners = new CopyOnWriteArrayList<ChangeListener>();
protected List<ModelChangedListener> registeredJenaListeners = new CopyOnWriteArrayList<ModelChangedListener>();
@Override
public void newIndividual(String individualURI,
@ -87,7 +90,6 @@ public abstract class RDFServiceImpl implements RDFService {
@Override
public synchronized void registerListener(ChangeListener changeListener) throws RDFServiceException {
if (!registeredListeners.contains(changeListener)) {
registeredListeners.add(changeListener);
}
@ -97,41 +99,86 @@ public abstract class RDFServiceImpl implements RDFService {
public synchronized void unregisterListener(ChangeListener changeListener) throws RDFServiceException {
registeredListeners.remove(changeListener);
}
@Override
public synchronized void registerJenaModelChangedListener(ModelChangedListener changeListener) throws RDFServiceException {
if (!registeredJenaListeners.contains(changeListener)) {
registeredJenaListeners.add(changeListener);
}
}
@Override
public synchronized void unregisterJenaModelChangedListener(ModelChangedListener changeListener) throws RDFServiceException {
registeredJenaListeners.remove(changeListener);
}
public synchronized List<ChangeListener> getRegisteredListeners() {
return this.registeredListeners;
}
public synchronized List<ModelChangedListener> getRegisteredJenaModelChangedListeners() {
return this.registeredJenaListeners;
}
@Override
public ChangeSet manufactureChangeSet() {
return new ChangeSetImpl();
}
// I switched the following two methods back to public so they could be
// used by the ListeningGraph, which is common to both implementations.
// This could probably be improved later. BJL
protected void notifyListenersOfChanges(ChangeSet changeSet)
throws IOException {
if (registeredListeners.isEmpty() && registeredJenaListeners.isEmpty()) {
return;
}
for (ModelChange modelChange: changeSet.getModelChanges()) {
modelChange.getSerializedModel().reset();
notifyListeners(modelChange);
}
}
public void notifyListeners(Triple triple, ModelChange.Operation operation, String graphURI) {
protected void notifyListeners(ModelChange modelChange) {
Iterator<ChangeListener> iter = registeredListeners.iterator();
while (iter.hasNext()) {
ChangeListener listener = iter.next();
if (operation == ModelChange.Operation.ADD) {
listener.addedStatement(sparqlTriple(triple), graphURI);
} else {
listener.removedStatement(sparqlTriple(triple).toString(), graphURI);
}
listener.notifyModelChange(modelChange);
}
log.debug(registeredJenaListeners.size() + " registered Jena listeners");
if (registeredJenaListeners.isEmpty()) {
return;
}
Model tempModel = ModelFactory.createDefaultModel();
Iterator<ModelChangedListener> jenaIter = registeredJenaListeners.iterator();
while (jenaIter.hasNext()) {
ModelChangedListener listener = jenaIter.next();
log.debug("\t" + listener.getClass().getSimpleName());
tempModel.register(listener);
}
if (Operation.ADD.equals(modelChange.getOperation())) {
tempModel.read(modelChange.getSerializedModel(), null,
RDFServiceUtils.getSerializationFormatString(
modelChange.getSerializationFormat()));
} else if (Operation.REMOVE.equals(modelChange.getOperation())) {
tempModel.remove(RDFServiceUtils.parseModel(
modelChange.getSerializedModel(),
modelChange.getSerializationFormat()));
}
while (jenaIter.hasNext()) {
tempModel.unregister(jenaIter.next());
}
}
public void notifyListenersOfEvent(Object event) {
Iterator<ChangeListener> iter = registeredListeners.iterator();
while (iter.hasNext()) {
ChangeListener listener = iter.next();
// TODO what is the graphURI parameter for?
listener.notifyEvent(null, event);
}
Iterator<ModelChangedListener> jenaIter = registeredJenaListeners.iterator();
while (jenaIter.hasNext()) {
ModelChangedListener listener = jenaIter.next();
listener.notifyEvent(null, event);
}
}
protected boolean isPreconditionSatisfied(String query,

236
webapp/src/edu/cornell/mannlib/vitro/webapp/rdfservice/impl/jena/ListeningGraph.java
View file

@ -1,236 +0,0 @@
/* $This file is distributed under the terms of the license in /doc/license.txt$ */
package edu.cornell.mannlib.vitro.webapp.rdfservice.impl.jena;
import java.util.ArrayList;
import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import com.hp.hpl.jena.graph.BulkUpdateHandler;
import com.hp.hpl.jena.graph.Capabilities;
import com.hp.hpl.jena.graph.Graph;
import com.hp.hpl.jena.graph.GraphEventManager;
import com.hp.hpl.jena.graph.GraphStatisticsHandler;
import com.hp.hpl.jena.graph.Node;
import com.hp.hpl.jena.graph.TransactionHandler;
import com.hp.hpl.jena.graph.Triple;
import com.hp.hpl.jena.graph.TripleMatch;
import com.hp.hpl.jena.graph.impl.GraphWithPerform;
import com.hp.hpl.jena.graph.impl.SimpleBulkUpdateHandler;
import com.hp.hpl.jena.graph.impl.SimpleEventManager;
import com.hp.hpl.jena.shared.AddDeniedException;
import com.hp.hpl.jena.shared.DeleteDeniedException;
import com.hp.hpl.jena.shared.PrefixMapping;
import com.hp.hpl.jena.shared.impl.PrefixMappingImpl;
import com.hp.hpl.jena.util.iterator.ExtendedIterator;
import com.hp.hpl.jena.util.iterator.WrappedIterator;
import edu.cornell.mannlib.vitro.webapp.rdfservice.ModelChange;
import edu.cornell.mannlib.vitro.webapp.rdfservice.impl.RDFServiceImpl;
import edu.cornell.mannlib.vitro.webapp.utils.logging.ToString;
public class ListeningGraph implements GraphWithPerform {
private static final Log log = LogFactory.getLog(ListeningGraph.class);
private RDFServiceImpl rdfServiceImpl;
private String graphURI;
private BulkUpdateHandler bulkUpdateHandler;
private GraphEventManager eventManager;
private PrefixMapping prefixMapping = new PrefixMappingImpl();
public ListeningGraph(String graphURI, RDFServiceImpl rdfServiceImpl) {
this.graphURI = graphURI;
this.rdfServiceImpl = rdfServiceImpl;
}
@Override
public void add(Triple triple) throws AddDeniedException {
performAdd(triple);
}
@Override
public void performAdd(Triple triple) throws AddDeniedException {
if (log.isDebugEnabled()) {
log.debug("adding " + triple + " to " + graphURI);
}
this.rdfServiceImpl.notifyListeners(triple, ModelChange.Operation.ADD, graphURI);
}
@Override
public void delete(Triple triple) throws DeleteDeniedException {
performDelete(triple);
}
@Override
public void performDelete(Triple triple) throws DeleteDeniedException {
if (log.isDebugEnabled()) {
log.debug("deleting " + triple + " from " + graphURI);
}
this.rdfServiceImpl.notifyListeners(triple, ModelChange.Operation.REMOVE, graphURI);
}
@Override
public void close() {
// Nothing to close.
}
@Override
public boolean contains(Triple arg0) {
return contains(arg0.getSubject(), arg0.getPredicate(), arg0.getObject());
}
@Override
public boolean contains(Node subject, Node predicate, Node object) {
return false;
}
@Override
public boolean dependsOn(Graph arg0) {
return false; // who knows?
}
@Override
public ExtendedIterator<Triple> find(TripleMatch arg0) {
Triple t = arg0.asTriple();
return find(t.getSubject(), t.getPredicate(), t.getObject());
}
@Override
public ExtendedIterator<Triple> find(Node subject, Node predicate, Node object) {
List<Triple> triplist = new ArrayList<Triple>();
return WrappedIterator.create(triplist.iterator());
}
@Override
public void clear() {
for (Triple t: find(null, null, null).toList()) {
delete(t);
}
}
@Override
public void remove(Node subject, Node predicate, Node object) {
for (Triple t: find(subject, predicate, object).toList()) {
delete(t);
}
}
@Override
@Deprecated
public BulkUpdateHandler getBulkUpdateHandler() {
if (this.bulkUpdateHandler == null) {
this.bulkUpdateHandler = new SimpleBulkUpdateHandler(this);
}
return this.bulkUpdateHandler;
}
@Override
public Capabilities getCapabilities() {
return capabilities;
}
@Override
public GraphEventManager getEventManager() {
if (eventManager == null) {
eventManager = new SimpleEventManager(this);
}
return eventManager;
}
@Override
public PrefixMapping getPrefixMapping() {
return prefixMapping;
}
@Override
public GraphStatisticsHandler getStatisticsHandler() {
return null;
}
@Override
public TransactionHandler getTransactionHandler() {
return null;
}
@Override
public boolean isClosed() {
return false;
}
@Override
public boolean isEmpty() {
return !contains(null, null, null);
}
@Override
public boolean isIsomorphicWith(Graph arg0) {
throw new UnsupportedOperationException("isIsomorphicWith() not supported " +
"by SPARQL graphs");
}
@Override
public int size() {
int size = find(null, null, null).toList().size();
return size;
}
private final static Capabilities capabilities = new Capabilities() {
@Override
public boolean addAllowed() {
return false;
}
@Override
public boolean addAllowed(boolean everyTriple) {
return false;
}
@Override
public boolean canBeEmpty() {
return true;
}
@Override
public boolean deleteAllowed() {
return false;
}
@Override
public boolean deleteAllowed(boolean everyTriple) {
return false;
}
@Override
public boolean findContractSafe() {
return true;
}
@Override
public boolean handlesLiteralTyping() {
return true;
}
@Override
public boolean iteratorRemoveAllowed() {
return false;
}
@Override
public boolean sizeAccurate() {
return true;
}
};
@Override
public String toString() {
return "ListeningGraph["+ToString.hashHex(this)
+ ", " + rdfServiceImpl
+ ", " + ToString.modelName(graphURI) + "]";
}
}

10
webapp/src/edu/cornell/mannlib/vitro/webapp/rdfservice/impl/jena/RDFServiceJena.java
View file

@ -91,16 +91,6 @@ public abstract class RDFServiceJena extends RDFServiceImpl implements RDFServic
}
}
protected void notifyListenersOfChanges(ChangeSet changeSet)
throws IOException {
for (ModelChange modelChange: changeSet.getModelChanges()) {
modelChange.getSerializedModel().reset();
Model model = ModelFactory.createModelForGraph(
new ListeningGraph(modelChange.getGraphURI(), this));
operateOnModel(model, modelChange, null);
}
}
protected void notifyListenersOfPostChangeEvents(ChangeSet changeSet) {
for (Object o : changeSet.getPostChangeEvents()) {
this.notifyListenersOfEvent(o);

17
webapp/src/edu/cornell/mannlib/vitro/webapp/rdfservice/impl/jena/model/RDFServiceModel.java
View file

@ -102,14 +102,15 @@ public class RDFServiceModel extends RDFServiceJena implements RDFService {
}
// notify listeners of triple changes
csIt = changeSet.getModelChanges().iterator();
while (csIt.hasNext()) {
ModelChange modelChange = csIt.next();
modelChange.getSerializedModel().reset();
Model model = ModelFactory.createModelForGraph(
new ListeningGraph(modelChange.getGraphURI(), this));
operateOnModel(model, modelChange, null);
}
notifyListenersOfChanges(changeSet);
// csIt = changeSet.getModelChanges().iterator();
// while (csIt.hasNext()) {
// ModelChange modelChange = csIt.next();
// modelChange.getSerializedModel().reset();
// Model model = ModelFactory.createModelForGraph(
// new ListeningGraph(modelChange.getGraphURI(), this));
// operateOnModel(model, modelChange, null);
// }
for (Object o : changeSet.getPostChangeEvents()) {
this.notifyListenersOfEvent(o);

17
webapp/src/edu/cornell/mannlib/vitro/webapp/rdfservice/impl/jena/sdb/RDFServiceFactorySDB.java
View file

@ -7,6 +7,7 @@ import javax.sql.DataSource;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import com.hp.hpl.jena.rdf.model.ModelChangedListener;
import com.hp.hpl.jena.sdb.StoreDesc;
import edu.cornell.mannlib.vitro.webapp.rdfservice.ChangeListener;
@ -41,6 +42,10 @@ public class RDFServiceFactorySDB implements RDFServiceFactory {
.getRegisteredListeners() ) {
rdfService.registerListener(cl);
}
for (ModelChangedListener cl : ((RDFServiceSDB) longTermRDFService)
.getRegisteredJenaModelChangedListeners() ) {
rdfService.registerJenaModelChangedListener(cl);
}
return rdfService;
} catch (Exception e) {
log.error(e,e);
@ -59,5 +64,17 @@ public class RDFServiceFactorySDB implements RDFServiceFactory {
throws RDFServiceException {
this.longTermRDFService.unregisterListener(changeListener);
}
@Override
public void registerJenaModelChangedListener(ModelChangedListener changeListener)
throws RDFServiceException {
this.longTermRDFService.registerJenaModelChangedListener(changeListener);
}
@Override
public void unregisterJenaModelChangedListener(ModelChangedListener changeListener)
throws RDFServiceException {
this.longTermRDFService.unregisterJenaModelChangedListener(changeListener);
}
}

14
webapp/src/edu/cornell/mannlib/vitro/webapp/rdfservice/impl/logging/LoggingRDFService.java
View file

@ -7,6 +7,8 @@ import java.io.OutputStream;
import java.util.List;
import com.hp.hpl.jena.rdf.model.Model;
import com.hp.hpl.jena.rdf.model.ModelChangedListener;
import edu.cornell.mannlib.vitro.webapp.rdfservice.ChangeListener;
import edu.cornell.mannlib.vitro.webapp.rdfservice.ChangeSet;
import edu.cornell.mannlib.vitro.webapp.rdfservice.RDFService;
@ -154,6 +156,18 @@ public class LoggingRDFService implements RDFService {
innerService.unregisterListener(changeListener);
}
@Override
public void registerJenaModelChangedListener(ModelChangedListener changeListener)
throws RDFServiceException {
innerService.registerJenaModelChangedListener(changeListener);
}
@Override
public void unregisterJenaModelChangedListener(ModelChangedListener changeListener)
throws RDFServiceException {
innerService.unregisterJenaModelChangedListener(changeListener);
}
@Override
public ChangeSet manufactureChangeSet() {
return innerService.manufactureChangeSet();

14
webapp/src/edu/cornell/mannlib/vitro/webapp/rdfservice/impl/logging/LoggingRDFServiceFactory.java
View file

@ -2,6 +2,8 @@
package edu.cornell.mannlib.vitro.webapp.rdfservice.impl.logging;
import com.hp.hpl.jena.rdf.model.ModelChangedListener;
import edu.cornell.mannlib.vitro.webapp.rdfservice.ChangeListener;
import edu.cornell.mannlib.vitro.webapp.rdfservice.RDFService;
import edu.cornell.mannlib.vitro.webapp.rdfservice.RDFServiceException;
@ -40,6 +42,18 @@ public class LoggingRDFServiceFactory implements RDFServiceFactory {
factory.unregisterListener(changeListener);
}
@Override
public void registerJenaModelChangedListener(ModelChangedListener changeListener)
throws RDFServiceException {
factory.registerJenaModelChangedListener(changeListener);
}
@Override
public void unregisterJenaModelChangedListener(ModelChangedListener changeListener)
throws RDFServiceException {
factory.unregisterJenaModelChangedListener(changeListener);
}
@Override
public String toString() {
return "LoggingRDFServiceFactory[factory=" + factory + "]";

140
webapp/src/edu/cornell/mannlib/vitro/webapp/rdfservice/impl/sparql/RDFServiceSparql.java
View file

@ -192,38 +192,40 @@ public class RDFServiceSparql extends RDFServiceImpl implements RDFService {
modelChange.getSerializedModel().mark(Integer.MAX_VALUE);
performChange(modelChange);
}
notifyListenersOfChanges(changeSet);
// notify listeners of triple changes
csIt = changeSet.getModelChanges().iterator();
while (csIt.hasNext()) {
ModelChange modelChange = csIt.next();
modelChange.getSerializedModel().reset();
Model model = ModelFactory.createModelForGraph(
new ListeningGraph(modelChange.getGraphURI(), this));
long start = System.currentTimeMillis();
if (modelChange.getOperation() == ModelChange.Operation.ADD) {
Model temp = ModelFactory.createDefaultModel();
temp.read(modelChange.getSerializedModel(), null,
getSerializationFormatString(
modelChange.getSerializationFormat()));
StmtIterator sit = temp.listStatements();
while(sit.hasNext()) {
Triple triple = sit.nextStatement().asTriple();
this.notifyListeners(triple, ModelChange.Operation.ADD, modelChange.getGraphURI());
}
//model.add(temp);
} else if (modelChange.getOperation() == ModelChange.Operation.REMOVE){
Model temp = ModelFactory.createDefaultModel();
temp.read(modelChange.getSerializedModel(), null,
getSerializationFormatString(
modelChange.getSerializationFormat()));
model.remove(temp);
} else {
log.error("Unsupported model change type " +
modelChange.getOperation().getClass().getName());
}
log.info((System.currentTimeMillis() - start) + " ms to notify " + this.getRegisteredListeners().size() + " listeners");
}
// csIt = changeSet.getModelChanges().iterator();
// while (csIt.hasNext()) {
// ModelChange modelChange = csIt.next();
// modelChange.getSerializedModel().reset();
// Model model = ModelFactory.createModelForGraph(
// new ListeningGraph(modelChange.getGraphURI(), this));
// long start = System.currentTimeMillis();
// if (modelChange.getOperation() == ModelChange.Operation.ADD) {
// Model temp = ModelFactory.createDefaultModel();
// temp.read(modelChange.getSerializedModel(), null,
// getSerializationFormatString(
// modelChange.getSerializationFormat()));
// StmtIterator sit = temp.listStatements();
// while(sit.hasNext()) {
// Triple triple = sit.nextStatement().asTriple();
// this.notifyListeners(triple, ModelChange.Operation.ADD, modelChange.getGraphURI());
// }
// //model.add(temp);
// } else if (modelChange.getOperation() == ModelChange.Operation.REMOVE){
// Model temp = ModelFactory.createDefaultModel();
// temp.read(modelChange.getSerializedModel(), null,
// getSerializationFormatString(
// modelChange.getSerializationFormat()));
// model.remove(temp);
// } else {
// log.error("Unsupported model change type " +
// modelChange.getOperation().getClass().getName());
// }
// log.info((System.currentTimeMillis() - start) + " ms to notify " + this.getRegisteredListeners().size() + " listeners");
// }
for (Object o : changeSet.getPostChangeEvents()) {
this.notifyListenersOfEvent(o);
@ -606,45 +608,45 @@ public class RDFServiceSparql extends RDFServiceImpl implements RDFService {
}
}
protected void addTriple(Triple t, String graphURI) throws RDFServiceException {
try {
StringBuffer updateString = new StringBuffer();
updateString.append("INSERT DATA { ");
updateString.append((graphURI != null) ? "GRAPH <" + graphURI + "> { " : "");
updateString.append(sparqlNodeUpdate(t.getSubject(), ""));
updateString.append(" ");
updateString.append(sparqlNodeUpdate(t.getPredicate(), ""));
updateString.append(" ");
updateString.append(sparqlNodeUpdate(t.getObject(), ""));
updateString.append(" }");
updateString.append((graphURI != null) ? " } " : "");
executeUpdate(updateString.toString());
notifyListeners(t, ModelChange.Operation.ADD, graphURI);
} finally {
rebuildGraphURICache = true;
}
}
protected void removeTriple(Triple t, String graphURI) throws RDFServiceException {
try {
StringBuffer updateString = new StringBuffer();
updateString.append("DELETE DATA { ");
updateString.append((graphURI != null) ? "GRAPH <" + graphURI + "> { " : "");
updateString.append(sparqlNodeUpdate(t.getSubject(), ""));
updateString.append(" ");
updateString.append(sparqlNodeUpdate(t.getPredicate(), ""));
updateString.append(" ");
updateString.append(sparqlNodeUpdate(t.getObject(), ""));
updateString.append(" }");
updateString.append((graphURI != null) ? " } " : "");
executeUpdate(updateString.toString());
notifyListeners(t, ModelChange.Operation.REMOVE, graphURI);
} finally {
rebuildGraphURICache = true;
}
}
// protected void addTriple(Triple t, String graphURI) throws RDFServiceException {
// try {
// StringBuffer updateString = new StringBuffer();
// updateString.append("INSERT DATA { ");
// updateString.append((graphURI != null) ? "GRAPH <" + graphURI + "> { " : "");
// updateString.append(sparqlNodeUpdate(t.getSubject(), ""));
// updateString.append(" ");
// updateString.append(sparqlNodeUpdate(t.getPredicate(), ""));
// updateString.append(" ");
// updateString.append(sparqlNodeUpdate(t.getObject(), ""));
// updateString.append(" }");
// updateString.append((graphURI != null) ? " } " : "");
//
// executeUpdate(updateString.toString());
// notifyListeners(t, ModelChange.Operation.ADD, graphURI);
// } finally {
// rebuildGraphURICache = true;
// }
// }
//
// protected void removeTriple(Triple t, String graphURI) throws RDFServiceException {
// try {
// StringBuffer updateString = new StringBuffer();
// updateString.append("DELETE DATA { ");
// updateString.append((graphURI != null) ? "GRAPH <" + graphURI + "> { " : "");
// updateString.append(sparqlNodeUpdate(t.getSubject(), ""));
// updateString.append(" ");
// updateString.append(sparqlNodeUpdate(t.getPredicate(), ""));
// updateString.append(" ");
// updateString.append(sparqlNodeUpdate(t.getObject(), ""));
// updateString.append(" }");
// updateString.append((graphURI != null) ? " } " : "");
//
// executeUpdate(updateString.toString());
// notifyListeners(t, ModelChange.Operation.REMOVE, graphURI);
// } finally {
// rebuildGraphURICache = true;
// }
// }
@Override
protected boolean isPreconditionSatisfied(String query,

83
webapp/src/edu/cornell/mannlib/vitro/webapp/reasoner/ABoxRecomputer.java
View file

@ -5,13 +5,13 @@ package edu.cornell.mannlib.vitro.webapp.reasoner;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.InputStream;
import java.io.StringWriter;
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Queue;
@ -93,15 +93,10 @@ public class ABoxRecomputer {
return recomputing;
}
public void recompute() {
recompute(null);
}
/**
* Recompute inferences for specified collection of individual URIs,
* or all URIs if parameter is null
* Recompute all individuals
*/
public void recompute(Queue<String> individualURIs) {
public void recompute() {
synchronized (lock1) {
if (recomputing) {
return;
@ -109,51 +104,70 @@ public class ABoxRecomputer {
recomputing = true;
}
}
boolean fullRecompute = (individualURIs == null);
boolean sizableRecompute = (!fullRecompute && individualURIs.size() > 2);
try {
if(fullRecompute || sizableRecompute) { // if doing a full rebuild
if (searchIndexer != null) {
searchIndexer.pause();
// Register now that we want to rebuild the index when we unpause
// This allows the indexer to optimize behaviour whilst paused
if(fullRecompute) {
searchIndexer.rebuildIndex();
}
}
if (searchIndexer != null) {
searchIndexer.pause();
// Register now that we want to rebuild the index when we unpause
// This allows the indexer to optimize behaviour whilst paused
searchIndexer.rebuildIndex();
}
log.info("Recomputing ABox inferences.");
log.info("Finding individuals in ABox.");
Queue<String>individualURIs = this.getAllIndividualURIs();
log.info("Recomputing inferences for " + individualURIs.size() + " individuals");
// Create a type cache for this execution and pass it to the recompute function
// Ensures that caches are only valid for the length of one recompute
recomputeABox(individualURIs, new TypeCaches());
recomputeIndividuals(individualURIs, new TypeCaches());
log.info("Finished recomputing inferences");
} finally {
if ((fullRecompute || sizableRecompute) && searchIndexer != null) {
if(searchIndexer != null) {
searchIndexer.unpause();
}
synchronized (lock1) {
recomputing = false;
recomputing = false;
}
}
}
/**
* Recompute inferences for specified collection of individual URIs,
* or all URIs if parameter is null
*/
public void recompute(Queue<String> individualURIs) {
boolean sizableRecompute = (individualURIs.size() > 20);
try {
if(sizableRecompute && searchIndexer != null) {
searchIndexer.pause();
}
recomputeIndividuals(individualURIs);
} finally {
if (sizableRecompute && searchIndexer != null) {
searchIndexer.unpause();
}
}
}
/*
* Recompute the ABox inference graph for the specified collection of
* individual URIs, or all individuals if the collection is null.
* individual URIs
*/
protected void recomputeABox(Queue<String> individuals, TypeCaches caches) {
boolean printLog = false;
private void recomputeIndividuals(Queue<String> individuals) {
recomputeIndividuals(individuals, new TypeCaches());
}
/*
* Recompute the ABox inference graph for the specified collection of
* individual URIs
*/
protected void recomputeIndividuals(Queue<String> individuals, TypeCaches caches) {
if (individuals == null) {
printLog = true;
log.info("Recomputing ABox inferences.");
log.info("Finding individuals in ABox.");
individuals = this.getAllIndividualURIs();
log.info("Recomputing inferences for " + individuals.size() + " individuals");
return;
}
long start = System.currentTimeMillis();
int numInds = 0;
Model rebuildModel = ModelFactory.createDefaultModel();
Model additionalInferences = ModelFactory.createDefaultModel();
List<String> individualsInBatch = new ArrayList<String>();
//Iterator<String> individualIt = individuals.iterator();
while (!individuals.isEmpty()) {
String individualURI = individuals.poll();
try {
@ -175,7 +189,7 @@ public class ABoxRecomputer {
log.info((System.currentTimeMillis() - start) / numInds + " ms per individual");
}
if (stopRequested) {
log.info("a stopRequested signal was received during recomputeABox. Halting Processing.");
log.info("a stopRequested signal was received during recomputeIndividuals. Halting Processing.");
return;
}
} catch (Exception e) {
@ -193,16 +207,13 @@ public class ABoxRecomputer {
log.error("Unable to write additional inferences from reasoner plugins", e);
}
}
if (printLog) {
log.info("Finished recomputing inferences");
}
}
private static final boolean RUN_PLUGINS = true;
private static final boolean SKIP_PLUGINS = !RUN_PLUGINS;
private Model recomputeIndividual(String individualURI,
Model rebuildModel, TypeCaches caches, Queue<String> individualQueue)
Model rebuildModel, TypeCaches caches, Collection<String> individualQueue)
throws RDFServiceException {
long start = System.currentTimeMillis();
Model assertions = getAssertions(individualURI);

126
webapp/src/edu/cornell/mannlib/vitro/webapp/reasoner/SimpleReasoner.java
View file

@ -25,9 +25,12 @@ import com.hp.hpl.jena.query.DatasetFactory;
import com.hp.hpl.jena.rdf.listeners.StatementListener;
import com.hp.hpl.jena.rdf.model.Literal;
import com.hp.hpl.jena.rdf.model.Model;
import com.hp.hpl.jena.rdf.model.ModelChangedListener;
import com.hp.hpl.jena.rdf.model.ModelFactory;
import com.hp.hpl.jena.rdf.model.NodeIterator;
import com.hp.hpl.jena.rdf.model.Property;
import com.hp.hpl.jena.rdf.model.RDFNode;
import com.hp.hpl.jena.rdf.model.ResIterator;
import com.hp.hpl.jena.rdf.model.Resource;
import com.hp.hpl.jena.rdf.model.ResourceFactory;
import com.hp.hpl.jena.rdf.model.Statement;
@ -38,15 +41,16 @@ import com.hp.hpl.jena.vocabulary.OWL;
import com.hp.hpl.jena.vocabulary.RDF;
import com.hp.hpl.jena.vocabulary.RDFS;
import edu.cornell.mannlib.vitro.webapp.dao.jena.ABoxJenaChangeListener;
import edu.cornell.mannlib.vitro.webapp.dao.jena.DifferenceGraph;
import edu.cornell.mannlib.vitro.webapp.dao.jena.RDFServiceGraph;
import edu.cornell.mannlib.vitro.webapp.dao.jena.event.BulkUpdateEvent;
import edu.cornell.mannlib.vitro.webapp.modelaccess.ModelNames;
import edu.cornell.mannlib.vitro.webapp.modules.searchIndexer.SearchIndexer;
import edu.cornell.mannlib.vitro.webapp.rdfservice.ChangeListener;
import edu.cornell.mannlib.vitro.webapp.rdfservice.ModelChange;
import edu.cornell.mannlib.vitro.webapp.rdfservice.RDFService;
import edu.cornell.mannlib.vitro.webapp.rdfservice.RDFServiceException;
import edu.cornell.mannlib.vitro.webapp.rdfservice.adapters.VitroModelFactory;
import edu.cornell.mannlib.vitro.webapp.rdfservice.impl.RDFServiceUtils;
import edu.cornell.mannlib.vitro.webapp.rdfservice.impl.jena.model.RDFServiceModel;
/**
@ -56,7 +60,8 @@ import edu.cornell.mannlib.vitro.webapp.rdfservice.impl.jena.model.RDFServiceMod
* @author sjm222
*/
public class SimpleReasoner extends StatementListener {
public class SimpleReasoner extends StatementListener
implements ModelChangedListener, ChangeListener {
private static final Log log = LogFactory.getLog(SimpleReasoner.class);
@ -74,9 +79,6 @@ public class SimpleReasoner extends StatementListener {
VitroModelFactory.createOntologyModel())
.createAnnotationProperty(mostSpecificTypePropertyURI);
private Queue<String> individualURIqueue = new IndividualURIQueue<String>();
private boolean accumulateChanges = false;
// Recomputer
private ABoxRecomputer recomputer = null;
private List<ReasonerPlugin> pluginList = new CopyOnWriteArrayList<ReasonerPlugin>();
@ -121,7 +123,7 @@ public class SimpleReasoner extends StatementListener {
aboxModel.register(this);
} else {
try {
rdfService.registerListener(new ABoxJenaChangeListener(this));
rdfService.registerListener(this);
} catch (RDFServiceException e) {
throw new RuntimeException("Unable to register change listener", e);
}
@ -169,30 +171,65 @@ public class SimpleReasoner extends StatementListener {
}
private void listenToStatement(Statement stmt) {
if(stmt.getSubject().isURIResource()) {
individualURIqueue.add(stmt.getSubject().getURI());
}
if(stmt.getObject().isURIResource() && !(RDF.type.equals(stmt.getPredicate()))) {
individualURIqueue.add(stmt.getObject().asResource().getURI());
}
if(!accumulateChanges) {
recomputeIndividuals();
}
Queue<String> individualURIs = new IndividualURIQueue<String>();
listenToStatement(stmt, individualURIs);
}
private void recomputeIndividuals() {
if(recomputer.isRecomputing()) {
return;
private void listenToStatement(Statement stmt, Queue<String> individualURIs) {
if(stmt.getSubject().isURIResource()) {
individualURIs.add(stmt.getSubject().getURI());
}
if(stmt.getObject().isURIResource() && !(RDF.type.equals(stmt.getPredicate()))) {
individualURIs.add(stmt.getObject().asResource().getURI());
}
recomputeIndividuals(individualURIs);
}
private void recomputeIndividuals(Queue<String> individualURIs) {
long start = System.currentTimeMillis();
int size = individualURIqueue.size();
recomputer.recompute(individualURIqueue);
int size = individualURIs.size();
recomputer.recompute(individualURIs);
if(size > 2) {
log.info((System.currentTimeMillis() - start) + " ms to recompute "
+ size + " individuals");
}
}
boolean isABoxInferenceGraph(String graphURI) {
return ModelNames.ABOX_INFERENCES.equals(graphURI);
}
boolean isTBoxGraph(String graphURI) {
return ( ModelNames.TBOX_ASSERTIONS.equals(graphURI)
|| ModelNames.TBOX_INFERENCES.equals(graphURI)
|| (graphURI != null && graphURI.contains("tbox")) );
}
public void notifyModelChange(ModelChange modelChange) {
if(isABoxInferenceGraph(modelChange.getGraphURI())
|| isTBoxGraph(modelChange.getGraphURI())) {
return;
}
Queue<String> individualURIs = new IndividualURIQueue<String>();
Model m = RDFServiceUtils.parseModel(modelChange.getSerializedModel(),
modelChange.getSerializationFormat());
ResIterator subjIt = m.listSubjects();
while(subjIt.hasNext()) {
Resource subj = subjIt.next();
if(subj.isURIResource()) {
individualURIs.add(subj.getURI());
}
}
NodeIterator objIt = m.listObjects();
while(objIt.hasNext()) {
RDFNode obj = objIt.next();
if(obj.isURIResource()) {
individualURIs.add(obj.asResource().getURI());
}
}
recomputeIndividuals(individualURIs);
}
/*
* Performs incremental ABox reasoning based
* on the addition of a new statement
@ -201,7 +238,7 @@ public class SimpleReasoner extends StatementListener {
@Override
public void addedStatement(Statement stmt) {
doPlugins(ModelUpdate.Operation.ADD,stmt);
listenToStatement(stmt);;
listenToStatement(stmt);
}
/*
@ -212,17 +249,18 @@ public class SimpleReasoner extends StatementListener {
@Override
public void removedStatement(Statement stmt) {
doPlugins(ModelUpdate.Operation.RETRACT,stmt);
Queue<String> individualURIs = new IndividualURIQueue<String>();
if(doSameAs && OWL.sameAs.equals(stmt.getPredicate())) {
if (stmt.getSubject().isURIResource()) {
individualURIqueue.addAll(this.recomputer.getSameAsIndividuals(
individualURIs.addAll(this.recomputer.getSameAsIndividuals(
stmt.getSubject().getURI()));
}
if (stmt.getObject().isURIResource()) {
individualURIqueue.addAll(this.recomputer.getSameAsIndividuals(
individualURIs.addAll(this.recomputer.getSameAsIndividuals(
stmt.getObject().asResource().getURI()));
}
}
listenToStatement(stmt);
listenToStatement(stmt, individualURIs);
}
/**
@ -1119,33 +1157,37 @@ public class SimpleReasoner extends StatementListener {
}
}
// DeltaComputer
/**
* Asynchronous reasoning mode (DeltaComputer) was used in the case of batch removals.
* Asynchronous reasoning mode (DeltaComputer) no longer used in the case of batch removals.
*/
public boolean isABoxReasoningAsynchronous() {
return false;
}
@Override
public void notifyEvent(String string, Object event) {
// don't care
}
@Override
public void notifyEvent(Model model, Object event) {
if (event instanceof BulkUpdateEvent) {
handleBulkUpdateEvent(event);
}
// if (event instanceof BulkUpdateEvent) {
// handleBulkUpdateEvent(event);
// }
}
public synchronized void handleBulkUpdateEvent(Object event) {
if (event instanceof BulkUpdateEvent) {
if (((BulkUpdateEvent) event).getBegin()) {
this.accumulateChanges = true;
} else {
log.debug("received a bulk update end event");
this.accumulateChanges = false;
recomputeIndividuals();
}
}
}
// public synchronized void handleBulkUpdateEvent(Object event) {
//
// if (event instanceof BulkUpdateEvent) {
// if (((BulkUpdateEvent) event).getBegin()) {
// this.accumulateChanges = true;
// } else {
// log.debug("received a bulk update end event");
// this.accumulateChanges = false;
// recomputeIndividuals();
// }
// }
// }
/**
* Utility method for logging

77
webapp/src/edu/cornell/mannlib/vitro/webapp/searchindex/IndexingChangeListener.java
View file

@ -2,6 +2,9 @@
package edu.cornell.mannlib.vitro.webapp.searchindex;
import static edu.cornell.mannlib.vitro.webapp.modules.searchIndexer.SearchIndexer.Event.Type.REBUILD_REQUESTED;
import static edu.cornell.mannlib.vitro.webapp.modules.searchIndexer.SearchIndexer.Event.Type.START_REBUILD;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
@ -16,18 +19,17 @@ import org.apache.jena.riot.tokens.Tokenizer;
import org.apache.jena.riot.tokens.TokenizerFactory;
import com.hp.hpl.jena.graph.Triple;
import com.hp.hpl.jena.rdf.listeners.StatementListener;
import com.hp.hpl.jena.rdf.model.Model;
import com.hp.hpl.jena.rdf.model.ModelChangedListener;
import com.hp.hpl.jena.rdf.model.ModelFactory;
import com.hp.hpl.jena.rdf.model.Statement;
import edu.cornell.mannlib.vitro.webapp.dao.jena.event.EditEvent;
import edu.cornell.mannlib.vitro.webapp.modules.searchIndexer.SearchIndexer;
import edu.cornell.mannlib.vitro.webapp.modules.searchIndexer.SearchIndexer.Event;
import edu.cornell.mannlib.vitro.webapp.rdfservice.ChangeListener;
import edu.cornell.mannlib.vitro.webapp.utils.threads.VitroBackgroundThread;
import static edu.cornell.mannlib.vitro.webapp.modules.searchIndexer.SearchIndexer.Event.Type.*;
/**
* When a change is heard, wait for an interval to see if more changes come in.
* When changes stop coming in for a specified interval, send what has
@ -52,8 +54,8 @@ import static edu.cornell.mannlib.vitro.webapp.modules.searchIndexer.SearchIndex
* which is semantically equivalent to the original, and add that to the list
* instead. The original statement is released.
*/
public class IndexingChangeListener implements ChangeListener,
SearchIndexer.Listener {
public class IndexingChangeListener extends StatementListener
implements ModelChangedListener, SearchIndexer.Listener {
private static final Log log = LogFactory
.getLog(IndexingChangeListener.class);
@ -105,16 +107,16 @@ public class IndexingChangeListener implements ChangeListener,
}
@Override
public void addedStatement(String serializedTriple, String graphURI) {
public void addedStatement(Statement stmt) {
if (!rebuildScheduled) {
noteChange(parseTriple(serializedTriple));
noteChange(stmt);
}
}
@Override
public void removedStatement(String serializedTriple, String graphURI) {
public void removedStatement(Statement stmt) {
if (!rebuildScheduled) {
noteChange(parseTriple(serializedTriple));
noteChange(stmt);
}
}
@ -122,7 +124,7 @@ public class IndexingChangeListener implements ChangeListener,
* We only care about events that signal the end of an edit operation.
*/
@Override
public void notifyEvent(String graphURI, Object event) {
public void notifyEvent(Model model, Object event) {
if ((event instanceof EditEvent)) {
EditEvent editEvent = (EditEvent) event;
if (!editEvent.getBegin()) { // editEvent is the end of an edit
@ -135,34 +137,33 @@ public class IndexingChangeListener implements ChangeListener,
}
}
// TODO avoid duplication with JenaChangeListener
private Statement parseTriple(String serializedTriple) {
try {
// Use RiotReader to parse a Triple
// NB A Triple can be serialized correctly with: FmtUtils.stringForTriple(triple, PrefixMapping.Factory.create()) + " .";'
Tokenizer tokenizer = TokenizerFactory.makeTokenizerString(serializedTriple);
Iterator<Triple> it = RiotReader.createParserNTriples(tokenizer, null);
if (it.hasNext()) {
Triple triple = it.next();
if (it.hasNext()) {
log.warn("More than one triple parsed from change event: '" + serializedTriple + "'");
}
// Use the retained defaultModel instance to convert the Triple to a Statement
// This does not add the Statement to the Model, so the Statement can be disposed when unused
// And whilst the Model is attached to the Statement, using a single instance means only one Model
// is created and attached to all of the Statements created by this instance
return defaultModel.asStatement(triple);
} else {
throw new RuntimeException("no triple parsed from change event: '" + serializedTriple + "'");
}
} catch (RuntimeException riot) {
log.error("Failed to parse triple " + serializedTriple, riot);
throw riot;
}
}
// private Statement parseTriple(String serializedTriple) {
// try {
// // Use RiotReader to parse a Triple
// // NB A Triple can be serialized correctly with: FmtUtils.stringForTriple(triple, PrefixMapping.Factory.create()) + " .";'
// Tokenizer tokenizer = TokenizerFactory.makeTokenizerString(serializedTriple);
// Iterator<Triple> it = RiotReader.createParserNTriples(tokenizer, null);
//
// if (it.hasNext()) {
// Triple triple = it.next();
//
// if (it.hasNext()) {
// log.warn("More than one triple parsed from change event: '" + serializedTriple + "'");
// }
//
// // Use the retained defaultModel instance to convert the Triple to a Statement
// // This does not add the Statement to the Model, so the Statement can be disposed when unused
// // And whilst the Model is attached to the Statement, using a single instance means only one Model
// // is created and attached to all of the Statements created by this instance
// return defaultModel.asStatement(triple);
// } else {
// throw new RuntimeException("no triple parsed from change event: '" + serializedTriple + "'");
// }
// } catch (RuntimeException riot) {
// log.error("Failed to parse triple " + serializedTriple, riot);
// throw riot;
// }
// }
// ----------------------------------------------------------------------
// helper classes

4
webapp/src/edu/cornell/mannlib/vitro/webapp/searchindex/SearchIndexerSetup.java
View file

@ -55,7 +55,7 @@ public class SearchIndexerSetup implements ServletContextListener {
listener = new IndexingChangeListener(searchIndexer);
listenerWrapper = new DeveloperDisabledChangeListener(listener,
Key.SEARCH_INDEX_SUPPRESS_MODEL_CHANGE_LISTENER);
RDFServiceUtils.getRDFServiceFactory(ctx).registerListener(
RDFServiceUtils.getRDFServiceFactory(ctx).registerJenaModelChangedListener(
listenerWrapper);
this.history = new IndexHistory();
@ -78,7 +78,7 @@ public class SearchIndexerSetup implements ServletContextListener {
searchIndexer.removeListener(this.history);
try {
RDFServiceUtils.getRDFServiceFactory(ctx).unregisterListener(
RDFServiceUtils.getRDFServiceFactory(ctx).unregisterJenaModelChangedListener(
listenerWrapper);
} catch (RDFServiceException e) {
log.warn("Failed to unregister the indexing listener.");

25
webapp/src/edu/cornell/mannlib/vitro/webapp/utils/developer/listeners/DeveloperDisabledChangeListener.java
View file

@ -2,7 +2,11 @@
package edu.cornell.mannlib.vitro.webapp.utils.developer.listeners;
import edu.cornell.mannlib.vitro.webapp.rdfservice.ChangeListener;
import com.hp.hpl.jena.rdf.listeners.StatementListener;
import com.hp.hpl.jena.rdf.model.Model;
import com.hp.hpl.jena.rdf.model.ModelChangedListener;
import com.hp.hpl.jena.rdf.model.Statement;
import edu.cornell.mannlib.vitro.webapp.utils.developer.DeveloperSettings;
import edu.cornell.mannlib.vitro.webapp.utils.developer.Key;
@ -11,11 +15,12 @@ import edu.cornell.mannlib.vitro.webapp.utils.developer.Key;
*
* Set the flag and this becomes opaque, passing no events through.
*/
public class DeveloperDisabledChangeListener implements ChangeListener {
private final ChangeListener inner;
public class DeveloperDisabledChangeListener extends StatementListener
implements ModelChangedListener {
private final ModelChangedListener inner;
private final Key disablingKey;
public DeveloperDisabledChangeListener(ChangeListener inner,
public DeveloperDisabledChangeListener(ModelChangedListener inner,
Key disablingKey) {
this.inner = inner;
this.disablingKey = disablingKey;
@ -30,23 +35,23 @@ public class DeveloperDisabledChangeListener implements ChangeListener {
// ----------------------------------------------------------------------
@Override
public void addedStatement(String serializedTriple, String graphURI) {
public void addedStatement(Statement stmt) {
if (isEnabled()) {
inner.addedStatement(serializedTriple, graphURI);
inner.addedStatement(stmt);
}
}
@Override
public void removedStatement(String serializedTriple, String graphURI) {
public void removedStatement(Statement stmt) {
if (isEnabled()) {
inner.removedStatement(serializedTriple, graphURI);
inner.removedStatement(stmt);
}
}
@Override
public void notifyEvent(String graphURI, Object event) {
public void notifyEvent(Model model, Object event) {
if (isEnabled()) {
inner.notifyEvent(graphURI, event);
inner.notifyEvent(model, event);
}
}