angular - 我如何扩展 RxJS？

我正在寻找一种方法，以便每当有人在实现了 Observable 的某个类的实例上调用 subscribe 时执行回调。

在 RxJS 库中没有钩子的情况下，在几乎任何其他面向对象的语言中，我会：

• 继承我想自定义的类（在这种情况下是ReplaySubject）；
• 重写 subscribe 方法；并
• 执行我的回调，然后将调用委托给超类的 subscribe 方法。

像这样的东西：

export class CallbackReplaySubject<T> extends ReplaySubject<T> { subscribe(observer: Partial<Observer<T>>): Unsubscribable { this.triggerCallback();

阅读更多 →

javascript - 如何使用 RXJS 动态链式调用函数？

我需要在第一次调用时缓存请求的结果，然后在随后的调用中读取缓存的值。

为了实现那个目标，我正在使用 Promise 并将它们进行链式调用。我有一个可用的解决方案，但我想将其转换为 RxJS 的 Observable 而不是 Promise。

这是我的可行方案：

private currentPromise: Promise<{ [key: string]: any }>; private cache: any; public getSomething(name: string): Promise<number>{ return this.c

阅读更多 →

typescript - Angular 2 RC4 RxJS 崩溃

我有一段代码，在最新的 Angular 2 更新之前运行得非常完美。除此之外，它还使用了 Http 请求和 Observables。

我更新到了 RC4，导入了所有 RxJS 操作符（为了确保它能工作），然后我收到错误：SecurityError: DOM Exception 18

有办法知道为什么以及我该如何修复这个吗？

简化的代码片段：

logIn(email: string, pass: string): Observable<boolean> { return Observable.create((observer: Observer<boolean>) => { let body = { email: email, pass: pass }; this.http.post(this.origin + "/auth/logIn", JSON.stringify(body), this.requestArguments())

阅读更多 →

typescript - 动态筛选 rxjs 流

我正在使用 RXJS，并且我希望动态过滤数据，但我遇到了问题：

let numberSource: ReplaySubject<Number> = new ReplaySubject<Number>(); let numberFilter: BehaviorSubject<Number> = new BehaviorSubject<Number>(5); let filteredData = numberSource.filter(n => n < numberFilter.value); numberFilter.subscribe(newFilter => { filteredDat

阅读更多 →

typescript - rxjs5：延迟共享可观察对象的取消订阅

我有一个创建成本很高的可观察对象，所以我已经shared它。然而，在某些情况下，所有订阅者都取消订阅，然后立即（或在短暂延迟后）有一个新的订阅者订阅。

实际可观察到的现象在这里太复杂而无法复制，但为了讨论起见：

const heavyObservable = Rx.Observable.create((observer) => { console.log('I am expensive, avoid hitting this code'); return Rx.Observable .interval(500)

阅读更多 →

angular - RxJs 操作符导致 TypeScript 错误

工作版本:

const observable = Observable .from([Promise.resolve('1'), Promise.resolve('2')]) .subscribe((results) => {})

损坏的版本（添加了 mergeAll 操作符）：

const observable = Observable .from([Promise.resolve('1'), Promise.resolve('2')])

阅读更多 →

angular - 在指定时间发起 API 请求以更新当前广播节目 Angular2 RxJS

我正在为一个社区广播电台制定时间表，并想知道如何更新当前正在播放的节目。

我已经做到了每分钟轮询服务器一次（不确定这是正确的方法，但紧急情况下可以用），但既然我从 API 得到了结束时间，我想知道是否可以设置在结束时间过后才更新，然后使用下一个结束时间设置一个新的观察者。

有人有办法实现这个目标吗，或者有更好的方法来更新当前的节目？

我真正感到困惑的是理解 Observables 和 Subscribers。我如何才能访问 radioShow$ 的当前

阅读更多 →

如何对 rxjs debounceTime 逻辑进行单元测试（Angular）？

由于我对 RxJs 的缺乏了解，我复制粘贴了使用它的一些代码，修改了它，现在测试失败了...

这是整个组件：

import { Component } from '@angular/core'; import { Observable } from 'rxjs/Observable'; import { map } from 'rxjs/operator/map'; import { debounceTime } from 'rxjs/operator/debounceTime'; import { distinctUntilChanged } from 'rxjs/operator/d

阅读更多 →

javascript - 如果条件满足则订阅，否则返回/退出 rxjs

我如何在操作符链中订阅之前检查条件？我有类似这样的情况。

parentElem:boolean = false; // the parent elem show/hide; // inside the ngAfterViewInit(); this.myForm.get('grandParent').valueChanges //.takeWhile(...) no idea if it works with my condition or not. //.skipWhile(...) no idea if it works with my condition or no

阅读更多 →

typescript - 我如何在 RxJS/Angular 中的两个不同点捕获错误？

假设这个 Angular 2 代码片段：

export class ApiService { // do stuff request(...): Observable<Response> { return this.http.request(...) .share() .catch((error: any) => { let errMsg = 'my error'; // do more stuff... // tried: throw

阅读更多 →