如何根据HttpOutBoundGateway失败的Httpstatus（401,400）重试消息

Question

基本上，我的用例是在HttpOutboundGateway请求中出现401时重试http请求。该请求来自jms代理，进入集成流。

 @Bean
  IntegrationFlow bank2wallet(ConnectionFactory jmsConnectionFactory,
      MessageHandler creditWalletHttpGateway) {
    return IntegrationFlows.from(Jms.messageDrivenChannelAdapter(jmsConnectionFactory)
        .destination(cp.getTransactionIn()))
        .<String, CreditRequest>transform(
            request -> new Gson().fromJson(request, CreditRequest.class))

        .enrichHeaders((headerEnricherSpec -> {
          // Todo get token from cache
          headerEnricherSpec.header(HttpHeaders.AUTHORIZATION, String.join(" ", "Bearer", ""));
          headerEnricherSpec.header(HttpHeaders.ACCEPT, "application/json");
          headerEnricherSpec.header(HttpHeaders.CONTENT_TYPE, "application/json");
        }))
        .handle(creditWalletHttpGateway, (e) -> e.advice(retryAdvice()))
        .get();
  }


@Bean
  MessageHandler creditWalletHttpGateway( @Value("${api.base.uri:https:/localhost/v3/sync}") URI uri) {
    HttpRequestExecutingMessageHandler httpHandler = new HttpRequestExecutingMessageHandler(uri);
    httpHandler.setExpectedResponseType(CreditResponse.class);
    httpHandler.setHttpMethod(HttpMethod.POST);
    return httpHandler;
  }

 @Bean
  RequestHandlerRetryAdvice retryAdvice() {
    RequestHandlerRetryAdvice requestHandlerRetryAdvice = new RequestHandlerRetryAdvice();
    requestHandlerRetryAdvice.setRecoveryCallback(errorMessageSendingRecoverer());
    return requestHandlerRetryAdvice;
  }

  @Bean
  ErrorMessageSendingRecoverer errorMessageSendingRecoverer() {
    return new ErrorMessageSendingRecoverer(recoveryChannel());
  }

  @Bean
  MessageChannel recoveryChannel() {
    return new DirectChannel();
  }

  @Bean
  MessageChannel retryChannel() {
    return new DirectChannel();
  }

  @Bean
  IntegrationFlow handleRecovery() {
    return IntegrationFlows.from("recoveryChannel")
        .log(Level.ERROR, "error", m -> m.getPayload())
        .<RuntimeException>handle((message) -> {

          
          MessagingException exception = (MessagingException) message.getPayload();
          Message<CreditRequest> originalCreditRequest = (Message<CreditRequest>) exception.getFailedMessage();
          // String token = gateway.getToken(configProperties);
          String token = UUID.randomUUID().toString();
          Message<CreditRequest> c = MessageBuilder.fromMessage(originalCreditRequest)
              .setHeader(ApiConstants.AUTHORIZATION, String.join(" ", "Bearer", token))
              .copyHeaders(message.getHeaders())
              .build();

          retryChannel().send(c);

        })
        .get();
  }

  @Bean
  IntegrationFlow creditRequestFlow() {

    return IntegrationFlows.from(retryChannel())
        .log(Level.INFO, "info", m ->  m.getPayload())
        .handle(Http.outboundGateway("https://localhost/v3/sync")
        .httpMethod(HttpMethod.POST)
        .expectedResponseType(CreditResponse.class))
        .get();

  }

标头使用适当的http标头进行了丰富，然后我有一个建议，使用默认的简单策略重试请求，RequestHandlerAdvice方法的问题是，它将HandlerRecovery流中的异常消息默认为非HttpException类（MessageException），因此我无法检查HttpStatus代码以重新路由消息。所以我的问题基本上是如何设计一个基于HttpStatus 401重试HttpOutBoundRequest的流。

Answer 1

我通过引入一个网关来进行出站http调用，并使用递归方式对其进行管理，从而解决了这个问题

@MessagingGateway
public interface B2WGateway {

    /**
     *
     * @param message
     * @return
     */
    @Gateway(requestChannel = "credit.input")
    CreditResponse bankToWallet(Message<CreditRequest> message);


}

然后隔离http出站集成流

 @Bean
  IntegrationFlow credit() {

    return f -> f.log(Level.INFO, "info", m -> m.getHeaders())
        .handle(Http.outboundGateway(configProperties.getBankToWalletUrl())
            .httpMethod(HttpMethod.POST)
            .expectedResponseType(CreditResponse.class)
        .errorHandler(new ResponseErrorHandler() {
          @Override
          public boolean hasError(ClientHttpResponse clientHttpResponse) throws IOException {
            return clientHttpResponse.getStatusCode().equals(HttpStatus.UNAUTHORIZED);
          }

          @Override
          public void handleError(ClientHttpResponse clientHttpResponse) throws IOException {
              throw new AuthenticationRequiredException("Authentication Required");
          }
        }));
  }

然后将消息解析为handleRecovery，以便在获得令牌刷新后发送消息

@Bean
  IntegrationFlow handleRecovery() {
    return IntegrationFlows.from("recoveryChannel")
        .log(Level.ERROR, "error", m -> m.getPayload())
        .<RuntimeException>handle((p, h) -> {
          MessageHandlingExpressionEvaluatingAdviceException exception = (MessageHandlingExpressionEvaluatingAdviceException) p;
          Message<CreditRequest> originalCreditRequest = (Message<CreditRequest>) exception
              .getFailedMessage();
          // String token = gateway.getToken(configProperties);
          String token = UUID.randomUUID().toString();
          Message<CreditRequest> c = MessageBuilder.fromMessage(originalCreditRequest)
              .setHeader(ApiConstants.AUTHORIZATION, String.join(" ", "Bearer", token))
              .copyHeaders(h)
              .build();
          return c;
        })
        .channel("credit.input")
        .get();
  }

然后修改了流的初始值，以使用网关服务和表达式建议。

 @Bean
  IntegrationFlow bank2wallet(ConnectionFactory jmsConnectionFactory) {
    return IntegrationFlows.from(Jms.messageDrivenChannelAdapter(jmsConnectionFactory)
        .destination(cp.getTransactionIn()))
        .<String, CreditRequest>transform(
            request -> new Gson().fromJson(request, CreditRequest.class))

        .enrichHeaders((headerEnricherSpec -> {
          // Todo get token from cache
          headerEnricherSpec.header(HttpHeaders.AUTHORIZATION, String.join(" ", "Bearer", ""));
          headerEnricherSpec.header(HttpHeaders.ACCEPT, "application/json");
          headerEnricherSpec.header(HttpHeaders.CONTENT_TYPE, "application/json");
        }))
        .handle((GenericHandler<CreditRequest>) (creditRequest, headers) -> gateway
            .bankToWallet(MessageBuilder.withPayload(creditRequest)
                .copyHeaders(headers)
                .build()), (e) -> e.advice(retryAdvice()))
        .get();
  }

Spring集成的启示-管理超文本传输协议出站适配器调用中的401错误