GoogleCloudStorageGrpcReadChannel implements SeekableByteChannel { @Override public int read(ByteBuffer byteBuffer) throws IOException { logger.atFine().log( "GCS gRPC read request for up to %d bytes at offset %d from object %s", byteBuffer.remaining(), position(), objectName); if (!isOpen()) { throw new ClosedChannelException(); } int bytesRead = 0; if (bufferedContent != null) { bytesRead += readBufferedContentInto(byteBuffer); } if (!byteBuffer.hasRemaining()) { return bytesRead; } if (position == objectSize) { return bytesRead > 0 ? bytesRead : -1; } if (resIterator == null) { OptionalLong bytesToRead; if (readStrategy == Fadvise.RANDOM) { long randomRangeSize = Math.max(byteBuffer.remaining(), readOptions.getMinRangeRequestSize()); bytesToRead = OptionalLong.of(randomRangeSize); } else { bytesToRead = OptionalLong.empty(); } requestObjectMedia(bytesToRead); } while (moreServerContent() && byteBuffer.hasRemaining()) { GetObjectMediaResponse res = resIterator.next(); ByteString content = res.getChecksummedData().getContent(); if (bytesToSkipBeforeReading >= 0 && bytesToSkipBeforeReading < content.size()) { content = res.getChecksummedData().getContent().substring((int) bytesToSkipBeforeReading); bytesToSkipBeforeReading = 0; } else if (bytesToSkipBeforeReading >= content.size()) { bytesToSkipBeforeReading -= content.size(); continue; } if (readOptions.isGrpcChecksumsEnabled() && res.getChecksummedData().hasCrc32C()) { Hasher messageHasher = Hashing.crc32c().newHasher(); messageHasher.putBytes(res.getChecksummedData().getContent().toByteArray()); int calculatedChecksum = messageHasher.hash().asInt(); int expectedChecksum = res.getChecksummedData().getCrc32C().getValue(); if (calculatedChecksum != expectedChecksum) { throw new IOException( String.format( "For %s: Message checksum didn't match. Expected %s, got %s.", this, expectedChecksum, calculatedChecksum)); } } boolean responseSizeLargerThanRemainingBuffer = content.size() > byteBuffer.remaining(); int bytesToWrite = responseSizeLargerThanRemainingBuffer ? byteBuffer.remaining() : content.size(); put(content, 0, bytesToWrite, byteBuffer); bytesRead += bytesToWrite; position += bytesToWrite; if (responseSizeLargerThanRemainingBuffer) { bufferedContent = content; bufferedContentReadOffset = bytesToWrite; } } return bytesRead; } private  GoogleCloudStorageGrpcReadChannel(
      StorageBlockingStub gcsGrpcBlockingStub,
      String bucketName,
      String objectName,
      long objectGeneration,
      long objectSize,
      GoogleCloudStorageReadOptions readOptions); static GoogleCloudStorageGrpcReadChannel open(
      StorageBlockingStub stub,
      String bucketName,
      String objectName,
      GoogleCloudStorageReadOptions readOptions); @Override int read(ByteBuffer byteBuffer); @Override int write(ByteBuffer byteBuffer); @Override long position(); @Override SeekableByteChannel position(long newPosition); @Override long size(); @Override SeekableByteChannel truncate(long l); @Override boolean isOpen(); @Override void close(); @Override String toString();  }
@Test public void readSingleChunkSucceeds() throws Exception { fakeService.setObject(DEFAULT_OBJECT.toBuilder().setSize(100).build()); GoogleCloudStorageGrpcReadChannel readChannel = newReadChannel(); ByteBuffer buffer = ByteBuffer.allocate(100); readChannel.read(buffer); verify(fakeService, times(1)).getObject(eq(GET_OBJECT_REQUEST), any()); verify(fakeService, times(1)).getObjectMedia(eq(GET_OBJECT_MEDIA_REQUEST), any()); assertArrayEquals(fakeService.data.substring(0, 100).toByteArray(), buffer.array()); }
@Test public void readMultipleChunksSucceeds() throws Exception { fakeService.setObject(DEFAULT_OBJECT.toBuilder().setSize(FakeService.CHUNK_SIZE * 2).build()); GoogleCloudStorageGrpcReadChannel readChannel = newReadChannel(); ByteBuffer buffer = ByteBuffer.allocate(FakeService.CHUNK_SIZE * 2); readChannel.read(buffer); verify(fakeService, times(1)).getObject(eq(GET_OBJECT_REQUEST), any()); verify(fakeService, times(1)).getObjectMedia(eq(GET_OBJECT_MEDIA_REQUEST), any()); assertArrayEquals( fakeService.data.substring(0, FakeService.CHUNK_SIZE * 2).toByteArray(), buffer.array()); }
@Test public void multipleSequentialReads() throws Exception { fakeService.setObject(DEFAULT_OBJECT.toBuilder().setSize(100).build()); GoogleCloudStorageGrpcReadChannel readChannel = newReadChannel(); ByteBuffer first_buffer = ByteBuffer.allocate(10); ByteBuffer second_buffer = ByteBuffer.allocate(20); readChannel.read(first_buffer); readChannel.read(second_buffer); verify(fakeService, times(1)).getObject(eq(GET_OBJECT_REQUEST), any()); verify(fakeService, times(1)).getObjectMedia(eq(GET_OBJECT_MEDIA_REQUEST), any()); assertArrayEquals(fakeService.data.substring(0, 10).toByteArray(), first_buffer.array()); assertArrayEquals(fakeService.data.substring(10, 30).toByteArray(), second_buffer.array()); }
@Test public void readToBufferWithArrayOffset() throws Exception { fakeService.setObject(DEFAULT_OBJECT.toBuilder().setSize(100).build()); GoogleCloudStorageGrpcReadChannel readChannel = newReadChannel(); byte[] array = new byte[200]; ByteBuffer buffer = ByteBuffer.wrap(array, 50, 150).slice(); readChannel.read(buffer); verify(fakeService, times(1)).getObject(eq(GET_OBJECT_REQUEST), any()); verify(fakeService, times(1)).getObjectMedia(eq(GET_OBJECT_MEDIA_REQUEST), any()); byte[] expected = ByteString.copyFrom(array, 50, 100).toByteArray(); assertArrayEquals(fakeService.data.substring(0, 100).toByteArray(), expected); }
@Test public void singleReadSucceedsWithValidObjectChecksum() throws Exception { fakeService.setObject( DEFAULT_OBJECT.toBuilder() .setCrc32C(UInt32Value.newBuilder().setValue(DEFAULT_OBJECT_CRC32C)) .build()); GoogleCloudStorageReadOptions options = GoogleCloudStorageReadOptions.builder().setGrpcChecksumsEnabled(true).build(); GoogleCloudStorageGrpcReadChannel readChannel = newReadChannel(options); ByteBuffer buffer = ByteBuffer.allocate(OBJECT_SIZE); readChannel.read(buffer); assertArrayEquals(fakeService.data.toByteArray(), buffer.array()); }
@Test public void partialReadSucceedsWithInvalidObjectChecksum() throws Exception { fakeService.setObject( DEFAULT_OBJECT.toBuilder().setCrc32C(UInt32Value.newBuilder().setValue(0)).build()); GoogleCloudStorageReadOptions options = GoogleCloudStorageReadOptions.builder().setGrpcChecksumsEnabled(true).build(); GoogleCloudStorageGrpcReadChannel readChannel = newReadChannel(options); ByteBuffer buffer = ByteBuffer.allocate(OBJECT_SIZE - 10); readChannel.read(buffer); assertArrayEquals( fakeService.data.substring(0, OBJECT_SIZE - 10).toByteArray(), buffer.array()); }
@Test public void multipleSequentialsReadsSucceedWithValidObjectChecksum() throws Exception { fakeService.setObject( DEFAULT_OBJECT.toBuilder() .setCrc32C(UInt32Value.newBuilder().setValue(DEFAULT_OBJECT_CRC32C)) .build()); GoogleCloudStorageReadOptions options = GoogleCloudStorageReadOptions.builder().setGrpcChecksumsEnabled(true).build(); GoogleCloudStorageGrpcReadChannel readChannel = newReadChannel(options); ByteBuffer firstBuffer = ByteBuffer.allocate(100); ByteBuffer secondBuffer = ByteBuffer.allocate(OBJECT_SIZE - 100); readChannel.read(firstBuffer); readChannel.read(secondBuffer); assertArrayEquals(fakeService.data.substring(0, 100).toByteArray(), firstBuffer.array()); assertArrayEquals(fakeService.data.substring(100).toByteArray(), secondBuffer.array()); }
@Test public void readFailsWithInvalidMessageChecksum() throws Exception { GoogleCloudStorageReadOptions options = GoogleCloudStorageReadOptions.builder().setGrpcChecksumsEnabled(true).build(); GoogleCloudStorageGrpcReadChannel readChannel = newReadChannel(options); fakeService.setReturnIncorrectMessageChecksum(); ByteBuffer buffer = ByteBuffer.allocate(10); IOException thrown = assertThrows(IOException.class, () -> readChannel.read(buffer)); assertTrue(thrown.getMessage().contains("checksum")); }
@Test public void readToBufferWithArrayOffsetSucceeds() throws Exception { GoogleCloudStorageReadOptions options = GoogleCloudStorageReadOptions.builder().setGrpcChecksumsEnabled(true).build(); GoogleCloudStorageGrpcReadChannel readChannel = newReadChannel(options); byte[] array = new byte[OBJECT_SIZE + 100]; ByteBuffer buffer = ByteBuffer.wrap(array, 50, OBJECT_SIZE).slice(); readChannel.read(buffer); byte[] expected = ByteString.copyFrom(array, 50, OBJECT_SIZE).toByteArray(); assertArrayEquals(fakeService.data.toByteArray(), expected); }
@Test public void readToBufferWithArrayOffsetFailsWithInvalidMessageChecksum() throws Exception { fakeService.setReturnIncorrectMessageChecksum(); GoogleCloudStorageReadOptions options = GoogleCloudStorageReadOptions.builder().setGrpcChecksumsEnabled(true).build(); GoogleCloudStorageGrpcReadChannel readChannel = newReadChannel(options); byte[] array = new byte[OBJECT_SIZE + 100]; ByteBuffer buffer = ByteBuffer.wrap(array, 50, OBJECT_SIZE).slice(); IOException thrown = assertThrows(IOException.class, () -> readChannel.read(buffer)); assertTrue( thrown.getMessage() + " should have contained 'checksum'", thrown.getMessage().contains("checksum")); }
@Test public void multipleReadsIgnoreObjectChecksumForLatestGenerationReads() throws Exception { fakeService.setObject( DEFAULT_OBJECT.toBuilder().setCrc32C(UInt32Value.newBuilder().setValue(0)).build()); GoogleCloudStorageReadOptions options = GoogleCloudStorageReadOptions.builder().setGrpcChecksumsEnabled(true).build(); GoogleCloudStorageGrpcReadChannel readChannel = newReadChannel(options); ByteBuffer firstBuffer = ByteBuffer.allocate(100); ByteBuffer secondBuffer = ByteBuffer.allocate(OBJECT_SIZE - 100); readChannel.read(firstBuffer); readChannel.read(secondBuffer); assertArrayEquals(fakeService.data.substring(0, 100).toByteArray(), firstBuffer.array()); assertArrayEquals(fakeService.data.substring(100).toByteArray(), secondBuffer.array()); }
@Test public void readHandlesGetMediaError() throws Exception { fakeService.setGetMediaException( Status.fromCode(Status.Code.INTERNAL) .withDescription("Custom error message.") .asException()); GoogleCloudStorageGrpcReadChannel readChannel = newReadChannel(); ByteBuffer buffer = ByteBuffer.allocate(10); IOException thrown = assertThrows(IOException.class, () -> readChannel.read(buffer)); assertTrue(thrown.getCause().getMessage().contains("Custom error message.")); }
@Test public void fastFailOnNotFoundFailsByReadWhenDisabled() { fakeService.setGetException(Status.NOT_FOUND.asException()); GoogleCloudStorageReadOptions options = GoogleCloudStorageReadOptions.builder().setFastFailOnNotFound(false).build(); ByteBuffer buffer = ByteBuffer.allocate(10); assertThrows(FileNotFoundException.class, () -> newReadChannel(options).read(buffer)); }