如何在一段时间后取消RXJS订阅

Question

如果用户的网络速度较慢，订阅过程超过30秒以上，我希望能够取消订阅。

const k = this.firebase(user)
        .subscribe(data => {

           //some instructions

        },
        error => alert(error),
        () => console.log("finished"));
}
k.unsubscribe();

- user2243952

Answer 1

查看操作符文档，你会发现很多有趣的东西。

只需使用Timeout操作符，它专门针对这种情况而设计：

const k = this.firebase(user)
    .timeout(30 * 1000)
    .subscribe(
        data => { /* do stuff*/ },
        error => { /* handle it */ },
        () => { /* finished */ }
    );

timeout会等待值被发射，直到时间限制，此时它将以失败的方式结束observable。

这意味着如果在30秒内没有收到任何内容，就会调用error处理程序，您可以使用它来通知用户（如果需要）。

更新：显然 Firebase 客户端在收到值后将observable保持运行状态（可能是为了让您获得进一步的更新通知）。由于Observable从未完成，因此Timeout将在接收数据后的30秒（或传递的任何其他持续时间）之后出现，导致流失败。

要将“流式”Observable转换为单事件Observable，请在超时之前使用Take操作符。

this.firebase(user)
    .take(1)
    .timeout(wait)
    .subscribe(/* etc */);

- Kroltan

---

错误提示将始终显示，即使用户在5秒后获取了数据。超时后，我会收到以下错误：TimeoutError：已发生超时。 - user2243952

@user2243952 哦，我明白了，显然Firebase在接收完数据后仍保持其可观察性（所以，当然，您稍后可以接收更新的数据）。尝试使用.take(1).timeout(...) 以便在单个事件后完成成功。 - Kroltan

非常感谢！但这是否意味着即使超时时间尚未结束，Firebase的可观察对象在第一次尝试后就停止了？ - user2243952

@user2243952 不，没有“尝试”。 firebase 返回的可观察对象每次更改时都会输出一个值。 take 仅通过单个更改，并立即“完成”流。 timeout 将等待30秒，如果发射在此之前发生，则成功，否则将失败。但请记住，流在发出后立即结束，因此超时会看到不会再有数据进入，因此不会出错。 - Kroltan

感谢您详细而清晰的解释，非常有帮助。 - user2243952

Answer 2

您可以使用计时器操作符创建另一个Observable流，并在计时器流完成之前监听k流。

例如：

const timer$ = Observable.timer(30000) // time in ms
const k$ = this.firebase(user)
    .takeUntil(timer$) // subscribe to k$ until the timer$ finishes
    .subscribe( ... )

- vma_93

---

如果时间> 30000，我想显示警报。如何知道时间是否超过了？ - user2243952

是的，@Kroltan 的答案对这种情况更好。很高兴知道这个运算符存在！虽然谢谢你的点赞。 - vma_93