没有价值的超时可观察订阅答案

【问题标题】：Timeout observable subscription without value没有价值的超时可观察订阅
【发布时间】：2018-03-15 02:09:00
【问题描述】：

我正在订阅 Firebase 的可观察返回值。如果连接出现问题，我希望订阅超时。这样做的正确方法是什么？我尝试了以下方法，但这在我收到最后一个值后 20 秒超时：

let chatSubscription = this.getChats().timeoutWith(20000, Observable.throw(new Error('timeout'))).subscribe(chats => { ... });

//编辑：getChats()

getChats() {
        return Observable.create(observer => {
            let chatList = this.db.list('/users/' + this.auth.user.uid + '/chats').timeoutWith(20000, Observable.throw(new Error('timeout'))).subscribe(chats => {
                observer.next(chats);
            });  
            //keep track of this subscription
            this.openSubscriptions.push(chatList);  
        });
    }

【问题讨论】：

你能发布你的 getChats() 方法吗
当然，之前应该这样做
你需要 Observable.create() 包装器吗？你可以只返回 this.db.list(...) observable，然后在你的组件中订阅它吗？
不清楚您当前的代码有什么问题。你自己说，它在最后一个值之后超时。那么期望的行为是什么？
我省略了一些代码以使其清楚，这使情况变得更糟:) 现在再次添加。 @Tamas，我想在 20 秒后没有收到单个值时超时。

标签： angular ionic-framework firebase-realtime-database rxjs angularfire

【解决方案1】：

您可以使用race 来收听首先产生的任何 observable：

const chats = this.db.list('/users/' + this.auth.user.uid + '/chats');
const timeout = Observable.throw(new Error("timed out")).delay(20000);
const chatWithTimeout = Observable.race(chats, timeout);
chatWithTimeout.subscribe(msg => ..., err => ...);

另外，您对Observable.create 的使用似乎有点不正统。我建议使用上面的代码并将其用作您的getChats：

getChats() {
    const chats = this.db.list('/users/' + this.auth.user.uid + '/chats');
    const timeout = Observable.throw(new Error("timed out")) .delay(20000);
    const chatWithTimeout = Observable.race(chats, timeout);
    return chatWithTimeout;
}

// usage
const subscription = foo.getChats().subscribe(...);

使用此版本，您无需保留开放订阅列表。让观察者自己跟踪这个订阅。

【讨论】：

我最终使用了这个解决方案，谢谢！一个改进建议：延迟不能直接作用于“抛出”，因为它只是延迟 onNext 和 onComplete 调用（github.com/Reactive-Extensions/RxJS/issues/648）： const timeout = Observable.throw(new Error("timed out")).materialize() . .delay(20000).dematerialize();

【解决方案2】：

您可以从超时 Observable 和 switchMap 开始，从初始超时到您在第一个发出项目时的原始值，例如：

const chats$ = this.getChats();
chats$.map(v => Observable.just(v))
      .startWith(Observable.never().timeoutWith(20000, new Error()))
      .switchMap(v => v)
      .subscribe(...);

如果在初始化后 20 秒内发出一个值，switchMap 只是身份运算符，并继续提供原始值流中的项目。

【讨论】：