![[Jenkins]](/jenkins/static/c1408a8d/images/svgs/logo.svg){kind=logo}
Package: RepositoryConnectorInitializer
RepositoryConnectorInitializer
| name | instruction | branch | complexity | line | method | ||||||||||||||||||||
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| RepositoryConnectorInitializer(DriverConfiguration) |
|
|
|
|
|
||||||||||||||||||||
| connectToRemote(String, int) |
|
|
|
|
|
||||||||||||||||||||
| connectToRemoteRepository(String) |
|
|
|
|
|
||||||||||||||||||||
| createInMemoryRepository() |
|
|
|
|
|
||||||||||||||||||||
| createLocalNativeRepositoryConfig(String, DriverConfiguration) |
|
|
|
|
|
||||||||||||||||||||
| createLocalRepository() |
|
|
|
|
|
||||||||||||||||||||
| createNativeRepository(DriverConfiguration, String) |
|
|
|
|
|
||||||||||||||||||||
| createRepositoryFromConfig() |
|
|
|
|
|
||||||||||||||||||||
| getConfigFileContent() |
|
|
|
|
|
||||||||||||||||||||
| getManager() |
|
|
|
|
|
||||||||||||||||||||
| getMaxReconnectAttempts() |
|
|
|
|
|
||||||||||||||||||||
| getRepository() |
|
|
|
|
|
||||||||||||||||||||
| getRepositoryId() |
|
|
|
|
|
||||||||||||||||||||
| getRepositoryManagerBaseDir() |
|
|
|
|
|
||||||||||||||||||||
| initializeRepository() |
|
|
|
|
|
||||||||||||||||||||
| invalidReconnectAttemptsConfig() |
|
|
|
|
|
||||||||||||||||||||
| isFileUri(URI) |
|
|
|
|
|
||||||||||||||||||||
| isRemoteRepository(URI) |
|
|
|
|
|
||||||||||||||||||||
| loadRepositoryConfig() |
|
|
|
|
|
||||||||||||||||||||
| resolveMaxReconnectAttempts() |
|
|
|
|
|
||||||||||||||||||||
| static {...} |
|
|
|
|
|
||||||||||||||||||||
| validateNativeStorePath(String) |
|
|
|
|
|
||||||||||||||||||||
| verifyRepositoryCreated(URI, boolean) |
|
|
|
|
|
||||||||||||||||||||
Coverage
1: /*
2: * JOPA
3: * Copyright (C) 2023 Czech Technical University in Prague
4: *
5: * This library is free software; you can redistribute it and/or
6: * modify it under the terms of the GNU Lesser General Public
7: * License as published by the Free Software Foundation; either
8: * version 3.0 of the License, or (at your option) any later version.
9: *
10: * This library is distributed in the hope that it will be useful,
11: * but WITHOUT ANY WARRANTY; without even the implied warranty of
12: * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
13: * Lesser General Public License for more details.
14: *
15: * You should have received a copy of the GNU Lesser General Public
16: * License along with this library.
17: */
18: package cz.cvut.kbss.ontodriver.rdf4j.connector.init;
19:
20: import cz.cvut.kbss.ontodriver.config.DriverConfiguration;
21: import cz.cvut.kbss.ontodriver.rdf4j.config.Constants;
22: import cz.cvut.kbss.ontodriver.rdf4j.config.Rdf4jConfigParam;
23: import cz.cvut.kbss.ontodriver.rdf4j.config.Rdf4jOntoDriverProperties;
24: import cz.cvut.kbss.ontodriver.rdf4j.exception.Rdf4jDriverException;
25: import cz.cvut.kbss.ontodriver.rdf4j.exception.RepositoryCreationException;
26: import cz.cvut.kbss.ontodriver.rdf4j.exception.RepositoryNotFoundException;
27: import org.eclipse.rdf4j.model.Model;
28: import org.eclipse.rdf4j.model.Resource;
29: import org.eclipse.rdf4j.model.vocabulary.CONFIG;
30: import org.eclipse.rdf4j.model.vocabulary.RDF;
31: import org.eclipse.rdf4j.repository.Repository;
32: import org.eclipse.rdf4j.repository.RepositoryException;
33: import org.eclipse.rdf4j.repository.config.RepositoryConfig;
34: import org.eclipse.rdf4j.repository.config.RepositoryConfigException;
35: import org.eclipse.rdf4j.repository.config.RepositoryConfigSchema;
36: import org.eclipse.rdf4j.repository.http.HTTPRepository;
37: import org.eclipse.rdf4j.repository.manager.RemoteRepositoryManager;
38: import org.eclipse.rdf4j.repository.manager.RepositoryManager;
39: import org.eclipse.rdf4j.repository.manager.RepositoryProvider;
40: import org.eclipse.rdf4j.repository.sail.SailRepository;
41: import org.eclipse.rdf4j.repository.sail.config.SailRepositoryConfig;
42: import org.eclipse.rdf4j.rio.RDFFormat;
43: import org.eclipse.rdf4j.rio.Rio;
44: import org.eclipse.rdf4j.sail.config.SailImplConfig;
45: import org.eclipse.rdf4j.sail.inferencer.fc.SchemaCachingRDFSInferencer;
46: import org.eclipse.rdf4j.sail.inferencer.fc.config.SchemaCachingRDFSInferencerConfig;
47: import org.eclipse.rdf4j.sail.memory.MemoryStore;
48: import org.eclipse.rdf4j.sail.nativerdf.config.NativeStoreConfig;
49: import org.slf4j.Logger;
50: import org.slf4j.LoggerFactory;
51:
52: import java.io.FileInputStream;
53: import java.io.FileNotFoundException;
54: import java.io.IOException;
55: import java.io.InputStream;
56: import java.net.URI;
57: import java.util.Optional;
58: import java.util.Set;
59:
60: public class RepositoryConnectorInitializer {
61:
62: private static final Logger LOG = LoggerFactory.getLogger(RepositoryConnectorInitializer.class);
63:
64: private static final String[] KNOWN_REMOTE_SCHEMES = {"http", "https", "ftp"};
65: private static final String LOCAL_NATIVE_REPO = "repositories/";
66: private static final String FILE_SCHEME = "file";
67: private static final String CLASSPATH_PREFIX = "classpath:";
68:
69: private final DriverConfiguration configuration;
70: private final int maxReconnectAttempts;
71:
72: private RepositoryManager manager;
73: private Repository repository;
74:
75: public RepositoryConnectorInitializer(DriverConfiguration configuration) throws Rdf4jDriverException {
76: this.configuration = configuration;
77: this.maxReconnectAttempts = resolveMaxReconnectAttempts();
78: }
79:
80: private int resolveMaxReconnectAttempts() throws Rdf4jDriverException {
81: try {
82:• final int attempts = configuration.isSet(Rdf4jConfigParam.RECONNECT_ATTEMPTS) ? Integer.parseInt(
83: configuration.getProperty(Rdf4jConfigParam.RECONNECT_ATTEMPTS)) :
84: Constants.DEFAULT_RECONNECT_ATTEMPTS_COUNT;
85:• if (attempts < 0) {
86: throw invalidReconnectAttemptsConfig();
87: }
88: return attempts;
89: } catch (NumberFormatException e) {
90: throw invalidReconnectAttemptsConfig();
91: }
92: }
93:
94: private static Rdf4jDriverException invalidReconnectAttemptsConfig() {
95: return new Rdf4jDriverException(
96: "Invalid value of configuration parameter " + Rdf4jOntoDriverProperties.RECONNECT_ATTEMPTS +
97: ". Must be a non-negative integer.");
98: }
99:
100: public void initializeRepository() throws Rdf4jDriverException {
101: final URI serverUri = configuration.getStorageProperties().getPhysicalURI();
102: LOG.debug("Initializing connector to repository at {}", serverUri);
103: try {
104: final boolean isRemote = isRemoteRepository(serverUri);
105:• if (isRemote) {
106: this.repository = connectToRemoteRepository(serverUri.toString());
107: } else {
108: this.repository = createLocalRepository();
109: }
110: verifyRepositoryCreated(serverUri, isRemote);
111: repository.init();
112: } catch (RepositoryException | RepositoryConfigException e) {
113: throw new Rdf4jDriverException("Failed to acquire RDF4J repository connection.", e);
114: }
115: }
116:
117: private static boolean isRemoteRepository(URI uri) {
118: final String scheme = uri.getScheme();
119:• for (String s : KNOWN_REMOTE_SCHEMES) {
120:• if (s.equals(scheme)) {
121: return true;
122: }
123: }
124: return false;
125: }
126:
127: private Repository connectToRemoteRepository(String repoUri) {
128: this.manager = RepositoryProvider.getRepositoryManagerOfRepository(repoUri);
129: final RemoteRepositoryManager remoteManager = (RemoteRepositoryManager) manager;
130: final String username = configuration.getStorageProperties().getUsername();
131:• if (username != null) {
132: final String password = configuration.getStorageProperties().getPassword();
133: remoteManager.setUsernameAndPassword(username, password);
134: }
135: return connectToRemote(repoUri, 1);
136: }
137:
138: private Repository connectToRemote(String repoUri, int attempts) {
139: try {
140: return new RemoteRepositoryWrapper((HTTPRepository) manager.getRepository(RepositoryProvider.getRepositoryIdOfRepository(repoUri)), configuration);
141: } catch (RepositoryException e) {
142:• if (attempts < maxReconnectAttempts) {
143: LOG.warn("Unable to connect to repository {}. Error is: {}. Retrying...", repoUri, e.getMessage());
144: return connectToRemote(repoUri, attempts + 1);
145: }
146: LOG.error("Threshold of failed connection attempts reached, throwing exception.");
147: throw e;
148: }
149: }
150:
151: private Repository createLocalRepository() {
152:• if (configuration.isSet(Rdf4jConfigParam.REPOSITORY_CONFIG)) {
153: return createRepositoryFromConfig();
154: }
155: final URI localUri = configuration.getStorageProperties().getPhysicalURI();
156:• if (!isFileUri(localUri) && configuration.is(Rdf4jConfigParam.USE_VOLATILE_STORAGE)) {
157: return createInMemoryRepository();
158: } else {
159: return createNativeRepository(configuration, localUri.toString());
160: }
161: }
162:
163: private Repository createRepositoryFromConfig() {
164: LOG.trace("Creating local repository from repository config file.");
165: final RepositoryConfig repoConfig = loadRepositoryConfig();
166: this.manager = RepositoryProvider.getRepositoryManager(getRepositoryManagerBaseDir().orElse(""));
167: manager.addRepositoryConfig(repoConfig);
168: return manager.getRepository(getRepositoryId());
169: }
170:
171: @SuppressWarnings("deprecated")
172: private RepositoryConfig loadRepositoryConfig() {
173: try (final InputStream is = getConfigFileContent()) {
174: final Model configModel = Rio.parse(is, "", RDFFormat.TURTLE);
175: Set<Resource> resources =
176: configModel.filter(null, RDF.TYPE, CONFIG.Rep.Repository).subjects();
177:• if (resources.isEmpty()) {
178: // Support for legacy repository configuration vocabulary.
179: // https://rdf4j.org/documentation/reference/configuration/#migrating-old-configurations
180: resources = configModel.filter(null, RDF.TYPE, RepositoryConfigSchema.REPOSITORY).subjects();
181: }
182:• assert resources.size() == 1;
183: return RepositoryConfig.create(configModel, resources.iterator().next());
184: } catch (IOException e) {
185: throw new RepositoryCreationException("Unable to create repository from the specified configuration.", e);
186: }
187: }
188:
189: private InputStream getConfigFileContent() {
190: final String configPath = configuration.getProperty(Rdf4jConfigParam.REPOSITORY_CONFIG);
191: LOG.trace("Loading repository configuration file content from {}.", configPath);
192:• if (configPath.startsWith(CLASSPATH_PREFIX)) {
193: final InputStream is =
194: getClass().getClassLoader().getResourceAsStream(configPath.substring(CLASSPATH_PREFIX.length()));
195:• if (is == null) {
196: throw new RepositoryCreationException(
197: "Unable to find repository configuration file on classpath location " + configPath);
198: }
199: return is;
200: } else {
201: try {
202: return new FileInputStream(configPath);
203: } catch (FileNotFoundException e) {
204: throw new RepositoryCreationException("Unable to find repository configuration file at " + configPath,
205: e);
206: }
207: }
208: }
209:
210: private Optional<String> getRepositoryManagerBaseDir() {
211: final String physicalUri = configuration.getStorageProperties().getPhysicalURI().toString();
212: final String[] tmp = physicalUri.split(LOCAL_NATIVE_REPO);
213:• return tmp.length == 2 ? Optional.of(tmp[0]) : Optional.empty();
214: }
215:
216: private String getRepositoryId() {
217: final String physicalUri = configuration.getStorageProperties().getPhysicalURI().toString();
218: final String[] tmp = physicalUri.split(LOCAL_NATIVE_REPO);
219:• if (tmp.length != 2) {
220: return physicalUri;
221: }
222: String repoId = tmp[1];
223: // Get rid of the trailing slash if necessary
224:• return repoId.charAt(repoId.length() - 1) == '/' ? repoId.substring(0, repoId.length() - 1) : repoId;
225: }
226:
227: private static boolean isFileUri(URI uri) {
228:• return uri.getScheme() != null && uri.getScheme().equals(FILE_SCHEME);
229: }
230:
231: /**
232: * Creates a local in-memory RDF4J repository which is disposed of when the VM shuts down.
233: */
234: private Repository createInMemoryRepository() {
235: LOG.trace("Creating local in-memory repository.");
236: final MemoryStore ms = new MemoryStore();
237:• if (configuration.is(Rdf4jConfigParam.USE_INFERENCE)) {
238: return new SailRepository(new SchemaCachingRDFSInferencer(ms));
239: } else {
240: return new SailRepository(ms);
241: }
242: }
243:
244: /**
245: * Creates native repository.
246: * <p>
247: * This kind of repository stores data in files and is persistent after the VM shuts down.
248: */
249: private Repository createNativeRepository(DriverConfiguration configuration, String localUri) {
250: LOG.trace("Creating local native repository at " + localUri);
251: validateNativeStorePath(localUri);
252: try {
253: this.manager = RepositoryProvider.getRepositoryManagerOfRepository(localUri);
254: final String repoId = getRepositoryId();
255: final RepositoryConfig cfg = createLocalNativeRepositoryConfig(repoId, configuration);
256: manager.addRepositoryConfig(cfg);
257: return manager.getRepository(repoId);
258: } catch (RepositoryConfigException | RepositoryException e) {
259: throw new RepositoryCreationException("Unable to create local repository at " + localUri, e);
260: }
261: }
262:
263: private static void validateNativeStorePath(String path) {
264:• if (path.split(LOCAL_NATIVE_REPO).length != 2) {
265: throw new RepositoryCreationException(
266: "Unsupported local RDF4J repository path. Expected file://path/repositories/id but got " +
267: path);
268: }
269: }
270:
271: private static RepositoryConfig createLocalNativeRepositoryConfig(String repoId,
272: DriverConfiguration configuration) {
273: SailImplConfig backend = new NativeStoreConfig();
274:• if (configuration.is(Rdf4jConfigParam.USE_INFERENCE)) {
275: backend = new SchemaCachingRDFSInferencerConfig(backend);
276: }
277: final SailRepositoryConfig repoType = new SailRepositoryConfig(backend);
278: return new RepositoryConfig(repoId, repoType);
279: }
280:
281: private void verifyRepositoryCreated(URI serverUri, boolean isRemote) {
282:• if (repository == null) {
283:• if (isRemote) {
284: throw new RepositoryNotFoundException("Unable to reach repository at " + serverUri);
285: } else {
286: throw new RepositoryCreationException("Unable to create local repository at " + serverUri);
287: }
288: }
289: }
290:
291: public RepositoryManager getManager() {
292: return manager;
293: }
294:
295: public Repository getRepository() {
296: return repository;
297: }
298:
299: public int getMaxReconnectAttempts() {
300: return maxReconnectAttempts;
301: }
302: }