angular6 - 用于下载媒体文件和调度进度的 NgRX Effect；如何处理流和节流

Question

我在为我的剧集下载选项理解和应用@Effects 时遇到了一些困难。我在另一个问题上得到了一些帮助，这是我最新的混合物。

快速概览:用户点击下载会调度 DOWNLOAD_EPISODE 操作，该操作在第一个 Effect 中捕获。下载调用返回一个 HttpEvent 流和一个最终的 HttpResponse。

在 event.type === 3 期间，我想报告下载进度。当 event.type === 4 主体到达时，我可以调用成功，例如可以创建一个 Blob。

服务 episodesService:

download( episode: Episode ): Observable<HttpEvent<any>> | Observable<HttpResponse<any>> {

  // const url = encodeURIComponent( episode.url.url );
  const url = 'https%3A%2F%2Fwww.sample-videos.com%2Faudio%2Fmp3%2Fcrowd-cheering.mp3';
  const req = new HttpRequest( 'GET', 'http://localhost:3000/episodes/' + url, {
    reportProgress: true,
    responseType: 'blob'
  } );
  return this.http.request( req );
}

downloadSuccess( response: any ): Observable<any> {
  console.log( 'calling download success', response );
  if ( response.body ) {
    var blob = new Blob( [ response.body ], { type: response.body.type } );
    console.log( 'blob', blob );
  }
  return of( { status: 'done' } );
}

getHttpProgress( event: HttpEvent<any> | HttpResponse<Blob> ): Observable<DownloadProgress> {

  switch ( event.type ) {
    case HttpEventType.DownloadProgress:
      const progress = Math.round( 100 * event.loaded / event.total );
      return of( { ...event, progress } );

    case HttpEventType.Response:
      const { body, type } = event;
      return of( { body, type, progress: 100 } );

    default:
      return of( { ...event, progress: 0 } );
  }
}

效果:

@Effect()
downloadEpisode$ = this.actions$.pipe(
  ofType<episodeActions.DownloadEpisodes>( episodeActions.DOWNLOAD_EPISODE ),
  switchMap( ( { payload } ) => this.episodesService.download( payload )
    .pipe(
      switchMap( (response: HttpEvent<any> | HttpResponse<any>) => this.episodesService.getHttpProgress( response ) ), //merge in the progress
      map( ( response: fromServices.DownloadProgress ) => {

        // update the progress in the episode
        //
        if ( response.type <= 3 ) {
          return new episodeActions.DownloadProgressEpisodes( { ...payload, download: {
            progress: response.progress
          } }  );

        // pass the Blob on the download response
        //
        } else if ( response.type === 4 ){
          return new episodeActions.DownloadEpisodesSuccess( response );
        }
      } ),
      catchError( error => of( new episodeActions.DownloadEpisodesFail( error ) ) ),
    )
  )
)

@Effect( { dispatch: false } )
processDownloadEpisodeSuccess$ = this.actions$.pipe(
  ofType<any>( episodeActions.DOWNLOAD_EPISODE_SUCCESS ),
  switchMap( ( { payload } ) => this.episodesService
    .downloadSuccess( payload ).pipe(
      tap( response => console.log( 'response', payload,response ) ),

      //  catchError(err => of(new episodeActions.ProcessEpisodesFail(error))),
    )
  )
)

我对这个解决方案非常满意，但是我不喜欢 MAP 中的 If ELSE 子句作为 DOWNLOAD_EPISODE 效果的一部分。

理想情况下，我想在那里拆分流，如果类型是 3，我想走路由 A，它总是调度 episodeActions.DownloadProgressEpisodes 和 Episode 负载。 每当它是类型 4(最后发出的事件)时，我想在流中获取 B 路由并使用响应主体调用 episodeActions.DownloadEpisodesSuccess。

我尝试添加更多效果，但我总是遇到这样一种情况:this.episodesService.download 的结果流要么是进度更新，要么是响应主体。对于进度更新，我要求 Episode 记录能够调用 reducer 并更新 Episode 的进度。

我尝试在 DownloadProgressEpisodes 之后但在 DownloadEpisodesSuccess 之前使用设置 filter( response => response.type === 4 )希望它允许在过滤器之前进行迭代以处理进度，然后将其余部分过滤到成功操作。 然后我了解到这不是流的工作方式。

我还尝试了 last() 确实有效，但它没有让我发送响应操作(控制台日志确实有效)，但只有 last( ) 将被调度。

因此，如果可以拆分流并以不同于event.type 4 的方式处理event.type 3，我会对此感兴趣。

*****更新*******

我想保留原来的问题，但是我确实遇到了限制要求，因为我正在为大文件分派(dispatch)很多操作。 我尝试了 throttle 运算符，但这会抑制发出，但也可能抑制响应主体的最后一次发出。我尝试用 partition 拆分它，但我认为这是不可能的。在一个灯泡时刻之后，我决定为 DOWNLOAD_EPISODE 创建 2 个效果，一个只捕获所有 event.type === 3 并限制它，另一个效果使用 last 运算符，仅处理 event.type === 4。

查看代码:

  @Effect( )
  downloadEpisode$ = this.actions$.pipe(
    ofType<episodeActions.DownloadEpisodes>( episodeActions.DOWNLOAD_EPISODE ),
    switchMap( ( { payload } ) => this.episodesService.download( payload )
      .pipe(
        switchMap( (response: HttpEvent<any> | HttpResponse<any>) => this.episodesService.getHttpProgress( response ) ),
        throttleTime( 500 ),
        map( ( response: fromServices.DownloadProgress ) => {
          console.log('Type 3', response);
          // update the progress in the episode
          if ( response.type <= 3) {
            return new episodeActions.DownloadProgressEpisodes( { ...payload, download: {
              progress: response.progress
            } }  );
          }
        } ),
        catchError( error => of( new episodeActions.DownloadEpisodesFail( error ) ) ),
      )
    )
  )

  @Effect( )
  downloadEpisodeLast$ = this.actions$.pipe(
    ofType<episodeActions.DownloadEpisodes>( episodeActions.DOWNLOAD_EPISODE ),
    switchMap( ( { payload } ) => this.episodesService.download( payload )
      .pipe(
        switchMap( (response: HttpEvent<any> | HttpResponse<any>) => this.episodesService.getHttpProgress( response ) ),
        last(),
        map( ( response: fromServices.DownloadProgress ) => {
          console.log('Type 4', response);
          if ( response.type === 4 ){
            return new episodeActions.DownloadEpisodesSuccess( response, { ...payload, download: {
              progress: response.progress
            } } );
          }
        } ),

        catchError( error => of( new episodeActions.DownloadEpisodesFail( error ) ) ),
      )
    )
  )

您怎么看，这是最好的解决方案吗？类型 3 和类型 4 是分开的，我可以限制它。

*更新2: 它确实会产生一个问题，即在进度为 100% 的 Download Success 操作之后可以触发进度为 97% 的 Progress Action。 每次遇到新的挑战...

SampleTime(500) 似乎可以工作，因为在 Type 4 事件进入后它似乎没有抛出 Type 3 事件。

*update3:我刚刚发现我在效果中调用了两次Download，唉。我回到原点，试图限制来自 episodeService.download 的 HttpEvent 流。

Answer 1

我想如果你不想拥有 if else语句，你必须创建不同的效果。

在当前操作中，您将发送一个新操作，例如DownloadEpisodeProgess ，有效载荷中的类型。 然后您将创建两个监听此操作的效果并通过类型相应地过滤它们，一个效果将用于调度 DownloadProgressEpisodes , 另一个为 DownloadEpisodesSuccess .

另一种可能的解决方案是 partition operator ，但我还没有将它与 NgRx Effects 结合使用。

同样要记住，对于您进行的每个订阅，都会有性能成本，我个人并不介意 if else声明。