如何使用Mono和state实现重复

Question

我有一个方法，它获取一个项目列表，发出一个web请求，然后返回处理失败的项目(有些项目可能已经正确处理，有些项目可能已经失败，只返回失败的项目或一个空列表):

// returns list of of items that failed processing or empty list.
private List<String> sendsItems(List<String> items);

我想使用Mono并尝试发送失败的项目最多N次。使用阻塞代码，它看起来像这样：

public void sendWithRetries() {
    List<String> items = List.of("one", "two", "three");
    int retryNo = 0;
    while (!items.isEmpty() && retryNo < 3) {
        items = sendsItems(items);
        retryNo++;
    }
}

我真的很难用反应器将其转换为代码。我一直在考虑使用Mono.repeat或Mono.retry系列函数，但我没有看到任何将失败的项目传递回sendItems的好方法，除了做一些丑陋的事情，比如：

    var items = new AtomicReference<>(List.of("one", "two", "three"));
    Mono.fromCallable(() -> {
                List<String> failedItems = sendsItems(items.get());
                items.set(failedItems);
                if (!failedItems.isEmpty()) {
                    throw new RuntimeException("retry me");
                }
                return Collections.emptyList();
            })
            .retry(3)

有更好的方法吗？

注意：我需要它的真正用例是kinesis put records api call。它返回有关哪些记录未得到处理的信息，我想重试仅发送它们。

Answer 1

您可以使用扩展运算符来维护状态：

Mono.fromCallable(() -> sendsItems(new Attempt(List.of("one", "two", "three"), 0)))
    .expand(attempt -> {
        if (attempt.getItems().isEmpty() || attempt.getRetry() > 3) {
            return Mono.empty();
        } else {
            return Mono.fromCallable(() -> sendsItems(attempt));
        }
    });

private Attempt sendsItems(Attempt previousAttempt) {
    System.out.println(previousAttempt);
    // implement actual sending logic here, below is just some dummy
    List<String> failedItems = previousAttempt.getItems().subList(0, previousAttempt.getItems().size() - 1);
    return new Attempt(failedItems, previousAttempt.retry + 1);
}

@Value
static class Attempt {
    List<String> items;
    int retry;
}