可使用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

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