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 } 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   ) {
  35     this.existsObservable = merge(
  36       buildBulkObservable({
  37         time: 500,
  38         notifierObservable: this.existsSubject,
  39         bulkGet: this.doSubscriptionsExist.bind(this)
  40       }).pipe(map(r => r.response)),
  41
  42       this.myAccountSubscriptionCacheSubject
  43     )
  44   }
  45
  46   getUserSubscriptionVideos (parameters: {
  47     videoPagination: ComponentPaginationLight
  48     sort: VideoSortField
  49     skipCount?: boolean
  50   }): Observable<ResultList<Video>> {
  51     const { videoPagination, sort, skipCount } = parameters
  52     const pagination = this.restService.componentPaginationToRestPagination(videoPagination)
  53
  54     let params = new HttpParams()
  55     params = this.restService.addRestGetParams(params, pagination, sort)
  56
  57     if (skipCount) params = params.set('skipCount', skipCount + '')
  58
  59     return this.authHttp
  60                .get<ResultList<Video>>(UserSubscriptionService.BASE_USER_SUBSCRIPTIONS_URL + '/videos', { params })
  61                .pipe(
  62                  switchMap(res => this.videoService.extractVideos(res)),
  63                  catchError(err => this.restExtractor.handleError(err))
  64                )
  65   }
  66
  67   /**
  68    * Subscription part
  69    */
  70
  71   deleteSubscription (nameWithHost: string) {
  72     const url = UserSubscriptionService.BASE_USER_SUBSCRIPTIONS_URL + '/' + nameWithHost
  73
  74     return this.authHttp.delete(url)
  75                .pipe(
  76                  map(this.restExtractor.extractDataBool),
  77                  tap(() => {
  78                    this.myAccountSubscriptionCache[nameWithHost] = false
  79
  80                    this.myAccountSubscriptionCacheSubject.next(this.myAccountSubscriptionCache)
  81                  }),
  82                  catchError(err => this.restExtractor.handleError(err))
  83                )
  84   }
  85
  86   addSubscription (nameWithHost: string) {
  87     const url = UserSubscriptionService.BASE_USER_SUBSCRIPTIONS_URL
  88
  89     const body = { uri: nameWithHost }
  90     return this.authHttp.post(url, body)
  91                .pipe(
  92                  map(this.restExtractor.extractDataBool),
  93                  tap(() => {
  94                    this.myAccountSubscriptionCache[nameWithHost] = true
  95
  96                    this.myAccountSubscriptionCacheSubject.next(this.myAccountSubscriptionCache)
  97                  }),
  98                  catchError(err => this.restExtractor.handleError(err))
  99                )
 100   }
 101
 102   listSubscriptions (parameters: {
 103     pagination: ComponentPaginationLight
 104     search: string
 105   }): Observable<ResultList<VideoChannel>> {
 106     const { pagination, search } = parameters
 107     const url = UserSubscriptionService.BASE_USER_SUBSCRIPTIONS_URL
 108
 109     const restPagination = this.restService.componentPaginationToRestPagination(pagination)
 110
 111     let params = new HttpParams()
 112     params = this.restService.addRestGetParams(params, restPagination)
 113     if (search) params = params.append('search', search)
 114
 115     return this.authHttp.get<ResultList<VideoChannelServer>>(url, { params })
 116                .pipe(
 117                  map(res => VideoChannelService.extractVideoChannels(res)),
 118                  catchError(err => this.restExtractor.handleError(err))
 119                )
 120   }
 121
 122   /**
 123    * SubscriptionExist part
 124    */
 125
 126   listenToMyAccountSubscriptionCacheSubject () {
 127     return this.myAccountSubscriptionCacheSubject.asObservable()
 128   }
 129
 130   listenToSubscriptionCacheChange (nameWithHost: string) {
 131     if (nameWithHost in this.myAccountSubscriptionCacheObservable) {
 132       return this.myAccountSubscriptionCacheObservable[nameWithHost]
 133     }
 134
 135     const obs = this.existsObservable
 136                     .pipe(
 137                       filter(existsResult => existsResult[nameWithHost] !== undefined),
 138                       map(existsResult => existsResult[nameWithHost])
 139                     )
 140
 141     this.myAccountSubscriptionCacheObservable[nameWithHost] = obs
 142     return obs
 143   }
 144
 145   doesSubscriptionExist (nameWithHost: string) {
 146     logger('Running subscription check for %d.', nameWithHost)
 147
 148     if (nameWithHost in this.myAccountSubscriptionCache) {
 149       logger('Found cache for %d.', nameWithHost)
 150
 151       return of(this.myAccountSubscriptionCache[nameWithHost])
 152     }
 153
 154     this.existsSubject.next(nameWithHost)
 155
 156     logger('Fetching from network for %d.', nameWithHost)
 157     return this.existsObservable.pipe(
 158       filter(existsResult => existsResult[nameWithHost] !== undefined),
 159       map(existsResult => existsResult[nameWithHost]),
 160       tap(result => this.myAccountSubscriptionCache[nameWithHost] = result)
 161     )
 162   }
 163
 164   private doSubscriptionsExist (uris: string[]): Observable<SubscriptionExistResult> {
 165     const url = UserSubscriptionService.BASE_USER_SUBSCRIPTIONS_URL + '/exist'
 166     let params = new HttpParams()
 167
 168     params = this.restService.addObjectParams(params, { uris })
 169
 170     return this.authHttp.get<SubscriptionExistResult>(url, { params })
 171                .pipe(
 172                  tap(res => {
 173                    this.myAccountSubscriptionCache = {
 174                      ...this.myAccountSubscriptionCache,
 175                      ...res
 176                    }
 177                  }),
 178                  catchError(err => this.restExtractor.handleError(err))
 179                )
 180   }
 181 }