diff --git a/spring-messaging/src/main/java/org/springframework/messaging/handler/annotation/reactive/PayloadMethodArgumentResolver.java b/spring-messaging/src/main/java/org/springframework/messaging/handler/annotation/reactive/PayloadMethodArgumentResolver.java index b057201230b..53004446d1c 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/handler/annotation/reactive/PayloadMethodArgumentResolver.java +++ b/spring-messaging/src/main/java/org/springframework/messaging/handler/annotation/reactive/PayloadMethodArgumentResolver.java @@ -231,7 +231,7 @@ public class PayloadMethodArgumentResolver implements HandlerMethodArgumentResol Flux flux = content .filter(this::nonEmptyDataBuffer) .map(buffer -> decoder.decode(buffer, elementType, mimeType, hints)) - .onErrorResume(ex -> Flux.error(handleReadError(parameter, message, ex))); + .onErrorMap(ex -> handleReadError(parameter, message, ex)); if (isContentRequired) { flux = flux.switchIfEmpty(Flux.error(() -> handleMissingBody(parameter, message))); } @@ -245,7 +245,7 @@ public class PayloadMethodArgumentResolver implements HandlerMethodArgumentResol Mono mono = content.next() .filter(this::nonEmptyDataBuffer) .map(buffer -> decoder.decode(buffer, elementType, mimeType, hints)) - .onErrorResume(ex -> Mono.error(handleReadError(parameter, message, ex))); + .onErrorMap(ex -> handleReadError(parameter, message, ex)); if (isContentRequired) { mono = mono.switchIfEmpty(Mono.error(() -> handleMissingBody(parameter, message))); } diff --git a/spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/ReactorNetty2TcpClient.java b/spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/ReactorNetty2TcpClient.java index a333e08e26c..61622a35e9c 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/ReactorNetty2TcpClient.java +++ b/spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/ReactorNetty2TcpClient.java @@ -256,12 +256,12 @@ public class ReactorNetty2TcpClient

implements TcpOperations

{ this.channelGroup.close().addListener(future -> channnelGroupCloseSink.tryEmitEmpty()); result = channnelGroupCloseSink.asMono(); if (this.loopResources != null) { - result = result.onErrorResume(ex -> Mono.empty()).then(this.loopResources.disposeLater()); + result = result.onErrorComplete().then(this.loopResources.disposeLater()); } if (this.poolResources != null) { - result = result.onErrorResume(ex -> Mono.empty()).then(this.poolResources.disposeLater()); + result = result.onErrorComplete().then(this.poolResources.disposeLater()); } - result = result.onErrorResume(ex -> Mono.empty()).then(stopScheduler()); + result = result.onErrorComplete().then(stopScheduler()); } else { result = stopScheduler(); diff --git a/spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/ReactorNettyTcpClient.java b/spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/ReactorNettyTcpClient.java index 5b56c35f7f9..337b7b31135 100644 --- a/spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/ReactorNettyTcpClient.java +++ b/spring-messaging/src/main/java/org/springframework/messaging/tcp/reactor/ReactorNettyTcpClient.java @@ -254,12 +254,12 @@ public class ReactorNettyTcpClient

implements TcpOperations

{ if (this.channelGroup != null) { result = FutureMono.from(this.channelGroup.close()); if (this.loopResources != null) { - result = result.onErrorResume(ex -> Mono.empty()).then(this.loopResources.disposeLater()); + result = result.onErrorComplete().then(this.loopResources.disposeLater()); } if (this.poolResources != null) { - result = result.onErrorResume(ex -> Mono.empty()).then(this.poolResources.disposeLater()); + result = result.onErrorComplete().then(this.poolResources.disposeLater()); } - result = result.onErrorResume(ex -> Mono.empty()).then(stopScheduler()); + result = result.onErrorComplete().then(stopScheduler()); } else { result = stopScheduler(); diff --git a/spring-r2dbc/src/main/java/org/springframework/r2dbc/connection/R2dbcTransactionManager.java b/spring-r2dbc/src/main/java/org/springframework/r2dbc/connection/R2dbcTransactionManager.java index 38ef488cae5..8537aa4d8d1 100644 --- a/spring-r2dbc/src/main/java/org/springframework/r2dbc/connection/R2dbcTransactionManager.java +++ b/spring-r2dbc/src/main/java/org/springframework/r2dbc/connection/R2dbcTransactionManager.java @@ -228,8 +228,8 @@ public class R2dbcTransactionManager extends AbstractReactiveTransactionManager .then(Mono.error(ex)); } return Mono.error(ex); - })).onErrorResume(ex -> Mono.error(new CannotCreateTransactionException( - "Could not open R2DBC Connection for transaction", ex))); + })).onErrorMap(ex -> new CannotCreateTransactionException( + "Could not open R2DBC Connection for transaction", ex)); }).then(); } diff --git a/spring-r2dbc/src/test/java/org/springframework/r2dbc/core/AbstractDatabaseClientIntegrationTests.java b/spring-r2dbc/src/test/java/org/springframework/r2dbc/core/AbstractDatabaseClientIntegrationTests.java index d25e21cb8b4..7fdde3c3bea 100644 --- a/spring-r2dbc/src/test/java/org/springframework/r2dbc/core/AbstractDatabaseClientIntegrationTests.java +++ b/spring-r2dbc/src/test/java/org/springframework/r2dbc/core/AbstractDatabaseClientIntegrationTests.java @@ -50,7 +50,7 @@ abstract class AbstractDatabaseClientIntegrationTests { Mono.from(connectionFactory.create()) .flatMapMany(connection -> Flux.from(connection.createStatement("DROP TABLE legoset").execute()) .flatMap(Result::getRowsUpdated) - .onErrorResume(e -> Mono.empty()) + .onErrorComplete() .thenMany(connection.createStatement(getCreateTableStatement()).execute()) .flatMap(Result::getRowsUpdated).thenMany(connection.close())).as(StepVerifier::create) .verifyComplete(); diff --git a/spring-r2dbc/src/test/java/org/springframework/r2dbc/core/AbstractTransactionalDatabaseClientIntegrationTests.java b/spring-r2dbc/src/test/java/org/springframework/r2dbc/core/AbstractTransactionalDatabaseClientIntegrationTests.java index 8bcb4c3f49c..8973f553a2c 100644 --- a/spring-r2dbc/src/test/java/org/springframework/r2dbc/core/AbstractTransactionalDatabaseClientIntegrationTests.java +++ b/spring-r2dbc/src/test/java/org/springframework/r2dbc/core/AbstractTransactionalDatabaseClientIntegrationTests.java @@ -69,7 +69,7 @@ abstract class AbstractTransactionalDatabaseClientIntegrationTests { Mono.from(connectionFactory.create()) .flatMapMany(connection -> Flux.from(connection.createStatement("DROP TABLE legoset").execute()) .flatMap(Result::getRowsUpdated) - .onErrorResume(e -> Mono.empty()) + .onErrorComplete() .thenMany(connection.createStatement(getCreateTableStatement()).execute()) .flatMap(Result::getRowsUpdated).thenMany(connection.close())).as(StepVerifier::create).verifyComplete(); diff --git a/spring-web/src/main/java/org/springframework/http/codec/ResourceHttpMessageWriter.java b/spring-web/src/main/java/org/springframework/http/codec/ResourceHttpMessageWriter.java index 5ab892c8d93..f1517acfda0 100644 --- a/spring-web/src/main/java/org/springframework/http/codec/ResourceHttpMessageWriter.java +++ b/spring-web/src/main/java/org/springframework/http/codec/ResourceHttpMessageWriter.java @@ -191,7 +191,7 @@ public class ResourceHttpMessageWriter implements HttpMessageWriter { if (InputStreamResource.class != resource.getClass()) { return Mono.fromCallable(resource::contentLength) .filter(length -> length != -1) - .onErrorResume(IOException.class, t -> Mono.empty()) + .onErrorComplete(IOException.class) .subscribeOn(Schedulers.boundedElastic()); } else { diff --git a/spring-web/src/main/java/org/springframework/web/server/ServerWebExchange.java b/spring-web/src/main/java/org/springframework/web/server/ServerWebExchange.java index 56c5cef3a62..da7a3bfb520 100644 --- a/spring-web/src/main/java/org/springframework/web/server/ServerWebExchange.java +++ b/spring-web/src/main/java/org/springframework/web/server/ServerWebExchange.java @@ -148,10 +148,10 @@ public interface ServerWebExchange { */ default Mono cleanupMultipart() { return getMultipartData() - .onErrorResume(t -> Mono.empty()) // ignore errors reading multipart data + .onErrorComplete() // ignore errors reading multipart data .flatMapIterable(Map::values) .flatMapIterable(Function.identity()) - .flatMap(part -> part.delete().onErrorResume(ex -> Mono.empty())) + .flatMap(part -> part.delete().onErrorComplete()) .then(); } diff --git a/spring-web/src/main/java/org/springframework/web/server/adapter/DefaultServerWebExchange.java b/spring-web/src/main/java/org/springframework/web/server/adapter/DefaultServerWebExchange.java index b7b63d0def0..07fe8e3e781 100644 --- a/spring-web/src/main/java/org/springframework/web/server/adapter/DefaultServerWebExchange.java +++ b/spring-web/src/main/java/org/springframework/web/server/adapter/DefaultServerWebExchange.java @@ -249,11 +249,11 @@ public class DefaultServerWebExchange implements ServerWebExchange { public Mono cleanupMultipart() { if (this.multipartRead) { return getMultipartData() - .onErrorResume(t -> Mono.empty()) // ignore errors reading multipart data + .onErrorComplete() // ignore errors reading multipart data .flatMapIterable(Map::values) .flatMapIterable(Function.identity()) .flatMap(part -> part.delete() - .onErrorResume(ex -> Mono.empty())) + .onErrorComplete()) .then(); } else { diff --git a/spring-web/src/test/java/org/springframework/http/server/reactive/ServerHttpResponseTests.java b/spring-web/src/test/java/org/springframework/http/server/reactive/ServerHttpResponseTests.java index cfcb0253998..910bf7ae47b 100644 --- a/spring-web/src/test/java/org/springframework/http/server/reactive/ServerHttpResponseTests.java +++ b/spring-web/src/test/java/org/springframework/http/server/reactive/ServerHttpResponseTests.java @@ -105,7 +105,7 @@ class ServerHttpResponseTests { headers.setContentType(MediaType.APPLICATION_JSON); headers.set(HttpHeaders.CONTENT_ENCODING, "gzip"); headers.setContentLength(12); - response.writeWith(body).onErrorResume(ex -> Mono.empty()).block(); + response.writeWith(body).onErrorComplete().block(); assertThat(response.statusCodeWritten).isFalse(); assertThat(response.headersWritten).isFalse(); diff --git a/spring-webflux/src/main/java/org/springframework/web/reactive/function/client/DefaultWebClient.java b/spring-webflux/src/main/java/org/springframework/web/reactive/function/client/DefaultWebClient.java index 854cba139e8..79246fbf4d3 100644 --- a/spring-webflux/src/main/java/org/springframework/web/reactive/function/client/DefaultWebClient.java +++ b/spring-webflux/src/main/java/org/springframework/web/reactive/function/client/DefaultWebClient.java @@ -186,11 +186,11 @@ final class DefaultWebClient implements WebClient { } private static Mono releaseIfNotConsumed(ClientResponse response) { - return response.releaseBody().onErrorResume(ex2 -> Mono.empty()); + return response.releaseBody().onErrorComplete(); } private static Mono releaseIfNotConsumed(ClientResponse response, Throwable ex) { - return response.releaseBody().onErrorResume(ex2 -> Mono.empty()).then(Mono.error(ex)); + return response.releaseBody().onErrorComplete().then(Mono.error(ex)); } diff --git a/spring-webflux/src/main/java/org/springframework/web/reactive/result/method/annotation/AbstractMessageReaderArgumentResolver.java b/spring-webflux/src/main/java/org/springframework/web/reactive/result/method/annotation/AbstractMessageReaderArgumentResolver.java index db735c3e872..5765945f023 100644 --- a/spring-webflux/src/main/java/org/springframework/web/reactive/result/method/annotation/AbstractMessageReaderArgumentResolver.java +++ b/spring-webflux/src/main/java/org/springframework/web/reactive/result/method/annotation/AbstractMessageReaderArgumentResolver.java @@ -183,7 +183,7 @@ public abstract class AbstractMessageReaderArgumentResolver extends HandlerMetho logger.debug(exchange.getLogPrefix() + "0..N [" + elementType + "]"); } Flux flux = reader.read(actualType, elementType, request, response, readHints); - flux = flux.onErrorResume(ex -> Flux.error(handleReadError(bodyParam, ex))); + flux = flux.onErrorMap(ex -> handleReadError(bodyParam, ex)); if (isBodyRequired) { flux = flux.switchIfEmpty(Flux.error(() -> handleMissingBody(bodyParam))); } @@ -199,7 +199,7 @@ public abstract class AbstractMessageReaderArgumentResolver extends HandlerMetho logger.debug(exchange.getLogPrefix() + "0..1 [" + elementType + "]"); } Mono mono = reader.readMono(actualType, elementType, request, response, readHints); - mono = mono.onErrorResume(ex -> Mono.error(handleReadError(bodyParam, ex))); + mono = mono.onErrorMap(ex -> handleReadError(bodyParam, ex)); if (isBodyRequired) { mono = mono.switchIfEmpty(Mono.error(() -> handleMissingBody(bodyParam))); } diff --git a/spring-webflux/src/main/java/org/springframework/web/reactive/result/view/freemarker/FreeMarkerView.java b/spring-webflux/src/main/java/org/springframework/web/reactive/result/view/freemarker/FreeMarkerView.java index 38d5894436d..50a556de640 100644 --- a/spring-webflux/src/main/java/org/springframework/web/reactive/result/view/freemarker/FreeMarkerView.java +++ b/spring-webflux/src/main/java/org/springframework/web/reactive/result/view/freemarker/FreeMarkerView.java @@ -223,7 +223,7 @@ public class FreeMarkerView extends AbstractUrlBasedView { return lookupTemplate(locale) .map(template -> Boolean.TRUE) .switchIfEmpty(Mono.just(Boolean.FALSE)) - .onErrorResume(FileNotFoundException.class, t -> Mono.just(Boolean.FALSE)) + .onErrorReturn(FileNotFoundException.class, Boolean.FALSE) .onErrorMap(ParseException.class, ex -> new ApplicationContextException( "Failed to parse FreeMarker template for URL [" + getUrl() + "]", ex)) .onErrorMap(IOException.class, ex -> new ApplicationContextException(