看著很复杂的样子，既要 Loading，又要 Reload，还要 Retry，如果用命令式写法可能会很蛋疼，要处理各种分支，而今天要讲的 rxAsync 指令就是用来优雅地解决这个问题的。

How

我们来思考下如果解决这个问题，至少有如下四个点需要考虑。

1. 发起请求有如下三种情况：
1. 第一次渲染主动载入
2. 用户点击重新载入
3. 载入出错自动重试
2. 渲染的过程中需要根据请求的三种状态 —— loading, success, error (类似 Promise 的 pending, resolved, rejected) —— 动态渲染不同的内容
3. 输入的参数发生变化时我们需要根据最新参数重新发起请求，但是当用户输入的重试次数变化时应该忽略，因为重试次数只影响 Error 状态
4. 用户点击重新载入可能在我们的指令内部，也可能在指令外部

Show Me the Code

话不多说，上代码：

@Directive({
selector: [rxAsync],
})
export class AsyncDirective<T, P, E = HttpErrorResponse>
implements OnInit, OnDestroy {
@ObservableInput()
@Input(rxAsyncContext)
private context$!: Observable<any> // 自定义 fetcher 调用时的 this 上下文，还可以通过箭头函数、fetcher.bind(this) 等方式解决

@ObservableInput()
@Input(rxAsyncFetcher)
private fetcher$!: Observable<Callback<[P], Observable<T>>> // 自动发起请求的回调函数，参数是下面的 params，应该返回 Observable

@ObservableInput()
@Input(rxAsyncParams)
private params$!: Observable<P> // fetcher 调用时传入的参数

@Input(rxAsyncRefetch)
private refetch$$ = new Subject<void>() // 支持用户在指令外部重新发起请求，用户可能不需要，所以设置一个默认值

@ObservableInput()
@Input(rxAsyncRetryTimes)
private retryTimes$!: Observable<number> // 发送 Error 时自动重试的次数，默认不重试

private destroy$$ = new Subject<void>()
private reload$$ = new Subject<void>()

private context = {
reload: this.reload.bind(this), // 将 reload 绑定到 template 上下文中，方便用户在指令内重新发起请求
} as IAsyncDirectiveContext<T, E>

private viewRef: Nullable<ViewRef>
private sub: Nullable<Subscription>

constructor(
private templateRef: TemplateRef<any>,
private viewContainerRef: ViewContainerRef,
) {}

reload() {
this.reload$$.next()
}

ngOnInit() {
// 得益于 ObservableInput ，我们可以一次性响应所有参数的变化
combineLatest([
this.context$,
this.fetcher$,
this.params$,
this.refetch$$.pipe(startWith(null)), // 需要 startWith(null) 触发第一次请求
this.reload$$.pipe(startWith(null)), // 同上
])
.pipe(
takeUntil(this.destroy$$),
withLatestFrom(this.retryTimes$), // 忽略 retryTimes 的变更，我们只需要取得它的最新值即可
)
.subscribe(([[context, fetcher, params], retryTimes]) => {
// 如果参数变化且上次请求还没有完成时，自动取消请求忽略掉
this.disposeSub()

// 每次发起请求前都重置 loading 和 error 的状态
Object.assign(this.context, {
loading: true,
error: null,
})

this.sub = fetcher
.call(context, params)
.pipe(
retry(retryTimes), // 错误时重试
finalize(() => {
// 无论是成功还是失败，都取消 loading，并重新触发渲染
this.context.loading = false
if (this.viewRef) {
this.viewRef.detectChanges()
}
}),
)
.subscribe(
data => (this.context.$implicit = data),
error => (this.context.error = error),
)

if (this.viewRef) {
return this.viewRef.markForCheck()
}

this.viewRef = this.viewContainerRef.createEmbeddedView(
this.templateRef,
this.context,
)
})
}

ngOnDestroy() {
this.disposeSub()

this.destroy$$.next()
this.destroy$$.complete()

if (this.viewRef) {
this.viewRef.destroy()
this.viewRef = null
}
}

disposeSub() {
if (this.sub) {
this.sub.unsubscribe()
this.sub = null
}
}
}

Usage

总共 100 多行的源码，说是很优雅，那到底使用的时候优不优雅呢？来个实例看看：

@Component({
selector: rx-async-directive-demo,
template: `
<button (click)="refetch$$.next()">Refetch (Outside rxAsync)</button>
<div
*rxAsync="
let todo;
let loading = loading;
let error = error;
let reload = reload;
context: context;
fetcher: fetchTodo;
params: todoId;
refetch: refetch$$;
retryTimes: retryTimes
"
>
<button (click)="reload()">Reload</button>
loading: {{ loading }} error: {{ error | json }}
<br />
todo: {{ todo | json }}
</div>
`,
preserveWhitespaces: false,
changeDetection: ChangeDetectionStrategy.OnPush,
})
class AsyncDirectiveComponent {
context = this

@Input()
todoId = 1

@Input()
retryTimes = 0

refetch$$ = new Subject<void>()

constructor(private http: HttpClient) {}

fetchTodo(todoId: string) {
return typeof todoId === number
? this.http.get(//jsonplaceholder.typicode.com/todos/ + todoId)
: EMPTY
}
}

Online Demo

各位看官觉得如何呢？

Think More

在不使用 rxjs 的 React, Vue 中我们是如何解决这些问题的？

Advertisement

ObservableInput 与本文的 rxAsync 已发布到 @rxts/ngrx ，欢迎试用。

源码仓库：GitHub，欢迎 PR 或提交 issue 。

推荐阅读：