mirror of
https://github.com/mautrix/whatsapp.git
synced 2026-08-24 13:04:56 -04:00
720 lines
20 KiB
Go
720 lines
20 KiB
Go
package voip
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
lkpcm "github.com/livekit/media-sdk"
|
|
livekitproto "github.com/livekit/protocol/livekit"
|
|
"github.com/livekit/protocol/logger"
|
|
lksdk "github.com/livekit/server-sdk-go/v2"
|
|
lkmedia "github.com/livekit/server-sdk-go/v2/pkg/media"
|
|
"github.com/pion/rtp/codecs"
|
|
"github.com/pion/webrtc/v4"
|
|
"github.com/pion/webrtc/v4/pkg/media/samplebuilder"
|
|
"github.com/rs/zerolog"
|
|
)
|
|
|
|
type LiveKitParticipant struct {
|
|
cfg LiveKitConfig
|
|
videoCfg VideoConfig
|
|
log zerolog.Logger
|
|
room *lksdk.Room
|
|
audio *lkmedia.PCMLocalTrack
|
|
audioPub *lksdk.LocalTrackPublication
|
|
audioSrc *MeowcallerAudioSource
|
|
video *lksdk.LocalTrack
|
|
videoPub *lksdk.LocalTrackPublication
|
|
screen *lksdk.LocalTrack
|
|
screenPub *lksdk.LocalTrackPublication
|
|
|
|
mu sync.Mutex
|
|
remoteAudio []*lkmedia.PCMRemoteTrack
|
|
remoteMediaCancel context.CancelFunc
|
|
remoteVideoPLI lksdk.PLIWriter
|
|
remoteVideoSSRC webrtc.SSRC
|
|
remoteScreenPLI lksdk.PLIWriter
|
|
remoteScreenSSRC webrtc.SSRC
|
|
remoteScreenActive bool
|
|
remoteVideoKeyframePending [2]bool
|
|
remoteVideoKeyframeAwaited [2]bool
|
|
disconnected bool
|
|
selectedRemoteParticipant string
|
|
remoteAudioMuteStateChange func(muted bool)
|
|
remoteVideoFrame func(frame LiveKitVideoFrame) error
|
|
remoteVideoMuteStateChange func(muted bool)
|
|
remoteScreenFrame func(frame LiveKitVideoFrame) error
|
|
remoteScreenMuteStateChange func(muted bool)
|
|
}
|
|
|
|
type liveKitVideoSource uint8
|
|
|
|
const (
|
|
liveKitVideoSourceCamera liveKitVideoSource = iota
|
|
liveKitVideoSourceScreenShare
|
|
)
|
|
|
|
func ConnectLiveKitParticipant(ctx context.Context, authResp *LiveKitAuthResponse, cfg LiveKitConfig, videoCfg VideoConfig, log zerolog.Logger) (*LiveKitParticipant, error) {
|
|
if authResp == nil {
|
|
return nil, fmt.Errorf("livekit auth response is nil")
|
|
}
|
|
if authResp.ConnectionURL() == "" || authResp.JWT() == "" {
|
|
return nil, fmt.Errorf("livekit auth response did not include both URL and token")
|
|
}
|
|
remoteMediaCtx, remoteMediaCancel := context.WithCancel(context.Background())
|
|
participant := &LiveKitParticipant{
|
|
cfg: cfg,
|
|
videoCfg: videoCfg,
|
|
log: log,
|
|
audioSrc: NewMeowcallerAudioSource(12),
|
|
remoteMediaCancel: remoteMediaCancel,
|
|
}
|
|
callback := &lksdk.RoomCallback{
|
|
ParticipantCallback: lksdk.ParticipantCallback{
|
|
OnTrackSubscribed: func(track *webrtc.TrackRemote, publication *lksdk.RemoteTrackPublication, rp *lksdk.RemoteParticipant) {
|
|
participant.onTrackSubscribed(remoteMediaCtx, track, publication, rp)
|
|
},
|
|
OnTrackUnsubscribed: participant.onTrackUnsubscribed,
|
|
OnTrackMuted: participant.onTrackMuted,
|
|
OnTrackUnmuted: participant.onTrackUnmuted,
|
|
},
|
|
OnDisconnected: func() {
|
|
participant.closeRemoteTracks()
|
|
},
|
|
OnDisconnectedWithReason: func(reason lksdk.DisconnectionReason) {
|
|
log.Info().Str("reason", string(reason)).Msg("Disconnected from LiveKit")
|
|
participant.closeRemoteTracks()
|
|
},
|
|
}
|
|
opts := []lksdk.ConnectOption{
|
|
lksdk.WithAutoSubscribe(cfg.AutoSubscribe),
|
|
}
|
|
if cfg.ConnectTimeout > 0 {
|
|
opts = append(opts, lksdk.WithConnectTimeout(cfg.ConnectTimeout))
|
|
}
|
|
room, err := connectLiveKit(ctx, authResp.ConnectionURL(), authResp.JWT(), callback, opts...)
|
|
if err != nil {
|
|
remoteMediaCancel()
|
|
return nil, err
|
|
}
|
|
participant.room = room
|
|
return participant, nil
|
|
}
|
|
|
|
func connectLiveKit(ctx context.Context, url, token string, callback *lksdk.RoomCallback, opts ...lksdk.ConnectOption) (*lksdk.Room, error) {
|
|
room := lksdk.NewRoom(callback)
|
|
if err := room.JoinWithContextAndToken(ctx, url, token, opts...); err != nil {
|
|
return nil, err
|
|
}
|
|
return room, nil
|
|
}
|
|
|
|
func (p *LiveKitParticipant) SetRemoteAudioMuteHandler(selectedParticipant string, handler func(muted bool)) {
|
|
p.mu.Lock()
|
|
p.selectedRemoteParticipant = selectedParticipant
|
|
p.remoteAudioMuteStateChange = handler
|
|
p.mu.Unlock()
|
|
}
|
|
|
|
func (p *LiveKitParticipant) SetRemoteVideoHandlers(selectedParticipant string, frameHandler func(frame LiveKitVideoFrame) error, muteHandler func(muted bool)) {
|
|
p.mu.Lock()
|
|
p.selectedRemoteParticipant = selectedParticipant
|
|
p.remoteVideoFrame = frameHandler
|
|
p.remoteVideoMuteStateChange = muteHandler
|
|
p.mu.Unlock()
|
|
}
|
|
|
|
func (p *LiveKitParticipant) SetRemoteScreenShareHandlers(frameHandler func(frame LiveKitVideoFrame) error, muteHandler func(muted bool)) {
|
|
p.mu.Lock()
|
|
p.remoteScreenFrame = frameHandler
|
|
p.remoteScreenMuteStateChange = muteHandler
|
|
p.mu.Unlock()
|
|
}
|
|
|
|
func (p *LiveKitParticipant) requestRemoteVideoKeyframe() bool {
|
|
p.mu.Lock()
|
|
if p.disconnected {
|
|
p.mu.Unlock()
|
|
return false
|
|
}
|
|
pli := p.remoteVideoPLI
|
|
ssrc := p.remoteVideoSSRC
|
|
source := liveKitVideoSourceCamera
|
|
if p.remoteScreenActive && p.remoteScreenPLI != nil && p.remoteScreenSSRC != 0 {
|
|
pli = p.remoteScreenPLI
|
|
ssrc = p.remoteScreenSSRC
|
|
source = liveKitVideoSourceScreenShare
|
|
} else if p.remoteScreenActive {
|
|
source = liveKitVideoSourceScreenShare
|
|
pli = nil
|
|
ssrc = 0
|
|
}
|
|
if pli == nil || ssrc == 0 {
|
|
p.remoteVideoKeyframePending[source] = true
|
|
p.mu.Unlock()
|
|
return false
|
|
}
|
|
p.remoteVideoKeyframePending[source] = false
|
|
p.mu.Unlock()
|
|
p.sendRemoteVideoPLI(source, pli, ssrc)
|
|
return true
|
|
}
|
|
|
|
func (p *LiveKitParticipant) setRemoteVideoPLI(pli lksdk.PLIWriter, ssrc webrtc.SSRC) {
|
|
p.setRemoteVideoPLIForSource(liveKitVideoSourceCamera, pli, ssrc)
|
|
}
|
|
|
|
func (p *LiveKitParticipant) setRemoteVideoPLIForSource(source liveKitVideoSource, pli lksdk.PLIWriter, ssrc webrtc.SSRC) {
|
|
p.mu.Lock()
|
|
if p.disconnected {
|
|
p.mu.Unlock()
|
|
return
|
|
}
|
|
if source == liveKitVideoSourceScreenShare {
|
|
p.remoteScreenPLI = pli
|
|
p.remoteScreenSSRC = ssrc
|
|
} else {
|
|
p.remoteVideoPLI = pli
|
|
p.remoteVideoSSRC = ssrc
|
|
}
|
|
pending := p.remoteVideoKeyframePending[source] && pli != nil && ssrc != 0
|
|
if pending {
|
|
p.remoteVideoKeyframePending[source] = false
|
|
}
|
|
p.mu.Unlock()
|
|
if pending {
|
|
p.sendRemoteVideoPLI(source, pli, ssrc)
|
|
}
|
|
}
|
|
|
|
func (p *LiveKitParticipant) sendRemoteVideoPLI(source liveKitVideoSource, pli lksdk.PLIWriter, ssrc webrtc.SSRC) {
|
|
p.mu.Lock()
|
|
if !p.disconnected {
|
|
p.remoteVideoKeyframeAwaited[source] = true
|
|
}
|
|
p.mu.Unlock()
|
|
pli(ssrc)
|
|
p.log.Info().Uint32("ssrc", uint32(ssrc)).Msg("Requested LiveKit H.264 keyframe for WhatsApp peer")
|
|
}
|
|
|
|
func (p *LiveKitParticipant) PublishAudioTrack(name string) error {
|
|
p.mu.Lock()
|
|
defer p.mu.Unlock()
|
|
if p.room == nil {
|
|
return fmt.Errorf("livekit room is not connected")
|
|
}
|
|
if p.audio != nil {
|
|
return nil
|
|
}
|
|
track, err := lkmedia.NewPCMLocalTrack(meowcallerSampleRate, 1, logger.GetLogger())
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if name == "" {
|
|
name = "whatsapp-audio"
|
|
}
|
|
pub, err := p.room.LocalParticipant.PublishTrack(track, &lksdk.TrackPublicationOptions{Name: name})
|
|
if err != nil {
|
|
track.Close()
|
|
return err
|
|
}
|
|
p.audio = track
|
|
p.audioPub = pub
|
|
return nil
|
|
}
|
|
|
|
func (p *LiveKitParticipant) PublishVideoTrack(name string) error {
|
|
p.mu.Lock()
|
|
defer p.mu.Unlock()
|
|
if p.room == nil {
|
|
return fmt.Errorf("livekit room is not connected")
|
|
}
|
|
if p.video != nil {
|
|
return nil
|
|
}
|
|
track, err := lksdk.NewLocalTrack(webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeH264, ClockRate: liveKitH264ClockRate})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if name == "" {
|
|
name = "whatsapp-video"
|
|
}
|
|
pub, err := p.room.LocalParticipant.PublishTrack(
|
|
track,
|
|
videoTrackPublicationOptions(name, livekitproto.TrackSource_CAMERA, p.videoCfg),
|
|
)
|
|
if err != nil {
|
|
_ = track.Close()
|
|
return err
|
|
}
|
|
p.video = track
|
|
p.videoPub = pub
|
|
return nil
|
|
}
|
|
|
|
func (p *LiveKitParticipant) PublishScreenShareTrack(name string) error {
|
|
p.mu.Lock()
|
|
defer p.mu.Unlock()
|
|
if p.room == nil {
|
|
return fmt.Errorf("livekit room is not connected")
|
|
}
|
|
if p.screen != nil {
|
|
return nil
|
|
}
|
|
track, err := lksdk.NewLocalTrack(webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeH264, ClockRate: liveKitH264ClockRate})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if name == "" {
|
|
name = "whatsapp-screen"
|
|
}
|
|
pub, err := p.room.LocalParticipant.PublishTrack(
|
|
track,
|
|
videoTrackPublicationOptions(name, livekitproto.TrackSource_SCREEN_SHARE, p.videoCfg),
|
|
)
|
|
if err != nil {
|
|
_ = track.Close()
|
|
return err
|
|
}
|
|
pub.SetMuted(true)
|
|
p.screen = track
|
|
p.screenPub = pub
|
|
return nil
|
|
}
|
|
|
|
func videoTrackPublicationOptions(name string, source livekitproto.TrackSource, cfg VideoConfig) *lksdk.TrackPublicationOptions {
|
|
return &lksdk.TrackPublicationOptions{
|
|
Name: name,
|
|
Source: source,
|
|
VideoWidth: cfg.MaxWidth,
|
|
VideoHeight: cfg.MaxHeight,
|
|
}
|
|
}
|
|
|
|
func (p *LiveKitParticipant) SetWhatsAppAudioMuted(muted bool) {
|
|
p.mu.Lock()
|
|
pub := p.audioPub
|
|
p.mu.Unlock()
|
|
if pub == nil {
|
|
return
|
|
}
|
|
pub.SetMuted(muted)
|
|
p.log.Debug().Bool("muted", muted).Msg("Set LiveKit WhatsApp audio mute state")
|
|
}
|
|
|
|
func (p *LiveKitParticipant) SetWhatsAppVideoMuted(muted bool) {
|
|
p.mu.Lock()
|
|
pub := p.videoPub
|
|
p.mu.Unlock()
|
|
if pub == nil {
|
|
return
|
|
}
|
|
pub.SetMuted(muted)
|
|
p.log.Debug().Bool("muted", muted).Msg("Set LiveKit WhatsApp video mute state")
|
|
}
|
|
|
|
func (p *LiveKitParticipant) SetWhatsAppScreenShareMuted(muted bool) {
|
|
p.mu.Lock()
|
|
pub := p.screenPub
|
|
p.mu.Unlock()
|
|
if pub == nil {
|
|
return
|
|
}
|
|
pub.SetMuted(muted)
|
|
p.log.Debug().Bool("muted", muted).Msg("Set LiveKit WhatsApp screen-share mute state")
|
|
}
|
|
|
|
func (p *LiveKitParticipant) SetWhatsAppVideoOrientation(orientation int) {
|
|
p.mu.Lock()
|
|
video := p.video
|
|
p.mu.Unlock()
|
|
if video == nil {
|
|
return
|
|
}
|
|
if setLiveKitVideoOrientation(video, orientation) {
|
|
p.log.Debug().Int("orientation", orientation&0x03).Msg("Set LiveKit WhatsApp video orientation")
|
|
}
|
|
}
|
|
|
|
func (p *LiveKitParticipant) WhatsAppSink() *LiveKitPCMWriter {
|
|
p.mu.Lock()
|
|
defer p.mu.Unlock()
|
|
return NewLiveKitPCMWriter(p.audio)
|
|
}
|
|
|
|
func (p *LiveKitParticipant) WhatsAppVideoSink() *LiveKitH264Writer {
|
|
p.mu.Lock()
|
|
defer p.mu.Unlock()
|
|
return NewLiveKitH264Writer(p.video, videoFrameDuration(p.videoCfg))
|
|
}
|
|
|
|
func (p *LiveKitParticipant) WhatsAppScreenShareSink() *LiveKitH264Writer {
|
|
p.mu.Lock()
|
|
defer p.mu.Unlock()
|
|
return NewLiveKitH264Writer(p.screen, videoFrameDuration(p.videoCfg))
|
|
}
|
|
|
|
func (p *LiveKitParticipant) MatrixAudioSource() *MeowcallerAudioSource {
|
|
return p.audioSrc
|
|
}
|
|
|
|
func (p *LiveKitParticipant) WriteWhatsAppFrame(frame []float32) error {
|
|
p.mu.Lock()
|
|
audio := p.audio
|
|
p.mu.Unlock()
|
|
if audio == nil {
|
|
return nil
|
|
}
|
|
return audio.WriteSample(Float32FrameToPCM16(frame))
|
|
}
|
|
|
|
func (p *LiveKitParticipant) Close() {
|
|
p.mu.Lock()
|
|
if p.disconnected {
|
|
p.mu.Unlock()
|
|
return
|
|
}
|
|
p.disconnected = true
|
|
room := p.room
|
|
audio := p.audio
|
|
audioPub := p.audioPub
|
|
video := p.video
|
|
videoPub := p.videoPub
|
|
screen := p.screen
|
|
screenPub := p.screenPub
|
|
p.room = nil
|
|
p.audio = nil
|
|
p.audioPub = nil
|
|
p.video = nil
|
|
p.videoPub = nil
|
|
p.screen = nil
|
|
p.screenPub = nil
|
|
p.mu.Unlock()
|
|
p.closeRemoteTracks()
|
|
if audioPub != nil {
|
|
audioPub.SetMuted(true)
|
|
}
|
|
if videoPub != nil {
|
|
videoPub.SetMuted(true)
|
|
}
|
|
if screenPub != nil {
|
|
screenPub.SetMuted(true)
|
|
}
|
|
if audio != nil {
|
|
audio.ClearQueue()
|
|
_ = audio.Close()
|
|
}
|
|
if video != nil {
|
|
_ = video.Close()
|
|
}
|
|
if screen != nil {
|
|
_ = screen.Close()
|
|
}
|
|
if room != nil {
|
|
room.Disconnect()
|
|
}
|
|
_ = p.audioSrc.Close()
|
|
}
|
|
|
|
func (p *LiveKitParticipant) onTrackSubscribed(ctx context.Context, track *webrtc.TrackRemote, publication *lksdk.RemoteTrackPublication, rp *lksdk.RemoteParticipant) {
|
|
switch track.Kind() {
|
|
case webrtc.RTPCodecTypeAudio:
|
|
p.onAudioTrackSubscribed(track, publication, rp)
|
|
case webrtc.RTPCodecTypeVideo:
|
|
p.onVideoTrackSubscribed(ctx, track, publication, rp)
|
|
}
|
|
}
|
|
|
|
func (p *LiveKitParticipant) onAudioTrackSubscribed(track *webrtc.TrackRemote, publication *lksdk.RemoteTrackPublication, rp *lksdk.RemoteParticipant) {
|
|
if track.Codec().MimeType != webrtc.MimeTypeOpus {
|
|
p.log.Warn().
|
|
Str("codec", track.Codec().MimeType).
|
|
Str("participant", string(rp.Identity())).
|
|
Msg("Ignoring non-Opus LiveKit audio track")
|
|
return
|
|
}
|
|
remote, err := lkmedia.NewPCMRemoteTrack(
|
|
track,
|
|
p.audioSrc,
|
|
lkmedia.WithTargetSampleRate(meowcallerSampleRate),
|
|
lkmedia.WithTargetChannels(1),
|
|
lkmedia.WithLogger(logger.GetLogger()),
|
|
)
|
|
if err != nil {
|
|
p.log.Warn().
|
|
Err(err).
|
|
Str("participant", string(rp.Identity())).
|
|
Msg("Failed to subscribe LiveKit audio track")
|
|
return
|
|
}
|
|
p.mu.Lock()
|
|
p.remoteAudio = append(p.remoteAudio, remote)
|
|
p.mu.Unlock()
|
|
p.handleRemoteAudioMuteState(publication, rp, publication.IsMuted())
|
|
_ = publication
|
|
}
|
|
|
|
func (p *LiveKitParticipant) onVideoTrackSubscribed(ctx context.Context, track *webrtc.TrackRemote, publication *lksdk.RemoteTrackPublication, rp *lksdk.RemoteParticipant) {
|
|
if !p.videoCfg.Enabled {
|
|
return
|
|
}
|
|
if !remoteParticipantSelected(p.selectedParticipant(), string(rp.Identity())) {
|
|
p.log.Debug().
|
|
Str("participant", string(rp.Identity())).
|
|
Str("selected_participant", p.selectedParticipant()).
|
|
Msg("Ignoring LiveKit video track from non-selected participant")
|
|
return
|
|
}
|
|
if !strings.EqualFold(track.Codec().MimeType, webrtc.MimeTypeH264) {
|
|
p.log.Warn().
|
|
Str("codec", track.Codec().MimeType).
|
|
Str("participant", string(rp.Identity())).
|
|
Msg("Ignoring unsupported LiveKit video track; only H.264 passthrough is implemented")
|
|
p.handleRemoteVideoMuteState(publication, rp, true)
|
|
return
|
|
}
|
|
source := liveKitVideoSourceRole(publication.Source())
|
|
p.setRemoteVideoPLIForSource(source, rp.WritePLI, track.SSRC())
|
|
p.handleRemoteVideoMuteState(publication, rp, publication.IsMuted())
|
|
go p.forwardRemoteH264Track(ctx, track, rp, source)
|
|
}
|
|
|
|
func (p *LiveKitParticipant) onTrackUnsubscribed(track *webrtc.TrackRemote, publication *lksdk.RemoteTrackPublication, rp *lksdk.RemoteParticipant) {
|
|
if track.Kind() == webrtc.RTPCodecTypeVideo {
|
|
source := liveKitVideoSourceRole(publication.Source())
|
|
p.mu.Lock()
|
|
if source == liveKitVideoSourceScreenShare && p.remoteScreenSSRC == track.SSRC() {
|
|
p.remoteScreenPLI = nil
|
|
p.remoteScreenSSRC = 0
|
|
p.remoteScreenActive = false
|
|
p.remoteVideoKeyframePending[liveKitVideoSourceScreenShare] = true
|
|
} else if source == liveKitVideoSourceCamera && p.remoteVideoSSRC == track.SSRC() {
|
|
p.remoteVideoPLI = nil
|
|
p.remoteVideoSSRC = 0
|
|
p.remoteVideoKeyframePending[liveKitVideoSourceCamera] = true
|
|
}
|
|
p.mu.Unlock()
|
|
p.handleRemoteVideoMuteState(publication, rp, true)
|
|
}
|
|
}
|
|
|
|
func (p *LiveKitParticipant) onTrackMuted(pub lksdk.TrackPublication, participant lksdk.Participant) {
|
|
p.handleRemoteAudioMuteState(pub, participant, true)
|
|
p.handleRemoteVideoMuteState(pub, participant, true)
|
|
}
|
|
|
|
func (p *LiveKitParticipant) onTrackUnmuted(pub lksdk.TrackPublication, participant lksdk.Participant) {
|
|
p.handleRemoteAudioMuteState(pub, participant, false)
|
|
p.handleRemoteVideoMuteState(pub, participant, false)
|
|
}
|
|
|
|
func (p *LiveKitParticipant) handleRemoteAudioMuteState(pub lksdk.TrackPublication, participant lksdk.Participant, muted bool) {
|
|
if pub == nil || participant == nil || pub.Kind() != lksdk.TrackKindAudio {
|
|
return
|
|
}
|
|
if _, ok := participant.(*lksdk.RemoteParticipant); !ok {
|
|
return
|
|
}
|
|
identity := participant.Identity()
|
|
p.mu.Lock()
|
|
selected := p.selectedRemoteParticipant
|
|
handler := p.remoteAudioMuteStateChange
|
|
p.mu.Unlock()
|
|
if selected != "" && identity != selected {
|
|
p.log.Debug().
|
|
Str("participant", identity).
|
|
Str("selected_participant", selected).
|
|
Bool("muted", muted).
|
|
Msg("Ignoring LiveKit mute state from non-selected participant")
|
|
return
|
|
}
|
|
p.log.Debug().
|
|
Str("participant", identity).
|
|
Str("track_id", pub.SID()).
|
|
Bool("muted", muted).
|
|
Msg("Observed LiveKit remote audio mute state")
|
|
if handler != nil {
|
|
handler(muted)
|
|
}
|
|
}
|
|
|
|
func (p *LiveKitParticipant) handleRemoteVideoMuteState(pub lksdk.TrackPublication, participant lksdk.Participant, muted bool) {
|
|
if pub == nil || participant == nil || pub.Kind() != lksdk.TrackKindVideo {
|
|
return
|
|
}
|
|
if _, ok := participant.(*lksdk.RemoteParticipant); !ok {
|
|
return
|
|
}
|
|
identity := participant.Identity()
|
|
p.mu.Lock()
|
|
selected := p.selectedRemoteParticipant
|
|
p.mu.Unlock()
|
|
if !remoteParticipantSelected(selected, identity) {
|
|
p.log.Debug().
|
|
Str("participant", identity).
|
|
Str("selected_participant", selected).
|
|
Bool("muted", muted).
|
|
Msg("Ignoring LiveKit video mute state from non-selected participant")
|
|
return
|
|
}
|
|
source := liveKitVideoSourceRole(pub.Source())
|
|
p.mu.Lock()
|
|
handler := p.remoteVideoMuteStateChange
|
|
if source == liveKitVideoSourceScreenShare {
|
|
p.remoteScreenActive = !muted
|
|
handler = p.remoteScreenMuteStateChange
|
|
}
|
|
p.mu.Unlock()
|
|
p.log.Debug().
|
|
Str("participant", identity).
|
|
Str("track_id", pub.SID()).
|
|
Bool("muted", muted).
|
|
Msg("Observed LiveKit remote video mute state")
|
|
if handler != nil {
|
|
handler(muted)
|
|
}
|
|
}
|
|
|
|
func (p *LiveKitParticipant) forwardRemoteH264Track(ctx context.Context, track *webrtc.TrackRemote, rp *lksdk.RemoteParticipant, source liveKitVideoSource) {
|
|
builder := samplebuilder.New(
|
|
liveKitH264MaxLatePackets,
|
|
&codecs.H264Packet{},
|
|
track.Codec().ClockRate,
|
|
)
|
|
var parameterSets h264ParameterSetRepeater
|
|
loggedIDR := false
|
|
p.log.Info().
|
|
Str("participant", string(rp.Identity())).
|
|
Str("track_id", track.ID()).
|
|
Str("source", source.String()).
|
|
Str("fmtp", track.Codec().SDPFmtpLine).
|
|
Msg("Started forwarding LiveKit H.264 video to WhatsApp")
|
|
for {
|
|
if ctx.Err() != nil {
|
|
return
|
|
}
|
|
packet, _, err := track.ReadRTP()
|
|
if err != nil {
|
|
if !errors.Is(err, io.EOF) && ctx.Err() == nil {
|
|
p.log.Debug().
|
|
Err(err).
|
|
Str("participant", string(rp.Identity())).
|
|
Str("track_id", track.ID()).
|
|
Msg("Stopped reading LiveKit H.264 video track")
|
|
}
|
|
return
|
|
}
|
|
builder.Push(packet)
|
|
for sample := builder.Pop(); sample != nil; sample = builder.Pop() {
|
|
if len(sample.Data) == 0 {
|
|
continue
|
|
}
|
|
accessUnit, repeatedParameterSets := parameterSets.Normalize(sample.Data)
|
|
nalTypes, profileLevelID, hasIDR, hasSPS, hasPPS := h264AccessUnitMetadata(accessUnit)
|
|
p.mu.Lock()
|
|
afterPLI := hasIDR && p.remoteVideoKeyframeAwaited[source]
|
|
if afterPLI {
|
|
p.remoteVideoKeyframeAwaited[source] = false
|
|
}
|
|
p.mu.Unlock()
|
|
if hasIDR && (!loggedIDR || afterPLI || repeatedParameterSets) {
|
|
p.log.Info().
|
|
Str("participant", string(rp.Identity())).
|
|
Str("track_id", track.ID()).
|
|
Ints("nal_types", nalTypes).
|
|
Str("profile_level_id", profileLevelID).
|
|
Int("bytes", len(accessUnit)).
|
|
Bool("has_sps", hasSPS).
|
|
Bool("has_pps", hasPPS).
|
|
Bool("after_pli", afterPLI).
|
|
Bool("repeated_parameter_sets", repeatedParameterSets).
|
|
Msg("Forwarding decoder-safe LiveKit H.264 keyframe to WhatsApp")
|
|
loggedIDR = true
|
|
}
|
|
p.mu.Lock()
|
|
handler := p.remoteVideoFrame
|
|
if source == liveKitVideoSourceScreenShare {
|
|
handler = p.remoteScreenFrame
|
|
}
|
|
p.mu.Unlock()
|
|
if handler == nil {
|
|
continue
|
|
}
|
|
if err = handler(LiveKitVideoFrame{
|
|
AccessUnit: accessUnit,
|
|
Duration: sample.Duration,
|
|
}); err != nil {
|
|
p.log.Warn().
|
|
Err(err).
|
|
Str("participant", string(rp.Identity())).
|
|
Str("track_id", track.ID()).
|
|
Str("source", source.String()).
|
|
Msg("Failed to forward LiveKit H.264 frame to WhatsApp")
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func liveKitVideoSourceRole(source livekitproto.TrackSource) liveKitVideoSource {
|
|
if source == livekitproto.TrackSource_SCREEN_SHARE {
|
|
return liveKitVideoSourceScreenShare
|
|
}
|
|
return liveKitVideoSourceCamera
|
|
}
|
|
|
|
func (s liveKitVideoSource) String() string {
|
|
if s == liveKitVideoSourceScreenShare {
|
|
return "screen_share"
|
|
}
|
|
return "camera"
|
|
}
|
|
|
|
func (p *LiveKitParticipant) selectedParticipant() string {
|
|
p.mu.Lock()
|
|
defer p.mu.Unlock()
|
|
return p.selectedRemoteParticipant
|
|
}
|
|
|
|
func remoteParticipantSelected(selected, identity string) bool {
|
|
return selected == "" || identity == selected
|
|
}
|
|
|
|
func (p *LiveKitParticipant) closeRemoteTracks() {
|
|
p.mu.Lock()
|
|
tracks := p.remoteAudio
|
|
p.remoteAudio = nil
|
|
p.remoteVideoPLI = nil
|
|
p.remoteVideoSSRC = 0
|
|
p.remoteScreenPLI = nil
|
|
p.remoteScreenSSRC = 0
|
|
p.remoteScreenActive = false
|
|
p.remoteVideoKeyframePending = [2]bool{}
|
|
p.remoteVideoKeyframeAwaited = [2]bool{}
|
|
cancel := p.remoteMediaCancel
|
|
p.remoteMediaCancel = nil
|
|
p.mu.Unlock()
|
|
if cancel != nil {
|
|
cancel()
|
|
}
|
|
for _, track := range tracks {
|
|
track.Close()
|
|
}
|
|
}
|
|
|
|
const meowcallerSampleRate = 16000
|
|
const liveKitH264ClockRate = 90000
|
|
const liveKitH264MaxLatePackets = 1000
|
|
|
|
func videoFrameDuration(cfg VideoConfig) time.Duration {
|
|
if cfg.MaxFPS <= 0 {
|
|
return time.Second / 30
|
|
}
|
|
return time.Second / time.Duration(cfg.MaxFPS)
|
|
}
|
|
|
|
var _ lkpcm.PCM16Writer = (*MeowcallerAudioSource)(nil)
|