Skip to content
This repository has been archived by the owner on Dec 27, 2023. It is now read-only.

Probably better fix for conda s3 bug #505

Open
wants to merge 2 commits into
base: master
Choose a base branch
from
Open
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
15 changes: 0 additions & 15 deletions .github/workflows/pdd.yml

This file was deleted.

Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,6 @@
import com.artipie.asto.Content;
import com.artipie.asto.Key;
import com.artipie.asto.Storage;
import com.artipie.asto.misc.UncheckedIOConsumer;
import com.artipie.asto.misc.UncheckedIOSupplier;
import com.artipie.asto.misc.UncheckedRunnable;
import io.reactivex.Flowable;
Expand All @@ -33,6 +32,7 @@
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
import org.apache.commons.lang3.tuple.ImmutablePair;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;

Expand Down Expand Up @@ -113,6 +113,7 @@ public CompletionStage<Void> process(
* output stream.
* @return Completion action with the result
* @throws ArtipieIOException On Error
* @checkstyle ExecutableStatementCountCheck (100 lines)
*/
public CompletionStage<R> processWithResult(
final BiFunction<Optional<InputStream>, OutputStream, R> action
Expand All @@ -135,18 +136,45 @@ public CompletionStage<R> processWithResult(
}
return stage;
}
).thenCompose(
optional -> {
try (PublishingOutputStream output = new PublishingOutputStream()) {
res.set(action.apply(optional, output));
return this.asto.save(this.write, new Content.From(output.publisher()));
} catch (final IOException err) {
throw new ArtipieIOException(err);
} finally {
optional.ifPresent(new UncheckedIOConsumer<>(InputStream::close));
}
)
.thenCompose(
input -> {
final PublishingOutputStream output = new PublishingOutputStream();
return CompletableFuture.runAsync(() -> res.set(action.apply(input, output)))
.thenApply(unused -> new ImmutablePair<>(input, output));
}
).thenApply(nothing -> res.get());
)
.thenCompose(
streams -> this.asto.save(
this.write, new Content.From(streams.getRight().publisher())
).thenApply(unused -> streams)
)
.handle(
(streams, throwable) -> {
Throwable last = throwable;
try {
if (streams.getLeft().isPresent()) {
streams.getLeft().get().close();
}
} catch (final IOException ex) {
if (last != null) {
ex.addSuppressed(last);
}
last = ex;
}
try {
streams.getRight().close();
} catch (final IOException ex) {
if (last != null) {
ex.addSuppressed(last);
}
last = ex;
}
if (last != null) {
throw new ArtipieIOException(last);
}
return res.get();
});
}

/**
Expand Down
Loading