Cache the success routes until next successful refresh. (#1602)

* refreshWorksWhenFirstRefreshSuccessAndOtherError

* update format

* use isSuccess instred of Enum

Co-authored-by: 田雪峰 <tianxuefeng@shuidihuzhu.com>

Fixes gh-600

spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/event/RefreshRoutesResultEvent.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2013-2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.gateway.event;
+
+import org.springframework.context.ApplicationEvent;
+
+/**
+ * @author alvin
+ */
+public class RefreshRoutesResultEvent extends ApplicationEvent {
+
+	private Throwable throwable;
+
+	public RefreshRoutesResultEvent(Object source, Throwable throwable) {
+		super(source);
+		this.throwable = throwable;
+	}
+
+	public RefreshRoutesResultEvent(Object source) {
+		super(source);
+	}
+
+	public Throwable getThrowable() {
+		return throwable;
+	}
+
+	public boolean isSuccess() {
+		return throwable == null;
+	}
+
+}

spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/route/CachingRouteLocator.java

@@ -21,10 +21,15 @@ import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.stream.Collectors;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 import reactor.cache.CacheFlux;
 import reactor.core.publisher.Flux;
 
 import org.springframework.cloud.gateway.event.RefreshRoutesEvent;
+import org.springframework.cloud.gateway.event.RefreshRoutesResultEvent;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.context.ApplicationEventPublisherAware;
 import org.springframework.context.ApplicationListener;
 import org.springframework.core.Ordered;
 import org.springframework.core.annotation.AnnotationAwareOrderComparator;
@@ -32,8 +37,10 @@ import org.springframework.core.annotation.AnnotationAwareOrderComparator;
 /**
  * @author Spencer Gibb
  */
-public class CachingRouteLocator
-		implements Ordered, RouteLocator, ApplicationListener<RefreshRoutesEvent> {
+public class CachingRouteLocator implements Ordered, RouteLocator,
+		ApplicationListener<RefreshRoutesEvent>, ApplicationEventPublisherAware {
+
+	private static final Log log = LogFactory.getLog(CachingRouteLocator.class);
 
 	private static final String CACHE_KEY = "routes";
 
@@ -43,6 +50,8 @@ public class CachingRouteLocator
 
 	private final Map<String, List> cache = new ConcurrentHashMap<>();
 
+	private ApplicationEventPublisher applicationEventPublisher;
+
 	public CachingRouteLocator(RouteLocator delegate) {
 		this.delegate = delegate;
 		routes = CacheFlux.lookup(cache, CACHE_KEY, Route.class)
@@ -69,8 +78,25 @@ public class CachingRouteLocator
 
 	@Override
 	public void onApplicationEvent(RefreshRoutesEvent event) {
-		fetch().materialize().collect(Collectors.toList())
-				.doOnNext(routes -> cache.put(CACHE_KEY, routes)).subscribe();
+		try {
+			fetch().collect(Collectors.toList()).subscribe(list -> Flux.fromIterable(list)
+					.materialize().collect(Collectors.toList()).subscribe(signals -> {
+						applicationEventPublisher
+								.publishEvent(new RefreshRoutesResultEvent(this));
+						cache.put(CACHE_KEY, signals);
+					}, throwable -> handleRefreshError(throwable)));
+		}
+		catch (Throwable e) {
+			handleRefreshError(e);
+		}
+	}
+
+	private void handleRefreshError(Throwable throwable) {
+		if (log.isErrorEnabled()) {
+			log.error("Refresh routes error !!!", throwable);
+		}
+		applicationEventPublisher
+				.publishEvent(new RefreshRoutesResultEvent(this, throwable));
 	}
 
 	@Deprecated
@@ -83,4 +109,10 @@ public class CachingRouteLocator
 		return 0;
 	}
 
+	@Override
+	public void setApplicationEventPublisher(
+			ApplicationEventPublisher applicationEventPublisher) {
+		this.applicationEventPublisher = applicationEventPublisher;
+	}
+
 }

spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/route/CachingRouteLocatorTests.java

@@ -16,11 +16,17 @@
 
 package org.springframework.cloud.gateway.route;
 
+import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 
 import org.junit.Test;
 import reactor.core.publisher.Flux;
 
+import org.springframework.cloud.gateway.event.RefreshRoutesEvent;
+import org.springframework.cloud.gateway.event.RefreshRoutesResultEvent;
+
 import static org.assertj.core.api.Assertions.assertThat;
 
 public class CachingRouteLocatorTests {
@@ -60,6 +66,73 @@ public class CachingRouteLocatorTests {
 		assertThat(routes).containsExactly(route1, route2);
 	}
 
+	@Test
+	public void refreshWorksWhenFirstRefreshSuccessAndOtherError()
+			throws InterruptedException {
+		Route route1 = route(1);
+		Route route2 = route(2);
+
+		CachingRouteLocator locator = new CachingRouteLocator(new RouteLocator() {
+			int i = 0;
+
+			@Override
+			public Flux<Route> getRoutes() {
+				if (i == 0) {
+					i++;
+					return Flux.just(route1);
+				}
+				else if (i == 1) {
+					i++;
+					return Flux.just(route2).map(route -> {
+						throw new RuntimeException("in chain.");
+					});
+				}
+				else if (i == 2) {
+					i++;
+					throw new RuntimeException("call getRoutes error.");
+				}
+				return Flux.just(route2);
+			}
+		});
+
+		List<Route> routes = locator.getRoutes().collectList().block();
+		assertThat(routes).containsExactly(route1);
+
+		List<RefreshRoutesResultEvent> resultEvents = new ArrayList<>();
+
+		waitUntilRefreshFinished(locator, resultEvents);
+		assertThat(resultEvents).hasSize(1);
+		assertThat(resultEvents.get(0).getThrowable().getCause().getMessage())
+				.isEqualTo("in chain.");
+		assertThat(resultEvents.get(0).isSuccess()).isEqualTo(false);
+		assertThat(locator.getRoutes().collectList().block()).containsExactly(route1);
+
+		waitUntilRefreshFinished(locator, resultEvents);
+		assertThat(resultEvents).hasSize(2);
+		assertThat(resultEvents.get(1).getThrowable().getMessage())
+				.isEqualTo("call getRoutes error.");
+		assertThat(resultEvents.get(1).isSuccess()).isEqualTo(false);
+		assertThat(locator.getRoutes().collectList().block()).containsExactly(route1);
+
+		waitUntilRefreshFinished(locator, resultEvents);
+		assertThat(resultEvents).hasSize(3);
+		assertThat(resultEvents.get(2).isSuccess()).isEqualTo(true);
+		assertThat(locator.getRoutes().collectList().block()).containsExactly(route2);
+
+	}
+
+	private void waitUntilRefreshFinished(CachingRouteLocator locator,
+			List<RefreshRoutesResultEvent> resultEvents) throws InterruptedException {
+		CountDownLatch cdl = new CountDownLatch(1);
+		locator.setApplicationEventPublisher(o -> {
+			resultEvents.add((RefreshRoutesResultEvent) o);
+			cdl.countDown();
+		});
+		locator.onApplicationEvent(new RefreshRoutesEvent(this));
+
+		assertThat(cdl.await(5, TimeUnit.SECONDS)).isTrue();
+	}
+
 	Route route(int id) {
 		return Route.async().id(String.valueOf(id)).uri("http://localhost/" + id)
 				.order(id).predicate(exchange -> true).build();