I有一个外部API，该API返回学生列表以及计数和偏移参数，该参数代表了数据库中剩余的学生数量。 它类似于分页的响应bu ...

Question

和

Student参数一起返回acount

offset

，该参数代表数据库中留在数据库中的学生数量。

它类似于分页的响应，但不会在响应中发送

Pageable

信息。

响应以下格式。

{
"students": 
[
    {
     "id":1, 
     "name":"Adam"
     }, 

    {
     "id":2, 
      "name":"Alan"
     }
], 

"count":2,

"offset":10

}

问题：

我需要在春季反应式中编写递归功能，该功能称为此API并累积所有学生

Flux<Student>

并将其发送到前端。

限制：

我每次通话我可以获取的最大学生数为2。 数据库中可能有大约20名学生。 ROUGH算法：

do(getStudents(offset))
while(response.count <2);

我希望能够使用反应性弹簧进行此操作。 我猜我可以使用

Flux.generate.takeUntil

等。

我不确定正确的语法，因为

takeUntil接受最终值，而不是变量。 振动提出了一种实现上述功能的方法。

任何帮助都会受到赞赏。预先感谢

对于重新质量用用例，反应堆提供mono.expand

（以及其他变体，例如flux..endpand或flux.expanddeep）。

expand

运算符就像flatmap一样，除了它也重新应用于其产生的元素（因此，产生递归效果）。

对于您的用例，伪代码可能是：

int count = 2;
Mono<Response> firstPage = service.getFirstPage(count);
Flux<Response> allPages = firstPage.expand(response -> {
  if (response.moreAvailable) return service.nextPage(response.offset, count);
  else return Mono.empty();
});

thehere是一个完整的最小可重复的示例，该示例模拟了使用内存记录的服务返回页面：

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.Collections;
import java.util.List;

import static java.lang.Math.min;

public class ExpandPages {

    // Data models
    record Student(int id, String name) {}
    record Page(int remaining, int nextOffset, List<Student> students) {}

    /**
     * Mock a service that sends pages of students
     * @param all All available students
     */
    record StudentRegistry(List<Student> all) {
        StudentRegistry {
            if (all == null || all.isEmpty()) throw new IllegalArgumentException("Null or empty student list");
            all = Collections.unmodifiableList(all);
        }

        /**
         * Request a page of students.
         *
         * @return A single page of students, starting at provided offset, with a maximum count of provided count.
         */
        public Mono<Page> next(int offset, int count) {
            if (offset < 0 || offset >= all.size()) throw new IllegalArgumentException("Bad offset");
            if (count < 1) throw new IllegalArgumentException("Bad count");

            count = min(count, all.size() - offset);
            int nextOffset = offset + count;
            int remaining = all.size() - nextOffset;
            return Mono.just(new Page(remaining, nextOffset, all.subList(offset, offset + count)));
        }
    }

    public static void main(String[] args) {
        final var registry = new StudentRegistry(List.of(
                new Student(1, "John"),
                new Student(2, "Jane"),
                new Student(3, "Jack"),
                new Student(4, "Jules"),
                new Student(5, "Julie"),
                new Student(6, "James"),
                new Student(7, "Joe"),
                new Student(8, "Johanna"),
                new Student(9, "Jolly Jumper")
        ));

        final int queriedCount = 2;
        Flux<Page> pages = registry
                // Get first page
                .next(0, queriedCount)
                // Recurse on each received page: check if there's more, then ask for the next available page
                .expand(response -> {
                    System.out.println("Received page "+response);
                    if (response.remaining() <= 0) {
                        System.out.println("No more page to fetch.");
                        return Mono.empty(); // Ends recursion
                    } else {
                        return registry.next(response.nextOffset(), min(queriedCount, response.remaining()));
                    }
                });

        // Trigger flow consumption: print all received students
        pages.flatMapIterable(Page::students)
             .doOnNext(System.out::println)
             .blockLast();
    }
}