import * as debug from 'debug'
-import { uniq } from 'lodash-es'
-import { asyncScheduler, merge, Observable, of, ReplaySubject, Subject } from 'rxjs'
-import { bufferTime, catchError, filter, map, observeOn, share, switchMap, tap } from 'rxjs/operators'
+import { merge, Observable, of, ReplaySubject, Subject } from 'rxjs'
+import { catchError, filter, map, switchMap, tap } from 'rxjs/operators'
import { HttpClient, HttpParams } from '@angular/common/http'
-import { Injectable, NgZone } from '@angular/core'
+import { Injectable } from '@angular/core'
import { ComponentPaginationLight, RestExtractor, RestService } from '@app/core'
-import { enterZone, leaveZone } from '@app/helpers'
+import { buildBulkObservable } from '@app/helpers'
import { Video, VideoChannel, VideoChannelService, VideoService } from '@app/shared/shared-main'
-import { ResultList, VideoChannel as VideoChannelServer, VideoSortField } from '@shared/models'
+import { ActorFollow, ResultList, VideoChannel as VideoChannelServer, VideoSortField } from '@shared/models'
import { environment } from '../../../environments/environment'
-const logger = debug('peertube:subscriptions:UserSubscriptionService')
+const debugLogger = debug('peertube:subscriptions:UserSubscriptionService')
type SubscriptionExistResult = { [ uri: string ]: boolean }
type SubscriptionExistResultObservable = { [ uri: string ]: Observable<boolean> }
@Injectable()
export class UserSubscriptionService {
static BASE_USER_SUBSCRIPTIONS_URL = environment.apiUrl + '/api/v1/users/me/subscriptions'
+ static BASE_VIDEO_CHANNELS_URL = environment.apiUrl + '/api/v1/video-channels'
+ static BASE_ACCOUNTS_URL = environment.apiUrl + '/api/v1/accounts'
// Use a replay subject because we "next" a value before subscribing
private existsSubject = new ReplaySubject<string>(1)
private authHttp: HttpClient,
private restExtractor: RestExtractor,
private videoService: VideoService,
- private restService: RestService,
- private ngZone: NgZone
+ private restService: RestService
) {
this.existsObservable = merge(
- this.existsSubject.pipe(
- // We leave Angular zone so Protractor does not get stuck
- bufferTime(500, leaveZone(this.ngZone, asyncScheduler)),
- filter(uris => uris.length !== 0),
- map(uris => uniq(uris)),
- observeOn(enterZone(this.ngZone, asyncScheduler)),
- switchMap(uris => this.doSubscriptionsExist(uris)),
- share()
- ),
+ buildBulkObservable({
+ time: 500,
+ notifierObservable: this.existsSubject,
+ bulkGet: this.doSubscriptionsExist.bind(this)
+ }).pipe(map(r => r.response)),
this.myAccountSubscriptionCacheSubject
)
}
+ listFollowers (parameters: {
+ pagination: ComponentPaginationLight
+ nameWithHost: string
+ search?: string
+ }) {
+ const { pagination, nameWithHost, search } = parameters
+
+ let url = `${UserSubscriptionService.BASE_ACCOUNTS_URL}/${nameWithHost}/followers`
+
+ let params = new HttpParams()
+ params = this.restService.addRestGetParams(params, this.restService.componentToRestPagination(pagination), '-createdAt')
+
+ if (search) {
+ const filters = this.restService.parseQueryStringFilter(search, {
+ channel: {
+ prefix: 'channel:'
+ }
+ })
+
+ if (filters.channel) {
+ url = `${UserSubscriptionService.BASE_VIDEO_CHANNELS_URL}/${filters.channel}/followers`
+ }
+
+ params = this.restService.addObjectParams(params, { search: filters.search })
+ }
+
+ return this.authHttp
+ .get<ResultList<ActorFollow>>(url, { params })
+ .pipe(
+ catchError(err => this.restExtractor.handleError(err))
+ )
+ }
+
getUserSubscriptionVideos (parameters: {
- videoPagination: ComponentPaginationLight,
- sort: VideoSortField,
+ videoPagination: ComponentPaginationLight
+ sort: VideoSortField
skipCount?: boolean
}): Observable<ResultList<Video>> {
const { videoPagination, sort, skipCount } = parameters
- const pagination = this.restService.componentPaginationToRestPagination(videoPagination)
+ const pagination = this.restService.componentToRestPagination(videoPagination)
let params = new HttpParams()
params = this.restService.addRestGetParams(params, pagination, sort)
return this.authHttp.delete(url)
.pipe(
- map(this.restExtractor.extractDataBool),
tap(() => {
this.myAccountSubscriptionCache[nameWithHost] = false
const body = { uri: nameWithHost }
return this.authHttp.post(url, body)
.pipe(
- map(this.restExtractor.extractDataBool),
tap(() => {
this.myAccountSubscriptionCache[nameWithHost] = true
)
}
- listSubscriptions (componentPagination: ComponentPaginationLight): Observable<ResultList<VideoChannel>> {
+ listSubscriptions (parameters: {
+ pagination: ComponentPaginationLight
+ search: string
+ }): Observable<ResultList<VideoChannel>> {
+ const { pagination, search } = parameters
const url = UserSubscriptionService.BASE_USER_SUBSCRIPTIONS_URL
- const pagination = this.restService.componentPaginationToRestPagination(componentPagination)
+ const restPagination = this.restService.componentToRestPagination(pagination)
let params = new HttpParams()
- params = this.restService.addRestGetParams(params, pagination)
+ params = this.restService.addRestGetParams(params, restPagination)
+ if (search) params = params.append('search', search)
return this.authHttp.get<ResultList<VideoChannelServer>>(url, { params })
.pipe(
listenToSubscriptionCacheChange (nameWithHost: string) {
if (nameWithHost in this.myAccountSubscriptionCacheObservable) {
- return this.myAccountSubscriptionCacheObservable[ nameWithHost ]
+ return this.myAccountSubscriptionCacheObservable[nameWithHost]
}
const obs = this.existsObservable
.pipe(
- filter(existsResult => existsResult[ nameWithHost ] !== undefined),
- map(existsResult => existsResult[ nameWithHost ])
+ filter(existsResult => existsResult[nameWithHost] !== undefined),
+ map(existsResult => existsResult[nameWithHost])
)
- this.myAccountSubscriptionCacheObservable[ nameWithHost ] = obs
+ this.myAccountSubscriptionCacheObservable[nameWithHost] = obs
return obs
}
doesSubscriptionExist (nameWithHost: string) {
- logger('Running subscription check for %d.', nameWithHost)
+ debugLogger('Running subscription check for %d.', nameWithHost)
if (nameWithHost in this.myAccountSubscriptionCache) {
- logger('Found cache for %d.', nameWithHost)
+ debugLogger('Found cache for %d.', nameWithHost)
- return of(this.myAccountSubscriptionCache[ nameWithHost ])
+ return of(this.myAccountSubscriptionCache[nameWithHost])
}
this.existsSubject.next(nameWithHost)
- logger('Fetching from network for %d.', nameWithHost)
+ debugLogger('Fetching from network for %d.', nameWithHost)
return this.existsObservable.pipe(
- filter(existsResult => existsResult[ nameWithHost ] !== undefined),
- map(existsResult => existsResult[ nameWithHost ]),
- tap(result => this.myAccountSubscriptionCache[ nameWithHost ] = result)
+ filter(existsResult => existsResult[nameWithHost] !== undefined),
+ map(existsResult => existsResult[nameWithHost]),
+ tap(result => this.myAccountSubscriptionCache[nameWithHost] = result)
)
}