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