Adding support for Bidi read

gcsio/src/main/java/com/google/cloud/hadoop/gcsio/GoogleCloudStorageBidiReadChannel.java

@@ -1,51 +1,60 @@
 package com.google.cloud.hadoop.gcsio;
 
+import static com.google.common.base.Strings.nullToEmpty;
+
 import com.google.api.core.ApiFuture;
 import com.google.api.core.ApiFutureCallback;
 import com.google.api.core.ApiFutures;
-import com.google.cloud.storage.BlobId;
-import com.google.cloud.storage.BlobReadSession;
-import com.google.cloud.storage.RangeSpec;
-import com.google.cloud.storage.ReadProjectionConfigs;
-import com.google.cloud.storage.Storage;
+import com.google.cloud.hadoop.util.GoogleCloudStorageEventBus;
+import com.google.cloud.storage.*;
 import com.google.cloud.storage.ZeroCopySupport.DisposableByteString;
+import com.google.common.annotations.VisibleForTesting;
 import com.google.protobuf.ByteString;
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.nio.channels.ClosedChannelException;
 import java.nio.channels.SeekableByteChannel;
 import java.util.List;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.function.IntFunction;
+import javax.annotation.Nullable;
 
 public class GoogleCloudStorageBidiReadChannel implements ReadVectoredSeekableByteChannel {
-
-  private final Storage storage;
-  private final StorageResourceId resourceId;
-  private final BlobId blobId;
-  private final GoogleCloudStorageReadOptions readOptions;
   private final BlobReadSession blobReadSession;
-  private ExecutorService boundedThreadPool;
+  private final ExecutorService boundedThreadPool;
+  private static final String GZIP_ENCODING = "gzip";
+  private long objectSize;
+  private boolean isOpen = true;
+  private boolean gzipEncoded = false;
+  @VisibleForTesting public SeekableByteChannel contentReadChannel;
+  @VisibleForTesting public long currentPosition = 0;
 
   public GoogleCloudStorageBidiReadChannel(
       Storage storage,
       GoogleCloudStorageItemInfo itemInfo,
       GoogleCloudStorageReadOptions readOptions,
       ExecutorService boundedThreadPool)
       throws IOException {
-    this.storage = storage;
-    this.resourceId =
+    StorageResourceId resourceId =
         new StorageResourceId(
             itemInfo.getBucketName(), itemInfo.getObjectName(), itemInfo.getContentGeneration());
-    this.blobId =
+    BlobId blobId =
         BlobId.of(
             resourceId.getBucketName(), resourceId.getObjectName(), resourceId.getGenerationId());
-    this.readOptions = readOptions;
     this.blobReadSession =
         initializeBlobReadSession(storage, blobId, readOptions.getBidiClientTimeout());
     this.boundedThreadPool = boundedThreadPool;
+    initMetadata(itemInfo.getContentEncoding(), itemInfo.getSize());
+    initializeReadSession();
+  }
+
+  private void initializeReadSession() {
+    ReadAsSeekableChannel seekableChannelConfig = ReadProjectionConfigs.asSeekableChannel();
+    this.contentReadChannel = blobReadSession.readAs(seekableChannelConfig);
+    this.currentPosition = 0;
   }
 
   private static BlobReadSession initializeBlobReadSession(
@@ -59,8 +68,23 @@ private static BlobReadSession initializeBlobReadSession(
 
   @Override
   public int read(ByteBuffer dst) throws IOException {
-    // TODO(dhritichopra) Add read flow
-    return 0;
+    throwIfNotOpen();
+    if (!dst.hasRemaining()) {
+      return 0;
+    }
+
+    if (currentPosition >= objectSize) {
+      return -1;
+    }
+    int bytesRead = contentReadChannel.read(dst);
+
+    if (bytesRead == -1) {
+      this.currentPosition = objectSize;
+      return -1;
+    }
+    currentPosition = currentPosition + bytesRead;
+
+    return bytesRead;
   }
 
   @Override
@@ -70,37 +94,61 @@ public int write(ByteBuffer src) throws IOException {
 
   @Override
   public long position() throws IOException {
-    // TODO(dhritichopra) Add read flow
-    return 0;
+    throwIfNotOpen();
+    return currentPosition;
   }
 
   @Override
   public SeekableByteChannel position(long newPosition) throws IOException {
-    // TODO(dhritichopra) Add read flow
-    return null;
+    throwIfNotOpen();
+    if (newPosition < 0) {
+      throw new IOException(String.format("Invalid seek position: %d", newPosition));
+    }
+    if (newPosition > objectSize) {
+      throw new java.io.EOFException(
+          String.format("Seek position %d is beyond file size %d", newPosition, objectSize));
+    }
+    if (gzipEncoded) {
+      throw new IOException("Gzip is not supported");
+    }
+    if (newPosition == this.currentPosition) {
+      return this;
+    }
+    contentReadChannel.position(newPosition);
+    this.currentPosition = newPosition;
+    return this;
   }
 
   @Override
   public long size() throws IOException {
-    // TODO(dhritichopra) Add read flow
-    return 0;
+    throwIfNotOpen();
+    if (objectSize == -1) {
+      throw new IOException("Size of file is not available");
+    }
+    return objectSize;
   }
 
   @Override
   public SeekableByteChannel truncate(long size) throws IOException {
-    // TODO(dhritichopra) Add read flow
-    return null;
+    throw new UnsupportedOperationException("Cannot truncate a read-only channel");
   }
 
   @Override
   public boolean isOpen() {
-    // TODO(dhritichopra) Add read flow
-    return false;
+    return isOpen;
   }
 
   @Override
   public void close() throws IOException {
-    blobReadSession.close();
+    if (isOpen) {
+      isOpen = false;
+      if (contentReadChannel != null) {
+        contentReadChannel.close();
+      }
+      if (blobReadSession != null) {
+        blobReadSession.close();
+      }
+    }
   }
 
   @Override
@@ -151,4 +199,20 @@ private void processBytesAndCompleteRange(
       range.getData().complete(buf);
     }
   }
+
+  private void throwIfNotOpen() throws IOException {
+    if (!isOpen()) {
+      GoogleCloudStorageEventBus.postOnException();
+      throw new ClosedChannelException();
+    }
+  }
+
+  protected void initMetadata(@Nullable String encoding, long sizeFromMetadata)
+      throws UnsupportedOperationException {
+    gzipEncoded = nullToEmpty(encoding).contains(GZIP_ENCODING);
+    if (gzipEncoded) {
+      throw new UnsupportedOperationException("Gzip Encoded Files are not supported");
+    }
+    objectSize = sizeFromMetadata;
+  }
 }

gcsio/src/test/java/com/google/cloud/hadoop/gcsio/FakeBlobReadSession.java

@@ -1,23 +1,22 @@
 package com.google.cloud.hadoop.gcsio;
 
-import static com.google.common.truth.Truth.assertThat;
-
 import com.google.api.core.ApiFutures;
-import com.google.cloud.storage.BlobInfo;
-import com.google.cloud.storage.BlobReadSession;
-import com.google.cloud.storage.RangeSpec;
-import com.google.cloud.storage.ReadAsFutureByteString;
-import com.google.cloud.storage.ReadProjectionConfig;
+import com.google.cloud.storage.*;
 import com.google.cloud.storage.ZeroCopySupport.DisposableByteString;
 import com.google.protobuf.ByteString;
 import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.NonWritableChannelException;
+import java.nio.channels.SeekableByteChannel;
 import java.nio.charset.StandardCharsets;
+import java.util.concurrent.Future;
 
 public class FakeBlobReadSession implements BlobReadSession {
 
-  private static final String TEST_STRING =
+  public static final String TEST_STRING =
       "Lorem ipsum dolor sit amet. Qui esse voluptatum qui tempora quia quo maiores galisum. Et officia cum";
 
+  public static final int TEST_STRING_SIZE = TEST_STRING.length();
   // The following are substring used in Test cases following the format SUBSTRING_{Offset}_{Length}
   public static final String SUBSTRING_20_10 = TEST_STRING.substring(20, 30);
   public static final String SUBSTRING_50_7 = TEST_STRING.substring(50, 57);
@@ -30,19 +29,78 @@ public BlobInfo getBlobInfo() {
 
   @Override
   public <Projection> Projection readAs(ReadProjectionConfig<Projection> readProjectionConfig) {
-    assertThat(readProjectionConfig).isInstanceOf(ReadAsFutureByteString.class);
-    RangeSpec range = ((ReadAsFutureByteString) readProjectionConfig).getRange();
-    return (Projection)
-        ApiFutures.immediateFuture(
-            new DisposableByteString() {
-              @Override
-              public ByteString byteString() {
-                return ByteString.copyFrom(getSubString(range).getBytes(StandardCharsets.UTF_8));
-              }
-
-              @Override
-              public void close() throws IOException {}
-            });
+    if (readProjectionConfig instanceof ReadAsSeekableChannel) {
+      return (Projection) createFakeSeekableByteChannel();
+    } else if (readProjectionConfig instanceof ReadAsFutureByteString) {
+      RangeSpec range = ((ReadAsFutureByteString) readProjectionConfig).getRange();
+      return (Projection) createAsFutureByteString(range);
+    }
+    throw new UnsupportedOperationException("Projection not Supported");
+  }
+
+  private Future<DisposableByteString> createAsFutureByteString(RangeSpec range) {
+    return ApiFutures.immediateFuture(
+        new DisposableByteString() {
+          @Override
+          public ByteString byteString() {
+            return ByteString.copyFrom(getSubString(range).getBytes(StandardCharsets.UTF_8));
+          }
+
+          @Override
+          public void close() throws IOException {}
+        });
+  }
+
+  private SeekableByteChannel createFakeSeekableByteChannel() {
+    return new SeekableByteChannel() {
+      private final byte[] data = TEST_STRING.getBytes(StandardCharsets.UTF_8);
+      private int position = 0;
+      private boolean open = true;
+
+      @Override
+      public int read(ByteBuffer dst) {
+        int bytesToRead = Math.min(dst.remaining(), data.length - position);
+        dst.put(data, position, bytesToRead);
+        position += bytesToRead;
+        return bytesToRead;
+      }
+
+      @Override
+      public long position() {
+        return position;
+      }
+
+      @Override
+      public SeekableByteChannel position(long newPosition) {
+        this.position = (int) newPosition;
+        return this;
+      }
+
+      @Override
+      public long size() {
+        return data.length;
+      }
+
+      @Override
+      public boolean isOpen() {
+        return open;
+      }
+
+      @Override
+      public void close() {
+        open = false;
+      }
+
+      @Override
+      public int write(ByteBuffer src) {
+        throw new NonWritableChannelException();
+      }
+
+      @Override
+      public SeekableByteChannel truncate(long size) {
+        throw new NonWritableChannelException();
+      }
+    };
   }
 
   private String getSubString(RangeSpec range) {