<commit_before>/* $This file is distributed under the terms of the license in LICENSE$ */

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 org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import org.apache.jena.atlas.io.StringWriterI;
import org.apache.jena.graph.Node;
import org.apache.jena.graph.NodeFactory;
import org.apache.jena.graph.Triple;
import org.apache.jena.query.Query;
import org.apache.jena.query.QueryFactory;
import org.apache.jena.query.QueryParseException;
import org.apache.jena.query.QuerySolution;
import org.apache.jena.query.Syntax;
import org.apache.jena.rdf.model.Literal;
import org.apache.jena.rdf.model.Model;
import org.apache.jena.rdf.model.ModelChangedListener;
import org.apache.jena.rdf.model.ModelFactory;
import org.apache.jena.rdf.model.RDFNode;
import org.apache.jena.rdf.model.Statement;
import org.apache.jena.rdf.model.StmtIterator;
import org.apache.jena.riot.out.NodeFormatter;
import org.apache.jena.riot.out.NodeFormatterTTL;
import org.apache.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.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 {
	
	private static final Log log = LogFactory.getLog(RDFServiceImpl.class);
	protected static final String BNODE_ROOT_QUERY = 
	        "SELECT DISTINCT ?s WHERE { ?s ?p ?o OPTIONAL { ?ss ?pp ?s } FILTER (!isBlank(?s) || !bound(?ss)) }";
	
	protected String defaultWriteGraphURI;
	protected List<ChangeListener> registeredListeners = new CopyOnWriteArrayList<ChangeListener>();
	protected List<ModelChangedListener> registeredJenaListeners = new CopyOnWriteArrayList<ModelChangedListener>();
    	
	@Override
	public void newIndividual(String individualURI, 
			                  String individualTypeURI) throws RDFServiceException {
	
       newIndividual(individualURI, individualTypeURI, defaultWriteGraphURI);
	}
		
    @Override
    public void newIndividual(String individualURI, 
                              String individualTypeURI, 
                              String graphURI) throws RDFServiceException {
    
       StringBuilder containsQuery = new StringBuilder("ASK { \n");
       if (graphURI != null) {
           containsQuery.append("  GRAPH <").append(graphURI).append("> { ");
       }
       containsQuery.append("<");   
       containsQuery.append(individualURI);
       containsQuery.append("> ");  
       containsQuery.append("?p ?o");
       if (graphURI != null) {
           containsQuery.append(" } \n");
       }
       containsQuery.append("\n}");
       
       if (sparqlAskQuery(containsQuery.toString())) {
            throw new RDFServiceException("individual already exists");
       } else {
            Triple triple = new Triple(NodeFactory.createURI(individualURI), RDF.type.asNode(), NodeFactory.createURI(individualTypeURI));
            //addTriple(triple, graphURI);
            ChangeSet cs = this.manufactureChangeSet();
            cs.addAddition(new ByteArrayInputStream(
                    sparqlTriple(triple).getBytes()), ModelSerializationFormat.N3, graphURI);
            changeSetUpdate(cs);
       }    
    }
	
	@Override
	public String getDefaultWriteGraphURI() throws RDFServiceException {
        return defaultWriteGraphURI;
	}
		
	@Override
	public synchronized void registerListener(ChangeListener changeListener) throws RDFServiceException {
		if (!registeredListeners.contains(changeListener)) {
		   registeredListeners.add(changeListener);
		}
	}
	
	@Override
	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();
	}    

    protected void notifyListenersOfChanges(ChangeSet changeSet)
            throws IOException {
        if (registeredListeners.isEmpty() && registeredJenaListeners.isEmpty()) {
            return;
        }
        for (ModelChange modelChange: changeSet.getModelChanges()) {
            notifyListeners(modelChange);
        }
    }
	
    protected void notifyListeners(ModelChange modelChange) throws IOException {
        modelChange.getSerializedModel().reset();
        for (ChangeListener listener : registeredListeners) {
            listener.notifyModelChange(modelChange);
        }
        log.debug(registeredJenaListeners.size() + " registered Jena listeners");
        if (registeredJenaListeners.isEmpty()) {
            return;
        }
        modelChange.getSerializedModel().reset();
        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()));<commit_msg>Moved stream reset inside loop so that every listener gets the same state (#79)<commit_after>/* $This file is distributed under the terms of the license in LICENSE$ */

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 org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import org.apache.jena.atlas.io.StringWriterI;
import org.apache.jena.graph.Node;
import org.apache.jena.graph.NodeFactory;
import org.apache.jena.graph.Triple;
import org.apache.jena.query.Query;
import org.apache.jena.query.QueryFactory;
import org.apache.jena.query.QueryParseException;
import org.apache.jena.query.QuerySolution;
import org.apache.jena.query.Syntax;
import org.apache.jena.rdf.model.Literal;
import org.apache.jena.rdf.model.Model;
import org.apache.jena.rdf.model.ModelChangedListener;
import org.apache.jena.rdf.model.ModelFactory;
import org.apache.jena.rdf.model.RDFNode;
import org.apache.jena.rdf.model.Statement;
import org.apache.jena.rdf.model.StmtIterator;
import org.apache.jena.riot.out.NodeFormatter;
import org.apache.jena.riot.out.NodeFormatterTTL;
import org.apache.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.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 {
	
	private static final Log log = LogFactory.getLog(RDFServiceImpl.class);
	protected static final String BNODE_ROOT_QUERY = 
	        "SELECT DISTINCT ?s WHERE { ?s ?p ?o OPTIONAL { ?ss ?pp ?s } FILTER (!isBlank(?s) || !bound(?ss)) }";
	
	protected String defaultWriteGraphURI;
	protected List<ChangeListener> registeredListeners = new CopyOnWriteArrayList<ChangeListener>();
	protected List<ModelChangedListener> registeredJenaListeners = new CopyOnWriteArrayList<ModelChangedListener>();
    	
	@Override
	public void newIndividual(String individualURI, 
			                  String individualTypeURI) throws RDFServiceException {
	
       newIndividual(individualURI, individualTypeURI, defaultWriteGraphURI);
	}
		
    @Override
    public void newIndividual(String individualURI, 
                              String individualTypeURI, 
                              String graphURI) throws RDFServiceException {
    
       StringBuilder containsQuery = new StringBuilder("ASK { \n");
       if (graphURI != null) {
           containsQuery.append("  GRAPH <").append(graphURI).append("> { ");
       }
       containsQuery.append("<");   
       containsQuery.append(individualURI);
       containsQuery.append("> ");  
       containsQuery.append("?p ?o");
       if (graphURI != null) {
           containsQuery.append(" } \n");
       }
       containsQuery.append("\n}");
       
       if (sparqlAskQuery(containsQuery.toString())) {
            throw new RDFServiceException("individual already exists");
       } else {
            Triple triple = new Triple(NodeFactory.createURI(individualURI), RDF.type.asNode(), NodeFactory.createURI(individualTypeURI));
            //addTriple(triple, graphURI);
            ChangeSet cs = this.manufactureChangeSet();
            cs.addAddition(new ByteArrayInputStream(
                    sparqlTriple(triple).getBytes()), ModelSerializationFormat.N3, graphURI);
            changeSetUpdate(cs);
       }    
    }
	
	@Override
	public String getDefaultWriteGraphURI() throws RDFServiceException {
        return defaultWriteGraphURI;
	}
		
	@Override
	public synchronized void registerListener(ChangeListener changeListener) throws RDFServiceException {
		if (!registeredListeners.contains(changeListener)) {
		   registeredListeners.add(changeListener);
		}
	}
	
	@Override
	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();
	}    

    protected void notifyListenersOfChanges(ChangeSet changeSet)
            throws IOException {
        if (registeredListeners.isEmpty() && registeredJenaListeners.isEmpty()) {
            return;
        }
        for (ModelChange modelChange: changeSet.getModelChanges()) {
            notifyListeners(modelChange);
        }
    }
	
    protected void notifyListeners(ModelChange modelChange) throws IOException {
        for (ChangeListener listener : registeredListeners) {
            modelChange.getSerializedModel().reset();
            listener.notifyModelChange(modelChange);
        }
        log.debug(registeredJenaListeners.size() + " registered Jena listeners");
        if (registeredJenaListeners.isEmpty()) {
            return;
        }
        modelChange.getSerializedModel().reset();
        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()));
{'fraction_non_alphanumeric': 0.09662152322702781, 'fraction_numerical': 0.000291247997670016, 'mean_word_length': 3.671768707482993, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6082168', 'n_tokens_mistral': 3879, 'n_tokens_neox': 3717, 'n_words': 666}