可管道操作员

操作员堆叠

管道操作员在事件实际上到达观察者之前进行一些操作。

[事件源 - 操作员1运算符2运算符3ââââ€]观察者

重要的是观察者只能在所有这些操作员之后访问最后一个值。

这就像JavaScript数组实用程序功能一样工作。

注意：应用可管道的操作员会创建一个具有其他逻辑的新观测，它不修改现有的可观察到。

筛选

当一个值传递给过滤器函数时，它将传递给（其他操作员或可观察的）或将被丢弃。

即使下一个发出的过滤器函数过滤器通知，它始终通过错误并完成

我们可以从任何可观察的可观察到的管道方法调用管道方法，然后我们可以调用我们想要的任何操作员。

import { Observable } from 'rxjs';
import { filter } from 'rxjs/operators';

interface NewsItem {
  category: 'Bussiness' | 'Sports';
  content: string;
}

const newsFeed$ = new Observable<NewsItem>((subscriber) => {
  setTimeout(() => {
    subscriber.next({ category: 'Bussiness', content: 'A' });
  }, 1000);
  setTimeout(() => {
    subscriber.next({ category: 'Sports', content: 'B' });
  }, 3000);
  setTimeout(() => {
    subscriber.next({ category: 'Bussiness', content: 'C' });
  }, 4000);
  setTimeout(() => {
    subscriber.next({ category: 'Sports', content: 'D' });
  }, 5000);
  setTimeout(() => {
    subscriber.next({ category: 'Bussiness', content: 'E' });
  }, 6000);
});

newsFeed$
  .pipe(filter((item) => item.category === 'Sports'))
  .subscribe((item) => console.log(item));

如果我们想要我们只能在我们想要的任何地方提取运动饲料

import { Observable } from "rxjs";
import { filter } from "rxjs/operators";

interface NewsItem {
  category: 'Business' | 'Sports';
  content: string;
}

const newsFeed$ = new Observable<NewsItem>(subscriber => {
  setTimeout(() => 
    subscriber.next({ category: 'Business', content: 'A' }), 1000);
  setTimeout(() => 
    subscriber.next({ category: 'Sports', content: 'B' }), 3000);
  setTimeout(() => 
    subscriber.next({ category: 'Business', content: 'C' }), 4000);
  setTimeout(() => 
    subscriber.next({ category: 'Sports', content: 'D' }), 6000);
  setTimeout(() => 
    subscriber.next({ category: 'Business', content: 'E' }), 7000);
});

const sportsNewsFeed$ = newsFeed$.pipe(
  filter(item => item.category === 'Sports')
);

newsFeed$.subscribe(
  item => console.log(item)
);

地图

地图函数的工作确实类似于JS地图功能

import { forkJoin } from "rxjs";
// Mike is from New Delhi and likes to eat pasta.

import { ajax } from "rxjs/ajax";
import { map } from "rxjs/operators";

const randomFirstName$ = ajax<any>('https://random-data-api.com/api/name/random_name').pipe(
  map(ajaxResponse => ajaxResponse.response.first_name)
);

const randomCapital$ = ajax<any>('https://random-data-api.com/api/nation/random_nation').pipe(
  map(ajaxResponse => ajaxResponse.response.capital)
);

const randomDish$ = ajax<any>('https://random-data-api.com/api/food/random_food').pipe(
  map(ajaxResponse => ajaxResponse.response.dish)
);

forkJoin([randomFirstName$, randomCapital$, randomDish$]).subscribe(
  ([firstName, capital, dish]) =>
    console.log(`${firstName} is from ${capital} and likes to eat ${dish}.`)
);

轻敲

*** tab ***操作员像间谍一样工作，让我们在不与通知互动的情况下产生副作用。

由于在地图中，我们返回的值将成为下一个链式函数的输入，有时是出于调试目的，例如，我们可以在中间传递tap（）函数，并间谍发生了什么

import { of } from "rxjs";
import { filter, map, tap } from "rxjs/operators";

of(1, 7, 3, 6, 2).pipe(
  filter(value => value > 5),
  map(value => value * 2),
  tap({
    next: value => console.log('Spy:', value)
  }),
).subscribe(value => console.log('Output:', value));

但是，如果我们将MAP函数与Curly Brances一起使用，并带有返回关键字的返回值，就在返回它们之前，我们还可以安装日志。

，但是在中间保持点击操作员不仅是控制台日志，还可以在那里制作任何副作用，因此它是一个非常有用的工具。

另外一件事

从 rxjs 7.3.0 开始，tap()操作员可以做更多的事情。您可以查看订阅何时在tap()运算符的级别开始和结束。

在上面讨论过的Next，错误和完整处理程序的顶部，tap()运营商介绍了另外三个处理程序：

订阅'当新订阅时称为
取消订阅'订阅通过取消订阅（呼叫unsubscribe）的订阅时；这将不是 '在发出错误或完整时执行
最终确定在订阅结束时，无论原因是什么：它都可以取消订阅，错误或完整的通知；所有这些都会导致最终的处理程序被调用。

...
  tap({
    subscribe: () => { console.log('New Subscription'); },
    unsubscribe: () => { console.log('Unsubscribed'); },
    finalize: () => { console.log('Subscription ended'); }
  }),
...

debouncetime

****************************** debouncetime *************** **********延迟发出的通知，仅在一定阈值后发出最后发射的值。

例如，我们有多次通知，让我们说每秒5个通知，我们只对最后一个值感兴趣。当我们聆听输入字段时，我们可以有这样的方案，用户可以迅速写入，当用户停止稍等时，我们最终会发出价值并处理它。

通过这种方式，我们可以避免出现的处理。

import { fromEvent } from 'rxjs';
import { map, debounceTime } from 'rxjs/operators';

const sliderInput = document.querySelector('input#slider');

fromEvent<any>(sliderInput, 'input')
  .pipe(
    debounceTime(500),
    map((event) => event.target['value'])
  )
  .subscribe((value) => console.log(value));

在此示例中，而不是发出数千个通知，它只是发出3，这是一个了不起的性能增长。

接管器

Catcheror函数在错误的情况下可观察到可观察到的后备。它会自动通过Next（）和完整（）功能。

如果发生错误通知而不是将其传递到下一个链式函数，则将其传递给可观察的。

可能可观察到的可能直接发射下一个（）和完整（）通知，然后将这些通知传递给下一个链式功能。

import { Observable, of } from 'rxjs';
import { catchError } from 'rxjs/operators';

const failingHttpRequest$ = new Observable((subscriber) => {
  setTimeout(() => {
    subscriber.error(new Error('Timeout'));
  }, 3000);

  return () => console.log('Teardown');
});

console.log('App started');

failingHttpRequest$
  .pipe(
    catchError((error) =>
      of(`Fallback value and err message: ${error.message}`)
    )
  )
  .subscribe({
    next: (value) => console.log(value),
    error: (err) => console.log(err.message),
    complete: () => console.log('Completed'),
  });

我们看到错误并没有传递给其他链式功能，我们只是创建了一个可观察到的新观察，并用 * *的 * ，并通过了下一个（）函数还有错误消息。

我们还可以看到，功能后发送完整的通知后，它也向我们的主管道发出了完整的通知。

有时我们可能想隐藏错误消息，或者在发生错误时不想做任何事情，对于这些情况，RXJS具有********* empty *******可观察到的，该名称表明是空的，并且仅发出完整（）通知。

import { EMPTY, Observable, of } from "rxjs"; import { catchError } from "rxjs/operators"; const failingHttpRequest$ = new Observable(subscriber => { setTimeout(() => { subscriber.error(new Error('Timeout')); }, 3000); }); console.log('App started'); failingHttpRequest$.pipe( catchError(error => EMPTY) ).subscribe({ next: value => console.log(value), complete: () => console.log('Completed') });

平坦的操作员

扁平操作员像捕获错误一样工作，但要进行下一个通知。

扁平操作员通过订阅可观察的新通知对下一个通知做出反应，只要源发射新值扁平操作员将不断订阅新的可观察结果。

平坦的操作员静态示例

在变平的操作员中，首先发出了一个值，并使用该值创建一个新值，并且它也开始发出新值，那么，如果我们有来自新创建的可观察的完整通知，则不会发出它们到主要通知。

让我们以concatmap
为例

import { Observable, of } from 'rxjs'; import { concatMap } from 'rxjs/operators'; const source$ = new Observable((subscriber) => { setTimeout(() => subscriber.next('A'), 2000); setTimeout(() => subscriber.next('B'), 3000); }); console.log('App has started'); source$ .pipe(concatMap((value) => of(1, 2))) .subscribe((value) => console.log(value));

我们可以看到，我们将其完全映射到了那些可观察到的值。

即使对于静态数字，它也不是一个非常有效的用例，如果我们可以处理热可观察物的创建等。这些变平的操作员可能非常有用。

变平操作员 - 动态HTTP请求

让我们这次做一个***************************************************************************************************************** ******示例将发送带有提供值的HTTP请求

import { fromEvent } from "rxjs"; import { ajax } from "rxjs/ajax"; import { concatMap, map } from "rxjs/operators"; const endpointInput: HTMLInputElement = document.querySelector('input#endpoint'); const fetchButton = document.querySelector('button#fetch'); fromEvent(fetchButton, 'click').pipe( map(() => endpointInput.value), concatMap(value => ajax(`https://random-data-api.com/api/${value}/random_${value}`) ) ).subscribe( value => console.log(value) );

变平操作员 - 错误处理＃1

即使完整的通知未传递给主流，错误通知也可以！

因此，我们需要仔细处理TEM，以免杀死我们的主要管道

import { EMPTY, fromEvent } from "rxjs"; import { ajax } from "rxjs/ajax"; import { catchError, concatMap, map } from "rxjs/operators"; const endpointInput: HTMLInputElement = document.querySelector('input#endpoint'); const fetchButton = document.querySelector('button#fetch'); fromEvent(fetchButton, 'click').pipe( map(() => endpointInput.value), concatMap(value => ajax(`https://random-data-api.com/api/${value}/random_${value}`) ), catchError(() => EMPTY) ).subscribe({ next: value => console.log(value), error: err => console.log('Error:', err), complete: () => console.log('Completed') });

我们对这种方法有问题，即使我们发现错误，我们也会在主管道中捕获它，并将其转换为完整的通知，这也杀死了主要管道，我们无法继续搜索。

变平操作员 - 错误处理＃2

为了解决上述问题，我们需要将Catcheror链接到内部可观察的

import { EMPTY, fromEvent, of } from 'rxjs'; import { ajax } from 'rxjs/ajax'; import { catchError, concatMap, map } from 'rxjs/operators'; const endpointInput: HTMLInputElement = document.querySelector('input#endpoint'); const fetchButton = document.querySelector('button#fetch'); fromEvent(fetchButton, 'click') .pipe( map(() => endpointInput.value), concatMap((value) => ajax(`https://random-data-api.com/api/${value}/random_${value}`).pipe( catchError((error) => of(`Could not fetch data: ${error}`)) ) ) ) .subscribe({ next: (value) => console.log(value), error: (err) => console.log('Error:', err), complete: () => console.log('Completed'), });

变平操作员 - 并发 - 辅助

ConcatMap操作员在第一个可观察到的完整之前，直到下一个可观察的可观察到。

这可能是一个很好的功能，因为未完成的可观察物将长期造成内存泄漏。

ConcatMap始终保证将以输入的顺序执行可观察到的。这非常重要（例如，考虑多个httprequests或鼠标位置）

变平操作员 - switchmap

switchmap为我们提供了与condapmap几乎相同的解决方案，但有一个很大的差异。

例如，如果以前可观察到的可观察到的新值仍未提供完整的通知，则switchmap自动取消了先前的观察值，并创建具有新值的新值。
。

它会创建不重要的内存泄漏，因为程序在取消操作后清洁此内存

这样，它可以阻止内存泄漏，并且可以根据情况非常快速行动。

平坦的操作员 - 合并

Mergemap也喜欢con缩放，但是它允许多个开放的可观察到，它不取消以前可观察到的SwitchMap，它允许它们同时工作，

这有时是我们需要的确切的东西。

让我们保持联系

嘿，如果您读了这么远，请非常感谢。
我希望您保持联系以获取更多甜蜜的内容。
请在Dev.To和其他频道中订阅（ð°特别twitter）

twitterðhttps://twitter.com/barisbll_dev
LinkedIn - https://www.linkedin.com/in/barisbll-dev/
githubð»ð»https://github.com/barisbll
中等ð°https://medium.com/@barisbll