diff --git a/src/main/java/com/google/gcloud/examples/StorageExample.java b/src/main/java/com/google/gcloud/examples/StorageExample.java index b6f54efa1d93..a4550f19d2de 100644 --- a/src/main/java/com/google/gcloud/examples/StorageExample.java +++ b/src/main/java/com/google/gcloud/examples/StorageExample.java @@ -16,6 +16,7 @@ package com.google.gcloud.examples; +import com.google.gcloud.RetryParams; import com.google.gcloud.spi.StorageRpc.Tuple; import com.google.gcloud.storage.BatchRequest; import com.google.gcloud.storage.BatchResponse; @@ -498,7 +499,8 @@ public static void main(String... args) throws Exception { printUsage(); return; } - StorageServiceOptions.Builder optionsBuilder = StorageServiceOptions.builder(); + StorageServiceOptions.Builder optionsBuilder = + StorageServiceOptions.builder().retryParams(RetryParams.getDefaultInstance()); StorageAction action; if (args.length >= 2 && !ACTIONS.containsKey(args[0])) { optionsBuilder.projectId(args[0]); diff --git a/src/main/java/com/google/gcloud/storage/BlobReadChannelImpl.java b/src/main/java/com/google/gcloud/storage/BlobReadChannelImpl.java new file mode 100644 index 000000000000..8ca8a01f2df3 --- /dev/null +++ b/src/main/java/com/google/gcloud/storage/BlobReadChannelImpl.java @@ -0,0 +1,140 @@ +/* + * Copyright 2015 Google Inc. All Rights Reserved. + * + * Licensed 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 com.google.gcloud.storage; + +import static com.google.gcloud.RetryHelper.runWithRetries; + +import com.google.api.services.storage.model.StorageObject; +import com.google.gcloud.spi.StorageRpc; + +import java.io.IOException; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; +import java.nio.ByteBuffer; +import java.util.Map; +import java.util.concurrent.Callable; + +/** + * Default implementation for BlobReadChannel. + */ +class BlobReadChannelImpl implements BlobReadChannel { + + private static final int MIN_BUFFER_SIZE = 2 * 1024 * 1024; + private static final long serialVersionUID = 4821762590742862669L; + + private final StorageServiceOptions serviceOptions; + private final Blob blob; + private final Map requestOptions; + private int position; + private boolean isOpen; + private boolean endOfStream; + + private transient StorageRpc storageRpc; + private transient StorageObject storageObject; + private transient int bufferPos; + private transient byte[] buffer; + + BlobReadChannelImpl(StorageServiceOptions serviceOptions, Blob blob, + Map requestOptions) { + this.serviceOptions = serviceOptions; + this.blob = blob; + this.requestOptions = requestOptions; + isOpen = true; + initTransients(); + } + + private void writeObject(ObjectOutputStream out) throws IOException { + if (buffer != null) { + position += bufferPos; + buffer = null; + bufferPos = 0; + endOfStream = false; + } + out.defaultWriteObject(); + } + + private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException { + in.defaultReadObject(); + initTransients(); + } + + private void initTransients() { + storageRpc = serviceOptions.storageRpc(); + storageObject = blob.toPb(); + } + + @Override + public boolean isOpen() { + return isOpen; + } + + @Override + public void close() { + if (isOpen) { + buffer = null; + isOpen = false; + } + } + + private void validateOpen() throws IOException { + if (!isOpen) { + throw new IOException("stream is closed"); + } + } + + @Override + public void seek(int position) throws IOException { + validateOpen(); + this.position = position; + buffer = null; + bufferPos = 0; + endOfStream = false; + } + + @Override + public int read(ByteBuffer byteBuffer) throws IOException { + validateOpen(); + if (buffer == null) { + if (endOfStream) { + return -1; + } + final int toRead = Math.max(byteBuffer.remaining(), MIN_BUFFER_SIZE); + buffer = runWithRetries(new Callable() { + @Override + public byte[] call() { + return storageRpc.read(storageObject, requestOptions, position, toRead); + } + }, serviceOptions.retryParams(), StorageServiceImpl.EXCEPTION_HANDLER); + if (toRead > buffer.length) { + endOfStream = true; + if (buffer.length == 0) { + buffer = null; + return -1; + } + } + } + int toWrite = Math.min(buffer.length - bufferPos, byteBuffer.remaining()); + byteBuffer.put(buffer, bufferPos, toWrite); + bufferPos += toWrite; + if (bufferPos >= buffer.length) { + position += buffer.length; + buffer = null; + bufferPos = 0; + } + return toWrite; + } +} diff --git a/src/main/java/com/google/gcloud/storage/BlobWriterChannelImpl.java b/src/main/java/com/google/gcloud/storage/BlobWriterChannelImpl.java new file mode 100644 index 000000000000..b7736346bba0 --- /dev/null +++ b/src/main/java/com/google/gcloud/storage/BlobWriterChannelImpl.java @@ -0,0 +1,138 @@ +/* + * Copyright 2015 Google Inc. All Rights Reserved. + * + * Licensed 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 com.google.gcloud.storage; + +import static com.google.gcloud.RetryHelper.runWithRetries; +import static java.util.concurrent.Executors.callable; + +import com.google.api.services.storage.model.StorageObject; +import com.google.gcloud.spi.StorageRpc; + +import java.io.IOException; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; +import java.nio.ByteBuffer; +import java.util.Arrays; +import java.util.Map; + +/** + * Default implementation for BlobWriteChannel. + */ +class BlobWriterChannelImpl implements BlobWriteChannel { + + private static final long serialVersionUID = 8675286882724938737L; + private static final int CHUNK_SIZE = 256 * 1024; + private static final int MIN_BUFFER_SIZE = 8 * CHUNK_SIZE; + + private final StorageServiceOptions options; + private final Blob blob; + private final String uploadId; + private int position; + private byte[] buffer = new byte[0]; + private int limit; + private boolean isOpen = true; + + private transient StorageRpc storageRpc; + private transient StorageObject storageObject; + + public BlobWriterChannelImpl(StorageServiceOptions options, Blob blob, + Map optionsMap) { + this.options = options; + this.blob = blob; + initTransients(); + uploadId = options.storageRpc().open(storageObject, optionsMap); + } + + private void writeObject(ObjectOutputStream out) throws IOException { + if (isOpen) { + flush(true); + } + out.defaultWriteObject(); + } + + private void flush(boolean compact) { + if (limit >= MIN_BUFFER_SIZE || compact && limit >= CHUNK_SIZE) { + final int length = limit - limit % CHUNK_SIZE; + runWithRetries(callable(new Runnable() { + @Override + public void run() { + storageRpc.write(uploadId, buffer, 0, storageObject, position, length, false); + } + }), options.retryParams(), StorageServiceImpl.EXCEPTION_HANDLER); + position += length; + limit -= length; + byte[] temp = new byte[compact ? limit : MIN_BUFFER_SIZE]; + System.arraycopy(buffer, length, temp, 0, limit); + buffer = temp; + } + } + + private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException { + in.defaultReadObject(); + if (isOpen) { + initTransients(); + } + } + + private void initTransients() { + storageRpc = options.storageRpc(); + storageObject = blob.toPb(); + } + + private void validateOpen() throws IOException { + if (!isOpen) { + throw new IOException("stream is closed"); + } + } + + @Override + public int write(ByteBuffer byteBuffer) throws IOException { + validateOpen(); + int toWrite = byteBuffer.remaining(); + int spaceInBuffer = buffer.length - limit; + if (spaceInBuffer >= toWrite) { + byteBuffer.get(buffer, limit, toWrite); + } else { + buffer = Arrays.copyOf(buffer, + Math.max(MIN_BUFFER_SIZE, buffer.length + toWrite - spaceInBuffer)); + byteBuffer.get(buffer, limit, toWrite); + } + limit += toWrite; + flush(false); + return toWrite; + } + + @Override + public boolean isOpen() { + return isOpen; + } + + @Override + public void close() throws IOException { + if (isOpen) { + runWithRetries(callable(new Runnable() { + @Override + public void run() { + storageRpc.write(uploadId, buffer, 0, storageObject, position, limit, true); + } + }), options.retryParams(), StorageServiceImpl.EXCEPTION_HANDLER); + position += buffer.length; + isOpen = false; + buffer = null; + } + } +} diff --git a/src/main/java/com/google/gcloud/storage/StorageServiceImpl.java b/src/main/java/com/google/gcloud/storage/StorageServiceImpl.java index 377ee7251f10..95b95141be14 100644 --- a/src/main/java/com/google/gcloud/storage/StorageServiceImpl.java +++ b/src/main/java/com/google/gcloud/storage/StorageServiceImpl.java @@ -46,11 +46,7 @@ import com.google.gcloud.spi.StorageRpc; import com.google.gcloud.spi.StorageRpc.Tuple; -import java.io.IOException; -import java.io.ObjectInputStream; -import java.io.ObjectOutputStream; import java.io.Serializable; -import java.nio.ByteBuffer; import java.util.Arrays; import java.util.List; import java.util.Map; @@ -77,7 +73,7 @@ public RetryResult beforeEval(Exception exception) { return null; } }; - private static final ExceptionHandler EXCEPTION_HANDLER = ExceptionHandler.builder() + static final ExceptionHandler EXCEPTION_HANDLER = ExceptionHandler.builder() .abortOn(RuntimeException.class).interceptor(EXCEPTION_HANDLER_INTERCEPTOR).build(); private static final byte[] EMPTY_BYTE_ARRAY = {}; @@ -421,230 +417,13 @@ private List> transformBatch return response; } - private static class BlobReadChannelImpl implements BlobReadChannel { - - private static final long serialVersionUID = 1612561791239832259L; - - private final StorageServiceOptions serviceOptions; - private final Blob blob; - private final Map requestOptions; - private int position; - private boolean isOpen; - private boolean endOfStream; - - private transient StorageRpc storageRpc; - private transient StorageObject storageObject; - private transient int bufferPos; - private transient byte[] buffer; - - BlobReadChannelImpl(StorageServiceOptions serviceOptions, Blob blob, - Map requestOptions) { - this.serviceOptions = serviceOptions; - this.blob = blob; - this.requestOptions = requestOptions; - isOpen = true; - initTransients(); - } - - private void writeObject(ObjectOutputStream out) throws IOException { - if (buffer != null) { - position += bufferPos; - buffer = null; - bufferPos = 0; - endOfStream = false; - } - out.defaultWriteObject(); - } - - private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException { - in.defaultReadObject(); - initTransients(); - } - - private void initTransients() { - storageRpc = serviceOptions.storageRpc(); - storageObject = blob.toPb(); - } - - @Override - public boolean isOpen() { - return isOpen; - } - - @Override - public void close() { - if (isOpen) { - buffer = null; - isOpen = false; - } - } - - private void validateOpen() throws IOException { - if (!isOpen) { - throw new IOException("stream is closed"); - } - } - - @Override - public void seek(int position) throws IOException { - validateOpen(); - this.position = position; - buffer = null; - bufferPos = 0; - endOfStream = false; - } - - @Override - public int read(ByteBuffer byteBuffer) throws IOException { - validateOpen(); - if (buffer == null) { - if (endOfStream) { - return -1; - } - final int toRead = Math.max(byteBuffer.remaining(), 256 * 1024); - buffer = runWithRetries(new Callable() { - @Override - public byte[] call() { - return storageRpc.read(storageObject, requestOptions, position, toRead); - } - }, serviceOptions.retryParams(), EXCEPTION_HANDLER); - if (toRead > buffer.length) { - endOfStream = true; - } - } - int toWrite = Math.min(buffer.length - bufferPos, byteBuffer.remaining()); - byteBuffer.put(buffer, bufferPos, toWrite); - bufferPos += toWrite; - if (bufferPos >= buffer.length) { - position += buffer.length; - buffer = null; - bufferPos = 0; - } - return toWrite; - } - } - @Override public BlobReadChannel reader(String bucket, String blob, BlobSourceOption... options) { Map optionsMap = optionMap(options); return new BlobReadChannelImpl(options(), Blob.of(bucket, blob), optionsMap); } - private static class BlobWriterChannelImpl implements BlobWriteChannel { - - private static final int CHUNK_SIZE = 256 * 1024; - private static final int COMPACT_THRESHOLD = (int) Math.round(CHUNK_SIZE * 0.8); - private static final long serialVersionUID = -4067648781804698786L; - - private final StorageServiceOptions options; - private final Blob blob; - private final String uploadId; - private int position; - private byte[] buffer = new byte[CHUNK_SIZE]; - private int limit; - private boolean isOpen = true; - - private transient StorageRpc storageRpc; - private transient StorageObject storageObject; - - public BlobWriterChannelImpl(StorageServiceOptions options, Blob blob, - Map optionsMap) { - this.options = options; - this.blob = blob; - initTransients(); - uploadId = options.storageRpc().open(storageObject, optionsMap); - } - - private void writeObject(ObjectOutputStream out) throws IOException { - if (!isOpen) { - out.defaultWriteObject(); - return; - } - flush(); - byte[] temp = buffer; - if (limit < COMPACT_THRESHOLD) { - buffer = Arrays.copyOf(buffer, limit); - } - out.defaultWriteObject(); - buffer = temp; - } - - private void flush() { - if (limit >= CHUNK_SIZE) { - final int length = limit - limit % CHUNK_SIZE; - runWithRetries(callable(new Runnable() { - @Override - public void run() { - storageRpc.write(uploadId, buffer, 0, storageObject, position, length, false); - } - }), options.retryParams(), EXCEPTION_HANDLER); - position += length; - limit -= length; - byte[] temp = new byte[CHUNK_SIZE]; - System.arraycopy(buffer, length, temp, 0, limit); - buffer = temp; - } - } - - private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException { - in.defaultReadObject(); - if (isOpen) { - if (buffer.length < CHUNK_SIZE) { - buffer = Arrays.copyOf(buffer, CHUNK_SIZE); - } - initTransients(); - } - } - - private void initTransients() { - storageRpc = options.storageRpc(); - storageObject = blob.toPb(); - } - - private void validateOpen() throws IOException { - if (!isOpen) { - throw new IOException("stream is closed"); - } - } - - @Override - public int write(ByteBuffer byteBuffer) throws IOException { - validateOpen(); - int toWrite = byteBuffer.remaining(); - int spaceInBuffer = buffer.length - limit; - if (spaceInBuffer >= toWrite) { - byteBuffer.get(buffer, limit, toWrite); - } else { - buffer = Arrays.copyOf(buffer, buffer.length + toWrite - spaceInBuffer); - byteBuffer.get(buffer, limit, toWrite); - } - limit += toWrite; - flush(); - return toWrite; - } - - @Override - public boolean isOpen() { - return isOpen; - } - - @Override - public void close() throws IOException { - if (isOpen) { - runWithRetries(callable(new Runnable() { - @Override - public void run() { - storageRpc.write(uploadId, buffer, 0, storageObject, position, limit, true); - } - }), options.retryParams(), EXCEPTION_HANDLER); - position += buffer.length; - isOpen = false; - buffer = null; - } - } - } - - @Override + @Override public BlobWriteChannel writer(Blob blob, BlobTargetOption... options) { final Map optionsMap = optionMap(blob, options); return new BlobWriterChannelImpl(options(), blob, optionsMap);