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

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