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

问题描述

我在理解和应用@Effects作为我的剧集下载"选项时有些挣扎.我在另一个问题上得到了一些帮助，这是我最近的调酒.

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.

快速概述:用户单击下载将调度DOWNLOAD_EPISODE操作，该操作在第一个效果中捕获.下载调用返回HttpEvents流和最终的HttpResponse.

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.

在event.type === 3期间，我要报告下载进度.当event.type === 4尸体到达时，我可以打电话给成功人员，例如可以创建一个Blob.

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.

服务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效果的一部分.

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.

理想情况下，我想在那儿拆分流，如果类型为3，我想走路线A，该路线始终以Episode有效载荷调度episodeActions.DownloadProgressEpisodes. 每当它是类型4(最后一个发出的事件)时，我都希望在流中获取B路由，并使用响应主体调用episodeActions.DownloadEpisodesSuccess.

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.

我试图添加更多效果，但是我总是最终遇到this.episodesService.download的结果流要么是进度更新，要么是响应正文的情况.对于进度更新，我要求Episode记录能够调用reducer并更新Episode的进度.

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.

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

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.

我还尝试了last()，它确实起作用，但是它不允许我调度响应操作(控制台日志确实起作用)，但是仅调度last()之后的DownloadEpisodesSuccess操作.

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.

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

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*******

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

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.

查看代码:

  @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被拆分了，我可以限制它.

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

* update2: 确实会产生一个问题，在进度为100％的Download Success操作之后可以触发进度为97％的Progress Action. 每当我遇到新的挑战...

*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)似乎可以正常工作，因为在出​​现Type 4事件之后似乎并没有引发Type 3事件.

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

* update3:实际上，我发现我现在两次打电话给Download，叹了口气.我回到第一个方格，试图限制来自EpisodeService.download的HttpEvent流.

*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.

推荐答案

我认为，如果您不想使用if else语句，则必须创建不同的效果.

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

在当前操作中，您将分派新操作，例如DownloadEpisodeProgess，其类型在有效载荷中. 然后，您将创建两个监听此动作的效果，并通过类型对它们进行相应的过滤，一个效果将用于调度DownloadProgressEpisodes，另一个效果用于调度DownloadEpisodesSuccess.

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.

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

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

请记住，对于您进行的每个订阅都会有性能成本，就我个人而言，我并不介意if else声明.

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 Effect，用于下载媒体文件和调度进度；如何处理流和节流的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！