-
-
Notifications
You must be signed in to change notification settings - Fork 7.4k
/
interceptors-consumer.ts
62 lines (58 loc) 路 1.8 KB
/
interceptors-consumer.ts
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
import { NestInterceptor, Type } from '@nestjs/common';
import {
CallHandler,
ContextType,
Controller,
} from '@nestjs/common/interfaces';
import { isEmpty } from '@nestjs/common/utils/shared.utils';
import { defer, from as fromPromise, Observable } from 'rxjs';
import { mergeAll, switchMap } from 'rxjs/operators';
import { ExecutionContextHost } from '../helpers/execution-context-host';
export class InterceptorsConsumer {
public async intercept<TContext extends string = ContextType>(
interceptors: NestInterceptor[],
args: unknown[],
instance: Controller,
callback: (...args: unknown[]) => unknown,
next: () => Promise<unknown>,
type?: TContext,
): Promise<unknown> {
if (isEmpty(interceptors)) {
return next();
}
const context = this.createContext(args, instance, callback);
context.setType<TContext>(type);
const start$ = defer(() => this.transformDeferred(next));
const nextFn =
(i = 0) =>
async () => {
if (i >= interceptors.length) {
return start$;
}
const handler: CallHandler = {
handle: () => fromPromise(nextFn(i + 1)()).pipe(mergeAll()),
};
return interceptors[i].intercept(context, handler);
};
return nextFn()();
}
public createContext(
args: unknown[],
instance: Controller,
callback: (...args: unknown[]) => unknown,
): ExecutionContextHost {
return new ExecutionContextHost(
args,
instance.constructor as Type<unknown>,
callback,
);
}
public transformDeferred(next: () => Promise<any>): Observable<any> {
return fromPromise(next()).pipe(
switchMap(res => {
const isDeferred = res instanceof Promise || res instanceof Observable;
return isDeferred ? res : Promise.resolve(res);
}),
);
}
}