RxJS和Interceptor

RxJS 是一个组织异步逻辑的库，它有很多 operator，可以极大的简化异步逻辑的编写。

它是由数据源产生数据，经过一系列 operator 的处理，最后传给接收者。

这个数据源叫做 observable。

比如这样：

import { of, filter, map } from 'rxjs';

of(1, 2, 3)
.pipe(map((x) => x * x))
.pipe(filter((x) => x % 2 !== 0))
.subscribe((v) => console.log(`value: ${v}`));

1
2
3
4
5
6

用 node 跑一下，结果如下：

这里 node 能直接解析 esm 需要在 package.json 里设置 type 为 module：

这就是 map、filter 的 operator 的作用。

还是很容易理解的。

有同学说，这样的逻辑自己写也行呀。

那这种呢：

import { of, scan, map } from 'rxjs';

const numbers$ = of(1, 2, 3);

numbers$
  .pipe(
    scan((total, n) => total + n),
    map((sum, index) => sum / (index + 1))
  )
  .subscribe(console.log);

1
2
3
4
5
6
7
8
9
10

scan 是计数，map 是转换，结果如下：

或者是节流、防抖：

import { fromEvent, throttleTime } from 'rxjs';

const clicks = fromEvent(document, 'click');
const result = clicks.pipe(throttleTime(1000));

result.subscribe(x => console.log(x));

1
2
3
4
5
6

import { fromEvent, debounceTime } from 'rxjs';

const clicks = fromEvent(document, 'click');
const result = clicks.pipe(debounceTime(1000));
result.subscribe(x => console.log(x));

1
2
3
4
5

有同学说，这些逻辑都很简单呀，完全可以自己写。

没错，一般异步逻辑自己写也行。

但是架不住 RxJS 的 operator 多呀，组合起来可以实现非常复杂的异步逻辑处理。

可以在官网文档看到所有的 operator (opens new window)。

所以说，如果异步逻辑复杂度高了，那上 RxJS 收益还是很高的，异步逻辑的编写就变成了 operator 的组合，少写很多代码。

感受到为啥要用 RxJS 了么？

也是因为这个原因，Nest 的 interceptor 集成了 RxJS，可以用它来处理响应。

当然，也有人觉得这里没必要用 RxJS。

但既然 Nest 支持了，我们就用用看，基于那一堆 operator 确实是能简化异步逻辑的。

创建一个测试项目：

nest new interceptor-test -p npm

进入目录执行 nest g interceptor：

nest g interceptor aaa --flat --no-spec

我们可以这样实现接口耗时统计：

import { CallHandler, ExecutionContext, Injectable, NestInterceptor } from '@nestjs/common';
import { Observable, tap } from '0rxjs';

@Injectable()
export class AaaInterceptor implements NestInterceptor {

  intercept(context: ExecutionContext, next: CallHandler): Observable<any> {
    const now = Date.now();
    return next
      .handle()
      .pipe(
        tap(() => console.log(`After... ${Date.now() - now}ms`)),
      );
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

tap operator 不会改变数据，只是额外执行一段逻辑。

在 handler 上启用 interceptor：

然后浏览器访问 http://localhost:3000 (opens new window)

就会看到打印的耗时数据：

或者全局启用这个 interceptor：

路由级别和全局级别的 interceptor 还是有区别的，路由级别的可以注入依赖，而全局的不行：

我们再来使用下别的 RxJS operator：

其实适合在 Nest 的 interceptor 里用的 operator 还真不多，也就这么几个：

# map

再生成一个 interceptor：

nest g interceptor map-test --flat --no-spec

使用 map operator 来对 controller 返回的数据做一些修改：

import { CallHandler, ExecutionContext, Injectable, NestInterceptor } from '@nestjs/common';
import { map, Observable } from 'rxjs';

@Injectable()
export class MapTestInterceptor implements NestInterceptor {
  intercept(context: ExecutionContext, next: CallHandler): Observable<any> {
    return next.handle().pipe(map(data => {
      return {
        code: 200,
        message: 'success',
        data
      }
    }))
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

在 controller 里引入下：

跑下试试：

现在返回的数据就变成了这样。

map 算是在 nest interceptor 里必用的 rxjs operator 了

# tap

再生成个 interceptor

nest g interceptor tap-test --flat --no-spec

使用 tap operator 来添加一些日志、缓存等逻辑：

import { AppService } from './app.service';
import { CallHandler, ExecutionContext, Injectable, Logger, NestInterceptor } from '@nestjs/common';
import { Observable, tap } from 'rxjs';

@Injectable()
export class TapTestInterceptor implements NestInterceptor {
  constructor(private appService: AppService) {}

  private readonly logger = new Logger(TapTestInterceptor.name);

  intercept(context: ExecutionContext, next: CallHandler): Observable<any> {
    return next.handle().pipe(tap((data) => {
      
      // 这里是更新缓存的操作，这里模拟下
      this.appService.getHello();

      this.logger.log(`log something`, data);
    }))
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

因为还没讲到缓存那块，这里就调用 service 方法模拟了下。

日志记录我们用的 nest 内置的 Logger，在 controller 返回响应的时候记录一些东西。

浏览器访问这个接口，会打印日志：

这里我们用的是 Nest 内置的 Logger，所以打印格式是这样的。

# catchError

controller 里很可能会抛出错误，这些错误会被 exception filter 处理，返回不同的响应，但在那之前，我们可以在 interceptor 里先处理下。

生成 interceptor：

nest g interceptor catch-error-test --flat --no-spec

使用 catchError 处理抛出的异常：

import { CallHandler, ExecutionContext, Injectable, Logger, NestInterceptor } from '@nestjs/common';
import { catchError, Observable, throwError } from 'rxjs';

@Injectable()
export class CatchErrorTestInterceptor implements NestInterceptor {
  private readonly logger = new Logger(CatchErrorTestInterceptor.name)

  intercept (context: ExecutionContext, next: CallHandler): Observable<any> {
    return next.handle().pipe(catchError(err => {
      this.logger.error(err.message, err.stack)
      return throwError(() => err)
    }))
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14

这里我们就是日志记录了一下，当然你也可以改成另一种错误，重新 throwError。

在 controller 里用一下：

浏览器访问下，可以看到返回 500 的错误：

打印了两次错误：

一次是我们在 interceptor 里打印的，一次是 exception filter 打印的。

其实我们能看到这个 500 错误，就是内置的 exception filter 处理的：

对应的 Nest 源码如下：

# timeout

接口如果长时间没返回，要给用户一个接口超时的响应，这时候就可以用 timeout operator。

我们再创建个 nest interceptor

nest g interceptor timeout --flat --no-spec

添加如下逻辑：

import { CallHandler, ExecutionContext, Injectable, NestInterceptor, RequestTimeoutException } from '@nestjs/common';
import { catchError, Observable, throwError, timeout, TimeoutError } from 'rxjs';

@Injectable()
export class TimeoutInterceptor implements NestInterceptor {
  intercept(context: ExecutionContext, next: CallHandler): Observable<any> {
    return next.handle().pipe(
      timeout(3000),
      catchError(err => {
        if(err instanceof TimeoutError) {
          console.log(err);
          return throwError(() => new RequestTimeoutException());
        }
        return throwError(() => err);
      })
    )
  }
}