WebFlux应用程序中的WebFilter

Question

@Component
public class TraceIdFilter implements WebFilter {

@Override
public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {
    return chain.filter(exchange).subscriberContext((Context context) ->
        context.put(AuditContext.class, getAuditContext(exchange.getRequest().getHeaders()))
    );
}

@GetMapping(value = "/some_mapping")
public Mono<ResponseEntity<WrappedResponse>> getResource(@PathVariable("resourceId") String id) {
    Mono.subscriberContext().flatMap(context -> {
        AuditContext auditContext = context.get(AuditContext.class);
        ...
    });

我遇到的问题是filter方法永远不会执行，并且上下文也没有设置。我已经确认Webfilter是在启动时加载的。还有什么需要让过滤器工作吗？

Answer 1

我有一大堆问题弄清楚了，所以希望它能帮助别人。我的用例是验证请求上的签名。这要求我解析put/post的请求正文。我看到的另一个主要用例是日志记录，所以下面的内容也会有所帮助。

java

@Component
public class MiddlewareAuthenticator implements WebFilter { 

    @Autowired private RequestValidationService requestValidationService;

@Override
public Mono<Void> filter(ServerWebExchange serverWebExchange, WebFilterChain chain) {
  return HEALTH_ENDPOINTS
      .matches(serverWebExchange)
      .flatMap(
          matches -> {
            if (matches.isMatch()) {
              return chain.filter(serverWebExchange);
            } else {
              return requestValidationService
                  .validate(serverWebExchange, 
                       new BiPredicate<ServerWebExchange, String> { 
                         @Override
                         public boolean test(ServerWebExchange e, String body) {
                             /** application logic can go here. few points:
                              1. I used a BiPredicate because I just need a true or false if the request should be passed to the controller. 
                              2. If you want todo other mutations you could swap the predicate to a normal function and return a mutated ServerWebExchange. 
                              3. I pass body separately here to ensure safety of accessing the request body and not having to rewrap the ServerWebExchange. A side affect of this though is any mutations to the String body do not affect downstream.
                              **/
                              return true;
                            }

                      })
                 .flatMap((ServerWebExchange r) -> chain.filter(r));
            }});
}

RequestValidationService.java

@Service
public class RequestValidationService {
private DataBuffer stringBuffer(String value) {
  byte[] bytes = value.getBytes(StandardCharsets.UTF_8);

  NettyDataBufferFactory nettyDataBufferFactory =
      new NettyDataBufferFactory(ByteBufAllocator.DEFAULT);
  DataBuffer buffer = nettyDataBufferFactory.allocateBuffer(bytes.length);
  buffer.write(bytes);
  return buffer;
}

private String bodyToString(InputStream bodyBytes) {
  byte[] currArr = null;
  try {
    currArr = bodyBytes.readAllBytes();
    bodyBytes.read(currArr);
  } catch (IOException ioe) {
    throw new RuntimeException("could not parse body");
  }

  if (currArr.length == 0) {
    return null;
  }

  return new String(currArr, StandardCharsets.UTF_8);
}

private ServerHttpRequestDecorator requestWrapper(ServerHttpRequest request, String bodyStr) {
  URI uri = request.getURI();
  ServerHttpRequest newRequest = request.mutate().uri(uri).build();
  final DataBuffer bodyDataBuffer = stringBuffer(bodyStr);
  Flux<DataBuffer> newBodyFlux = Flux.just(bodyDataBuffer);
  ServerHttpRequestDecorator requestDecorator =
      new ServerHttpRequestDecorator(newRequest) {
        @Override
        public Flux<DataBuffer> getBody() {
          return newBodyFlux;
        }
      };

  return requestDecorator;
}

private InputStream newInputStream() {
  return new InputStream() {
    public int read() {
      return -1;
    }
  };
}

private InputStream processRequestBody(InputStream s, DataBuffer d) {
  SequenceInputStream seq = new SequenceInputStream(s, d.asInputStream());
  return seq;
}

private Mono<ServerWebExchange> processInputStream(
    InputStream aggregatedBodyBytes,
    ServerWebExchange exchange,
    BiPredicate<ServerHttpRequest, String> predicate) {

  ServerHttpRequest request = exchange.getRequest();
  HttpHeaders headers = request.getHeaders();

  String bodyStr = bodyToString(aggregatedBodyBytes);

  ServerWebExchange mutatedExchange = exchange;

  // if the body exists on the request we need to mutate the ServerWebExchange to not
  // reparse the body because DataBuffers can only be read once;
  if (bodyStr != null) {
    mutatedExchange = exchange.mutate().request(requestWrapper(request, bodyStr)).build();
  }

  ServerHttpRequest mutatedRequest = mutatedExchange.getRequest();

  if (predicate.test(mutatedRequest, bodyStr)) {
    return Mono.just(mutatedExchange);
  }

  return Mono.error(new RuntimeException("invalid signature"));
}

/*
 * Because the DataBuffer is in a Flux we must reduce it to a Mono type via Flux.reduce
 * This covers large payloads or requests bodies that get sent in multiple byte chunks
 * and need to be concatentated.
 *
 * 1. The reduce is initialized with a newInputStream
 * 2. processRequestBody is called on each step of the Flux where a step is a body byte
 *    chunk. The method processRequestBody casts the Inbound DataBuffer to a InputStream
 *    and concats the new InputStream with the existing one
 * 3. Once the Flux is complete flatMap is executed with the resulting InputStream which is
 *    passed with the ServerWebExchange to processInputStream which will do the request validation
 */
public Mono<ServerWebExchange> validate(
    ServerWebExchange exchange, BiPredicate<ServerHttpRequest, String> p) {
  Flux<DataBuffer> body = exchange.getRequest().getBody();

  return body.reduce(newInputStream(), this::processRequestBody)
      .flatMap((InputStream b) -> processInputStream(b, exchange, p));
}

}

双谓词文档：https://docs.oracle.com/javase/8/docs/api/java/util/function/BiPredicate.html