You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@sis.apache.org by am...@apache.org on 2020/09/28 14:25:54 UTC
[sis] branch refactor/strict_storage_connector created (now dbeef90)
This is an automated email from the ASF dual-hosted git repository.
amanin pushed a change to branch refactor/strict_storage_connector
in repository https://gitbox.apache.org/repos/asf/sis.git.
at dbeef90 feat(Storage): Add an extension of storage connector with fail-fast behaviors
This branch includes the following new commits:
new dbeef90 feat(Storage): Add an extension of storage connector with fail-fast behaviors
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[sis] 01/01: feat(Storage): Add an extension of storage connector
with fail-fast behaviors
Posted by am...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
amanin pushed a commit to branch refactor/strict_storage_connector
in repository https://gitbox.apache.org/repos/asf/sis.git
commit dbeef9069be6fe6209969857c0fbc756b11a101e
Author: Alexis Manin <am...@apache.org>
AuthorDate: Wed Apr 29 20:44:09 2020 +0200
feat(Storage): Add an extension of storage connector with fail-fast behaviors
---
.../apache/sis/storage/StrictStorageConnector.java | 238 +++++++++++++++++++++
.../apache/sis/storage/StorageConnectorTest.java | 68 +++++-
.../sis/storage/StrictStorageConnectorTest.java | 177 +++++++++++++++
.../apache/sis/test/suite/StorageTestSuite.java | 1 +
4 files changed, 475 insertions(+), 9 deletions(-)
diff --git a/storage/sis-storage/src/main/java/org/apache/sis/storage/StrictStorageConnector.java b/storage/sis-storage/src/main/java/org/apache/sis/storage/StrictStorageConnector.java
new file mode 100644
index 0000000..ea9975c
--- /dev/null
+++ b/storage/sis-storage/src/main/java/org/apache/sis/storage/StrictStorageConnector.java
@@ -0,0 +1,238 @@
+package org.apache.sis.storage;
+
+import java.io.Closeable;
+import java.io.File;
+import java.io.IOException;
+import java.net.URI;
+import java.nio.ByteBuffer;
+import java.nio.file.Path;
+import java.util.Optional;
+import java.util.concurrent.Callable;
+import javax.imageio.stream.ImageInputStream;
+import javax.sql.DataSource;
+import org.apache.sis.util.UnconvertibleObjectException;
+import org.apache.sis.util.collection.BackingStoreException;
+
+/**
+ * Extension of a storage connector providing strong encapsulation of "views". This allows to:
+ * <ul>
+ * <li>Adopt a <em>fail-fast</em> behavior in case storage view is corrupted by a user</li>
+ * <li>
+ * Provide easier usage:
+ * <ul>
+ * <li>Initial mark/final rewind is performed internally, user do not need to care about it.</li>
+ * <li>Provide strongly typed operators, to guide user on how to use this object.</li>
+ * </ul>
+ * </li>
+ * </ul>
+ * The purpose of this class is to be merged in StorageConnector once its principle has been validated.
+ *
+ * <em>Guarantees</em>:
+ * <ul>
+ * <li>This object is <em>not</em> concurrent, and ensure a <em>fail-fast</em> behavior in such cases.</li>
+ * <li>
+ * useAs* methods will enforce following behavior:
+ * <ul>
+ * <li>If possible, rewind properly consumed storage view to its initial state</li>
+ * <li>If above statement is not possible, an error will be immediately propagated, and the connector will be marked as closed.</li>
+ * </ul>
+ * </li>
+ * </ul>
+ *
+ * Typical usage:
+ * <ol>
+ * <li>Check storage compatibility through `useAs*` methods</li>
+ * <li>If the storage is compatible, commit our choice by locking a storage view, closing the connector in the process.</li>
+ * </ol>
+ *
+ * Example:
+ * <pre>
+ * final Path file = Paths.get("path/to/file");
+ * try (var c = new StrictStorageConnector(new StorageConnector(file)) {
+ *
+ * // Use connector automatically reset buffering to check support
+ * Boolean isSupported = c.useAsBuffer((buffer) -%gt; buffer.get() == SEARCHED_KEY);
+ *
+ * // Once support is validated, acquire real storage connection. At this point,
+ * // storage life cycle becomes the responsability of the caller, allowing it
+ * // to survive beyond the connector scope.
+ * if (supported) {
+ * try ( InputStream stream = c.commit( InputStream.class ) ) {
+ * // read all needed data from acquired stream
+ * }
+ * }
+ * }
+ * </pre>
+ */
+public class StrictStorageConnector implements AutoCloseable {
+
+ private final StorageConnector storage;
+
+ private Object committedStorage;
+ private volatile int concurrentFlag;
+
+ public StrictStorageConnector(StorageConnector storage) {
+ this.storage = storage;
+ }
+
+ public void closeAllExcept(Object view) throws DataStoreException {
+ // Closing multiple times is OK. However, if the view is not null, we will let control raise an error.
+ if (concurrentFlag < 0 && view == null) return;
+ try {
+ doUnderControl(() -> {
+ concurrentFlag = -1;
+ storage.closeAllExcept(view);
+ committedStorage = view;
+ return null;
+ });
+ } catch (IOException e) {
+ throw new DataStoreException(e);
+ }
+ }
+
+ /**
+ * Provides an in-memory byte buffer containing first bytes of the source storage.
+ * To know how many bytes are available, refer to the buffer {@link ByteBuffer#remaining() remaining byte count}.
+ * User <em>do not</em> need to rewind buffer after use. It is the storage connector responsability.
+ *
+ * @param operator User operation to perform against preovided buffer.
+ * @param <T>
+ * @return
+ * @throws DataStoreException
+ * @throws IOException
+ */
+ public <T> T useAsBuffer(StorageOperatingFunction<ByteBuffer, T> operator) throws DataStoreException, IOException {
+ return doUnderControl(() -> {
+ final ByteBuffer buffer = getOrFail(ByteBuffer.class);
+ try ( Closeable rewindOnceDone = buffer::rewind ) {
+ return operator.apply(buffer);
+ }
+ });
+ }
+
+ public <T> T useAsImageInputStream(StorageOperatingFunction<ImageInputStream, T> operator) throws IOException, DataStoreException {
+ return doUnderControl(() -> {
+ ImageInputStream stream = getOrFail(ImageInputStream.class);
+ final long positionCtrl = stream.getStreamPosition();
+ stream.mark();
+ T result;
+ try ( Closeable rewindOnceDone = stream::reset ) {
+ result = operator.apply(stream);
+ }
+ if (stream.getStreamPosition() != positionCtrl) {
+ concurrentFlag = -1; // mark this connector as closed/not valid anymore
+ throw new DataStoreException("Operator has messed with stream marks");
+ }
+ return result;
+ });
+ }
+
+ /**
+ * Temporarily expose storage through queried interface/class to be used by a user defined operator.
+ * Note that provided storage is checked after use, to ensure it has not been corrupted by input operator.
+ *
+ * @param storageType Storage access interface to provide to the operator.
+ * @param operator The operator that will access storage to compute a result.
+ * @param <S> Storage class
+ * @param <T> Type of result computed by user operator.
+ * @return The value computed by user operator.
+ * @throws IOException If given operator throws IOException on execution.
+ * @throws UnsupportedStorageException If queried storage type cannot be accessed in current context.
+ * @throws DataStoreException If an error occurs while fetching queried storage.
+ */
+ public <S, T> T useAs(Class<T> storageType, StorageOperatingFunction<S, T> operator) throws IOException, DataStoreException {
+ if (ByteBuffer.class.isAssignableFrom(storageType)) return useAsBuffer((StorageOperatingFunction<ByteBuffer, T>) operator);
+ else if (ImageInputStream.class.isAssignableFrom(storageType)) return useAsImageInputStream((StorageOperatingFunction<ImageInputStream, T>) operator);
+ else if (URI.class.isAssignableFrom(storageType)) return ((StorageOperatingFunction<URI, T>) operator).apply(getURI().orElseThrow(() -> new UnsupportedStorageException("Cannot acquire an URI")));
+ else if (Path.class.isAssignableFrom(storageType)) return ((StorageOperatingFunction<Path, T>) operator).apply(getPath().orElseThrow(() -> new UnsupportedStorageException("Cannot acquire a path")));
+ else if (File.class.isAssignableFrom(storageType)) return ((StorageOperatingFunction<File, T>) operator).apply(getPath().map(p -> p.toFile()).orElseThrow(() -> new UnsupportedStorageException("Cannot acquire a file")));
+ else throw new UnsupportedStorageException("Queried storage type is not supported yet: "+storageType);
+ }
+
+ /**
+ * Ensure only one storage operation is running at any time against this storage connector. It allows fail-fast
+ * behavior if this connector is used in concurrent context.
+ *
+ * @param operator The operation to perform once we checked no other operation is running.
+ *
+ * @param <V> Type of result value produced by given operator.
+ * @return The result produced by given operator.
+ * @throws IOException If anything wrong happens while input operator consumes storage, or we can mark/rewind storage.
+ * @throws DataStoreException Same reasons as for IOException + can happen if queried storage is of unsupported type.
+ * @throws IllegalStateException If this connector is already closed.
+ */
+ protected <V> V doUnderControl(StorageCallable<V> operator) throws IOException, DataStoreException {
+ if (concurrentFlag < 0) throw new IllegalStateException("...");
+ if (concurrentFlag != 0) throw new ConcurrentReadException("...");
+ concurrentFlag++;
+ try {
+ return operator.call();
+ } finally {
+ concurrentFlag--;
+ }
+ }
+
+ public Optional<Path> getPath() { return getSilently(Path.class); }
+
+ public Optional<URI> getURI() { return getSilently(URI.class); }
+
+ public Optional<DataSource> getSQLDatasource() { return getSilently(DataSource.class); }
+
+ public Optional<String> getPathAsString() { return getSilently(String.class); }
+
+ /**
+ * Retrieve storage in the queried form, closing all other opened view in the same time.
+ * <em>Warning</em>: This method also closes this storage connector, making invalid any more calls on it.
+ *
+ * @param target Type of the view to get back / keep opened.
+ * @param <T> Type of the wanted storage connection.
+ * @return Underlying storage in the requested form. Never null.
+ *
+ * @throws IOException If anything goes wrong while initializing storage access.
+ * @throws DataStoreException If this connector is used concurrently, or if any problem occurs while initializing view.
+ * @throws IllegalStateException If this connector is already closed.
+ */
+ public <T> T commit(Class<T> target) throws IOException, DataStoreException {
+ return doUnderControl(() -> {
+ final T result = getOrFail(target);
+ concurrentFlag = -1; //close flag
+ storage.closeAllExcept(result);
+ return result;
+ });
+ }
+
+ private <T> T getOrFail(Class<T> target) throws DataStoreException {
+ T view = storage.getStorageAs(target);
+ if (view == null) throw new UnsupportedStorageException();
+ return view;
+ }
+
+ private <T> Optional<T> getSilently(Class<T> target) {
+ try {
+ return Optional.ofNullable(storage.getStorageAs(target));
+ } catch (UnconvertibleObjectException e) {
+ // TODO: log fine
+ return Optional.empty();
+ } catch (DataStoreException e) {
+ // According to current implementation, that should never happen.
+ // Moreover, it is not really logic to propagate DataStoreException, as this operation should not involve
+ // any "storage" logic (only in-memory path/uri conversion if needed).
+ throw new BackingStoreException(e);
+ }
+ }
+
+ @Override
+ public void close() throws IOException, DataStoreException {
+ storage.closeAllExcept(committedStorage);
+ }
+
+ private interface StorageCallable<V> extends Callable<V> {
+ @Override
+ V call() throws IOException, DataStoreException;
+ }
+
+ @FunctionalInterface
+ public interface StorageOperatingFunction<I, O> {
+ O apply(I storage) throws IOException, DataStoreException;
+ }
+}
diff --git a/storage/sis-storage/src/test/java/org/apache/sis/storage/StorageConnectorTest.java b/storage/sis-storage/src/test/java/org/apache/sis/storage/StorageConnectorTest.java
index a7d33ea..381bec8 100644
--- a/storage/sis-storage/src/test/java/org/apache/sis/storage/StorageConnectorTest.java
+++ b/storage/sis-storage/src/test/java/org/apache/sis/storage/StorageConnectorTest.java
@@ -16,29 +16,40 @@
*/
package org.apache.sis.storage;
-import java.net.URI;
+import java.io.ByteArrayInputStream;
import java.io.DataInput;
+import java.io.IOException;
import java.io.InputStream;
import java.io.Reader;
-import java.io.IOException;
+import java.net.URI;
import java.nio.ByteBuffer;
-import java.nio.charset.StandardCharsets;
import java.nio.channels.ReadableByteChannel;
-import javax.imageio.stream.ImageInputStream;
-import javax.imageio.ImageIO;
+import java.nio.charset.StandardCharsets;
import java.sql.Connection;
-import org.apache.sis.setup.OptionKey;
-import org.apache.sis.util.UnconvertibleObjectException;
+import java.util.Random;
+import javax.imageio.ImageIO;
+import javax.imageio.stream.ImageInputStream;
import org.apache.sis.internal.storage.io.ChannelDataInput;
import org.apache.sis.internal.storage.io.ChannelImageInputStream;
import org.apache.sis.internal.storage.io.InputStreamAdapter;
-import org.apache.sis.test.DependsOnMethod;
+import org.apache.sis.setup.OptionKey;
import org.apache.sis.test.DependsOn;
+import org.apache.sis.test.DependsOnMethod;
import org.apache.sis.test.TestCase;
+import org.apache.sis.util.UnconvertibleObjectException;
import org.junit.Test;
import static org.junit.Assume.assumeTrue;
-import static org.opengis.test.Assert.*;
+import static org.opengis.test.Assert.assertArrayEquals;
+import static org.opengis.test.Assert.assertEquals;
+import static org.opengis.test.Assert.assertFalse;
+import static org.opengis.test.Assert.assertInstanceOf;
+import static org.opengis.test.Assert.assertNotNull;
+import static org.opengis.test.Assert.assertNotSame;
+import static org.opengis.test.Assert.assertNull;
+import static org.opengis.test.Assert.assertSame;
+import static org.opengis.test.Assert.assertTrue;
+import static org.opengis.test.Assert.fail;
/**
@@ -390,4 +401,43 @@ public final strictfp class StorageConnectorTest extends TestCase {
assertTrue("channel.isOpen()", channel.isOpen());
channel.close();
}
+
+ @Test
+ public void getting_buffer_should_not_change_underlying_stream_position() throws Exception {
+ final StorageConnector con = create(false);
+
+ final ImageInputStream stream = con.getStorageAs(ImageInputStream.class);
+
+ final ByteBuffer buffer = con.getStorageAs(ByteBuffer.class);
+ assertEquals(0, stream.getStreamPosition());
+
+ buffer.get();
+ assertEquals(0, stream.getStreamPosition());
+
+ con.closeAllExcept(null);
+ }
+
+ @Test
+ public void moving_stream_should_not_impact_buffer() throws Exception {
+ final byte[] data = new byte[(int) Math.pow(2, 16)];
+ new Random().nextBytes(data);
+ final StorageConnector connector = new StorageConnector(new ByteArrayInputStream(data));
+ final ByteBuffer buffer = connector.getStorageAs(ByteBuffer.class);
+ final int blockSize = buffer.remaining();
+ final byte[] ctrl = new byte[blockSize];
+ buffer.get(ctrl).rewind();
+
+ InputStream stream = connector.getStorageAs(InputStream.class);
+ stream.mark(blockSize *2);
+ stream.skip(blockSize);
+ stream.read(new byte[blockSize]);
+ stream.reset();
+
+ final byte[] afterValue = new byte[blockSize];
+ buffer.get(afterValue).rewind();
+
+ assertArrayEquals(ctrl, afterValue);
+
+ connector.closeAllExcept(null);
+ }
}
diff --git a/storage/sis-storage/src/test/java/org/apache/sis/storage/StrictStorageConnectorTest.java b/storage/sis-storage/src/test/java/org/apache/sis/storage/StrictStorageConnectorTest.java
new file mode 100644
index 0000000..40110a4
--- /dev/null
+++ b/storage/sis-storage/src/test/java/org/apache/sis/storage/StrictStorageConnectorTest.java
@@ -0,0 +1,177 @@
+package org.apache.sis.storage;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.URISyntaxException;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import org.apache.sis.setup.OptionKey;
+import org.apache.sis.test.DependsOn;
+import org.junit.Test;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+@DependsOn(org.apache.sis.storage.StorageConnectorTest.class)
+public class StrictStorageConnectorTest {
+ /**
+ * Name of the test file, in the same directory than this {@code StorageConnectorTest} file.
+ */
+ private static final String FILENAME = "Any.txt";
+
+ /**
+ * Creates the instance to test. This method uses the {@code "test.txt"} ASCII file as
+ * the resource to test. The resource can be provided either as a URL or as a stream.
+ */
+ private static StrictStorageConnector create(final boolean asStream) {
+ final Class<?> c = StorageConnectorTest.class;
+ final Object storage = asStream ? c.getResourceAsStream(FILENAME) : c.getResource(FILENAME);
+ assertNotNull(storage);
+ final StorageConnector unsafeConnector = new StorageConnector(storage);
+ unsafeConnector.setOption(OptionKey.ENCODING, StandardCharsets.US_ASCII);
+ unsafeConnector.setOption(OptionKey.URL_ENCODING, "UTF-8");
+ final StrictStorageConnector connector = new StrictStorageConnector(unsafeConnector);
+ return connector;
+ }
+
+ private static byte[] getFileBytes() throws URISyntaxException, IOException {
+ final Path filePath = Paths.get(StorageConnector.class.getResource(FILENAME).toURI());
+ return Files.readAllBytes(filePath);
+ }
+
+ @Test
+ public void acquiring_path_works() {
+ final StrictStorageConnector connector = create(false);
+ assertTrue(connector.getPath().isPresent());
+ assertTrue(connector.getURI().isPresent());
+ assertTrue(connector.getPathAsString().isPresent());
+ assertFalse(connector.getSQLDatasource().isPresent());
+ }
+
+ @Test
+ public void stream_based_connector_return_empty_path() {
+ final StrictStorageConnector connector = create(true);
+ assertFalse(connector.getPath().isPresent());
+ assertFalse(connector.getURI().isPresent());
+ assertFalse(connector.getPathAsString().isPresent());
+ assertFalse(connector.getSQLDatasource().isPresent());
+ }
+
+ @Test
+ public void byte_buffer_is_rewind_after_use() throws Exception {
+ final byte[] ctrl = getFileBytes();
+ try (final StrictStorageConnector connector = create(false)) {
+ // Mess with internal buffer
+ connector.useAsBuffer(buffer -> {
+ // mess with it
+ return buffer.get(new byte[10]);
+ });
+ // ensure it has been properly rewind
+ connector.useAsBuffer(buffer -> {
+ assertEquals(0, buffer.position());
+ byte[] readValue = new byte[buffer.remaining()];
+ buffer.get(readValue);
+ assertArrayEquals(ctrl, readValue);
+ return null;
+ });
+ }
+ }
+
+ @Test
+ public void fail_fast_when_user_corrupts_stream_mark() throws IOException, DataStoreException {
+ try (final StrictStorageConnector c = create(false)) {
+ try {
+ c.useAsImageInputStream(stream -> {
+ stream.skipBytes(1);
+ stream.mark();
+ return 0;
+ });
+ fail("We should have detected something has gone wrong");
+ } catch (DataStoreException e) {
+ // Expected behavior: connector has detected that rewind did not work properly.
+ }
+ }
+ }
+
+ @Test
+ public void no_concurrency_allowed() throws Exception {
+ try (final StrictStorageConnector c = create(false)) {
+ synchronized (c) {
+ new Thread(() -> {
+ try {
+ c.useAsBuffer(buffer -> {
+ synchronized (c) {
+ try {
+ c.notifyAll();
+ c.wait(1000);
+ } catch (InterruptedException e) {
+ // Do not matter here.
+ }
+ }
+ return null;
+ });
+
+ } catch (IOException | DataStoreException e) {
+ // Do not matter here.
+ }
+ }).start();
+
+ // Ensure above operation is
+ c.wait(100);
+ }
+
+ try {
+ c.useAsBuffer(buffer -> null);
+ fail("Concurrency error should have been raised.");
+ } catch (ConcurrentReadException e) {
+ // Expected behavior: fail-fast to prevent concurrency.
+ }
+ synchronized (c) {
+ c.notifyAll();
+ }
+ }
+ }
+
+ @Test
+ public void commit_close_all_resources_but_chosen() throws Exception {
+ final InputStream is;
+ try (final StrictStorageConnector c = create(false)) {
+
+ is = c.commit(InputStream.class);
+
+ try {
+ c.useAsBuffer(buffer -> null);
+ fail("connector should be closed");
+ } catch (IllegalStateException e) {
+ // Expected behavior
+ try {
+ is.read();
+ } catch (IOException bis) {
+ fail("We queried for the input stream to stay open.");
+ }
+ }
+ }
+
+ try ( final InputStream close = is ) {
+ is.read();
+ } catch (IOException e) {
+ fail("Committed storage view should still be opened.");
+ }
+ }
+
+ @Test
+ public void closing_multiple_times_causes_no_error() throws Exception {
+ try ( StrictStorageConnector c = create(true) ) {
+
+ c.commit(InputStream.class);
+ c.closeAllExcept(null);
+ }
+ }
+}
diff --git a/storage/sis-storage/src/test/java/org/apache/sis/test/suite/StorageTestSuite.java b/storage/sis-storage/src/test/java/org/apache/sis/test/suite/StorageTestSuite.java
index b89e232..3d50486 100644
--- a/storage/sis-storage/src/test/java/org/apache/sis/test/suite/StorageTestSuite.java
+++ b/storage/sis-storage/src/test/java/org/apache/sis/test/suite/StorageTestSuite.java
@@ -44,6 +44,7 @@ import org.junit.BeforeClass;
org.apache.sis.storage.FeatureNamingTest.class,
org.apache.sis.storage.ProbeResultTest.class,
org.apache.sis.storage.StorageConnectorTest.class,
+ org.apache.sis.storage.StrictStorageConnectorTest.class,
org.apache.sis.storage.event.StoreListenersTest.class,
org.apache.sis.internal.storage.query.CoverageQueryTest.class,
org.apache.sis.internal.storage.query.SimpleQueryTest.class,