1 import { bufferTime, catchError, filter, map, tap, share, switchMap } from 'rxjs/operators'
2 import { Observable, ReplaySubject, Subject, of, merge } from 'rxjs'
3 import { HttpClient, HttpParams } from '@angular/common/http'
4 import { Injectable } from '@angular/core'
5 import { ResultList } from '../../../../../shared'
6 import { environment } from '../../../environments/environment'
7 import { RestExtractor, RestService } from '../rest'
8 import { VideoChannel } from '@app/shared/video-channel/video-channel.model'
9 import { VideoChannelService } from '@app/shared/video-channel/video-channel.service'
10 import { VideoChannel as VideoChannelServer } from '../../../../../shared/models/videos'
11 import { ComponentPaginationLight } from '@app/shared/rest/component-pagination.model'
12 import { uniq } from 'lodash-es'
13 import * as debug from 'debug'
15 const logger = debug('peertube:subscriptions:UserSubscriptionService')
17 type SubscriptionExistResult = { [ uri: string ]: boolean }
18 type SubscriptionExistResultObservable = { [ uri: string ]: Observable<boolean> }
21 export class UserSubscriptionService {
22 static BASE_USER_SUBSCRIPTIONS_URL = environment.apiUrl + '/api/v1/users/me/subscriptions'
24 // Use a replay subject because we "next" a value before subscribing
25 private existsSubject = new ReplaySubject<string>(1)
26 private readonly existsObservable: Observable<SubscriptionExistResult>
28 private myAccountSubscriptionCache: SubscriptionExistResult = {}
29 private myAccountSubscriptionCacheObservable: SubscriptionExistResultObservable = {}
30 private myAccountSubscriptionCacheSubject = new Subject<SubscriptionExistResult>()
33 private authHttp: HttpClient,
34 private restExtractor: RestExtractor,
35 private restService: RestService
37 this.existsObservable = merge(
38 this.existsSubject.pipe(
40 filter(uris => uris.length !== 0),
41 map(uris => uniq(uris)),
42 switchMap(uris => this.doSubscriptionsExist(uris)),
46 this.myAccountSubscriptionCacheSubject
54 deleteSubscription (nameWithHost: string) {
55 const url = UserSubscriptionService.BASE_USER_SUBSCRIPTIONS_URL + '/' + nameWithHost
57 return this.authHttp.delete(url)
59 map(this.restExtractor.extractDataBool),
61 this.myAccountSubscriptionCache[nameWithHost] = false
63 this.myAccountSubscriptionCacheSubject.next(this.myAccountSubscriptionCache)
65 catchError(err => this.restExtractor.handleError(err))
69 addSubscription (nameWithHost: string) {
70 const url = UserSubscriptionService.BASE_USER_SUBSCRIPTIONS_URL
72 const body = { uri: nameWithHost }
73 return this.authHttp.post(url, body)
75 map(this.restExtractor.extractDataBool),
77 this.myAccountSubscriptionCache[nameWithHost] = true
79 this.myAccountSubscriptionCacheSubject.next(this.myAccountSubscriptionCache)
81 catchError(err => this.restExtractor.handleError(err))
85 listSubscriptions (componentPagination: ComponentPaginationLight): Observable<ResultList<VideoChannel>> {
86 const url = UserSubscriptionService.BASE_USER_SUBSCRIPTIONS_URL
88 const pagination = this.restService.componentPaginationToRestPagination(componentPagination)
90 let params = new HttpParams()
91 params = this.restService.addRestGetParams(params, pagination)
93 return this.authHttp.get<ResultList<VideoChannelServer>>(url, { params })
95 map(res => VideoChannelService.extractVideoChannels(res)),
96 catchError(err => this.restExtractor.handleError(err))
101 * SubscriptionExist part
104 listenToMyAccountSubscriptionCacheSubject () {
105 return this.myAccountSubscriptionCacheSubject.asObservable()
108 listenToSubscriptionCacheChange (nameWithHost: string) {
109 if (nameWithHost in this.myAccountSubscriptionCacheObservable) {
110 return this.myAccountSubscriptionCacheObservable[ nameWithHost ]
113 const obs = this.existsObservable
115 filter(existsResult => existsResult[ nameWithHost ] !== undefined),
116 map(existsResult => existsResult[ nameWithHost ])
119 this.myAccountSubscriptionCacheObservable[ nameWithHost ] = obs
123 doesSubscriptionExist (nameWithHost: string) {
124 logger('Running subscription check for %d.', nameWithHost)
126 if (nameWithHost in this.myAccountSubscriptionCache) {
127 logger('Found cache for %d.', nameWithHost)
129 return of(this.myAccountSubscriptionCache[ nameWithHost ])
132 this.existsSubject.next(nameWithHost)
134 logger('Fetching from network for %d.', nameWithHost)
135 return this.existsObservable.pipe(
136 filter(existsResult => existsResult[ nameWithHost ] !== undefined),
137 map(existsResult => existsResult[ nameWithHost ]),
138 tap(result => this.myAccountSubscriptionCache[ nameWithHost ] = result)
142 private doSubscriptionsExist (uris: string[]): Observable<SubscriptionExistResult> {
143 const url = UserSubscriptionService.BASE_USER_SUBSCRIPTIONS_URL + '/exist'
144 let params = new HttpParams()
146 params = this.restService.addObjectParams(params, { uris })
148 return this.authHttp.get<SubscriptionExistResult>(url, { params })
151 this.myAccountSubscriptionCache = {
152 ...this.myAccountSubscriptionCache,
156 catchError(err => this.restExtractor.handleError(err))