![[Jenkins]](/jenkins/static/ab118e47/images/svgs/logo.svg){kind=logo}
Package: StorageConnector
StorageConnector
| name | instruction | branch | complexity | line | method | ||||||||||||||||||||
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| StorageConnector(DriverConfiguration) |
|
|
|
|
|
||||||||||||||||||||
| acquire(int) |
|
|
|
|
|
||||||||||||||||||||
| acquireConnection() |
|
|
|
|
|
||||||||||||||||||||
| addStatements(Collection) |
|
|
|
|
|
||||||||||||||||||||
| begin() |
|
|
|
|
|
||||||||||||||||||||
| close() |
|
|
|
|
|
||||||||||||||||||||
| commit() |
|
|
|
|
|
||||||||||||||||||||
| containsStatement(Resource, IRI, Value, boolean, Collection) |
|
|
|
|
|
||||||||||||||||||||
| executeBooleanQuery(String) |
|
|
|
|
|
||||||||||||||||||||
| executeSelectQuery(String) |
|
|
|
|
|
||||||||||||||||||||
| executeUpdate(String) |
|
|
|
|
|
||||||||||||||||||||
| findStatements(Resource, IRI, Value, boolean) |
|
|
|
|
|
||||||||||||||||||||
| findStatements(Resource, IRI, Value, boolean, Collection) |
|
|
|
|
|
||||||||||||||||||||
| getContexts() |
|
|
|
|
|
||||||||||||||||||||
| getValueFactory() |
|
|
|
|
|
||||||||||||||||||||
| initialize() |
|
|
|
|
|
||||||||||||||||||||
| invalidReconnectAttemptsConfig() |
|
|
|
|
|
||||||||||||||||||||
| isInMemoryRepository(Repository) |
|
|
|
|
|
||||||||||||||||||||
| releaseConnection(RepositoryConnection) |
|
|
|
|
|
||||||||||||||||||||
| removeStatements(Collection) |
|
|
|
|
|
||||||||||||||||||||
| resolveMaxReconnectAttempts() |
|
|
|
|
|
||||||||||||||||||||
| rollback() |
|
|
|
|
|
||||||||||||||||||||
| setRepository(Repository) |
|
|
|
|
|
||||||||||||||||||||
| static {...} |
|
|
|
|
|
||||||||||||||||||||
| unwrap(Class) |
|
|
|
|
|
||||||||||||||||||||
Coverage
1: /**
2: * Copyright (C) 2022 Czech Technical University in Prague
3: *
4: * This program is free software: you can redistribute it and/or modify it under
5: * the terms of the GNU General Public License as published by the Free Software
6: * Foundation, either version 3 of the License, or (at your option) any
7: * later version.
8: *
9: * This program is distributed in the hope that it will be useful, but WITHOUT
10: * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
11: * FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
12: * details. You should have received a copy of the GNU General Public License
13: * along with this program. If not, see <http://www.gnu.org/licenses/>.
14: */
15: package cz.cvut.kbss.ontodriver.sesame.connector;
16:
17: import cz.cvut.kbss.ontodriver.config.DriverConfiguration;
18: import cz.cvut.kbss.ontodriver.exception.OntoDriverException;
19: import cz.cvut.kbss.ontodriver.sesame.config.Constants;
20: import cz.cvut.kbss.ontodriver.sesame.config.SesameConfigParam;
21: import cz.cvut.kbss.ontodriver.sesame.config.SesameOntoDriverProperties;
22: import cz.cvut.kbss.ontodriver.sesame.exceptions.SesameDriverException;
23: import org.eclipse.rdf4j.common.iteration.Iterations;
24: import org.eclipse.rdf4j.model.*;
25: import org.eclipse.rdf4j.query.TupleQueryResult;
26: import org.eclipse.rdf4j.repository.Repository;
27: import org.eclipse.rdf4j.repository.RepositoryConnection;
28: import org.eclipse.rdf4j.repository.RepositoryException;
29: import org.eclipse.rdf4j.repository.RepositoryResult;
30: import org.eclipse.rdf4j.repository.manager.RepositoryManager;
31: import org.eclipse.rdf4j.repository.sail.SailRepository;
32: import org.eclipse.rdf4j.sail.Sail;
33: import org.eclipse.rdf4j.sail.helpers.SailWrapper;
34: import org.eclipse.rdf4j.sail.memory.MemoryStore;
35: import org.slf4j.Logger;
36: import org.slf4j.LoggerFactory;
37:
38: import java.util.Collection;
39: import java.util.Collections;
40: import java.util.List;
41: import java.util.Objects;
42:
43: class StorageConnector extends AbstractConnector {
44:
45: private static final Logger LOG = LoggerFactory.getLogger(StorageConnector.class);
46:
47: private final DriverConfiguration configuration;
48:
49: private final int maxReconnectAttempts;
50:
51: private Repository repository;
52: private RepositoryManager manager;
53: private RepositoryConnection connection;
54:
55: StorageConnector(DriverConfiguration configuration) throws SesameDriverException {
56:• assert configuration != null;
57:
58: this.configuration = configuration;
59: this.maxReconnectAttempts = resolveMaxReconnectAttempts();
60: initialize();
61: this.open = true;
62: }
63:
64: private int resolveMaxReconnectAttempts() throws SesameDriverException {
65: try {
66:• final int attempts = configuration.isSet(SesameConfigParam.RECONNECT_ATTEMPTS) ? Integer.parseInt(
67: configuration.getProperty(SesameConfigParam.RECONNECT_ATTEMPTS)) :
68: Constants.DEFAULT_RECONNECT_ATTEMPTS_COUNT;
69:• if (attempts < 0) {
70: throw invalidReconnectAttemptsConfig();
71: }
72: return attempts;
73: } catch (NumberFormatException e) {
74: throw invalidReconnectAttemptsConfig();
75: }
76: }
77:
78: private static SesameDriverException invalidReconnectAttemptsConfig() {
79: return new SesameDriverException(
80: "Invalid value of configuration parameter " + SesameOntoDriverProperties.SESAME_RECONNECT_ATTEMPTS +
81: ". Must be a non-negative integer.");
82: }
83:
84: private void initialize() throws SesameDriverException {
85: final RepositoryConnectorInitializer initializer = new RepositoryConnectorInitializer(configuration,
86: maxReconnectAttempts);
87: initializer.initializeRepository();
88: this.manager = initializer.getManager();
89: this.repository = initializer.getRepository();
90: }
91:
92: @Override
93: public void close() throws SesameDriverException {
94:• if (!open) {
95: return;
96: }
97: LOG.debug("Closing connector to repository {}.", configuration.getStorageProperties().getPhysicalURI());
98: try {
99: repository.shutDown();
100:• if (manager != null) {
101: manager.shutDown();
102: }
103: } catch (RepositoryException e) {
104: throw new SesameDriverException("Exception caught when closing Sesame repository connection.", e);
105: } finally {
106: this.open = false;
107: }
108: }
109:
110: @Override
111: public TupleQueryResult executeSelectQuery(String query) throws SesameDriverException {
112: final RepositoryConnection conn = acquireConnection();
113: return new ConnectionStatementExecutor(conn).executeSelectQuery(query);
114: // The connection is released by the result set once it is closed
115: }
116:
117: RepositoryConnection acquireConnection() throws SesameDriverException {
118: // Workaround for local native storage being reset when multiple drivers access it
119:• if (!repository.isInitialized()) {
120: initialize();
121: }
122: LOG.trace("Acquiring repository connection.");
123: return acquire(1);
124: }
125:
126: private RepositoryConnection acquire(int attempts) throws SesameDriverException {
127: try {
128: return repository.getConnection();
129: } catch (RepositoryException e) {
130:• if (attempts < maxReconnectAttempts) {
131: LOG.warn("Unable to acquire repository connection. Error is: {}. Retrying...", e.getMessage());
132: return acquire(attempts + 1);
133: }
134: LOG.error("Threshold of failed connection acquisition attempts reached, throwing exception.");
135: throw new SesameDriverException(e);
136: }
137: }
138:
139: void releaseConnection(RepositoryConnection conn) throws SesameDriverException {
140: try {
141:• if (conn != null) {
142: LOG.trace("Releasing repository connection.");
143: conn.close();
144: }
145: } catch (RepositoryException e) {
146: throw new SesameDriverException(e);
147: }
148: }
149:
150: @Override
151: public boolean executeBooleanQuery(String query) throws SesameDriverException {
152: try (final RepositoryConnection conn = acquireConnection()) {
153: return new ConnectionStatementExecutor(conn).executeBooleanQuery(query);
154: }
155: }
156:
157: @Override
158: public void executeUpdate(String query) throws SesameDriverException {
159: try (final RepositoryConnection conn = acquireConnection()) {
160: new ConnectionStatementExecutor(conn).executeUpdate(query);
161: }
162: }
163:
164: @Override
165: public List<Resource> getContexts() throws SesameDriverException {
166: try (final RepositoryConnection conn = acquireConnection()) {
167: final RepositoryResult<Resource> res = conn.getContextIDs();
168: return Iterations.asList(res);
169: } catch (RepositoryException e) {
170: throw new SesameDriverException(e);
171: }
172: }
173:
174: @Override
175: public ValueFactory getValueFactory() {
176: return repository.getValueFactory();
177: }
178:
179: @Override
180: public void begin() throws SesameDriverException {
181: super.begin();
182: this.connection = acquireConnection();
183: try {
184: connection.begin();
185: } catch (RepositoryException e) {
186: transaction.rollback();
187: throw new SesameDriverException(e);
188: }
189: }
190:
191: @Override
192: public void commit() throws SesameDriverException {
193:• assert connection != null;
194:
195: transaction.commit();
196: try {
197: connection.commit();
198: connection.close();
199: this.connection = null;
200: transaction.afterCommit();
201: } catch (RepositoryException e) {
202: transaction.rollback();
203: throw new SesameDriverException(e);
204: }
205: }
206:
207: @Override
208: public void rollback() throws SesameDriverException {
209:• assert connection != null;
210: transaction.rollback();
211: try {
212: connection.rollback();
213: connection.close();
214: this.connection = null;
215: } catch (RepositoryException e) {
216: throw new SesameDriverException(e);
217: } finally {
218: transaction.afterRollback();
219: }
220: }
221:
222: @Override
223: public void addStatements(Collection<Statement> statements) throws SesameDriverException {
224: verifyTransactionActive();
225:• assert connection != null;
226:
227: try {
228: connection.add(statements);
229: } catch (RepositoryException e) {
230: throw new SesameDriverException(e);
231: }
232: }
233:
234: @Override
235: public void removeStatements(Collection<Statement> statements) throws SesameDriverException {
236: verifyTransactionActive();
237:• assert connection != null;
238:
239: try {
240: connection.remove(statements);
241: } catch (RepositoryException e) {
242: throw new SesameDriverException(e);
243: }
244: }
245:
246: @Override
247: public Collection<Statement> findStatements(Resource subject, IRI property, Value value, boolean includeInferred)
248: throws SesameDriverException {
249: return findStatements(subject, property, value, includeInferred, Collections.emptySet());
250: }
251:
252: @Override
253: public Collection<Statement> findStatements(Resource subject, org.eclipse.rdf4j.model.IRI property,
254: Value value, boolean includeInferred, Collection<IRI> context)
255: throws SesameDriverException {
256: try (final RepositoryConnection conn = acquireConnection()) {
257: final RepositoryResult<Statement> m;
258: m = conn.getStatements(subject, property, null, includeInferred, context.toArray(new IRI[0]));
259: return Iterations.asList(m);
260: } catch (RepositoryException e) {
261: throw new SesameDriverException(e);
262: }
263: }
264:
265: @Override
266: public boolean containsStatement(Resource subject, IRI property, Value value, boolean includeInferred,
267: Collection<IRI> contexts)
268: throws SesameDriverException {
269: try (final RepositoryConnection conn = acquireConnection()) {
270: return conn.hasStatement(subject, property, value, includeInferred, contexts.toArray(new IRI[0]));
271: } catch (RepositoryException e) {
272: throw new SesameDriverException(e);
273: }
274: }
275:
276: @Override
277: public <T> T unwrap(Class<T> cls) throws OntoDriverException {
278:• if (cls.isAssignableFrom(getClass())) {
279: return cls.cast(this);
280: }
281:• if (cls.isAssignableFrom(repository.getClass())) {
282: return cls.cast(repository);
283: }
284: throw new SesameDriverException("No instance of class " + cls + " found.");
285: }
286:
287: /**
288: * Replaces the currently open repository with the specified one.
289: * <p>
290: * Note that this functionality is only supported for in-memory stores.
291: *
292: * @param newRepository The new repository to set
293: */
294: public void setRepository(Repository newRepository) {
295: Objects.requireNonNull(newRepository);
296:• if (!isInMemoryRepository(repository)) {
297: throw new UnsupportedOperationException("Cannot replace repository which is not in-memory.");
298: }
299:• if (transaction.isActive()) {
300: throw new IllegalStateException("Cannot replace repository in transaction.");
301: }
302: repository.shutDown();
303:• assert newRepository.isInitialized();
304: this.repository = newRepository;
305: // Since in-memory repositories are not registered in RepositoryManager, we shouldn't need to deal with it
306: }
307:
308: private static boolean isInMemoryRepository(Repository repo) {
309:• if (!(repo instanceof SailRepository)) {
310: return false;
311: }
312: final Sail sail = ((SailRepository) repo).getSail();
313:• return sail instanceof SailWrapper ? ((SailWrapper) sail).getBaseSail() instanceof MemoryStore :
314: sail instanceof MemoryStore;
315: }
316: }