-
Notifications
You must be signed in to change notification settings - Fork 196
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
AwsSdk2Transport implementation (#177)
* Implement AwsSdk2Transport Implementation of automatic request and response compression. Integration test Make sure bulk requests work Properly parse 403 errors from OpenSearch service (they don't follow OS format) Ensure that every transport error with a status code is reported as an OpenSearchError Signed-off-by: Matt Timmermans <mtimmermans@tripadvisor.com> * Fix license headers on new files Signed-off-by: Matt Timmermans <mtimmermans@tripadvisor.com> * New AWS sdk release isn't all in the cache yet Signed-off-by: Matt Timmermans <mtimmermans@tripadvisor.com> * Fix license headers in new file Signed-off-by: Matt Timmermans <mtimmermans@tripadvisor.com> Co-authored-by: Matt Timmermans <mtimmermans@tripadvisor.com>
- Loading branch information
1 parent
f8d9a5e
commit 2336053
Showing
12 changed files
with
1,791 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
43 changes: 43 additions & 0 deletions
43
...ent/src/main/java/org/opensearch/client/transport/aws/AsyncByteArrayContentPublisher.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,43 @@ | ||
| /* | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| * | ||
| * The OpenSearch Contributors require contributions made to | ||
| * this file be licensed under the Apache-2.0 license or a | ||
| * compatible open source license. | ||
| */ | ||
|
|
||
| package org.opensearch.client.transport.aws; | ||
|
|
||
| import org.reactivestreams.Subscriber; | ||
| import software.amazon.awssdk.core.async.AsyncRequestBody; | ||
| import software.amazon.awssdk.http.async.SdkHttpContentPublisher; | ||
|
|
||
| import javax.annotation.CheckForNull; | ||
| import java.nio.ByteBuffer; | ||
| import java.util.Optional; | ||
|
|
||
| /** | ||
| * An implementation of AWS {@SdkHttpContentPublisher} that transfers a pre-existing | ||
| * byte array | ||
| */ | ||
| class AsyncByteArrayContentPublisher implements SdkHttpContentPublisher { | ||
| private final AsyncRequestBody delegate; | ||
|
|
||
| AsyncByteArrayContentPublisher(@CheckForNull byte[] data) { | ||
| if (data == null) { | ||
| delegate = AsyncRequestBody.empty(); | ||
| } else { | ||
| delegate = AsyncRequestBody.fromBytes(data); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public Optional<Long> contentLength() { | ||
| return delegate.contentLength(); | ||
| } | ||
|
|
||
| @Override | ||
| public void subscribe(Subscriber<? super ByteBuffer> s) { | ||
| delegate.subscribe(s); | ||
| } | ||
| } |
56 changes: 56 additions & 0 deletions
56
...ient/src/main/java/org/opensearch/client/transport/aws/AsyncCapturingResponseHandler.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,56 @@ | ||
| /* | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| * | ||
| * The OpenSearch Contributors require contributions made to | ||
| * this file be licensed under the Apache-2.0 license or a | ||
| * compatible open source license. | ||
| */ | ||
|
|
||
| package org.opensearch.client.transport.aws; | ||
|
|
||
| import org.reactivestreams.Publisher; | ||
| import software.amazon.awssdk.http.SdkHttpResponse; | ||
| import software.amazon.awssdk.http.async.SdkAsyncHttpResponseHandler; | ||
|
|
||
| import java.nio.ByteBuffer; | ||
| import java.util.concurrent.CompletableFuture; | ||
| import java.util.concurrent.atomic.AtomicBoolean; | ||
|
|
||
| /** | ||
| * An implementation of AWS {@link SdkAsyncHttpResponseHandler} that captures the response, | ||
| * and the content as a byte array. | ||
| */ | ||
| final class AsyncCapturingResponseHandler implements SdkAsyncHttpResponseHandler { | ||
| private final CompletableFuture<SdkHttpResponse> responseFuture; | ||
| private final AsyncCapturingSubscriber bodySubscriber = new AsyncCapturingSubscriber(); | ||
| private final AtomicBoolean subscribed = new AtomicBoolean(false); | ||
|
|
||
| AsyncCapturingResponseHandler() { | ||
| responseFuture = new CompletableFuture<>(); | ||
| } | ||
|
|
||
| public CompletableFuture<SdkHttpResponse> getHeaderPromise() { | ||
| return responseFuture; | ||
| } | ||
|
|
||
| public CompletableFuture<byte[]> getBodyPromise() { | ||
| return bodySubscriber.getPromise(); | ||
| } | ||
|
|
||
| @Override | ||
| public void onHeaders(SdkHttpResponse response) { | ||
| responseFuture.complete(response); | ||
| } | ||
|
|
||
| @Override | ||
| public void onStream(Publisher<ByteBuffer> publisher) { | ||
| if (!subscribed.getAndSet(true)) { | ||
| publisher.subscribe(bodySubscriber); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public void onError(Throwable e) { | ||
| responseFuture.completeExceptionally(e); | ||
| } | ||
| } |
71 changes: 71 additions & 0 deletions
71
java-client/src/main/java/org/opensearch/client/transport/aws/AsyncCapturingSubscriber.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,71 @@ | ||
| /* | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| * | ||
| * The OpenSearch Contributors require contributions made to | ||
| * this file be licensed under the Apache-2.0 license or a | ||
| * compatible open source license. | ||
| */ | ||
|
|
||
| package org.opensearch.client.transport.aws; | ||
|
|
||
| import org.reactivestreams.Subscriber; | ||
| import org.reactivestreams.Subscription; | ||
|
|
||
| import java.io.ByteArrayOutputStream; | ||
| import java.nio.ByteBuffer; | ||
| import java.util.concurrent.CompletableFuture; | ||
|
|
||
| /** | ||
| * A reactive subscriber that captures a byte stream into a byte array | ||
| */ | ||
| class AsyncCapturingSubscriber implements Subscriber<ByteBuffer> { | ||
| private final ByteArrayOutputStream buffer; | ||
| private final CompletableFuture<byte[]> promise; | ||
| private Subscription subscription; | ||
|
|
||
| AsyncCapturingSubscriber() { | ||
| buffer = new ByteArrayOutputStream(); | ||
| promise = new CompletableFuture<>(); | ||
| } | ||
|
|
||
| public CompletableFuture<byte[]> getPromise() { | ||
| return promise; | ||
| } | ||
|
|
||
| @Override | ||
| public void onSubscribe(Subscription subscription) { | ||
| this.subscription = subscription; | ||
| subscription.request(Long.MAX_VALUE); | ||
| } | ||
|
|
||
| @Override | ||
| public void onNext(ByteBuffer buf) { | ||
| try { | ||
| if (buf != null && buf.remaining() > 0) { | ||
| if (buf.hasArray()) { | ||
| buffer.write(buf.array(), buf.arrayOffset() + buf.position(), buf.remaining()); | ||
| } else { | ||
| byte[] data = new byte[buf.remaining()]; | ||
| buf.asReadOnlyBuffer().get(data); | ||
| buffer.write(data); | ||
| } | ||
| } | ||
| this.subscription.request(1); | ||
| } catch (Throwable e) { | ||
| promise.completeExceptionally(e); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public void onError(Throwable e) { | ||
| if (e == null) { | ||
| e = new IllegalArgumentException("Subscriber.onError called with null paramter"); | ||
| } | ||
| promise.completeExceptionally(e); | ||
| } | ||
|
|
||
| @Override | ||
| public void onComplete() { | ||
| promise.complete(buffer.toByteArray()); | ||
| } | ||
| } |
Oops, something went wrong.