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'
13 const logger = debug('peertube:subscriptions:UserSubscriptionService')
15 type SubscriptionExistResult = { [ uri: string ]: boolean }
16 type SubscriptionExistResultObservable = { [ uri: string ]: Observable<boolean> }
19 export class UserSubscriptionService {
20 static BASE_USER_SUBSCRIPTIONS_URL = environment.apiUrl + '/api/v1/users/me/subscriptions'
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>
26 private myAccountSubscriptionCache: SubscriptionExistResult = {}
27 private myAccountSubscriptionCacheObservable: SubscriptionExistResultObservable = {}
28 private myAccountSubscriptionCacheSubject = new Subject<SubscriptionExistResult>()
31 private authHttp: HttpClient,
32 private restExtractor: RestExtractor,
33 private videoService: VideoService,
34 private restService: RestService,
35 private ngZone: NgZone
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)),
48 this.myAccountSubscriptionCacheSubject
52 getUserSubscriptionVideos (parameters: {
53 videoPagination: ComponentPaginationLight,
56 }): Observable<ResultList<Video>> {
57 const { videoPagination, sort, skipCount } = parameters
58 const pagination = this.restService.componentPaginationToRestPagination(videoPagination)
60 let params = new HttpParams()
61 params = this.restService.addRestGetParams(params, pagination, sort)
63 if (skipCount) params = params.set('skipCount', skipCount + '')
66 .get<ResultList<Video>>(UserSubscriptionService.BASE_USER_SUBSCRIPTIONS_URL + '/videos', { params })
68 switchMap(res => this.videoService.extractVideos(res)),
69 catchError(err => this.restExtractor.handleError(err))
77 deleteSubscription (nameWithHost: string) {
78 const url = UserSubscriptionService.BASE_USER_SUBSCRIPTIONS_URL + '/' + nameWithHost
80 return this.authHttp.delete(url)
82 map(this.restExtractor.extractDataBool),
84 this.myAccountSubscriptionCache[nameWithHost] = false
86 this.myAccountSubscriptionCacheSubject.next(this.myAccountSubscriptionCache)
88 catchError(err => this.restExtractor.handleError(err))
92 addSubscription (nameWithHost: string) {
93 const url = UserSubscriptionService.BASE_USER_SUBSCRIPTIONS_URL
95 const body = { uri: nameWithHost }
96 return this.authHttp.post(url, body)
98 map(this.restExtractor.extractDataBool),
100 this.myAccountSubscriptionCache[nameWithHost] = true
102 this.myAccountSubscriptionCacheSubject.next(this.myAccountSubscriptionCache)
104 catchError(err => this.restExtractor.handleError(err))
108 listSubscriptions (componentPagination: ComponentPaginationLight): Observable<ResultList<VideoChannel>> {
109 const url = UserSubscriptionService.BASE_USER_SUBSCRIPTIONS_URL
111 const pagination = this.restService.componentPaginationToRestPagination(componentPagination)
113 let params = new HttpParams()
114 params = this.restService.addRestGetParams(params, pagination)
116 return this.authHttp.get<ResultList<VideoChannelServer>>(url, { params })
118 map(res => VideoChannelService.extractVideoChannels(res)),
119 catchError(err => this.restExtractor.handleError(err))
124 * SubscriptionExist part
127 listenToMyAccountSubscriptionCacheSubject () {
128 return this.myAccountSubscriptionCacheSubject.asObservable()
131 listenToSubscriptionCacheChange (nameWithHost: string) {
132 if (nameWithHost in this.myAccountSubscriptionCacheObservable) {
133 return this.myAccountSubscriptionCacheObservable[ nameWithHost ]
136 const obs = this.existsObservable
138 filter(existsResult => existsResult[ nameWithHost ] !== undefined),
139 map(existsResult => existsResult[ nameWithHost ])
142 this.myAccountSubscriptionCacheObservable[ nameWithHost ] = obs
146 doesSubscriptionExist (nameWithHost: string) {
147 logger('Running subscription check for %d.', nameWithHost)
149 if (nameWithHost in this.myAccountSubscriptionCache) {
150 logger('Found cache for %d.', nameWithHost)
152 return of(this.myAccountSubscriptionCache[ nameWithHost ])
155 this.existsSubject.next(nameWithHost)
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)
165 private doSubscriptionsExist (uris: string[]): Observable<SubscriptionExistResult> {
166 const url = UserSubscriptionService.BASE_USER_SUBSCRIPTIONS_URL + '/exist'
167 let params = new HttpParams()
169 params = this.restService.addObjectParams(params, { uris })
171 return this.authHttp.get<SubscriptionExistResult>(url, { params })
174 this.myAccountSubscriptionCache = {
175 ...this.myAccountSubscriptionCache,
179 catchError(err => this.restExtractor.handleError(err))