Add publisher wrapping when handling async checksums

[haydenbaker]

Motivation and Context

Modifications

Testing

Screenshots (if appropriate)

Types of changes

• Bug fix (non-breaking change which fixes an issue)
• New feature (non-breaking change which adds functionality)

Checklist

• I have read the CONTRIBUTING document
• Local run of mvn install succeeds
• My code follows the code style of this project
• My change requires a change to the Javadoc documentation
• I have updated the Javadoc documentation accordingly
• I have added tests to cover my changes
• All new and existing tests passed
• I have added a changelog entry. Adding a new entry must be accomplished by running the scripts/new-change script and following the instructions. Commit the new file created by the script in .changes/next-release with your changes.
• My change is to implement 1.11 parity feature and I have updated LaunchChangelog

License

• I confirm that this pull request can be released under the Apache 2 license

[millems]

How about:

@SdkInternalApi
public final class ChecksumPublisher implements Publisher<ByteBuffer> {
    private final CompletableFuture<Void> signal = new CompletableFuture<>();
    private final Publisher<ByteBuffer> publisher;
    private final Collection<? extends Checksum> consumers;

    public ChecksumPublisher(Publisher<ByteBuffer> publisher, Collection<? extends Checksum> consumers) {
        this.publisher = publisher;
        this.consumers = consumers;
    }

    @Override
    public void subscribe(Subscriber<? super ByteBuffer> subscriber) {
        publisher.subscribe(new ChecksumSubscriber(subscriber, consumers, signal));
    }

    public CompletableFuture<Void> checksum() {
        return signal;
    }
}

You might also want to add validation that subscribe() is only called once. If it gets called again (allowed for publishers), weird stuff will happen since we're reusing signal.

[millems]

Should we copy the default implementation of update(ByteBuffer) from Java 9+? It honors the bounds of the byte buffer properly. This implementation doesn't seem to reset the buffer when byteBuffer.hasArray() is false, either. Here's a version where I use update(ByteBuffer) from Java 9+ and reset the buffer after using it:

@SdkInternalApi
public final class ChecksumSubscriber extends DelegatingSubscriber<ByteBuffer, ByteBuffer> {
    private final CompletableFuture<Void> signal;
    private final byte[] copyBuffer;
    private final Collection<Checksum> checksums = new ArrayList<>();

    public ChecksumSubscriber(Subscriber<? super ByteBuffer> subscriber, 
                              Collection<? extends Checksum> consumers,
                              CompletableFuture<Void> signal) {
        super(subscriber);

        this.checksums.addAll(consumers);
        this.signal = signal;
    }

    @Override
    public void onNext(ByteBuffer byteBuffer) {
        updateChecksumsAndReset(byteBuffer);
        subscriber.onNext(byteBuffer);
    }

    private void updateChecksumsAndReset(ByteBuffer buffer) {
        int position = buffer.position();
        int limit = buffer.limit();
        int remaining = limit - position;
        if (remaining <= 0) {
            return;
        }
        if (buffer.hasArray()) {
            checksums.forEach(c -> c.update(buffer.array(), position + buffer.arrayOffset(), remaining));
        } else {
            if (copyBuffer == null) {
                copyBuffer = new byte[4096];
            }
            while (buffer.hasRemaining()) {
                int length = Math.min(buffer.remaining(), copyBuffer.length);
                buffer.get(copyBuffer, 0, length);
                checksums.forEach(c -> c.update(copyBuffer, 0, length));
            }
        }
        buffer.position(position);
        buffer.limit(limit);
    }

    @Override
    public void onError(Throwable t) {
        super.onError(t);
        signal.completeExceptionally(t);
    }

    @Override
    public void onComplete() {
        super.onComplete();
        signal.complete(null);
    }
}

[haydenbaker]

Closed in favor of #4567