From fa70b31f1ed80bf8a30e598b8dc62b9cd977547e Mon Sep 17 00:00:00 2001 From: Alvin <329772643@qq.com> Date: Wed, 20 May 2020 03:24:53 +0800 Subject: [PATCH] Cache the success routes until next successful refresh. (#1602) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * refreshWorksWhenFirstRefreshSuccessAndOtherError * update format * use isSuccess instred of Enum Co-authored-by: 田雪峰 Fixes gh-600 --- .../event/RefreshRoutesResultEvent.java | 45 ++++++++++++ .../gateway/route/CachingRouteLocator.java | 40 +++++++++- .../route/CachingRouteLocatorTests.java | 73 +++++++++++++++++++ 3 files changed, 154 insertions(+), 4 deletions(-) create mode 100644 spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/event/RefreshRoutesResultEvent.java diff --git a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/event/RefreshRoutesResultEvent.java b/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/event/RefreshRoutesResultEvent.java new file mode 100644 index 00000000..031b0d92 --- /dev/null +++ b/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; + } + +} diff --git a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/route/CachingRouteLocator.java b/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/route/CachingRouteLocator.java index f5c8d636..cda9daf5 100644 --- a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/route/CachingRouteLocator.java +++ b/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 { +public class CachingRouteLocator implements Ordered, RouteLocator, + ApplicationListener, 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 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; + } + } diff --git a/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/route/CachingRouteLocatorTests.java b/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/route/CachingRouteLocatorTests.java index 4a0a408f..7308e655 100644 --- a/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/route/CachingRouteLocatorTests.java +++ b/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 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 routes = locator.getRoutes().collectList().block(); + assertThat(routes).containsExactly(route1); + + List 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 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();