client/src/app/shared/shared-user-subscription/user-subscription.service.ts

   1 import * as debug from 'debug'
   2 import { uniq } from 'lodash-es'
   3 import { asyncScheduler, merge, Observable, of, ReplaySubject, Subject } from 'rxjs'
   4 import { bufferTime, catchError, filter, map, observeOn, share, switchMap, tap } from 'rxjs/operators'
   5 import { HttpClient, HttpParams } from '@angular/common/http'
   6 import { Injectable, NgZone } from '@angular/core'
   7 import { ComponentPaginationLight, RestExtractor, RestService } from '@app/core'
   8 import { enterZone, leaveZone } from '@app/helpers'
   9 import { Video, VideoChannel, VideoChannelService, VideoService } from '@app/shared/shared-main'
  10 import { ResultList, VideoChannel as VideoChannelServer, VideoSortField } from '@shared/models'
  11 import { environment } from '../../../environments/environment'
  12
  13 const logger = debug('peertube:subscriptions:UserSubscriptionService')
  14
  15 type SubscriptionExistResult = { [ uri: string ]: boolean }
  16 type SubscriptionExistResultObservable = { [ uri: string ]: Observable<boolean> }
  17
  18 @Injectable()
  19 export class UserSubscriptionService {
  20   static BASE_USER_SUBSCRIPTIONS_URL = environment.apiUrl + '/api/v1/users/me/subscriptions'
  21
  22   // Use a replay subject because we "next" a value before subscribing
  23   private existsSubject = new ReplaySubject<string>(1)
  24   private readonly existsObservable: Observable<SubscriptionExistResult>
  25
  26   private myAccountSubscriptionCache: SubscriptionExistResult = {}
  27   private myAccountSubscriptionCacheObservable: SubscriptionExistResultObservable = {}
  28   private myAccountSubscriptionCacheSubject = new Subject<SubscriptionExistResult>()
  29
  30   constructor (
  31     private authHttp: HttpClient,
  32     private restExtractor: RestExtractor,
  33     private videoService: VideoService,
  34     private restService: RestService,
  35     private ngZone: NgZone
  36   ) {
  37     this.existsObservable = merge(
  38       this.existsSubject.pipe(
  39         // We leave Angular zone so Protractor does not get stuck
  40         bufferTime(500, leaveZone(this.ngZone, asyncScheduler)),
  41         filter(uris => uris.length !== 0),
  42         map(uris => uniq(uris)),
  43         observeOn(enterZone(this.ngZone, asyncScheduler)),
  44         switchMap(uris => this.doSubscriptionsExist(uris)),
  45         share()
  46       ),
  47
  48       this.myAccountSubscriptionCacheSubject
  49     )
  50   }
  51
  52   getUserSubscriptionVideos (parameters: {
  53     videoPagination: ComponentPaginationLight,
  54     sort: VideoSortField,
  55     skipCount?: boolean
  56   }): Observable<ResultList<Video>> {
  57     const { videoPagination, sort, skipCount } = parameters
  58     const pagination = this.restService.componentPaginationToRestPagination(videoPagination)
  59
  60     let params = new HttpParams()
  61     params = this.restService.addRestGetParams(params, pagination, sort)
  62
  63     if (skipCount) params = params.set('skipCount', skipCount + '')
  64
  65     return this.authHttp
  66                .get<ResultList<Video>>(UserSubscriptionService.BASE_USER_SUBSCRIPTIONS_URL + '/videos', { params })
  67                .pipe(
  68                  switchMap(res => this.videoService.extractVideos(res)),
  69                  catchError(err => this.restExtractor.handleError(err))
  70                )
  71   }
  72
  73   /**
  74    * Subscription part
  75    */
  76
  77   deleteSubscription (nameWithHost: string) {
  78     const url = UserSubscriptionService.BASE_USER_SUBSCRIPTIONS_URL + '/' + nameWithHost
  79
  80     return this.authHttp.delete(url)
  81                .pipe(
  82                  map(this.restExtractor.extractDataBool),
  83                  tap(() => {
  84                    this.myAccountSubscriptionCache[nameWithHost] = false
  85
  86                    this.myAccountSubscriptionCacheSubject.next(this.myAccountSubscriptionCache)
  87                  }),
  88                  catchError(err => this.restExtractor.handleError(err))
  89                )
  90   }
  91
  92   addSubscription (nameWithHost: string) {
  93     const url = UserSubscriptionService.BASE_USER_SUBSCRIPTIONS_URL
  94
  95     const body = { uri: nameWithHost }
  96     return this.authHttp.post(url, body)
  97                .pipe(
  98                  map(this.restExtractor.extractDataBool),
  99                  tap(() => {
 100                    this.myAccountSubscriptionCache[nameWithHost] = true
 101
 102                    this.myAccountSubscriptionCacheSubject.next(this.myAccountSubscriptionCache)
 103                  }),
 104                  catchError(err => this.restExtractor.handleError(err))
 105                )
 106   }
 107
 108   listSubscriptions (componentPagination: ComponentPaginationLight): Observable<ResultList<VideoChannel>> {
 109     const url = UserSubscriptionService.BASE_USER_SUBSCRIPTIONS_URL
 110
 111     const pagination = this.restService.componentPaginationToRestPagination(componentPagination)
 112
 113     let params = new HttpParams()
 114     params = this.restService.addRestGetParams(params, pagination)
 115
 116     return this.authHttp.get<ResultList<VideoChannelServer>>(url, { params })
 117                .pipe(
 118                  map(res => VideoChannelService.extractVideoChannels(res)),
 119                  catchError(err => this.restExtractor.handleError(err))
 120                )
 121   }
 122
 123   /**
 124    * SubscriptionExist part
 125    */
 126
 127   listenToMyAccountSubscriptionCacheSubject () {
 128     return this.myAccountSubscriptionCacheSubject.asObservable()
 129   }
 130
 131   listenToSubscriptionCacheChange (nameWithHost: string) {
 132     if (nameWithHost in this.myAccountSubscriptionCacheObservable) {
 133       return this.myAccountSubscriptionCacheObservable[ nameWithHost ]
 134     }
 135
 136     const obs = this.existsObservable
 137                     .pipe(
 138                       filter(existsResult => existsResult[ nameWithHost ] !== undefined),
 139                       map(existsResult => existsResult[ nameWithHost ])
 140                     )
 141
 142     this.myAccountSubscriptionCacheObservable[ nameWithHost ] = obs
 143     return obs
 144   }
 145
 146   doesSubscriptionExist (nameWithHost: string) {
 147     logger('Running subscription check for %d.', nameWithHost)
 148
 149     if (nameWithHost in this.myAccountSubscriptionCache) {
 150       logger('Found cache for %d.', nameWithHost)
 151
 152       return of(this.myAccountSubscriptionCache[ nameWithHost ])
 153     }
 154
 155     this.existsSubject.next(nameWithHost)
 156
 157     logger('Fetching from network for %d.', nameWithHost)
 158     return this.existsObservable.pipe(
 159       filter(existsResult => existsResult[ nameWithHost ] !== undefined),
 160       map(existsResult => existsResult[ nameWithHost ]),
 161       tap(result => this.myAccountSubscriptionCache[ nameWithHost ] = result)
 162     )
 163   }
 164
 165   private doSubscriptionsExist (uris: string[]): Observable<SubscriptionExistResult> {
 166     const url = UserSubscriptionService.BASE_USER_SUBSCRIPTIONS_URL + '/exist'
 167     let params = new HttpParams()
 168
 169     params = this.restService.addObjectParams(params, { uris })
 170
 171     return this.authHttp.get<SubscriptionExistResult>(url, { params })
 172                .pipe(
 173                  tap(res => {
 174                    this.myAccountSubscriptionCache = {
 175                      ...this.myAccountSubscriptionCache,
 176                      ...res
 177                    }
 178                  }),
 179                  catchError(err => this.restExtractor.handleError(err))
 180                )
 181   }
 182 }