Allow per-filesystem specification of channel reopen count

jean-philippe-martin · jean-philippe-martin · commit 52606c259d25 · 2017-03-10T15:42:08.000-08:00
diff --git a/google-cloud-contrib/google-cloud-nio/src/main/java/com/google/cloud/storage/contrib/nio/CloudStorageConfiguration.java b/google-cloud-contrib/google-cloud-nio/src/main/java/com/google/cloud/storage/contrib/nio/CloudStorageConfiguration.java
@@ -59,6 +59,12 @@ public abstract class CloudStorageConfiguration {
    */
   public abstract int blockSize();
 
+  /**
+   * Returns the number of times we try re-opening a channel if it's closed unexpectedly
+   * while reading.
+   */
+  public abstract int channelReopen();
+
   /**
    * Creates a new builder, initialized with the following settings:
    *
@@ -82,6 +88,7 @@ public static final class Builder {
     private boolean stripPrefixSlash = true;
     private boolean usePseudoDirectories = true;
     private int blockSize = CloudStorageFileSystem.BLOCK_SIZE_DEFAULT;
+    private int channelReopen = 0;
 
     /**
      * Changes current working directory for new filesystem. This defaults to the root directory.
@@ -134,6 +141,11 @@ public Builder blockSize(int value) {
       return this;
     }
 
+    public Builder channelReopen(int value) {
+      channelReopen = value;
+      return this;
+    }
+
     /**
      * Creates new instance without destroying builder.
      */
@@ -143,7 +155,8 @@ public CloudStorageConfiguration build() {
           permitEmptyPathComponents,
           stripPrefixSlash,
           usePseudoDirectories,
-          blockSize);
+          blockSize,
+          channelReopen);
     }
 
     Builder() {}
@@ -168,6 +181,9 @@ static CloudStorageConfiguration fromMap(Map<String, ?> env) {
         case "blockSize":
           builder.blockSize((Integer) entry.getValue());
           break;
+        case "channelReopen":
+          builder.channelReopen((Integer) entry.getValue());
+          break;
         default:
           throw new IllegalArgumentException(entry.getKey());
       }
diff --git a/google-cloud-contrib/google-cloud-nio/src/main/java/com/google/cloud/storage/contrib/nio/CloudStorageFileSystemProvider.java b/google-cloud-contrib/google-cloud-nio/src/main/java/com/google/cloud/storage/contrib/nio/CloudStorageFileSystemProvider.java
@@ -139,7 +139,6 @@ public CloudStorageFileSystemProvider() {
 
   CloudStorageFileSystemProvider(@Nullable StorageOptions gcsStorageOptions) {
     this.storageOptions = gcsStorageOptions;
-
   }
 
   // Initialize this.storage, once. This may throw an exception if default authentication
@@ -226,7 +225,7 @@ public SeekableByteChannel newByteChannel(
   private SeekableByteChannel newReadChannel(Path path, Set<? extends OpenOption> options)
       throws IOException {
     initStorage();
-    int channelReopen = 0;
+    int channelReopen = ((CloudStorageFileSystem)path.getFileSystem()).config().channelReopen();
     for (OpenOption option : options) {
       if (option instanceof StandardOpenOption) {
         switch ((StandardOpenOption) option) {
diff --git a/google-cloud-contrib/google-cloud-nio/src/main/java/com/google/cloud/storage/contrib/nio/CloudStorageReadChannel.java b/google-cloud-contrib/google-cloud-nio/src/main/java/com/google/cloud/storage/contrib/nio/CloudStorageReadChannel.java
@@ -44,13 +44,13 @@ final class CloudStorageReadChannel implements SeekableByteChannel {
 
   private final Storage gcsStorage;
   private final BlobId file;
-  private ReadChannel channel;
-  private long position;
-  private long size;
   // max # of times we may reopen the file
   private final int maxReopen;
   // how many times we re-opened the file
   private int reopens;
+  private ReadChannel channel;
+  private long position;
+  private long size;
 
   /**
    * @param maxReopen max. number of times to try re-opening the channel if it closes on us unexpectedly.
@@ -66,6 +66,7 @@ private CloudStorageReadChannel(Storage gcsStorage, BlobId file, long position,
     this.gcsStorage = gcsStorage;
     this.file = file;
     this.position = position;
+    this.reopens = 0;
     this.maxReopen = maxReopen;
     // XXX: Reading size and opening file should be atomic.
     this.size = fetchSize(gcsStorage, file);
@@ -107,9 +108,9 @@ public int read(ByteBuffer dst) throws IOException {
           amt = channel.read(dst);
           break;
         } catch (StorageException exs) {
-          // this error isn't marked as retryable since the channel is closed;
-          // but here at this higher level we can retry it.
           if (exs.getMessage().contains("Connection closed prematurely") && reopens < maxReopen) {
+            // this error isn't marked as retryable since the channel is closed;
+            // but here at this higher level we can retry it.
             reopens++;
             retryDelay(reopens);
             innerOpen();
