|
18 | 18 |
|
19 | 19 | import com.google.api.services.storage.model.StorageObject; |
20 | 20 | import com.google.cloud.storage.Conversions.Decoder; |
| 21 | +import com.google.cloud.storage.Crc32cValue.Crc32cLengthKnown; |
| 22 | +import com.google.cloud.storage.HttpContentRange.HasRange; |
21 | 23 | import com.google.cloud.storage.Retrying.RetrierWithAlg; |
22 | 24 | import com.google.cloud.storage.spi.v1.HttpRpcContext; |
23 | 25 | import com.google.cloud.storage.spi.v1.HttpStorageRpc; |
24 | 26 | import io.opencensus.trace.EndSpanOptions; |
| 27 | +import java.nio.ByteBuffer; |
| 28 | +import java.nio.channels.GatheringByteChannel; |
25 | 29 | import java.util.Locale; |
26 | 30 | import java.util.concurrent.atomic.AtomicBoolean; |
27 | 31 | import org.checkerframework.checker.nullness.qual.Nullable; |
@@ -54,28 +58,118 @@ final class JsonResumableSession { |
54 | 58 |
|
55 | 59 | ResumableOperationResult<@Nullable StorageObject> put( |
56 | 60 | RewindableContent content, HttpContentRange contentRange) { |
| 61 | + Crc32cValue<?> crc32cSoFar = resumableWrite.getCumulativeCrc32c(); |
| 62 | + @Nullable Crc32cValue<?> nextCumulativeCrc32c = |
| 63 | + resumableWrite.getHasher().nullSafeConcat(crc32cSoFar, content.getCrc32c()); |
| 64 | + @Nullable Crc32cValue<?> finalChecksum = |
| 65 | + contentRange.isFinalizing() ? nextCumulativeCrc32c : null; |
57 | 66 | JsonResumableSessionPutTask task = |
58 | | - new JsonResumableSessionPutTask(context, resumableWrite, content, contentRange); |
| 67 | + new JsonResumableSessionPutTask( |
| 68 | + context, resumableWrite, content, contentRange, finalChecksum); |
59 | 69 | HttpRpcContext httpRpcContext = HttpRpcContext.getInstance(); |
60 | 70 | try { |
61 | 71 | httpRpcContext.newInvocationId(); |
62 | 72 | AtomicBoolean dirty = new AtomicBoolean(false); |
63 | | - return retrier.run( |
64 | | - () -> { |
65 | | - if (dirty.getAndSet(true)) { |
66 | | - ResumableOperationResult<@Nullable StorageObject> query = query(); |
67 | | - long persistedSize = query.getPersistedSize(); |
68 | | - if (contentRange.endOffsetEquals(persistedSize) || query.getObject() != null) { |
69 | | - return query; |
70 | | - } else { |
71 | | - task.rewindTo(persistedSize); |
72 | | - } |
73 | | - } |
74 | | - return task.call(); |
75 | | - }, |
76 | | - Decoder.identity()); |
| 73 | + ResumableOperationResult<@Nullable StorageObject> result = |
| 74 | + retrier.run( |
| 75 | + () -> { |
| 76 | + if (dirty.getAndSet(true)) { |
| 77 | + ResumableOperationResult<@Nullable StorageObject> query = query(); |
| 78 | + long persistedSize = query.getPersistedSize(); |
| 79 | + if (contentRange.endOffsetEquals(persistedSize) || query.getObject() != null) { |
| 80 | + return query; |
| 81 | + } else { |
| 82 | + task.rewindTo(persistedSize); |
| 83 | + } |
| 84 | + } |
| 85 | + return task.call(); |
| 86 | + }, |
| 87 | + Decoder.identity()); |
| 88 | + |
| 89 | + if (nextCumulativeCrc32c != null) { |
| 90 | + long persistedSize = result.getPersistedSize(); |
| 91 | + if (contentRange.endOffsetEquals(persistedSize) || result.getObject() != null) { |
| 92 | + resumableWrite.setCumulativeCrc32c(nextCumulativeCrc32c); |
| 93 | + } else if (contentRange instanceof HasRange) { |
| 94 | + ByteRangeSpec range = ((HasRange<?>) contentRange).range(); |
| 95 | + content.rewindTo(0); |
| 96 | + long serverConsumedBytes = persistedSize - range.beginOffset(); |
| 97 | + try (HashingGatheringByteChannel hashingChannel = |
| 98 | + new HashingGatheringByteChannel(serverConsumedBytes)) { |
| 99 | + StorageException.wrapIOException(() -> content.writeTo(hashingChannel)); |
| 100 | + resumableWrite.setCumulativeCrc32c( |
| 101 | + resumableWrite.getHasher().nullSafeConcat(crc32cSoFar, hashingChannel.cumulative)); |
| 102 | + } |
| 103 | + } else { |
| 104 | + throw new StorageException( |
| 105 | + 0, |
| 106 | + String.format( |
| 107 | + Locale.US, |
| 108 | + "Result persistedSize (%d) did not match expected end of contentRange (%s) and contentRange does not have range to allow automatic recovery", |
| 109 | + persistedSize, |
| 110 | + contentRange)); |
| 111 | + } |
| 112 | + } |
| 113 | + return result; |
77 | 114 | } finally { |
78 | 115 | httpRpcContext.clearInvocationId(); |
79 | 116 | } |
80 | 117 | } |
| 118 | + |
| 119 | + private static final class HashingGatheringByteChannel implements GatheringByteChannel { |
| 120 | + private final long maxBytesToConsume; |
| 121 | + |
| 122 | + private Crc32cLengthKnown cumulative; |
| 123 | + |
| 124 | + private HashingGatheringByteChannel(long maxBytesToConsume) { |
| 125 | + this.maxBytesToConsume = maxBytesToConsume; |
| 126 | + this.cumulative = Crc32cValue.zero(); |
| 127 | + } |
| 128 | + |
| 129 | + @Override |
| 130 | + public int write(ByteBuffer src) { |
| 131 | + return Math.toIntExact(write(new ByteBuffer[] {src}, 0, 1)); |
| 132 | + } |
| 133 | + |
| 134 | + @Override |
| 135 | + public long write(ByteBuffer[] srcs) { |
| 136 | + return write(srcs, 0, srcs.length); |
| 137 | + } |
| 138 | + |
| 139 | + @Override |
| 140 | + public long write(ByteBuffer[] srcs, int offset, int length) { |
| 141 | + Crc32cLengthKnown cum = Crc32cValue.zero(); |
| 142 | + for (int i = offset; i < length; i++) { |
| 143 | + long toConsume = maxBytesToConsume - cum.getLength(); |
| 144 | + if (toConsume <= 0) { |
| 145 | + if (cum.getLength() == 0) { |
| 146 | + return -1; |
| 147 | + } else { |
| 148 | + break; |
| 149 | + } |
| 150 | + } |
| 151 | + |
| 152 | + ByteBuffer buf = srcs[i]; |
| 153 | + if (buf.remaining() <= toConsume) { |
| 154 | + cum = cum.concat(Hasher.enabled().hash(buf)); |
| 155 | + } else { |
| 156 | + ByteBuffer slice = buf.slice(); |
| 157 | + int limit = Math.toIntExact(toConsume); |
| 158 | + slice.limit(limit); |
| 159 | + cum = cum.concat(Hasher.enabled().hash(slice)); |
| 160 | + buf.position(buf.position() + limit); |
| 161 | + } |
| 162 | + } |
| 163 | + cumulative = cumulative.concat(cum); |
| 164 | + return cum.getLength(); |
| 165 | + } |
| 166 | + |
| 167 | + @Override |
| 168 | + public boolean isOpen() { |
| 169 | + return true; |
| 170 | + } |
| 171 | + |
| 172 | + @Override |
| 173 | + public void close() {} |
| 174 | + } |
81 | 175 | } |
0 commit comments