Skip to content

Commit f2b8f6b

Browse files
aozarovaozarov
committed
---
yaml --- r: 1963 b: refs/heads/pubsub-alpha c: 02694ca h: refs/heads/master i: 1961: 6b61328 1959: 86a1429
1 parent 2726f8e commit f2b8f6b

11 files changed

Lines changed: 35 additions & 130 deletions

File tree

‎[refs]‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@ refs/heads/master: 689bbb466df4b2d5d2483d6edb8ac5c7c7f7c6fa
33
refs/heads/travis: e21ee7b88a5edc3f3d8c71f90c3fc32abf7e8dd6
44
refs/heads/gh-pages: 4e0561bb4504bf647db669a14417b2b2c87ba45d
55
refs/heads/bigquery: 762fa5830e6c398c0396177e3e7fd243bd62cfc3
6-
refs/heads/pubsub-alpha: 39829a0fb487481a348496b5d930dd4c379a8cf8
6+
refs/heads/pubsub-alpha: 02694ca9280c9b0c482afcdd10d98b8afe42d50f
77
refs/heads/resource-manager: ebf4adc5ee835cd2086c4ac5b4e78d01a5a005a7
88
refs/heads/update-datastore: 482954f2c5055231e5b3122ea91d2ba00ce8187c
99
refs/tags/0.0.9: 22f1839238f66c39e67ed4dfdcd273b1ae2e8444

‎branches/pubsub-alpha/gcloud-java-datastore/src/main/java/com/google/gcloud/datastore/FullEntity.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,8 @@
1919
import com.google.api.services.datastore.DatastoreV1;
2020

2121
/**
22-
* A full entity is a {@link BaseEntity} that with a complete set of properties.
22+
* A full entity is a {@link BaseEntity} that holds all the properties associated with a
23+
* Datastore entity (as opposed to {@link ProjectionEntity}).
2324
*/
2425
public class FullEntity<K extends IncompleteKey> extends BaseEntity<K> {
2526

‎branches/pubsub-alpha/gcloud-java-storage/src/main/java/com/google/gcloud/spi/DefaultStorageRpc.java‎

Lines changed: 7 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -429,8 +429,8 @@ public void onFailure(GoogleJsonError e, HttpHeaders responseHeaders) {
429429
}
430430

431431
@Override
432-
public Tuple<String, byte[]> read(StorageObject from, Map<Option, ?> options, long position,
433-
int bytes) throws StorageException {
432+
public byte[] read(StorageObject from, Map<Option, ?> options, long position, int bytes)
433+
throws StorageException {
434434
try {
435435
Get req = storage.objects()
436436
.get(from.getBucket(), from.getName())
@@ -439,13 +439,12 @@ public Tuple<String, byte[]> read(StorageObject from, Map<Option, ?> options, lo
439439
.setIfMetagenerationNotMatch(IF_METAGENERATION_NOT_MATCH.getLong(options))
440440
.setIfGenerationMatch(IF_GENERATION_MATCH.getLong(options))
441441
.setIfGenerationNotMatch(IF_GENERATION_NOT_MATCH.getLong(options));
442-
StringBuilder range = new StringBuilder();
443-
range.append("bytes=").append(position).append("-").append(position + bytes - 1);
444-
req.getRequestHeaders().setRange(range.toString());
442+
MediaHttpDownloader downloader = req.getMediaHttpDownloader();
443+
downloader.setContentRange(position, Ints.checkedCast(position + bytes - 1));
444+
downloader.setDirectDownloadEnabled(true);
445445
ByteArrayOutputStream output = new ByteArrayOutputStream();
446-
req.executeMedia().download(output);
447-
String etag = req.getLastResponseHeaders().getETag();
448-
return Tuple.of(etag, output.toByteArray());
446+
req.executeMediaAndDownloadTo(output);
447+
return output.toByteArray();
449448
} catch (IOException ex) {
450449
throw translate(ex);
451450
}

‎branches/pubsub-alpha/gcloud-java-storage/src/main/java/com/google/gcloud/spi/StorageRpc.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -259,7 +259,7 @@ StorageObject compose(Iterable<StorageObject> sources, StorageObject target,
259259
byte[] load(StorageObject storageObject, Map<Option, ?> options)
260260
throws StorageException;
261261

262-
Tuple<String, byte[]> read(StorageObject from, Map<Option, ?> options, long position, int bytes)
262+
byte[] read(StorageObject from, Map<Option, ?> options, long position, int bytes)
263263
throws StorageException;
264264

265265
String open(StorageObject object, Map<Option, ?> options) throws StorageException;

‎branches/pubsub-alpha/gcloud-java-storage/src/main/java/com/google/gcloud/storage/BlobReadChannel.java‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -26,8 +26,9 @@
2626
/**
2727
* A channel for reading data from a Google Cloud Storage object.
2828
*
29-
* Implementations of this class may buffer data internally to reduce remote calls. This interface
30-
* implements {@link Restorable} to allow saving the reader's state to continue reading afterwards.
29+
* Implementations of this class may buffer data internally to reduce remote calls.
30+
*
31+
* This class is @{link Serializable}, which allows incremental reads.
3132
*/
3233
public interface BlobReadChannel extends ReadableByteChannel, Closeable,
3334
Restorable<BlobReadChannel> {

‎branches/pubsub-alpha/gcloud-java-storage/src/main/java/com/google/gcloud/storage/BlobReadChannelImpl.java‎

Lines changed: 4 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,6 @@
2323
import com.google.gcloud.RestorableState;
2424
import com.google.gcloud.RetryHelper;
2525
import com.google.gcloud.spi.StorageRpc;
26-
import com.google.gcloud.spi.StorageRpc.Tuple;
2726

2827
import java.io.IOException;
2928
import java.io.Serializable;
@@ -42,7 +41,6 @@ class BlobReadChannelImpl implements BlobReadChannel {
4241
private final StorageOptions serviceOptions;
4342
private final BlobId blob;
4443
private final Map<StorageRpc.Option, ?> requestOptions;
45-
private String lastEtag;
4644
private int position;
4745
private boolean isOpen;
4846
private boolean endOfStream;
@@ -119,19 +117,12 @@ public int read(ByteBuffer byteBuffer) throws IOException {
119117
}
120118
final int toRead = Math.max(byteBuffer.remaining(), chunkSize);
121119
try {
122-
Tuple<String, byte[]> result = runWithRetries(new Callable<Tuple<String, byte[]>>() {
120+
buffer = runWithRetries(new Callable<byte[]>() {
123121
@Override
124-
public Tuple<String, byte[]> call() {
122+
public byte[] call() {
125123
return storageRpc.read(storageObject, requestOptions, position, toRead);
126124
}
127125
}, serviceOptions.retryParams(), StorageImpl.EXCEPTION_HANDLER);
128-
if (lastEtag != null && !Objects.equals(result.x(), lastEtag)) {
129-
StringBuilder messageBuilder = new StringBuilder();
130-
messageBuilder.append("Blob ").append(blob).append(" was updated while reading");
131-
throw new StorageException(0, messageBuilder.toString(), false);
132-
}
133-
lastEtag = result.x();
134-
buffer = result.y();
135126
} catch (RetryHelper.RetryHelperException e) {
136127
throw StorageException.translateAndThrow(e);
137128
}
@@ -161,7 +152,6 @@ static class StateImpl implements RestorableState<BlobReadChannel>, Serializable
161152
private final StorageOptions serviceOptions;
162153
private final BlobId blob;
163154
private final Map<StorageRpc.Option, ?> requestOptions;
164-
private final String lastEtag;
165155
private final int position;
166156
private final boolean isOpen;
167157
private final boolean endOfStream;
@@ -171,7 +161,6 @@ static class StateImpl implements RestorableState<BlobReadChannel>, Serializable
171161
this.serviceOptions = builder.serviceOptions;
172162
this.blob = builder.blob;
173163
this.requestOptions = builder.requestOptions;
174-
this.lastEtag = builder.lastEtag;
175164
this.position = builder.position;
176165
this.isOpen = builder.isOpen;
177166
this.endOfStream = builder.endOfStream;
@@ -182,7 +171,6 @@ static class Builder {
182171
private final StorageOptions serviceOptions;
183172
private final BlobId blob;
184173
private final Map<StorageRpc.Option, ?> requestOptions;
185-
private String lastEtag;
186174
private int position;
187175
private boolean isOpen;
188176
private boolean endOfStream;
@@ -194,11 +182,6 @@ private Builder(StorageOptions options, BlobId blob, Map<StorageRpc.Option, ?> r
194182
this.requestOptions = reqOptions;
195183
}
196184

197-
Builder lastEtag(String lastEtag) {
198-
this.lastEtag = lastEtag;
199-
return this;
200-
}
201-
202185
Builder position(int position) {
203186
this.position = position;
204187
return this;
@@ -232,7 +215,6 @@ static Builder builder(
232215
@Override
233216
public BlobReadChannel restore() {
234217
BlobReadChannelImpl channel = new BlobReadChannelImpl(serviceOptions, blob, requestOptions);
235-
channel.lastEtag = lastEtag;
236218
channel.position = position;
237219
channel.isOpen = isOpen;
238220
channel.endOfStream = endOfStream;
@@ -242,8 +224,8 @@ public BlobReadChannel restore() {
242224

243225
@Override
244226
public int hashCode() {
245-
return Objects.hash(serviceOptions, blob, requestOptions, lastEtag, position, isOpen,
246-
endOfStream, chunkSize);
227+
return Objects.hash(serviceOptions, blob, requestOptions, position, isOpen, endOfStream,
228+
chunkSize);
247229
}
248230

249231
@Override
@@ -258,7 +240,6 @@ public boolean equals(Object obj) {
258240
return Objects.equals(this.serviceOptions, other.serviceOptions)
259241
&& Objects.equals(this.blob, other.blob)
260242
&& Objects.equals(this.requestOptions, other.requestOptions)
261-
&& Objects.equals(this.lastEtag, other.lastEtag)
262243
&& this.position == other.position
263244
&& this.isOpen == other.isOpen
264245
&& this.endOfStream == other.endOfStream

‎branches/pubsub-alpha/gcloud-java-storage/src/main/java/com/google/gcloud/storage/BlobWriteChannel.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -26,8 +26,8 @@
2626
* A channel for writing data to a Google Cloud Storage object.
2727
*
2828
* Implementations of this class may further buffer data internally to reduce remote calls. Written
29-
* data will only be visible after calling {@link #close()}. This interface implements
30-
* {@link Restorable} to allow saving the writer's state to continue writing afterwards.
29+
* data will only be visible after calling {@link #close()}. This class is serializable, to allow
30+
* incremental writes.
3131
*/
3232
public interface BlobWriteChannel extends WritableByteChannel, Closeable,
3333
Restorable<BlobWriteChannel> {

‎branches/pubsub-alpha/gcloud-java-storage/src/main/java/com/google/gcloud/storage/Storage.java‎

Lines changed: 2 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -33,7 +33,6 @@
3333
import java.io.InputStream;
3434
import java.io.Serializable;
3535
import java.net.URL;
36-
import java.nio.ByteBuffer;
3736
import java.util.Arrays;
3837
import java.util.Collections;
3938
import java.util.HashSet;
@@ -1406,29 +1405,14 @@ private static void checkContentType(BlobInfo blobInfo) throws IllegalArgumentEx
14061405
BatchResponse apply(BatchRequest batchRequest);
14071406

14081407
/**
1409-
* Return a channel for reading the blob's content. The blob's latest generation is read. If the
1410-
* blob changes while reading (i.e. {@link BlobInfo#etag()} changes), subsequent calls to
1411-
* {@code blobReadChannel.read(ByteBuffer)} may throw {@link StorageException}.
1412-
*
1413-
* <p>The {@link BlobSourceOption#generationMatch(long)} option can be provided to ensure that
1414-
* {@code blobReadChannel.read(ByteBuffer)} calls will throw {@link StorageException} if blob`s
1415-
* generation differs from the expected one.
1408+
* Return a channel for reading the blob's content.
14161409
*
14171410
* @throws StorageException upon failure
14181411
*/
14191412
BlobReadChannel reader(String bucket, String blob, BlobSourceOption... options);
14201413

14211414
/**
1422-
* Return a channel for reading the blob's content. If {@code blob.generation()} is set
1423-
* data corresponding to that generation is read. If {@code blob.generation()} is {@code null}
1424-
* the blob's latest generation is read. If the blob changes while reading (i.e.
1425-
* {@link BlobInfo#etag()} changes), subsequent calls to {@code blobReadChannel.read(ByteBuffer)}
1426-
* may throw {@link StorageException}.
1427-
*
1428-
* <p>The {@link BlobSourceOption#generationMatch()} and
1429-
* {@link BlobSourceOption#generationMatch(long)} options can be used to ensure that
1430-
* {@code blobReadChannel.read(ByteBuffer)} calls will throw {@link StorageException} if the
1431-
* blob`s generation differs from the expected one.
1415+
* Return a channel for reading the blob's content.
14321416
*
14331417
* @throws StorageException upon failure
14341418
*/

‎branches/pubsub-alpha/gcloud-java-storage/src/test/java/com/google/gcloud/storage/BlobReadChannelImplTest.java‎

Lines changed: 11 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919
import static org.easymock.EasyMock.anyObject;
2020
import static org.easymock.EasyMock.createMock;
2121
import static org.easymock.EasyMock.expect;
22+
import static org.easymock.EasyMock.expectLastCall;
2223
import static org.easymock.EasyMock.replay;
2324
import static org.easymock.EasyMock.verify;
2425
import static org.junit.Assert.assertArrayEquals;
@@ -45,7 +46,7 @@ public class BlobReadChannelImplTest {
4546

4647
private static final String BUCKET_NAME = "b";
4748
private static final String BLOB_NAME = "n";
48-
private static final BlobId BLOB_ID = BlobId.of(BUCKET_NAME, BLOB_NAME, -1L);
49+
private static final BlobId BLOB_ID = BlobId.of(BUCKET_NAME, BLOB_NAME);
4950
private static final Map<StorageRpc.Option, ?> EMPTY_RPC_OPTIONS = ImmutableMap.of();
5051
private static final int DEFAULT_CHUNK_SIZE = 2 * 1024 * 1024;
5152
private static final int CUSTOM_CHUNK_SIZE = 2 * 1024 * 1024;
@@ -87,7 +88,7 @@ public void testReadBuffered() throws IOException {
8788
ByteBuffer firstReadBuffer = ByteBuffer.allocate(42);
8889
ByteBuffer secondReadBuffer = ByteBuffer.allocate(42);
8990
expect(storageRpcMock.read(BLOB_ID.toPb(), EMPTY_RPC_OPTIONS, 0, DEFAULT_CHUNK_SIZE))
90-
.andReturn(StorageRpc.Tuple.of("etag", result));
91+
.andReturn(result);
9192
replay(storageRpcMock);
9293
reader.read(firstReadBuffer);
9394
reader.read(secondReadBuffer);
@@ -106,11 +107,10 @@ public void testReadBig() throws IOException {
106107
byte[] secondResult = randomByteArray(DEFAULT_CHUNK_SIZE);
107108
ByteBuffer firstReadBuffer = ByteBuffer.allocate(DEFAULT_CHUNK_SIZE);
108109
ByteBuffer secondReadBuffer = ByteBuffer.allocate(42);
109-
expect(storageRpcMock.read(BLOB_ID.toPb(), EMPTY_RPC_OPTIONS, 0, DEFAULT_CHUNK_SIZE))
110-
.andReturn(StorageRpc.Tuple.of("etag", firstResult));
111-
expect(storageRpcMock.read(
112-
BLOB_ID.toPb(), EMPTY_RPC_OPTIONS, DEFAULT_CHUNK_SIZE, CUSTOM_CHUNK_SIZE))
113-
.andReturn(StorageRpc.Tuple.of("etag", secondResult));
110+
storageRpcMock.read(BLOB_ID.toPb(), EMPTY_RPC_OPTIONS, 0, DEFAULT_CHUNK_SIZE);
111+
expectLastCall().andReturn(firstResult);
112+
storageRpcMock.read(BLOB_ID.toPb(), EMPTY_RPC_OPTIONS, DEFAULT_CHUNK_SIZE, CUSTOM_CHUNK_SIZE);
113+
expectLastCall().andReturn(secondResult);
114114
replay(storageRpcMock);
115115
reader.read(firstReadBuffer);
116116
reader.read(secondReadBuffer);
@@ -125,7 +125,7 @@ public void testReadFinish() throws IOException {
125125
byte[] result = {};
126126
ByteBuffer readBuffer = ByteBuffer.allocate(DEFAULT_CHUNK_SIZE);
127127
expect(storageRpcMock.read(BLOB_ID.toPb(), EMPTY_RPC_OPTIONS, 0, DEFAULT_CHUNK_SIZE))
128-
.andReturn(StorageRpc.Tuple.of("etag", result));
128+
.andReturn(result);
129129
replay(storageRpcMock);
130130
assertEquals(-1, reader.read(readBuffer));
131131
}
@@ -137,7 +137,7 @@ public void testSeek() throws IOException {
137137
byte[] result = randomByteArray(DEFAULT_CHUNK_SIZE);
138138
ByteBuffer readBuffer = ByteBuffer.allocate(DEFAULT_CHUNK_SIZE);
139139
expect(storageRpcMock.read(BLOB_ID.toPb(), EMPTY_RPC_OPTIONS, 42, DEFAULT_CHUNK_SIZE))
140-
.andReturn(StorageRpc.Tuple.of("etag", result));
140+
.andReturn(result);
141141
replay(storageRpcMock);
142142
reader.read(readBuffer);
143143
assertArrayEquals(result, readBuffer.array());
@@ -166,41 +166,16 @@ public void testReadClosed() {
166166
}
167167
}
168168

169-
@Test
170-
public void testReadGenerationChanged() throws IOException {
171-
BlobId blobId = BlobId.of(BUCKET_NAME, BLOB_NAME);
172-
reader = new BlobReadChannelImpl(options, blobId, EMPTY_RPC_OPTIONS);
173-
byte[] firstResult = randomByteArray(DEFAULT_CHUNK_SIZE);
174-
byte[] secondResult = randomByteArray(DEFAULT_CHUNK_SIZE);
175-
ByteBuffer firstReadBuffer = ByteBuffer.allocate(DEFAULT_CHUNK_SIZE);
176-
ByteBuffer secondReadBuffer = ByteBuffer.allocate(DEFAULT_CHUNK_SIZE);
177-
expect(storageRpcMock.read(blobId.toPb(), EMPTY_RPC_OPTIONS, 0, DEFAULT_CHUNK_SIZE))
178-
.andReturn(StorageRpc.Tuple.of("etag1", firstResult));
179-
expect(storageRpcMock.read(
180-
blobId.toPb(), EMPTY_RPC_OPTIONS, DEFAULT_CHUNK_SIZE, DEFAULT_CHUNK_SIZE))
181-
.andReturn(StorageRpc.Tuple.of("etag2", firstResult));
182-
replay(storageRpcMock);
183-
reader.read(firstReadBuffer);
184-
try {
185-
reader.read(secondReadBuffer);
186-
fail("Expected BlobReadChannel read to throw StorageException");
187-
} catch (StorageException ex) {
188-
StringBuilder messageBuilder = new StringBuilder();
189-
messageBuilder.append("Blob ").append(blobId).append(" was updated while reading");
190-
assertEquals(messageBuilder.toString(), ex.getMessage());
191-
}
192-
}
193-
194169
@Test
195170
public void testSaveAndRestore() throws IOException, ClassNotFoundException {
196171
byte[] firstResult = randomByteArray(DEFAULT_CHUNK_SIZE);
197172
byte[] secondResult = randomByteArray(DEFAULT_CHUNK_SIZE);
198173
ByteBuffer firstReadBuffer = ByteBuffer.allocate(42);
199174
ByteBuffer secondReadBuffer = ByteBuffer.allocate(DEFAULT_CHUNK_SIZE);
200175
expect(storageRpcMock.read(BLOB_ID.toPb(), EMPTY_RPC_OPTIONS, 0, DEFAULT_CHUNK_SIZE))
201-
.andReturn(StorageRpc.Tuple.of("etag", firstResult));
176+
.andReturn(firstResult);
202177
expect(storageRpcMock.read(BLOB_ID.toPb(), EMPTY_RPC_OPTIONS, 42, DEFAULT_CHUNK_SIZE))
203-
.andReturn(StorageRpc.Tuple.of("etag", secondResult));
178+
.andReturn(secondResult);
204179
replay(storageRpcMock);
205180
reader = new BlobReadChannelImpl(options, BLOB_ID, EMPTY_RPC_OPTIONS);
206181
reader.read(firstReadBuffer);

‎branches/pubsub-alpha/gcloud-java-storage/src/test/java/com/google/gcloud/storage/ITStorageTest.java‎

Lines changed: 0 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -49,7 +49,6 @@
4949
import java.util.Iterator;
5050
import java.util.List;
5151
import java.util.Map;
52-
import java.util.Random;
5352
import java.util.concurrent.ExecutionException;
5453
import java.util.concurrent.TimeUnit;
5554
import java.util.concurrent.TimeoutException;
@@ -732,41 +731,6 @@ public void testReadChannelFail() throws IOException {
732731
assertTrue(storage.delete(BUCKET, blobName));
733732
}
734733

735-
@Test
736-
public void testReadChannelFailUpdatedGeneration() throws IOException {
737-
String blobName = "test-read-blob-fail-updated-generation";
738-
BlobInfo blob = BlobInfo.builder(BUCKET, blobName).build();
739-
Random random = new Random();
740-
int chunkSize = 1024;
741-
int blobSize = 2 * chunkSize;
742-
byte[] content = new byte[blobSize];
743-
random.nextBytes(content);
744-
BlobInfo remoteBlob = storage.create(blob, content);
745-
assertNotNull(remoteBlob);
746-
assertEquals(blobSize, (long) remoteBlob.size());
747-
try (BlobReadChannel reader = storage.reader(blob.blobId())) {
748-
reader.chunkSize(chunkSize);
749-
ByteBuffer readBytes = ByteBuffer.allocate(chunkSize);
750-
int numReadBytes = reader.read(readBytes);
751-
assertEquals(chunkSize, numReadBytes);
752-
assertArrayEquals(Arrays.copyOf(content, chunkSize), readBytes.array());
753-
try (BlobWriteChannel writer = storage.writer(blob)) {
754-
byte[] newContent = new byte[blobSize];
755-
random.nextBytes(newContent);
756-
int numWrittenBytes = writer.write(ByteBuffer.wrap(newContent));
757-
assertEquals(blobSize, numWrittenBytes);
758-
}
759-
readBytes = ByteBuffer.allocate(chunkSize);
760-
reader.read(readBytes);
761-
fail("StorageException was expected");
762-
} catch(StorageException ex) {
763-
StringBuilder messageBuilder = new StringBuilder();
764-
messageBuilder.append("Blob ").append(blob.blobId()).append(" was updated while reading");
765-
assertEquals(messageBuilder.toString(), ex.getMessage());
766-
}
767-
assertTrue(storage.delete(BUCKET, blobName));
768-
}
769-
770734
@Test
771735
public void testWriteChannelFail() throws IOException {
772736
String blobName = "test-write-channel-blob-fail";

0 commit comments

Comments
 (0)