标签: rxjs-pipeable-operators

RxJs 5.5版引入了管道运算符，以使其更易于组合RxJs运算符，并使这些情况下的树震动更有效。我的问题是，如果只打算使用一个运算符，应该使用管道运算符吗？

考虑以下示例：

someObservable.map(mappingLogicMethod).subscribe(x => console.log(x));

与

someObservable.pipe(map(mappingLogicMethod)).subscribe(x => console.log(x));

在仅使用一个操作员的情况下，最合适的方法是什么？

我对使用Angular和RxJ的takeUntil运算符取消订阅的常见模式之一有疑问。在本文中，它位于第三位置。例如，我们在组件类中有这样的代码：

  private destroy$: Subject<boolean> = new Subject();

  ngOnInit() {
     this.control.
     .pipe(takeUntil(this.destroy$)
     .subscribe(doSmthngFunc); 
  }

  ngOnDestroy() {
    this.destroy$.next(true);
    // Which next line of code is correct?
    // this.destroy$.complete()     // this one?
    // this.destroy$.unsubscribe()  // or this one?
  }

第一行this.destroy $ .next（true）非常清楚。但是第二个不是。如果我们研究这些方法的实现，就会发现它们具有类似的行为。 complete（）： unsubscribe（）：

据我了解，语义上complete（）是可取的，因为我们在组件生命周期中第一次和最后一次调用next（），然后我们完成了此Subject，将其视为Observable并可以调用complete（）。这些方法属于观察者，而取消订阅则属于可观察者，我们没有要取消订阅的订阅。但实际上，这些方法具有类似的代码：

    this.isStopped = true; // both

    this.observers.length = 0; // complete
    this.observers = null;     // unsubscribe

    this.closed = true;        // only unsubscribe

理论上，complete（）可能会延迟作用，因为它可能会在订阅的每个观察者上调用complete（），但是我们在destroy $上没有观察者。因此，问题-哪种方法更可取，更不易出错，为什么？

我使用 angular 6，我想过滤异步管道的结果，然后再在 UI 中呈现它们。

这是我现在的代码

this.results = this.form.get('name').valueChanges.pipe(           
  filter(formdata => formdata.name.length > 0), 
  switchMap( formdata => this.service.getNames(formdata.name)) 
);

和模板

  <div *ngIf='results | async ; let items'>
       <div *ngFor='let item of items'>{{item.id}} {{item.name}} </div>               
  </div> 

从管道中，我得到了一些 ID 和名称。我已经有一个 id 数组。我想过滤管道的 id，而不是渲染已经在数组中的那些。

所以，这就是我尝试做的。

array = [{id:1,name:'one'},{id:2,name:'two'}];//I already have this

新版管道过滤器

this.results = this.form.get('name').valueChanges.pipe(           
  filter(formdata => formdata.name.length > 0), 
  switchMap( formdata => this.service.getNames(formdata.name)) ,
  filter(checkIfResultIdInArray())//pseudocode
);

假设这checkIfResultIdInArray是我创建的函数。过滤并返回所有不在array. 所以最终出现在模板中的 ids/names 不是{id:1,name:'one'},{id:2,name:'two'}.

或者我可以以某种方式在模板中过滤？

更改页面时取消挂起的 HTTP 请求

我们有一个 Angular 服务，它有一个昂贵的 HTTP 查询，可供 3 个不同的消费者访问。每个消费者都可以随时修改此查询，发出新请求，并且所有其他消费者都必须使用新数据进行更新。

因为 HTTP 订阅在完成后立即关闭，所以我们使用带有行为主体的内部可观察模式来保持消费者的连接（见下文）。

那么问题是，当用户更改页面时，当前挂起的 HTTP 响应无法取消。

通常，我认为在后台丢弃 HTTP 请求不会有太大问题……但除了这是一项昂贵的操作之外，我发现在用户返回后是否确实解决了挂起的响应到页面，它将使用旧查询的数据更新消费者。

没有布埃诺。

服务电话

private dataSubject = new BehaviorSubject<MyData>(...);
public data$ = this.dataSubject.asObservable();
...

getData(): Observable<MyData> {
  if (this.dataSubject)
    return this.data$;
  } else {
    const http$ = this.http.post(...))
      .pipe(map(response => response as MyData),
        takeUntil(this.unsubscribe$)); // see tearDown() below
    http$.subscribe(
      (availableDevices: MyData) => {
        this.dataSubject.next(availableDevices);
      }
    );
    return this.data$;
  }
}

我试图在每个消费者在 ngDestroy() 期间调用的服务中创建一个拆卸方法，但除非我完成了流，否则它不起作用。但那时，当用户返回页面时，我无法再次重新启动流。

tearDown(): void {
  this.unsubscribe$.next();
  this.unsubscribe$.complete();
  // this.dataSubject.next(null);
  // this.dataSubject.complete(); -- …

我对 rxjs operator 有点困惑delay。

当我使用创建的假 observable 对其进行测试时from，我只看到初始延迟：

const { from } = Rx;
const { delay, tap } = RxOperators;

from([1, 2, 3, 4]).pipe(
  tap(console.log),
  delay(1000));

（您可以将此代码片段复制并粘贴到rxviz 中。）

我tap在那里放置了一个以确保from实际将数组项作为单独的值而不是单个数组值发出。

最初的延迟不是我所期望的，但至少文档是这样说的：

[...] 此运算符将源 Observable 时间偏移以毫秒表示的时间量。保留值之间的相对时间间隔。

但是，当我使用从事件创建的 observable 对其进行测试时，我会在每个发出的值之前看到延迟：

const { fromEvent } = Rx;
const { delay } = RxOperators;

fromEvent(document, 'click')
  .pipe(delay(1000))

这里发生了什么？为什么delay在这两种情况下表现不同？

• 我有一个自动完成输入。
• 每次输入或删除字母时，我都会发出 HTTP 请求。
• 响应被映射到建议列表，我将其显示在输入下方的下拉列表中。

但是：如果最后一个字母被删除，使输入为空，我想跳过 HTTP 请求等并返回一个空数组。

因此，我需要一个在管道中首先调用的运算符，每次满足条件时都会跳过下面的所有运算符并“提前返回”，就像 for 循环中的“break”或“return”语句一样。

我不能使用 filter()，因为该运算符会阻止生成的可观察值发出。但我需要它发出来清除下拉列表。

<input [formControl]="formGroup.searchTerm">
<ul>
 <li *ngFor="let suggestion of suggestions$ | async">{{suggestion}}</li>
</ul>

suggetions$ = this.formGroup.valueChanges.pipe(
    pluck('searchString')
    // filter(searchString => searchString.length > 0) // does not help
    unknownOperator(searchString => {
        if(searchString.length === 0) {
            skipOperatorsBelowAndReturnThisInstead([])
        } else {
            continueWithOperatorsBelow(searchTerm)
        }
    })
    switchMap(values => this.http.get(this.url + values.term)),
    map(this.buildSuggestions),
),

谢谢！

我有两个可观察流，它们执行非常独立的映射逻辑，但最终以以下 3 个运算符结束：

  this.selection
    .pipe(
      ..Custom mapping operators
      tap(_ => this.devicesLoading = true),
      switchMap(d => this.mapService.findLocationForDevices(d)),
      map(loc => marker([loc.latitude, loc.longitude])
    )
    .subscribe(markers => this.plotMarkers(markers));

我想将最后一个tap, switchMap, map运算符移动到一个通用函数，这样我就可以在我的两个可观察流中应用它们。

我想过这样做：

  private resolveLocationsAndConvertToMarkers = (devices: String[]) => [
    tap(_ => this.devicesLoading = true),
    switchMap((devices: string[]) => this.mapService.findLocationForDevices(devices)),
    map(loc => marker([loc.latitude, loc.longitude])
  ];

但我不确定如何将这些运算符扩展到管道参数中，例如：#

      this.selection
        .pipe(
          // Custom mapping operators
          ... this.resolveLocationsAndConvertToMarkers
        )
        .subscribe(markers => this.plotMarkers(markers));

这个错误there are no overloads that expect 3 or 5 arguments...

使用 RxJs 时的一个非常常见的问题似乎是希望一个或多个 observables 的结果然后在后续的观察中使用它们。

例如在伪代码中（这不是故意的 rx 或有效的 js 语法）

var someResult = $observable-A; // wait to complete
var finalResult = $observable-B(someResult.aValueINeed);

这可以以一种丑陋的方式完成，您可以订阅两者并在另一个内部调用一个..但是，这非常混乱并且不能为您提供很大的灵活性。

例如（真正的语法）

$observable-A.subscribe(resultA => { 
    $observable-B(resultA.aValueINeed)
        .subscribe(resultB => { 
            console.log('After everything completes: ', resultB); 
        }
}

这也意味着当您完成两个 observable 时，没有其他任何东西可以轻松消耗此流。

我的特定用例需要以下内容：

1. 对 $observable-A 的初始调用，因为它不需要其他任何东西（这是一个基本的 http GET 调用）
2. 对另一个服务的 http GET 调用需要来自 $observable-A 的数据，该服务返回 $observable-B
3. 使用两个可观察结果 (A + B) 为我的服务（在我的情况下，Angular 服务）创建一个对象以返回单个列表。

我还需要能够在我的服务中订阅这个功能，这就是为什么使用上面的订阅方法对我不起作用。

我知道这个问题以前有人问在这里。但是接受的解决方案不适用于我，或者我无法很好地理解它。

我正在使用ng-7，我有一个简单的用例：

我有2个API，第2个取决于1st的响应。我订阅第一个API的结果，然后使用管道订阅第二个API的结果。

我的代码如下所示；

this._SomeService
        .addUserToDb(payload)
        .pipe(
          map(res => res),
          mergeMap(db1Response =>
            this._SomeService.addUserToDb2(db1Response
            )
          ),
          catchError(errodb1 => {

            return Observable.throw(new 
            Error(errorSso));
          })
        )
        .subscribe(
          resDb2 => {
              // Here I get response of addUserToDb2
          },
          errDb2 => {


          }
        )

现在，在订阅第二个API响应之前，我想订阅另一个可观察的说法：

this._tokenService.getToken.pipe（

）

并希望在服务2中使用它的响应。这样：

API1 =>令牌=> API2

请建议如何实施。

更新：

我尝试实现，以下是我的实现：

  this._service.addUserToDB1(payload).pipe(
          map(resp => this.resDB1 = resp) // Adding to global variable because I need this response while subscribing to DB2 service.
          ,mergeMap(resdb1=>this._tokenService.getToken.pipe(
            mergeMap(token => this._service.addUserToDb2(
              this.resDB1,
              this.organizationId,
              this.practitionerId, …

我有多个在页面生命周期内发出值的可观察对象。例如：

chartData$: Observable;
tableData$: Observable;
filterData$: Observable;

用户可以随时单击“下载”按钮，并获取结合了每个可观察量最后发出的值的 JSON：

downloadButtonClicked$.pipe(
    combine chartData$, tableData$ and filterData$    // <- how do I get latest values here?
).subscribe(([chart, table, filter]) => downloadJson(chart, table, filter))

但是，downloadJson当这 3 个可观察量中的任何一个在页面生命周期中发出值时，不应调用该函数，仅在单击“下载”时调用。

太长了；

工作最优雅的解决方案（如迈克建议） https://stackblitz.com/edit/typescript-jm3zma?file=index.ts