Skip to content

Ensure Sink.contextView is propagated #1450

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 10 commits into from
Aug 1, 2024
Merged

Ensure Sink.contextView is propagated #1450

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
30f3de9
Ensure Sink.contextView is propagated
rozza Jul 15, 2024
bce9f74
Code review updates
rozza Jul 25, 2024
12083ad
Code review explicitly set the context
rozza Jul 25, 2024
905202f
Merge branch 'master' into JAVA-5345
rozza Jul 25, 2024
96da0b8
Replace use of Mono.create in GridFSUploadPublisherImpl
rozza Jul 26, 2024
55cdbf7
Remove nested subscription from within Mono.
rozza Jul 26, 2024
57dbf6a
Pass context to cancellation mono
rozza Jul 26, 2024
2d0db20
Update context propagation in BatchCursorFlux
rozza Jul 26, 2024
38109c9
Unify ordering in Crypt.java
rozza Jul 26, 2024
cc75540
More idiomatic change of empty mono type
rozza Jul 29, 2024
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
1 change: 1 addition & 0 deletions ...ve-streams/src/main/com/mongodb/reactivestreams/client/internal/BatchCursorPublisher.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -106,6 +106,7 @@ public Publisher<T> first() {
.flatMap(batchCursor -> Mono.create(sink -> {
batchCursor.setBatchSize(1);
Mono.from(batchCursor.next())
.contextWrite(sink.contextView())
.doOnTerminate(batchCursor::close)
.doOnError(sink::error)
.doOnSuccess(results -> {
Expand Down
7 changes: 7 additions & 0 deletions ...rc/main/com/mongodb/reactivestreams/client/internal/gridfs/GridFSUploadPublisherImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -103,21 +103,25 @@ public void subscribe(final Subscriber<? super Void> s) {
sink.onCancel(() -> createCancellationMono(terminated).subscribe());

Consumer<Throwable> errorHandler = e -> createCancellationMono(terminated)
.contextWrite(sink.contextView())
.doOnError(i -> sink.error(e))
.doOnSuccess(i -> sink.error(e))
.subscribe();

Consumer<Long> saveFileDataMono = l -> createSaveFileDataMono(terminated, l)
.contextWrite(sink.contextView())
.doOnError(errorHandler)
.doOnSuccess(i -> sink.success())
.subscribe();

Consumer<Void> saveChunksMono = i -> createSaveChunksMono(terminated)
.contextWrite(sink.contextView())
.doOnError(errorHandler)
.doOnSuccess(saveFileDataMono)
.subscribe();

createCheckAndCreateIndexesMono()
.contextWrite(sink.contextView())
.doOnError(errorHandler)
.doOnSuccess(saveChunksMono)
.subscribe();
Expand Down Expand Up @@ -159,6 +163,7 @@ private Mono<Void> createCheckAndCreateIndexesMono() {
AtomicBoolean collectionExists = new AtomicBoolean(false);

return Mono.create(sink -> Mono.from(findPublisher.projection(PROJECTION).first())
.contextWrite(sink.contextView())
.subscribe(
d -> collectionExists.set(true),
sink::error,
Expand All @@ -167,6 +172,7 @@ private Mono<Void> createCheckAndCreateIndexesMono() {
sink.success();
} else {
checkAndCreateIndex(filesCollection.withReadPreference(primary()), FILES_INDEX)
.contextWrite(sink.contextView())
.doOnError(sink::error)
.doOnSuccess(i -> {
checkAndCreateIndex(chunksCollection.withReadPreference(primary()), CHUNKS_INDEX)
Expand Down Expand Up @@ -227,6 +233,7 @@ private Mono<Long> createSaveChunksMono(final AtomicBoolean terminated) {
AtomicLong lengthInBytes = new AtomicLong(0);
AtomicInteger chunkIndex = new AtomicInteger(0);
new ResizingByteBufferFlux(source, chunkSizeBytes)
.contextWrite(sink.contextView())
.flatMap((Function<ByteBuffer, Publisher<InsertOneResult>>) byteBuffer -> {
if (terminated.get()) {
return Mono.empty();
Expand Down