在 api-gateway 中提取 Flux<Databuffer> 请求正文后出现 SocketTimeoutException

如何解决在 api-gateway 中提取 Flux<Databuffer> 请求正文后出现 SocketTimeoutException

早上好。 我在 api-gateway 和 spring 的上下文中 反应性授权管理器。我必须提取正文请求并过滤一些字段以实现请求的安全性（如果用户没有基于某些正文字段的值的权限，则不允许他获得响应）。所以我首先将身体作为 Flux。 然后我将其转换为 InputStream 并使用 IOUtils 将 InputStream 转换为 String 以便我可以使用 JsonPath 提取字段并将其用于实现安全性的目的。 在 api-gateway 中似乎一切正常，我进行了调试并获得了我期望的值，但是当对权限的检查全部成功并且因此请求转到服务器微服务时，我得到了 java.net.SocketTimeoutException . 这是将 Flux 转换为 InputStream 的代码：

 static InputStream getInputStreamFromFluxDataBuffer(Flux<DataBuffer> data) throws IOException {
    PipedOutputStream osPipe = new PipedOutputStream();
    PipedInputStream isPipe = new PipedInputStream(osPipe);

    DataBufferUtils.write(data,osPipe)
            .subscribeOn(Schedulers.elastic())
            .doOnComplete(() -> {
                try {
                    osPipe.close();
                } catch (IOException ignored) {
                   
                }
            })
            .subscribe(DataBufferUtils.releaseConsumer());
    return isPipe;
}

然后我检索请求体，它是 Flux 的一个实例，我使用这个方法来获取一个 InputStream。所以我把它转成String来得到body为String（我不想在api-gateway微服务中导入实体的模型）。

    public static String getBodyAsString(AuthorizationContext authorizationContext) throws IOException {
    Flux<DataBuffer> body = authorizationContext.getExchange()
            .getRequest().getBody();

    InputStream inputStream = getInputStreamFromFluxDataBuffer(body);

   String bodyAsString = IOUtils.inputStreamAsString(inputStream,UTF_8.name());
    inputStream.close();

    return bodyAsString;

}

以下步骤包括使用 JsonPath 检索我感兴趣的主体 json 的值：

    public static String getFieldfromBody(String field,AuthorizationContext authorizationContext) throws IOException {
    String body =  getBodyAsString(authorizationContext);
    System.out.println(body);
    DocumentContext docCtx = JsonPath.parse(body);
    String JsonExp="$."+field;
    JsonPath jsonPath = JsonPath.compile(JsonExp);
    return docCtx.read(jsonPath);

}

然后我实现了安全的业务逻辑：

@Component

公共类 UserMgtUpdateUserReactiveAuthorizationManager 实现 ReactiveAuthorizationManager {

@Override
public Mono<AuthorizationDecision> check(Mono<Authentication> authenticationMono,AuthorizationContext authorizationContext) {
    return authenticationMono.map( authentication -> Tuples.of(authentication.getAuthorities(),(String) authentication.getPrincipal()))
            .map( authoritiesAndUserDetails -> {
                String roleValueField = null;
                String username = SecurityUtils.getUsernamePathVariable(authorizationContext);
                try {
                    roleValueField = SecurityUtils.getFieldfromBody("role",authorizationContext);
                } catch (Exception e) {
                    e.printStackTrace();
                }

                return new AuthorizationDecision(checkPermissions(authoritiesAndUserDetails,username,roleValueField));
            });
}

private static boolean checkPermissions(Tuple2<? extends Collection<? extends GrantedAuthority>,String> authoritiesAndUserDetails,String username,String role) {
    Collection<? extends GrantedAuthority> authorities = authoritiesAndUserDetails.getT1();
    String loggedUser = authoritiesAndUserDetails.getT2();
    if(loggedUser.equals(username))
        return false;
    if (StringUtils.isEmpty(role))
        return UserPermissionsUtils.hasUserUpdatePermission(authorities);
    else
        return UserPermissionsUtils.hasUserUpdateRole(authorities) && UserPermissionsUtils.hasUserUpdatePermission(authorities);
}

}

服务器微服务中的API是这样的：

    @RequestMapping(value = "/users/{username}",method = RequestMethod.PATCH)
public ResponseEntity<UserNew> updateUser(@PathVariable("username") String username,@RequestBody UserNewDto userDto) throws Exception {
    userDto.setUsername(username);
    userNewDtoUtils.validateRequiredFieldsForUpdate(userDto);
    userNewDtoUtils.validateInput(userDto);
    userNewDtoUtils.throwExceptionIfCompanyDoesNotExist(userDto);
    userNewDtoUtils.throwExceptionIfAtLeastOneSiteDoesNotExist(userDto);
    userNewDtoUtils.throwExceptionIfAtLeastOneDataloggerDoesNotExists(userDto);

    UserNew userNew = null;
    try {
        userNew = userRegistrationService.updateUser(userDto);
    } catch (UserRepositoryException e) {
        if (UserRepositoryException.USER_NOT_FOUND_ERROR_CODE.equals(e.getErrorCode()))
            throw new ResponseStatusException(HttpStatus.NOT_FOUND,e.getMessage());
        throw new ResponseStatusException(HttpStatus.INTERNAL_SERVER_ERROR,e.getMessage());
    }
    if (userNew == null)
        throw new ResponseStatusException(HttpStatus.NOT_FOUND,userErrorMessageService.buildUserNotFoundErrorMessage(userDto.getUsername()));
    return ResponseEntity.ok()
            .header(HttpHeaders.LOCATION,"/" + userNew.getId())
            .body(userNew);
}

我在服务器微服务控制台中收到以下异常：

 Resolved [org.springframework.http.converter.HttpMessageNotReadableException: I/O error while reading input message; nested exception is org.apache.catalina.connector.ClientAbortException: java.net.SocketTimeoutException]

没有更多的堆栈跟踪。我正在使用邮递员执行其余调用。输入的json是：

{
"email": "[e-mail]","status": "INACTIVE","role": "admin"

}

输出json为

{
"timestamp": 1608628117729,"status": 400,"error": "Bad Request","message": "I/O error while reading input message; nested exception is org.apache.catalina.connector.ClientAbortException: java.net.SocketTimeoutException","path": "/users/use04"

}

所以它包含对 tomcat 的引用。

有人可以帮我吗？ 提前致谢。