Spring Webflux取消管道并返回flux

Question

我将解释我不知道如何在不使用异常的情况下解决的情况。 情况如下，我将通过测试来解释它，但想象一下有一个控制器返回调用“测试方法”方法的 Flux

public class TestDelete {

    @Test
    void testFinal(){
        Flux<ClassWithFlag> fluxWithAdditionalsPass = testReturnWhenConditionPass();
        List<ClassWithFlag> receivedItemsPass = new ArrayList<>();
        fluxWithAdditionalsPass.subscribe( receivedItemsPass::add );
        System.out.println(receivedItemsPass);

        Flux<ClassWithFlag> fluxWithAdditionalsNoPass = testReturnWhenConditionNoPass();
        List<ClassWithFlag> receivedItemsNoPass = new ArrayList<>();
        fluxWithAdditionalsNoPass.subscribe( receivedItemsNoPass::add );
        System.out.println(receivedItemsNoPass);
    }

    private Flux<ClassWithFlag> testReturnWhenConditionPass() {
        List<ClassWithFlag> listOfObject =
                List.of(ClassWithFlag.builder().build(), ClassWithFlag.builder().build());
        Flux<ClassWithFlag> fluxOfObjects = Flux.fromIterable(listOfObject);

        Flux<ClassWithFlag> fluxOfObjectsWIthMappings = fluxOfObjects
                .map( FakeService::mapMessageToPass )
                .map(FakeService::setFlagAccordingToMessage)
                .collectList()
                .flatMapMany(list -> {
                    boolean hasErrors = ClassWithFlag.anyHasError(list);
                    if (hasErrors) {
                        return Flux.error(new CustomException(list));
                    }
                    return Flux.fromIterable(list);
                });
        return fluxOfObjectsWIthMappings.collectList()
            .flatMapMany( Flux::fromIterable )
            .map( FakeService::additional);
    }

    private Flux<ClassWithFlag> testReturnWhenConditionNoPass() {
        List<ClassWithFlag> listOfObject =
                List.of(ClassWithFlag.builder().build(), ClassWithFlag.builder().build());
        Flux<ClassWithFlag> fluxOfObjects = Flux.fromIterable(listOfObject);

        Flux<ClassWithFlag> fluxOfObjectsWithMapping = fluxOfObjects
                .map(FakeService::toNotPass)
                .map(FakeService::setFlagAccordingToMessage)
                .collectList()
                .flatMapMany(list -> {
                    boolean hasErrors = ClassWithFlag.anyHasError(list);
                    if (hasErrors) {
                        return Flux.error(new CustomException(list));
                    }
                    return Flux.fromIterable(list);
                });

        return fluxOfObjectsWithMapping
                .collectList()
                .flatMapMany( Flux::fromIterable ).map(FakeService::additional);
    }



    @AllArgsConstructor
    public class CustomException extends RuntimeException {
        List<ClassWithFlag> listOfObjectWithError;
    }
}

所以这是主要测试，这是假服务和课程

class FakeService {

    static ClassWithFlag mapMessageToPass(ClassWithFlag classWithFlag) {
        classWithFlag.setMessage("Hello");
        return classWithFlag;
    }

    static ClassWithFlag toNotPass(ClassWithFlag classWithFlag) {
        classWithFlag.setMessage("Bye Bye");
        return classWithFlag;
    }

    static ClassWithFlag additional(ClassWithFlag classWithFlag) {
        classWithFlag.setAdditionalMessage("Additional things done");
        return classWithFlag;
    }

    static ClassWithFlag setFlagAccordingToMessage(ClassWithFlag classWithFlag) {
        if (classWithFlag.getMessage().equals("Hello")) {
            classWithFlag.setError(false);
            return classWithFlag;
        }
        classWithFlag.setError(true);
        return classWithFlag;
    }
}

@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class ClassWithFlag {
    private Boolean error;
    private String message;
    private String additionalMessage;

    public static boolean anyHasError(List<ClassWithFlag> classWithFlags) {
        ClassWithFlag classWithFlag = classWithFlags.stream()
                .filter(ClassWithFlag::getError)
                .findFirst()
                .orElse(null);
        return classWithFlag != null;
    }
}

正如您在测试中看到的，我有两种方法“testReturnWhenConditionPass”和“testReturnWhenConditionNoPass”都是相同的，唯一的区别是一个会输入条件以返回 Flux.error 而另一个则不会，所以不这样做的人会做一些额外的地图方法。

我想通过此实现的主要目的是，在任何类的“错误”标志为 true 的情况下，我不想继续流程并返回通量。 我实现它的唯一方法是使用带有自定义异常的 Flux 错误来包装响应，然后我有一个异常处理程序将返回这样的响应

@ResponseStatus(HttpStatus.OK)
@ExceptionHandler(CustomException.class)
public Flux<ClassWithFlag> handleException(CustomExceptionex) {
    return Flux.fromIterable(ex.getListOfObjectWithError());
}

我想知道是否有更好的方法来处理 webflux 上的这个问题，而不是使用和异常，并且如果出现任何错误，直接返回通量，而不是继续“附加”步骤，因为通量类型是相同的。

谢谢大家并致以诚挚的问候

重要提示：我知道这可以通过像这样的嵌套或私有方法来实现，但我不想避免超级嵌套或私有方法

我不想这样做

private Flux<ClassWithFlag> testReturnWhenConditionNoPass() {
        List<ClassWithFlag> listOfObject =
                List.of(ClassWithFlag.builder().build(), ClassWithFlag.builder().build());
        Flux<ClassWithFlag> fluxOfObjects = Flux.fromIterable(listOfObject);

        return fluxOfObjects
                .map(FakeService::toNotPass)
                .map(FakeService::setFlagAccordingToMessage)
                .collectList()
                .flatMapMany(list -> {
                    boolean hasErrors = ClassWithFlag.anyHasError(list);
                    if (hasErrors) {
                        return Flux.fromIterable(new list);
                    }
                    return Flux.fromIterable(list)
                               .collectList()
                               .flatMapMany(Flux::fromIterable)
                               .map(FakeService::additional);
                });

    }

想象一下相同的情况下大多数步骤，最终会出现令人难以置信的嵌套 还必须添加 private 不是有效的解决方案。

TakeWhile 之类的东西不起作用 aqlso 因为它会产生松散的对象

Answer 1

    private Flux<ClassWithFlag> testReturnWhenConditionPass() {
        List<ClassWithFlag> listOfObject =
                List.of(ClassWithFlag.builder().build(), ClassWithFlag.builder().build());
        Flux<ClassWithFlag> fluxOfObjects = Flux.fromIterable(listOfObject);
        return fluxOfObjects
                .map(FakeService::mapMessageToPass)
                .map(FakeService::setFlagAccordingToMessage)
                .collectList()
                .flatMapMany(list -> {
                    boolean hasErrors = ClassWithFlag.anyHasError(list);
                    if (hasErrors) {
                        /* return Flux.error(new CustomException(list));
                        return without further processing...*/
                        return Flux.fromIterable(list);
                    }
                    return continueProcessingIfNoErrorWasDetected(Flux.fromIterable(list));
                });

    }

    private Flux<ClassWithFlag> continueProcessingIfNoErrorWasDetected(Flux<ClassWithFlag> fluxOfObjectsWIthMappings) {
        return fluxOfObjectsWIthMappings.collectList()
                .flatMapMany(Flux::fromIterable)
                .map(FakeService::additional);
    }