org.springframework.http.server.reactive.ServerHttpRequestDecorator#org.springframework.web.reactive.function.server.HandlerStrategies源码实例Demo

@Test
public void webFilter() {

	RouterFunction<ServerResponse> routerFunction = RouterFunctions.route()
			.GET("/", request -> ServerResponse.ok().build())
			.build();

	new DefaultRouterFunctionSpec(routerFunction)
			.handlerStrategies(HandlerStrategies.builder()
					.webFilter((exchange, chain) -> {
						exchange.getResponse().getHeaders().set("foo", "123");
						return chain.filter(exchange);
					})
					.build())
			.build()
			.get()
			.uri("/")
			.exchange()
			.expectStatus().isOk()
			.expectHeader().valueEquals("foo", "123");
}
 

@Test
public void exceptionHandler() {

	RouterFunction<ServerResponse> routerFunction = RouterFunctions.route()
			.GET("/error", request -> Mono.error(new IllegalStateException("boo")))
			.build();

	new DefaultRouterFunctionSpec(routerFunction)
			.handlerStrategies(HandlerStrategies.builder()
					.exceptionHandler((exchange, ex) -> {
						exchange.getResponse().setStatusCode(HttpStatus.BAD_REQUEST);
						return Mono.empty();
					})
					.build())
			.build()
			.get()
			.uri("/error")
			.exchange()
			.expectStatus().isBadRequest();
}
 

@Test
public void webFilter() {

	RouterFunction<ServerResponse> routerFunction = RouterFunctions.route()
			.GET("/", request -> ServerResponse.ok().build())
			.build();

	new DefaultRouterFunctionSpec(routerFunction)
			.handlerStrategies(HandlerStrategies.builder()
					.webFilter((exchange, chain) -> {
						exchange.getResponse().getHeaders().set("foo", "123");
						return chain.filter(exchange);
					})
					.build())
			.build()
			.get()
			.uri("/")
			.exchange()
			.expectStatus().isOk()
			.expectHeader().valueEquals("foo", "123");
}
 

@Test
public void exceptionHandler() {

	RouterFunction<ServerResponse> routerFunction = RouterFunctions.route()
			.GET("/error", request -> Mono.error(new IllegalStateException("boo")))
			.build();

	new DefaultRouterFunctionSpec(routerFunction)
			.handlerStrategies(HandlerStrategies.builder()
					.exceptionHandler((exchange, ex) -> {
						exchange.getResponse().setStatusCode(HttpStatus.BAD_REQUEST);
						return Mono.empty();
					})
					.build())
			.build()
			.get()
			.uri("/error")
			.exchange()
			.expectStatus().isBadRequest();
}
 

@Test
public void shouldSetModifyBodyResponseFilterWithRewriteFunction() {
	ConfigurableApplicationContext context = mock(
			ConfigurableApplicationContext.class);
	Route.AsyncBuilder routeBuilder = Route.async().id("123").uri("abc:123")
			.predicate(exchange -> true);

	when(context.getBean(ModifyResponseBodyGatewayFilterFactory.class))
			.thenReturn(new ModifyResponseBodyGatewayFilterFactory(
					HandlerStrategies.withDefaults().messageReaders(),
					Collections.emptySet(), Collections.emptySet()));

	RouteLocatorBuilder.Builder routes = new RouteLocatorBuilder(context).routes();
	GatewayFilterSpec spec = new GatewayFilterSpec(routeBuilder, routes);
	spec.modifyResponseBody(String.class, String.class,
			(exchange, s) -> Mono.just(s));

	Route route = routeBuilder.build();
	assertThat(route.getFilters()).hasSize(1);

	assertFilter(route.getFilters().get(0),
			ModifyResponseBodyGatewayFilterFactory.ModifyResponseGatewayFilter.class,
			NettyWriteResponseFilter.WRITE_RESPONSE_FILTER_ORDER - 1);
}
 

@Test
public void shouldSetModifyBodyResponseFilterWithRewriteFunctionAndEmptyBodySupplier() {
	ConfigurableApplicationContext context = mock(
			ConfigurableApplicationContext.class);
	Route.AsyncBuilder routeBuilder = Route.async().id("123").uri("abc:123")
			.predicate(exchange -> true);

	when(context.getBean(ModifyResponseBodyGatewayFilterFactory.class))
			.thenReturn(new ModifyResponseBodyGatewayFilterFactory(
					HandlerStrategies.withDefaults().messageReaders(),
					Collections.emptySet(), Collections.emptySet()));

	RouteLocatorBuilder.Builder routes = new RouteLocatorBuilder(context).routes();
	GatewayFilterSpec spec = new GatewayFilterSpec(routeBuilder, routes);
	spec.modifyResponseBody(String.class, String.class,
			(exchange, s) -> Mono.just(s == null ? "emptybody" : s));

	Route route = routeBuilder.build();
	assertThat(route.getFilters()).hasSize(1);

	assertFilter(route.getFilters().get(0),
			ModifyResponseBodyGatewayFilterFactory.ModifyResponseGatewayFilter.class,
			NettyWriteResponseFilter.WRITE_RESPONSE_FILTER_ORDER - 1);
}
 

@Test
public void shouldSetModifyBodyResponseFilterWithRewriteFunctionAndNewContentType() {
	ConfigurableApplicationContext context = mock(
			ConfigurableApplicationContext.class);
	Route.AsyncBuilder routeBuilder = Route.async().id("123").uri("abc:123")
			.predicate(exchange -> true);

	when(context.getBean(ModifyResponseBodyGatewayFilterFactory.class))
			.thenReturn(new ModifyResponseBodyGatewayFilterFactory(
					HandlerStrategies.withDefaults().messageReaders(),
					Collections.emptySet(), Collections.emptySet()));

	RouteLocatorBuilder.Builder routes = new RouteLocatorBuilder(context).routes();
	GatewayFilterSpec spec = new GatewayFilterSpec(routeBuilder, routes);
	spec.modifyResponseBody(String.class, String.class,
			MediaType.APPLICATION_JSON_VALUE, (exchange, s) -> Mono.just(s));

	Route route = routeBuilder.build();
	assertThat(route.getFilters()).hasSize(1);

	assertFilter(route.getFilters().get(0),
			ModifyResponseBodyGatewayFilterFactory.ModifyResponseGatewayFilter.class,
			NettyWriteResponseFilter.WRITE_RESPONSE_FILTER_ORDER - 1);
}
 

@Test
public void shouldSetModifyBodyResponseFilterWithConfigConsumer() {
	ConfigurableApplicationContext context = mock(
			ConfigurableApplicationContext.class);
	Route.AsyncBuilder routeBuilder = Route.async().id("123").uri("abc:123")
			.predicate(exchange -> true);

	when(context.getBean(ModifyResponseBodyGatewayFilterFactory.class))
			.thenReturn(new ModifyResponseBodyGatewayFilterFactory(
					HandlerStrategies.withDefaults().messageReaders(),
					Collections.emptySet(), Collections.emptySet()));

	RouteLocatorBuilder.Builder routes = new RouteLocatorBuilder(context).routes();
	GatewayFilterSpec spec = new GatewayFilterSpec(routeBuilder, routes);
	spec.modifyResponseBody(
			(smth) -> new ModifyResponseBodyGatewayFilterFactory.Config()
					.setRewriteFunction(String.class, String.class,
							(exchange, s) -> Mono.just(s)));

	Route route = routeBuilder.build();
	assertThat(route.getFilters()).hasSize(1);

	assertFilter(route.getFilters().get(0),
			ModifyResponseBodyGatewayFilterFactory.ModifyResponseGatewayFilter.class,
			NettyWriteResponseFilter.WRITE_RESPONSE_FILTER_ORDER - 1);
}
 

/**
 * 输出请求体
 *
 * @param exchange ServerWebExchange
 * @param chain    GatewayFilterChain
 * @param log      日志DTO
 * @return Mono
 */
private Mono<Void> readBody(ServerWebExchange exchange, GatewayFilterChain chain, LogDto log) {
    Flux<DataBuffer> dataBufferFlux = exchange.getRequest().getBody();
    return DataBufferUtils.join(dataBufferFlux).flatMap(dataBuffer -> {
        byte[] bytes = new byte[dataBuffer.readableByteCount()];
        dataBuffer.read(bytes);
        DataBufferUtils.release(dataBuffer);

        // 重新构造请求
        ServerHttpRequest mutatedRequest = new ServerHttpRequestDecorator(exchange.getRequest()) {
            @Override
            public Flux<DataBuffer> getBody() {
                return Flux.defer(() -> {
                    DataBuffer buffer = exchange.getResponse().bufferFactory().wrap(bytes);
                    DataBufferUtils.retain(buffer);

                    return Mono.just(buffer);
                });
            }
        };

        ServerWebExchange mutatedExchange = exchange.mutate().request(mutatedRequest).build();
        return ServerRequest.create(mutatedExchange, HandlerStrategies.withDefaults().messageReaders()).bodyToMono(String.class).doOnNext(body -> {
            if (Pattern.matches("^\\[.*]$", body)) {
                List<Object> list = Json.toList(body, Object.class);
                log.setBody(list == null ? body : list);
            } else if (Pattern.matches("^\\{.*}$", body)) {
                Map obj = Json.toMap(body);
                log.setBody(obj == null ? body : obj);
            } else {
                log.setBody(body);
            }

            logger.info("请求参数：{}", log.toString());
        }).then(chain.filter(mutatedExchange));
    });
}
 

public static void main(String[] args) throws Exception {
    AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
    PostRepository posts = new PostRepository();
    PostHandler postHandler = new PostHandler(posts);
    Routes routesBean = new Routes(postHandler);

    context.registerBean(PostRepository.class, () -> posts);
    context.registerBean(PostHandler.class, () -> postHandler);
    context.registerBean(Routes.class, () -> routesBean);
    context.registerBean(WebHandler.class, () -> RouterFunctions.toWebHandler(routesBean.routes(), HandlerStrategies.builder().build()));
    context.refresh();

    nettyServer(context).onDispose().block();
}
 

/**
 * Instantiates a new Body param plugin.
 */
public BodyParamPlugin() {
    this.messageReaders = HandlerStrategies.withDefaults().messageReaders();
}
 

/**
 * Instantiates a new Body param plugin.
 */
public BodyParamPlugin() {
    this.messageReaders = HandlerStrategies.withDefaults().messageReaders();
}
 

public FileSizeFilter() {
    HandlerStrategies.builder().codecs(configurer -> configurer.defaultCodecs().maxInMemorySize(maxSize * BYTES_PER_MB));
    this.messageReaders = HandlerStrategies.withDefaults().messageReaders();
}
 

public ReadBodyPredicateFactory() {
	super(Config.class);
	this.messageReaders = HandlerStrategies.withDefaults().messageReaders();
}
 

public ModifyRequestBodyGatewayFilterFactory() {
	super(Config.class);
	this.messageReaders = HandlerStrategies.withDefaults().messageReaders();
}
 

private HttpWebHandlerAdapter httpHandler(GenericApplicationContext context) {
	return (HttpWebHandlerAdapter) RouterFunctions.toHttpHandler(context.getBean(RouterFunction.class),
			HandlerStrategies.empty().exceptionHandler(context.getBean(WebExceptionHandler.class))
					.codecs(config -> config.registerDefaults(true)).build());
}
 

/**
 * Handle the given exception. A completion signal through the return value
 * indicates error handling is complete while an error signal indicates the
 * exception is still not handled.
 *
 * @param exchange the current exchange
 * @param ex       the exception to handle
 * @return {@code Mono<Void>} to indicate when exception handling is complete
 */
@Override
public Mono<Void> handle(final ServerWebExchange exchange, final Throwable ex) {
    return handle(ex)
            .flatMap(it -> it.writeTo(exchange,new HandlerStrategiesResponseContext(HandlerStrategies.withDefaults())))
            .flatMap(i -> Mono.empty());
}
 

/**
 * Instantiates a new Handler strategies response context.
 *
 * @param handlerStrategies the handler strategies
 */
public HandlerStrategiesResponseContext(final HandlerStrategies handlerStrategies) {
    this.handlerStrategies = handlerStrategies;
}