我正在编写一个 Angular 应用程序,它使用 ReactiveX API 来处理异步操作。我之前在一个 Android 项目中使用过这个 API,我非常喜欢它如何简化并发任务处理。但是有一件事我不确定如何以正确的方式解决。
如何从正在进行的任务更新观察者?在这种情况下,任务需要时间来加载/创建一个复杂/大的对象,我能够返回中间进度,但不能返回对象本身。 observable 只能返回一种数据类型。因此我知道两种可能性。
我使用了这两种技术,它们都有效,但我想知道是否有一个统一的标准,一个干净的方法,如何解决这个任务。当然,它也可以是一个全新的。我对每个解决方案持开放态度。
最佳答案
经过仔细考虑,我使用了一个
解决方案类似于我的问题中的选项二。
主要的 observable 与实际结果有关
操作。
本例中为 http 请求,但文件迭代示例类似。
它由“工作”函数返回。
可以通过函数参数添加第二个观察者/订阅者。这个订阅者只关心
进度信息。这样所有操作都是空安全的,不需要类型检查。
工作函数的第二个版本,没有进度观察者,
如果不需要进度 UI 更新,则可以使用。
export class FileUploadService {
doWork(formData: FormData, url: string): Subject<Response> {
return this.privateDoWork(formData, url, null);
}
doWorkWithProgress(formData: FormData, url: string, progressObserver: Observer<number>): Subject<Response> {
return this.privateDoWork(formData, url, progressObserver);
}
private privateDoWork(formData: FormData, url: string, progressObserver: Observer<number> | null): Subject<Response> {
return Observable.create(resultObserver => {
let xhr: XMLHttpRequest = new XMLHttpRequest();
xhr.open("POST", url);
xhr.onload = (evt) => {
if (progressObserver) {
progressObserver.next(1);
progressObserver.complete();
}
resultObserver.next((<any>evt.target).response);
resultObserver.complete()
};
xhr.upload.onprogress = (evt) => {
if (progressObserver) {
progressObserver.next(evt.loaded / evt.total);
}
};
xhr.onabort = (evt) => resultObserver.error("Upload aborted by user");
xhr.onerror = (evt) => resultObserver.error("Error");
xhr.send(formData);
});
}
这是对包括进度订阅者在内的函数的调用。使用此解决方案,上传功能的调用者必须
创建/处理/拆除进度订阅者。
this.fileUploadService.doWorkWithProgress(this.chosenSerie.formData, url, new Subscriber((progress) => console.log(progress * 100)).subscribe(
(result) => console.log(result),
(error) => console.log(error),
() => console.log("request Completed")
);
总的来说,我更喜欢这个解决方案,而不是具有单个订阅的“配对”对象。没有必要的空处理,并且
我得到了一个明确的关注点。
该示例是用 Typescript 编写的,但其他 ReactiveX 实现应该可以实现类似的解决方案。
关于rxjs - 如何使用 ReactiveX Observables/Subjects 处理进度更新?,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/46151594/