Improve route predicate handling of errors in filterWhen

This commit slightly changes the way filterWhen is used in route
predicate mapping, in order to avoid swallowing of errors due to the
combination of `filterWhen` and `next()`. This is due to `filterWhen`
delaying the emission of async predicate errors to the end of the
Flux, and `next()` potentially cancelling said Flux before the error(s)
are propagated.

The fix involves applying `filterWhen` inside a `concatMap`, on a value
per value basis rather than on the whole original Flux. This results
in predicate errors being propagated as they happen, at which point
we can log them and swallow them in the concatMap.

spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/handler/RoutePredicateHandlerMapping.java

@@ -19,6 +19,8 @@ package org.springframework.cloud.gateway.handler;
 
 import java.util.function.Function;
 
+import reactor.core.publisher.Mono;
+
 import org.springframework.cloud.gateway.route.Route;
 import org.springframework.cloud.gateway.route.RouteLocator;
 import org.springframework.web.cors.CorsConfiguration;
@@ -29,8 +31,6 @@ import static org.springframework.cloud.gateway.support.ServerWebExchangeUtils.G
 import static org.springframework.cloud.gateway.support.ServerWebExchangeUtils.GATEWAY_PREDICATE_ROUTE_ATTR;
 import static org.springframework.cloud.gateway.support.ServerWebExchangeUtils.GATEWAY_ROUTE_ATTR;
 
-import reactor.core.publisher.Mono;
-
 /**
  * @author Spencer Gibb
  */
@@ -88,17 +88,20 @@ public class RoutePredicateHandlerMapping extends AbstractHandlerMapping {
 	}
 
 	protected Mono<Route> lookupRoute(ServerWebExchange exchange) {
-		return this.routeLocator.getRoutes()
-				.filterWhen(route ->  {
-					// add the current route we are testing
-					exchange.getAttributes().put(GATEWAY_PREDICATE_ROUTE_ATTR, route.getId());
-					try {
-						return route.getPredicate().apply(exchange);
-					} catch (Exception e) {
-						logger.error("Error applying predicate for route: "+route.getId(), e);
-					}
-					return Mono.just(false);
-				})
+		return this.routeLocator
+				.getRoutes()
+				//individually filter routes so that filterWhen error delaying is not a problem
+				.concatMap(route -> Mono
+						.just(route)
+						.filterWhen(r -> {
+							// add the current route we are testing
+							exchange.getAttributes().put(GATEWAY_PREDICATE_ROUTE_ATTR, r.getId());
+							return r.getPredicate().apply(exchange);
+						})
+						//instead of immediately stopping main flux due to error, log and swallow it
+						.doOnError(e -> logger.error("Error applying predicate for route: "+route.getId(), e))
+						.onErrorResume(e -> Mono.empty())
+				)
 				// .defaultIfEmpty() put a static Route not found
 				// or .switchIfEmpty()
 				// .switchIfEmpty(Mono.<Route>empty().log("noroute"))

spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/handler/RoutePredicateHandlerMappingTest.java

@@ -0,0 +1,97 @@
+package org.springframework.cloud.gateway.handler;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.mockito.Mockito;
+import org.springframework.boot.test.rule.OutputCapture;
+import org.springframework.cloud.gateway.route.Route;
+import org.springframework.cloud.gateway.route.RouteLocator;
+import org.springframework.web.server.ServerWebExchange;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
+import static org.hamcrest.Matchers.containsString;
+
+/**
+ * @author Simon Baslé
+ */
+public class RoutePredicateHandlerMappingTest {
+
+	@Rule
+	public OutputCapture outputCapture = new OutputCapture();
+
+	@Test
+	public void lookupRouteFromSyncPredicates() {
+		Route routeFalse = Route.async()
+				.id("routeFalse")
+				.uri("http://localhost")
+				.predicate(swe -> false)
+				.build();
+		Route routeFail = Route.async()
+				.id("routeFail")
+				.uri("http://localhost")
+				.predicate(swe -> { throw new IllegalStateException("boom"); })
+				.build();
+		Route routeTrue = Route.async()
+				.id("routeTrue")
+				.uri("http://localhost")
+				.predicate(swe -> true)
+				.build();
+		RouteLocator routeLocator =
+				() -> Flux.just(routeFalse, routeFail, routeTrue).hide();
+		RoutePredicateHandlerMapping mapping =
+				new RoutePredicateHandlerMapping(null, routeLocator);
+
+		final Mono<Route> routeMono =
+				mapping.lookupRoute(Mockito.mock(ServerWebExchange.class));
+
+		StepVerifier.create(routeMono.map(Route::getId))
+				.expectNext("routeTrue")
+				.verifyComplete();
+
+		outputCapture.expect(containsString("Error applying predicate for route: routeFail"));
+		outputCapture.expect(containsString("java.lang.IllegalStateException: boom"));
+	}
+
+	@Test
+	public void lookupRouteFromAsyncPredicates() {
+		Route routeFalse = Route.async()
+				.id("routeFalse")
+				.uri("http://localhost")
+				.asyncPredicate(swe -> Mono.just(false))
+				.build();
+		Route routeError = Route.async()
+				.id("routeError")
+				.uri("http://localhost")
+				.asyncPredicate(swe -> Mono.error(new IllegalStateException("boom1")))
+				.build();
+		Route routeFail = Route.async()
+				.id("routeFail")
+				.uri("http://localhost")
+				.asyncPredicate(swe -> { throw new IllegalStateException("boom2"); })
+				.build();
+		Route routeTrue = Route.async()
+				.id("routeTrue")
+				.uri("http://localhost")
+				.asyncPredicate(swe -> Mono.just(true))
+				.build();
+		RouteLocator routeLocator =
+				() -> Flux.just(routeFalse, routeError, routeFail, routeTrue).hide();
+		RoutePredicateHandlerMapping mapping =
+				new RoutePredicateHandlerMapping(null, routeLocator);
+
+		final Mono<Route> routeMono =
+				mapping.lookupRoute(Mockito.mock(ServerWebExchange.class));
+
+		StepVerifier.create(routeMono.map(Route::getId))
+				.expectNext("routeTrue")
+				.verifyComplete();
+
+		outputCapture.expect(containsString("Error applying predicate for route: routeError"));
+		outputCapture.expect(containsString("java.lang.IllegalStateException: boom1"));
+
+		outputCapture.expect(containsString("Error applying predicate for route: routeFail"));
+		outputCapture.expect(containsString("java.lang.IllegalStateException: boom2"));
+	}
+}