springcloud gateway讀取請求body中的內容

request body中的內容只允許讀取一次，若是多次讀取會報錯，本章中簡單介紹了如何在springcloud gateway中讀取body中的內容。

配置方法

首先先介紹java代碼配置的方法

@Bean
public RouteLocator routes(RouteLocatorBuilder builder) {
    return builder.routes()
            .route("openApi", r -> r.path("/openApi")
                    .and()
                    .readBody(JSONObject.class, requestBody -> true)
                    .filters(f -> f.filter(new OpenApiFilter()))
                    .uri("lb://openApi"))
            .build();
}

其中調用了readBody方法，這個方法就是讀取body的核心方法，為gateway提供的，將body參數轉為JSONObject並方入緩存中。
之后調用了名為OpenApiFilter的過濾器，這個是自定義的過濾器。我回在OpenApiFilter中用到body。

使用方法

在routes中使用了readBody方法后，就可以在緩存中獲取到body參數。以下為Filter中的摘要。

@Override
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
    ServerHttpRequest request = exchange.getRequest();
    ServerHttpResponse response = exchange.getResponse();
    if (!HttpMethod.POST.equals(request.getMethod())) {
        // 不是POST請求直接返回
    }
    JSONObject jsonObject = exchange.getAttribute("cachedRequestBodyObject");
    // 后面可以根據body中的內容進行處理
    // 省略返回方法
    return null;
}

源碼簡單解析

首先看一下核心的readBody方法。

public <T> BooleanSpec readBody(Class<T> inClass, Predicate<T> predicate) {
    return asyncPredicate(getBean(ReadBodyPredicateFactory.class)
            .applyAsync(c -> c.setPredicate(inClass, predicate)));
}

可以看出其調用了ReadBodyPredicateFactory類的applyAsync方法。

@Override
@SuppressWarnings("unchecked")
public AsyncPredicate<ServerWebExchange> applyAsync(Config config) {
    return new AsyncPredicate<ServerWebExchange>() {
        @Override
        public Publisher<Boolean> apply(ServerWebExchange exchange) {
            Class inClass = config.getInClass();
            // 首先從exchange中獲得cachedBody緩存中的body,CACHE_REQUEST_BODY_OBJECT_KEY的值即為cachedRequestBodyObject
            // 所以在自定義Filter的中我們也同樣以此方法獲取body參數。
            Object cachedBody = exchange.getAttribute(CACHE_REQUEST_BODY_OBJECT_KEY);
            Mono<?> modifiedBody;
            // We can only read the body from the request once, once that happens if
            // we try to read the body again an exception will be thrown. The below
            // if/else caches the body object as a request attribute in the
            // ServerWebExchange so if this filter is run more than once (due to more
            // than one route using it) we do not try to read the request body
            // multiple times
            // 注釋翻譯結果
            // 我們只能從請求中讀取一次內容，一旦我們嘗試再次讀取正文，將引發異常。 下面
            // if / else將主體對象作為請求屬性緩存在ServerWebExchange，因此如果此篩選器運行了多次（由於運行了多次）
            // 而不是使用它的一條路線），我們不會嘗試讀取請求正文多次

            if (cachedBody != null) {
                // 如果緩存中的已經存在則去執行配置中的predicate，本案例中沒有配置predicate。
                try {
                    boolean test = config.predicate.test(cachedBody);
                    exchange.getAttributes().put(TEST_ATTRIBUTE, test);
                    return Mono.just(test);
                }
                catch (ClassCastException e) {
                    if (log.isDebugEnabled()) {
                        log.debug("Predicate test failed because class in predicate "
                                + "does not match the cached body object", e);
                    }
                }
                return Mono.just(false);
            }
            else {
                // 如果不存在則讀取body並方入CACHE_REQUEST_BODY_OBJECT_KEY中，然后執行配置中的predicate.
                // 讀取body的操作在ServerWebExchangeUtils.cacheRequestBodyAndRequest中。
                return ServerWebExchangeUtils.cacheRequestBodyAndRequest(exchange,
                        (serverHttpRequest) -> ServerRequest
                                .create(exchange.mutate().request(serverHttpRequest)
                                        .build(), messageReaders)
                                .bodyToMono(inClass)
                                .doOnNext(objectValue -> exchange.getAttributes().put(
                                        CACHE_REQUEST_BODY_OBJECT_KEY, objectValue))
                                .map(objectValue -> config.getPredicate()
                                        .test(objectValue)));
            }
        }

        @Override
        public String toString() {
            return String.format("ReadBody: %s", config.getInClass());
        }
    };
}

ServerWebExchangeUtils.cacheRequestBodyAndRequest方法

public static <T> Mono<T> cacheRequestBodyAndRequest(ServerWebExchange exchange,
        Function<ServerHttpRequest, Mono<T>> function) {
    return cacheRequestBody(exchange, true, function);
}

繼續往下看cacheRequestBody

private static <T> Mono<T> cacheRequestBody(ServerWebExchange exchange,
        boolean cacheDecoratedRequest,
        Function<ServerHttpRequest, Mono<T>> function) {
    // Join all the DataBuffers so we have a single DataBuffer for the body
    return DataBufferUtils.join(exchange.getRequest().getBody())
            .flatMap(dataBuffer -> {
                if (dataBuffer.readableByteCount() > 0) {
                    if (log.isTraceEnabled()) {
                        log.trace("retaining body in exchange attribute");
                    }
                    exchange.getAttributes().put(CACHED_REQUEST_BODY_ATTR,
                            dataBuffer);
                }

                ServerHttpRequestDecorator decorator = new ServerHttpRequestDecorator(
                        exchange.getRequest()) {
                    @Override
                    public Flux<DataBuffer> getBody() {
                        return Mono.<DataBuffer>fromSupplier(() -> {
                            if (exchange.getAttributeOrDefault(
                                    CACHED_REQUEST_BODY_ATTR, null) == null) {
                                // probably == downstream closed
                                return null;
                            }
                            // TODO: deal with Netty
                            NettyDataBuffer pdb = (NettyDataBuffer) dataBuffer;
                            return pdb.factory()
                                    .wrap(pdb.getNativeBuffer().retainedSlice());
                        }).flux();
                    }
                };
                if (cacheDecoratedRequest) {
                    exchange.getAttributes().put(
                            CACHED_SERVER_HTTP_REQUEST_DECORATOR_ATTR, decorator);
                }
                return function.apply(decorator);
            });
}

這里運用了裝飾器設計模式將body中讀取到的dataBuffer數據包裝了下往下傳遞。這里本人也只是粗略了解。先記錄一下。