client/src/app/core/notification/peertube-socket.service.ts

   1 import { Subject } from 'rxjs'
   2 import { ManagerOptions, Socket, SocketOptions } from 'socket.io-client'
   3 import { Injectable } from '@angular/core'
   4 import { LiveVideoEventPayload, LiveVideoEventType, UserNotification as UserNotificationServer } from '@shared/models'
   5 import { environment } from '../../../environments/environment'
   6 import { AuthService } from '../auth'
   7
   8 export type NotificationEvent = 'new' | 'read' | 'read-all'
   9
  10 @Injectable()
  11 export class PeerTubeSocket {
  12   private io: (uri: string, opts?: Partial<ManagerOptions & SocketOptions>) => Socket
  13
  14   private notificationSubject = new Subject<{ type: NotificationEvent, notification?: UserNotificationServer }>()
  15   private liveVideosSubject = new Subject<{ type: LiveVideoEventType, payload: LiveVideoEventPayload }>()
  16
  17   private notificationSocket: Socket
  18   private liveVideosSocket: Socket
  19
  20   constructor (
  21     private auth: AuthService
  22   ) {}
  23
  24   async getMyNotificationsSocket () {
  25     await this.initNotificationSocket()
  26
  27     return this.notificationSubject.asObservable()
  28   }
  29
  30   getLiveVideosObservable () {
  31     return this.liveVideosSubject.asObservable()
  32   }
  33
  34   async subscribeToLiveVideosSocket (videoId: number) {
  35     await this.initLiveVideosSocket()
  36
  37     this.liveVideosSocket.emit('subscribe', { videoId })
  38   }
  39
  40   unsubscribeLiveVideos (videoId: number) {
  41     if (!this.liveVideosSocket) return
  42
  43     this.liveVideosSocket.emit('unsubscribe', { videoId })
  44   }
  45
  46   dispatchNotificationEvent (type: NotificationEvent, notification?: UserNotificationServer) {
  47     this.notificationSubject.next({ type, notification })
  48   }
  49
  50   private async initNotificationSocket () {
  51     if (this.notificationSocket) return
  52
  53     await this.importIOIfNeeded()
  54
  55     this.notificationSocket = this.io(environment.apiUrl + '/user-notifications', {
  56       query: { accessToken: this.auth.getAccessToken() }
  57     })
  58
  59     this.notificationSocket.on('new-notification', (n: UserNotificationServer) => {
  60       this.dispatchNotificationEvent('new', n)
  61     })
  62   }
  63
  64   private async initLiveVideosSocket () {
  65     if (this.liveVideosSocket) return
  66
  67     await this.importIOIfNeeded()
  68
  69     this.liveVideosSocket = this.io(environment.apiUrl + '/live-videos')
  70
  71     const types: LiveVideoEventType[] = [ 'views-change', 'state-change' ]
  72
  73     for (const type of types) {
  74       this.liveVideosSocket.on(type, (payload: LiveVideoEventPayload) => {
  75         this.dispatchLiveVideoEvent(type, payload)
  76       })
  77     }
  78   }
  79
  80   private async importIOIfNeeded () {
  81     if (this.io) return
  82
  83     this.io = (await import('socket.io-client')).io
  84   }
  85
  86   private dispatchLiveVideoEvent (type: LiveVideoEventType, payload: LiveVideoEventPayload) {
  87     this.liveVideosSubject.next({ type, payload })
  88   }
  89 }