NgRX 下载媒体文件和调度进度的效果；如何处理流和节流 [英] NgRX Effect for downloading media file and dispatching progress; how to deal with the stream and throttling

问题描述

我在理解和应用 @Effects 为我的剧集下载选项方面有点挣扎.我在另一个问题上得到了一些帮助，这是我最新的混合物.

快速概览:用户点击下载，分派一个 DOWNLOAD_EPISODE 动作，该动作在第一个 Effect 中捕获.下载调用返回一个 HttpEvents 流和一个最终的 HttpResponse.

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

服务episodesService:

download( episode: Episode ): 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, {报告进度:真，响应类型:'blob'});返回 this.http.request(req);}downloadSuccess( response: any ): Observable{console.log('调用下载成功'，响应);如果(响应.正文){var blob = new Blob( [ response.body ], { type: response.body.type } );console.log('blob', blob);}返回({状态:'完成'});}getHttpProgress( event: HttpEvent | HttpResponse ): Observable{开关(事件类型){案例HttpEventType.DownloadProgress:const progress = Math.round( 100 * event.loaded/event.total );返回({ ...事件，进度})；案例 HttpEventType.Response:const { 正文，类型 } = 事件；return of( { body, type, progress: 100 } );默认:返回({ ...事件，进度:0})；}}

效果:

@Effect()下载Episode$ = this.actions$.pipe(ofType( episodeActions.DOWNLOAD_EPISODE ),switchMap( ( { payload } ) => this.episodesService.download( payload ).管道(switchMap( (response: HttpEvent | HttpResponse) => this.episodesService.getHttpProgress( response ) ),//合并进度地图((响应: fromServices.DownloadProgress )=> {//更新剧集的进度//如果(响应.类型<= 3){return new episodeActions.DownloadProgressEpisodes( { ...有效载荷，下载:{进度: response.progress} });//在下载响应中传递 Blob//} 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( episodeActions.DOWNLOAD_EPISODE_SUCCESS ),switchMap( ( { payload } ) => this.episodesService.downloadSuccess( 有效载荷 ).pipe(点击( response => console.log( 'response', payload,response ) ),//catchError(err => of(new episodeActions.ProcessEpisodesFail(error))),)))

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

理想情况下，我想在那里拆分流，如果类型为 3，我想走路线 A，该路线始终使用剧集有效负载分派 episodeActions.DownloadProgressEpisodes.每当它是类型 4，即最后发出的事件时，我想在流中获取 B 路由并使用响应正文调用 episodeActions.DownloadEpisodesSuccess.

我尝试添加更多效果，但我总是遇到这样的情况，this.episodesService.download 的结果流要么是进度更新，要么是响应正文.对于进度更新，我要求剧集记录能够调用减速器并更新剧集的进度.

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

我也尝试过 last() ，它确实有效，但它没有让我分派响应操作(控制台日志确实有效)，但只有 last() 之后的 DownloadEpisodesSuccess 操作) 将被分派.

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

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

我想保留原来的问题，但是我确实遇到了限制要求，因为我为大文件调度了很多操作.我尝试了 throttle 运算符，但这会抑制发射，但也可以抑制响应主体的最后一次发射.我尝试用 partition 拆分它，但我认为这是不可能的.灯泡瞬间后，我决定为 DOWNLOAD_EPISODE 创建 2 种效果，一种仅捕获所有 event.type === 3 并对其进行限制，另一种效果使用 last 操作符，只处理 event.type === 4.

查看代码:

 @Effect( )下载Episode$ = this.actions$.pipe(ofType( episodeActions.DOWNLOAD_EPISODE ),switchMap( ( { payload } ) => this.episodesService.download( payload ).管道(switchMap( (response: HttpEvent | HttpResponse) => this.episodesService.getHttpProgress( response ) ),油门时间(500)，地图((响应: fromServices.DownloadProgress )=> {console.log('Type 3', response);//更新剧集的进度如果(响应.类型<= 3){return new episodeActions.DownloadProgressEpisodes( { ...有效载荷，下载:{进度: response.progress} });}}),catchError( error => of( new episodeActions.DownloadEpisodesFail( error ) ) ),)))@影响( )下载EpisodeLast$ = this.actions$.pipe(ofType( episodeActions.DOWNLOAD_EPISODE ),switchMap( ( { payload } ) => this.episodesService.download( payload ).管道(switchMap((响应: HttpEvent | HttpResponse) => this.episodesService.getHttpProgress( response ) ),最后的()，地图((响应: fromServices.DownloadProgress )=> {console.log('Type 4', response);如果(响应.类型 === 4){return new episodeActions.DownloadEpisodesSuccess( response, { ...payload, download: {进度: response.progress} });}}),catchError( error => of( new episodeActions.DownloadEpisodesFail( error ) ) ),)))

您认为这是最好的解决方案吗?类型 3 和类型 4 是分开的，我可以对其进行节流.

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

SampleTime(500) 似乎有效，因为它似乎在 Type 4 事件进来后没有抛出 Type 3 事件.

*update3:我刚刚发现我现在调用了两次 Download 效果，叹气.我又回到了第一阶段，试图限制来自 episodeService.download 的 HttpEvent 流.

解决方案

我觉得如果你不想有 if else 语句，你就必须创造不同的效果.

在当前操作中，您将发送一个新操作，例如DownloadEpisodeProgess，包含有效负载中的类型.然后，您将创建两个效果来侦听此操作并通过类型相应地对其进行过滤，一个效果将用于分派 DownloadProgressEpisodes，另一个用于 DownloadEpisodesSuccess.

另一种可能的解决方案是 分区运算符，但我没有t 与 NgRx 效果结合使用.

请记住，对于您进行的每次订阅，都会产生性能成本，我个人并不介意 if else 语句.

I am struggling a bit with understanding and applying @Effects for my episode Download option. I got some help in another question and this is my latest concoction.

Quick overview: User clicks download which dispatches a DOWNLOAD_EPISODE action which is captured in the first Effect. The download call returns a stream of HttpEvents and a final HttpResponse.

During the event.type === 3 I want to report download progress. When event.type === 4 the body has arrived and I can call the success which for example can create a Blob.

Service 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 } );
  }
}

The effects:

@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))),
    )
  )
)

I am pretty happy with this solution, however i do not like the If ELSE clause in the MAP as part of the DOWNLOAD_EPISODE Effect.

Ideally I want to split the stream there, if type is 3 I want to go route A which always dispatches episodeActions.DownloadProgressEpisodes with an Episode payload. Whenever it is type 4, the last emitted event, I want to got the B route in the stream and call episodeActions.DownloadEpisodesSuccess with the response body.

I tried to add more effects but I always end up in a situation the the result stream of the this.episodesService.download will either be a progress update OR a response body. For the progress update, I require the Episode record to be able to call the reducer and update the Episode's progress.

I tried to use things as setting filter( response => response.type === 4 ) after the DownloadProgressEpisodes but before DownloadEpisodesSuccess hoping it would allow for an iteration before the filter to deal with the progress and then filter the rest through to the Success Action. I then learned that is not how streams work.

I also tried last() which did work but it didn't let me dispatch the Response Actions (the console log did work), but only the DownloadEpisodesSuccess action after the last() would be dispatched.

So, if it is possible to split the stream and deal with event.type 3 differently then event.type 4 I would be interested in that.

*****UPDATE*******

I want to keep the original question, however I did run into a throttling requirement because I was dispatching a lot of actions for large files. I tried the throttle operator but that would suppress emits but could also suppress the last emit with the response body. I tried splitting it with partition but I don't think that is possible. After a light bulb moment I decided to create 2 effects for DOWNLOAD_EPISODE, one which only captures all event.type === 3 and throttles it and another effect that uses the last operator and only deals with event.type === 4.

See code:

  @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 ) ) ),
      )
    )
  )

What do you think, is this the best solution? Type 3 and 4 are split and I can throttle it.

*update2: It does create an issue where the Progress Action with progress 97% can be triggered after the Download Success action with progress 100%. Every time I run into a new challenge...

SampleTime(500) seems to work, since it doesnt seem to throw a Type 3 event after the Type 4 event came in.

*update3: I just found out that I am calling Download twice now in the effect, sigh. I am back at square one trying to throttle the HttpEvent stream coming from episodeService.download.

解决方案

I think if you don't want to have if else statement, you'll have to create different effects.

In the current one you'll dispatch a new action, e.g. DownloadEpisodeProgess, with the type in the payload. You will then create two effects that listens to this action and filter them accordingly via the type, one effect will be used to dispatch DownloadProgressEpisodes, the other one for DownloadEpisodesSuccess.

Another possible solution would be the partition operator, but I haven't used it in combination with NgRx Effects.

It is also good to keep in mind that for each subscription you make, there will be a performance cost, personally I don't really mind the if else statement.

这篇关于NgRX 下载媒体文件和调度进度的效果；如何处理流和节流的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！