Adapt to API changes in the latest reactor snapshot

snicoll · snicoll · commit 673f83e388e5 · 2020-08-04T11:41:30.000+02:00
diff --git a/spring-r2dbc/src/test/java/org/springframework/r2dbc/connection/lookup/AbstractRoutingConnectionFactoryUnitTests.java b/spring-r2dbc/src/test/java/org/springframework/r2dbc/connection/lookup/AbstractRoutingConnectionFactoryUnitTests.java
@@ -63,7 +63,7 @@ public void shouldDetermineRoutedFactory() {
 		connectionFactory.afterPropertiesSet();
 
 		connectionFactory.determineTargetConnectionFactory()
-				.subscriberContext(Context.of(ROUTING_KEY, "key"))
+				.contextWrite(Context.of(ROUTING_KEY, "key"))
 				.as(StepVerifier::create)
 				.expectNext(routedConnectionFactory)
 				.verifyComplete();
@@ -109,7 +109,7 @@ public void unresolvableConnectionFactoryRetrievalShouldFail() {
 		connectionFactory.afterPropertiesSet();
 
 		connectionFactory.determineTargetConnectionFactory()
-				.subscriberContext(Context.of(ROUTING_KEY, "unknown"))
+				.contextWrite(Context.of(ROUTING_KEY, "unknown"))
 				.as(StepVerifier::create)
 				.verifyError(IllegalStateException.class);
 	}
@@ -122,7 +122,7 @@ public void connectionFactoryRetrievalWithUnknownLookupKeyShouldReturnDefaultCon
 		connectionFactory.afterPropertiesSet();
 
 		connectionFactory.determineTargetConnectionFactory()
-				.subscriberContext(Context.of(ROUTING_KEY, "unknown"))
+				.contextWrite(Context.of(ROUTING_KEY, "unknown"))
 				.as(StepVerifier::create)
 				.expectNext(defaultConnectionFactory)
 				.verifyComplete();
@@ -153,7 +153,7 @@ public void shouldLookupFromMap() {
 		connectionFactory.afterPropertiesSet();
 
 		connectionFactory.determineTargetConnectionFactory()
-				.subscriberContext(Context.of(ROUTING_KEY, "my-key"))
+				.contextWrite(Context.of(ROUTING_KEY, "my-key"))
 				.as(StepVerifier::create)
 				.expectNext(routedConnectionFactory)
 				.verifyComplete();
@@ -168,7 +168,7 @@ public void shouldAllowModificationsAfterInitialization() {
 		connectionFactory.afterPropertiesSet();
 
 		connectionFactory.determineTargetConnectionFactory()
-				.subscriberContext(Context.of(ROUTING_KEY, "lookup-key"))
+				.contextWrite(Context.of(ROUTING_KEY, "lookup-key"))
 				.as(StepVerifier::create)
 				.expectNext(defaultConnectionFactory)
 				.verifyComplete();
@@ -177,7 +177,7 @@ public void shouldAllowModificationsAfterInitialization() {
 		connectionFactory.afterPropertiesSet();
 
 		connectionFactory.determineTargetConnectionFactory()
-				.subscriberContext(Context.of(ROUTING_KEY, "lookup-key"))
+				.contextWrite(Context.of(ROUTING_KEY, "lookup-key"))
 				.as(StepVerifier::create)
 				.expectNext(routedConnectionFactory)
 				.verifyComplete();
@@ -187,8 +187,12 @@ static class DummyRoutingConnectionFactory extends AbstractRoutingConnectionFact
 
 		@Override
 		protected Mono<Object> determineCurrentLookupKey() {
-			return Mono.subscriberContext().filter(context -> context.hasKey(ROUTING_KEY))
-					.map(context -> context.get(ROUTING_KEY));
+			return Mono.deferContextual(context -> {
+				if (context.hasKey(ROUTING_KEY)) {
+					return Mono.just(context.get(ROUTING_KEY));
+				}
+				return Mono.empty();
+			});
 		}
 	}
 
diff --git a/spring-tx/src/main/java/org/springframework/transaction/interceptor/TransactionAspectSupport.java b/spring-tx/src/main/java/org/springframework/transaction/interceptor/TransactionAspectSupport.java
@@ -887,8 +887,8 @@ public Object invokeWithinTransaction(Method method, @Nullable Class<?> targetCl
 								// target invocation exception
 								return completeTransactionAfterThrowing(it, ex).then(Mono.error(ex));
 							}
-						})).subscriberContext(TransactionContextManager.getOrCreateContext())
-						.subscriberContext(TransactionContextManager.getOrCreateContextHolder());
+						})).contextWrite(TransactionContextManager.getOrCreateContext())
+						.contextWrite(TransactionContextManager.getOrCreateContextHolder());
 			}
 
 			// Any other reactive type, typically a Flux
@@ -917,8 +917,8 @@ public Object invokeWithinTransaction(Method method, @Nullable Class<?> targetCl
 							// target invocation exception
 							return completeTransactionAfterThrowing(it, ex).then(Mono.error(ex));
 						}
-					})).subscriberContext(TransactionContextManager.getOrCreateContext())
-					.subscriberContext(TransactionContextManager.getOrCreateContextHolder()));
+					})).contextWrite(TransactionContextManager.getOrCreateContext())
+					.contextWrite(TransactionContextManager.getOrCreateContextHolder()));
 		}
 
 		@SuppressWarnings("serial")
diff --git a/spring-tx/src/main/java/org/springframework/transaction/reactive/TransactionContextManager.java b/spring-tx/src/main/java/org/springframework/transaction/reactive/TransactionContextManager.java
@@ -50,19 +50,17 @@ private TransactionContextManager() {
 	 * or no context found in a holder
 	 */
 	public static Mono<TransactionContext> currentContext() throws NoTransactionException {
-		return Mono.subscriberContext().handle((ctx, sink) -> {
+		return Mono.deferContextual(ctx -> {
 			if (ctx.hasKey(TransactionContext.class)) {
-				sink.next(ctx.get(TransactionContext.class));
-				return;
+				return Mono.just(ctx.get(TransactionContext.class));
 			}
 			if (ctx.hasKey(TransactionContextHolder.class)) {
 				TransactionContextHolder holder = ctx.get(TransactionContextHolder.class);
 				if (holder.hasContext()) {
-					sink.next(holder.currentContext());
-					return;
+					return Mono.just(holder.currentContext());
 				}
 			}
-			sink.error(new NoTransactionInContextException());
+			return Mono.error(new NoTransactionInContextException());
 		});
 	}
 
diff --git a/spring-tx/src/main/java/org/springframework/transaction/reactive/TransactionalOperatorImpl.java b/spring-tx/src/main/java/org/springframework/transaction/reactive/TransactionalOperatorImpl.java
@@ -82,8 +82,8 @@ public <T> Mono<T> transactional(Mono<T> mono) {
 					this.transactionManager::commit, (res, err) -> Mono.empty(), this.transactionManager::rollback)
 					.onErrorResume(ex -> rollbackOnException(it, ex).then(Mono.error(ex))));
 		})
-		.subscriberContext(TransactionContextManager.getOrCreateContext())
-		.subscriberContext(TransactionContextManager.getOrCreateContextHolder());
+		.contextWrite(TransactionContextManager.getOrCreateContext())
+		.contextWrite(TransactionContextManager.getOrCreateContextHolder());
 	}
 
 	@Override
@@ -104,8 +104,8 @@ public <T> Flux<T> execute(TransactionCallback<T> action) throws TransactionExce
 					.onErrorResume(ex ->
 							rollbackOnException(it, ex).then(Mono.error(ex))));
 		})
-		.subscriberContext(TransactionContextManager.getOrCreateContext())
-		.subscriberContext(TransactionContextManager.getOrCreateContextHolder());
+		.contextWrite(TransactionContextManager.getOrCreateContext())
+		.contextWrite(TransactionContextManager.getOrCreateContextHolder());
 	}
 
 	/**
diff --git a/spring-web/src/main/java/org/springframework/web/filter/reactive/ServerWebExchangeContextFilter.java b/spring-web/src/main/java/org/springframework/web/filter/reactive/ServerWebExchangeContextFilter.java
@@ -47,7 +47,7 @@ public class ServerWebExchangeContextFilter implements WebFilter {
 	@Override
 	public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {
 		return chain.filter(exchange)
-				.subscriberContext(cxt -> cxt.put(EXCHANGE_CONTEXT_ATTRIBUTE, exchange));
+				.contextWrite(cxt -> cxt.put(EXCHANGE_CONTEXT_ATTRIBUTE, exchange));
 	}
 
 

Original file line number	Diff line number	Diff line change
`@@ -887,8 +887,8 @@ public Object invokeWithinTransaction(Method method, @Nullable Class<?> targetCl`
`887`	`887`	`// target invocation exception`
`888`	`888`	`return completeTransactionAfterThrowing(it, ex).then(Mono.error(ex));`
`889`	`889`	`}`
`890`		`- })).subscriberContext(TransactionContextManager.getOrCreateContext())`
`891`		`- .subscriberContext(TransactionContextManager.getOrCreateContextHolder());`
	`890`	`+ })).contextWrite(TransactionContextManager.getOrCreateContext())`
	`891`	`+ .contextWrite(TransactionContextManager.getOrCreateContextHolder());`
`892`	`892`	`}`
`893`	`893`
`894`	`894`	`// Any other reactive type, typically a Flux`
`@@ -917,8 +917,8 @@ public Object invokeWithinTransaction(Method method, @Nullable Class<?> targetCl`
`917`	`917`	`// target invocation exception`
`918`	`918`	`return completeTransactionAfterThrowing(it, ex).then(Mono.error(ex));`
`919`	`919`	`}`
`920`		`- })).subscriberContext(TransactionContextManager.getOrCreateContext())`
`921`		`- .subscriberContext(TransactionContextManager.getOrCreateContextHolder()));`
	`920`	`+ })).contextWrite(TransactionContextManager.getOrCreateContext())`
	`921`	`+ .contextWrite(TransactionContextManager.getOrCreateContextHolder()));`
`922`	`922`	`}`
`923`	`923`
`924`	`924`	`@SuppressWarnings("serial")`
Original file line number	Diff line number	Diff line change
`@@ -50,19 +50,17 @@ private TransactionContextManager() {`
`50`	`50`	`* or no context found in a holder`
`51`	`51`	`*/`
`52`	`52`	`public static Mono<TransactionContext> currentContext() throws NoTransactionException {`
`53`		`- return Mono.subscriberContext().handle((ctx, sink) -> {`
	`53`	`+ return Mono.deferContextual(ctx -> {`
`54`	`54`	`if (ctx.hasKey(TransactionContext.class)) {`
`55`		`- sink.next(ctx.get(TransactionContext.class));`
`56`		`- return;`
	`55`	`+ return Mono.just(ctx.get(TransactionContext.class));`
`57`	`56`	`}`
`58`	`57`	`if (ctx.hasKey(TransactionContextHolder.class)) {`
`59`	`58`	`TransactionContextHolder holder = ctx.get(TransactionContextHolder.class);`
`60`	`59`	`if (holder.hasContext()) {`
`61`		`- sink.next(holder.currentContext());`
`62`		`- return;`
	`60`	`+ return Mono.just(holder.currentContext());`
`63`	`61`	`}`
`64`	`62`	`}`
`65`		`- sink.error(new NoTransactionInContextException());`
	`63`	`+ return Mono.error(new NoTransactionInContextException());`
`66`	`64`	`});`
`67`	`65`	`}`
`68`	`66`
Original file line number	Diff line number	Diff line change
`@@ -47,7 +47,7 @@ public class ServerWebExchangeContextFilter implements WebFilter {`
`47`	`47`	`@Override`
`48`	`48`	`public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {`
`49`	`49`	`return chain.filter(exchange)`
`50`		`- .subscriberContext(cxt -> cxt.put(EXCHANGE_CONTEXT_ATTRIBUTE, exchange));`
	`50`	`+ .contextWrite(cxt -> cxt.put(EXCHANGE_CONTEXT_ATTRIBUTE, exchange));`
`51`	`51`	`}`
`52`	`52`
`53`	`53`