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

如何解决Flowable 使用 RxJava ReactiveX 执行任务和返回值

我有下面的方法来检查文件格式是否正确，如果它们正确，它会添加到 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,.gif"));

我写的 if (true) 总是正确的，但我需要在这里检查或如何从 .collect() 函数返回。我怎样才能用反应式 java 做到这一点

解决方法

您可以在收集和最后一个 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"));
                });
    });
}