Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -645,6 +645,9 @@ public Blob compose(ComposeRequest composeRequest) {
.forEach(builder::addSourceObjects);
final Object target = codecs.blobInfo().encode(composeRequest.getTarget());
builder.setDestination(target);
if (composeRequest.deleteSourceObjects()) {
builder.setDeleteSourceObjects(true);
}
ComposeObjectRequest req = opts.composeObjectsRequest().apply(builder).build();
GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext());
return retrier.run(
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -3185,6 +3185,7 @@ class ComposeRequest implements Serializable {
private final List<SourceBlob> sourceBlobs;
private final BlobInfo target;
private final List<BlobTargetOption> targetOptions;
private final boolean deleteSourceObjects;

private transient Opts<ObjectTargetOpt> targetOpts;

Expand Down Expand Up @@ -3222,6 +3223,7 @@ public static class Builder {
private final Set<BlobTargetOption> targetOptions = new LinkedHashSet<>();
private BlobInfo target;
private Opts<ObjectTargetOpt> opts = Opts.empty();
private boolean deleteSourceObjects;

/** Add source blobs for compose operation. */
public Builder addSource(Iterable<String> blobs) {
Expand Down Expand Up @@ -3265,6 +3267,16 @@ public Builder setTargetOptions(Iterable<BlobTargetOption> options) {
return this;
}

/**
* Sets whether to delete the source objects after the compose operation.
*
* @return the builder
*/
public Builder setDeleteSourceObjects(boolean deleteSourceObjects) {
this.deleteSourceObjects = deleteSourceObjects;
return this;
}

/** Creates a {@code ComposeRequest} object. */
public ComposeRequest build() {
checkArgument(!sourceBlobs.isEmpty());
Expand All @@ -3280,6 +3292,7 @@ private ComposeRequest(Builder builder) {
// keep targetOptions for serialization even though we will read targetOpts
targetOptions = ImmutableList.copyOf(builder.targetOptions);
targetOpts = builder.opts.prepend(Opts.unwrap(targetOptions).resolveFrom(target));
deleteSourceObjects = builder.deleteSourceObjects;
}

/** Returns compose operation's source blobs. */
Expand All @@ -3297,6 +3310,11 @@ public List<BlobTargetOption> getTargetOptions() {
return targetOptions;
}

/** Returns whether to delete the source objects after the compose operation. */
public boolean deleteSourceObjects() {
return deleteSourceObjects;
}

@InternalApi
Opts<ObjectTargetOpt> getTargetOpts() {
return targetOpts;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -651,7 +651,14 @@ public Blob compose(final ComposeRequest composeRequest) {
}
Opts<ObjectTargetOpt> targetOpts = composeRequest.getTargetOpts();
StorageObject targetPb = codecs.blobInfo().encode(composeRequest.getTarget());
Map<StorageRpc.Option, ?> targetOptions = targetOpts.getRpcOptions();
final Map<StorageRpc.Option, ?> targetOptions;
if (composeRequest.deleteSourceObjects()) {
Map<StorageRpc.Option, Object> mutableOptions = new HashMap<>(targetOpts.getRpcOptions());
mutableOptions.put(StorageRpc.Option.DELETE_SOURCE_OBJECTS, true);
targetOptions = Collections.unmodifiableMap(mutableOptions);
} else {
targetOptions = targetOpts.getRpcOptions();
}
ResultRetryAlgorithm<?> algorithm =
retryAlgorithmManager.getForObjectsCompose(sources, targetPb, targetOptions);
return run(
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -817,6 +817,9 @@ public StorageObject compose(
sourceObjects.add(sourceObject);
}
request.setSourceObjects(sourceObjects);
if (Boolean.TRUE.equals(targetOptions.get(Option.DELETE_SOURCE_OBJECTS))) {
request.setDeleteSourceObjects(true);
}
Span span = startSpan(HttpStorageRpcSpans.SPAN_NAME_COMPOSE);
Scope scope = tracer.withSpan(span);
try {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -82,6 +82,7 @@ enum Option {
INCLUDE_TRAILING_DELIMITER("includeTrailingDelimiter"),
X_UPLOAD_CONTENT_LENGTH("x-upload-content-length"),
OBJECT_FILTER("objectFilter"),
DELETE_SOURCE_OBJECTS("deleteSourceObjects"),
/**
* An {@link com.google.common.collect.ImmutableMap ImmutableMap&lt;String, String>} of values
* which will be set as additional headers on the request.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -361,6 +361,20 @@ public void composeRequest() throws IOException, ClassNotFoundException {
}
}

@Test
public void testComposeRequestSerialization() throws Exception {
Storage.ComposeRequest request =
Storage.ComposeRequest.newBuilder()
.setTarget(BLOB_INFO)
.addSource("s1", "s2")
.setDeleteSourceObjects(true)
.build();
Storage.ComposeRequest copy = serializeAndDeserialize(request);
assertThat(copy.getTarget()).isEqualTo(request.getTarget());
assertThat(copy.getSourceBlobs().size()).isEqualTo(request.getSourceBlobs().size());
assertThat(copy.deleteSourceObjects()).isTrue();
}

/**
* Here we override the super classes implementation to remove the "assertNotSame".
*
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,7 @@
import java.security.spec.PKCS8EncodedKeySpec;
import java.security.spec.X509EncodedKeySpec;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import javax.crypto.spec.SecretKeySpec;
Expand Down Expand Up @@ -134,6 +135,12 @@ public class StorageImplMockitoTest {
Storage.BlobTargetOption.doesNotExist();
private static final Storage.BlobTargetOption BLOB_TARGET_PREDEFINED_ACL =
Storage.BlobTargetOption.predefinedAcl(Storage.PredefinedAcl.PRIVATE);
private static final Storage.ComposeRequest COMPOSE_REQUEST =
Storage.ComposeRequest.newBuilder()
.setTarget(BLOB_INFO1)
.addSource(BLOB_NAME2, BLOB_NAME3)
.setDeleteSourceObjects(true)
.build();
private static final Map<StorageRpc.Option, ?> BLOB_TARGET_OPTIONS_CREATE =
ImmutableMap.of(
StorageRpc.Option.IF_METAGENERATION_MATCH, BLOB_INFO1.getMetageneration(),
Expand Down Expand Up @@ -1028,6 +1035,55 @@ public void testDeleteNotification() {
assertEquals(isDeleted, Boolean.TRUE);
}

@Test
public void testCompose() {
List<StorageObject> sources =
ImmutableList.of(
Conversions.json()
.blobInfo()
.encode(BlobInfo.newBuilder(BUCKET_NAME1, BLOB_NAME2).build()),
Conversions.json()
.blobInfo()
.encode(BlobInfo.newBuilder(BUCKET_NAME1, BLOB_NAME3).build()));
StorageObject target = Conversions.json().blobInfo().encode(BLOB_INFO1);
Map<StorageRpc.Option, Object> targetOptions = new HashMap<>();
targetOptions.put(StorageRpc.Option.DELETE_SOURCE_OBJECTS, true);
doReturn(target)
.doThrow(UNEXPECTED_CALL_EXCEPTION)
.when(storageRpcMock)
.compose(sources, target, targetOptions);
initializeService();
Blob blob = storage.compose(COMPOSE_REQUEST);
assertEquals(expectedBlob1, blob);
}

@Test
public void testComposeDeleteSourceObjectsFalse() {
List<StorageObject> sources =
ImmutableList.of(
Conversions.json()
.blobInfo()
.encode(BlobInfo.newBuilder(BUCKET_NAME1, BLOB_NAME2).build()),
Conversions.json()
.blobInfo()
.encode(BlobInfo.newBuilder(BUCKET_NAME1, BLOB_NAME3).build()));
StorageObject target = Conversions.json().blobInfo().encode(BLOB_INFO1);
Map<StorageRpc.Option, Object> targetOptions = new HashMap<>();
doReturn(target)
.doThrow(UNEXPECTED_CALL_EXCEPTION)
.when(storageRpcMock)
.compose(sources, target, targetOptions);
initializeService();
Storage.ComposeRequest request =
Storage.ComposeRequest.newBuilder()
.setTarget(BLOB_INFO1)
.addSource(BLOB_NAME2, BLOB_NAME3)
.setDeleteSourceObjects(false)
.build();
Blob blob = storage.compose(request);
assertEquals(expectedBlob1, blob);
}

private void verifyBucketNotification(Notification value) {
assertNull(value.getNotificationId());
assertEquals(CUSTOM_ATTRIBUTES, value.getCustomAttributes());
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -812,6 +812,34 @@ public void testComposeBlobWithContentType() {
assertArrayEquals(composedBytes, readBytes);
}

@Test
public void testComposeBlobDeleteSource() {
String baseName = generator.randomObjectName();
String sourceBlobName1 = baseName + "-1";
String sourceBlobName2 = baseName + "-2";
BlobInfo sourceBlob1 = BlobInfo.newBuilder(bucket, sourceBlobName1).build();
BlobInfo sourceBlob2 = BlobInfo.newBuilder(bucket, sourceBlobName2).build();
storage.create(sourceBlob1, BLOB_BYTE_CONTENT);
storage.create(sourceBlob2, BLOB_BYTE_CONTENT);

String targetBlobName = baseName + "-target";
BlobInfo targetBlob = BlobInfo.newBuilder(bucket, targetBlobName).build();
ComposeRequest req =
ComposeRequest.newBuilder()
.addSource(sourceBlobName1, sourceBlobName2)
.setTarget(targetBlob)
.setDeleteSourceObjects(true)
.build();
Blob remoteTargetBlob = storage.compose(req);
assertNotNull(remoteTargetBlob);

assertNull(storage.get(bucket.getName(), sourceBlobName1));
assertNull(storage.get(bucket.getName(), sourceBlobName2));

byte[] readBytes = storage.readAllBytes(bucket.getName(), targetBlobName);
assertThat(readBytes.length).isEqualTo(BLOB_BYTE_CONTENT.length * 2);
}

@Test
public void testComposeBlobFail() {
String baseName = generator.randomObjectName();
Expand Down
Loading