无限调用链

Question

我有一个需要输入并返回输出的服务：

Item callService(Item input);

输入始终是可观察对象发出的前一个元素，服务调用的输出应该是可观察对象发出的下一个元素。并且应该重复这个序列，直到一个项目满足某个条件。

实际上，我有一片树叶，我将树叶发送到一个服务，该服务告诉我它的父级是什么。我会继续这样做，直到找到根。

我的第一个解决方案：

我递归调用一个 flatMap 函数。由于递归，我不喜欢这个解决方案：

Observable.just(Item.of("start"))
          .map(Result::of)
          .flatMap(App::fallbackUntilIsRoot)
          .blockingSingle();

private static Observable<Result> fallbackUntilIsRoot(Result previousResult)
{
    if (previousResult.isRoot())
    {
        return Observable.just(previousResult);
    } else
    {
        return Service.call(previousResult.getItem())
                      .flatMap(App::fallbackUntilIsRoot); // recursion
    }
}

我的第二个解决方案：

我维护一个外部队列，基本上我用这个队列提供 observable。看起来还是不太好。

Queue<Item> todoList = new LinkedList<>(Collections.singleton(Item.of("start")));

Result result = Observable.range(1, Integer.MAX_VALUE) //max tries
                          .concatMap(attempt -> Service.call(todoList))
                          .takeUntil(Result::isEmpty)
                          .takeLast(2)
                          .firstElement()
                          .blockingGet();

public class Service
{
    private static final Queue<Item> itemsProvidedByService = new LinkedList<>(
            Arrays.asList(Item.of("first"), Item.of("second"), Item.of("third"), Item.of("fourth"), Item.of("root")));

    public static Observable<Result> call(Queue<Item> todoList)
    {
        Item previousItem = todoList.poll();

        Item newItem = itemsProvidedByService.poll(); // we could pass previousItem to a real service here

        if (newItem == null)
        {
            return Observable.just(Result.empty());
        }

        todoList.add(newItem);

        return Observable.just(Result.of(newItem));
    }
}

有没有一种 Rx 方式来进行这种链接？

Answer 1

与此同时，我找到了我的问题的名称。它被称为核心递归。 RxJava 2 为这个问题提供了一个干净（无副作用）的解决方案。

private static void solutionWithGenerate()
{
    Result startElement = Result.of(Item.of("start"));

    Result result = Observable.generate(() -> startElement, App::generate)
                              .startWith(startElement)
                              .lastElement()
                              .blockingGet();

    System.out.println(result);
}

private static Result generate(Result previousResult, Emitter<Result> emitter)
{
    Result newResult = Service.callDirectly(previousResult.getItem());

    if (newResult.isEmpty())
    {
        emitter.onComplete();
    } else
    {
        emitter.onNext(newResult);
    }

    return newResult;
}

也可以在 RxJava 1 中完成。更详细一点： Observable.Generate in RxJava?

Answer 2

您可以通过保留当前迭代项的单个副作用字段来做到这一点。
将repeat() 与takeUntil() 组合在一起将使while 循环与fromCallable() 一起使迭代器在循环中可变（即使用just() 将永远重复初始值）：

it = Result.of(Item.of("start"));
Observable.fromCallable(() -> it)
        .map(result -> {
            it = service.call(result);
            return result;
        })
        .repeat()
        .takeUntil(Item -> it.isRoot())
        .subscribe(item -> .....);