如何解决在RXJS中，如何使用retryWhen强制重新执行通过bindNodeCallback创建的Observable？ 编辑
我有一个Subject，它接收对象，然后调用用bindNodeCallback包装的函数。如果由于错误而失败，或者结果返回错误，我想使用retryWhen重试该函数的执行。
我尝试了几种不同的方法，但是未能成功使绑定函数重新启动。
function fakeSend(
task: string,cb: (err: Error | null,result?: boolean) => void
) {
console.log("fakesend",task);
setTimeout(() => {
const hasError = Math.random() < 0.5;
const res = Math.random() < 0.5;
console.log(hasError ? "hasError" : `responding with ${res}`);
if (hasError) {
return cb(new Error("error"));
}
return cb(null,res);
},100);
}
const boundSend = bindNodeCallback(fakeSend);
const subject = new Subject<string>();
subject.subscribe(
(task) => {
boundSend(task)
.pipe(
tap((status) => {
if (!status) {
throw new Error("Did not send");
}
return status;
}),retryWhen((errs) =>
errs.pipe(
delay(1000),tap((err) => console.log)
)
)
)
.subscribe({
next: console.log,error: console.error,complete: () => {
console.log("complete",task);
}
});
},(error) => {
console.log("error in subject subscribe");
}
);
subject.next("test1");
subject.next("test2");
subject.next("test3");
subject.next("test4");
谢谢！
const { Subject,bindNodeCallback } = rxjs;
const { tap,retryWhen,delay } = rxjs.operators;
function fakeSend(
task,cb
) {
console.log("fakesend",100);
}
const boundSend = bindNodeCallback(fakeSend);
const subject = new Subject();
subject.subscribe(
(task) => {
boundSend(task)
.pipe(
tap((status) => {
console.log('tap');
if (!status) {
throw new Error("Did not send");
}
return status;
}),(error) => {
console.log("error in subject subscribe");
}
);
subject.next("test1");
subject.next("test2");
subject.next("test3");
subject.next("test4");<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.6.2/rxjs.umd.min.js"></script>编辑
我可以使用它：
subject.subscribe(
(task) => {
of(task).pipe(
mergeMap((v) => boundSend(v)),tap((val) => {
if (!val) throw new Error("Did not send");
}),retryWhen((errs) => errs.pipe(delay(300)))
).subscribe();
});
但是，我不明白为什么这不起作用：
subject
.pipe(
mergeMap((task) => boundSend(task)),tap((val) => {
if (!val) throw new Error("Did not send");
}),retryWhen((errs) => errs.pipe(delay(300)))
)
.subscribe();
Edit2
好吧。我想这可以解释：https://github.com/ReactiveX/rxjs/issues/1401
我在想这是因为我是RX菜鸟：/
解决方法
我认为，如果您希望能够在通过{出现错误的情况下重试基于基于回调的函数，可以尝试使用new Observable构造函数而不是bindNodeCallback {1}}操作员。一些细节如下。
首先，您可以使用retryWhen构造函数，从基于基于回调的函数创建一个Observable，
new Observable函数function newSend(task: string) {
return new Observable(
(subscriber: Subscriber<string>): TeardownLogic => {
fakeSend(task,(err: Error | null,result?: string) => {
if (err) {
subscriber.error(err);
} else {
subscriber.next(result);
subscriber.complete();
}
});
}
);
}
返回一个Observable，该Observable内部调用newSend，并根据传递给fakeSend的回调收到的参数发出结果或错误。
现在，我们有了创建包裹基于基于回调的函数的Observable的方法，我们可以使用fakeSend从通知流中创建回调的结果流通过主题生成的，就像这样
mergeMap如果执行上面的代码，我们会看到主题上的订阅将所有成功调用记录到subject
.pipe(
mergeMap((task) => {
return newSend(task);
})
)
.subscribe(
(data) => console.log("Notification",data),(error) => {
console.log("Error in final subscription",error);
},() => console.log("DONE")
);
subject.next("test1");
subject.next("test2");
subject.next("test3");
subject.next("test4");
subject.complete(); // added just to show that the complete function of the subscriber can be invoked
，直到遇到错误为止，这时它将记录错误并终止（请参阅this stackblitz）。
然后我们可以添加fakeSend运算符，以便在retryWhen回调中发生错误的情况下真正重试
fakeSend至此，我们的逻辑已经完成。每当错误地调用subject
.pipe(
mergeMap((task) => {
return newSend(task).pipe(
retryWhen((errs) =>
errs.pipe(
delay(1000),tap((err) => console.log("Retry",err.message))
)
)
);
})
)
回调函数时，fakeSend运算符将确保执行对具有相同参数的retryWhen的新调用。完整的代码可以在this stackblitz中看到。
为什么不起作用 fakeSend
原因是bindNodeCallback的实现在内部使用bindNodeCallback来缓存最后的结果，并在随后的订阅中重播。因此，如果基于基于回调的函数出错，并且AsyncSubject运算符再次订阅，则该订阅会再次生成相同的错误，从而开始无限循环。如果您将retryWhen替换为newSend(task)（第46和47行），可以看到in the stackblitz
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 dio@foxmail.com 举报，一经查实，本站将立刻删除。