You are viewing a plain text version of this content. The canonical link for it is here.
Posted to oak-commits@jackrabbit.apache.org by re...@apache.org on 2026/03/21 10:37:16 UTC
(jackrabbit-oak) 01/01: OAK-8840: copy jackrabbit-data into Oak: copy java code (not tests)
This is an automated email from the ASF dual-hosted git repository.
reschke pushed a commit to branch OAK-8840
in repository https://gitbox.apache.org/repos/asf/jackrabbit-oak.git
commit 675d7043407546c45a1ab44c260a44ffe41773af
Author: Julian Reschke <ju...@gmx.de>
AuthorDate: Sat Mar 21 11:36:50 2026 +0100
OAK-8840: copy jackrabbit-data into Oak: copy java code (not tests)
---
oak-blob/pom.xml | 14 +-
.../core/config/ConfigurationException.java | 45 +
.../jackrabbit/core/config/DataSourceConfig.java | 225 ++++
.../jackrabbit/core/config/package-info.java | 19 +
.../jackrabbit/core/data/AbstractBackend.java | 191 +++
.../jackrabbit/core/data/AbstractDataRecord.java | 94 ++
.../jackrabbit/core/data/AbstractDataStore.java | 147 ++
.../jackrabbit/core/data/AsyncTouchCallback.java | 41 +
.../jackrabbit/core/data/AsyncTouchResult.java | 50 +
.../jackrabbit/core/data/AsyncUploadCache.java | 351 +++++
.../core/data/AsyncUploadCacheResult.java | 70 +
.../jackrabbit/core/data/AsyncUploadCallback.java | 40 +
.../jackrabbit/core/data/AsyncUploadResult.java | 62 +
.../org/apache/jackrabbit/core/data/Backend.java | 186 +++
.../core/data/BackendResourceAbortable.java | 42 +
.../jackrabbit/core/data/CachingDataRecord.java | 62 +
.../jackrabbit/core/data/CachingDataStore.java | 1399 ++++++++++++++++++++
.../apache/jackrabbit/core/data/CachingFDS.java | 51 +
.../jackrabbit/core/data/DataIdentifier.java | 80 ++
.../apache/jackrabbit/core/data/DataRecord.java | 63 +
.../org/apache/jackrabbit/core/data/DataStore.java | 154 +++
.../jackrabbit/core/data/DataStoreException.java | 56 +
.../jackrabbit/core/data/DataStoreFactory.java | 43 +
.../org/apache/jackrabbit/core/data/FSBackend.java | 477 +++++++
.../jackrabbit/core/data/FileDataRecord.java | 71 +
.../apache/jackrabbit/core/data/FileDataStore.java | 500 +++++++
.../jackrabbit/core/data/LazyFileInputStream.java | 167 +++
.../apache/jackrabbit/core/data/LocalCache.java | 668 ++++++++++
.../jackrabbit/core/data/MultiDataStore.java | 722 ++++++++++
.../jackrabbit/core/data/MultiDataStoreAware.java | 40 +
.../jackrabbit/core/data/ScanEventListener.java | 26 +
.../jackrabbit/core/data/db/DbDataRecord.java | 71 +
.../jackrabbit/core/data/db/DbDataStore.java | 1016 ++++++++++++++
.../jackrabbit/core/data/db/DbInputStream.java | 191 +++
.../jackrabbit/core/data/db/DerbyDataStore.java | 53 +
.../data/db/ResettableTempFileInputStream.java | 64 +
.../core/data/db/TempFileInputStream.java | 57 +
.../jackrabbit/core/data/db/package-info.java | 19 +
.../apache/jackrabbit/core/data/package-info.java | 19 +
.../core/data/util/NamedThreadFactory.java | 44 +
.../jackrabbit/core/data/util/package-info.java | 19 +
.../apache/jackrabbit/core/fs/BasedFileSystem.java | 187 +++
.../org/apache/jackrabbit/core/fs/FileSystem.java | 206 +++
.../jackrabbit/core/fs/FileSystemException.java | 49 +
.../jackrabbit/core/fs/FileSystemFactory.java | 40 +
.../jackrabbit/core/fs/FileSystemPathUtil.java | 229 ++++
.../jackrabbit/core/fs/FileSystemResource.java | 226 ++++
.../core/fs/RandomAccessOutputStream.java | 40 +
.../apache/jackrabbit/core/fs/local/FileUtil.java | 99 ++
.../jackrabbit/core/fs/local/HandleMonitor.java | 217 +++
.../jackrabbit/core/fs/local/LocalFileSystem.java | 388 ++++++
.../jackrabbit/core/fs/local/package-info.java | 19 +
.../apache/jackrabbit/core/fs/package-info.java | 19 +
.../core/util/db/CheckSchemaOperation.java | 113 ++
.../jackrabbit/core/util/db/ConnectionFactory.java | 382 ++++++
.../jackrabbit/core/util/db/ConnectionHelper.java | 608 +++++++++
.../jackrabbit/core/util/db/DataSourceWrapper.java | 119 ++
.../jackrabbit/core/util/db/DatabaseAware.java | 30 +
.../apache/jackrabbit/core/util/db/DbUtility.java | 98 ++
.../core/util/db/DerbyConnectionHelper.java | 97 ++
.../core/util/db/Oracle10R1ConnectionHelper.java | 165 +++
.../core/util/db/OracleConnectionHelper.java | 82 ++
.../core/util/db/PostgreSQLConnectionHelper.java | 35 +
.../jackrabbit/core/util/db/ResultSetWrapper.java | 70 +
.../jackrabbit/core/util/db/StreamWrapper.java | 106 ++
.../jackrabbit/core/util/db/package-info.java | 19 +
.../jackrabbit/data/core/InternalXAResource.java | 72 +
.../jackrabbit/data/core/TransactionContext.java | 400 ++++++
.../jackrabbit/data/core/TransactionException.java | 44 +
.../apache/jackrabbit/data/core/package-info.java | 19 +
70 files changed, 11883 insertions(+), 4 deletions(-)
diff --git a/oak-blob/pom.xml b/oak-blob/pom.xml
index 70a8a5aa0f..4ab1e98887 100644
--- a/oak-blob/pom.xml
+++ b/oak-blob/pom.xml
@@ -81,15 +81,16 @@
<artifactId>jcr</artifactId>
<version>2.0</version>
</dependency>
+
+ <!-- Dependencies to other Oak components -->
<dependency>
<groupId>org.apache.jackrabbit</groupId>
- <artifactId>jackrabbit-data</artifactId>
- <version>${jackrabbit.version}</version>
+ <artifactId>oak-commons</artifactId>
+ <version>${project.version}</version>
</dependency>
- <!-- Dependencies to other Oak components -->
<dependency>
<groupId>org.apache.jackrabbit</groupId>
- <artifactId>oak-commons</artifactId>
+ <artifactId>oak-jackrabbit-api</artifactId>
<version>${project.version}</version>
</dependency>
@@ -107,6 +108,11 @@
<groupId>commons-codec</groupId>
<artifactId>commons-codec</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.commons</groupId>
+ <artifactId>commons-dbcp2</artifactId>
+ <version>2.14.0</version>
+ </dependency>
<!-- Optional dependency for jclouds s3 to enable s3 cloud store -->
<dependency>
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/config/ConfigurationException.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/config/ConfigurationException.java
new file mode 100644
index 0000000000..2e10f7d712
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/config/ConfigurationException.java
@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.jackrabbit.core.config;
+
+import javax.jcr.RepositoryException;
+
+/**
+ * Exception class used for configuration errors.
+ */
+public class ConfigurationException extends RepositoryException {
+
+ /**
+ * Creates a configuration exception.
+ *
+ * @param message configuration message
+ */
+ public ConfigurationException(String message) {
+ super(message);
+ }
+
+ /**
+ * Creates a configuration exception that is caused by another exception.
+ *
+ * @param message configuration error message
+ * @param cause root cause of the configuration error
+ */
+ public ConfigurationException(String message, Exception cause) {
+ super(message, cause);
+ }
+
+}
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/config/DataSourceConfig.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/config/DataSourceConfig.java
new file mode 100644
index 0000000000..6df55b73e7
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/config/DataSourceConfig.java
@@ -0,0 +1,225 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.jackrabbit.core.config;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Properties;
+
+import javax.naming.Context;
+
+/**
+ * This class contains list of definitions for {@code DataSource} instances.
+ */
+public class DataSourceConfig {
+
+ public static final String DRIVER = "driver";
+
+ public static final String URL = "url";
+
+ public static final String USER = "user";
+
+ public static final String PASSWORD = "password";
+
+ public static final String DB_TYPE = "databaseType";
+
+ public static final String VALIDATION_QUERY = "validationQuery";
+
+ public static final String MAX_POOL_SIZE = "maxPoolSize";
+
+ private final List<DataSourceDefinition> defs = new ArrayList<DataSourceDefinition>();
+
+ /**
+ * Adds a DataSourceDefinition from the given properties.
+ *
+ * @param props the properties (key and values must be strings)
+ * @throws ConfigurationException on error
+ */
+ public void addDataSourceDefinition(String name, Properties props) throws ConfigurationException {
+ DataSourceDefinition def = new DataSourceDefinition(name, props);
+ for (DataSourceDefinition existing : defs) {
+ if (existing.getLogicalName().equals(def.getLogicalName())) {
+ throw new ConfigurationException("Duplicate logicalName for a DataSource: "
+ + def.getLogicalName());
+ }
+ }
+ defs.add(def);
+ }
+
+ /**
+ * @return the unmodifiable list of the current {@link DataSourceDefinition}s
+ */
+ public List<DataSourceDefinition> getDefinitions() {
+ return Collections.unmodifiableList(defs);
+ }
+
+ /**
+ * The definition of a DataSource.
+ */
+ public static final class DataSourceDefinition {
+
+ private static final List<String> allPropNames =
+ Arrays.asList(DRIVER, URL, USER, PASSWORD, DB_TYPE, VALIDATION_QUERY, MAX_POOL_SIZE);
+
+ private static final List<String> allJndiPropNames =
+ Arrays.asList(DRIVER, URL, USER, PASSWORD, DB_TYPE);
+
+ private final String logicalName;
+
+ private final String driver;
+
+ private final String url;
+
+ private final String user;
+
+ private final String password;
+
+ private final String dbType;
+
+ private final String validationQuery;
+
+ private final int maxPoolSize;
+
+ /**
+ * Creates a DataSourceDefinition from the given properties and
+ * throws a {@link ConfigurationException} when the set of properties does not
+ * satisfy some validity constraints.
+ *
+ * @param name the logical name of the data source
+ * @param props the properties (string keys and values)
+ * @throws ConfigurationException on error
+ */
+ public DataSourceDefinition(String name, Properties props) throws ConfigurationException {
+ this.logicalName = name;
+ this.driver = (String) props.getProperty(DRIVER);
+ this.url = (String) props.getProperty(URL);
+ this.user = (String) props.getProperty(USER);
+ this.password = (String) props.getProperty(PASSWORD);
+ this.dbType = (String) props.getProperty(DB_TYPE);
+ this.validationQuery = (String) props.getProperty(VALIDATION_QUERY);
+ try {
+ this.maxPoolSize = Integer.parseInt((String) props.getProperty(MAX_POOL_SIZE, "-1"));
+ } catch (NumberFormatException e) {
+ throw new ConfigurationException("failed to parse " + MAX_POOL_SIZE
+ + " property for DataSource " + logicalName);
+ }
+ verify(props);
+ }
+
+ private void verify(Properties props) throws ConfigurationException {
+ // Check required properties
+ if (logicalName == null || "".equals(logicalName.trim())) {
+ throw new ConfigurationException("DataSource logical name must not be null or empty");
+ }
+ if (driver == null || "".equals(driver)) {
+ throw new ConfigurationException("DataSource driver must not be null or empty");
+ }
+ if (url == null || "".equals(url)) {
+ throw new ConfigurationException("DataSource URL must not be null or empty");
+ }
+ if (dbType == null || "".equals(dbType)) {
+ throw new ConfigurationException("DataSource databaseType must not be null or empty");
+ }
+ // Check unknown properties
+ for (Object propName : props.keySet()) {
+ if (!allPropNames.contains((String) propName)) {
+ throw new ConfigurationException("Unknown DataSource property: " + propName);
+ }
+ }
+ // Check JNDI config:
+ if (isJndiConfig()) {
+ for (Object propName : props.keySet()) {
+ if (!allJndiPropNames.contains((String) propName)) {
+ throw new ConfigurationException("Property " + propName
+ + " is not allowed for a DataSource obtained through JNDI"
+ + ", DataSource logicalName = " + logicalName);
+ }
+ }
+ }
+ }
+
+ private boolean isJndiConfig() throws ConfigurationException {
+ Class<?> driverClass = null;
+ try {
+ if (driver.length() > 0) {
+ driverClass = Class.forName(driver);
+ }
+ } catch (ClassNotFoundException e) {
+ throw new ConfigurationException("Could not load JDBC driver class " + driver, e);
+ }
+ return driverClass != null && Context.class.isAssignableFrom(driverClass);
+ }
+
+ /**
+ * @return the logicalName
+ */
+ public String getLogicalName() {
+ return logicalName;
+ }
+
+ /**
+ * @return the driver
+ */
+ public String getDriver() {
+ return driver;
+ }
+
+ /**
+ * @return the url
+ */
+ public String getUrl() {
+ return url;
+ }
+
+ /**
+ * @return the user
+ */
+ public String getUser() {
+ return user;
+ }
+
+ /**
+ * @return the dbType
+ */
+ public String getDbType() {
+ return dbType;
+ }
+
+ /**
+ * @return the password
+ */
+ public String getPassword() {
+ return password;
+ }
+
+ /**
+ * @return the validationQuery
+ */
+ public String getValidationQuery() {
+ return validationQuery;
+ }
+
+ /**
+ * @return the maxPoolSize
+ */
+ public int getMaxPoolSize() {
+ return maxPoolSize;
+ }
+ }
+}
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/config/package-info.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/config/package-info.java
new file mode 100755
index 0000000000..0db1e22b55
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/config/package-info.java
@@ -0,0 +1,19 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+/* see JCR-4060 */
+@org.osgi.annotation.versioning.Version("2.13.5")
+package org.apache.jackrabbit.core.config;
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/data/AbstractBackend.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/AbstractBackend.java
new file mode 100644
index 0000000000..3258962cb9
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/AbstractBackend.java
@@ -0,0 +1,191 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.jackrabbit.core.data;
+
+import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadPoolExecutor;
+
+import org.apache.jackrabbit.core.data.util.NamedThreadFactory;
+
+/**
+ * Abstract Backend which has a reference to the underlying {@link CachingDataStore} and is
+ * maintaining the lifecycle of the internal asynchronous write executor.
+ */
+public abstract class AbstractBackend implements Backend {
+
+ /**
+ * {@link CachingDataStore} instance using this backend.
+ */
+ private CachingDataStore dataStore;
+
+ /**
+ * path of repository home dir.
+ */
+ private String homeDir;
+
+ /**
+ * path of config property file.
+ */
+ private String config;
+
+ /**
+ * The pool size of asynchronous write pooling executor.
+ */
+ private int asyncWritePoolSize = 10;
+
+ /**
+ * Asynchronous write pooling executor.
+ */
+ private volatile Executor asyncWriteExecutor;
+
+ /**
+ * Returns the pool size of the asynchronous write pool executor.
+ * @return the pool size of the asynchronous write pool executor
+ */
+ public int getAsyncWritePoolSize() {
+ return asyncWritePoolSize;
+ }
+
+ /**
+ * Sets the pool size of the asynchronous write pool executor.
+ * @param asyncWritePoolSize pool size of the async write pool executor
+ */
+ public void setAsyncWritePoolSize(int asyncWritePoolSize) {
+ this.asyncWritePoolSize = asyncWritePoolSize;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void init(CachingDataStore dataStore, String homeDir, String config) throws DataStoreException {
+ this.dataStore = dataStore;
+ this.homeDir = homeDir;
+ this.config = config;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void close() throws DataStoreException {
+ Executor asyncExecutor = getAsyncWriteExecutor();
+
+ if (asyncExecutor != null && asyncExecutor instanceof ExecutorService) {
+ ((ExecutorService) asyncExecutor).shutdownNow();
+ }
+ }
+
+ /**
+ * Returns the {@link CachingDataStore} instance using this backend.
+ * @return the {@link CachingDataStore} instance using this backend
+ */
+ protected CachingDataStore getDataStore() {
+ return dataStore;
+ }
+
+ /**
+ * Sets the {@link CachingDataStore} instance using this backend.
+ * @param dataStore the {@link CachingDataStore} instance using this backend
+ */
+ protected void setDataStore(CachingDataStore dataStore) {
+ this.dataStore = dataStore;
+ }
+
+ /**
+ * Returns path of repository home dir.
+ * @return path of repository home dir
+ */
+ protected String getHomeDir() {
+ return homeDir;
+ }
+
+ /**
+ * Sets path of repository home dir.
+ * @param homeDir path of repository home dir
+ */
+ protected void setHomeDir(String homeDir) {
+ this.homeDir = homeDir;
+ }
+
+ /**
+ * Returns path of config property file.
+ * @return path of config property file
+ */
+ protected String getConfig() {
+ return config;
+ }
+
+ /**
+ * Sets path of config property file.
+ * @param config path of config property file
+ */
+ protected void setConfig(String config) {
+ this.config = config;
+ }
+
+ /**
+ * Returns Executor used to execute asynchronous write or touch jobs.
+ * @return Executor used to execute asynchronous write or touch jobs
+ */
+ protected Executor getAsyncWriteExecutor() {
+ Executor executor = asyncWriteExecutor;
+
+ if (executor == null) {
+ synchronized (this) {
+ executor = asyncWriteExecutor;
+ if (executor == null) {
+ asyncWriteExecutor = executor = createAsyncWriteExecutor();
+ }
+ }
+ }
+
+ return executor;
+ }
+
+ /**
+ * Creates an {@link Executor}.
+ * This method is invoked during the initialization for asynchronous write/touch job executions.
+ * @return an {@link Executor}
+ */
+ protected Executor createAsyncWriteExecutor() {
+ Executor asyncExecutor;
+
+ if (dataStore.getAsyncUploadLimit() > 0 && getAsyncWritePoolSize() > 0) {
+ asyncExecutor = (ThreadPoolExecutor) Executors.newFixedThreadPool(getAsyncWritePoolSize(),
+ new NamedThreadFactory(getClass().getSimpleName() + "-write-worker"));
+ } else {
+ asyncExecutor = new ImmediateExecutor();
+ }
+
+ return asyncExecutor;
+ }
+
+ /**
+ * This class implements {@link Executor} interface to run {@code command} right away,
+ * resulting in non-asynchronous mode executions.
+ */
+ private class ImmediateExecutor implements Executor {
+ @Override
+ public void execute(Runnable command) {
+ command.run();
+ }
+ }
+
+}
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/data/AbstractDataRecord.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/AbstractDataRecord.java
new file mode 100644
index 0000000000..3c7f7d9edf
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/AbstractDataRecord.java
@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.jackrabbit.core.data;
+
+
+/**
+ * Abstract data record base class. This base class contains only
+ * a reference to the data identifier of the record and implements
+ * the standard {@link Object} equality, hash code, and string
+ * representation methods based on the identifier.
+ */
+public abstract class AbstractDataRecord implements DataRecord {
+
+ /**
+ * The data store that contains this record.
+ */
+ private final AbstractDataStore store;
+
+ /**
+ * The binary identifier;
+ */
+ private final DataIdentifier identifier;
+
+ /**
+ * Creates a data record with the given identifier.
+ *
+ * @param identifier data identifier
+ */
+ public AbstractDataRecord(
+ AbstractDataStore store, DataIdentifier identifier) {
+ this.store = store;
+ this.identifier = identifier;
+ }
+
+ /**
+ * Returns the data identifier.
+ *
+ * @return data identifier
+ */
+ public DataIdentifier getIdentifier() {
+ return identifier;
+ }
+
+ public String getReference() {
+ return store.getReferenceFromIdentifier(identifier);
+ }
+
+ /**
+ * Returns the string representation of the data identifier.
+ *
+ * @return string representation
+ */
+ public String toString() {
+ return identifier.toString();
+ }
+
+ /**
+ * Checks if the given object is a data record with the same identifier
+ * as this one.
+ *
+ * @param object other object
+ * @return <code>true</code> if the other object is a data record and has
+ * the same identifier as this one, <code>false</code> otherwise
+ */
+ public boolean equals(Object object) {
+ return (object instanceof DataRecord)
+ && identifier.equals(((DataRecord) object).getIdentifier());
+ }
+
+ /**
+ * Returns the hash code of the data identifier.
+ *
+ * @return hash code
+ */
+ public int hashCode() {
+ return identifier.hashCode();
+ }
+
+}
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/data/AbstractDataStore.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/AbstractDataStore.java
new file mode 100644
index 0000000000..21f852beb6
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/AbstractDataStore.java
@@ -0,0 +1,147 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.jackrabbit.core.data;
+
+import java.nio.charset.StandardCharsets;
+import java.security.SecureRandom;
+
+import javax.crypto.Mac;
+import javax.crypto.spec.SecretKeySpec;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public abstract class AbstractDataStore implements DataStore {
+
+ private static Logger LOG = LoggerFactory.getLogger(AbstractDataStore.class);
+
+ private static final String ALGORITHM = "HmacSHA1";
+
+ /**
+ * Array of hexadecimal digits.
+ */
+ private static final char[] HEX = "0123456789abcdef".toCharArray();
+
+ /**
+ * The digest algorithm used to uniquely identify records.
+ */
+ protected String DIGEST = System.getProperty("ds.digest.algorithm", "SHA-256");
+
+ /**
+ * Cached copy of the reference key of this data store. Initialized in
+ * {@link #getReferenceKey()} when the key is first accessed.
+ */
+ private byte[] referenceKey = null;
+
+ //---------------------------------------------------------< DataStore >--
+
+ public DataRecord getRecord(DataIdentifier identifier)
+ throws DataStoreException {
+ DataRecord record = getRecordIfStored(identifier);
+ if (record != null) {
+ return record;
+ } else {
+ throw new DataStoreException(
+ "Record " + identifier + " does not exist");
+ }
+ }
+
+ public DataRecord getRecordFromReference(String reference)
+ throws DataStoreException {
+ if (reference != null) {
+ int colon = reference.indexOf(':');
+ if (colon != -1) {
+ DataIdentifier identifier =
+ new DataIdentifier(reference.substring(0, colon));
+ if (reference.equals(getReferenceFromIdentifier(identifier))) {
+ return getRecordIfStored(identifier);
+ }
+ }
+ }
+ return null;
+ }
+
+ //---------------------------------------------------------< protected >--
+
+ /**
+ * Returns the hex encoding of the given bytes.
+ *
+ * @param value value to be encoded
+ * @return encoded value
+ */
+ protected static String encodeHexString(byte[] value) {
+ char[] buffer = new char[value.length * 2];
+ for (int i = 0; i < value.length; i++) {
+ buffer[2 * i] = HEX[(value[i] >> 4) & 0x0f];
+ buffer[2 * i + 1] = HEX[value[i] & 0x0f];
+ }
+ return new String(buffer);
+ }
+
+ protected String getReferenceFromIdentifier(DataIdentifier identifier) {
+ try {
+ String id = identifier.toString();
+
+ Mac mac = Mac.getInstance(ALGORITHM);
+ mac.init(new SecretKeySpec(getReferenceKey(), ALGORITHM));
+ byte[] hash = mac.doFinal(id.getBytes(StandardCharsets.UTF_8));
+
+ return id + ':' + encodeHexString(hash);
+ } catch (Exception e) {
+ LOG.error("Failed to hash identifier using MAC (Message Authentication Code) algorithm.", e);
+ }
+ return null;
+ }
+
+ /**
+ * Returns the reference key of this data store. If one does not already
+ * exist, it is automatically created in an implementation-specific way.
+ * The default implementation simply creates a temporary random key that's
+ * valid only until the data store gets restarted. Subclasses can override
+ * and/or decorate this method to support a more persistent reference key.
+ * <p>
+ * This method is called only once during the lifetime of a data store
+ * instance and the return value is cached in memory, so it's no problem
+ * if the implementation is slow.
+ *
+ * @return reference key
+ * @throws DataStoreException if the key is not available
+ */
+ protected byte[] getOrCreateReferenceKey() throws DataStoreException {
+ byte[] referenceKeyValue = new byte[256];
+ new SecureRandom().nextBytes(referenceKeyValue);
+ return referenceKeyValue;
+ }
+
+ //-----------------------------------------------------------< private >--
+
+ /**
+ * Returns the reference key of this data store. Synchronized to
+ * control concurrent access to the cached {@link #referenceKey} value.
+ *
+ * @return reference key
+ * @throws DataStoreException if the key is not available
+ */
+ private synchronized byte[] getReferenceKey() throws DataStoreException {
+ if (referenceKey == null) {
+ referenceKey = getOrCreateReferenceKey();
+ }
+ return referenceKey;
+ }
+
+}
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/data/AsyncTouchCallback.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/AsyncTouchCallback.java
new file mode 100644
index 0000000000..2084000b27
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/AsyncTouchCallback.java
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.jackrabbit.core.data;
+/**
+ * This interface defines callback methods to reflect the status of asynchronous
+ * touch.
+ */
+public interface AsyncTouchCallback {
+
+
+ /**
+ * Callback method for successful asynchronous touch.
+ */
+ public void onSuccess(AsyncTouchResult result);
+
+ /**
+ * Callback method for failed asynchronous touch.
+ */
+ public void onFailure(AsyncTouchResult result);
+
+ /**
+ * Callback method for aborted asynchronous touch.
+ */
+ public void onAbort(AsyncTouchResult result);
+
+}
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/data/AsyncTouchResult.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/AsyncTouchResult.java
new file mode 100644
index 0000000000..014db6553a
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/AsyncTouchResult.java
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.jackrabbit.core.data;
+
+/**
+ *
+ * The class holds the result of asynchronous touch to {@link Backend}
+ */
+public class AsyncTouchResult {
+ /**
+ * {@link DataIdentifier} on which asynchronous touch is initiated.
+ */
+ private final DataIdentifier identifier;
+ /**
+ * Any {@link Exception} which is raised in asynchronously touch.
+ */
+ private Exception exception;
+
+ public AsyncTouchResult(DataIdentifier identifier) {
+ super();
+ this.identifier = identifier;
+ }
+
+ public DataIdentifier getIdentifier() {
+ return identifier;
+ }
+
+ public Exception getException() {
+ return exception;
+ }
+
+ public void setException(Exception exception) {
+ this.exception = exception;
+ }
+
+}
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/data/AsyncUploadCache.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/AsyncUploadCache.java
new file mode 100644
index 0000000000..2eeeb9e6d9
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/AsyncUploadCache.java
@@ -0,0 +1,351 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.jackrabbit.core.data;
+
+import java.io.BufferedInputStream;
+import java.io.BufferedOutputStream;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.ObjectInput;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutput;
+import java.io.ObjectOutputStream;
+import java.io.OutputStream;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.commons.io.IOUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This class holds all in progress uploads. This class contains two data
+ * structures, one is {@link #asyncUploadMap} which is {@link Map}
+ * of file path vs lastModified of upload. The second {@link #toBeDeleted} is
+ * {@link Set} of upload which is marked for delete, while it is already
+ * in progress. Before starting an asynchronous upload, it requires to invoke
+ * {@link #add(String)} to add entry to {@link #asyncUploadMap}. After
+ * asynchronous upload completes, it requires to invoke
+ * {@link #remove(String)} to remove entry from
+ * {@link #asyncUploadMap} Any modification to this class are immediately
+ * persisted to local file system. {@link #asyncUploadMap} is persisted to /
+ * {@link homeDir}/ {@link #PENDIND_UPLOAD_FILE}. {@link #toBeDeleted} is
+ * persisted to / {@link homeDir}/ {@link #TO_BE_DELETED_UPLOAD_FILE}. The /
+ * {@link homeDir} refer to ${rep.home}.
+ */
+public class AsyncUploadCache {
+ private static final Logger LOG = LoggerFactory.getLogger(AsyncUploadCache.class);
+
+ /**
+ * {@link Map} of fileName Vs lastModified to store asynchronous upload.
+ */
+ Map<String, Long> asyncUploadMap = new HashMap<String, Long>();
+
+ /**
+ * {@link Set} of fileName which are mark for delete during asynchronous
+ * Upload.
+ */
+ Set<String> toBeDeleted = new HashSet<String>();
+
+ String path;
+
+ String homeDir;
+
+ int asyncUploadLimit;
+
+ private File pendingUploads;
+
+ private File toBeDeletedUploads;
+
+ private static final String PENDIND_UPLOAD_FILE = "async-pending-uploads.ser";
+
+ private static final String TO_BE_DELETED_UPLOAD_FILE = "async-tobedeleted-uploads.ser";
+
+ /**
+ * This methods checks if file can be added to {@link #asyncUploadMap}. If
+ * yes it adds to {@link #asyncUploadMap} and
+ * {@link #serializeAsyncUploadMap()} the {@link #asyncUploadMap} to disk.
+ *
+ * @return {@link AsyncUploadCacheResult} if successfully added to
+ * asynchronous uploads it sets
+ * {@link AsyncUploadCacheResult#setAsyncUpload(boolean)} to true
+ * else sets to false.
+ */
+ public synchronized AsyncUploadCacheResult add(String fileName)
+ throws IOException {
+ AsyncUploadCacheResult result = new AsyncUploadCacheResult();
+ if (asyncUploadMap.entrySet().size() >= asyncUploadLimit) {
+ LOG.info(
+ "Async write limit [{}] reached. File [{}] not added to async write cache.",
+ asyncUploadLimit, fileName);
+ LOG.debug("AsyncUploadCache size=[{}] and entries =[{}]",
+ asyncUploadMap.size(), asyncUploadMap.keySet());
+ result.setAsyncUpload(false);
+ } else {
+ long startTime = System.currentTimeMillis();
+ if (toBeDeleted.remove(fileName)) {
+ serializeToBeDeleted();
+ }
+ asyncUploadMap.put(fileName, System.currentTimeMillis());
+ serializeAsyncUploadMap();
+ LOG.debug("added file [{}] to asyncUploadMap upoad took [{}] sec",
+ fileName, ((System.currentTimeMillis() - startTime) / 1000));
+ LOG.debug("AsyncUploadCache size=[{}] and entries =[{}]",
+ asyncUploadMap.size(), asyncUploadMap.keySet());
+ result.setAsyncUpload(true);
+ }
+ return result;
+ }
+
+ /**
+ * This methods removes file (if found) from {@link #asyncUploadMap}. If
+ * file is found, it immediately serializes the {@link #asyncUploadMap} to
+ * disk. This method sets
+ * {@link AsyncUploadCacheResult#setRequiresDelete(boolean)} to true, if
+ * asynchronous upload found to be in {@link #toBeDeleted} set i.e. marked
+ * for delete.
+ */
+ public synchronized AsyncUploadCacheResult remove(String fileName)
+ throws IOException {
+ long startTime = System.currentTimeMillis();
+ Long retVal = asyncUploadMap.remove(fileName);
+ if (retVal != null) {
+ serializeAsyncUploadMap();
+ LOG.debug("removed file [{}] from asyncUploadMap took [{}] sec",
+ fileName, ((System.currentTimeMillis() - startTime) / 1000));
+ LOG.debug("AsyncUploadCache size=[{}] and entries =[{}]",
+ asyncUploadMap.size(), asyncUploadMap.keySet());
+ } else {
+ LOG.debug("cannot removed file [{}] from asyncUploadMap took [{}] sec. File not found.",
+ fileName, ((System.currentTimeMillis() - startTime) / 1000));
+ LOG.debug("AsyncUploadCache size=[{}] and entries =[{}]",
+ asyncUploadMap.size(), asyncUploadMap.keySet());
+ }
+ AsyncUploadCacheResult result = new AsyncUploadCacheResult();
+ result.setRequiresDelete(toBeDeleted.contains(fileName));
+ return result;
+ }
+
+ /**
+ * This methods returns the in progress asynchronous uploads which are not
+ * marked for delete.
+ */
+ public synchronized Set<String> getAll() {
+ Set<String> retVal = new HashSet<String>();
+ retVal.addAll(asyncUploadMap.keySet());
+ retVal.removeAll(toBeDeleted);
+ return retVal;
+ }
+
+ /**
+ * This methos checks if asynchronous upload is in progress for @param
+ * fileName. If @param touch is true, the lastModified is updated to current
+ * time.
+ */
+ public synchronized boolean hasEntry(String fileName, boolean touch)
+ throws IOException {
+ boolean contains = asyncUploadMap.containsKey(fileName)
+ && !toBeDeleted.contains(fileName);
+ if (touch && contains) {
+ long timeStamp = System.currentTimeMillis();
+ asyncUploadMap.put(fileName, timeStamp);
+ serializeAsyncUploadMap();
+ }
+ return contains;
+ }
+
+ /**
+ * Returns lastModified from {@link #asyncUploadMap} if found else returns
+ * 0.
+ */
+ public synchronized long getLastModified(String fileName) {
+ return asyncUploadMap.get(fileName) != null
+ && !toBeDeleted.contains(fileName)
+ ? asyncUploadMap.get(fileName)
+ : 0;
+ }
+
+ /**
+ * This methods deletes asynchronous upload for @param fileName if there
+ * exists asynchronous upload for @param fileName.
+ */
+ public synchronized void delete(String fileName) throws IOException {
+ boolean serialize = false;
+ if (toBeDeleted.remove(fileName)) {
+ serialize = true;
+ }
+ if (asyncUploadMap.containsKey(fileName) && toBeDeleted.add(fileName)) {
+ serialize = true;
+ }
+ if (serialize) {
+ serializeToBeDeleted();
+ }
+ }
+
+ /**
+ * Delete in progress asynchronous uploads which are older than @param min.
+ * This method leverage lastModified stored in {@link #asyncUploadMap}
+ */
+ public synchronized Set<String> deleteOlderThan(long min)
+ throws IOException {
+ min = min - 1000;
+ LOG.info("deleteOlderThan min [{}]", min);
+ Set<String> deleteSet = new HashSet<String>();
+ for (Map.Entry<String, Long> entry : asyncUploadMap.entrySet()) {
+ if (entry.getValue() < min) {
+ deleteSet.add(entry.getKey());
+ }
+ }
+ if (deleteSet.size() > 0) {
+ LOG.debug("deleteOlderThan set [{}]", deleteSet);
+ toBeDeleted.addAll(deleteSet);
+ serializeToBeDeleted();
+ }
+ return deleteSet;
+ }
+
+ /**
+ * @param homeDir
+ * home directory of repository.
+ * @param path
+ * path of the {@link LocalCache}
+ * @param asyncUploadLimit
+ * the maximum number of asynchronous uploads
+ */
+ public synchronized void init(String homeDir, String path,
+ int asyncUploadLimit) throws IOException, ClassNotFoundException {
+ this.homeDir = homeDir;
+ this.path = path;
+ this.asyncUploadLimit = asyncUploadLimit;
+ LOG.info(
+ "AsynWriteCache:homeDir=[{}], path=[{}], asyncUploadLimit=[{}].",
+ new Object[] { homeDir, path, asyncUploadLimit });
+ pendingUploads = new File(homeDir + "/" + PENDIND_UPLOAD_FILE);
+ toBeDeletedUploads = new File(homeDir + "/" + TO_BE_DELETED_UPLOAD_FILE);
+ if (pendingUploads.exists()) {
+ deserializeAsyncUploadMap();
+ } else {
+ pendingUploads.createNewFile();
+ asyncUploadMap = new HashMap<String, Long>();
+ serializeAsyncUploadMap();
+ }
+
+ if (toBeDeletedUploads.exists()) {
+ deserializeToBeDeleted();
+ } else {
+ toBeDeletedUploads.createNewFile();
+ asyncUploadMap = new HashMap<String, Long>();
+ serializeToBeDeleted();
+ }
+ }
+
+ /**
+ * Reset the {@link AsyncUploadCache} to empty {@link #asyncUploadMap} and
+ * {@link #toBeDeleted}
+ */
+ public synchronized void reset() throws IOException {
+ if (!pendingUploads.exists()) {
+ pendingUploads.createNewFile();
+ }
+ pendingUploads.createNewFile();
+ asyncUploadMap = new HashMap<String, Long>();
+ serializeAsyncUploadMap();
+
+ if (!toBeDeletedUploads.exists()) {
+ toBeDeletedUploads.createNewFile();
+ }
+ toBeDeletedUploads.createNewFile();
+ toBeDeleted = new HashSet<String>();
+ serializeToBeDeleted();
+ }
+
+ /**
+ * Serialize {@link #asyncUploadMap} to local file system.
+ */
+ private synchronized void serializeAsyncUploadMap() throws IOException {
+
+ // use buffering
+ OutputStream fos = new FileOutputStream(pendingUploads);
+ OutputStream buffer = new BufferedOutputStream(fos);
+ ObjectOutput output = new ObjectOutputStream(buffer);
+ try {
+ output.writeObject(asyncUploadMap);
+ output.flush();
+ } finally {
+ output.close();
+ IOUtils.closeQuietly(buffer);
+
+ }
+ }
+
+ /**
+ * Deserialize {@link #asyncUploadMap} from local file system.
+ */
+ private synchronized void deserializeAsyncUploadMap() throws IOException,
+ ClassNotFoundException {
+ // use buffering
+ InputStream fis = new FileInputStream(pendingUploads);
+ InputStream buffer = new BufferedInputStream(fis);
+ ObjectInput input = new ObjectInputStream(buffer);
+ try {
+ asyncUploadMap = (Map<String, Long>) input.readObject();
+ } finally {
+ input.close();
+ IOUtils.closeQuietly(buffer);
+ }
+ }
+
+ /**
+ * Serialize {@link #toBeDeleted} to local file system.
+ */
+ private synchronized void serializeToBeDeleted() throws IOException {
+
+ // use buffering
+ OutputStream fos = new FileOutputStream(toBeDeletedUploads);
+ OutputStream buffer = new BufferedOutputStream(fos);
+ ObjectOutput output = new ObjectOutputStream(buffer);
+ try {
+ output.writeObject(toBeDeleted);
+ output.flush();
+ } finally {
+ output.close();
+ IOUtils.closeQuietly(buffer);
+ }
+ }
+
+ /**
+ * Deserialize {@link #toBeDeleted} from local file system.
+ */
+ private synchronized void deserializeToBeDeleted() throws IOException,
+ ClassNotFoundException {
+ // use buffering
+ InputStream fis = new FileInputStream(toBeDeletedUploads);
+ InputStream buffer = new BufferedInputStream(fis);
+ ObjectInput input = new ObjectInputStream(buffer);
+ try {
+ toBeDeleted = (Set<String>) input.readObject();
+ } finally {
+ input.close();
+ IOUtils.closeQuietly(buffer);
+ }
+ }
+}
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/data/AsyncUploadCacheResult.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/AsyncUploadCacheResult.java
new file mode 100644
index 0000000000..fb5613fc3f
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/AsyncUploadCacheResult.java
@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.jackrabbit.core.data;
+
+import java.io.File;
+
+/**
+ * This class holds result of asynchronous upload from {@link AsyncUploadCache}
+ */
+public class AsyncUploadCacheResult {
+
+ /**
+ * flag to indicate that asynchronous upload can be started on file.
+ */
+ private boolean asyncUpload;
+
+ /**
+ * flag to indicate that cached file requires to be deleted. It is
+ * applicable in case where file marked for delete before asynchronous
+ * upload completes.
+ */
+ private boolean requiresDelete;
+
+ private File file;
+
+ /**
+ * Flag to denote that asynchronous upload can be started on file.
+ */
+ public boolean canAsyncUpload() {
+ return asyncUpload;
+ }
+
+ public void setAsyncUpload(boolean asyncUpload) {
+ this.asyncUpload = asyncUpload;
+ }
+
+ /**
+ * Flag to indicate that record to be deleted from {@link DataStore}.
+ */
+ public boolean doRequiresDelete() {
+ return requiresDelete;
+ }
+
+ public void setRequiresDelete(boolean requiresDelete) {
+ this.requiresDelete = requiresDelete;
+ }
+
+ public File getFile() {
+ return file;
+ }
+
+ public void setFile(File file) {
+ this.file = file;
+ }
+
+}
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/data/AsyncUploadCallback.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/AsyncUploadCallback.java
new file mode 100644
index 0000000000..af18c03249
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/AsyncUploadCallback.java
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.jackrabbit.core.data;
+
+/**
+ * This interface defines callback methods to reflect the status of asynchronous
+ * upload.
+ */
+public interface AsyncUploadCallback {
+
+ /**
+ * Callback method for successful asynchronous upload.
+ */
+ public void onSuccess(AsyncUploadResult result);
+
+ /**
+ * Callback method for failed asynchronous upload.
+ */
+ public void onFailure(AsyncUploadResult result);
+
+ /**
+ * Callback method for aborted asynchronous upload.
+ */
+ public void onAbort(AsyncUploadResult result);
+}
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/data/AsyncUploadResult.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/AsyncUploadResult.java
new file mode 100644
index 0000000000..bdbbb4c1f2
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/AsyncUploadResult.java
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.jackrabbit.core.data;
+
+import java.io.File;
+
+/**
+ *
+ * The class holds the result of asynchronous upload to {@link Backend}
+ */
+public class AsyncUploadResult {
+ /**
+ * {@link DataIdentifier} on which asynchronous upload is initiated.
+ */
+ private final DataIdentifier identifier;
+
+ /**
+ * {@link File} which is asynchronously uploaded.
+ */
+ private final File file;
+
+ /**
+ * Any {@link Exception} which is raised in asynchronously upload.
+ */
+ private Exception exception;
+
+ public AsyncUploadResult(DataIdentifier identifier, File file) {
+ super();
+ this.identifier = identifier;
+ this.file = file;
+ }
+
+ public DataIdentifier getIdentifier() {
+ return identifier;
+ }
+
+ public File getFile() {
+ return file;
+ }
+
+ public Exception getException() {
+ return exception;
+ }
+
+ public void setException(Exception exception) {
+ this.exception = exception;
+ }
+}
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/data/Backend.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/Backend.java
new file mode 100644
index 0000000000..840b52ae70
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/Backend.java
@@ -0,0 +1,186 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.jackrabbit.core.data;
+
+import java.io.File;
+import java.io.InputStream;
+import java.util.Iterator;
+import java.util.Set;
+
+/**
+ * The interface defines the backend which can be plugged into
+ * {@link CachingDataStore}.
+ */
+public interface Backend {
+
+ /**
+ * This method initialize backend with the configuration.
+ *
+ * @param store
+ * {@link CachingDataStore}
+ * @param homeDir
+ * path of repository home dir.
+ * @param config
+ * path of config property file.
+ * @throws DataStoreException
+ */
+ void init(CachingDataStore store, String homeDir, String config)
+ throws DataStoreException;
+
+ /**
+ * Return inputstream of record identified by identifier.
+ *
+ * @param identifier
+ * identifier of record.
+ * @return inputstream of the record.
+ * @throws DataStoreException
+ * if record not found or any error.
+ */
+ InputStream read(DataIdentifier identifier) throws DataStoreException;
+
+ /**
+ * Return length of record identified by identifier.
+ *
+ * @param identifier
+ * identifier of record.
+ * @return length of the record.
+ * @throws DataStoreException
+ * if record not found or any error.
+ */
+ long getLength(DataIdentifier identifier) throws DataStoreException;
+
+ /**
+ * Return lastModified of record identified by identifier.
+ *
+ * @param identifier
+ * identifier of record.
+ * @return lastModified of the record.
+ * @throws DataStoreException
+ * if record not found or any error.
+ */
+ long getLastModified(DataIdentifier identifier) throws DataStoreException;
+
+ /**
+ * Stores file to backend with identifier used as key. If key pre-exists, it
+ * updates the timestamp of the key.
+ *
+ * @param identifier
+ * key of the file
+ * @param file
+ * file that would be stored in backend.
+ * @throws DataStoreException
+ * for any error.
+ */
+ void write(DataIdentifier identifier, File file) throws DataStoreException;
+
+ /**
+ * Write file to backend in asynchronous mode.
+ *
+ * @param identifier
+ * @param file
+ * @param callback
+ * Callback interface to called after upload succeed or failed.
+ * @throws DataStoreException
+ */
+ void writeAsync(DataIdentifier identifier, File file,
+ AsyncUploadCallback callback) throws DataStoreException;
+
+ /**
+ * Returns identifiers of all records that exists in backend.
+ *
+ * @return iterator consisting of all identifiers
+ * @throws DataStoreException
+ */
+ Iterator<DataIdentifier> getAllIdentifiers() throws DataStoreException;
+
+ /**
+ * This method check the existence of record in backend. Return true if
+ * records exists else false. This method also touch record identified by
+ * identifier if touch is true.
+ *
+ * @param identifier
+ * @throws DataStoreException
+ */
+ boolean exists(DataIdentifier identifier, boolean touch)
+ throws DataStoreException;
+
+ /**
+ * This method check the existence of record in backend.
+ *
+ * @param identifier
+ * identifier to be checked.
+ * @return true if records exists else false.
+ * @throws DataStoreException
+ */
+ boolean exists(DataIdentifier identifier) throws DataStoreException;
+
+ /**
+ * Update the lastModified of record if it's lastModified < minModifiedDate.
+ *
+ * @param identifier
+ * @param minModifiedDate
+ * @throws DataStoreException
+ */
+ void touch(final DataIdentifier identifier, long minModifiedDate)
+ throws DataStoreException;
+
+ /**
+ * Update the lastModified of record if it's lastModified < minModifiedDate
+ * asynchronously. Result of update is passed using appropriate
+ * {@link AsyncTouchCallback} methods. If identifier's lastModified >
+ * minModified {@link AsyncTouchCallback#onAbort(AsyncTouchResult)} is
+ * called. Any exception is communicated through
+ * {@link AsyncTouchCallback#onFailure(AsyncTouchResult)} . On successful
+ * update of lastModified,
+ * {@link AsyncTouchCallback#onSuccess(AsyncTouchResult)}
+ * is invoked.
+ *
+ * @param identifier
+ * @param minModifiedDate
+ * @param callback
+ * @throws DataStoreException
+ */
+ void touchAsync(final DataIdentifier identifier, long minModifiedDate,
+ final AsyncTouchCallback callback) throws DataStoreException;
+
+ /**
+ * Close backend and release resources like database connection if any.
+ *
+ * @throws DataStoreException
+ */
+ void close() throws DataStoreException;
+
+ /**
+ * Delete all records which are older than timestamp.
+ *
+ * @param timestamp
+ * @return {@link Set} of identifiers which are deleted.
+ * @throws DataStoreException
+ */
+ Set<DataIdentifier> deleteAllOlderThan(long timestamp)
+ throws DataStoreException;
+
+ /**
+ * Delete record identified by identifier. No-op if identifier not found.
+ *
+ * @param identifier
+ * @throws DataStoreException
+ */
+ void deleteRecord(DataIdentifier identifier) throws DataStoreException;
+}
+
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/data/BackendResourceAbortable.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/BackendResourceAbortable.java
new file mode 100644
index 0000000000..e1b66ceb46
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/BackendResourceAbortable.java
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.jackrabbit.core.data;
+
+import java.io.InputStream;
+
+/**
+ * {@link Backend} resource abstraction, such as {@link InputStream}, which can be aborted without consuming it
+ * fully for efficiency.
+ * <p>
+ * Some {@link Backend} implementations such as <code>S3Backend</code> may return an abortable <code>InputStream</code>
+ * for a more optimal resource use. <code>S3Backend</code> internally uses Apache HttpClient library which tries
+ * to reuse HTTP connections by reading data fully to the end of an attached <code>InputStream</code> on {@link InputStream#close()}
+ * by default. It can be efficient from a socket pool management perspective, but possibly a significant overhead
+ * while bytes are read from S3 just to be discarded. So, a {@link Backend} implementation that retrieves an abortable
+ * resource may decide to wrap the underlying resource (e.g, <code>InputStream</code>) by this interface (e.g,
+ * <code>S3BackendResourceAbortableInputStream</code>) in order to abort the underlying resources (e.g, http request
+ * object) without having to read data fully.
+ */
+public interface BackendResourceAbortable {
+
+ /**
+ * Abort the underlying backend resource(s).
+ */
+ public void abort();
+
+}
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/data/CachingDataRecord.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/CachingDataRecord.java
new file mode 100644
index 0000000000..881911709a
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/CachingDataRecord.java
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.jackrabbit.core.data;
+
+import java.io.InputStream;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * CachingDataRecord which stores reference to {@link CachingDataStore}. This
+ * class doesn't store any references to attributes but attributes are fetched
+ * on demand from {@link CachingDataStore}.
+ */
+public class CachingDataRecord extends AbstractDataRecord {
+
+ private static final Logger LOG = LoggerFactory.getLogger(CachingDataRecord.class);
+
+ private final CachingDataStore store;
+
+ public CachingDataRecord(CachingDataStore store, DataIdentifier identifier) {
+ super(store, identifier);
+ this.store = store;
+ }
+
+ @Override
+ public long getLastModified() {
+ try {
+ return store.getLastModified(getIdentifier());
+ } catch (DataStoreException dse) {
+ LOG.info("exception in getLastModified for identifier ["
+ + getIdentifier() + "]. returning 0.", dse);
+ return 0;
+ }
+ }
+
+ @Override
+ public long getLength() throws DataStoreException {
+ return store.getLength(getIdentifier());
+ }
+
+ @Override
+ public InputStream getStream() throws DataStoreException {
+ return store.getStream(getIdentifier());
+ }
+
+}
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/data/CachingDataStore.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/CachingDataStore.java
new file mode 100644
index 0000000000..a95c2893c4
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/CachingDataStore.java
@@ -0,0 +1,1399 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.jackrabbit.core.data;
+
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.io.UnsupportedEncodingException;
+import java.lang.ref.WeakReference;
+import java.nio.charset.StandardCharsets;
+import java.security.DigestOutputStream;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.WeakHashMap;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+
+import javax.jcr.RepositoryException;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.commons.io.IOUtils;
+import org.apache.jackrabbit.core.data.util.NamedThreadFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A caching data store that consists of {@link LocalCache} and {@link Backend}.
+ * {@link Backend} is single source of truth. All methods first try to fetch
+ * information from {@link LocalCache}. If record is not available in
+ * {@link LocalCache}, then it is fetched from {@link Backend} and saved to
+ * {@link LocalCache} for further access. This class is designed to work without
+ * {@link LocalCache} and then all information is fetched from {@link Backend}.
+ * To disable {@link LocalCache} set {@link #setCacheSize(long)} to 0. *
+ * Configuration:
+ *
+ * <pre>
+ * <DataStore class="org.apache.jackrabbit.aws.ext.ds.CachingDataStore">
+ *
+ * <param name="{@link #setPath(String) path}" value="/data/datastore"/>
+ * <param name="{@link #setConfig(String) config}" value="${rep.home}/backend.properties"/>
+ * <param name="{@link #setCacheSize(long) cacheSize}" value="68719476736"/>
+ * <param name="{@link #setSecret(String) secret}" value="123456"/>
+ * <param name="{@link #setCachePurgeTrigFactor(double)}" value="0.95d"/>
+ * <param name="{@link #setCachePurgeResizeFactor(double) cacheSize}" value="0.85d"/>
+ * <param name="{@link #setMinRecordLength(int) minRecordLength}" value="1024"/>
+ * <param name="{@link #setContinueOnAsyncUploadFailure(boolean) continueOnAsyncUploadFailure}" value="false"/>
+ * <param name="{@link #setConcurrentUploadsThreads(int) concurrentUploadsThreads}" value="10"/>
+ * <param name="{@link #setAsyncUploadLimit(int) asyncUploadLimit}" value="100"/>
+ * <param name="{@link #setUploadRetries(int) uploadRetries}" value="3"/>
+ * <param name="{@link #setTouchAsync(boolean) touchAsync}" value="false"/>
+ * <param name="{@link #setProactiveCaching(boolean) proactiveCaching}" value="true"/>
+ * <param name="{@link #setRecLengthCacheSize(int) recLengthCacheSize}" value="200"/>
+ * </DataStore>
+ * </pre>
+ */
+public abstract class CachingDataStore extends AbstractDataStore implements
+ MultiDataStoreAware, AsyncUploadCallback, AsyncTouchCallback {
+
+ /**
+ * Logger instance.
+ */
+ private static final Logger LOG = LoggerFactory.getLogger(CachingDataStore.class);
+
+ private static final String DS_STORE = ".DS_Store";
+
+ /**
+ * Name of the directory used for temporary files. Must be at least 3
+ * characters.
+ */
+ private static final String TMP = "tmp";
+
+ /**
+ * All data identifiers that are currently in use are in this set until they
+ * are garbage collected.
+ */
+ protected Map<DataIdentifier, WeakReference<DataIdentifier>> inUse = Collections.synchronizedMap(new WeakHashMap<DataIdentifier, WeakReference<DataIdentifier>>());
+
+ /**
+ * In memory map to hold failed asynchronous upload {@link DataIdentifier}
+ * and its retry count. Once if all retries are exhausted or file is
+ * successfully uploaded, then corresponding entry is flushed from the map.
+ * As all failed uploads are synchronously uploaded at startup, this map
+ * is not required to be persisted.
+ */
+ protected final Map<DataIdentifier, Integer> uploadRetryMap = new ConcurrentHashMap<DataIdentifier, Integer>(5);
+
+ /**
+ * In memory map to hold in-progress asynchronous touch. Once touch is
+ * successful corresponding entry is flushed from the map.
+ */
+ protected final Map<DataIdentifier, Long> asyncTouchCache = new ConcurrentHashMap<DataIdentifier, Long>(5);
+
+ /**
+ * In memory map to hold in-progress asynchronous downloads. Once
+ * download is finished corresponding entry is flushed from the map.
+ */
+ protected final Map<DataIdentifier, Long> asyncDownloadCache = new ConcurrentHashMap<DataIdentifier, Long>(5);
+
+ /**
+ * In memory cache to hold {@link DataRecord#getLength()} against
+ * {@link DataIdentifier}
+ */
+ protected Map<DataIdentifier, Long> recLenCache = null;
+
+ protected Backend backend;
+
+ /**
+ * The minimum size of an object that should be stored in this data store.
+ */
+ private int minRecordLength = 16 * 1024;
+
+ private String path;
+
+ private File directory;
+
+ private File tmpDir;
+
+ private String secret;
+
+ /**
+ * Flag to indicate if lastModified is updated asynchronously.
+ */
+ private boolean touchAsync = false;
+
+ /**
+ * Flag to indicate that binary content will be cached proactively and
+ * asynchronously when binary metadata is retrieved from {@link Backend}.
+ */
+ private boolean proactiveCaching = true;
+
+ /**
+ * The optional backend configuration.
+ */
+ private String config;
+
+ /**
+ * The minimum modified date. If a file is accessed (read or write) with a
+ * modified date older than this value, the modified date is updated to the
+ * current time.
+ */
+ private long minModifiedDate;
+
+ /**
+ * Cache purge trigger factor. Cache will undergo in auto-purge mode if
+ * cache current size is greater than cachePurgeTrigFactor * cacheSize
+ */
+ private double cachePurgeTrigFactor = 0.95d;
+
+ /**
+ * Cache resize factor. After auto-purge mode, cache current size would just
+ * greater than cachePurgeResizeFactor * cacheSize cacheSize
+ */
+ private double cachePurgeResizeFactor = 0.85d;
+
+ /**
+ * The number of bytes in the cache. The default value is 64 GB.
+ */
+ private long cacheSize = 64L * 1024 * 1024 * 1024;
+
+ /**
+ * The number of retries for failed upload.
+ */
+ private int uploadRetries = 3;
+
+ /**
+ * The local file system cache.
+ */
+ private LocalCache cache;
+
+ /**
+ * Caching holding pending uploads
+ */
+ private AsyncUploadCache asyncWriteCache;
+
+ /**
+ * {@link ExecutorService} to asynchronous downloads
+ */
+ private ExecutorService downloadExecService;
+
+ protected abstract Backend createBackend();
+
+ protected abstract String getMarkerFile();
+
+ /**
+ * In {@link #init(String)},it resumes all incomplete asynchronous upload
+ * from {@link AsyncUploadCache} and uploads them concurrently in multiple
+ * threads. It throws {@link RepositoryException}, if file is not found in
+ * local cache for that asynchronous upload. As far as code is concerned, it
+ * is only possible when somebody has removed files from local cache
+ * manually. If there is an exception and user want to proceed with
+ * inconsistencies, set parameter continueOnAsyncUploadFailure to true in
+ * repository.xml. This will ignore {@link RepositoryException} and log all
+ * missing files and proceed after resetting {@link AsyncUploadCache} .
+ */
+ private boolean continueOnAsyncUploadFailure;
+
+ /**
+ * The {@link #init(String)} methods checks for {@link #getMarkerFile()} and
+ * if it doesn't exists migrates all files from fileystem to {@link Backend}
+ * . This parameter governs number of threads which will upload files
+ * concurrently to {@link Backend}.
+ */
+ private int concurrentUploadsThreads = 10;
+
+ /**
+ * This parameter limits the number of asynchronous uploads slots to
+ * {@link Backend}. Once this limit is reached, further uploads to
+ * {@link Backend} are synchronous, till one of asynchronous uploads
+ * completes and make asynchronous uploads slot available. To disable
+ * asynchronous upload, set {@link #asyncUploadLimit} parameter to 0 in
+ * repository.xml. By default it is 100
+ */
+ private int asyncUploadLimit = 100;
+
+ /**
+ * Size of {@link #recLenCache}. Each entry consumes of approx 140 bytes.
+ * Default total memory consumption of {@link #recLenCache} 28KB.
+ */
+ private int recLengthCacheSize = 200;
+
+ /**
+ * Initialized the data store. If the path is not set, <repository
+ * home>/repository/datastore is used. This directory is automatically
+ * created if it does not yet exist. During first initialization, it upload
+ * all files from local datastore to backed and local datastore act as a
+ * local cache.
+ */
+ @Override
+ public void init(String homeDir) throws RepositoryException {
+ try {
+ if (path == null) {
+ path = homeDir + "/repository/datastore";
+ }
+ // create tmp inside path
+ tmpDir = new File(path, "tmp");
+ LOG.info("path=[{}], tmpPath=[{}]", path, tmpDir.getAbsolutePath());
+ directory = new File(path);
+ mkdirs(directory);
+ mkdirs(new File(homeDir));
+
+ if (!mkdirs(tmpDir)) {
+ FileUtils.cleanDirectory(tmpDir);
+ LOG.info("tmp=[{}] cleaned.", tmpDir.getPath());
+ }
+ boolean asyncWriteCacheInitStatus = true;
+ try {
+ asyncWriteCache = new AsyncUploadCache();
+ asyncWriteCache.init(homeDir, path, asyncUploadLimit);
+ } catch (Exception e) {
+ LOG.warn("Failed to initialize asyncWriteCache", e);
+ asyncWriteCacheInitStatus = false;
+ }
+ backend = createBackend();
+ backend.init(this, path, config);
+ String markerFileName = getMarkerFile();
+ if (markerFileName != null && !"".equals(markerFileName.trim())) {
+ // create marker file in homeDir to avoid deletion in cache
+ // cleanup.
+ File markerFile = new File(homeDir, markerFileName);
+ if (!markerFile.exists()) {
+ LOG.info("load files from local cache");
+ uploadFilesFromCache();
+ try {
+ markerFile.createNewFile();
+ } catch (IOException e) {
+ throw new DataStoreException(
+ "Could not create marker file "
+ + markerFile.getAbsolutePath(), e);
+ }
+ } else {
+ LOG.info("marker file = [{}] exists ",
+ markerFile.getAbsolutePath());
+ if (!asyncWriteCacheInitStatus) {
+ LOG.info("Initialization of asyncWriteCache failed. "
+ + "Re-loading all files from local cache");
+ uploadFilesFromCache();
+ asyncWriteCache.reset();
+ }
+ }
+ } else {
+ throw new DataStoreException("Failed to intialized DataStore."
+ + " MarkerFileName is null or empty. ");
+ }
+ // upload any leftover async uploads to backend during last shutdown
+ Set<String> fileList = asyncWriteCache.getAll();
+ if (fileList != null && !fileList.isEmpty()) {
+ List<String> errorFiles = new ArrayList<String>();
+ LOG.info("Uploading [{}] and size=[{}] from AsyncUploadCache.",
+ fileList, fileList.size());
+ long totalSize = 0;
+ List<File> files = new ArrayList<File>(fileList.size());
+ for (String fileName : fileList) {
+ File f = new File(path, fileName);
+ if (!f.exists()) {
+ errorFiles.add(fileName);
+ LOG.error(
+ "Cannot upload pending file [{}]. File doesn't exist.",
+ f.getAbsolutePath());
+ } else {
+ totalSize += f.length();
+ files.add(new File(path, fileName));
+ }
+ }
+ new FilesUploader(files, totalSize, concurrentUploadsThreads,
+ true).upload();
+ if (!continueOnAsyncUploadFailure && errorFiles.size() > 0) {
+ LOG.error(
+ "Pending uploads of files [{}] failed. Files do not exist in Local cache.",
+ errorFiles);
+ LOG.error("To continue set [continueOnAsyncUploadFailure] "
+ + "to true in Datastore configuration in "
+ + "repository.xml. There would be inconsistent data "
+ + "in repository due the missing files. ");
+ throw new RepositoryException(
+ "Cannot upload async uploads from local cache. Files not found.");
+ } else {
+ if (errorFiles.size() > 0) {
+ LOG.error(
+ "Pending uploads of files [{}] failed. Files do" +
+ " not exist in Local cache. Continuing as " +
+ "[continueOnAsyncUploadFailure] is set to true.",
+ errorFiles);
+ }
+ LOG.info("Reseting AsyncWrite Cache list.");
+ asyncWriteCache.reset();
+ }
+ }
+ downloadExecService = Executors.newFixedThreadPool(5,
+ new NamedThreadFactory("backend-file-download-worker"));
+ cache = new LocalCache(path, tmpDir.getAbsolutePath(), cacheSize,
+ cachePurgeTrigFactor, cachePurgeResizeFactor, asyncWriteCache);
+ /*
+ * Initialize LRU cache of size {@link #recLengthCacheSize}
+ */
+ recLenCache = Collections.synchronizedMap(new LinkedHashMap<DataIdentifier, Long>(
+ recLengthCacheSize, 0.75f, true) {
+
+ private static final long serialVersionUID = -8752749075395630485L;
+
+ @Override
+ protected boolean removeEldestEntry(
+ Map.Entry<DataIdentifier, Long> eldest) {
+ if (size() > recLengthCacheSize) {
+ LOG.trace("evicted from recLengthCache [{}]",
+ eldest.getKey());
+ return true;
+ }
+ return false;
+ }
+ });
+ } catch (Exception e) {
+ throw new RepositoryException(e);
+ }
+ }
+
+ /**
+ * Creates a new data record in {@link Backend}. The stream is first
+ * consumed and the contents are saved in a temporary file and the {@link #DIGEST}
+ * message digest of the stream is calculated. If a record with the same
+ * {@link #DIGEST} digest (and length) is found then it is returned. Otherwise new
+ * record is created in {@link Backend} and the temporary file is moved in
+ * place to {@link LocalCache}.
+ *
+ * @param input
+ * binary stream
+ * @return {@link CachingDataRecord}
+ * @throws DataStoreException
+ * if the record could not be created.
+ */
+ @Override
+ public DataRecord addRecord(InputStream input) throws DataStoreException {
+ File temporary = null;
+ long startTime = System.currentTimeMillis();
+ long length = 0;
+ try {
+ temporary = newTemporaryFile();
+ DataIdentifier tempId = new DataIdentifier(temporary.getName());
+ usesIdentifier(tempId);
+ // Copy the stream to the temporary file and calculate the
+ // stream length and the message digest of the stream
+ MessageDigest digest = MessageDigest.getInstance(DIGEST);
+ OutputStream output = new DigestOutputStream(new FileOutputStream(
+ temporary), digest);
+ try {
+ length = IOUtils.copyLarge(input, output);
+ } finally {
+ output.close();
+ }
+ long currTime = System.currentTimeMillis();
+ DataIdentifier identifier = new DataIdentifier(
+ encodeHexString(digest.digest()));
+ LOG.debug("Digest of [{}], length =[{}] took [{}]ms ",
+ new Object[] { identifier, length, (currTime - startTime) });
+ String fileName = getFileName(identifier);
+ AsyncUploadCacheResult result = null;
+ synchronized (this) {
+ usesIdentifier(identifier);
+ // check if async upload is already in progress
+ if (!asyncWriteCache.hasEntry(fileName, true)) {
+ result = cache.store(fileName, temporary, true);
+ }
+ }
+ LOG.debug("storing [{}] in localCache took [{}] ms", identifier,
+ (System.currentTimeMillis() - currTime));
+ if (result != null) {
+ if (result.canAsyncUpload()) {
+ backend.writeAsync(identifier, result.getFile(), this);
+ } else {
+ backend.write(identifier, result.getFile());
+ }
+ }
+ // this will also make sure that
+ // tempId is not garbage collected until here
+ inUse.remove(tempId);
+ LOG.debug("addRecord [{}] of length [{}] took [{}]ms.",
+ new Object[] { identifier, length,
+ (System.currentTimeMillis() - startTime) });
+ return new CachingDataRecord(this, identifier);
+ } catch (NoSuchAlgorithmException e) {
+ throw new DataStoreException(DIGEST + " not available", e);
+ } catch (IOException e) {
+ throw new DataStoreException("Could not add record", e);
+ } finally {
+ if (temporary != null) {
+ // try to delete - but it's not a big deal if we can't
+ temporary.delete();
+ }
+ }
+ }
+
+ @Override
+ public DataRecord getRecord(DataIdentifier identifier)
+ throws DataStoreException {
+ String fileName = getFileName(identifier);
+ try {
+ if (getLength(identifier) > -1) {
+ LOG.trace("getRecord: [{}] retrieved using getLength",
+ identifier);
+ if (minModifiedDate > 0) {
+ touchInternal(identifier);
+ }
+ usesIdentifier(identifier);
+ return new CachingDataRecord(this, identifier);
+ } else if (asyncWriteCache.hasEntry(fileName, minModifiedDate > 0)) {
+ LOG.trace("getRecord: [{}] retrieved from asyncUploadmap",
+ identifier);
+ usesIdentifier(identifier);
+ return new CachingDataRecord(this, identifier);
+ }
+ } catch (IOException ioe) {
+ throw new DataStoreException("error in getting record ["
+ + identifier + "]", ioe);
+ }
+ throw new DataStoreException("Record not found: " + identifier);
+ }
+
+ /**
+ * Get a data record for the given identifier or null it data record doesn't
+ * exist in {@link Backend}
+ *
+ * @param identifier identifier of record.
+ * @return the {@link CachingDataRecord} or null.
+ */
+ @Override
+ public DataRecord getRecordIfStored(DataIdentifier identifier)
+ throws DataStoreException {
+ String fileName = getFileName(identifier);
+ try {
+ if (asyncWriteCache.hasEntry(fileName, minModifiedDate > 0)) {
+ LOG.trace(
+ "getRecordIfStored: [{}] retrieved from asyncuploadmap",
+ identifier);
+ usesIdentifier(identifier);
+ return new CachingDataRecord(this, identifier);
+ } else if (recLenCache.containsKey(identifier)) {
+ LOG.trace(
+ "getRecordIfStored: [{}] retrieved using recLenCache",
+ identifier);
+ if (minModifiedDate > 0) {
+ touchInternal(identifier);
+ }
+ usesIdentifier(identifier);
+ return new CachingDataRecord(this, identifier);
+ } else {
+ try {
+ long length = backend.getLength(identifier);
+ LOG.debug(
+ "getRecordIfStored :[{}] retrieved from backend",
+ identifier);
+ recLenCache.put(identifier, length);
+ if (minModifiedDate > 0) {
+ touchInternal(identifier);
+ }
+ usesIdentifier(identifier);
+ return new CachingDataRecord(this, identifier);
+ } catch (DataStoreException ignore) {
+ LOG.warn(" getRecordIfStored: [{}] not found", identifier);
+ }
+
+ }
+ } catch (IOException ioe) {
+ throw new DataStoreException(ioe);
+ }
+ return null;
+ }
+
+ @Override
+ public void updateModifiedDateOnAccess(long before) {
+ LOG.info("minModifiedDate set to [{}]", before);
+ minModifiedDate = before;
+ }
+
+ /**
+ * Retrieves all identifiers from {@link Backend}.
+ */
+ @Override
+ public Iterator<DataIdentifier> getAllIdentifiers()
+ throws DataStoreException {
+ Set<DataIdentifier> ids = new HashSet<DataIdentifier>();
+ for (String fileName : asyncWriteCache.getAll()) {
+ ids.add(getIdentifier(fileName));
+ }
+ Iterator<DataIdentifier> itr = backend.getAllIdentifiers();
+ while (itr.hasNext()) {
+ ids.add(itr.next());
+ }
+ return ids.iterator();
+ }
+
+ /**
+ * This method deletes record from {@link Backend} and then from
+ * {@link LocalCache}
+ */
+ @Override
+ public void deleteRecord(DataIdentifier identifier)
+ throws DataStoreException {
+ String fileName = getFileName(identifier);
+ synchronized (this) {
+ try {
+ // order is important here
+ recLenCache.remove(identifier);
+ asyncWriteCache.delete(fileName);
+ backend.deleteRecord(identifier);
+ cache.delete(fileName);
+ } catch (IOException ioe) {
+ throw new DataStoreException(ioe);
+ }
+ }
+ }
+
+ @Override
+ public synchronized int deleteAllOlderThan(long min)
+ throws DataStoreException {
+ Set<DataIdentifier> diSet = backend.deleteAllOlderThan(min);
+
+ // remove entries from local cache
+ for (DataIdentifier identifier : diSet) {
+ recLenCache.remove(identifier);
+ cache.delete(getFileName(identifier));
+ }
+ try {
+ for (String fileName : asyncWriteCache.deleteOlderThan(min)) {
+ diSet.add(getIdentifier(fileName));
+ }
+ } catch (IOException e) {
+ throw new DataStoreException(e);
+ }
+ LOG.info(
+ "deleteAllOlderThan exit. Deleted [{}]records. Number of records deleted [{}]",
+ diSet, diSet.size());
+ return diSet.size();
+ }
+
+ /**
+ * Get stream of record from {@link LocalCache}. If record is not available
+ * in {@link LocalCache}, this method fetches record from {@link Backend}
+ * and stores it to {@link LocalCache}. Stream is then returned from cached
+ * record.
+ */
+ InputStream getStream(DataIdentifier identifier) throws DataStoreException {
+ InputStream in = null;
+ try {
+ String fileName = getFileName(identifier);
+ InputStream cached = cache.getIfStored(fileName);
+ if (cached != null) {
+ return cached;
+ }
+ in = backend.read(identifier);
+ return cache.store(fileName, in);
+ } catch (IOException e) {
+ throw new DataStoreException("IO Exception: " + identifier, e);
+ } finally {
+ IOUtils.closeQuietly(in);
+ }
+ }
+
+ /**
+ * Return lastModified of record from {@link Backend} assuming
+ * {@link Backend} as a single source of truth.
+ */
+ public long getLastModified(DataIdentifier identifier)
+ throws DataStoreException {
+ String fileName = getFileName(identifier);
+ long lastModified = asyncWriteCache.getLastModified(fileName);
+ if (lastModified != 0) {
+ LOG.trace(
+ "identifier [{}], lastModified=[{}] retrireved from AsyncUploadCache ",
+ identifier, lastModified);
+
+ } else if (asyncTouchCache.get(identifier) != null) {
+ lastModified = asyncTouchCache.get(identifier);
+ LOG.trace(
+ "identifier [{}], lastModified=[{}] retrireved from asyncTouchCache ",
+ identifier, lastModified);
+ } else {
+ lastModified = backend.getLastModified(identifier);
+ LOG.debug(
+ "identifier [{}], lastModified=[{}] retrireved from backend ",
+ identifier, lastModified);
+ asyncDownload(identifier);
+ }
+ return lastModified;
+ }
+
+ /**
+ * Return the length of record from {@link LocalCache} if available,
+ * otherwise retrieve it from {@link Backend}.
+ */
+ public long getLength(final DataIdentifier identifier)
+ throws DataStoreException {
+ String fileName = getFileName(identifier);
+
+ Long length = recLenCache.get(identifier);
+ if (length != null) {
+ LOG.trace(" identifier [{}] length fetched from recLengthCache",
+ identifier);
+ return length;
+ } else if ((length = cache.getFileLength(fileName)) != null) {
+ LOG.trace(" identifier [{}] length fetched from local cache",
+ identifier);
+ recLenCache.put(identifier, length);
+ return length;
+ } else {
+ length = backend.getLength(identifier);
+ LOG.debug(" identifier [{}] length fetched from backend",
+ identifier);
+ recLenCache.put(identifier, length);
+ asyncDownload(identifier);
+ return length;
+ }
+ }
+
+ @Override
+ protected byte[] getOrCreateReferenceKey() throws DataStoreException {
+ return secret.getBytes(StandardCharsets.UTF_8);
+ }
+
+ public Set<String> getPendingUploads() {
+ return asyncWriteCache.getAll();
+ }
+
+
+ public void deleteFromCache(DataIdentifier identifier)
+ throws DataStoreException {
+ try {
+ // order is important here
+ recLenCache.remove(identifier);
+ String fileName = getFileName(identifier);
+ asyncWriteCache.delete(fileName);
+ cache.delete(fileName);
+ } catch (IOException ioe) {
+ throw new DataStoreException(ioe);
+ }
+ }
+
+ @Override
+ public void onSuccess(AsyncUploadResult result) {
+ DataIdentifier identifier = result.getIdentifier();
+ File file = result.getFile();
+ String fileName = getFileName(identifier);
+ try {
+ LOG.debug("Upload completed for [{}]", identifier);
+ // remove from failed upload map if any.
+ uploadRetryMap.remove(identifier);
+ AsyncUploadCacheResult cachedResult = asyncWriteCache.remove(fileName);
+ if (cachedResult.doRequiresDelete()) {
+ // added record already marked for delete
+ deleteRecord(identifier);
+ } else {
+ // async upload took lot of time.
+ // getRecord to touch if required.
+ getRecord(identifier);
+ }
+ } catch (IOException ie) {
+ LOG.warn("Cannot remove pending file upload. Dataidentifer [ "
+ + identifier + "], file [" + file.getAbsolutePath() + "]", ie);
+ } catch (DataStoreException dse) {
+ LOG.warn("Cannot remove pending file upload. Dataidentifer [ "
+ + identifier + "], file [" + file.getAbsolutePath() + "]", dse);
+ }
+ }
+
+ @Override
+ public void onFailure(AsyncUploadResult result) {
+ DataIdentifier identifier = result.getIdentifier();
+ File file = result.getFile();
+ String fileName = getFileName(identifier);
+ if (result.getException() != null) {
+ LOG.warn("Async Upload failed. Dataidentifer [ " + identifier
+ + "], file [" + file.getAbsolutePath() + "]",
+ result.getException());
+ } else {
+ LOG.warn("Async Upload failed. Dataidentifer [ " + identifier
+ + "], file [" + file.getAbsolutePath() + "]");
+ }
+ // Retry failed upload upto uploadRetries times.
+ try {
+ if (asyncWriteCache.hasEntry(fileName, false)) {
+ synchronized (uploadRetryMap) {
+ Integer retry = uploadRetryMap.get(identifier);
+ if (retry == null) {
+ retry = new Integer(1);
+ } else {
+ retry++;
+ }
+ if (retry <= uploadRetries) {
+ uploadRetryMap.put(identifier, retry);
+ LOG.info(
+ "Retrying [{}] times failed upload for dataidentifer {}",
+ retry, identifier);
+ try {
+ backend.writeAsync(identifier, file, this);
+ } catch (DataStoreException e) {
+ LOG.warn("exception", e);
+ }
+ } else {
+ LOG.info("Retries [{}] exhausted for dataidentifer {}.",
+ (retry - 1), identifier);
+ uploadRetryMap.remove(identifier);
+ }
+ }
+ }
+ } catch (IOException ie) {
+ LOG.warn("Cannot retry failed async file upload. Dataidentifer [ "
+ + identifier + "], file [" + file.getAbsolutePath() + "]", ie);
+ }
+ }
+
+ @Override
+ public void onAbort(AsyncUploadResult result) {
+ DataIdentifier identifier = result.getIdentifier();
+ File file = result.getFile();
+ String fileName = getFileName(identifier);
+ try {
+ // remove from failed upload map if any.
+ uploadRetryMap.remove(identifier);
+ asyncWriteCache.remove(fileName);
+ LOG.info(
+ "Async Upload Aborted. Dataidentifer [{}], file [{}] removed from AsyncCache.",
+ identifier, file.getAbsolutePath());
+ } catch (IOException ie) {
+ LOG.warn("Cannot remove pending file upload. Dataidentifer [ "
+ + identifier + "], file [" + file.getAbsolutePath() + "]", ie);
+ }
+ }
+
+
+ @Override
+ public void onSuccess(AsyncTouchResult result) {
+ asyncTouchCache.remove(result.getIdentifier());
+ LOG.debug(" Async Touch succeed. Removed [{}] from asyncTouchCache",
+ result.getIdentifier());
+
+ }
+
+ @Override
+ public void onFailure(AsyncTouchResult result) {
+ LOG.warn(" Async Touch failed. Not removing [{}] from asyncTouchCache",
+ result.getIdentifier());
+ if (result.getException() != null) {
+ LOG.debug(" Async Touch failed. exception", result.getException());
+ }
+ }
+
+ @Override
+ public void onAbort(AsyncTouchResult result) {
+ asyncTouchCache.remove(result.getIdentifier());
+ LOG.debug(" Async Touch aborted. Removed [{}] from asyncTouchCache",
+ result.getIdentifier());
+ }
+
+ /**
+ * Method to confirm that identifier can be deleted from {@link Backend}
+ *
+ * @param identifier
+ * @return
+ */
+ public boolean confirmDelete(DataIdentifier identifier) {
+ if (isInUse(identifier)) {
+ LOG.debug("identifier [{}] is inUse confirmDelete= false ",
+ identifier);
+ return false;
+ }
+
+ String fileName = getFileName(identifier);
+ long lastModified = asyncWriteCache.getLastModified(fileName);
+ if (lastModified != 0) {
+ LOG.debug(
+ "identifier [{}] is asyncWriteCache map confirmDelete= false ",
+ identifier);
+ return false;
+
+ }
+ if (asyncTouchCache.get(identifier) != null) {
+ LOG.debug(
+ "identifier [{}] is asyncTouchCache confirmDelete = false ",
+ identifier);
+ return false;
+ }
+
+ return true;
+ }
+
+ /**
+ * Internal method to touch identifier in @link {@link Backend}. if
+ * {@link #touchAsync}, the record is updated asynchronously.
+ *
+ * @param identifier
+ * @throws DataStoreException
+ */
+ private void touchInternal(DataIdentifier identifier)
+ throws DataStoreException {
+
+ if (touchAsync) {
+ Long lastModified = asyncTouchCache.put(identifier,
+ System.currentTimeMillis());
+
+ if (lastModified == null) {
+ LOG.debug("Async touching [{}] ", identifier);
+ backend.touchAsync(identifier, minModifiedDate, this);
+ } else {
+ LOG.debug( "Touched in asyncTouchMap [{}]", identifier);
+ }
+
+ } else {
+ backend.touch(identifier, minModifiedDate);
+ }
+ }
+
+ /**
+ * Invoke {@link #getStream(DataIdentifier)} asynchronously to cache binary
+ * asynchronously.
+ */
+ private void asyncDownload(final DataIdentifier identifier) {
+ if (proactiveCaching
+ && cacheSize != 0
+ && asyncDownloadCache.put(identifier, System.currentTimeMillis()) == null) {
+ downloadExecService.execute(new Runnable() {
+ @Override
+ public void run() {
+ long startTime = System.currentTimeMillis();
+ InputStream input = null;
+ try {
+ LOG.trace("Async download [{}] started.", identifier);
+ input = getStream(identifier);
+ } catch (RepositoryException re) {
+ // ignore exception
+ } finally {
+ asyncDownloadCache.remove(identifier);
+ IOUtils.closeQuietly(input);
+ LOG.debug("Async download [{}] completed in [{}] ms.",
+ identifier,
+ (System.currentTimeMillis() - startTime));
+ }
+ }
+ });
+ }
+ }
+
+ /**
+ * Returns a unique temporary file to be used for creating a new data
+ * record.
+ */
+ private File newTemporaryFile() throws IOException {
+ return File.createTempFile(TMP, null, tmpDir);
+ }
+
+ /**
+ * Load files from {@link LocalCache} to {@link Backend}.
+ */
+ private void uploadFilesFromCache() throws RepositoryException {
+ ArrayList<File> files = new ArrayList<File>();
+ listRecursive(files, directory);
+ long totalSize = 0;
+ for (File f : files) {
+ totalSize += f.length();
+ }
+ if (files.size() > 0) {
+ if (concurrentUploadsThreads > 1) {
+ new FilesUploader(files, totalSize, concurrentUploadsThreads,
+ false).upload();
+ } else {
+ uploadFilesInSingleThread(files, totalSize);
+ }
+ }
+ }
+
+ private void uploadFilesInSingleThread(List<File> files, long totalSize)
+ throws RepositoryException {
+ long startTime = System.currentTimeMillis();
+ LOG.info("Upload: [{}] files in single thread.", files.size());
+ long currentCount = 0;
+ long currentSize = 0;
+ long time = System.currentTimeMillis();
+ for (File f : files) {
+ String name = f.getName();
+ LOG.debug("upload file [{}] ", name);
+ if (!name.startsWith(TMP) && !name.endsWith(DS_STORE)
+ && f.length() > 0) {
+ uploadFileToBackEnd(f, false);
+ }
+ currentSize += f.length();
+ currentCount++;
+ long now = System.currentTimeMillis();
+ if (now > time + 5000) {
+ LOG.info("Uploaded: [{}/{}] files, [{}/{}] size data",
+ new Object[] { currentCount, files.size(), currentSize,
+ totalSize });
+ time = now;
+ }
+ }
+ long endTime = System.currentTimeMillis();
+ LOG.info(
+ "Uploaded: [{}/{}] files, [{}/{}] size data, time taken = [{}] sec",
+ new Object[] { currentCount, files.size(), currentSize, totalSize,
+ ((endTime - startTime) / 1000) });
+ }
+
+ /**
+ * Traverse recursively and populate list with files.
+ */
+ private static void listRecursive(List<File> list, File file) {
+ File[] files = file.listFiles();
+ if (files != null) {
+ for (File f : files) {
+ if (f.isDirectory()) {
+ listRecursive(list, f);
+ } else {
+ list.add(f);
+ }
+ }
+ }
+ }
+
+ /**
+ * Upload file from {@link LocalCache} to {@link Backend}.
+ *
+ * @param f
+ * file to uploaded.
+ * @throws DataStoreException
+ */
+ private void uploadFileToBackEnd(File f, boolean updateAsyncUploadCache)
+ throws DataStoreException {
+ try {
+ DataIdentifier identifier = new DataIdentifier(f.getName());
+ usesIdentifier(identifier);
+ if (!backend.exists(identifier)) {
+ backend.write(identifier, f);
+ }
+ if (updateAsyncUploadCache) {
+ String fileName = getFileName(identifier);
+ asyncWriteCache.remove(fileName);
+ }
+ LOG.debug("uploaded [{}]", f.getName());
+ } catch (IOException ioe) {
+ throw new DataStoreException(ioe);
+ }
+ }
+
+ /**
+ * Derive file name from identifier.
+ */
+ private static String getFileName(DataIdentifier identifier) {
+ String name = identifier.toString();
+ return getFileName(name);
+ }
+
+ private static String getFileName(String name) {
+ return name.substring(0, 2) + "/" + name.substring(2, 4) + "/"
+ + name.substring(4, 6) + "/" + name;
+ }
+
+ private static DataIdentifier getIdentifier(String fileName) {
+ return new DataIdentifier(
+ fileName.substring(fileName.lastIndexOf("/") + 1));
+ }
+
+ private void usesIdentifier(DataIdentifier identifier) {
+ inUse.put(identifier, new WeakReference<DataIdentifier>(identifier));
+ }
+
+ private static boolean mkdirs(File dir) throws IOException {
+ if (dir.exists()) {
+ if (dir.isFile()) {
+ throw new IOException("Can not create a directory "
+ + "because a file exists with the same name: "
+ + dir.getAbsolutePath());
+ }
+ return false;
+ }
+ boolean created = dir.mkdirs();
+ if (!created) {
+ throw new IOException("Could not create directory: "
+ + dir.getAbsolutePath());
+ }
+ return created;
+ }
+
+ @Override
+ public void clearInUse() {
+ inUse.clear();
+ }
+
+ public boolean isInUse(DataIdentifier identifier) {
+ return inUse.containsKey(identifier);
+ }
+
+ @Override
+ public void close() throws DataStoreException {
+ cache.close();
+ backend.close();
+ downloadExecService.shutdown();
+ }
+
+ /**
+ * Setter for configuration based secret
+ *
+ * @param secret
+ * the secret used to sign reference binaries
+ */
+ public void setSecret(String secret) {
+ this.secret = secret;
+ }
+
+ /**
+ * Set the minimum object length.
+ *
+ * @param minRecordLength
+ * the length
+ */
+ public void setMinRecordLength(int minRecordLength) {
+ this.minRecordLength = minRecordLength;
+ }
+
+ /**
+ * Return mininum object length.
+ */
+ @Override
+ public int getMinRecordLength() {
+ return minRecordLength;
+ }
+
+ /**
+ * Return path of configuration properties.
+ *
+ * @return path of configuration properties.
+ */
+ public String getConfig() {
+ return config;
+ }
+
+ /**
+ * Set the configuration properties path.
+ *
+ * @param config
+ * path of configuration properties.
+ */
+ public void setConfig(String config) {
+ this.config = config;
+ }
+
+ /**
+ * @return size of {@link LocalCache}.
+ */
+ public long getCacheSize() {
+ return cacheSize;
+ }
+
+ /**
+ * Set size of {@link LocalCache}.
+ *
+ * @param cacheSize
+ * size of {@link LocalCache}.
+ */
+ public void setCacheSize(long cacheSize) {
+ this.cacheSize = cacheSize;
+ }
+
+ /**
+ * @return path of {@link LocalCache}.
+ */
+ public String getPath() {
+ return path;
+ }
+
+ /**
+ * Set path of {@link LocalCache}.
+ *
+ * @param path
+ * of {@link LocalCache}.
+ */
+ public void setPath(String path) {
+ this.path = path;
+ }
+
+ /**
+ * @return Purge trigger factor of {@link LocalCache}.
+ */
+ public double getCachePurgeTrigFactor() {
+ return cachePurgeTrigFactor;
+ }
+
+ /**
+ * Set purge trigger factor of {@link LocalCache}.
+ *
+ * @param cachePurgeTrigFactor
+ * purge trigger factor.
+ */
+ public void setCachePurgeTrigFactor(double cachePurgeTrigFactor) {
+ this.cachePurgeTrigFactor = cachePurgeTrigFactor;
+ }
+
+ /**
+ * @return Purge resize factor of {@link LocalCache}.
+ */
+ public double getCachePurgeResizeFactor() {
+ return cachePurgeResizeFactor;
+ }
+
+ /**
+ * Set purge resize factor of {@link LocalCache}.
+ *
+ * @param cachePurgeResizeFactor
+ * purge resize factor.
+ */
+ public void setCachePurgeResizeFactor(double cachePurgeResizeFactor) {
+ this.cachePurgeResizeFactor = cachePurgeResizeFactor;
+ }
+
+ public int getConcurrentUploadsThreads() {
+ return concurrentUploadsThreads;
+ }
+
+ public void setConcurrentUploadsThreads(int concurrentUploadsThreads) {
+ this.concurrentUploadsThreads = concurrentUploadsThreads;
+ }
+
+ public int getAsyncUploadLimit() {
+ return asyncUploadLimit;
+ }
+
+ public void setAsyncUploadLimit(int asyncUploadLimit) {
+ this.asyncUploadLimit = asyncUploadLimit;
+ }
+
+ public boolean isContinueOnAsyncUploadFailure() {
+ return continueOnAsyncUploadFailure;
+ }
+
+ public void setContinueOnAsyncUploadFailure(
+ boolean continueOnAsyncUploadFailure) {
+ this.continueOnAsyncUploadFailure = continueOnAsyncUploadFailure;
+ }
+
+ public int getUploadRetries() {
+ return uploadRetries;
+ }
+
+ public void setUploadRetries(int uploadRetries) {
+ this.uploadRetries = uploadRetries;
+ }
+
+ public void setTouchAsync(boolean touchAsync) {
+ this.touchAsync = touchAsync;
+ }
+
+ public void setProactiveCaching(boolean proactiveCaching) {
+ this.proactiveCaching = proactiveCaching;
+ }
+
+ public void setRecLengthCacheSize(int recLengthCacheSize) {
+ this.recLengthCacheSize = recLengthCacheSize;
+ }
+
+ public Backend getBackend() {
+ return backend;
+ }
+
+ /**
+ * This class initiates files upload in multiple threads to backend.
+ */
+ private class FilesUploader {
+ final List<File> files;
+
+ final long totalSize;
+
+ volatile AtomicInteger currentCount = new AtomicInteger();
+
+ volatile AtomicLong currentSize = new AtomicLong();
+
+ volatile AtomicBoolean exceptionRaised = new AtomicBoolean();
+
+ DataStoreException exception;
+
+ final int threads;
+
+ final boolean updateAsyncCache;
+
+ FilesUploader(List<File> files, long totalSize, int threads,
+ boolean updateAsyncCache) {
+ super();
+ this.files = files;
+ this.threads = threads;
+ this.totalSize = totalSize;
+ this.updateAsyncCache = updateAsyncCache;
+ }
+
+ void addCurrentCount(int delta) {
+ currentCount.addAndGet(delta);
+ }
+
+ void addCurrentSize(long delta) {
+ currentSize.addAndGet(delta);
+ }
+
+ synchronized void setException(DataStoreException exception) {
+ exceptionRaised.getAndSet(true);
+ this.exception = exception;
+ }
+
+ boolean isExceptionRaised() {
+ return exceptionRaised.get();
+ }
+
+ void logProgress() {
+ LOG.info("Uploaded: [{}/{}] files, [{}/{}] size data",
+ new Object[] { currentCount, files.size(), currentSize,
+ totalSize });
+ }
+
+ void upload() throws DataStoreException {
+ long startTime = System.currentTimeMillis();
+ LOG.info(" Uploading [{}] using [{}] threads.", files.size(), threads);
+ ExecutorService executor = Executors.newFixedThreadPool(threads,
+ new NamedThreadFactory("backend-file-upload-worker"));
+ int partitionSize = files.size() / (threads);
+ int startIndex = 0;
+ int endIndex = partitionSize;
+ for (int i = 1; i <= threads; i++) {
+ List<File> partitionFileList = Collections.unmodifiableList(files.subList(
+ startIndex, endIndex));
+ FileUploaderThread fut = new FileUploaderThread(
+ partitionFileList, startIndex, endIndex, this,
+ updateAsyncCache);
+ executor.execute(fut);
+
+ startIndex = endIndex;
+ if (i == (threads - 1)) {
+ endIndex = files.size();
+ } else {
+ endIndex = startIndex + partitionSize;
+ }
+ }
+ // This will make the executor accept no new threads
+ // and finish all existing threads in the queue
+ executor.shutdown();
+
+ try {
+ // Wait until all threads are finish
+ while (!isExceptionRaised()
+ && !executor.awaitTermination(15, TimeUnit.SECONDS)) {
+ logProgress();
+ }
+ } catch (InterruptedException ie) {
+
+ }
+ long endTime = System.currentTimeMillis();
+ LOG.info(
+ "Uploaded: [{}/{}] files, [{}/{}] size data, time taken = [{}] sec",
+ new Object[] { currentCount, files.size(), currentSize,
+ totalSize, ((endTime - startTime) / 1000) });
+ if (isExceptionRaised()) {
+ executor.shutdownNow(); // Cancel currently executing tasks
+ throw exception;
+ }
+ }
+
+ }
+
+ /**
+ * This class implements {@link Runnable} interface and uploads list of
+ * files from startIndex to endIndex to {@link Backend}
+ */
+ private class FileUploaderThread implements Runnable {
+ final List<File> files;
+
+ final FilesUploader filesUploader;
+
+ final int startIndex;
+
+ final int endIndex;
+
+ final boolean updateAsyncCache;
+
+ FileUploaderThread(List<File> files, int startIndex, int endIndex,
+ FilesUploader controller, boolean updateAsyncCache) {
+ super();
+ this.files = files;
+ this.filesUploader = controller;
+ this.startIndex = startIndex;
+ this.endIndex = endIndex;
+ this.updateAsyncCache = updateAsyncCache;
+ }
+
+ public void run() {
+ long time = System.currentTimeMillis();
+ LOG.debug(
+ "Thread [{}] : Uploading files from startIndex [{}] to endIndex [{}] both inclusive.",
+ new Object[] { Thread.currentThread().getName(), startIndex,
+ (endIndex - 1) });
+ int uploadCount = 0;
+ long uploadSize = 0;
+ try {
+ for (File f : files) {
+
+ if (filesUploader.isExceptionRaised()) {
+ break;
+ }
+ String name = f.getName();
+ LOG.debug("upload file [{}] ",name);
+ if (!name.startsWith(TMP) && !name.endsWith(DS_STORE)
+ && f.length() > 0) {
+ uploadFileToBackEnd(f, updateAsyncCache);
+ }
+ uploadCount++;
+ uploadSize += f.length();
+ // update upload status at every 15 seconds.
+ long now = System.currentTimeMillis();
+ if (now > time + 15000) {
+ filesUploader.addCurrentCount(uploadCount);
+ filesUploader.addCurrentSize(uploadSize);
+ uploadCount = 0;
+ uploadSize = 0;
+ time = now;
+ }
+ }
+ // update final state.
+ filesUploader.addCurrentCount(uploadCount);
+ filesUploader.addCurrentSize(uploadSize);
+ } catch (DataStoreException e) {
+ if (!filesUploader.isExceptionRaised()) {
+ filesUploader.setException(e);
+ }
+ }
+
+ }
+ }
+
+}
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/data/CachingFDS.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/CachingFDS.java
new file mode 100644
index 0000000000..1cfc50e137
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/CachingFDS.java
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * {@link CachingDataStore} with {@link FSBackend}. It is performant
+ * {@link DataStore} when {@link FSBackend} is hosted on network storage
+ * (SAN or NAS). It leverages all caching capabilites of
+ * {@link CachingDataStore}.
+ */
+package org.apache.jackrabbit.core.data;
+
+import java.util.Properties;
+
+public class CachingFDS extends CachingDataStore {
+ private Properties properties;
+
+ @Override
+ protected Backend createBackend() {
+ FSBackend backend = new FSBackend();
+ if (properties != null) {
+ backend.setProperties(properties);
+ }
+ return backend;
+ }
+
+ @Override
+ protected String getMarkerFile() {
+ return "fs.init.done";
+ }
+
+ /**
+ * Properties required to configure the S3Backend
+ */
+ public void setProperties(Properties properties) {
+ this.properties = properties;
+ }
+}
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/data/DataIdentifier.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/DataIdentifier.java
new file mode 100644
index 0000000000..0c0b36a20d
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/DataIdentifier.java
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.jackrabbit.core.data;
+
+import java.io.Serializable;
+
+/**
+ * Opaque data identifier used to identify records in a data store.
+ * All identifiers must be serializable and implement the standard
+ * object equality and hash code methods.
+ */
+public class DataIdentifier implements Serializable {
+
+ /**
+ * Serial version UID.
+ */
+ private static final long serialVersionUID = -9197191401131100016L;
+
+ /**
+ * Data identifier.
+ */
+ private final String identifier;
+
+ /**
+ * Creates a data identifier from the given string.
+ *
+ * @param identifier data identifier
+ */
+ public DataIdentifier(String identifier) {
+ this.identifier = identifier;
+ }
+
+ //-------------------------------------------------------------< Object >
+
+ /**
+ * Returns the identifier string.
+ *
+ * @return identifier string
+ */
+ public String toString() {
+ return identifier;
+ }
+
+ /**
+ * Checks if the given object is a data identifier and has the same
+ * string representation as this one.
+ *
+ * @param object other object
+ * @return <code>true</code> if the given object is the same identifier,
+ * <code>false</code> otherwise
+ */
+ public boolean equals(Object object) {
+ return (object instanceof DataIdentifier)
+ && identifier.equals(object.toString());
+ }
+
+ /**
+ * Returns the hash code of the identifier string.
+ *
+ * @return hash code
+ */
+ public int hashCode() {
+ return identifier.hashCode();
+ }
+
+}
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/data/DataRecord.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/DataRecord.java
new file mode 100644
index 0000000000..022b485a82
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/DataRecord.java
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.jackrabbit.core.data;
+
+import java.io.InputStream;
+
+/**
+ * Immutable data record that consists of a binary stream.
+ */
+public interface DataRecord {
+
+ /**
+ * Returns the identifier of this record.
+ *
+ * @return data identifier
+ */
+ DataIdentifier getIdentifier();
+
+ /**
+ * Returns a secure reference to this binary, or {@code null} if no such
+ * reference is available.
+ *
+ * @return binary reference, or {@code null}
+ */
+ String getReference();
+
+ /**
+ * Returns the length of the binary stream in this record.
+ *
+ * @return length of the binary stream
+ * @throws DataStoreException if the record could not be accessed
+ */
+ long getLength() throws DataStoreException;
+
+ /**
+ * Returns the the binary stream in this record.
+ *
+ * @return binary stream
+ * @throws DataStoreException if the record could not be accessed
+ */
+ InputStream getStream() throws DataStoreException;
+
+ /**
+ * Returns the last modified of the record.
+ *
+ * @return last modified time of the binary stream
+ */
+ long getLastModified();
+}
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/data/DataStore.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/DataStore.java
new file mode 100644
index 0000000000..d714cf7b0a
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/DataStore.java
@@ -0,0 +1,154 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.jackrabbit.core.data;
+
+import java.io.InputStream;
+import java.util.Iterator;
+
+import javax.jcr.RepositoryException;
+
+/**
+ * Append-only store for binary streams. A data store consists of a number
+ * of identifiable data records that each contain a distinct binary stream.
+ * New binary streams can be added to the data store, but existing streams
+ * are never removed or modified.
+ * <p>
+ * A data store should be fully thread-safe, i.e. it should be possible to
+ * add and access data records concurrently. Optimally even separate processes
+ * should be able to concurrently access the data store with zero interprocess
+ * synchronization.
+ */
+public interface DataStore {
+
+ /**
+ * Check if a record for the given identifier exists, and return it if yes.
+ * If no record exists, this method returns null.
+ *
+ * @param identifier data identifier
+ * @return the record if found, and null if not
+ * @throws DataStoreException if the data store could not be accessed
+ */
+ DataRecord getRecordIfStored(DataIdentifier identifier)
+ throws DataStoreException;
+
+ /**
+ * Returns the identified data record. The given identifier should be
+ * the identifier of a previously saved data record. Since records are
+ * never removed, there should never be cases where the identified record
+ * is not found. Abnormal cases like that are treated as errors and
+ * handled by throwing an exception.
+ *
+ * @param identifier data identifier
+ * @return identified data record
+ * @throws DataStoreException if the data store could not be accessed,
+ * or if the given identifier is invalid
+ */
+ DataRecord getRecord(DataIdentifier identifier) throws DataStoreException;
+
+ /**
+ * Returns the record that matches the given binary reference.
+ * Returns {@code null} if the reference is invalid, for example if it
+ * points to a record that does not exist.
+ *
+ * @param reference binary reference
+ * @return matching record, or {@code null}
+ * @throws DataStoreException if the data store could not be accessed
+ */
+ DataRecord getRecordFromReference(String reference)
+ throws DataStoreException;
+
+ /**
+ * Creates a new data record. The given binary stream is consumed and
+ * a binary record containing the consumed stream is created and returned.
+ * If the same stream already exists in another record, then that record
+ * is returned instead of creating a new one.
+ * <p>
+ * The given stream is consumed and <strong>not closed</strong> by this
+ * method. It is the responsibility of the caller to close the stream.
+ * A typical call pattern would be:
+ * <pre>
+ * InputStream stream = ...;
+ * try {
+ * record = store.addRecord(stream);
+ * } finally {
+ * stream.close();
+ * }
+ * </pre>
+ *
+ * @param stream binary stream
+ * @return data record that contains the given stream
+ * @throws DataStoreException if the data store could not be accessed
+ */
+ DataRecord addRecord(InputStream stream) throws DataStoreException;
+
+ /**
+ * From now on, update the modified date of an object even when accessing it.
+ * Usually, the modified date is only updated when creating a new object,
+ * or when a new link is added to an existing object. When this setting is enabled,
+ * even getLength() will update the modified date.
+ *
+ * @param before - update the modified date to the current time if it is older than this value
+ */
+ void updateModifiedDateOnAccess(long before);
+
+ /**
+ * Delete objects that have a modified date older than the specified date.
+ *
+ * @param min the minimum time
+ * @return the number of data records deleted
+ * @throws DataStoreException
+ */
+ int deleteAllOlderThan(long min) throws DataStoreException;
+
+ /**
+ * Get all identifiers.
+ *
+ * @return an iterator over all DataIdentifier objects
+ * @throws DataStoreException if the list could not be read
+ */
+ Iterator<DataIdentifier> getAllIdentifiers() throws DataStoreException;
+
+ /**
+ * Initialized the data store
+ *
+ * @param homeDir the home directory of the repository
+ * @throws RepositoryException
+ */
+ void init(String homeDir) throws RepositoryException;
+
+ /**
+ * Get the minimum size of an object that should be stored in this data store.
+ * Depending on the overhead and configuration, each store may return a different value.
+ *
+ * @return the minimum size in bytes
+ */
+ int getMinRecordLength();
+
+ /**
+ * Close the data store
+ *
+ * @throws DataStoreException if a problem occurred
+ */
+ void close() throws DataStoreException;
+
+ /**
+ * Clear the in-use list. This is only used for testing to make the the garbage collection
+ * think that objects are no longer in use.
+ */
+ void clearInUse();
+
+}
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/data/DataStoreException.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/DataStoreException.java
new file mode 100644
index 0000000000..4b173ac84a
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/DataStoreException.java
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.jackrabbit.core.data;
+
+import javax.jcr.RepositoryException;
+
+/**
+ * Exception thrown by the Data Store module.
+ */
+public class DataStoreException extends RepositoryException {
+
+ /**
+ * Constructs a new instance of this class with the specified detail
+ * message.
+ *
+ * @param message the detailed message.
+ */
+ public DataStoreException(String message) {
+ super(message);
+ }
+
+ /**
+ * Constructs a new instance of this class with the specified detail
+ * message and root cause.
+ *
+ * @param message the detailed message.
+ * @param cause root failure cause
+ */
+ public DataStoreException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ /**
+ * Constructs a new instance of this class with the specified root cause.
+ *
+ * @param rootCause root failure cause
+ */
+ public DataStoreException(Throwable rootCause) {
+ super(rootCause);
+ }
+
+}
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/data/DataStoreFactory.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/DataStoreFactory.java
new file mode 100644
index 0000000000..390f4c5ece
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/DataStoreFactory.java
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.jackrabbit.core.data;
+
+import javax.jcr.RepositoryException;
+
+
+/**
+ * Factory interface for creating {@link DataStore} instances. Used
+ * to decouple the repository internals from the repository configuration
+ * mechanism.
+ *
+ * @since Jackrabbit 1.5
+ * @see <a href="https://issues.apache.org/jira/browse/JCR-1438">JCR-1438</a>
+ */
+public interface DataStoreFactory {
+
+ /**
+ * Creates, initializes, and returns a {@link DataStore} instance
+ * for use by the repository. Note that no information is passed from
+ * the client, so all required configuration information must be
+ * encapsulated in the factory.
+ *
+ * @return initialized data store
+ * @throws RepositoryException if the data store can not be created
+ */
+ DataStore getDataStore() throws RepositoryException;
+
+}
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/data/FSBackend.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/FSBackend.java
new file mode 100644
index 0000000000..1437178d55
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/FSBackend.java
@@ -0,0 +1,477 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+/**
+ * File system {@link Backend} used with {@link CachingDataStore}.
+ * The file system can be network storage.
+ */
+package org.apache.jackrabbit.core.data;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.RandomAccessFile;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.commons.io.IOUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class FSBackend extends AbstractBackend {
+
+ private Properties properties;
+
+ private String fsPath;
+
+ File fsPathDir;
+
+ public static final String FS_BACKEND_PATH = "fsBackendPath";
+
+ /**
+ * Logger instance.
+ */
+ private static final Logger LOG = LoggerFactory.getLogger(FSBackend.class);
+
+ /**
+ * The maximum last modified time resolution of the file system.
+ */
+ private static final int ACCESS_TIME_RESOLUTION = 2000;
+
+ @Override
+ public void init(CachingDataStore store, String homeDir, String config)
+ throws DataStoreException {
+ super.init(store, homeDir, config);
+ Properties initProps = null;
+ // Check is configuration is already provided. That takes precedence
+ // over config provided via file based config
+ if (this.properties != null) {
+ initProps = this.properties;
+ } else {
+ initProps = new Properties();
+ InputStream in = null;
+ try {
+ in = new FileInputStream(config);
+ initProps.load(in);
+ } catch (IOException e) {
+ throw new DataStoreException(
+ "Could not initialize FSBackend from " + config, e);
+ } finally {
+ IOUtils.closeQuietly(in);
+ }
+ this.properties = initProps;
+ }
+ init(store, homeDir, initProps);
+
+ }
+
+ public void init(CachingDataStore store, String homeDir, Properties prop)
+ throws DataStoreException {
+ setDataStore(store);
+ setHomeDir(homeDir);
+ this.fsPath = prop.getProperty(FS_BACKEND_PATH);
+ if (this.fsPath == null || "".equals(this.fsPath)) {
+ throw new DataStoreException("Could not initialize FSBackend from "
+ + getConfig() + ". [" + FS_BACKEND_PATH + "] property not found.");
+ }
+ fsPathDir = new File(this.fsPath);
+ if (fsPathDir.exists() && fsPathDir.isFile()) {
+ throw new DataStoreException("Can not create a directory "
+ + "because a file exists with the same name: " + this.fsPath);
+ }
+ if (!fsPathDir.exists()) {
+ boolean created = fsPathDir.mkdirs();
+ if (!created) {
+ throw new DataStoreException("Could not create directory: "
+ + fsPathDir.getAbsolutePath());
+ }
+ }
+ }
+
+ @Override
+ public InputStream read(DataIdentifier identifier)
+ throws DataStoreException {
+ File file = getFile(identifier);
+ try {
+ return new LazyFileInputStream(file);
+ } catch (IOException e) {
+ throw new DataStoreException("Error opening input stream of "
+ + file.getAbsolutePath(), e);
+ }
+ }
+
+ @Override
+ public long getLength(DataIdentifier identifier) throws DataStoreException {
+ File file = getFile(identifier);
+ if (file.isFile()) {
+ return file.length();
+ }
+ throw new DataStoreException("Could not length of dataIdentifier ["
+ + identifier + "]");
+ }
+
+ @Override
+ public long getLastModified(DataIdentifier identifier)
+ throws DataStoreException {
+ long start = System.currentTimeMillis();
+ File f = getFile(identifier);
+ if (f.isFile()) {
+ return getLastModified(f);
+ }
+ LOG.info("getLastModified:Identifier [{}] not found. Took [{}] ms.",
+ identifier, (System.currentTimeMillis() - start));
+ throw new DataStoreException("Identifier [" + identifier
+ + "] not found.");
+ }
+
+ @Override
+ public void write(DataIdentifier identifier, File src)
+ throws DataStoreException {
+ File dest = getFile(identifier);
+ synchronized (this) {
+ if (dest.exists()) {
+ long now = System.currentTimeMillis();
+ if (getLastModified(dest) < now + ACCESS_TIME_RESOLUTION) {
+ setLastModified(dest, now + ACCESS_TIME_RESOLUTION);
+ }
+ } else {
+ try {
+ FileUtils.copyFile(src, dest);
+ } catch (IOException ioe) {
+ LOG.error("failed to copy [{}] to [{}]",
+ src.getAbsolutePath(), dest.getAbsolutePath());
+ throw new DataStoreException("Not able to write file ["
+ + identifier + "]", ioe);
+ }
+ }
+ }
+
+ }
+
+ @Override
+ public void writeAsync(final DataIdentifier identifier, final File src,
+ final AsyncUploadCallback callback)
+ throws DataStoreException {
+ if (callback == null) {
+ throw new IllegalArgumentException(
+ "callback parameter cannot be null in asyncUpload");
+ }
+ getAsyncWriteExecutor().execute(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ write(identifier, src);
+ callback.onSuccess(new AsyncUploadResult(identifier, src));
+ } catch (DataStoreException dse) {
+ AsyncUploadResult res = new AsyncUploadResult(identifier,
+ src);
+ res.setException(dse);
+ callback.onFailure(res);
+ }
+
+ }
+ });
+ }
+
+ @Override
+ public Iterator<DataIdentifier> getAllIdentifiers()
+ throws DataStoreException {
+ ArrayList<File> files = new ArrayList<File>();
+ for (File file : fsPathDir.listFiles()) {
+ if (file.isDirectory()) { // skip top-level files
+ listRecursive(files, file);
+ }
+ }
+
+ ArrayList<DataIdentifier> identifiers = new ArrayList<DataIdentifier>();
+ for (File f : files) {
+ String name = f.getName();
+ identifiers.add(new DataIdentifier(name));
+ }
+ LOG.debug("Found " + identifiers.size() + " identifiers.");
+ return identifiers.iterator();
+ }
+
+ @Override
+ public boolean exists(DataIdentifier identifier, boolean touch)
+ throws DataStoreException {
+ File file = getFile(identifier);
+ if (file.isFile()) {
+ if (touch) {
+ long now = System.currentTimeMillis();
+ setLastModified(file, now + ACCESS_TIME_RESOLUTION);
+ }
+ return true;
+ }
+ return false;
+ }
+
+ @Override
+ public boolean exists(DataIdentifier identifier) throws DataStoreException {
+ return exists(identifier, false);
+ }
+
+ @Override
+ public void touch(DataIdentifier identifier, long minModifiedDate)
+ throws DataStoreException {
+ File file = getFile(identifier);
+ long now = System.currentTimeMillis();
+ if (minModifiedDate > 0 && minModifiedDate > getLastModified(file)) {
+ setLastModified(file, now + ACCESS_TIME_RESOLUTION);
+ }
+ }
+
+ @Override
+ public void touchAsync(final DataIdentifier identifier,
+ final long minModifiedDate,
+ final AsyncTouchCallback callback)
+ throws DataStoreException {
+ try {
+ if (callback == null) {
+ throw new IllegalArgumentException(
+ "callback parameter cannot be null in touchAsync");
+ }
+ Thread.currentThread().setContextClassLoader(
+ getClass().getClassLoader());
+
+ getAsyncWriteExecutor().execute(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ touch(identifier, minModifiedDate);
+ callback.onSuccess(new AsyncTouchResult(identifier));
+ } catch (DataStoreException e) {
+ AsyncTouchResult result = new AsyncTouchResult(
+ identifier);
+ result.setException(e);
+ callback.onFailure(result);
+ }
+ }
+ });
+ } catch (Exception e) {
+ if (callback != null) {
+ callback.onAbort(new AsyncTouchResult(identifier));
+ }
+ throw new DataStoreException("Cannot touch the record "
+ + identifier.toString(), e);
+ }
+
+ }
+
+ @Override
+ public Set<DataIdentifier> deleteAllOlderThan(long min)
+ throws DataStoreException {
+ Set<DataIdentifier> deleteIdSet = new HashSet<DataIdentifier>(30);
+ for (File file : fsPathDir.listFiles()) {
+ if (file.isDirectory()) { // skip top-level files
+ deleteOlderRecursive(file, min, deleteIdSet);
+ }
+ }
+ return deleteIdSet;
+ }
+
+ @Override
+ public void deleteRecord(DataIdentifier identifier)
+ throws DataStoreException {
+ File file = getFile(identifier);
+ synchronized (this) {
+ if (file.exists()) {
+ if (file.delete()) {
+ deleteEmptyParentDirs(file);
+ } else {
+ LOG.warn("Failed to delete file " + file.getAbsolutePath());
+ }
+ }
+ }
+ }
+
+ /**
+ * Properties used to configure the backend. If provided explicitly before
+ * init is invoked then these take precedence
+ * @param properties to configure S3Backend
+ */
+ public void setProperties(Properties properties) {
+ this.properties = properties;
+ }
+
+ /**
+ * Returns the identified file. This method implements the pattern used to
+ * avoid problems with too many files in a single directory.
+ * <p>
+ * No sanity checks are performed on the given identifier.
+ * @param identifier data identifier
+ * @return identified file
+ */
+ private File getFile(DataIdentifier identifier) {
+ String string = identifier.toString();
+ File file = this.fsPathDir;
+ file = new File(file, string.substring(0, 2));
+ file = new File(file, string.substring(2, 4));
+ file = new File(file, string.substring(4, 6));
+ return new File(file, string);
+ }
+
+ /**
+ * Set the last modified date of a file, if the file is writable.
+ * @param file the file
+ * @param time the new last modified date
+ * @throws DataStoreException if the file is writable but modifying the date
+ * fails
+ */
+ private static void setLastModified(File file, long time)
+ throws DataStoreException {
+ if (!file.setLastModified(time)) {
+ if (!file.canWrite()) {
+ // if we can't write to the file, so garbage collection will
+ // also not delete it
+ // (read only files or file systems)
+ return;
+ }
+ try {
+ // workaround for Windows: if the file is already open for
+ // reading
+ // (in this or another process), then setting the last modified
+ // date
+ // doesn't work - see also JCR-2872
+ RandomAccessFile r = new RandomAccessFile(file, "rw");
+ try {
+ r.setLength(r.length());
+ } finally {
+ r.close();
+ }
+ } catch (IOException e) {
+ throw new DataStoreException(
+ "An IO Exception occurred while trying to set the last modified date: "
+ + file.getAbsolutePath(), e);
+ }
+ }
+ }
+
+ /**
+ * Get the last modified date of a file.
+ * @param file the file
+ * @return the last modified date
+ * @throws DataStoreException if reading fails
+ */
+ private static long getLastModified(File file) throws DataStoreException {
+ long lastModified = file.lastModified();
+ if (lastModified == 0) {
+ throw new DataStoreException(
+ "Failed to read record modified date: "
+ + file.getAbsolutePath());
+ }
+ return lastModified;
+ }
+
+ private void listRecursive(List<File> list, File file) {
+ File[] files = file.listFiles();
+ if (files != null) {
+ for (File f : files) {
+ if (f.isDirectory()) {
+ listRecursive(list, f);
+ } else {
+ list.add(f);
+ }
+ }
+ }
+ }
+
+ private void deleteEmptyParentDirs(File file) {
+ File parent = file.getParentFile();
+ try {
+ // Only iterate & delete if parent directory of the blob file is
+ // child
+ // of the base directory and if it is empty
+ while (FileUtils.directoryContains(fsPathDir, parent)) {
+ String[] entries = parent.list();
+ if (entries == null) {
+ LOG.warn("Failed to list directory {}",
+ parent.getAbsolutePath());
+ break;
+ }
+ if (entries.length > 0) {
+ break;
+ }
+ boolean deleted = parent.delete();
+ LOG.debug("Deleted parent [{}] of file [{}]: {}", new Object[] {
+ parent, file.getAbsolutePath(), deleted });
+ parent = parent.getParentFile();
+ }
+ } catch (IOException e) {
+ LOG.warn("Error in parents deletion for " + file.getAbsoluteFile(),
+ e);
+ }
+ }
+
+ private void deleteOlderRecursive(File file, long min,
+ Set<DataIdentifier> deleteIdSet) throws DataStoreException {
+ if (file.isFile() && file.exists() && file.canWrite()) {
+ synchronized (this) {
+ long lastModified;
+ try {
+ lastModified = getLastModified(file);
+ } catch (DataStoreException e) {
+ LOG.warn(
+ "Failed to read modification date; file not deleted", e);
+ // don't delete the file, since the lastModified date is
+ // uncertain
+ lastModified = min;
+ }
+ if (lastModified < min) {
+ DataIdentifier id = new DataIdentifier(file.getName());
+ if (getDataStore().confirmDelete(id)) {
+ getDataStore().deleteFromCache(id);
+ if (LOG.isInfoEnabled()) {
+ LOG.info("Deleting old file "
+ + file.getAbsolutePath() + " modified: "
+ + new Timestamp(lastModified).toString()
+ + " length: " + file.length());
+ }
+ if (file.delete()) {
+ deleteIdSet.add(id);
+ } else {
+ LOG.warn("Failed to delete old file "
+ + file.getAbsolutePath());
+ }
+ }
+ }
+ }
+ } else if (file.isDirectory()) {
+ File[] list = file.listFiles();
+ if (list != null) {
+ for (File f : list) {
+ deleteOlderRecursive(f, min, deleteIdSet);
+ }
+ }
+
+ // JCR-1396: FileDataStore Garbage Collector and empty directories
+ // Automatic removal of empty directories (but not the root!)
+ synchronized (this) {
+ list = file.listFiles();
+ if (list != null && list.length == 0) {
+ file.delete();
+ }
+ }
+ }
+ }
+}
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/data/FileDataRecord.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/FileDataRecord.java
new file mode 100644
index 0000000000..df0161db10
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/FileDataRecord.java
@@ -0,0 +1,71 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.jackrabbit.core.data;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+
+
+/**
+ * Data record that is based on a normal file.
+ */
+public class FileDataRecord extends AbstractDataRecord {
+
+ /**
+ * The file that contains the binary stream.
+ */
+ private final File file;
+
+ /**
+ * Creates a data record based on the given identifier and file.
+ *
+ * @param identifier data identifier
+ * @param file file that contains the binary stream
+ */
+ public FileDataRecord(
+ AbstractDataStore store, DataIdentifier identifier, File file) {
+ super(store, identifier);
+ assert file.isFile();
+ this.file = file;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public long getLength() {
+ return file.length();
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public InputStream getStream() throws DataStoreException {
+ try {
+ return new LazyFileInputStream(file);
+ } catch (IOException e) {
+ throw new DataStoreException("Error opening input stream of " + file.getAbsolutePath(), e);
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public long getLastModified() {
+ return file.lastModified();
+ }
+}
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/data/FileDataStore.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/FileDataStore.java
new file mode 100644
index 0000000000..397a805bb1
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/FileDataStore.java
@@ -0,0 +1,500 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.jackrabbit.core.data;
+
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.io.RandomAccessFile;
+import java.lang.ref.WeakReference;
+import java.nio.file.CopyOption;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardCopyOption;
+import java.security.DigestOutputStream;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.WeakHashMap;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.commons.io.IOUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Simple file-based data store. Data records are stored as normal files
+ * named using a message digest of the contained binary stream.
+ *
+ * Configuration:
+ * <pre>
+ * <DataStore class="org.apache.jackrabbit.core.data.FileDataStore">
+ * <param name="{@link #setPath(String) path}" value="/data/datastore"/>
+ * <param name="{@link #setMinRecordLength(int) minRecordLength}" value="1024"/>
+ * </DataStore>
+ * </pre>
+ * <p>
+ * If the directory is not set, the directory <repository home>/repository/datastore is used.
+ * <p>
+ * A three level directory structure is used to avoid placing too many
+ * files in a single directory. The chosen structure is designed to scale
+ * up to billions of distinct records.
+ * <p>
+ * This implementation relies on the underlying file system to support
+ * atomic O(1) move operations with {@link Files#move(Path, Path, CopyOption...)}.
+ */
+public class FileDataStore extends AbstractDataStore
+ implements MultiDataStoreAware {
+
+ /**
+ * Logger instance
+ */
+ private static Logger log = LoggerFactory.getLogger(FileDataStore.class);
+
+ /**
+ * The default value for the minimum object size.
+ */
+ private static final int DEFAULT_MIN_RECORD_LENGTH = 100;
+
+ /**
+ * The maximum last modified time resolution of the file system.
+ */
+ private static final int ACCESS_TIME_RESOLUTION = 2000;
+
+ /**
+ * Name of the directory used for temporary files.
+ * Must be at least 3 characters.
+ */
+ private static final String TMP = "tmp";
+
+ /**
+ * The minimum modified date. If a file is accessed (read or write) with a modified date
+ * older than this value, the modified date is updated to the current time.
+ */
+ private volatile long minModifiedDate;
+
+ /**
+ * The directory that contains all the data record files. The structure
+ * of content within this directory is controlled by this class.
+ */
+ private File directory;
+
+ /**
+ * The name of the directory that contains all the data record files. The structure
+ * of content within this directory is controlled by this class.
+ */
+ private String path;
+
+ /**
+ * The minimum size of an object that should be stored in this data store.
+ */
+ private int minRecordLength = DEFAULT_MIN_RECORD_LENGTH;
+
+ /**
+ * All data identifiers that are currently in use are in this set until they are garbage collected.
+ */
+ protected Map<DataIdentifier, WeakReference<DataIdentifier>> inUse =
+ Collections.synchronizedMap(new WeakHashMap<DataIdentifier, WeakReference<DataIdentifier>>());
+
+ /**
+ * Initialized the data store.
+ * If the path is not set, <repository home>/repository/datastore is used.
+ * This directory is automatically created if it does not yet exist.
+ *
+ * @param homeDir
+ */
+ public void init(String homeDir) {
+ if (path == null) {
+ path = homeDir + "/repository/datastore";
+ }
+ directory = new File(path);
+ directory.mkdirs();
+ }
+
+ /**
+ * Get a data record for the given identifier.
+ *
+ * @param identifier the identifier
+ * @return the data record or null
+ */
+ public DataRecord getRecordIfStored(DataIdentifier identifier) throws DataStoreException {
+ File file = getFile(identifier);
+ if (!file.exists()) {
+ return null;
+ }
+ if (minModifiedDate != 0) {
+ // only check when running garbage collection
+ synchronized (this) {
+ if (getLastModified(file) < minModifiedDate) {
+ setLastModified(file, System.currentTimeMillis() + ACCESS_TIME_RESOLUTION);
+ }
+ }
+ }
+ usesIdentifier(identifier);
+ return new FileDataRecord(this, identifier, file);
+ }
+
+ private void usesIdentifier(DataIdentifier identifier) {
+ inUse.put(identifier, new WeakReference<DataIdentifier>(identifier));
+ }
+
+ /**
+ * Creates a new data record.
+ * The stream is first consumed and the contents are saved in a temporary file
+ * and the {@link #DIGEST} message digest of the stream is calculated. If a
+ * record with the same {@link #DIGEST} digest (and length) is found then it is
+ * returned. Otherwise the temporary file is moved in place to become
+ * the new data record that gets returned.
+ *
+ * @param input binary stream
+ * @return data record that contains the given stream
+ * @throws DataStoreException if the record could not be created
+ */
+ public DataRecord addRecord(InputStream input) throws DataStoreException {
+ File temporary = null;
+ try {
+ temporary = newTemporaryFile();
+ DataIdentifier tempId = new DataIdentifier(temporary.getName());
+ usesIdentifier(tempId);
+ // Copy the stream to the temporary file and calculate the
+ // stream length and the message digest of the stream
+ long length = 0;
+ MessageDigest digest = MessageDigest.getInstance(DIGEST);
+ OutputStream output = new DigestOutputStream(
+ new FileOutputStream(temporary), digest);
+ try {
+ length = IOUtils.copyLarge(input, output);
+ } finally {
+ output.close();
+ }
+ DataIdentifier identifier =
+ new DataIdentifier(encodeHexString(digest.digest()));
+ File file;
+
+ synchronized (this) {
+ // Check if the same record already exists, or
+ // move the temporary file in place if needed
+ usesIdentifier(identifier);
+ file = getFile(identifier);
+ if (!file.exists()) {
+ File parent = file.getParentFile();
+ parent.mkdirs();
+ Files.move(temporary.toPath(), file.toPath(), StandardCopyOption.ATOMIC_MOVE);
+ // no longer need to delete the temporary file
+ temporary = null;
+ } else {
+ long now = System.currentTimeMillis();
+ if (getLastModified(file) < now + ACCESS_TIME_RESOLUTION) {
+ setLastModified(file, now + ACCESS_TIME_RESOLUTION);
+ }
+ }
+ if (file.length() != length) {
+ // Sanity checks on the record file. These should never fail,
+ // but better safe than sorry...
+ if (!file.isFile()) {
+ throw new IOException("Not a file: " + file);
+ }
+ throw new IOException(DIGEST + " collision: " + file);
+ }
+ }
+ // this will also make sure that
+ // tempId is not garbage collected until here
+ inUse.remove(tempId);
+ return new FileDataRecord(this, identifier, file);
+ } catch (NoSuchAlgorithmException e) {
+ throw new DataStoreException(DIGEST + " not available", e);
+ } catch (IOException e) {
+ throw new DataStoreException("Could not add record", e);
+ } finally {
+ if (temporary != null) {
+ temporary.delete();
+ }
+ }
+ }
+
+ /**
+ * Returns the identified file. This method implements the pattern
+ * used to avoid problems with too many files in a single directory.
+ * <p>
+ * No sanity checks are performed on the given identifier.
+ *
+ * @param identifier data identifier
+ * @return identified file
+ */
+ private File getFile(DataIdentifier identifier) {
+ usesIdentifier(identifier);
+ String string = identifier.toString();
+ File file = directory;
+ file = new File(file, string.substring(0, 2));
+ file = new File(file, string.substring(2, 4));
+ file = new File(file, string.substring(4, 6));
+ return new File(file, string);
+ }
+
+ /**
+ * Returns a unique temporary file to be used for creating a new
+ * data record.
+ *
+ * @return temporary file
+ * @throws IOException
+ */
+ private File newTemporaryFile() throws IOException {
+ // the directory is already created in the init method
+ return File.createTempFile(TMP, null, directory);
+ }
+
+ public void updateModifiedDateOnAccess(long before) {
+ minModifiedDate = before;
+ }
+
+ public void deleteRecord(DataIdentifier identifier)
+ throws DataStoreException {
+ File file = getFile(identifier);
+ synchronized (this) {
+ if (file.exists()) {
+ if (file.delete()) {
+ deleteEmptyParentDirs(file);
+ } else {
+ log.warn("Failed to delete file " + file.getAbsolutePath());
+ }
+ }
+ }
+ }
+
+ private void deleteEmptyParentDirs(File file) {
+ File parent = file.getParentFile();
+ try {
+ // Only iterate & delete if parent directory of the blob file is child
+ // of the base directory and if it is empty
+ while (FileUtils.directoryContains(directory, parent)) {
+ String[] entries = parent.list();
+ if (entries == null) {
+ log.warn("Failed to list directory {}", parent.getAbsolutePath());
+ break;
+ }
+ if (entries.length > 0) {
+ break;
+ }
+ boolean deleted = parent.delete();
+ log.debug("Deleted parent [{}] of file [{}]: {}",
+ new Object[]{parent, file.getAbsolutePath(), deleted});
+ parent = parent.getParentFile();
+ }
+ } catch (IOException e) {
+ log.warn("Error in parents deletion for " + file.getAbsoluteFile(), e);
+ }
+ }
+
+ public int deleteAllOlderThan(long min) {
+ int count = 0;
+ for (File file : directory.listFiles()) {
+ if (file.isDirectory()) { // skip top-level files
+ count += deleteOlderRecursive(file, min);
+ }
+ }
+ return count;
+ }
+
+ private int deleteOlderRecursive(File file, long min) {
+ int count = 0;
+ if (file.isFile() && file.exists() && file.canWrite()) {
+ synchronized (this) {
+ long lastModified;
+ try {
+ lastModified = getLastModified(file);
+ } catch (DataStoreException e) {
+ log.warn("Failed to read modification date; file not deleted", e);
+ // don't delete the file, since the lastModified date is uncertain
+ lastModified = min;
+ }
+ if (lastModified < min) {
+ DataIdentifier id = new DataIdentifier(file.getName());
+ if (!inUse.containsKey(id)) {
+ if (log.isInfoEnabled()) {
+ log.info("Deleting old file " + file.getAbsolutePath() +
+ " modified: " + new Timestamp(lastModified).toString() +
+ " length: " + file.length());
+ }
+ if (!file.delete()) {
+ log.warn("Failed to delete old file " + file.getAbsolutePath());
+ }
+ count++;
+ }
+ }
+ }
+ } else if (file.isDirectory()) {
+ File[] list = file.listFiles();
+ if (list != null) {
+ for (File f: list) {
+ count += deleteOlderRecursive(f, min);
+ }
+ }
+
+ // JCR-1396: FileDataStore Garbage Collector and empty directories
+ // Automatic removal of empty directories (but not the root!)
+ synchronized (this) {
+ list = file.listFiles();
+ if (list != null && list.length == 0) {
+ file.delete();
+ }
+ }
+ }
+ return count;
+ }
+
+ private void listRecursive(List<File> list, File file) {
+ File[] files = file.listFiles();
+ if (files != null) {
+ for (File f : files) {
+ if (f.isDirectory()) {
+ listRecursive(list, f);
+ } else {
+ list.add(f);
+ }
+ }
+ }
+ }
+
+ public Iterator<DataIdentifier> getAllIdentifiers() {
+ ArrayList<File> files = new ArrayList<File>();
+ for (File file : directory.listFiles()) {
+ if (file.isDirectory()) { // skip top-level files
+ listRecursive(files, file);
+ }
+ }
+
+ ArrayList<DataIdentifier> identifiers = new ArrayList<DataIdentifier>();
+ for (File f: files) {
+ String name = f.getName();
+ identifiers.add(new DataIdentifier(name));
+ }
+ log.debug("Found " + identifiers.size() + " identifiers.");
+ return identifiers.iterator();
+ }
+
+ public void clearInUse() {
+ inUse.clear();
+ }
+
+ /**
+ * Get the name of the directory where this data store keeps the files.
+ *
+ * @return the full path name
+ */
+ public String getPath() {
+ return path;
+ }
+
+ /**
+ * Set the name of the directory where this data store keeps the files.
+ *
+ * @param directoryName the path name
+ */
+ public void setPath(String directoryName) {
+ this.path = directoryName;
+ }
+
+ public int getMinRecordLength() {
+ return minRecordLength;
+ }
+
+ /**
+ * Set the minimum object length.
+ *
+ * @param minRecordLength the length
+ */
+ public void setMinRecordLength(int minRecordLength) {
+ this.minRecordLength = minRecordLength;
+ }
+
+ public void close() {
+ // nothing to do
+ }
+
+ //---------------------------------------------------------< protected >--
+
+ @Override
+ protected byte[] getOrCreateReferenceKey() throws DataStoreException {
+ File file = new File(directory, "reference.key");
+ try {
+ if (file.exists()) {
+ return FileUtils.readFileToByteArray(file);
+ } else {
+ byte[] key = super.getOrCreateReferenceKey();
+ FileUtils.writeByteArrayToFile(file, key);
+ return key;
+ }
+ } catch (IOException e) {
+ throw new DataStoreException(
+ "Unable to access reference key file " + file.getPath(), e);
+ }
+ }
+
+ //-----------------------------------------------------------< private >--
+
+ /**
+ * Get the last modified date of a file.
+ *
+ * @param file the file
+ * @return the last modified date
+ * @throws DataStoreException if reading fails
+ */
+ private static long getLastModified(File file) throws DataStoreException {
+ long lastModified = file.lastModified();
+ if (lastModified == 0) {
+ throw new DataStoreException("Failed to read record modified date: " + file.getAbsolutePath());
+ }
+ return lastModified;
+ }
+
+ /**
+ * Set the last modified date of a file, if the file is writable.
+ *
+ * @param file the file
+ * @param time the new last modified date
+ * @throws DataStoreException if the file is writable but modifying the date fails
+ */
+ private static void setLastModified(File file, long time) throws DataStoreException {
+ if (!file.setLastModified(time)) {
+ if (!file.canWrite()) {
+ // if we can't write to the file, so garbage collection will also not delete it
+ // (read only files or file systems)
+ return;
+ }
+ try {
+ // workaround for Windows: if the file is already open for reading
+ // (in this or another process), then setting the last modified date
+ // doesn't work - see also JCR-2872
+ RandomAccessFile r = new RandomAccessFile(file, "rw");
+ try {
+ r.setLength(r.length());
+ } finally {
+ r.close();
+ }
+ } catch (IOException e) {
+ throw new DataStoreException("An IO Exception occurred while trying to set the last modified date: " + file.getAbsolutePath(), e);
+ }
+ }
+ }
+}
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/data/LazyFileInputStream.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/LazyFileInputStream.java
new file mode 100644
index 0000000000..a9c0489b04
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/LazyFileInputStream.java
@@ -0,0 +1,167 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.jackrabbit.core.data;
+
+import java.io.File;
+import java.io.FileDescriptor;
+import java.io.FileInputStream;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+
+import org.apache.commons.io.input.AutoCloseInputStream;
+
+/**
+ * This input stream delays opening the file until the first byte is read, and
+ * closes and discards the underlying stream as soon as the end of input has
+ * been reached or when the stream is explicitly closed.
+ */
+public class LazyFileInputStream extends AutoCloseInputStream {
+
+ /**
+ * The file descriptor to use.
+ */
+ protected final FileDescriptor fd;
+
+ /**
+ * The file to read from.
+ */
+ protected final File file;
+
+ /**
+ * True if the input stream was opened. It is also set to true if the stream
+ * was closed without reading (to avoid opening the file after the stream
+ * was closed).
+ */
+ protected boolean opened;
+
+ /**
+ * Creates a new <code>LazyFileInputStream</code> for the given file. If the
+ * file is unreadable, a FileNotFoundException is thrown.
+ * The file is not opened until the first byte is read from the stream.
+ *
+ * @param file the file
+ * @throws java.io.FileNotFoundException
+ */
+ public LazyFileInputStream(File file)
+ throws FileNotFoundException {
+ super(null);
+ if (!file.canRead()) {
+ throw new FileNotFoundException(file.getPath());
+ }
+ this.file = file;
+ this.fd = null;
+ }
+
+ /**
+ * Creates a new <code>LazyFileInputStream</code> for the given file
+ * descriptor.
+ * The file is not opened until the first byte is read from the stream.
+ *
+ * @param fd
+ */
+ public LazyFileInputStream(FileDescriptor fd) {
+ super(null);
+ this.file = null;
+ this.fd = fd;
+ }
+
+ /**
+ * Creates a new <code>LazyFileInputStream</code> for the given file. If the
+ * file is unreadable, a FileNotFoundException is thrown.
+ *
+ * @param name
+ * @throws java.io.FileNotFoundException
+ */
+ public LazyFileInputStream(String name) throws FileNotFoundException {
+ this(new File(name));
+ }
+
+ /**
+ * Open the stream if required.
+ *
+ * @throws java.io.IOException
+ */
+ protected void open() throws IOException {
+ if (!opened) {
+ opened = true;
+ if (fd != null) {
+ in = new FileInputStream(fd);
+ } else {
+ in = new FileInputStream(file);
+ }
+ }
+ }
+
+ public int read() throws IOException {
+ open();
+ return super.read();
+ }
+
+ public int available() throws IOException {
+ open();
+ return super.available();
+ }
+
+ public void close() throws IOException {
+ // make sure the file is not opened afterwards
+ opened = true;
+
+ // only close the file if it was in fact opened
+ if (in != null) {
+ super.close();
+ }
+ }
+
+ public synchronized void reset() throws IOException {
+ open();
+ super.reset();
+ }
+
+ public boolean markSupported() {
+ try {
+ open();
+ } catch (IOException e) {
+ throw new IllegalStateException(e.toString());
+ }
+ return super.markSupported();
+ }
+
+ public synchronized void mark(int readlimit) {
+ try {
+ open();
+ } catch (IOException e) {
+ throw new IllegalStateException(e.toString());
+ }
+ super.mark(readlimit);
+ }
+
+ public long skip(long n) throws IOException {
+ open();
+ return super.skip(n);
+ }
+
+ public int read(byte[] b) throws IOException {
+ open();
+ return super.read(b, 0, b.length);
+ }
+
+ public int read(byte[] b, int off, int len) throws IOException {
+ open();
+ return super.read(b, off, len);
+ }
+
+}
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/data/LocalCache.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/LocalCache.java
new file mode 100644
index 0000000000..f043908784
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/LocalCache.java
@@ -0,0 +1,668 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.jackrabbit.core.data;
+
+import java.io.BufferedOutputStream;
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.commons.io.IOUtils;
+import org.apache.jackrabbit.util.TransientFileFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This class implements a LRU cache used by {@link CachingDataStore}. If cache
+ * size exceeds limit, this cache goes in purge mode. In purge mode any
+ * operation to cache is no-op. After purge cache size would be less than
+ * cachePurgeResizeFactor * maximum size.
+ */
+public class LocalCache {
+
+ /**
+ * Logger instance.
+ */
+ static final Logger LOG = LoggerFactory.getLogger(LocalCache.class);
+
+ /**
+ * The file names of the files that need to be deleted.
+ */
+ final Set<String> toBeDeleted = new HashSet<String>();
+
+ /**
+ * The filename Vs file size LRU cache.
+ */
+ LRUCache cache;
+
+ /**
+ * The directory where the files are created.
+ */
+ private final File directory;
+
+ /**
+ * The directory where tmp files are created.
+ */
+ private final File tmp;
+
+ /**
+ * If true cache is in purgeMode and not available. All operation would be
+ * no-op.
+ */
+ private volatile boolean purgeMode;
+
+ private AsyncUploadCache asyncUploadCache;
+
+ private AtomicLong cacheMissCounter = new AtomicLong();
+
+ private AtomicLong cacheMissDuration = new AtomicLong();
+
+
+ /**
+ * Build LRU cache of files located at 'path'. It uses lastModified property
+ * of file to build LRU cache. If cache size exceeds limit size, this cache
+ * goes in purge mode. In purge mode any operation to cache is no-op.
+ *
+ * @param path file system path
+ * @param tmpPath temporary directory used by cache.
+ * @param maxSizeInBytes maximum size of cache.
+ * @param cachePurgeTrigFactor factor which triggers cache to purge mode.
+ * That is if current size exceed (cachePurgeTrigFactor * maxSizeInBytes), the
+ * cache will go in auto-purge mode.
+ * @param cachePurgeResizeFactor after cache purge size of cache will be
+ * just less (cachePurgeResizeFactor * maxSizeInBytes).
+ * @param asyncUploadCache {@link AsyncUploadCache}
+ */
+ public LocalCache(String path, String tmpPath, long maxSizeInBytes, double cachePurgeTrigFactor,
+ double cachePurgeResizeFactor, AsyncUploadCache asyncUploadCache) {
+ directory = new File(path);
+ tmp = new File(tmpPath);
+ LOG.info(
+ "cachePurgeTrigFactor =[{}], cachePurgeResizeFactor =[{}], " +
+ "cachePurgeTrigFactorSize =[{}], cachePurgeResizeFactorSize =[{}]",
+ new Object[] { cachePurgeTrigFactor, cachePurgeResizeFactor,
+ (cachePurgeTrigFactor * maxSizeInBytes),
+ (cachePurgeResizeFactor * maxSizeInBytes) });
+ cache = new LRUCache(maxSizeInBytes, cachePurgeTrigFactor, cachePurgeResizeFactor);
+ this.asyncUploadCache = asyncUploadCache;
+ new Thread(new CacheBuildJob()).start();
+ }
+
+ /**
+ * Store an item in the cache and return the input stream. If cache is in
+ * purgeMode or file doesn't exists, inputstream from a
+ * {@link TransientFileFactory#createTransientFile(String, String, File)} is
+ * returned. Otherwise inputStream from cached file is returned. This method
+ * doesn't close the incoming inputstream.
+ *
+ * @param fileName the key of cache.
+ * @param in {@link InputStream}
+ * @return the (new) input stream.
+ */
+ public InputStream store(String fileName, final InputStream in)
+ throws IOException {
+ fileName = fileName.replace("\\", "/");
+ File f = getFile(fileName);
+ long length = 0;
+ if (!f.exists() || isInPurgeMode()) {
+ OutputStream out = null;
+ File transFile = null;
+ try {
+ TransientFileFactory tff = TransientFileFactory.getInstance();
+ transFile = tff.createTransientFile("s3-", "tmp", tmp);
+ out = new BufferedOutputStream(new FileOutputStream(transFile));
+ length = IOUtils.copyLarge(in, out);
+ } finally {
+ IOUtils.closeQuietly(out);
+ }
+ // rename the file to local fs cache
+ if (canAdmitFile(length)
+ && (f.getParentFile().exists() || f.getParentFile().mkdirs())
+ && transFile.renameTo(f) && f.exists()) {
+ if (transFile.exists() && transFile.delete()) {
+ LOG.info("tmp file [{}] not deleted successfully",
+ transFile.getAbsolutePath());
+ }
+ transFile = null;
+ LOG.debug(
+ "file [{}] doesn't exists. adding to local cache using inputstream.",
+ fileName);
+ cache.put(fileName, f.length());
+ } else {
+ LOG.debug(
+ "file [{}] doesn't exists. returning transient file [{}].",
+ fileName, transFile.getAbsolutePath());
+ f = transFile;
+ }
+ } else {
+ if (in instanceof BackendResourceAbortable) {
+ ((BackendResourceAbortable) in).abort();
+ }
+ f.setLastModified(System.currentTimeMillis());
+ LOG.debug(
+ "file [{}] exists. adding to local cache using inputstream.",
+ fileName);
+ cache.put(fileName, f.length());
+ }
+ tryPurge();
+ return new LazyFileInputStream(f);
+ }
+
+ /**
+ * Store an item along with file in cache. Cache size is increased by
+ * {@link File#length()} If file already exists in cache,
+ * {@link File#setLastModified(long)} is updated with current time.
+ *
+ * @param fileName the key of cache.
+ * @param src file to be added to cache.
+ */
+ public File store(String fileName, final File src) {
+ try {
+ return store(fileName, src, false).getFile();
+ } catch (IOException ioe) {
+ LOG.warn("Exception in addding file [" + fileName + "] to local cache.", ioe);
+ }
+ return null;
+ }
+
+ /**
+ * This method add file to {@link LocalCache} and tries that file can be
+ * added to {@link AsyncUploadCache}. If file is added to
+ * {@link AsyncUploadCache} successfully, it sets
+ * {@link AsyncUploadCacheResult#setAsyncUpload(boolean)} to true.
+ *
+ * @param fileName name of the file.
+ * @param src source file.
+ * @param tryForAsyncUpload If true it tries to add fileName to
+ * {@link AsyncUploadCache}
+ * @return {@link AsyncUploadCacheResult}. This method sets
+ * {@link AsyncUploadCacheResult#setAsyncUpload(boolean)} to true, if
+ * fileName is added to {@link AsyncUploadCache} successfully else
+ * it sets {@link AsyncUploadCacheResult#setAsyncUpload(boolean)} to
+ * false. {@link AsyncUploadCacheResult#getFile()} contains cached
+ * file, if it is added to {@link LocalCache} or original file.
+ * @throws IOException
+ */
+ public AsyncUploadCacheResult store(String fileName, File src,
+ boolean tryForAsyncUpload) throws IOException {
+ fileName = fileName.replace("\\", "/");
+ File dest = getFile(fileName);
+ File parent = dest.getParentFile();
+ AsyncUploadCacheResult result = new AsyncUploadCacheResult();
+ result.setFile(src);
+ result.setAsyncUpload(false);
+ boolean destExists = false;
+ if ((destExists = dest.exists())
+ || (src.exists() && !dest.exists() && !src.equals(dest)
+ && canAdmitFile(src.length())
+ && (parent.exists() || parent.mkdirs()) && (src.renameTo(dest)))) {
+ if (destExists) {
+ dest.setLastModified(System.currentTimeMillis());
+ }
+ LOG.debug("file [{}] moved to [{}] ", src.getAbsolutePath(), dest.getAbsolutePath());
+ LOG.debug(
+ "file [{}] exists= [{}] added to local cache, isLastModified [{}]",
+ new Object[] { dest.getAbsolutePath(), dest.exists(),
+ destExists });
+
+ cache.put(fileName, dest.length());
+ result.setFile(dest);
+ if (tryForAsyncUpload) {
+ result.setAsyncUpload(asyncUploadCache.add(fileName).canAsyncUpload());
+ }
+ } else {
+ LOG.info("file [{}] exists= [{}] not added to local cache.",
+ fileName, destExists);
+ }
+ tryPurge();
+ return result;
+ }
+ /**
+ * Return the inputstream from from cache, or null if not in the cache.
+ *
+ * @param fileName name of file.
+ * @return stream or null.
+ */
+ public InputStream getIfStored(String fileName) throws IOException {
+ File file = getFileIfStored(fileName);
+ return file == null ? null : new LazyFileInputStream(file);
+ }
+
+ public File getFileIfStored(String fileName) throws IOException {
+ fileName = fileName.replace("\\", "/");
+ File f = getFile(fileName);
+ long diff = (System.currentTimeMillis() - cacheMissDuration.get()) / 1000;
+ // logged at 5 minute interval minimum
+ if (diff > 5 * 60) {
+ LOG.info("local cache misses [{}] in [{}] sec", new Object[] {
+ cacheMissCounter.getAndSet(0), diff });
+ cacheMissDuration.set(System.currentTimeMillis());
+ }
+
+ // return file in purge mode = true and file present in asyncUploadCache
+ // as asyncUploadCache's files will be not be deleted in cache purge.
+ if (!f.exists() || (isInPurgeMode() && !asyncUploadCache.hasEntry(fileName, false))) {
+ LOG.debug(
+ "getFileIfStored returned: purgeMode=[{}], file=[{}] exists=[{}]",
+ new Object[] { isInPurgeMode(), f.getAbsolutePath(), f.exists() });
+ cacheMissCounter.incrementAndGet();
+ return null;
+ } else {
+ // touch entry in LRU caches
+ f.setLastModified(System.currentTimeMillis());
+ cache.get(fileName);
+ return f;
+ }
+ }
+
+ /**
+ * Delete file from cache. Size of cache is reduced by file length. The
+ * method is no-op if file doesn't exist in cache.
+ *
+ * @param fileName file name that need to be removed from cache.
+ */
+ public void delete(String fileName) {
+ if (isInPurgeMode()) {
+ LOG.debug("purgeMode true :delete returned");
+ return;
+ }
+ fileName = fileName.replace("\\", "/");
+ cache.remove(fileName);
+ }
+
+ /**
+ * Returns length of file if exists in cache else returns null.
+ * @param fileName name of the file.
+ */
+ public Long getFileLength(String fileName) {
+ Long length = null;
+ try {
+ length = cache.get(fileName);
+ if( length == null ) {
+ File f = getFileIfStored(fileName);
+ if (f != null) {
+ length = f.length();
+ }
+ }
+ } catch (IOException ignore) {
+
+ }
+ return length;
+ }
+
+ /**
+ * Close the cache. Cache maintain set of files which it was not able to
+ * delete successfully. This method will an attempt to delete all
+ * unsuccessful delete files.
+ */
+ public void close() {
+ LOG.debug("close");
+ deleteOldFiles();
+ }
+
+ /**
+ * Check if cache can admit file of given length.
+ * @param length of the file.
+ * @return true if yes else return false.
+ */
+ private boolean canAdmitFile(final long length) {
+ //order is important here
+ boolean value = !isInPurgeMode() && (cache.canAdmitFile(length));
+ if (!value) {
+ LOG.debug("cannot admit file of length=[{}] and currentSizeInBytes=[{}] ",
+ length, cache.currentSizeInBytes);
+ }
+ return value;
+ }
+
+ /**
+ * Return true if cache is in purge mode else return false.
+ */
+ synchronized boolean isInPurgeMode() {
+ return purgeMode;
+ }
+
+ /**
+ * Set purge mode. If set to true all cache operation will be no-op. If set
+ * to false, all operations to cache are available.
+ *
+ * @param purgeMode purge mode
+ */
+ synchronized void setPurgeMode(final boolean purgeMode) {
+ this.purgeMode = purgeMode;
+ }
+
+ File getFile(final String fileName) {
+ return new File(directory, fileName);
+ }
+
+ private void deleteOldFiles() {
+ int initialSize = toBeDeleted.size();
+ int count = 0;
+ for (String fileName : new ArrayList<String>(toBeDeleted)) {
+ fileName = fileName.replace("\\", "/");
+ if( cache.remove(fileName) != null) {
+ count++;
+ }
+ }
+ LOG.info("deleted [{}]/[{}] files.", count, initialSize);
+ }
+
+ /**
+ * This method tries to delete a file. If it is not able to delete file due
+ * to any reason, it add it toBeDeleted list.
+ *
+ * @param fileName name of the file which will be deleted.
+ * @return true if this method deletes file successfuly else return false.
+ */
+ boolean tryDelete(final String fileName) {
+ LOG.debug("try deleting file [{}]", fileName);
+ File f = getFile(fileName);
+ if (f.exists() && f.delete()) {
+ LOG.info("File [{}] deleted successfully", f.getAbsolutePath());
+ toBeDeleted.remove(fileName);
+ while (true) {
+ f = f.getParentFile();
+ if (f.equals(directory) || f.list().length > 0) {
+ break;
+ }
+ // delete empty parent folders (except the main directory)
+ f.delete();
+ }
+ return true;
+ } else if (f.exists()) {
+ LOG.info("not able to delete file [{}]", f.getAbsolutePath());
+ toBeDeleted.add(fileName);
+ return false;
+ }
+ return true;
+ }
+
+ static int maxSizeElements(final long bytes) {
+ // after a CQ installation, the average item in
+ // the data store is about 52 KB
+ int count = (int) (bytes / 65535);
+ count = Math.max(1024, count);
+ count = Math.min(64 * 1024, count);
+ return count;
+ }
+
+ /**
+ * This method tries purging of local cache. It checks if local cache
+ * has exceeded the defined limit then it triggers purge cache job in a
+ * seperate thread.
+ */
+ synchronized void tryPurge() {
+ if (!isInPurgeMode()
+ && cache.currentSizeInBytes > cache.cachePurgeTrigSize) {
+ setPurgeMode(true);
+ LOG.info(
+ "cache.entries = [{}], currentSizeInBytes=[{}] exceeds cachePurgeTrigSize=[{}]",
+ new Object[] { cache.size(), cache.currentSizeInBytes,
+ cache.cachePurgeTrigSize });
+ new Thread(new PurgeJob()).start();
+ } else {
+ LOG.debug(
+ "currentSizeInBytes=[{}],cachePurgeTrigSize=[{}], isInPurgeMode =[{}]",
+ new Object[] { cache.currentSizeInBytes,
+ cache.cachePurgeTrigSize, isInPurgeMode() });
+ }
+ }
+
+ /**
+ * A LRU based extension {@link LinkedHashMap}. The key is file name and
+ * value is length of file.
+ */
+ private class LRUCache extends LinkedHashMap<String, Long> {
+ private static final long serialVersionUID = 1L;
+
+ volatile long currentSizeInBytes;
+
+ final long maxSizeInBytes;
+
+ final long cachePurgeResize;
+
+ final long cachePurgeTrigSize;
+
+ LRUCache(final long maxSizeInBytes,
+ final double cachePurgeTrigFactor,
+ final double cachePurgeResizeFactor) {
+ super(maxSizeElements(maxSizeInBytes), (float) 0.75, true);
+ this.maxSizeInBytes = maxSizeInBytes;
+ this.cachePurgeTrigSize = new Double(cachePurgeTrigFactor
+ * maxSizeInBytes).longValue();
+ this.cachePurgeResize = new Double(cachePurgeResizeFactor
+ * maxSizeInBytes).longValue();
+ }
+
+ /**
+ * Overridden {@link Map#remove(Object)} to delete corresponding file
+ * from file system.
+ */
+ @Override
+ public synchronized Long remove(final Object key) {
+ String fileName = (String) key;
+ fileName = fileName.replace("\\", "/");
+ try {
+ // not removing file from local cache, if there is in progress
+ // async upload on it.
+ if (asyncUploadCache.hasEntry(fileName, false)) {
+ LOG.info(
+ "AsyncUploadCache upload contains file [{}]. Not removing it from LocalCache.",
+ fileName);
+ return null;
+ }
+ } catch (IOException e) {
+ LOG.debug("error: ", e);
+ return null;
+ }
+ Long flength = null;
+ if (tryDelete(fileName)) {
+ flength = super.remove(key);
+ if (flength != null) {
+ LOG.debug("cache entry [{}], with size [{}] removed.",
+ fileName, flength);
+ currentSizeInBytes -= flength.longValue();
+ }
+ } else if (!getFile(fileName).exists()) {
+ // second attempt. remove from cache if file doesn't exists
+ flength = super.remove(key);
+ if (flength != null) {
+ LOG.debug(
+ "file not exists. cache entry [{}], with size [{}] removed.",
+ fileName, flength);
+ currentSizeInBytes -= flength.longValue();
+ }
+ } else {
+ LOG.info("not able to remove cache entry [{}], size [{}]", key,
+ super.get(key));
+ }
+ return flength;
+ }
+
+ @Override
+ public Long put(final String fileName, final Long value) {
+ if( isInPurgeMode()) {
+ LOG.debug("cache is purge mode: put is no-op");
+ return null;
+ }
+ synchronized (this) {
+ Long oldValue = cache.get(fileName);
+ if (oldValue == null) {
+ long flength = value.longValue();
+ currentSizeInBytes += flength;
+ return super.put(fileName.replace("\\", "/"), value);
+ }
+ toBeDeleted.remove(fileName);
+ return oldValue;
+ }
+ }
+
+ @Override
+ public Long get(Object key) {
+ if( isInPurgeMode()) {
+ LOG.debug("cache is purge mode: get is no-op");
+ return null;
+ }
+ synchronized (this) {
+ return super.get(key);
+ }
+ }
+
+ /**
+ * This method check if cache can admit file of given length.
+ * @param length length of file.
+ * @return true if cache size + length is less than maxSize.
+ */
+ synchronized boolean canAdmitFile(final long length) {
+ return cache.currentSizeInBytes + length < cache.maxSizeInBytes;
+ }
+ }
+
+ /**
+ * This class performs purging of local cache. It implements
+ * {@link Runnable} and should be invoked in a separate thread.
+ */
+ private class PurgeJob implements Runnable {
+ public PurgeJob() {
+ // TODO Auto-generated constructor stub
+ }
+
+ /**
+ * This method purges local cache till its size is less than
+ * cacheResizefactor * maxSize
+ */
+ @Override
+ public void run() {
+ try {
+ synchronized (cache) {
+ // first try to delete toBeDeleted files
+ int initialSize = cache.size();
+ LOG.info(" cache purge job started. initial cache entries = [{}]", initialSize);
+ for (String fileName : new ArrayList<String>(toBeDeleted)) {
+ cache.remove(fileName);
+ }
+ int skipCount = 0;
+ Iterator<Map.Entry<String, Long>> itr = cache.entrySet().iterator();
+ while (itr.hasNext()) {
+ Map.Entry<String, Long> entry = itr.next();
+ if (entry.getKey() != null) {
+ if (cache.currentSizeInBytes > cache.cachePurgeResize) {
+ if (cache.remove(entry.getKey()) != null) {
+ itr = cache.entrySet().iterator();
+ for (int i = 0; i < skipCount && itr.hasNext(); i++) {
+ itr.next();
+ }
+ } else {
+ skipCount++;
+ }
+ } else {
+ break;
+ }
+ }
+ }
+ LOG.info(
+ " cache purge job completed: cleaned [{}] files and currentSizeInBytes = [{}]",
+ (initialSize - cache.size()), cache.currentSizeInBytes);
+ }
+ } catch (Exception e) {
+ LOG.error("error in purge jobs:", e);
+ } finally {
+ setPurgeMode(false);
+ }
+ }
+ }
+
+ /**
+ * This class implements {@link Runnable} interface to build LRU cache
+ * asynchronously.
+ */
+ private class CacheBuildJob implements Runnable {
+
+
+ public void run() {
+ long startTime = System.currentTimeMillis();
+ ArrayList<File> allFiles = new ArrayList<File>();
+ Iterator<File> it = FileUtils.iterateFiles(directory, null, true);
+ while (it.hasNext()) {
+ File f = it.next();
+ allFiles.add(f);
+ }
+ long t1 = System.currentTimeMillis();
+ LOG.debug("Time taken to recursive [{}] took [{}] sec",
+ allFiles.size(), ((t1 - startTime) / 1000));
+
+ String dataStorePath = directory.getAbsolutePath();
+ // convert to java path format
+ dataStorePath = dataStorePath.replace("\\", "/");
+ LOG.info("directoryPath = " + dataStorePath);
+
+ String tmpPath = tmp.getAbsolutePath();
+ tmpPath = tmpPath.replace("\\", "/");
+ LOG.debug("tmp path [{}]", tmpPath);
+ long time = System.currentTimeMillis();
+ int count = 0;
+ for (File f : allFiles) {
+ if (f.exists()) {
+ count++;
+ String name = f.getPath();
+ String filePath = f.getAbsolutePath();
+ // convert to java path format
+ name = name.replace("\\", "/");
+ filePath = filePath.replace("\\", "/");
+ // skipped any temp file
+ if(filePath.startsWith(tmpPath) ) {
+ LOG.info ("tmp file [{}] skipped ", filePath);
+ continue;
+ }
+ if (filePath.startsWith(dataStorePath)) {
+ name = filePath.substring(dataStorePath.length());
+ }
+ if (name.startsWith("/") || name.startsWith("\\")) {
+ name = name.substring(1);
+ }
+ store(name, f);
+ long now = System.currentTimeMillis();
+ if (now > time + 10000) {
+ LOG.info("Processed {" + (count) + "}/{" + allFiles.size() + "}");
+ time = now;
+ }
+ }
+ }
+ LOG.debug(
+ "Processed [{}]/[{}], currentSizeInBytes = [{}], maxSizeInBytes = [{}], cache.filecount = [{}]",
+ new Object[] { count, allFiles.size(),
+ cache.currentSizeInBytes, cache.maxSizeInBytes,
+ cache.size() });
+ long t3 = System.currentTimeMillis();
+ LOG.info("Time to build cache of [{}] files took [{}] sec",
+ allFiles.size(), ((t3 - startTime) / 1000));
+ }
+ }
+}
+
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/data/MultiDataStore.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/MultiDataStore.java
new file mode 100644
index 0000000000..aea8016a48
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/MultiDataStore.java
@@ -0,0 +1,722 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.jackrabbit.core.data;
+
+import java.io.BufferedReader;
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.FileWriter;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.OutputStreamWriter;
+import java.util.ArrayList;
+import java.util.Calendar;
+import java.util.Iterator;
+import java.util.concurrent.locks.ReentrantLock;
+
+import javax.jcr.RepositoryException;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.jackrabbit.core.fs.FileSystem;
+import org.apache.jackrabbit.core.fs.FileSystemException;
+import org.apache.jackrabbit.core.fs.FileSystemResource;
+import org.apache.jackrabbit.core.fs.local.LocalFileSystem;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A MultiDataStore can handle two independent DataStores.
+ * <p>
+ * <b>Attention:</b> You will lost the global single instance mechanism !
+ * </p>
+ * It can be used if you have two storage systems. One for fast access and a
+ * other one like a archive DataStore on a slower storage system. All Files will
+ * be added to the primary DataStore. On read operations first the primary
+ * dataStore will be used and if no Record is found the archive DataStore will
+ * be used. The GarabageCollector will only remove files from the archive
+ * DataStore.
+ * <p>
+ * The internal MoveDataTask will be started automatically and could be
+ * configured with the following properties.
+ * <p>
+ * The Configuration:
+ *
+ * <pre>
+ * <DataStore class="org.apache.jackrabbit.core.data.MultiDataStore">
+ * <param name="{@link #setMaxAge(int) maxAge}" value="60"/>
+ * <param name="{@link #setMoveDataTaskSleep(int) moveDataTaskSleep}" value="604800"/>
+ * <param name="{@link #setMoveDataTaskFirstRunHourOfDay(int) moveDataTaskFirstRunHourOfDay}" value="1"/>
+ * <param name="{@link #setSleepBetweenRecords(long) sleepBetweenRecords}" value="100"/>
+ * <param name="{@link #setDelayedDelete(boolean) delayedDelete}" value="false"/>
+ * <param name="{@link #setDelayedDeleteSleep(long) delayedDeleteSleep}" value="86400"/>
+ * <param name="primary" value="org.apache.jackrabbit.core.data.db.DbDataStore">
+ * <param .../>
+ * </param>
+ * <param name="archive" value="org.apache.jackrabbit.core.data.FileDataStore">
+ * <param .../>
+ * </param>
+ * </DataStore>
+ * </pre>
+ *
+ * <ul>
+ * <li><code>maxAge</code>: defines how many days the content will reside in the
+ * primary data store. DataRecords that have been added before this time span
+ * will be moved to the archive data store. (default = <code>60</code>)</li>
+ * <li><code>moveDataTaskSleep</code>: specifies the sleep time of the
+ * moveDataTaskThread in seconds. (default = 60 * 60 * 24 * 7, which equals 7
+ * days)</li>
+ * <li><code>moveDataTaskNextRunHourOfDay</code>: specifies the hour at which
+ * the moveDataTaskThread initiates its first run (default = <code>1</code>
+ * which means 01:00 at night)</li>
+ * <li><code>sleepBetweenRecords</code>: specifies the delay in milliseconds
+ * between scanning data records (default = <code>100</code>)</li>
+ * <li><code>delayedDelete</code>: its possible to delay the delete operation on
+ * the primary data store. The DataIdentifiers will be written to a temporary
+ * file. The file will be processed after a defined sleep (see
+ * <code>delayedDeleteSleep</code>) It's useful if you like to create a snapshot
+ * of the primary data store backend in the meantime before the data will be
+ * deleted. (default = <code>false</code>)</li>
+ * <li><code>delayedDeleteSleep</code>: specifies the sleep time of the
+ * delayedDeleteTaskThread in seconds. (default = 60 * 60 * 24, which equals 1
+ * day). This means the delayed delete from the primary data store will be
+ * processed after one day.</li>
+ * </ul>
+ */
+public class MultiDataStore implements DataStore {
+
+ /**
+ * Logger instance
+ */
+ private static Logger log = LoggerFactory.getLogger(MultiDataStore.class);
+
+ private DataStore primaryDataStore;
+ private DataStore archiveDataStore;
+
+ /**
+ * Max Age in days.
+ */
+ private int maxAge = 60;
+
+ /**
+ * ReentrantLock that is used while the MoveDataTask is running.
+ */
+ private ReentrantLock moveDataTaskLock = new ReentrantLock();
+ private boolean moveDataTaskRunning = false;
+ private Thread moveDataTaskThread;
+
+ /**
+ * The sleep time in seconds of the MoveDataTask, 7 day default.
+ */
+ private int moveDataTaskSleep = 60 * 60 * 24 * 7;
+
+ /**
+ * Indicates when the next run of the move task is scheduled. The first run
+ * is scheduled by default at 01:00 hours.
+ */
+ private Calendar moveDataTaskNextRun = Calendar.getInstance();
+
+ /**
+ * Its possible to delay the delete operation on the primary data store
+ * while move task is running. The delete will be executed after defined
+ * delayDeleteSleep.
+ */
+ private boolean delayedDelete = false;
+
+ /**
+ * The sleep time in seconds to delay remove operation on the primary data
+ * store, 1 day default.
+ */
+ private long delayedDeleteSleep = 60 * 60 * 24;
+
+ /**
+ * File that holds the data identifiers if delayDelete is enabled.
+ */
+ private FileSystemResource identifiersToDeleteFile = null;
+
+ private Thread deleteDelayedIdentifiersTaskThread;
+
+ /**
+ * Name of the file which holds the identifiers if deleayed delete is
+ * enabled
+ */
+ private final String IDENTIFIERS_TO_DELETE_FILE_KEY = "identifiersToDelete";
+
+ /**
+ * The delay time in milliseconds between scanning data records, 100
+ * default.
+ */
+ private long sleepBetweenRecords = 100;
+
+ {
+ if (moveDataTaskNextRun.get(Calendar.HOUR_OF_DAY) >= 1) {
+ moveDataTaskNextRun.add(Calendar.DAY_OF_MONTH, 1);
+ }
+ moveDataTaskNextRun.set(Calendar.HOUR_OF_DAY, 1);
+ moveDataTaskNextRun.set(Calendar.MINUTE, 0);
+ moveDataTaskNextRun.set(Calendar.SECOND, 0);
+ moveDataTaskNextRun.set(Calendar.MILLISECOND, 0);
+ }
+
+ /**
+ * Setter for the primary dataStore
+ *
+ * @param dataStore
+ */
+ public void setPrimaryDataStore(DataStore dataStore) {
+ this.primaryDataStore = dataStore;
+ }
+
+ /**
+ * Setter for the archive dataStore
+ *
+ * @param dataStore
+ */
+ public void setArchiveDataStore(DataStore dataStore) {
+ this.archiveDataStore = dataStore;
+ }
+
+ /**
+ * Check if a record for the given identifier exists in the primary data
+ * store. If not found there it will be returned from the archive data
+ * store. If no record exists, this method returns null.
+ *
+ * @param identifier
+ * data identifier
+ * @return the record if found, and null if not
+ */
+ public DataRecord getRecordIfStored(DataIdentifier identifier) throws DataStoreException {
+ if (moveDataTaskRunning) {
+ moveDataTaskLock.lock();
+ }
+ try {
+ DataRecord dataRecord = primaryDataStore.getRecordIfStored(identifier);
+ if (dataRecord == null) {
+ dataRecord = archiveDataStore.getRecordIfStored(identifier);
+ }
+ return dataRecord;
+ } finally {
+ if (moveDataTaskRunning) {
+ moveDataTaskLock.unlock();
+ }
+ }
+ }
+
+ /**
+ * Returns the identified data record from the primary data store. If not
+ * found there it will be returned from the archive data store. The given
+ * identifier should be the identifier of a previously saved data record.
+ * Since records are never removed, there should never be cases where the
+ * identified record is not found. Abnormal cases like that are treated as
+ * errors and handled by throwing an exception.
+ *
+ * @param identifier
+ * data identifier
+ * @return identified data record
+ * @throws DataStoreException
+ * if the data store could not be accessed, or if the given
+ * identifier is invalid
+ */
+ public DataRecord getRecord(DataIdentifier identifier) throws DataStoreException {
+ if (moveDataTaskRunning) {
+ moveDataTaskLock.lock();
+ }
+ try {
+ return primaryDataStore.getRecord(identifier);
+ } catch (DataStoreException e) {
+ return archiveDataStore.getRecord(identifier);
+ } finally {
+ if (moveDataTaskRunning) {
+ moveDataTaskLock.unlock();
+ }
+ }
+ }
+
+ /**
+ * Creates a new data record in the primary data store. The given binary
+ * stream is consumed and a binary record containing the consumed stream is
+ * created and returned. If the same stream already exists in another
+ * record, then that record is returned instead of creating a new one.
+ * <p>
+ * The given stream is consumed and <strong>not closed</strong> by this
+ * method. It is the responsibility of the caller to close the stream. A
+ * typical call pattern would be:
+ *
+ * <pre>
+ * InputStream stream = ...;
+ * try {
+ * record = store.addRecord(stream);
+ * } finally {
+ * stream.close();
+ * }
+ * </pre>
+ *
+ * @param stream
+ * binary stream
+ * @return data record that contains the given stream
+ * @throws DataStoreException
+ * if the data store could not be accessed
+ */
+ public DataRecord addRecord(InputStream stream) throws DataStoreException {
+ return primaryDataStore.addRecord(stream);
+ }
+
+ /**
+ * From now on, update the modified date of an object even when accessing it
+ * in the archive data store. Usually, the modified date is only updated
+ * when creating a new object, or when a new link is added to an existing
+ * object. When this setting is enabled, even getLength() will update the
+ * modified date.
+ *
+ * @param before
+ * - update the modified date to the current time if it is older
+ * than this value
+ */
+ public void updateModifiedDateOnAccess(long before) {
+ archiveDataStore.updateModifiedDateOnAccess(before);
+ }
+
+ /**
+ * Delete objects that have a modified date older than the specified date
+ * from the archive data store.
+ *
+ * @param min
+ * the minimum time
+ * @return the number of data records deleted
+ * @throws DataStoreException
+ */
+ public int deleteAllOlderThan(long min) throws DataStoreException {
+ return archiveDataStore.deleteAllOlderThan(min);
+ }
+
+ /**
+ * Get all identifiers from the archive data store.
+ *
+ * @return an iterator over all DataIdentifier objects
+ * @throws DataStoreException
+ * if the list could not be read
+ */
+ public Iterator<DataIdentifier> getAllIdentifiers() throws DataStoreException {
+ return archiveDataStore.getAllIdentifiers();
+ }
+
+ public DataRecord getRecordFromReference(String reference)
+ throws DataStoreException {
+ DataRecord record = primaryDataStore.getRecordFromReference(reference);
+ if (record == null) {
+ record = archiveDataStore.getRecordFromReference(reference);
+ }
+ return record;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public void init(String homeDir) throws RepositoryException {
+ if (delayedDelete) {
+ // First initialize the identifiersToDeleteFile
+ LocalFileSystem fileSystem = new LocalFileSystem();
+ fileSystem.setRoot(new File(homeDir));
+ identifiersToDeleteFile = new FileSystemResource(fileSystem, FileSystem.SEPARATOR
+ + IDENTIFIERS_TO_DELETE_FILE_KEY);
+ }
+ moveDataTaskThread = new Thread(new MoveDataTask(),
+ "Jackrabbit-MulitDataStore-MoveDataTaskThread");
+ moveDataTaskThread.setDaemon(true);
+ moveDataTaskThread.start();
+ log.info("MultiDataStore-MoveDataTask thread started; first run scheduled at "
+ + moveDataTaskNextRun.getTime());
+ if (delayedDelete) {
+ try {
+ // Run on startup the DeleteDelayedIdentifiersTask only if the
+ // file exists and modify date is older than the
+ // delayedDeleteSleep timeout ...
+ if (identifiersToDeleteFile != null
+ && identifiersToDeleteFile.exists()
+ && (identifiersToDeleteFile.lastModified() + (delayedDeleteSleep * 1000)) < System
+ .currentTimeMillis()) {
+ deleteDelayedIdentifiersTaskThread = new Thread(
+ //Start immediately ...
+ new DeleteDelayedIdentifiersTask(0L),
+ "Jackrabbit-MultiDataStore-DeleteDelayedIdentifiersTaskThread");
+ deleteDelayedIdentifiersTaskThread.setDaemon(true);
+ deleteDelayedIdentifiersTaskThread.start();
+ log.info("Old entries in the " + IDENTIFIERS_TO_DELETE_FILE_KEY
+ + " File found. DeleteDelayedIdentifiersTask-Thread started now.");
+ }
+ } catch (FileSystemException e) {
+ throw new RepositoryException("I/O error while reading from '"
+ + identifiersToDeleteFile.getPath() + "'", e);
+ }
+ }
+ }
+
+ /**
+ * Get the minimum size of an object that should be stored in the primary
+ * data store.
+ *
+ * @return the minimum size in bytes
+ */
+ public int getMinRecordLength() {
+ return primaryDataStore.getMinRecordLength();
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public void close() throws DataStoreException {
+ DataStoreException lastException = null;
+ // 1. close the primary data store
+ try {
+ primaryDataStore.close();
+ } catch (DataStoreException e) {
+ lastException = e;
+ }
+ // 2. close the archive data store
+ try {
+ archiveDataStore.close();
+ } catch (DataStoreException e) {
+ if (lastException != null) {
+ lastException = new DataStoreException(lastException);
+ }
+ }
+ // 3. if moveDataTaskThread is running interrupt it
+ try {
+ if (moveDataTaskRunning) {
+ moveDataTaskThread.interrupt();
+ }
+ } catch (Exception e) {
+ if (lastException != null) {
+ lastException = new DataStoreException(lastException);
+ }
+ }
+ // 4. if deleteDelayedIdentifiersTaskThread is running interrupt it
+ try {
+ if (deleteDelayedIdentifiersTaskThread != null
+ && deleteDelayedIdentifiersTaskThread.isAlive()) {
+ deleteDelayedIdentifiersTaskThread.interrupt();
+ }
+ } catch (Exception e) {
+ if (lastException != null) {
+ lastException = new DataStoreException(lastException);
+ }
+ }
+ if (lastException != null) {
+ throw lastException;
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public void clearInUse() {
+ archiveDataStore.clearInUse();
+ }
+
+ public int getMaxAge() {
+ return maxAge;
+ }
+
+ public void setMaxAge(int maxAge) {
+ this.maxAge = maxAge;
+ }
+
+ public int getMoveDataTaskSleep() {
+ return moveDataTaskSleep;
+ }
+
+ public int getMoveDataTaskFirstRunHourOfDay() {
+ return moveDataTaskNextRun.get(Calendar.HOUR_OF_DAY);
+ }
+
+ public void setMoveDataTaskSleep(int sleep) {
+ this.moveDataTaskSleep = sleep;
+ }
+
+ public void setMoveDataTaskFirstRunHourOfDay(int hourOfDay) {
+ moveDataTaskNextRun = Calendar.getInstance();
+ if (moveDataTaskNextRun.get(Calendar.HOUR_OF_DAY) >= hourOfDay) {
+ moveDataTaskNextRun.add(Calendar.DAY_OF_MONTH, 1);
+ }
+ moveDataTaskNextRun.set(Calendar.HOUR_OF_DAY, hourOfDay);
+ moveDataTaskNextRun.set(Calendar.MINUTE, 0);
+ moveDataTaskNextRun.set(Calendar.SECOND, 0);
+ moveDataTaskNextRun.set(Calendar.MILLISECOND, 0);
+ }
+
+ public void setSleepBetweenRecords(long millis) {
+ this.sleepBetweenRecords = millis;
+ }
+
+ public long getSleepBetweenRecords() {
+ return sleepBetweenRecords;
+ }
+
+ public boolean isDelayedDelete() {
+ return delayedDelete;
+ }
+
+ public void setDelayedDelete(boolean delayedDelete) {
+ this.delayedDelete = delayedDelete;
+ }
+
+ public long getDelayedDeleteSleep() {
+ return delayedDeleteSleep;
+ }
+
+ public void setDelayedDeleteSleep(long delayedDeleteSleep) {
+ this.delayedDeleteSleep = delayedDeleteSleep;
+ }
+
+ /**
+ * Writes the given DataIdentifier to the delayedDeletedFile.
+ *
+ * @param identifier
+ * @return boolean true if it was successful otherwise false
+ */
+ private boolean writeDelayedDataIdentifier(DataIdentifier identifier) {
+ BufferedWriter writer = null;
+ try {
+ File identifierFile = new File(
+ ((LocalFileSystem) identifiersToDeleteFile.getFileSystem()).getPath(),
+ identifiersToDeleteFile.getPath());
+ writer = new BufferedWriter(new FileWriter(identifierFile, true));
+ writer.write(identifier.toString());
+ return true;
+ } catch (Exception e) {
+ log.warn("I/O error while saving DataIdentifier (stacktrace on DEBUG log level) to '"
+ + identifiersToDeleteFile.getPath() + "': " + e.getMessage());
+ log.debug("Root cause: ", e);
+ return false;
+ } finally {
+ IOUtils.closeQuietly(writer);
+ }
+ }
+
+ /**
+ * Purges the delayedDeletedFile.
+ *
+ * @return boolean true if it was successful otherwise false
+ */
+ private boolean purgeDelayedDeleteFile() {
+ BufferedWriter writer = null;
+ try {
+ writer = new BufferedWriter(new OutputStreamWriter(
+ identifiersToDeleteFile.getOutputStream()));
+ writer.write("");
+ return true;
+ } catch (Exception e) {
+ log.warn("I/O error while purging (stacktrace on DEBUG log level) the "
+ + IDENTIFIERS_TO_DELETE_FILE_KEY + " file '"
+ + identifiersToDeleteFile.getPath() + "': " + e.getMessage());
+ log.debug("Root cause: ", e);
+ return false;
+ } finally {
+ IOUtils.closeQuietly(writer);
+ }
+ }
+
+ /**
+ * Class for maintaining the MultiDataStore. It will be used to move the
+ * content of the primary data store to the archive data store.
+ */
+ public class MoveDataTask implements Runnable {
+
+ /**
+ * {@inheritDoc}
+ */
+ public void run() {
+ while (!Thread.currentThread().isInterrupted()) {
+ try {
+ log.info("Next move-data task run scheduled at "
+ + moveDataTaskNextRun.getTime());
+ long sleepTime = moveDataTaskNextRun.getTimeInMillis()
+ - System.currentTimeMillis();
+ if (sleepTime > 0) {
+ Thread.sleep(sleepTime);
+ }
+ moveDataTaskRunning = true;
+ moveOutdatedData();
+ moveDataTaskRunning = false;
+ moveDataTaskNextRun.add(Calendar.SECOND, moveDataTaskSleep);
+ if (delayedDelete) {
+ if (deleteDelayedIdentifiersTaskThread != null
+ && deleteDelayedIdentifiersTaskThread.isAlive()) {
+ log.warn("The DeleteDelayedIdentifiersTask-Thread is already running.");
+ } else {
+ deleteDelayedIdentifiersTaskThread = new Thread(
+ new DeleteDelayedIdentifiersTask(delayedDeleteSleep),
+ "Jackrabbit-MultiDataStore-DeleteDelayedIdentifiersTaskThread");
+ deleteDelayedIdentifiersTaskThread.setDaemon(true);
+ deleteDelayedIdentifiersTaskThread.start();
+ }
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ log.warn("Interrupted: stopping move-data task.");
+ }
+
+ /**
+ * Moves outdated data from primary to archive data store
+ */
+ protected void moveOutdatedData() {
+ try {
+ long now = System.currentTimeMillis();
+ long maxAgeMilli = 1000L * 60 * 60 * 24 * maxAge;
+ log.debug("Collecting all Identifiers from PrimaryDataStore...");
+ Iterator<DataIdentifier> allIdentifiers = primaryDataStore.getAllIdentifiers();
+ int moved = 0;
+ while (allIdentifiers.hasNext()) {
+ DataIdentifier identifier = allIdentifiers.next();
+ DataRecord dataRecord = primaryDataStore.getRecord(identifier);
+ if ((dataRecord.getLastModified() + maxAgeMilli) < now) {
+ try {
+ moveDataTaskLock.lock();
+ if (delayedDelete) {
+ // first write it to the file and then add it to
+ // the archive data store ...
+ if (writeDelayedDataIdentifier(identifier)) {
+ archiveDataStore.addRecord(dataRecord.getStream());
+ moved++;
+ }
+ } else {
+ // first add it and then delete it .. not really
+ // atomic ...
+ archiveDataStore.addRecord(dataRecord.getStream());
+ ((MultiDataStoreAware) primaryDataStore).deleteRecord(identifier);
+ moved++;
+ }
+ if (moved % 100 == 0) {
+ log.debug("Moving DataRecord's... ({})", moved);
+ }
+ } catch (DataStoreException e) {
+ log.error("Failed to move DataRecord. DataIdentifier: " + identifier, e);
+ } finally {
+ moveDataTaskLock.unlock();
+ }
+ }
+ // Give other threads time to use the MultiDataStore while
+ // MoveDataTask is running..
+ Thread.sleep(sleepBetweenRecords);
+ }
+ if (delayedDelete) {
+ log.info("Moved "
+ + moved
+ + " DataRecords to the archive data store. The DataRecords in the primary data store will be removed in "
+ + delayedDeleteSleep + " seconds.");
+ } else {
+ log.info("Moved " + moved + " DataRecords to the archive data store.");
+ }
+ } catch (Exception e) {
+ log.warn("Failed to run move-data task.", e);
+ }
+ }
+ }
+
+ /**
+ * Class to clean up the delayed DataRecords from the primary data store.
+ */
+ public class DeleteDelayedIdentifiersTask implements Runnable {
+
+ boolean run = true;
+ private long sleepTime = 0L;
+
+ /**
+ * Constructor
+ * @param sleep how long this DeleteDelayedIdentifiersTask should sleep in seconds.
+ */
+ public DeleteDelayedIdentifiersTask(long sleep) {
+ this.sleepTime = (sleep * 1000L);
+ }
+
+ @Override
+ public void run() {
+ if (moveDataTaskRunning) {
+ log.warn("It's not supported to run the DeleteDelayedIdentifiersTask while the MoveDataTask is running.");
+ return;
+ }
+ while (run && !Thread.currentThread().isInterrupted()) {
+ if (sleepTime > 0) {
+ try {
+ Thread.sleep(sleepTime);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ log.info("Start to delete DataRecords from the primary data store.");
+ BufferedReader reader = null;
+ ArrayList<DataIdentifier> problemIdentifiers = new ArrayList<DataIdentifier>();
+ try {
+ int deleted = 0;
+ reader = new BufferedReader(new InputStreamReader(
+ identifiersToDeleteFile.getInputStream()));
+ while (true) {
+ String s = reader.readLine();
+ if (s == null || s.equals("")) {
+ break;
+ }
+ DataIdentifier identifier = new DataIdentifier(s);
+ try {
+ moveDataTaskLock.lock();
+ ((MultiDataStoreAware) primaryDataStore).deleteRecord(identifier);
+ deleted++;
+ } catch (DataStoreException e) {
+ log.error("Failed to delete DataRecord. DataIdentifier: " + identifier,
+ e);
+ problemIdentifiers.add(identifier);
+ } finally {
+ moveDataTaskLock.unlock();
+ }
+ // Give other threads time to use the MultiDataStore
+ // while
+ // DeleteDelayedIdentifiersTask is running..
+ Thread.sleep(sleepBetweenRecords);
+ }
+ log.info("Deleted " + deleted + " DataRecords from the primary data store.");
+ if (problemIdentifiers.isEmpty()) {
+ try {
+ identifiersToDeleteFile.delete();
+ } catch (FileSystemException e) {
+ log.warn("Unable to delete the " + IDENTIFIERS_TO_DELETE_FILE_KEY
+ + " File.");
+ if (!purgeDelayedDeleteFile()) {
+ log.error("Unable to purge the " + IDENTIFIERS_TO_DELETE_FILE_KEY
+ + " File.");
+ }
+ }
+ } else {
+ if (purgeDelayedDeleteFile()) {
+ for (int x = 0; x < problemIdentifiers.size(); x++) {
+ writeDelayedDataIdentifier(problemIdentifiers.get(x));
+ }
+ }
+ }
+ } catch (InterruptedException e) {
+ log.warn("Interrupted: stopping delayed-delete task.");
+ Thread.currentThread().interrupt();
+ } catch (Exception e) {
+ log.warn("Failed to run delayed-delete task.", e);
+ } finally {
+ IOUtils.closeQuietly(reader);
+ run = false;
+ }
+ }
+ }
+ }
+
+}
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/data/MultiDataStoreAware.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/MultiDataStoreAware.java
new file mode 100644
index 0000000000..b4bb5c20ec
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/MultiDataStoreAware.java
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.jackrabbit.core.data;
+
+import org.apache.jackrabbit.core.data.MultiDataStore.MoveDataTask;
+
+/**
+ * To use a DataStore within a MultiDataStore it must implement this
+ * MultiDataStoreAware Interface. It extends a DataStore to delete a single
+ * DataRecord.
+ */
+public interface MultiDataStoreAware {
+
+ /**
+ * Deletes a single DataRecord based on the given identifier. Delete will
+ * only be used by the {@link MoveDataTask}.
+ *
+ * @param identifier
+ * data identifier
+ * @throws DataStoreException
+ * if the data store could not be accessed, or if the given
+ * identifier is invalid
+ */
+ void deleteRecord(DataIdentifier identifier) throws DataStoreException;
+
+}
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/data/ScanEventListener.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/ScanEventListener.java
new file mode 100644
index 0000000000..f442f6f7d3
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/ScanEventListener.java
@@ -0,0 +1,26 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.jackrabbit.core.data;
+
+import org.apache.jackrabbit.api.management.MarkEventListener;
+
+/**
+ * The listener interface for receiving garbage collection scan events.
+ */
+public interface ScanEventListener extends MarkEventListener {
+
+}
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/data/db/DbDataRecord.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/db/DbDataRecord.java
new file mode 100644
index 0000000000..417b3ced77
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/db/DbDataRecord.java
@@ -0,0 +1,71 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.jackrabbit.core.data.db;
+
+import org.apache.jackrabbit.core.data.AbstractDataRecord;
+import org.apache.jackrabbit.core.data.DataIdentifier;
+import org.apache.jackrabbit.core.data.DataStoreException;
+
+import java.io.BufferedInputStream;
+import java.io.InputStream;
+
+/**
+ * Data record that is stored in a database
+ */
+public class DbDataRecord extends AbstractDataRecord {
+
+ protected final DbDataStore store;
+ protected final long length;
+ protected long lastModified;
+
+ /**
+ * Creates a data record based on the given identifier and length.
+ *
+ * @param identifier data identifier
+ * @param length the length
+ * @param lastModified
+ */
+ public DbDataRecord(DbDataStore store, DataIdentifier identifier, long length, long lastModified) {
+ super(store, identifier);
+ this.store = store;
+ this.length = length;
+ this.lastModified = lastModified;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public long getLength() throws DataStoreException {
+ lastModified = store.touch(getIdentifier(), lastModified);
+ return length;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public InputStream getStream() throws DataStoreException {
+ lastModified = store.touch(getIdentifier(), lastModified);
+ return new BufferedInputStream(new DbInputStream(store, getIdentifier()));
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public long getLastModified() {
+ return lastModified;
+ }
+}
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/data/db/DbDataStore.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/db/DbDataStore.java
new file mode 100644
index 0000000000..fff2cb8d8c
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/db/DbDataStore.java
@@ -0,0 +1,1016 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.jackrabbit.core.data.db;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.io.input.CountingInputStream;
+import org.apache.jackrabbit.core.data.AbstractDataStore;
+import org.apache.jackrabbit.core.data.DataIdentifier;
+import org.apache.jackrabbit.core.data.DataRecord;
+import org.apache.jackrabbit.core.data.DataStoreException;
+import org.apache.jackrabbit.core.data.MultiDataStoreAware;
+import org.apache.jackrabbit.core.util.db.CheckSchemaOperation;
+import org.apache.jackrabbit.core.util.db.ConnectionFactory;
+import org.apache.jackrabbit.core.util.db.ConnectionHelper;
+import org.apache.jackrabbit.core.util.db.DatabaseAware;
+import org.apache.jackrabbit.core.util.db.DbUtility;
+import org.apache.jackrabbit.core.util.db.StreamWrapper;
+import org.apache.jackrabbit.util.Text;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.BufferedInputStream;
+import java.io.ByteArrayInputStream;
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.lang.ref.WeakReference;
+import java.security.DigestInputStream;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.UUID;
+import java.util.WeakHashMap;
+
+import javax.jcr.RepositoryException;
+import javax.sql.DataSource;
+
+/**
+ * A data store implementation that stores the records in a database using JDBC.
+ *
+ * Configuration:
+ * <pre>
+ * <DataStore class="org.apache.jackrabbit.core.data.db.DbDataStore">
+ * <param name="{@link #setUrl(String) url}" value="jdbc:postgresql:test"/>
+ * <param name="{@link #setUser(String) user}" value="sa"/>
+ * <param name="{@link #setPassword(String) password}" value="sa"/>
+ * <param name="{@link #setDatabaseType(String) databaseType}" value="postgresql"/>
+ * <param name="{@link #setDriver(String) driver}" value="org.postgresql.Driver"/>
+ * <param name="{@link #setMinRecordLength(int) minRecordLength}" value="1024"/>
+ * <param name="{@link #setMaxConnections(int) maxConnections}" value="2"/>
+ * <param name="{@link #setCopyWhenReading(boolean) copyWhenReading}" value="true"/>
+ * <param name="{@link #setTablePrefix(String) tablePrefix}" value=""/>
+ * <param name="{@link #setSchemaObjectPrefix(String) schemaObjectPrefix}" value=""/>
+ * <param name="{@link #setSchemaCheckEnabled(boolean) schemaCheckEnabled}" value="true"/>
+ * </DataStore>
+ * </pre>
+ * <p>
+ * Only URL, user name and password usually need to be set.
+ * The remaining settings are generated using the database URL sub-protocol from the
+ * database type resource file.
+ * <p>
+ * JNDI can be used to get the connection. In this case, use the javax.naming.InitialContext as the driver,
+ * and the JNDI name as the URL. If the user and password are configured in the JNDI resource,
+ * they should not be configured here. Example JNDI settings:
+ * <pre>
+ * <param name="driver" value="javax.naming.InitialContext" />
+ * <param name="url" value="java:comp/env/jdbc/Test" />
+ * </pre>
+ * <p>
+ * For Microsoft SQL Server 2005, there is a problem reading large BLOBs. You will need to use
+ * the JDBC driver version 1.2 or newer, and append ;responseBuffering=adaptive to the database URL.
+ * Don't append ;selectMethod=cursor, otherwise it can still run out of memory.
+ * Example database URL: jdbc:sqlserver://localhost:4220;DatabaseName=test;responseBuffering=adaptive
+ * <p>
+ * By default, the data is copied to a temp file when reading, to avoid problems when reading multiple
+ * blobs at the same time.
+ * <p>
+ * The tablePrefix can be used to specify a schema and / or catalog name:
+ * <param name="tablePrefix" value="ds.">
+ */
+public class DbDataStore extends AbstractDataStore
+ implements DatabaseAware, MultiDataStoreAware {
+
+ /**
+ * The default value for the minimum object size.
+ */
+ public static final int DEFAULT_MIN_RECORD_LENGTH = 100;
+
+ /**
+ * Write to a temporary file to get the length (slow, but always works).
+ * This is the default setting.
+ */
+ public static final String STORE_TEMP_FILE = "tempFile";
+
+ /**
+ * Call PreparedStatement.setBinaryStream(..., -1)
+ */
+ public static final String STORE_SIZE_MINUS_ONE = "-1";
+
+ /**
+ * Call PreparedStatement.setBinaryStream(..., Integer.MAX_VALUE)
+ */
+ public static final String STORE_SIZE_MAX = "max";
+
+ /**
+ * The digest algorithm used to uniquely identify records.
+ */
+ protected static final String DIGEST = System.getProperty("ds.digest.algorithm", "SHA-256");
+
+ /**
+ * The prefix used for temporary objects.
+ */
+ protected static final String TEMP_PREFIX = "TEMP_";
+
+ /**
+ * Logger instance
+ */
+ private static Logger log = LoggerFactory.getLogger(DbDataStore.class);
+
+ /**
+ * The minimum modified date. If a file is accessed (read or write) with a modified date
+ * older than this value, the modified date is updated to the current time.
+ */
+ protected long minModifiedDate;
+
+ /**
+ * The database URL used.
+ */
+ protected String url;
+
+ /**
+ * The database driver.
+ */
+ protected String driver;
+
+ /**
+ * The user name.
+ */
+ protected String user;
+
+ /**
+ * The password
+ */
+ protected String password;
+
+ /**
+ * The database type used.
+ */
+ protected String databaseType;
+
+ /**
+ * The minimum size of an object that should be stored in this data store.
+ */
+ protected int minRecordLength = DEFAULT_MIN_RECORD_LENGTH;
+
+ /**
+ * The prefix for the datastore table, empty by default.
+ */
+ protected String tablePrefix = "";
+
+ /**
+ * The prefix of the table names. By default it is empty.
+ */
+ protected String schemaObjectPrefix = "";
+
+ /**
+ * Whether the schema check must be done during initialization.
+ */
+ private boolean schemaCheckEnabled = true;
+
+ /**
+ * The logical name of the DataSource to use.
+ */
+ protected String dataSourceName;
+
+ /**
+ * This is the property 'table'
+ * in the [databaseType].properties file, initialized with the default value.
+ */
+ protected String tableSQL = "DATASTORE";
+
+ /**
+ * This is the property 'createTable'
+ * in the [databaseType].properties file, initialized with the default value.
+ */
+ protected String createTableSQL =
+ "CREATE TABLE ${tablePrefix}${table}(ID VARCHAR(255) PRIMARY KEY, LENGTH BIGINT, LAST_MODIFIED BIGINT, DATA BLOB)";
+
+ /**
+ * This is the property 'insertTemp'
+ * in the [databaseType].properties file, initialized with the default value.
+ */
+ protected String insertTempSQL =
+ "INSERT INTO ${tablePrefix}${table} VALUES(?, 0, ?, NULL)";
+
+ /**
+ * This is the property 'updateData'
+ * in the [databaseType].properties file, initialized with the default value.
+ */
+ protected String updateDataSQL =
+ "UPDATE ${tablePrefix}${table} SET DATA=? WHERE ID=?";
+
+ /**
+ * This is the property 'updateLastModified'
+ * in the [databaseType].properties file, initialized with the default value.
+ */
+ protected String updateLastModifiedSQL =
+ "UPDATE ${tablePrefix}${table} SET LAST_MODIFIED=? WHERE ID=? AND LAST_MODIFIED<?";
+
+ /**
+ * This is the property 'update'
+ * in the [databaseType].properties file, initialized with the default value.
+ */
+ protected String updateSQL =
+ "UPDATE ${tablePrefix}${table} SET ID=?, LENGTH=?, LAST_MODIFIED=? " +
+ "WHERE ID=? AND LAST_MODIFIED=?";
+
+ /**
+ * This is the property 'delete'
+ * in the [databaseType].properties file, initialized with the default value.
+ */
+ protected String deleteSQL =
+ "DELETE FROM ${tablePrefix}${table} WHERE ID=?";
+
+ /**
+ * This is the property 'deleteOlder'
+ * in the [databaseType].properties file, initialized with the default value.
+ */
+ protected String deleteOlderSQL =
+ "DELETE FROM ${tablePrefix}${table} WHERE LAST_MODIFIED<?";
+
+ /**
+ * This is the property 'selectMeta'
+ * in the [databaseType].properties file, initialized with the default value.
+ */
+ protected String selectMetaSQL =
+ "SELECT LENGTH, LAST_MODIFIED FROM ${tablePrefix}${table} WHERE ID=?";
+
+ /**
+ * This is the property 'selectAll'
+ * in the [databaseType].properties file, initialized with the default value.
+ */
+ protected String selectAllSQL =
+ "SELECT ID FROM ${tablePrefix}${table}";
+
+ /**
+ * This is the property 'selectData'
+ * in the [databaseType].properties file, initialized with the default value.
+ */
+ protected String selectDataSQL =
+ "SELECT ID, DATA FROM ${tablePrefix}${table} WHERE ID=?";
+
+ /**
+ * The stream storing mechanism used.
+ */
+ protected String storeStream = STORE_TEMP_FILE;
+
+ /**
+ * Copy the stream to a temp file before returning it.
+ * Enabled by default to support concurrent reads.
+ */
+ protected boolean copyWhenReading = true;
+
+ /**
+ * All data identifiers that are currently in use are in this set until they are garbage collected.
+ */
+ protected Map<DataIdentifier, WeakReference<DataIdentifier>> inUse =
+ Collections.synchronizedMap(new WeakHashMap<DataIdentifier, WeakReference<DataIdentifier>>());
+
+ /**
+ * The temporary identifiers that are currently in use.
+ */
+ protected List<String> temporaryInUse = Collections.synchronizedList(new ArrayList<String>());
+
+ /**
+ * The {@link ConnectionHelper} set in the {@link #init(String)} method.
+ * */
+ protected ConnectionHelper conHelper;
+
+ /**
+ * The repositories {@link ConnectionFactory}.
+ */
+ private ConnectionFactory connectionFactory;
+
+ public void setConnectionFactory(ConnectionFactory connnectionFactory) {
+ this.connectionFactory = connnectionFactory;
+ }
+
+ public DataRecord addRecord(InputStream stream) throws DataStoreException {
+ InputStream fileInput = null;
+ String tempId = null;
+ ResultSet rs = null;
+ try {
+ long tempModified;
+ while (true) {
+ try {
+ tempModified = System.currentTimeMillis();
+ String id = UUID.randomUUID().toString();
+ tempId = TEMP_PREFIX + id;
+ temporaryInUse.add(tempId);
+ // SELECT LENGTH, LAST_MODIFIED FROM DATASTORE WHERE ID=?
+ rs = conHelper.query(selectMetaSQL, tempId);
+ boolean hasNext = rs.next();
+ DbUtility.close(rs);
+ rs = null;
+ if (hasNext) {
+ // re-try in the very, very unlikely event that the row already exists
+ continue;
+ }
+ // INSERT INTO DATASTORE VALUES(?, 0, ?, NULL)
+ conHelper.exec(insertTempSQL, tempId, tempModified);
+ break;
+ } catch (Exception e) {
+ throw convert("Can not insert new record", e);
+ } finally {
+ DbUtility.close(rs);
+ // prevent that rs.close() is called again
+ rs = null;
+ }
+ }
+ MessageDigest digest = getDigest();
+ DigestInputStream dIn = new DigestInputStream(stream, digest);
+ CountingInputStream in = new CountingInputStream(dIn);
+ StreamWrapper wrapper;
+ if (STORE_SIZE_MINUS_ONE.equals(storeStream)) {
+ wrapper = new StreamWrapper(in, -1);
+ } else if (STORE_SIZE_MAX.equals(storeStream)) {
+ wrapper = new StreamWrapper(in, Integer.MAX_VALUE);
+ } else if (STORE_TEMP_FILE.equals(storeStream)) {
+ File temp = moveToTempFile(in);
+ long length = temp.length();
+ wrapper = new StreamWrapper(new ResettableTempFileInputStream(temp), length);
+ } else {
+ throw new DataStoreException("Unsupported stream store algorithm: " + storeStream);
+ }
+ // UPDATE DATASTORE SET DATA=? WHERE ID=?
+ conHelper.exec(updateDataSQL, wrapper, tempId);
+ long length = in.getByteCount();
+ DataIdentifier identifier =
+ new DataIdentifier(encodeHexString(digest.digest()));
+ usesIdentifier(identifier);
+ String id = identifier.toString();
+ long newModified;
+ while (true) {
+ newModified = System.currentTimeMillis();
+ if (checkExisting(tempId, length, identifier)) {
+ touch(identifier, newModified);
+ conHelper.exec(deleteSQL, tempId);
+ break;
+ }
+ try {
+ // UPDATE DATASTORE SET ID=?, LENGTH=?, LAST_MODIFIED=?
+ // WHERE ID=? AND LAST_MODIFIED=?
+ int count = conHelper.update(updateSQL,
+ id, length, newModified, tempId, tempModified);
+ // If update count is 0, the last modified time of the
+ // temporary row was changed - which means we need to
+ // re-try using a new last modified date (a later one)
+ // because we need to ensure the new last modified date
+ // is _newer_ than the old (otherwise the garbage
+ // collection could delete rows)
+ if (count != 0) {
+ // update was successful
+ break;
+ }
+ } catch (SQLException e) {
+ // duplicate key (the row already exists) - repeat
+ // we use exception handling for flow control here, which is bad,
+ // but the alternative is to use UPDATE ... WHERE ... (SELECT ...)
+ // which could cause a deadlock in some databases - also,
+ // duplicate key will only occur if somebody else concurrently
+ // added the same record (which is very unlikely)
+ }
+ // SELECT LENGTH, LAST_MODIFIED FROM DATASTORE WHERE ID=?
+ rs = conHelper.query(selectMetaSQL, tempId);
+ if (!rs.next()) {
+ // the row was deleted, which is unexpected / not allowed
+ String msg =
+ DIGEST + " temporary entry deleted: " +
+ " id=" + tempId + " length=" + length;
+ log.error(msg);
+ throw new DataStoreException(msg);
+ }
+ tempModified = rs.getLong(2);
+ DbUtility.close(rs);
+ rs = null;
+ }
+ usesIdentifier(identifier);
+ DbDataRecord record = new DbDataRecord(this, identifier, length, newModified);
+ return record;
+ } catch (Exception e) {
+ throw convert("Can not insert new record", e);
+ } finally {
+ if (tempId != null) {
+ temporaryInUse.remove(tempId);
+ }
+ DbUtility.close(rs);
+ if (fileInput != null) {
+ try {
+ fileInput.close();
+ } catch (IOException e) {
+ throw convert("Can not close temporary file", e);
+ }
+ }
+ }
+ }
+
+ /**
+ * Check if a row with this ID already exists.
+ *
+ * @return true if the row exists and the length matches
+ * @throw DataStoreException if a row exists, but the length is different
+ */
+ private boolean checkExisting(String tempId, long length, DataIdentifier identifier) throws DataStoreException, SQLException {
+ String id = identifier.toString();
+ // SELECT LENGTH, LAST_MODIFIED FROM DATASTORE WHERE ID=?
+ ResultSet rs = null;
+ try {
+ rs = conHelper.query(selectMetaSQL, id);
+ if (rs.next()) {
+ long oldLength = rs.getLong(1);
+ long lastModified = rs.getLong(2);
+ if (oldLength != length) {
+ String msg =
+ DIGEST + " collision: temp=" + tempId
+ + " id=" + id + " length=" + length
+ + " oldLength=" + oldLength;
+ log.error(msg);
+ throw new DataStoreException(msg);
+ }
+ DbUtility.close(rs);
+ rs = null;
+ touch(identifier, lastModified);
+ // row already exists
+ conHelper.exec(deleteSQL, tempId);
+ return true;
+ }
+ } finally {
+ DbUtility.close(rs);
+ }
+ return false;
+ }
+
+ /**
+ * Creates a temp file and copies the data there.
+ * The input stream is closed afterwards.
+ *
+ * @param in the input stream
+ * @return the file
+ * @throws IOException
+ */
+ private File moveToTempFile(InputStream in) throws IOException {
+ File temp = File.createTempFile("dbRecord", null);
+ writeToFileAndClose(in, temp);
+ return temp;
+ }
+
+ private void writeToFileAndClose(InputStream in, File file) throws IOException {
+ OutputStream out = new FileOutputStream(file);
+ try {
+ IOUtils.copy(in, out);
+ } finally {
+ IOUtils.closeQuietly(out);
+ IOUtils.closeQuietly(in);
+ }
+ }
+
+ public synchronized void deleteRecord(DataIdentifier identifier) throws DataStoreException {
+ try {
+ conHelper.exec(deleteSQL, identifier.toString());
+ } catch (Exception e) {
+ throw convert("Can not delete record", e);
+ }
+ }
+
+ public synchronized int deleteAllOlderThan(long min) throws DataStoreException {
+ try {
+ ArrayList<String> touch = new ArrayList<String>();
+ ArrayList<DataIdentifier> ids = new ArrayList<DataIdentifier>(inUse.keySet());
+ for (DataIdentifier identifier: ids) {
+ if (identifier != null) {
+ touch.add(identifier.toString());
+ }
+ }
+ touch.addAll(temporaryInUse);
+ for (String key : touch) {
+ updateLastModifiedDate(key, 0);
+ }
+ // DELETE FROM DATASTORE WHERE LAST_MODIFIED<?
+ log.debug(deleteOlderSQL + " - Scanstart [" + min + "]");
+ return conHelper.update(deleteOlderSQL, min);
+ } catch (Exception e) {
+ throw convert("Can not delete records", e);
+ }
+ }
+
+ public Iterator<DataIdentifier> getAllIdentifiers() throws DataStoreException {
+ ArrayList<DataIdentifier> list = new ArrayList<DataIdentifier>();
+ ResultSet rs = null;
+ try {
+ // SELECT ID FROM DATASTORE
+ rs = conHelper.query(selectAllSQL);
+ while (rs.next()) {
+ String id = rs.getString(1);
+ if (!id.startsWith(TEMP_PREFIX)) {
+ DataIdentifier identifier = new DataIdentifier(id);
+ list.add(identifier);
+ }
+ }
+ log.debug("Found " + list.size() + " identifiers.");
+ return list.iterator();
+ } catch (Exception e) {
+ throw convert("Can not read records", e);
+ } finally {
+ DbUtility.close(rs);
+ }
+ }
+
+ public int getMinRecordLength() {
+ return minRecordLength;
+ }
+
+ /**
+ * Set the minimum object length.
+ * The maximum value is around 32000.
+ *
+ * @param minRecordLength the length
+ */
+ public void setMinRecordLength(int minRecordLength) {
+ this.minRecordLength = minRecordLength;
+ }
+
+ public DataRecord getRecordIfStored(DataIdentifier identifier) throws DataStoreException {
+ usesIdentifier(identifier);
+ ResultSet rs = null;
+ try {
+ String id = identifier.toString();
+ // SELECT LENGTH, LAST_MODIFIED FROM DATASTORE WHERE ID = ?
+ rs = conHelper.query(selectMetaSQL, id);
+ if (!rs.next()) {
+ return null;
+ }
+ long length = rs.getLong(1);
+ long lastModified = rs.getLong(2);
+ DbUtility.close(rs);
+ rs = null;
+ lastModified = touch(identifier, lastModified);
+ return new DbDataRecord(this, identifier, length, lastModified);
+ } catch (Exception e) {
+ throw convert("Can not read identifier " + identifier, e);
+ } finally {
+ DbUtility.close(rs);
+ }
+ }
+
+ /**
+ * Open the input stream. This method sets those fields of the caller
+ * that need to be closed once the input stream is read.
+ *
+ * @param inputStream the database input stream object
+ * @param identifier data identifier
+ * @throws DataStoreException if the data store could not be accessed,
+ * or if the given identifier is invalid
+ */
+ InputStream openStream(DbInputStream inputStream, DataIdentifier identifier) throws DataStoreException {
+ ResultSet rs = null;
+ try {
+ // SELECT ID, DATA FROM DATASTORE WHERE ID = ?
+ rs = conHelper.query(selectDataSQL, identifier.toString());
+ if (!rs.next()) {
+ throw new DataStoreException("Record not found: " + identifier);
+ }
+ InputStream stream = rs.getBinaryStream(2);
+ if (stream == null) {
+ stream = new ByteArrayInputStream(new byte[0]);
+ DbUtility.close(rs);
+ } else if (copyWhenReading) {
+ // If we copy while reading, create a temp file and close the stream
+ File temp = moveToTempFile(stream);
+ stream = new BufferedInputStream(new TempFileInputStream(temp));
+ DbUtility.close(rs);
+ } else {
+ stream = new BufferedInputStream(stream);
+ inputStream.setResultSet(rs);
+ }
+ return stream;
+ } catch (Exception e) {
+ DbUtility.close(rs);
+ throw convert("Retrieving database resource ", e);
+ }
+ }
+
+ public synchronized void init(String homeDir) throws DataStoreException {
+ try {
+ initDatabaseType();
+
+ conHelper = createConnectionHelper(getDataSource());
+
+ if (isSchemaCheckEnabled()) {
+ createCheckSchemaOperation().run();
+ }
+ } catch (Exception e) {
+ throw convert("Can not init data store, driver=" + driver + " url=" + url + " user=" + user +
+ " schemaObjectPrefix=" + schemaObjectPrefix + " tableSQL=" + tableSQL + " createTableSQL=" + createTableSQL, e);
+ }
+ }
+
+ private DataSource getDataSource() throws Exception {
+ if (getDataSourceName() == null || "".equals(getDataSourceName())) {
+ return connectionFactory.getDataSource(getDriver(), getUrl(), getUser(), getPassword());
+ } else {
+ return connectionFactory.getDataSource(dataSourceName);
+ }
+ }
+
+ /**
+ * This method is called from the {@link #init(String)} method of this class and returns a
+ * {@link ConnectionHelper} instance which is assigned to the {@code conHelper} field. Subclasses may
+ * override it to return a specialized connection helper.
+ *
+ * @param dataSrc the {@link DataSource} of this persistence manager
+ * @return a {@link ConnectionHelper}
+ * @throws Exception on error
+ */
+ protected ConnectionHelper createConnectionHelper(DataSource dataSrc) throws Exception {
+ return new ConnectionHelper(dataSrc, false);
+ }
+
+ /**
+ * This method is called from {@link #init(String)} after the
+ * {@link #createConnectionHelper(DataSource)} method, and returns a default {@link CheckSchemaOperation}.
+ *
+ * @return a new {@link CheckSchemaOperation} instance
+ */
+ protected final CheckSchemaOperation createCheckSchemaOperation() {
+ String tableName = tablePrefix + schemaObjectPrefix + tableSQL;
+ return new CheckSchemaOperation(conHelper, new ByteArrayInputStream(createTableSQL.getBytes()), tableName);
+ }
+
+ protected void initDatabaseType() throws DataStoreException {
+ boolean failIfNotFound = false;
+ if (databaseType == null) {
+ if (dataSourceName != null) {
+ try {
+ databaseType = connectionFactory.getDataBaseType(dataSourceName);
+ } catch (RepositoryException e) {
+ throw new DataStoreException(e);
+ }
+ } else {
+ if (!url.startsWith("jdbc:")) {
+ return;
+ }
+ int start = "jdbc:".length();
+ int end = url.indexOf(':', start);
+ databaseType = url.substring(start, end);
+ }
+ } else {
+ failIfNotFound = true;
+ }
+
+ InputStream in =
+ DbDataStore.class.getResourceAsStream(databaseType + ".properties");
+ if (in == null) {
+ if (failIfNotFound) {
+ String msg =
+ "Configuration error: The resource '" + databaseType
+ + ".properties' could not be found;"
+ + " Please verify the databaseType property";
+ log.debug(msg);
+ throw new DataStoreException(msg);
+ } else {
+ return;
+ }
+ }
+ Properties prop = new Properties();
+ try {
+ try {
+ prop.load(in);
+ } finally {
+ in.close();
+ }
+ } catch (IOException e) {
+ String msg = "Configuration error: Could not read properties '" + databaseType + ".properties'";
+ log.debug(msg);
+ throw new DataStoreException(msg, e);
+ }
+ if (driver == null) {
+ driver = getProperty(prop, "driver", driver);
+ }
+ tableSQL = getProperty(prop, "table", tableSQL);
+ createTableSQL = getProperty(prop, "createTable", createTableSQL);
+ insertTempSQL = getProperty(prop, "insertTemp", insertTempSQL);
+ updateDataSQL = getProperty(prop, "updateData", updateDataSQL);
+ updateLastModifiedSQL = getProperty(prop, "updateLastModified", updateLastModifiedSQL);
+ updateSQL = getProperty(prop, "update", updateSQL);
+ deleteSQL = getProperty(prop, "delete", deleteSQL);
+ deleteOlderSQL = getProperty(prop, "deleteOlder", deleteOlderSQL);
+ selectMetaSQL = getProperty(prop, "selectMeta", selectMetaSQL);
+ selectAllSQL = getProperty(prop, "selectAll", selectAllSQL);
+ selectDataSQL = getProperty(prop, "selectData", selectDataSQL);
+ storeStream = getProperty(prop, "storeStream", storeStream);
+ if (!STORE_SIZE_MINUS_ONE.equals(storeStream)
+ && !STORE_TEMP_FILE.equals(storeStream)
+ && !STORE_SIZE_MAX.equals(storeStream)) {
+ String msg = "Unsupported Stream store mechanism: " + storeStream
+ + " supported are: " + STORE_SIZE_MINUS_ONE + ", "
+ + STORE_TEMP_FILE + ", " + STORE_SIZE_MAX;
+ log.debug(msg);
+ throw new DataStoreException(msg);
+ }
+ }
+
+ /**
+ * Get the expanded property value. The following placeholders are supported:
+ * ${table}: the table name (the default is DATASTORE) and
+ * ${tablePrefix}: tablePrefix plus schemaObjectPrefix as set in the configuration
+ *
+ * @param prop the properties object
+ * @param key the key
+ * @param defaultValue the default value
+ * @return the property value (placeholders are replaced)
+ */
+ protected String getProperty(Properties prop, String key, String defaultValue) {
+ String sql = prop.getProperty(key, defaultValue);
+ sql = Text.replace(sql, "${table}", tableSQL).trim();
+ sql = Text.replace(sql, "${tablePrefix}", tablePrefix + schemaObjectPrefix).trim();
+ return sql;
+ }
+
+ /**
+ * Convert an exception to a data store exception.
+ *
+ * @param cause the message
+ * @param e the root cause
+ * @return the data store exception
+ */
+ protected DataStoreException convert(String cause, Exception e) {
+ log.warn(cause, e);
+ if (e instanceof DataStoreException) {
+ return (DataStoreException) e;
+ } else {
+ return new DataStoreException(cause, e);
+ }
+ }
+
+ public void updateModifiedDateOnAccess(long before) {
+ log.debug("Update modifiedDate on access before " + before);
+ minModifiedDate = before;
+ }
+
+ /**
+ * Update the modified date of an entry if required.
+ *
+ * @param identifier the entry identifier
+ * @param lastModified the current last modified date
+ * @return the new modified date
+ */
+ long touch(DataIdentifier identifier, long lastModified) throws DataStoreException {
+ usesIdentifier(identifier);
+ return updateLastModifiedDate(identifier.toString(), lastModified);
+ }
+
+ private long updateLastModifiedDate(String key, long lastModified) throws DataStoreException {
+ if (lastModified < minModifiedDate) {
+ long now = System.currentTimeMillis();
+ try {
+ // UPDATE DATASTORE SET LAST_MODIFIED = ? WHERE ID = ? AND LAST_MODIFIED < ?
+ conHelper.update(updateLastModifiedSQL, now, key, now);
+ return now;
+ } catch (Exception e) {
+ throw convert("Can not update lastModified", e);
+ }
+ }
+ return lastModified;
+ }
+
+ /**
+ * Get the database type (if set).
+ * @return the database type
+ */
+ public String getDatabaseType() {
+ return databaseType;
+ }
+
+ /**
+ * Set the database type. By default the sub-protocol of the JDBC database URL is used if it is not set.
+ * It must match the resource file [databaseType].properties. Example: mysql.
+ *
+ * @param databaseType
+ */
+ public void setDatabaseType(String databaseType) {
+ this.databaseType = databaseType;
+ }
+
+ /**
+ * Get the database driver
+ *
+ * @return the driver
+ */
+ public String getDriver() {
+ return driver;
+ }
+
+ /**
+ * Set the database driver class name.
+ * If not set, the default driver class name for the database type is used,
+ * as set in the [databaseType].properties resource; key 'driver'.
+ *
+ * @param driver
+ */
+ public void setDriver(String driver) {
+ this.driver = driver;
+ }
+
+ /**
+ * Get the password.
+ *
+ * @return the password
+ */
+ public String getPassword() {
+ return password;
+ }
+
+ /**
+ * Set the password.
+ *
+ * @param password
+ */
+ public void setPassword(String password) {
+ this.password = password;
+ }
+
+ /**
+ * Get the database URL.
+ *
+ * @return the URL
+ */
+ public String getUrl() {
+ return url;
+ }
+
+ /**
+ * Set the database URL.
+ * Example: jdbc:postgresql:test
+ *
+ * @param url
+ */
+ public void setUrl(String url) {
+ this.url = url;
+ }
+
+ /**
+ * Get the user name.
+ *
+ * @return the user name
+ */
+ public String getUser() {
+ return user;
+ }
+
+ /**
+ * Set the user name.
+ *
+ * @param user
+ */
+ public void setUser(String user) {
+ this.user = user;
+ }
+
+ /**
+ * @return whether the schema check is enabled
+ */
+ public final boolean isSchemaCheckEnabled() {
+ return schemaCheckEnabled;
+ }
+
+ /**
+ * @param enabled set whether the schema check is enabled
+ */
+ public final void setSchemaCheckEnabled(boolean enabled) {
+ schemaCheckEnabled = enabled;
+ }
+
+ public synchronized void close() throws DataStoreException {
+ // nothing to do
+ }
+
+ protected void usesIdentifier(DataIdentifier identifier) {
+ inUse.put(identifier, new WeakReference<DataIdentifier>(identifier));
+ }
+
+ public void clearInUse() {
+ inUse.clear();
+ }
+
+ protected synchronized MessageDigest getDigest() throws DataStoreException {
+ try {
+ return MessageDigest.getInstance(DIGEST);
+ } catch (NoSuchAlgorithmException e) {
+ throw convert("No such algorithm: " + DIGEST, e);
+ }
+ }
+
+ /**
+ * Get the maximum number of concurrent connections.
+ *
+ * @deprecated
+ * @return the maximum number of connections.
+ */
+ @Deprecated
+ public int getMaxConnections() {
+ return -1;
+ }
+
+ /**
+ * Set the maximum number of concurrent connections in the pool.
+ * At least 3 connections are required if the garbage collection process is used.
+ *
+ * @deprecated
+ * @param maxConnections the new value
+ */
+ @Deprecated
+ public void setMaxConnections(int maxConnections) {
+ // no effect
+ }
+
+ /**
+ * Is a stream copied to a temporary file before returning?
+ *
+ * @return the setting
+ */
+ public boolean getCopyWhenReading() {
+ return copyWhenReading;
+ }
+
+ /**
+ * The the copy setting. If enabled,
+ * a stream is always copied to a temporary file when reading a stream.
+ *
+ * @param copyWhenReading the new setting
+ */
+ public void setCopyWhenReading(boolean copyWhenReading) {
+ this.copyWhenReading = copyWhenReading;
+ }
+
+ /**
+ * Get the table prefix.
+ *
+ * @return the table prefix.
+ */
+ public String getTablePrefix() {
+ return tablePrefix;
+ }
+
+ /**
+ * Set the new table prefix. The default is empty.
+ * The table name is constructed like this:
+ * ${tablePrefix}${schemaObjectPrefix}${tableName}
+ *
+ * @param tablePrefix the new value
+ */
+ public void setTablePrefix(String tablePrefix) {
+ this.tablePrefix = tablePrefix;
+ }
+
+ /**
+ * Get the schema prefix.
+ *
+ * @return the schema object prefix
+ */
+ public String getSchemaObjectPrefix() {
+ return schemaObjectPrefix;
+ }
+
+ /**
+ * Set the schema object prefix. The default is empty.
+ * The table name is constructed like this:
+ * ${tablePrefix}${schemaObjectPrefix}${tableName}
+ *
+ * @param schemaObjectPrefix the new prefix
+ */
+ public void setSchemaObjectPrefix(String schemaObjectPrefix) {
+ this.schemaObjectPrefix = schemaObjectPrefix;
+ }
+
+ public String getDataSourceName() {
+ return dataSourceName;
+ }
+
+ public void setDataSourceName(String dataSourceName) {
+ this.dataSourceName = dataSourceName;
+ }
+}
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/data/db/DbInputStream.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/db/DbInputStream.java
new file mode 100644
index 0000000000..99a482c576
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/db/DbInputStream.java
@@ -0,0 +1,191 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.jackrabbit.core.data.db;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.sql.ResultSet;
+
+import org.apache.commons.io.input.AutoCloseInputStream;
+import org.apache.jackrabbit.core.data.DataIdentifier;
+import org.apache.jackrabbit.core.data.DataStoreException;
+import org.apache.jackrabbit.core.util.db.DbUtility;
+
+/**
+ * This class represents an input stream backed by a database. The database
+ * objects are only acquired when reading from the stream, and stay open until
+ * the stream is closed, fully read, or garbage collected.
+ * <p>
+ * This class does not support mark/reset. It is always to be wrapped
+ * using a BufferedInputStream.
+ */
+public class DbInputStream extends AutoCloseInputStream {
+
+ protected DbDataStore store;
+ protected DataIdentifier identifier;
+ protected boolean endOfStream;
+
+ protected ResultSet rs;
+ private InputStream initialStream;
+
+ /**
+ * Create a database input stream for the given identifier.
+ * Database access is delayed until the first byte is read from the stream.
+ *
+ * @param store the database data store
+ * @param identifier the data identifier
+ */
+ protected DbInputStream(DbDataStore store, DataIdentifier identifier) {
+ super(null);
+ this.store = store;
+ this.identifier = identifier;
+ this.initialStream = super.in;
+ }
+
+ /**
+ * Open the stream if required.
+ *
+ * @throws IOException
+ */
+ protected void openStream() throws IOException {
+ if (endOfStream) {
+ throw new EOFException();
+ }
+ if (in == initialStream) {
+ try {
+ in = store.openStream(this, identifier);
+ } catch (DataStoreException e) {
+ throw new IOException(e.getMessage(), e);
+ }
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ * When the stream is consumed, the database objects held by the instance are closed.
+ */
+ public int read() throws IOException {
+ if (endOfStream) {
+ return -1;
+ }
+ openStream();
+ int c = in.read();
+ if (c == -1) {
+ endOfStream = true;
+ close();
+ }
+ return c;
+ }
+
+ /**
+ * {@inheritDoc}
+ * When the stream is consumed, the database objects held by the instance are closed.
+ */
+ public int read(byte[] b) throws IOException {
+ return read(b, 0, b.length);
+ }
+
+ /**
+ * {@inheritDoc}
+ * When the stream is consumed, the database objects held by the instance are closed.
+ */
+ public int read(byte[] b, int off, int len) throws IOException {
+ if (endOfStream) {
+ return -1;
+ }
+ openStream();
+ int c = in.read(b, off, len);
+ if (c == -1) {
+ endOfStream = true;
+ close();
+ }
+ return c;
+ }
+
+ /**
+ * {@inheritDoc}
+ * When the stream is consumed, the database objects held by the instance are closed.
+ */
+ public void close() throws IOException {
+ if (in != initialStream) {
+ in.close();
+ in = initialStream;
+ // some additional database objects
+ // may need to be closed
+ if (rs != null) {
+ DbUtility.close(rs);
+ rs = null;
+ }
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public long skip(long n) throws IOException {
+ if (endOfStream) {
+ return -1;
+ }
+ openStream();
+ return in.skip(n);
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public int available() throws IOException {
+ if (endOfStream) {
+ return 0;
+ }
+ openStream();
+ return in.available();
+ }
+
+ /**
+ * This method does nothing.
+ */
+ public void mark(int readlimit) {
+ // do nothing
+ }
+
+ /**
+ * This method does nothing.
+ */
+ public void reset() throws IOException {
+ // do nothing
+ }
+
+ /**
+ * Check whether mark and reset are supported.
+ *
+ * @return false
+ */
+ public boolean markSupported() {
+ return false;
+ }
+
+ /**
+ * Set the result set of this input stream. This object must be closed once
+ * the stream is closed.
+ *
+ * @param rs the result set
+ */
+ void setResultSet(ResultSet rs) {
+ this.rs = rs;
+ }
+}
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/data/db/DerbyDataStore.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/db/DerbyDataStore.java
new file mode 100644
index 0000000000..c1a8ff665d
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/db/DerbyDataStore.java
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.jackrabbit.core.data.db;
+
+import java.sql.SQLException;
+
+import javax.sql.DataSource;
+
+import org.apache.jackrabbit.core.data.DataStoreException;
+import org.apache.jackrabbit.core.util.db.ConnectionHelper;
+import org.apache.jackrabbit.core.util.db.DerbyConnectionHelper;
+
+/**
+ * The Derby data store closes the database when the data store is closed
+ * (embedded databases only).
+ */
+public class DerbyDataStore extends DbDataStore {
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ protected ConnectionHelper createConnectionHelper(DataSource dataSrc) throws Exception {
+ return new DerbyConnectionHelper(dataSrc, false);
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public synchronized void close() throws DataStoreException {
+ super.close();
+ try {
+ ((DerbyConnectionHelper) conHelper).shutDown(getDriver());
+ } catch (SQLException e) {
+ throw new DataStoreException(e);
+ }
+ }
+}
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/data/db/ResettableTempFileInputStream.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/db/ResettableTempFileInputStream.java
new file mode 100644
index 0000000000..203c6dd8b6
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/db/ResettableTempFileInputStream.java
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.jackrabbit.core.data.db;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.nio.channels.FileChannel;
+
+/**
+ * TempFileInputStream that can be reset in order to allow the
+ * ConnectionHelper to retry SQL execution in case of failure.
+ */
+public class ResettableTempFileInputStream extends TempFileInputStream {
+
+ private final FileChannel fileChannel;
+ private long mark = 0;
+
+ public ResettableTempFileInputStream(final File file) throws FileNotFoundException {
+ this(new FileInputStream(file), file);
+ }
+
+ private ResettableTempFileInputStream(final FileInputStream in, final File file) {
+ super(in, file);
+ this.fileChannel = in.getChannel();
+ }
+
+ @Override
+ public boolean markSupported() {
+ return true;
+ }
+
+ @Override
+ public synchronized void mark(int readlimit) {
+ try {
+ mark = fileChannel.position();
+ } catch (IOException ex) {
+ mark = -1;
+ }
+ }
+
+ @Override
+ public synchronized void reset() throws IOException {
+ if (mark == -1) {
+ throw new IOException("Mark failed");
+ }
+ fileChannel.position(mark);
+ }
+}
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/data/db/TempFileInputStream.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/db/TempFileInputStream.java
new file mode 100644
index 0000000000..5419e2c4ca
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/db/TempFileInputStream.java
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.jackrabbit.core.data.db;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileNotFoundException;
+import java.io.FilterInputStream;
+import java.io.IOException;
+
+import org.apache.commons.io.input.ClosedInputStream;
+
+/**
+ * An input stream from a temporary file. The file is deleted when the stream is
+ * closed or garbage collected.
+ */
+public class TempFileInputStream extends FilterInputStream {
+
+ private final File file;
+
+ public TempFileInputStream(File file) throws FileNotFoundException {
+ this(new FileInputStream(file), file);
+ }
+
+ protected TempFileInputStream(FileInputStream in, File file) {
+ super(in);
+ this.file = file;
+ }
+
+ @Override
+ public void close() throws IOException {
+ in.close();
+ in = new ClosedInputStream();
+ file.delete();
+ }
+
+ @Override
+ protected void finalize() throws Throwable {
+ close();
+ super.finalize();
+ }
+
+}
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/data/db/package-info.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/db/package-info.java
new file mode 100755
index 0000000000..26277ba6fb
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/db/package-info.java
@@ -0,0 +1,19 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+/* see JCR-4060 */
+@org.osgi.annotation.versioning.Version("2.13.6")
+package org.apache.jackrabbit.core.data.db;
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/data/package-info.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/package-info.java
new file mode 100755
index 0000000000..cc474572b5
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/package-info.java
@@ -0,0 +1,19 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+/* see JCR-4060 */
+@org.osgi.annotation.versioning.Version("2.15.0")
+package org.apache.jackrabbit.core.data;
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/data/util/NamedThreadFactory.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/util/NamedThreadFactory.java
new file mode 100644
index 0000000000..e176db7243
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/util/NamedThreadFactory.java
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.jackrabbit.core.data.util;
+
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * This class extends {@link ThreadFactory} to creates named threads.
+ */
+public class NamedThreadFactory implements ThreadFactory {
+
+ private AtomicInteger threadCount = new AtomicInteger(1);
+
+ String threadPrefixName;
+
+ public NamedThreadFactory(String threadPrefixName) {
+ super();
+ this.threadPrefixName = threadPrefixName;
+ }
+
+ public Thread newThread(Runnable r) {
+ Thread thread = new Thread(r);
+ thread.setContextClassLoader(getClass().getClassLoader());
+ thread.setName(threadPrefixName + "-" + threadCount.getAndIncrement());
+ return thread;
+ }
+
+}
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/data/util/package-info.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/util/package-info.java
new file mode 100755
index 0000000000..e1d33779c4
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/data/util/package-info.java
@@ -0,0 +1,19 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+/* see JCR-4060 */
+@org.osgi.annotation.versioning.Version("2.13.5")
+package org.apache.jackrabbit.core.data.util;
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/fs/BasedFileSystem.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/fs/BasedFileSystem.java
new file mode 100644
index 0000000000..bf3bbdf334
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/fs/BasedFileSystem.java
@@ -0,0 +1,187 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.jackrabbit.core.fs;
+
+import java.io.InputStream;
+import java.io.OutputStream;
+
+
+/**
+ * A <code>BasedFileSystem</code> represents a 'file system in a file system'.
+ */
+public class BasedFileSystem implements FileSystem {
+
+ protected final FileSystem fsBase;
+
+ protected final String basePath;
+
+ /**
+ * Creates a new <code>BasedFileSystem</code>
+ *
+ * @param fsBase the <code>FileSystem</code> the new file system should be based on
+ * @param relRootPath the root path relative to <code>fsBase</code>'s root
+ */
+ public BasedFileSystem(FileSystem fsBase, String relRootPath) {
+ if (fsBase == null) {
+ throw new IllegalArgumentException("invalid file system argument");
+ }
+ this.fsBase = fsBase;
+
+ if (relRootPath == null) {
+ throw new IllegalArgumentException("invalid null path argument");
+ }
+ if (relRootPath.equals(SEPARATOR)) {
+ throw new IllegalArgumentException("invalid path argument");
+ }
+ if (!relRootPath.startsWith(SEPARATOR)) {
+ relRootPath = SEPARATOR + relRootPath;
+ }
+ if (relRootPath.endsWith(SEPARATOR)) {
+ relRootPath = relRootPath.substring(0, relRootPath.length() - 1);
+
+ }
+ this.basePath = relRootPath;
+ }
+
+ protected String buildBasePath(String path) {
+ if (path.startsWith(SEPARATOR)) {
+ if (path.length() == 1) {
+ return basePath;
+ } else {
+ return basePath + path;
+ }
+ } else {
+ return basePath + SEPARATOR + path;
+ }
+ }
+
+ //-----------------------------------------------------------< FileSystem >
+ /**
+ * {@inheritDoc}
+ */
+ public void init() throws FileSystemException {
+ // check base path
+ if (!fsBase.isFolder(basePath)) {
+ fsBase.createFolder(basePath);
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public void close() throws FileSystemException {
+ // do nothing; base file system should be closed explicitly
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public void createFolder(String folderPath) throws FileSystemException {
+ fsBase.createFolder(buildBasePath(folderPath));
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public void deleteFile(String filePath) throws FileSystemException {
+ fsBase.deleteFile(buildBasePath(filePath));
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public void deleteFolder(String folderPath) throws FileSystemException {
+ fsBase.deleteFolder(buildBasePath(folderPath));
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public boolean exists(String path) throws FileSystemException {
+ return fsBase.exists(buildBasePath(path));
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public InputStream getInputStream(String filePath) throws FileSystemException {
+ return fsBase.getInputStream(buildBasePath(filePath));
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public OutputStream getOutputStream(String filePath) throws FileSystemException {
+ return fsBase.getOutputStream(buildBasePath(filePath));
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public boolean hasChildren(String path) throws FileSystemException {
+ return fsBase.hasChildren(buildBasePath(path));
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public boolean isFile(String path) throws FileSystemException {
+ return fsBase.isFile(buildBasePath(path));
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public boolean isFolder(String path) throws FileSystemException {
+ return fsBase.isFolder(buildBasePath(path));
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public long lastModified(String path) throws FileSystemException {
+ return fsBase.lastModified(buildBasePath(path));
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public long length(String filePath) throws FileSystemException {
+ return fsBase.length(buildBasePath(filePath));
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public String[] list(String folderPath) throws FileSystemException {
+ return fsBase.list(buildBasePath(folderPath));
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public String[] listFiles(String folderPath) throws FileSystemException {
+ return fsBase.listFiles(buildBasePath(folderPath));
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public String[] listFolders(String folderPath) throws FileSystemException {
+ return fsBase.listFolders(buildBasePath(folderPath));
+ }
+}
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/fs/FileSystem.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/fs/FileSystem.java
new file mode 100644
index 0000000000..babc0f873b
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/fs/FileSystem.java
@@ -0,0 +1,206 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.jackrabbit.core.fs;
+
+import java.io.InputStream;
+import java.io.OutputStream;
+
+/**
+ * The <code>FileSystem</code> interface is an abstraction of a virtual
+ * file system. The similarities of its method names with with the methods
+ * of the <code>java.io.File</code> class are intentional.
+ * <br>
+ * Implementations of this interface expose a file system-like resource.
+ * File system-like resources include WebDAV-enabled servers, local file systems,
+ * and so forth.
+ */
+public interface FileSystem {
+
+ /**
+ * File separator
+ */
+ String SEPARATOR = "/";
+
+ /**
+ * File separator character
+ */
+ char SEPARATOR_CHAR = '/';
+
+ /**
+ * Initialize the file system
+ *
+ * @throws FileSystemException if the file system initialization fails
+ */
+ void init() throws FileSystemException;
+
+ /**
+ * Close the file system. After calling this method, the file system is no
+ * longer accessible.
+ *
+ * @throws FileSystemException
+ */
+ void close() throws FileSystemException;
+
+ /**
+ * Returns an input stream of the contents of the file denoted by this path.
+ *
+ * @param filePath the path of the file.
+ * @return an input stream of the contents of the file.
+ * @throws FileSystemException if the file does not exist
+ * or if it cannot be read from
+ */
+ InputStream getInputStream(String filePath) throws FileSystemException;
+
+ /**
+ * Returns an output stream for writing bytes to the file denoted by this path.
+ * The file will be created if it doesn't exist. If the file exists, its contents
+ * will be overwritten.
+ *
+ * @param filePath the path of the file.
+ * @return an output stream for writing bytes to the file.
+ * @throws FileSystemException if the file cannot be written to or created
+ */
+ OutputStream getOutputStream(String filePath) throws FileSystemException;
+
+ /**
+ * Creates the folder named by this path, including any necessary but
+ * nonexistent parent folders. Note that if this operation fails it
+ * may have succeeded in creating some of the necessary parent folders.
+ *
+ * @param folderPath the path of the folder to be created.
+ * @throws FileSystemException if a file system entry denoted by path
+ * already exists or if another error occurs.
+ */
+ void createFolder(String folderPath) throws FileSystemException;
+
+ /**
+ * Tests whether the file system entry denoted by this path exists.
+ *
+ * @param path the path of a file system entry.
+ * @return true if the file system entry at path exists; false otherwise.
+ * @throws FileSystemException
+ */
+ boolean exists(String path) throws FileSystemException;
+
+ /**
+ * Tests whether the file system entry denoted by this path exists and
+ * is a file.
+ *
+ * @param path the path of a file system entry.
+ * @return true if the file system entry at path is a file; false otherwise.
+ * @throws FileSystemException
+ */
+ boolean isFile(String path) throws FileSystemException;
+
+ /**
+ * Tests whether the file system entry denoted by this path exists and
+ * is a folder.
+ *
+ * @param path the path of a file system entry.
+ * @return true if the file system entry at path is a folder; false otherwise.
+ * @throws FileSystemException
+ */
+ boolean isFolder(String path) throws FileSystemException;
+
+ /**
+ * Tests whether the file system entry denoted by this path has child entries.
+ *
+ * @param path the path of a file system entry.
+ * @return true if the file system entry at path has child entries; false otherwise.
+ * @throws FileSystemException
+ */
+ boolean hasChildren(String path) throws FileSystemException;
+
+ /**
+ * Returns the length of the file denoted by this path.
+ *
+ * @param filePath the path of the file.
+ * @return The length, in bytes, of the file denoted by this path,
+ * or -1L if the length can't be determined.
+ * @throws FileSystemException if the path does not denote an existing file.
+ */
+ long length(String filePath) throws FileSystemException;
+
+ /**
+ * Returns the time that the file system entry denoted by this path
+ * was last modified.
+ *
+ * @param path the path of a file system entry.
+ * @return A long value representing the time the file system entry was
+ * last modified, measured in milliseconds since the epoch
+ * (00:00:00 GMT, January 1, 1970), or 0L if the modification
+ * time can't be determined.
+ * @throws FileSystemException if the file system entry does not exist.
+ */
+ long lastModified(String path) throws FileSystemException;
+
+ /**
+ * Returns an array of strings naming the files and folders
+ * in the folder denoted by this path.
+ *
+ * @param folderPath the path of the folder whose contents is to be listed.
+ * @return an array of strings naming the files and folders
+ * in the folder denoted by this path.
+ * @throws FileSystemException if this path does not denote a folder or if
+ * another error occurs.
+ */
+ String[] list(String folderPath) throws FileSystemException;
+
+ /**
+ * Returns an array of strings naming the files in the folder
+ * denoted by this path.
+ *
+ * @param folderPath the path of the folder whose contents is to be listed.
+ * @return an array of strings naming the files in the folder
+ * denoted by this path.
+ * @throws FileSystemException if this path does not denote a folder or if
+ * another error occurs.
+ */
+ String[] listFiles(String folderPath) throws FileSystemException;
+
+ /**
+ * Returns an array of strings naming the folders in the folder
+ * denoted by this path.
+ *
+ * @param folderPath the path of the folder whose contents is to be listed.
+ * @return an array of strings naming the folders in the folder
+ * denoted by this path.
+ * @throws FileSystemException if this path does not denote a folder or if
+ * another error occurs.
+ */
+ String[] listFolders(String folderPath) throws FileSystemException;
+
+ /**
+ * Deletes the file denoted by this path.
+ *
+ * @param filePath the path of the file to be deleted.
+ * @throws FileSystemException if this path does not denote a file or if
+ * another error occurs.
+ */
+ void deleteFile(String filePath) throws FileSystemException;
+
+ /**
+ * Deletes the folder denoted by this path. Any contents of this folder
+ * (folders and files) will be deleted recursively.
+ *
+ * @param folderPath the path of the folder to be deleted.
+ * @throws FileSystemException if this path does not denote a folder or if
+ * another error occurs.
+ */
+ void deleteFolder(String folderPath) throws FileSystemException;
+
+}
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/fs/FileSystemException.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/fs/FileSystemException.java
new file mode 100644
index 0000000000..620eeeb9aa
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/fs/FileSystemException.java
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.jackrabbit.core.fs;
+
+/**
+ * The <code>FileSystemException</code> signals an error within a file system
+ * operation. FileSystemExceptions are thrown by {@link FileSystem}
+ * implementations.
+ */
+public class FileSystemException extends Exception {
+
+ /**
+ * Constructs a new instance of this class with the specified detail
+ * message.
+ *
+ * @param message the detail message. The detail message is saved for
+ * later retrieval by the {@link #getMessage()} method.
+ */
+ public FileSystemException(String message) {
+ super(message);
+ }
+
+ /**
+ * Constructs a new instance of this class with the specified detail
+ * message and root cause.
+ *
+ * @param message the detail message. The detail message is saved for
+ * later retrieval by the {@link #getMessage()} method.
+ * @param rootCause root failure cause
+ */
+ public FileSystemException(String message, Throwable rootCause) {
+ super(message, rootCause);
+ }
+
+}
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/fs/FileSystemFactory.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/fs/FileSystemFactory.java
new file mode 100644
index 0000000000..012ab7d533
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/fs/FileSystemFactory.java
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.jackrabbit.core.fs;
+
+import javax.jcr.RepositoryException;
+
+
+/**
+ * Factory interface for creating {@link FileSystem} instances. Used
+ * to decouple the repository internals from the repository configuration
+ * mechanism.
+ */
+public interface FileSystemFactory {
+
+ /**
+ * Creates, initializes, and returns a {@link FileSystem} instance
+ * for use by the repository. Note that no information is passed from
+ * the client, so all required configuration information must be
+ * encapsulated in the factory.
+ *
+ * @return initialized file system
+ * @throws RepositoryException if the file system can not be created
+ */
+ FileSystem getFileSystem() throws RepositoryException;
+
+}
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/fs/FileSystemPathUtil.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/fs/FileSystemPathUtil.java
new file mode 100644
index 0000000000..af056fb37b
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/fs/FileSystemPathUtil.java
@@ -0,0 +1,229 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.jackrabbit.core.fs;
+
+import java.io.ByteArrayOutputStream;
+import java.util.BitSet;
+
+
+/**
+ * Utility class for handling paths in a file system.
+ */
+public final class FileSystemPathUtil {
+
+ /**
+ * Array of lowercase hexadecimal characters used in creating hex escapes.
+ */
+ private static final char[] HEX_TABLE = "0123456789abcdef".toCharArray();
+
+ /**
+ * The escape character used to mark hex escape sequences.
+ */
+ private static final char ESCAPE_CHAR = '%';
+
+ /**
+ * The list of characters that are not encoded by the <code>escapeName(String)</code>
+ * and <code>unescape(String)</code> methods. They contains the characters
+ * which can safely be used in file names:
+ */
+ public static final BitSet SAFE_NAMECHARS;
+
+ /**
+ * The list of characters that are not encoded by the <code>escapePath(String)</code>
+ * and <code>unescape(String)</code> methods. They contains the characters
+ * which can safely be used in file paths:
+ */
+ public static final BitSet SAFE_PATHCHARS;
+
+ static {
+ // build list of valid name characters
+ SAFE_NAMECHARS = new BitSet(256);
+ int i;
+ for (i = 'a'; i <= 'z'; i++) {
+ SAFE_NAMECHARS.set(i);
+ }
+ for (i = 'A'; i <= 'Z'; i++) {
+ SAFE_NAMECHARS.set(i);
+ }
+ for (i = '0'; i <= '9'; i++) {
+ SAFE_NAMECHARS.set(i);
+ }
+ SAFE_NAMECHARS.set('-');
+ SAFE_NAMECHARS.set('_');
+ SAFE_NAMECHARS.set('.');
+
+ // build list of valid path characters (includes name characters)
+ SAFE_PATHCHARS = (BitSet) SAFE_NAMECHARS.clone();
+ SAFE_PATHCHARS.set(FileSystem.SEPARATOR_CHAR);
+ }
+
+ /**
+ * private constructor
+ */
+ private FileSystemPathUtil() {
+ }
+
+ /**
+ * Escapes the given string using URL encoding for all bytes not included
+ * in the given set of safe characters.
+ *
+ * @param s the string to escape
+ * @param safeChars set of safe characters (bytes)
+ * @return escaped string
+ */
+ private static String escape(String s, BitSet safeChars) {
+ byte[] bytes = s.getBytes();
+ StringBuilder out = new StringBuilder(bytes.length);
+ for (int i = 0; i < bytes.length; i++) {
+ int c = bytes[i] & 0xff;
+ if (safeChars.get(c) && c != ESCAPE_CHAR) {
+ out.append((char) c);
+ } else {
+ out.append(ESCAPE_CHAR);
+ out.append(HEX_TABLE[(c >> 4) & 0x0f]);
+ out.append(HEX_TABLE[(c) & 0x0f]);
+ }
+ }
+ return out.toString();
+ }
+
+ /**
+ * Encodes the specified <code>path</code>. Same as
+ * <code>{@link #escapeName(String)}</code> except that the separator
+ * character <b><code>/</code></b> is regarded as a legal path character
+ * that needs no escaping.
+ *
+ * @param path the path to encode.
+ * @return the escaped path
+ */
+ public static String escapePath(String path) {
+ return escape(path, SAFE_PATHCHARS);
+ }
+
+ /**
+ * Encodes the specified <code>name</code>. Same as
+ * <code>{@link #escapePath(String)}</code> except that the separator character
+ * <b><code>/</code></b> is regarded as an illegal character that needs
+ * escaping.
+ *
+ * @param name the name to encode.
+ * @return the escaped name
+ */
+ public static String escapeName(String name) {
+ return escape(name, SAFE_NAMECHARS);
+ }
+
+ /**
+ * Decodes the specified path/name.
+ *
+ * @param pathOrName the escaped path/name
+ * @return the unescaped path/name
+ */
+ public static String unescape(String pathOrName) {
+ ByteArrayOutputStream out = new ByteArrayOutputStream(pathOrName.length());
+ for (int i = 0; i < pathOrName.length(); i++) {
+ char c = pathOrName.charAt(i);
+ if (c == ESCAPE_CHAR) {
+ try {
+ out.write(Integer.parseInt(pathOrName.substring(i + 1, i + 3), 16));
+ } catch (NumberFormatException e) {
+ IllegalArgumentException iae = new IllegalArgumentException("Failed to unescape escape sequence");
+ iae.initCause(e);
+ throw iae;
+ }
+ i += 2;
+ } else {
+ out.write(c);
+ }
+ }
+ return new String(out.toByteArray());
+ }
+
+ /**
+ * Tests whether the specified path represents the root path, i.e. "/".
+ *
+ * @param path path to test
+ * @return true if the specified path represents the root path; false otherwise.
+ */
+ public static boolean denotesRoot(String path) {
+ return path.equals(FileSystem.SEPARATOR);
+ }
+
+ /**
+ * Checks if <code>path</code> is a valid path.
+ *
+ * @param path the path to be checked
+ * @throws FileSystemException If <code>path</code> is not a valid path
+ */
+ public static void checkFormat(String path) throws FileSystemException {
+ if (path == null) {
+ throw new FileSystemException("null path");
+ }
+
+ // path must be absolute, i.e. starting with '/'
+ if (!path.startsWith(FileSystem.SEPARATOR)) {
+ throw new FileSystemException("not an absolute path: " + path);
+ }
+
+ // trailing '/' is not allowed (except for root path)
+ if (path.endsWith(FileSystem.SEPARATOR) && path.length() > 1) {
+ throw new FileSystemException("malformed path: " + path);
+ }
+
+ String[] names = path.split(FileSystem.SEPARATOR);
+ for (int i = 1; i < names.length; i++) {
+ // name must not be empty
+ if (names[i].length() == 0) {
+ throw new FileSystemException("empty name: " + path);
+ }
+ // leading/trailing whitespace is not allowed
+ String trimmed = names[i].trim();
+ if (!trimmed.equals(names[i])) {
+ throw new FileSystemException("illegal leading or trailing whitespace in name: " + path);
+ }
+ }
+ }
+
+ /**
+ * Returns the parent directory of the specified <code>path</code>.
+ *
+ * @param path a file system path denoting a directory or a file.
+ * @return the parent directory.
+ */
+ public static String getParentDir(String path) {
+ int pos = path.lastIndexOf(FileSystem.SEPARATOR_CHAR);
+ if (pos > 0) {
+ return path.substring(0, pos);
+ }
+ return FileSystem.SEPARATOR;
+ }
+
+ /**
+ * Returns the name of the specified <code>path</code>.
+ *
+ * @param path a file system path denoting a directory or a file.
+ * @return the name.
+ */
+ public static String getName(String path) {
+ int pos = path.lastIndexOf(FileSystem.SEPARATOR_CHAR);
+ if (pos != -1) {
+ return path.substring(pos + 1);
+ }
+ return path;
+ }
+
+}
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/fs/FileSystemResource.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/fs/FileSystemResource.java
new file mode 100644
index 0000000000..b429bf73fe
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/fs/FileSystemResource.java
@@ -0,0 +1,226 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.jackrabbit.core.fs;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.jackrabbit.core.fs.FileSystemPathUtil;
+
+/**
+ * A <code>FileSystemResource</code> represents a resource (i.e. file) in a
+ * <code>FileSystem</code>.
+ */
+public class FileSystemResource {
+
+ protected final FileSystem fs;
+
+ protected final String path;
+
+ static {
+ // preload FileSystemPathUtil to prevent classloader issues during shutdown
+ FileSystemPathUtil.class.hashCode();
+ }
+
+ /**
+ * Creates a new <code>FileSystemResource</code>
+ *
+ * @param fs the <code>FileSystem</code> where the resource is located
+ * @param path the path of the resource in the <code>FileSystem</code>
+ */
+ public FileSystemResource(FileSystem fs, String path) {
+ if (fs == null) {
+ throw new IllegalArgumentException("invalid file system argument");
+ }
+ this.fs = fs;
+
+ if (path == null) {
+ throw new IllegalArgumentException("invalid path argument");
+ }
+ this.path = path;
+ }
+
+ /**
+ * Returns the <code>FileSystem</code> where this resource is located.
+ *
+ * @return the <code>FileSystem</code> where this resource is located.
+ */
+ public FileSystem getFileSystem() {
+ return fs;
+ }
+
+ /**
+ * Returns the path of this resource.
+ *
+ * @return the path of this resource.
+ */
+ public String getPath() {
+ return path;
+ }
+
+ /**
+ * Returns the parent directory of this resource.
+ *
+ * @return the parent directory.
+ */
+ public String getParentDir() {
+ return FileSystemPathUtil.getParentDir(path);
+ }
+
+ /**
+ * Returns the name of this resource.
+ *
+ * @return the name.
+ */
+ public String getName() {
+ return FileSystemPathUtil.getName(path);
+ }
+
+ /**
+ * Creates the parent directory of this resource, including any necessary
+ * but nonexistent parent directories.
+ *
+ * @throws FileSystemException
+ */
+ public synchronized void makeParentDirs() throws FileSystemException {
+ String parentDir = getParentDir();
+ if (!fs.exists(parentDir)) {
+ fs.createFolder(parentDir);
+ }
+ }
+
+ /**
+ * Deletes this resource.
+ * Same as {@link #delete(boolean)} called with {@code false}.
+ *
+ * @see FileSystem#deleteFile
+ */
+ public void delete() throws FileSystemException {
+ delete(false);
+ }
+
+ /**
+ * Deletes this resource.
+ *
+ * @param pruneEmptyParentDirs if <code>true</code>, empty parent folders will
+ * automatically be deleted
+ * @see FileSystem#deleteFile
+ */
+ public synchronized void delete(boolean pruneEmptyParentDirs) throws FileSystemException {
+ fs.deleteFile(path);
+ if (pruneEmptyParentDirs) {
+ // prune empty parent folders
+ String parentDir = FileSystemPathUtil.getParentDir(path);
+ while (!parentDir.equals(FileSystem.SEPARATOR)
+ && fs.exists(parentDir)
+ && !fs.hasChildren(parentDir)) {
+ fs.deleteFolder(parentDir);
+ parentDir = FileSystemPathUtil.getParentDir(parentDir);
+ }
+ }
+ }
+
+ /**
+ * @see FileSystem#exists
+ */
+ public boolean exists() throws FileSystemException {
+ return fs.exists(path);
+ }
+
+ /**
+ * @see FileSystem#getInputStream
+ */
+ public InputStream getInputStream() throws FileSystemException {
+ return fs.getInputStream(path);
+ }
+
+ /**
+ * Spools this resource to the given output stream.
+ *
+ * @param out output stream where to spool the resource
+ * @throws FileSystemException if the input stream for this resource could
+ * not be obtained
+ * @throws IOException if an error occurs while while spooling
+ * @see FileSystem#getInputStream
+ */
+ public void spool(OutputStream out) throws FileSystemException, IOException {
+ InputStream in = fs.getInputStream(path);
+ try {
+ IOUtils.copy(in, out);
+ } finally {
+ IOUtils.closeQuietly(in);
+ }
+ }
+
+ /**
+ * @see FileSystem#getOutputStream
+ */
+ public OutputStream getOutputStream() throws FileSystemException {
+ return fs.getOutputStream(path);
+ }
+
+ /**
+ * @see FileSystem#lastModified
+ */
+ public long lastModified() throws FileSystemException {
+ return fs.lastModified(path);
+ }
+
+ /**
+ * @see FileSystem#length
+ */
+ public long length() throws FileSystemException {
+ return fs.length(path);
+ }
+
+ //-------------------------------------------< java.lang.Object overrides >
+ /**
+ * Returns the path string of this resource. This is just the
+ * string returned by the <code>{@link #getPath}</code> method.
+ *
+ * @return The path string of this resource
+ */
+ public String toString() {
+ return getPath();
+ }
+
+ public boolean equals(Object obj) {
+ if (this == obj) {
+ return true;
+ }
+ if (obj instanceof FileSystemResource) {
+ FileSystemResource other = (FileSystemResource) obj;
+ return (path == null ? other.path == null : path.equals(other.path))
+ && (fs == null ? other.fs == null : fs.equals(other.fs));
+ }
+ return false;
+ }
+
+ /**
+ * Returns zero to satisfy the Object equals/hashCode contract.
+ * This class is mutable and not meant to be used as a hash key.
+ *
+ * @return always zero
+ * @see Object#hashCode()
+ */
+ public int hashCode() {
+ return 0;
+ }
+
+}
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/fs/RandomAccessOutputStream.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/fs/RandomAccessOutputStream.java
new file mode 100644
index 0000000000..b540908671
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/fs/RandomAccessOutputStream.java
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.jackrabbit.core.fs;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
+/**
+ * Extends the regular <code>java.io.OutputStream</code> with a random
+ * access facility. Multiple <code>write()</code> operations can be
+ * positioned off sequence with the {@link #seek} method.
+ *
+ * @deprecated this class should no longer be used
+ */
+@Deprecated
+public abstract class RandomAccessOutputStream extends OutputStream {
+
+ /**
+ * Sets the current position in the resource where the next write
+ * will occur.
+ *
+ * @param position the new position in the resource.
+ * @throws IOException if an error occurs while seeking to the position.
+ */
+ public abstract void seek(long position) throws IOException;
+}
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/fs/local/FileUtil.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/fs/local/FileUtil.java
new file mode 100644
index 0000000000..5f9b830974
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/fs/local/FileUtil.java
@@ -0,0 +1,99 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.jackrabbit.core.fs.local;
+
+import java.io.File;
+import java.io.IOException;
+
+import org.apache.commons.io.FileUtils;
+
+/**
+ * Static utility methods for recursively copying and deleting files and
+ * directories.
+ */
+public final class FileUtil {
+
+ /**
+ * private constructor
+ */
+ private FileUtil() {
+ }
+
+ /**
+ * Recursively copies the given file or directory to the
+ * given destination.
+ *
+ * @param src source file or directory
+ * @param dest destination file or directory
+ * @throws IOException if the file or directory cannot be copied
+ */
+ public static void copy(File src, File dest) throws IOException {
+ if (!src.canRead()) {
+ throw new IOException(src.getPath() + " can't be read from.");
+ }
+ if (src.isDirectory()) {
+ // src is a folder
+ if (dest.isFile()) {
+ throw new IOException("can't copy a folder to a file");
+ }
+ if (!dest.exists()) {
+ dest.mkdirs();
+ }
+ if (!dest.canWrite()) {
+ throw new IOException("can't write to " + dest.getPath());
+ }
+ File[] children = src.listFiles();
+ for (int i = 0; i < children.length; i++) {
+ copy(children[i], new File(dest, children[i].getName()));
+ }
+ } else {
+ // src is a file
+ File destParent;
+ if (dest.isDirectory()) {
+ // dest is a folder
+ destParent = dest;
+ dest = new File(destParent, src.getName());
+ } else {
+ destParent = dest.getParentFile();
+ }
+ if (!destParent.canWrite()) {
+ throw new IOException("can't write to " + destParent.getPath());
+ }
+
+ FileUtils.copyFile(src, dest);
+ }
+ }
+
+ /**
+ * Recursively deletes the given file or directory.
+ *
+ * @param f file or directory
+ * @throws IOException if the file or directory cannot be deleted
+ */
+ public static void delete(File f) throws IOException {
+ if (f.isDirectory()) {
+ // it's a folder, list children first
+ File[] children = f.listFiles();
+ for (int i = 0; i < children.length; i++) {
+ delete(children[i]);
+ }
+ }
+ if (!f.delete()) {
+ throw new IOException("Unable to delete '" + f.getPath() + "'");
+ }
+ }
+}
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/fs/local/HandleMonitor.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/fs/local/HandleMonitor.java
new file mode 100644
index 0000000000..c6a6577ab8
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/fs/local/HandleMonitor.java
@@ -0,0 +1,217 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.jackrabbit.core.fs.local;
+
+import org.apache.jackrabbit.util.LazyFileInputStream;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.HashMap;
+import java.util.HashSet;
+
+/**
+ * This Class implements a very simple open handle monitor for the local
+ * file system. This is usefull, if the list of open handles, referenced by
+ * an open FileInputStream() should be tracked. This can cause problems on
+ * windows filesystems where open files cannot be deleted.
+ */
+public class HandleMonitor {
+
+ /**
+ * The default logger
+ */
+ private static Logger log = LoggerFactory.getLogger(HandleMonitor.class);
+
+ /**
+ * the map of open handles (key=File, value=Handle)
+ */
+ private HashMap<File, Handle> openHandles = new HashMap<File, Handle>();
+
+ /**
+ * Opens a file and returns an InputStream
+ *
+ * @param file
+ * @return
+ * @throws FileNotFoundException
+ */
+ public InputStream open(File file) throws FileNotFoundException {
+ Handle handle = getHandle(file);
+ InputStream in = handle.open();
+ return in;
+ }
+
+ /**
+ * Checks, if the file is open
+ * @param file
+ * @return
+ */
+ public boolean isOpen(File file) {
+ return openHandles.containsKey(file);
+ }
+
+ /**
+ * Closes a file
+ * @param file
+ */
+ private void close(File file) {
+ openHandles.remove(file);
+ }
+
+ /**
+ * Returns the handle for a file.
+ * @param file
+ * @return
+ */
+ private Handle getHandle(File file) {
+ Handle handle = openHandles.get(file);
+ if (handle == null) {
+ handle = new Handle(file);
+ openHandles.put(file, handle);
+ }
+ return handle;
+ }
+
+ /**
+ * Dumps the contents of this monitor
+ */
+ public void dump() {
+ log.info("Number of open files: " + openHandles.size());
+ for (File file : openHandles.keySet()) {
+ Handle handle = openHandles.get(file);
+ handle.dump();
+ }
+ }
+
+ /**
+ * Dumps the information for a file
+ * @param file
+ */
+ public void dump(File file) {
+ Handle handle = openHandles.get(file);
+ if (handle != null) {
+ handle.dump(true);
+ }
+ }
+
+ /**
+ * Class representing all open handles to a file
+ */
+ private class Handle {
+
+ /**
+ * the file of this handle
+ */
+ private File file;
+
+ /**
+ * all open streams of this handle
+ */
+ private HashSet<Handle.MonitoredInputStream> streams = new HashSet<Handle.MonitoredInputStream>();
+
+ /**
+ * Creates a new handle for a file
+ * @param file
+ */
+ private Handle(File file) {
+ this.file = file;
+ }
+
+ /**
+ * opens a stream for this handle
+ * @return
+ * @throws FileNotFoundException
+ */
+ private InputStream open() throws FileNotFoundException {
+ Handle.MonitoredInputStream in = new Handle.MonitoredInputStream(file);
+ streams.add(in);
+ return in;
+ }
+
+ /**
+ * Closes a stream
+ * @param in
+ */
+ private void close(MonitoredInputStream in) {
+ streams.remove(in);
+ if (streams.isEmpty()) {
+ HandleMonitor.this.close(file);
+ }
+ }
+
+ /**
+ * Dumps this handle
+ */
+ private void dump() {
+ dump(false);
+ }
+
+ /**
+ * Dumps this handle
+ */
+ private void dump(boolean detailed) {
+ if (detailed) {
+ log.info("- " + file.getPath() + ", " + streams.size());
+ for (Handle.MonitoredInputStream in : streams) {
+ in.dump();
+ }
+ } else {
+ log.info("- " + file.getPath() + ", " + streams.size());
+ }
+ }
+
+ /**
+ * Delegating input stream that registers/unregisters itself from the
+ * handle.
+ */
+ private class MonitoredInputStream extends LazyFileInputStream {
+
+ /**
+ * throwable of the time, the stream was created
+ */
+ private final Throwable throwable = new Exception();
+
+ /**
+ * {@inheritDoc}
+ */
+ private MonitoredInputStream(File file) throws FileNotFoundException {
+ super(file);
+ }
+
+ /**
+ * dumps this stream
+ */
+ private void dump() {
+ log.info("- opened by : ", throwable);
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public void close() throws IOException {
+ // remove myself from the set
+ Handle.this.close(this);
+ super.close();
+ }
+
+ }
+ }
+
+}
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/fs/local/LocalFileSystem.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/fs/local/LocalFileSystem.java
new file mode 100644
index 0000000000..d5d32aeb75
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/fs/local/LocalFileSystem.java
@@ -0,0 +1,388 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.jackrabbit.core.fs.local;
+
+import org.apache.jackrabbit.core.fs.FileSystem;
+import org.apache.jackrabbit.core.fs.FileSystemException;
+import org.apache.jackrabbit.core.fs.local.FileUtil;
+import org.apache.jackrabbit.core.fs.local.HandleMonitor;
+import org.apache.jackrabbit.util.LazyFileInputStream;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.FileFilter;
+import java.io.FileNotFoundException;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+
+/**
+ * A <code>LocalFileSystem</code> ...
+ */
+public class LocalFileSystem implements FileSystem {
+
+ private static Logger log = LoggerFactory.getLogger(LocalFileSystem.class);
+
+ private File root;
+
+ private HandleMonitor monitor;
+
+ /**
+ * Default constructor
+ */
+ public LocalFileSystem() {
+ }
+
+ public String getPath() {
+ if (root != null) {
+ return root.getPath();
+ } else {
+ return null;
+ }
+ }
+
+ /**
+ * Sets the path to the root directory of this local filesystem. please note
+ * that this method can be called via reflection during initialization and
+ * must not be altered.
+ *
+ * @param rootPath the path to the root directory
+ */
+ public void setPath(String rootPath) {
+ setRoot(new File(osPath(rootPath)));
+ }
+
+ public void setRoot(File root) {
+ this.root = root;
+ }
+
+ /**
+ * Enables/Disables the use of the handle monitor.
+ *
+ * @param enable
+ */
+ public void setEnableHandleMonitor(String enable) {
+ setEnableHandleMonitor(Boolean.valueOf(enable).booleanValue());
+ }
+
+ /**
+ * Enables/Disables the use of the handle monitor.
+ *
+ * @param enable flag
+ */
+ public void setEnableHandleMonitor(boolean enable) {
+ if (enable && monitor == null) {
+ monitor = new HandleMonitor();
+ }
+ if (!enable && monitor != null) {
+ monitor = null;
+ }
+ }
+
+ /**
+ * Returns <code>true</code> if use of the handle monitor is currently
+ * enabled, otherwise returns <code>false</code>.
+ *
+ * @see #setEnableHandleMonitor(boolean)
+ */
+ public String getEnableHandleMonitor() {
+ return monitor == null ? "false" : "true";
+ }
+
+ private String osPath(String genericPath) {
+ if (File.separator.equals(SEPARATOR)) {
+ return genericPath;
+ }
+ return genericPath.replace(SEPARATOR_CHAR, File.separatorChar);
+ }
+
+ //-------------------------------------------< java.lang.Object overrides >
+ public boolean equals(Object obj) {
+ if (this == obj) {
+ return true;
+ }
+ if (obj instanceof LocalFileSystem) {
+ LocalFileSystem other = (LocalFileSystem) obj;
+ if (root == null) {
+ return other.root == null;
+ } else {
+ return root.equals(other.root);
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Returns zero to satisfy the Object equals/hashCode contract.
+ * This class is mutable and not meant to be used as a hash key.
+ *
+ * @return always zero
+ * @see Object#hashCode()
+ */
+ public int hashCode() {
+ return 0;
+ }
+
+ //-----------------------------------------------------------< FileSystem >
+ /**
+ * {@inheritDoc}
+ */
+ public void init() throws FileSystemException {
+ if (root == null) {
+ String msg = "root directory not set";
+ log.debug(msg);
+ throw new FileSystemException(msg);
+ }
+
+ if (root.exists()) {
+ if (!root.isDirectory()) {
+ String msg = "path does not denote a folder";
+ log.debug(msg);
+ throw new FileSystemException(msg);
+ }
+ } else {
+ if (!root.mkdirs()) {
+ String msg = "failed to create root";
+ log.debug(msg);
+ throw new FileSystemException(msg);
+ }
+ }
+ log.info("LocalFileSystem initialized at path " + root.getPath());
+ if (monitor != null) {
+ log.info("LocalFileSystem using handle monitor");
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public void close() throws FileSystemException {
+ root = null;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public void createFolder(String folderPath) throws FileSystemException {
+ File f = new File(root, osPath(folderPath));
+ if (f.exists()) {
+ String msg = f.getPath() + " already exists";
+ log.debug(msg);
+ throw new FileSystemException(msg);
+ }
+ if (!f.mkdirs()) {
+ String msg = "failed to create folder " + f.getPath();
+ log.debug(msg);
+ throw new FileSystemException(msg);
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public void deleteFile(String filePath) throws FileSystemException {
+ File f = new File(root, osPath(filePath));
+ if (!f.isFile()) {
+ String msg = f.getPath() + " does not denote an existing file";
+ throw new FileSystemException(msg);
+ }
+ try {
+ FileUtil.delete(f);
+ } catch (IOException ioe) {
+ String msg = "failed to delete " + f.getPath();
+ if (monitor != null && monitor.isOpen(f)) {
+ log.error("Unable to delete. There are still open streams.");
+ monitor.dump(f);
+ }
+
+ throw new FileSystemException(msg, ioe);
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public void deleteFolder(String folderPath) throws FileSystemException {
+ File f = new File(root, osPath(folderPath));
+ if (!f.isDirectory()) {
+ String msg = f.getPath() + " does not denote an existing folder";
+ log.debug(msg);
+ throw new FileSystemException(msg);
+ }
+ try {
+ FileUtil.delete(f);
+ } catch (IOException ioe) {
+ String msg = "failed to delete " + f.getPath();
+ log.debug(msg);
+ throw new FileSystemException(msg, ioe);
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public boolean exists(String path) throws FileSystemException {
+ File f = new File(root, osPath(path));
+ return f.exists();
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public InputStream getInputStream(String filePath)
+ throws FileSystemException {
+ File f = new File(root, osPath(filePath));
+ try {
+ if (monitor == null) {
+ return new LazyFileInputStream(f);
+ } else {
+ return monitor.open(f);
+ }
+ } catch (FileNotFoundException fnfe) {
+ String msg = f.getPath() + " does not denote an existing file";
+ log.debug(msg);
+ throw new FileSystemException(msg, fnfe);
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public OutputStream getOutputStream(String filePath)
+ throws FileSystemException {
+ File f = new File(root, osPath(filePath));
+ try {
+ return new FileOutputStream(f);
+ } catch (FileNotFoundException fnfe) {
+ String msg = "failed to get output stream for " + f.getPath();
+ log.debug(msg);
+ throw new FileSystemException(msg, fnfe);
+ }
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public boolean hasChildren(String path) throws FileSystemException {
+ File f = new File(root, osPath(path));
+ if (!f.exists()) {
+ String msg = f.getPath() + " does not exist";
+ log.debug(msg);
+ throw new FileSystemException(msg);
+ }
+ if (f.isFile()) {
+ return false;
+ }
+ return (f.list().length > 0);
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public boolean isFile(String path) throws FileSystemException {
+ File f = new File(root, osPath(path));
+ return f.isFile();
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public boolean isFolder(String path) throws FileSystemException {
+ File f = new File(root, osPath(path));
+ return f.isDirectory();
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public long lastModified(String path) throws FileSystemException {
+ File f = new File(root, osPath(path));
+ return f.lastModified();
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public long length(String filePath) throws FileSystemException {
+ File f = new File(root, osPath(filePath));
+ if (!f.exists()) {
+ return -1;
+ }
+ return f.length();
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public String[] list(String folderPath) throws FileSystemException {
+ File f = new File(root, osPath(folderPath));
+ String[] entries = f.list();
+ if (entries == null) {
+ String msg = folderPath + " does not denote a folder";
+ log.debug(msg);
+ throw new FileSystemException(msg);
+ }
+ return entries;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public String[] listFiles(String folderPath) throws FileSystemException {
+ File folder = new File(root, osPath(folderPath));
+ File[] files = folder.listFiles(new FileFilter() {
+ public boolean accept(File f) {
+ return f.isFile();
+ }
+ });
+ if (files == null) {
+ String msg = folderPath + " does not denote a folder";
+ log.debug(msg);
+ throw new FileSystemException(msg);
+ }
+ String[] entries = new String[files.length];
+ for (int i = 0; i < files.length; i++) {
+ entries[i] = files[i].getName();
+ }
+ return entries;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ public String[] listFolders(String folderPath) throws FileSystemException {
+ File file = new File(root, osPath(folderPath));
+ File[] folders = file.listFiles(new FileFilter() {
+ public boolean accept(File f) {
+ return f.isDirectory();
+ }
+ });
+ if (folders == null) {
+ String msg = folderPath + " does not denote a folder";
+ log.debug(msg);
+ throw new FileSystemException(msg);
+ }
+ String[] entries = new String[folders.length];
+ for (int i = 0; i < folders.length; i++) {
+ entries[i] = folders[i].getName();
+ }
+ return entries;
+ }
+
+}
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/fs/local/package-info.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/fs/local/package-info.java
new file mode 100755
index 0000000000..ba4a9b106c
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/fs/local/package-info.java
@@ -0,0 +1,19 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+/* see JCR-4060 */
+@org.osgi.annotation.versioning.Version("2.13.5")
+package org.apache.jackrabbit.core.fs.local;
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/fs/package-info.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/fs/package-info.java
new file mode 100755
index 0000000000..15a8337e03
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/fs/package-info.java
@@ -0,0 +1,19 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+/* see JCR-4060 */
+@org.osgi.annotation.versioning.Version("2.13.5")
+package org.apache.jackrabbit.core.fs;
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/util/db/CheckSchemaOperation.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/util/db/CheckSchemaOperation.java
new file mode 100644
index 0000000000..ab69ec6116
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/util/db/CheckSchemaOperation.java
@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.jackrabbit.core.util.db;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.sql.SQLException;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.jackrabbit.util.Text;
+
+/**
+ * An operation which synchronously checks the DB schema in the {@link #run()} method. The
+ * {@link #addVariableReplacement(String, String)} method return the instance to enable method chaining.
+ */
+public class CheckSchemaOperation {
+
+ public static final String SCHEMA_OBJECT_PREFIX_VARIABLE = "${schemaObjectPrefix}";
+
+ public static final String TABLE_SPACE_VARIABLE = "${tableSpace}";
+
+ private final ConnectionHelper conHelper;
+
+ private final InputStream ddl;
+
+ private final String table;
+
+ private final Map<String, String> varReplacement = new HashMap<String, String>();
+
+ /**
+ * @param connectionhelper the connection helper
+ * @param ddlStream the stream of the DDL to use to create the schema if necessary (closed by the
+ * {@link #run()} method)
+ * @param tableName the name of the table to use for the schema-existence-check
+ */
+ public CheckSchemaOperation(ConnectionHelper connectionhelper, InputStream ddlStream, String tableName) {
+ conHelper = connectionhelper;
+ ddl = ddlStream;
+ table = tableName;
+ }
+
+ /**
+ * Adds a variable replacement mapping.
+ *
+ * @param var the variable
+ * @param replacement the replacement value
+ * @return this
+ */
+ public CheckSchemaOperation addVariableReplacement(String var, String replacement) {
+ varReplacement.put(var, replacement);
+ return this;
+ }
+
+ /**
+ * Checks if the required schema objects exist and creates them if they don't exist yet.
+ *
+ * @throws SQLException if an error occurs
+ * @throws IOException if an error occurs
+ */
+ public void run() throws SQLException, IOException {
+ try {
+ if (!conHelper.tableExists(table)) {
+ BufferedReader reader = new BufferedReader(new InputStreamReader(ddl));
+ String sql = reader.readLine();
+ while (sql != null) {
+ // Skip comments and empty lines
+ if (!sql.startsWith("#") && sql.length() > 0) {
+ // replace prefix variable
+ sql = replace(sql);
+ // execute sql stmt
+ conHelper.exec(sql);
+ }
+ // read next sql stmt
+ sql = reader.readLine();
+ }
+ }
+ } finally {
+ IOUtils.closeQuietly(ddl);
+ }
+ }
+
+ /**
+ * Applies the variable replacement to the given string.
+ *
+ * @param sql the string in which to replace variables
+ * @return the new string
+ */
+ private String replace(String sql) {
+ String result = sql;
+ for (Map.Entry<String, String> entry : varReplacement.entrySet()) {
+ result = Text.replace(result, entry.getKey(), entry.getValue()).trim();
+ }
+ return result;
+ }
+}
diff --git a/oak-blob/src/main/java/org/apache/jackrabbit/core/util/db/ConnectionFactory.java b/oak-blob/src/main/java/org/apache/jackrabbit/core/util/db/ConnectionFactory.java
new file mode 100644
index 0000000000..fe17311175
--- /dev/null
+++ b/oak-blob/src/main/java/org/apache/jackrabbit/core/util/db/ConnectionFactory.java
@@ -0,0 +1,382 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.jackrabbit.core.util.db;
+
+import java.lang.reflect.InvocationTargetException;
+import java.sql.Connection;
+import java.sql.Driver;
+import java.sql.SQLException;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import javax.jcr.RepositoryException;
+import javax.naming.Context;
+import javax.naming.NamingException;
+import javax.sql.DataSource;
+
+import org.apache.commons.dbcp2.BasicDataSource;
... 2402 lines suppressed ...