Skip to content

Commit fb7eea9

Browse files
committed
Merge branch '5.3.x'
2 parents c1856d2 + c99210c commit fb7eea9

File tree

4 files changed

+74
-0
lines changed

4 files changed

+74
-0
lines changed

spring-core/src/main/java/org/springframework/core/io/buffer/DataBufferUtils.java

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -47,6 +47,7 @@
4747
import reactor.core.publisher.FluxSink;
4848
import reactor.core.publisher.Mono;
4949
import reactor.core.publisher.SynchronousSink;
50+
import reactor.util.context.Context;
5051

5152
import org.springframework.core.io.Resource;
5253
import org.springframework.lang.Nullable;
@@ -1057,6 +1058,12 @@ protected void hookOnError(Throwable throwable) {
10571058
protected void hookOnComplete() {
10581059
this.sink.complete();
10591060
}
1061+
1062+
@Override
1063+
public Context currentContext() {
1064+
return this.sink.currentContext();
1065+
}
1066+
10601067
}
10611068

10621069

@@ -1148,6 +1155,12 @@ private void sinkDataBuffer() {
11481155
this.sink.next(dataBuffer);
11491156
this.dataBuffer.set(null);
11501157
}
1158+
1159+
@Override
1160+
public Context currentContext() {
1161+
return this.sink.currentContext();
1162+
}
1163+
11511164
}
11521165

11531166
}

spring-core/src/test/java/org/springframework/core/io/buffer/DataBufferUtilsTests.java

Lines changed: 49 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
import java.nio.channels.CompletionHandler;
2525
import java.nio.channels.FileChannel;
2626
import java.nio.channels.ReadableByteChannel;
27+
import java.nio.channels.SeekableByteChannel;
2728
import java.nio.channels.WritableByteChannel;
2829
import java.nio.charset.StandardCharsets;
2930
import java.nio.file.Files;
@@ -43,6 +44,7 @@
4344
import reactor.core.publisher.Flux;
4445
import reactor.core.publisher.Mono;
4546
import reactor.test.StepVerifier;
47+
import reactor.util.context.Context;
4648

4749
import org.springframework.core.io.ByteArrayResource;
4850
import org.springframework.core.io.ClassPathResource;
@@ -940,6 +942,53 @@ void matcher3(String displayName, DataBufferFactory bufferFactory) {
940942
release(foo);
941943
}
942944

945+
@ParameterizedDataBufferAllocatingTest
946+
void propagateContextByteChannel(String displayName, DataBufferFactory bufferFactory) throws IOException {
947+
Path path = Paths.get(this.resource.getURI());
948+
try (SeekableByteChannel out = Files.newByteChannel(this.tempFile, StandardOpenOption.WRITE, StandardOpenOption.TRUNCATE_EXISTING)) {
949+
Flux<DataBuffer> result = DataBufferUtils.read(path, bufferFactory, 1024, StandardOpenOption.READ)
950+
.transformDeferredContextual((f, ctx) -> {
951+
assertThat(ctx.getOrDefault("key", "EMPTY")).isEqualTo("TEST");
952+
return f;
953+
})
954+
.transform(f -> DataBufferUtils.write(f, out))
955+
.transformDeferredContextual((f, ctx) -> {
956+
assertThat(ctx.getOrDefault("key", "EMPTY")).isEqualTo("TEST");
957+
return f;
958+
})
959+
.contextWrite(Context.of("key", "TEST"));
960+
961+
StepVerifier.create(result)
962+
.consumeNextWith(DataBufferUtils::release)
963+
.verifyComplete();
964+
965+
966+
}
967+
}
968+
969+
@ParameterizedDataBufferAllocatingTest
970+
void propagateContextAsynchronousFileChannel(String displayName, DataBufferFactory bufferFactory) throws IOException {
971+
Path path = Paths.get(this.resource.getURI());
972+
try (AsynchronousFileChannel out = AsynchronousFileChannel.open(this.tempFile, StandardOpenOption.WRITE, StandardOpenOption.TRUNCATE_EXISTING)) {
973+
Flux<DataBuffer> result = DataBufferUtils.read(path, bufferFactory, 1024, StandardOpenOption.READ)
974+
.transformDeferredContextual((f, ctx) -> {
975+
assertThat(ctx.getOrDefault("key", "EMPTY")).isEqualTo("TEST");
976+
return f;
977+
})
978+
.transform(f -> DataBufferUtils.write(f, out))
979+
.transformDeferredContextual((f, ctx) -> {
980+
assertThat(ctx.getOrDefault("key", "EMPTY")).isEqualTo("TEST");
981+
return f;
982+
})
983+
.contextWrite(Context.of("key", "TEST"));
984+
985+
StepVerifier.create(result)
986+
.consumeNextWith(DataBufferUtils::release)
987+
.verifyComplete();
988+
989+
990+
}
991+
}
943992

944993
private static class ZeroDemandSubscriber extends BaseSubscriber<DataBuffer> {
945994

spring-web/src/main/java/org/springframework/http/codec/multipart/MultipartParser.java

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@
2929
import reactor.core.publisher.BaseSubscriber;
3030
import reactor.core.publisher.Flux;
3131
import reactor.core.publisher.FluxSink;
32+
import reactor.util.context.Context;
3233

3334
import org.springframework.core.codec.DecodingException;
3435
import org.springframework.core.io.buffer.DataBuffer;
@@ -98,6 +99,11 @@ public static Flux<Token> parse(Flux<DataBuffer> buffers, byte[] boundary, int m
9899
});
99100
}
100101

102+
@Override
103+
public Context currentContext() {
104+
return this.sink.currentContext();
105+
}
106+
101107
@Override
102108
protected void hookOnSubscribe(Subscription subscription) {
103109
requestBuffer();

spring-web/src/main/java/org/springframework/http/codec/multipart/PartGenerator.java

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,7 @@
4242
import reactor.core.publisher.FluxSink;
4343
import reactor.core.publisher.Mono;
4444
import reactor.core.scheduler.Scheduler;
45+
import reactor.util.context.Context;
4546

4647
import org.springframework.core.codec.DecodingException;
4748
import org.springframework.core.io.buffer.DataBuffer;
@@ -113,6 +114,11 @@ public static Flux<Part> createParts(Flux<MultipartParser.Token> tokens, int max
113114
});
114115
}
115116

117+
@Override
118+
public Context currentContext() {
119+
return this.sink.currentContext();
120+
}
121+
116122
@Override
117123
protected void hookOnSubscribe(Subscription subscription) {
118124
requestToken();

0 commit comments

Comments
 (0)