代码之家 › 专栏 › 技术社区 › thk

Rxjs init resource once with hot observable

system.reactive observable rxjs typescript angular

thk · 技术社区 · 6 年前

我有一个服务设置,代码如下:

@Injectable()
export class StaticResourcesService {

private resourcesTree$: Observable<any>;

getSpecificLabel(resourceName: string, type: string): Observable<string> {
    return this.getResources().map(resources => {
      return jp.value(resources, `$.${resourceName}[?(@.type=="${type}")].label`);
    });
  }

private getResources(): Observable<any> {
    if (this.resourcesTree$ != null) {
      return this.resourcesTree$;
    }
    let source = this.getResourcesFilePath() // get obs<string> with a file path
      .mergeMap(path => {
        console.log(`static-resources path: ${path}`);
        return this.http.get(path).map(res => res.json()); // get json obj
      }).do(res => {
        this.resourcesTree$ = Observable.of(res);
      });
    return source;
  }

具有以下变换方法的自定义角度管道:

transform(resourceType: string, resourceName: string): Observable<string> {
    return this.staticResourcesService.getSpecificLabel(resourceName, resourceType);
  }

我在一个简单的html页面中使用了这个自定义管道:

<div *ngFor="let item of items">{{ item?.status | staticResource:'MY_RES' | async }}</div>

当我第一次加载页面时,自定义管道将处理4次(ngFor循环4x),并且永远不会从resourcesTree$获取其源数据。我理解这是一种正常行为,因为订阅将同时侦听,do将执行4次。但我想要的是一种优化的方式,只访问一次数据,然后在接下来的3次中利用resourcesTree$。我试图通过在源代码上使用share()方法使其“热门”。但我不能让它正常工作。

有人能告诉我如何做到这一点吗?也许我都错了,我真的愿意接受这里的任何建议。

非常感谢。

1 回复 | 直到 6 年前

Richard Matsen 6 年前

异步服务的常见模式是将共享资源设为Subject()或ReplaySubject()。这样,你只需连接用户,他们就可以很好地等待数据的到来。

在您的示例中,它对资源进行切片,因此这是一个映射操作符(正如您已经做的那样)。还有一种方法可以发出回迁信号起动 (不在乎它是否已经完成,因为它是可以观察到的)。

@Injectable()
export class StaticResourcesService {

private resourcesTree$ = new ReplaySubject(1);
private status = 'unloaded';

getSpecificLabel(resourceName: string, type: string): Observable<string> {
  if( this.status === 'unloaded') {
    this.status = 'loading';
    getResources();   
  }
  return resourcesTree$.map(resources => {
    return jp.value(resources, `$.${resourceName}[?(@.type=="${type}")].label`);
  });
}

private getResources(): Observable<any> {
  this.getResourcesFilePath() // get obs<string> with a file path
    .mergeMap(path => {
      console.log(`static-resources path: ${path}`);
      return this.http.get(path).map(res => res.json()); // get json obj
    })
    .subscribe(res => {
        this.resourcesTree$.next(res);
        this.status = 'loaded';
      });
  }