Spring集成HTTP出站网关根据回复内容重试

Question

我正在使用分两步运行的 API：

1. 它以异步方式开始处理文档，并为您提供用于第 2 步的 ID
2. 它提供了一个端点，您可以在其中获取结果，但前提是它们准备就绪。所以基本上它总是会给你一个 200 响应，其中包含一些细节，比如处理状态。

所以问题是如何为 HTTP 出站网关实现自定义“成功”标准。我还想将它与我已经实现的 RetryAdvice 结合起来。

我尝试了以下方法，但首先在 HandleMessageAdvice 中提供的消息有效负载为空，其次未触发重试：

.handle(Http.outboundGateway("https://northeurope.api.cognitive.microsoft.com/vision/v3" +
        ".0/read/analyzeResults/abc")
        .mappedRequestHeaders("Ocp-Apim-Subscription-Key")
        .httpMethod(HttpMethod.GET), c -> c.advice(this.advices.retryAdvice())
              .handleMessageAdvice(new AbstractHandleMessageAdvice() {
    @Override
    protected Object doInvoke(MethodInvocation invocation, Message<?> message) throws Throwable {
        String body = (String) message.getPayload();
        if (StringUtils.isEmpty(body))
            throw new RuntimeException("Still analyzing");
        JSONObject document = new JSONObject(body);
        if (document.has("analyzeResult"))
            return message;
        else
            throw new RuntimeException("Still analyzing");
    }
}))

我从 4 年前的 Artem 那里找到了这个答案，但首先我没有在出站网关上找到回复通道方法，其次不确定这个场景是否已经在新版本的 Spring 集成中得到改进：http outbound retry with conditions (For checker condition).

更新

按照 Artem 的建议，我有以下几点：

.handle(Http.outboundGateway("https://northeurope.api.cognitive.microsoft.com/vision/v3" +
        ".0/read/analyzeResults/abc")
        .mappedRequestHeaders("Ocp-Apim-Subscription-Key")
        .httpMethod(HttpMethod.GET), c -> c.advice(advices.verifyReplySuccess())
        .advice(advices.retryUntilRequestCompleteAdvice()))

建议：

@Bean
public Advice verifyReplySuccess() {
    return new AbstractRequestHandlerAdvice() {
        @Override
        protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) {
            try {
                Object payload = ((MessageBuilder) callback.execute()).build().getPayload();
                String body = (String) ((ResponseEntity) payload).getBody();
                JSONObject document = new JSONObject(body);
                if (document.has("analyzeResult"))
                    return message;
            } catch (JSONException e) {
                throw new RuntimeException(e);
            }
            throw new RuntimeException("Still analyzing");
        }
    };
}

但是现在当我调试 doInvoke 方法时，payload 的主体是 null。奇怪的是，当我使用 Postman 执行相同的 GET 请求时，正文被正确返回。有什么想法吗？

使用 Postman 的响应正文如下所示：

{
    "status": "succeeded",
    "createdDateTime": "2020-09-01T10:55:52Z",
    "lastUpdatedDateTime": "2020-09-01T10:55:57Z",
    "analyzeResult": {
        "version": "3.0.0",
        "readResults": [
            {
                "page": 1,........

这是我使用回调从出站网关获取的有效负载：

<200,[Transfer-Encoding:"chunked", Content-Type:"application/json; charset=utf-8", x-envoy-upstream-service-time:"27", CSP-Billing-Usage:"CognitiveServices.ComputerVision.Transaction=1", apim-request-id:"a503c72f-deae-4299-9e32-625d831cfd91", Strict-Transport-Security:"max-age=31536000; includeSubDomains; preload", x-content-type-options:"nosniff", Date:"Tue, 01 Sep 2020 19:48:36 GMT"]>

Answer 1

在 Java DSL 中确实没有 request 和 reply 通道选项，因为您只需将 handle() 包装到 channel() 配置中，或者只是以流自然的方式链接端点，它们将使用隐式交换消息之间的直接渠道。您可以在 XML 配置中查看 Java DSL IntegrationFlow 作为 <chain>。

您的通知配置有点错误：您需要将自定义通知声明为链中的第一个，因此当从那里抛出异常时，重试将处理它。

您还应该考虑实现AbstractRequestHandlerAdvice 以使其与RequestHandlerRetryAdvice 逻辑保持一致。

您在那里实现doInvoke()，调用ExecutionCallback.execute() 并分析结果以按原样返回或抛出所需的异常。调用HttpRequestExecutingMessageHandler 的结果将是AbstractIntegrationMessageBuilder，并且可能是ResponseEntity 作为payload，以检查您的进一步逻辑。

Answer 2

按照 Artem 的建议，我提出了以下建议（附加技巧是将 expectedResponseType 设置为 String，否则使用 ResponseEntity 正文为空）：

.handle(Http.outboundGateway("https://northeurope.api.cognitive.microsoft.com/vision/v3" +
        ".0/read/analyzeResults/abc")
        .mappedRequestHeaders("Ocp-Apim-Subscription-Key")
        .httpMethod(HttpMethod.GET).expectedResponseType(String.class),
        c -> c.advice(advices.retryUntilRequestCompleteAdvice())
              .advice(advices.verifyReplySuccess()))

建议：

@Bean
public Advice verifyReplySuccess() {
    return new AbstractRequestHandlerAdvice() {
        @Override
        protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) {
            Object payload = ((MessageBuilder) callback.execute()).build().getPayload();
            if (((String) payload).contains("analyzeResult"))
                return payload;
            else
                throw new RuntimeException("Still analyzing");
        }
    };
}