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

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