Flowable 使用 RxJava ReactiveX 执行任务和返回值

Question

我有下面的方法来检查文件格式是否正确，如果正确，它会添加到requestBody，否则，它应该向客户端抛出一个错误消息，指出文件格式无效.

public Maybe<HttpResponse<?>> post(Publisher<CompletedFileUpload> images) {
        return Flowable.fromPublisher(images)
                .collect(MultipartBody::builder, (requestBody, file) -> {
                    if (new FileExtension().fileExtensionValidation(file.getFilename())) {
                        requestBody
                                .addPart("images", file.getFilename(), MediaType.TEXT_PLAIN_TYPE, file.getBytes());
                    }
                })
                .flatMapMaybe(requestBody -> {
                    if (true)
                        return iImageUploadClient.post(requestBody.build());
                    else
                        return Maybe.just(HttpResponse.serverError("Image file extension invalid, should ne .png, .jpg, .jpeg,.gif"));
                });
    }

此代码检查文件格式if (new FileExtension().fileExtensionValidation(file.getFilename()))，当失败时应返回消息为return Maybe.just(HttpResponse.serverError("Image file extension invalid, should ne .png, .jpg, .jpeg,.gif"));

我写了if (true)，这总是正确的，相反我需要在这里检查或者如何从.collect() 函数返回。我如何使用反应式 Java 来做到这一点

Answer 1

您可以在集合和最后一个 flatMapMaybe 之间使用共享的 AtomicBoolean。此外，如果您想在此处停止图像，请抛出异常并将其转换为中性多部分体，以便 flatMapMaybe 仍然运行。

public Maybe<HttpResponse<?>> post(Publisher<CompletedFileUpload> images) {
    return Maybe.<HttpResponse<?>>defer(() -> {
        AtomicBoolean formatError = new AtomicBoolean();
        return Flowable.fromPublisher(images)
                .collect(MultipartBody::builder, (requestBody, file) -> {
                    if (new FileExtension().fileExtensionValidation(file.getFilename())) {
                        requestBody
                                .addPart("images", file.getFilename(),
                                     MediaType.TEXT_PLAIN_TYPE, file.getBytes());
                    } else {
                        formatError.set(true);
                        throw new CancellationException();
                    }
                })
                .onErrorResumeNext(error -> {
                     if (error instanceof CancellationException) {
                         return Single.just(MultipartBody.builder());
                     }
                     return Single.error(error);
                })
                .flatMapMaybe(requestBody -> {
                    if (!formatError.get())
                        return iImageUploadClient.post(requestBody.build());
                    else
                        return Maybe.just(HttpResponse.serverError(
                            "Image file extension invalid, should ne .png, .jpg, .jpeg,.gif"));
                });
    });
}