import org.openrdf.sail.SailConnection; //导入依赖的package包/类
private static void loadTestStatements() throws Exception {
    final ValueFactory vf = sail.getValueFactory();

    final SailConnection sailConn = sail.getConnection();
    sailConn.begin();
    sailConn.addStatement(vf.createURI("urn:Alice"), vf.createURI("urn:likes"), vf.createURI("urn:icecream"));
    sailConn.addStatement(vf.createURI("urn:Bob"), vf.createURI("urn:likes"), vf.createURI("urn:icecream"));
    sailConn.addStatement(vf.createURI("urn:Charlie"), vf.createURI("urn:likes"), vf.createURI("urn:icecream"));
    sailConn.addStatement(vf.createURI("urn:David"), vf.createURI("urn:likes"), vf.createURI("urn:icecream"));
    sailConn.addStatement(vf.createURI("urn:Eve"), vf.createURI("urn:likes"), vf.createURI("urn:icecream"));
    sailConn.addStatement(vf.createURI("urn:Frank"), vf.createURI("urn:likes"), vf.createURI("urn:icecream"));
    sailConn.addStatement(vf.createURI("urn:George"), vf.createURI("urn:likes"), vf.createURI("urn:icecream"));
    sailConn.addStatement(vf.createURI("urn:Hillary"), vf.createURI("urn:likes"), vf.createURI("urn:icecream"));

    sailConn.addStatement(vf.createURI("urn:Alice"), vf.createURI("urn:hasEyeColor"), vf.createURI("urn:blue"));
    sailConn.addStatement(vf.createURI("urn:Bob"), vf.createURI("urn:hasEyeColor"), vf.createURI("urn:blue"));
    sailConn.addStatement(vf.createURI("urn:Charlie"), vf.createURI("urn:hasEyeColor"), vf.createURI("urn:blue"));
    sailConn.addStatement(vf.createURI("urn:David"), vf.createURI("urn:hasEyeColor"), vf.createURI("urn:blue"));
    sailConn.addStatement(vf.createURI("urn:Eve"), vf.createURI("urn:hasEyeColor"), vf.createURI("urn:blue"));
    sailConn.addStatement(vf.createURI("urn:Frank"), vf.createURI("urn:hasEyeColor"), vf.createURI("urn:blue"));
    sailConn.addStatement(vf.createURI("urn:George"), vf.createURI("urn:hasEyeColor"), vf.createURI("urn:green"));
    sailConn.addStatement(vf.createURI("urn:Hillary"), vf.createURI("urn:hasEyeColor"), vf.createURI("urn:brown"));
    sailConn.commit();
    sailConn.close();
}
 

import org.openrdf.sail.SailConnection; //导入依赖的package包/类
@Test
public void testGetStatements() throws SailException {
    final SailConnection con = cas.getConnection();
    try {
        con.begin();

        Assert.assertTrue(hasStatement(con, u1, p1, l1));
        Assert.assertFalse(hasStatement(con, u2, p2, l2));

        con.commit();
    } catch (final Throwable t) {
        con.rollback();
        throw t;
    } finally {
        con.close();
    }
}
 

import org.openrdf.sail.SailConnection; //导入依赖的package包/类
@Test
public void testGetContextIDs() throws SailException {
    final SailConnection con = cas.getConnection();
    try {
        con.begin();

        final CloseableIteration<? extends Resource, SailException> cid = con.getContextIDs();
        try {
            assertTrue(cid.hasNext());
            assertThat(cid.next(), CoreMatchers.is(c1));
            assertFalse(cid.hasNext());
        } finally {
            cid.close();
        }

        con.commit();
    } catch (final Throwable t) {
        con.rollback();
        throw t;
    } finally {
        con.close();
    }
}
 

import org.openrdf.sail.SailConnection; //导入依赖的package包/类
void initInference(SailConnection conn, ValueFactory vf) throws SailException {
    synchronized (this) {
        if (toSub == null) {
            boolean done = false;
            try {
                toSub = new HashMap<String, Collection<String>>();
                toSuper = new HashMap<String, String>();
                for (IndexConfiguration config: indexConfigurations.values()) {
                    String type = config.getMatchType();
                    initInference(conn, vf, type);
                }
                done = true;
            } finally {
                if (!done) {
                    //Don't leave half configured information one failure:
                    toSub = null;
                    toSuper = null;
                }
            }
        }
    }
}
 

import org.openrdf.sail.SailConnection; //导入依赖的package包/类
/**
 * Returns a wrapper for the Sail connections of the federation members.
 * 
 * TODO: currently opens connections to *all* federation members,
 *       should better use lazy initialization of the connections.
 *       
 * @return the Sail connection wrapper.
 */
@Override
protected SailConnection getConnectionInternal() throws SailException {
	
	if (!this.initialized)
		throw new IllegalStateException("Sail has not been initialized.");
	
	return new FederationSailConnection(this);
}
 

import org.openrdf.sail.SailConnection; //导入依赖的package包/类
/**
 * Cleanup inferred triples that are no longer supported by any justification.
 */
private void cleanupUnsupported(KiWiReasoningConnection connection) throws SQLException, SailException {
    updateTaskStatus("cleaning up unsupported triples");

    int count = 0, total = 0;

    startTask("Unsupported Triple Cleaner", TASK_GROUP);
    updateTaskStatus("loading unsupported triples");

    CloseableIteration<KiWiTriple,SQLException> tripleIterator = connection.listUnsupportedTriples();
    try {
        if(tripleIterator.hasNext()) {

            updateTaskStatus("deleting unsupported triples");
            SailConnection tc = store.getConnection();
            KiWiSailConnection ic = getWrappedConnection(tc);
            try {
                tc.begin();
                while(tripleIterator.hasNext()) {
                    ic.removeInferredStatement(tripleIterator.next());
                    count++;
                }
                log.debug("removed {} unsupported triples",count);
                tc.commit();
            } catch(SailException ex) {
                ic.rollback();
                throw ex;
            } finally {
                ic.close();
            }
        }
    } finally {
        Iterations.closeCloseable(tripleIterator);

    }
}
 

import org.openrdf.sail.SailConnection; //导入依赖的package包/类
/**
 * Return the KiWiSailConnection underlying a given sail connection. The method will follow wrapped
 * connections until it finds the KiWiSailConnection, or otherwise throws a SailException.
 * @param connection
 * @return
 */
private KiWiSailConnection getWrappedConnection(SailConnection connection) throws SailException {
    SailConnection it = connection;
    while(it instanceof SailConnectionWrapper) {
        it = ((SailConnectionWrapper) it).getWrappedConnection();
        if(it instanceof KiWiSailConnection) {
            return (KiWiSailConnection) it;
        }
    }
    throw new SailException("no underlying KiWiSailConnection found for connection");
}
 

import org.openrdf.sail.SailConnection; //导入依赖的package包/类
/**
 * Return the cache entry for the given resource, or null if this entry does not exist.
 *
 *
 * @param resource the resource to retrieve the cache entry for
 * @return
 */
@Override
public CacheEntry getEntry(URI resource) {
    try {
        try(LDCachingKiWiPersistenceConnection dbcon = persistence.getConnection()) {

            // load cache entry from database
            CacheEntry ce = dbcon.getCacheEntry(resource.stringValue());

            // if entry exists, load triples for the resource from the cache context of the repository
            if(ce != null) {
                SailConnection con = store.getConnection();
                try {
                    con.begin();

                    Model triples = new TreeModel();
                    ModelCommons.add(triples,con.getStatements(resource,null,null,true,store.getValueFactory().createURI(cacheContext)));
                    ce.setTriples(triples);

                    con.commit();
                } catch(SailException ex) {
                    con.rollback();
                } finally {
                    con.close();
                }
            }
            return ce;

        }

    } catch (SailException | SQLException e) {
        log.error("could not retrieve cached triples from repository",e);
    }

    return null;
}
 

import org.openrdf.sail.SailConnection; //导入依赖的package包/类
/**
 * Update the cache entry for the given resource with the given entry.
 *
 * @param resource the resource to update
 * @param entry    the entry for the resource
 */
@Override
public void putEntry(URI resource, CacheEntry entry) {
    try {
        try(LDCachingKiWiPersistenceConnection dbcon = persistence.getConnection()) {

            // store cache entry in database
            dbcon.removeCacheEntry(resource.stringValue());

            // update triples in cache
            SailConnection con = store.getConnection();
            try {
                con.begin();

                con.removeStatements(resource, null, null, store.getValueFactory().createURI(cacheContext));
                for(Statement stmt : entry.getTriples()) {
                    con.addStatement(stmt.getSubject(), stmt.getPredicate(), stmt.getObject(), store.getValueFactory().createURI(cacheContext));
                }

                con.commit();

                entry.setResource(store.getValueFactory().createURI(resource.stringValue()));

                dbcon.storeCacheEntry(entry);
            } catch(SailException ex) {
                con.rollback();
            } finally {
                con.close();
            }

        }

    } catch (SailException | SQLException e) {
        log.error("could not retrieve cached triples from repository",e);
    }

}
 

import org.openrdf.sail.SailConnection; //导入依赖的package包/类
/**
 * Remove the cache entry for the given resource if it exists. Does nothing otherwise.
 *
 * @param resource the resource to remove the entry for
 */
@Override
public void removeEntry(URI resource) {
    try {
        try(LDCachingKiWiPersistenceConnection dbcon = persistence.getConnection()) {

            // store cache entry in database
            dbcon.removeCacheEntry(resource.stringValue());

            // update triples in cache
            SailConnection con = store.getConnection();
            try {
                con.begin();

                con.removeStatements(resource, null, null, store.getValueFactory().createURI(cacheContext));

                con.commit();
            } catch(SailException ex) {
                con.rollback();
            } finally {
                con.close();
            }

        }

    } catch (SailException | SQLException e) {
        log.error("could not remove cached triples from repository",e);
    }
}
 

import org.openrdf.sail.SailConnection; //导入依赖的package包/类
/**
 * Clear all entries in the cache backend.
 */
@Override
public void clear() {
    try {
        try(LDCachingKiWiPersistenceConnection dbcon = persistence.getConnection()) {

            // list all entries and remove them
            CloseableIteration<KiWiCacheEntry, SQLException> entries = dbcon.listAll();
            while (entries.hasNext()) {
                dbcon.removeCacheEntry(entries.next());
            }

            // update triples in cache
            SailConnection con = store.getConnection();
            try {
                con.begin();

                con.removeStatements((Resource) null, null, null, store.getValueFactory().createURI(cacheContext));

                con.commit();
            } catch(SailException ex) {
                con.rollback();
            } finally {
                con.close();
            }

        }

    } catch (SailException | SQLException e) {
        log.error("could not remove cached triples from repository",e);
    }

}
 

import org.openrdf.sail.SailConnection; //导入依赖的package包/类
private static ResourceConnection getWrappedResourceConnection(SailConnection connection) {
    if(connection instanceof ResourceConnection) {
        return (ResourceConnection)connection;
    } else if(connection instanceof SailConnectionWrapper) {
        return getWrappedResourceConnection(((SailConnectionWrapper) connection).getWrappedConnection());
    } else {
        return null;
    }

}
 

import org.openrdf.sail.SailConnection; //导入依赖的package包/类
protected static boolean hasStatement(SailConnection con, Resource subj,
URI pred, Value object, Resource... contexts) throws SailException {
    final CloseableIteration<? extends Statement, SailException> stmts = con.getStatements(subj, pred, object, true, contexts);
    try {
        return stmts.hasNext();
    } finally {
        stmts.close();
    }
}
 

import org.openrdf.sail.SailConnection; //导入依赖的package包/类
private BigdataSail.BigdataSailConnection getBaseSailConnection(SailConnection con) {
    SailConnection wrapped = con;
    while(wrapped instanceof SailConnectionWrapper) {
        wrapped = ((SailConnectionWrapper) wrapped).getWrappedConnection();
    }
    if(wrapped instanceof BigDataSesame27Sail.BigDataSesame27SailConnection) {
        return ((BigDataSesame27Sail.BigDataSesame27SailConnection) wrapped).getWrapped();
    }
    if(wrapped instanceof BigdataSail.BigdataSailConnection) {
        return (BigdataSail.BigdataSailConnection) wrapped;
    }
    return null;
}
 

import org.openrdf.sail.SailConnection; //导入依赖的package包/类
protected RyaSailRepositoryConnection(SailRepository repository, SailConnection sailConnection) {
    super(repository, sailConnection);
}
 

import org.openrdf.sail.SailConnection; //导入依赖的package包/类
@Override
protected SailConnection getConnectionInternal() throws SailException {
    return new RdfCloudTripleStoreConnection(this, conf, VF);
}
 

import org.openrdf.sail.SailConnection; //导入依赖的package包/类
/**
 * Ensure a user that has not been added to the Rya instance can not interact with it.
 */
@Test
public void userNotAddedCanNotInsert() throws Exception {
    final String user = testInstance.createUniqueUser();
    final SecurityOperations secOps = super.getConnector().securityOperations();

    final RyaClient userAClient = AccumuloRyaClientFactory.build(
            new AccumuloConnectionDetails(ADMIN_USER, ADMIN_USER.toCharArray(), getInstanceName(), getZookeepers()),
            super.getClusterInstance().getCluster().getConnector(ADMIN_USER, ADMIN_USER));

    // Install the instance of Rya.
    userAClient.getInstall().install(getRyaInstanceName(), InstallConfiguration.builder().build());

    // Create the user that will not be added to the instance of Rya, but will try to scan it.
    secOps.createLocalUser(user, new PasswordToken(user));

    //Try to add a statement the Rya instance with the unauthorized user. This should fail.
    boolean securityExceptionThrown = false;

    Sail sail = null;
    SailConnection sailConn = null;
    try {
        final AccumuloRdfConfiguration userCConf = makeRyaConfig(getRyaInstanceName(), user, user, getInstanceName(), getZookeepers());
        sail = RyaSailFactory.getInstance(userCConf);
        sailConn = sail.getConnection();

        final ValueFactory vf = sail.getValueFactory();
        sailConn.addStatement(vf.createURI("urn:Alice"), vf.createURI("urn:talksTo"), vf.createURI("urn:Bob"));

    } catch(final RuntimeException e) {
        final Throwable cause = e.getCause();
        if(cause instanceof AccumuloSecurityException) {
            securityExceptionThrown = true;
        }
    } finally {
        if(sailConn != null) {
            sailConn.close();
        }
        if(sail != null) {
            sail.shutDown();
        }
    }

    assertTrue(securityExceptionThrown);
}
 

import org.openrdf.sail.SailConnection; //导入依赖的package包/类
/**
 * Ensure a user that has been added to the Rya instance can interact with it.
 */
@Test
public void userAddedCanInsert() throws Exception {
    final String user = testInstance.createUniqueUser();
    final SecurityOperations secOps = super.getConnector().securityOperations();

    final RyaClient userAClient = AccumuloRyaClientFactory.build(
            new AccumuloConnectionDetails(ADMIN_USER, ADMIN_USER.toCharArray(), getInstanceName(), getZookeepers()),
            super.getClusterInstance().getCluster().getConnector(ADMIN_USER, ADMIN_USER));

    // Create the user that will not be added to the instance of Rya, but will try to scan it.
    secOps.createLocalUser(user, new PasswordToken(user));

    // Install the instance of Rya.
    userAClient.getInstall().install(getRyaInstanceName(), InstallConfiguration.builder().build());

    // Add the user.
    userAClient.getAddUser().get().addUser(getRyaInstanceName(), user);

    // Try to add a statement to the Rya instance. This should succeed.
    Sail sail = null;
    SailConnection sailConn = null;

    try {
        final AccumuloRdfConfiguration userDConf = makeRyaConfig(getRyaInstanceName(), user, user, getInstanceName(), getZookeepers());
        sail = RyaSailFactory.getInstance(userDConf);
        sailConn = sail.getConnection();

        final ValueFactory vf = sail.getValueFactory();
        sailConn.begin();
        sailConn.addStatement(vf.createURI("urn:Alice"), vf.createURI("urn:talksTo"), vf.createURI("urn:Bob"));
        sailConn.close();

    } finally {
        if(sailConn != null) {
            sailConn.close();
        }
        if(sail != null) {
            sail.shutDown();
        }
    }
}
 

import org.openrdf.sail.SailConnection; //导入依赖的package包/类
/**
 * Ensure a user that has been removed from the Rya instance can not interact with it.
 */
@Test
public void removedUserCanNotInsert() throws Exception {
    final String adminUser = testInstance.createUniqueUser();
    final String user = testInstance.createUniqueUser();
    final SecurityOperations secOps = super.getConnector().securityOperations();

    // Create the user that will install the instance of Rya.
    secOps.createLocalUser(adminUser, new PasswordToken(adminUser));
    secOps.grantSystemPermission(adminUser, SystemPermission.CREATE_TABLE);

    final RyaClient userAClient = AccumuloRyaClientFactory.build(
            new AccumuloConnectionDetails(adminUser, adminUser.toCharArray(), getInstanceName(), getZookeepers()),
            super.getClusterInstance().getCluster().getConnector(adminUser, adminUser));

    // Create the user that will be added to the instance of Rya.
    secOps.createLocalUser(user, new PasswordToken(user));

    final RyaClient userCClient = AccumuloRyaClientFactory.build(
            new AccumuloConnectionDetails(user, user.toCharArray(), getInstanceName(), getZookeepers()),
            super.getClusterInstance().getCluster().getConnector(user, user));

    // Install the instance of Rya.
    userAClient.getInstall().install(getRyaInstanceName(), InstallConfiguration.builder().build());

    // Add userC.
    userAClient.getAddUser().get().addUser(getRyaInstanceName(), user);

    // Remove userA.
    userCClient.getRemoveUser().get().removeUser(getRyaInstanceName(), adminUser);

    // Show that userA can not insert anything.
    boolean securityExceptionThrown = false;

    Sail sail = null;
    SailConnection sailConn = null;
    try {
        final AccumuloRdfConfiguration userAConf = makeRyaConfig(getRyaInstanceName(), adminUser, adminUser, getInstanceName(), getZookeepers());
        sail = RyaSailFactory.getInstance(userAConf);
        sailConn = sail.getConnection();

        final ValueFactory vf = sail.getValueFactory();
        sailConn.addStatement(vf.createURI("urn:Alice"), vf.createURI("urn:talksTo"), vf.createURI("urn:Bob"));

    } catch(final RuntimeException e) {
        final Throwable cause = e.getCause();
        if(cause instanceof AccumuloSecurityException) {
            securityExceptionThrown = true;
        }
    } finally {
        if(sailConn != null) {
            sailConn.close();
        }
        if(sail != null) {
            sail.shutDown();
        }
    }

    assertTrue(securityExceptionThrown);
}
 

import org.openrdf.sail.SailConnection; //导入依赖的package包/类
@Override
public SailConnection createConnection() throws SailException {
  try {
    Connection db = getConnection();
    db.setAutoCommit(true);
    if (db.getTransactionIsolation() != TRANSACTION_READ_COMMITTED) {
      db.setTransactionIsolation(TRANSACTION_READ_COMMITTED);
    }
    TripleManager tripleManager = new TripleManager();
    GeneralDBTripleRepository s = new PostGISTripleRepository();
    s.setTripleManager(tripleManager);
    s.setValueFactory(vf);
    s.setConnection(db);
    s.setBNodeTable(bnodeTable);
    s.setURITable(uriTable);
    s.setLiteralTable(literalTable);
    s.setIdSequence(ids);
    DefaultSailChangedEvent sailChangedEvent = new DefaultSailChangedEvent(sail);
    s.setSailChangedEvent(sailChangedEvent);
    TableFactory tables = createTableFactory();
    TransTableManager trans = createTransTableManager();
    trans.setIdSequence(ids);
    tripleManager.setTransTableManager(trans);
    trans.setBatchQueue(tripleManager.getQueue());
    trans.setSailChangedEvent(sailChangedEvent);
    trans.setConnection(db);
    trans.setTemporaryTableFactory(tables);
    trans.setStatementsTable(tripleTableManager);
    trans.setFromDummyTable(getFromDummyTable());
    trans.initialize();
    s.setTransaction(trans);
    GeneralDBQueryBuilderFactory bfactory = createQueryBuilderFactory();
    bfactory.setValueFactory(vf);
    bfactory.setUsingHashTable(hashManager != null);
    s.setQueryBuilderFactory(bfactory);

    GeneralDBConnection conn = new GeneralDBConnection(sail, s);
    conn.setNamespaces(namespaces);
    GeneralDBEvaluationFactory efactory = new PostGISEvaluationFactory();
    efactory.setQueryBuilderFactory(bfactory);
    efactory.setRdbmsTripleRepository(s);
    efactory.setIdSequence(ids);
    conn.setRdbmsEvaluationFactory(efactory);
    GeneralDBQueryOptimizer optimizer = createOptimizer();
    GeneralDBSelectQueryOptimizerFactory selectOptimizerFactory =
        createSelectQueryOptimizerFactory();
    selectOptimizerFactory.setTransTableManager(trans);
    selectOptimizerFactory.setValueFactory(vf);
    selectOptimizerFactory.setIdSequence(ids);
    optimizer.setSelectQueryOptimizerFactory(selectOptimizerFactory);
    optimizer.setValueFactory(vf);
    optimizer.setBnodeTable(bnodeTable);
    optimizer.setUriTable(uriTable);
    optimizer.setLiteralTable(literalTable);
    optimizer.setHashTable(hashTable);
    conn.setRdbmsQueryOptimizer(optimizer);
    conn.setLockManager(lock);
    return conn;
  } catch (SQLException e) {
    throw new RdbmsException(e);
  }
}