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