mirror of
https://github.com/m1k1o/neko.git
synced 2024-07-24 14:40:50 +12:00
400 lines
10 KiB
Go
400 lines
10 KiB
Go
package webrtc
|
|
|
|
import (
|
|
"fmt"
|
|
"io"
|
|
"reflect"
|
|
"strings"
|
|
"sync"
|
|
|
|
"github.com/pion/webrtc/v3"
|
|
"github.com/pion/webrtc/v3/pkg/media"
|
|
"github.com/rs/zerolog"
|
|
"github.com/rs/zerolog/log"
|
|
|
|
"demodesk/neko/internal/config"
|
|
"demodesk/neko/internal/types"
|
|
"demodesk/neko/internal/types/event"
|
|
"demodesk/neko/internal/types/message"
|
|
)
|
|
|
|
func New(desktop types.DesktopManager, capture types.CaptureManager, config *config.WebRTC) *WebRTCManagerCtx {
|
|
return &WebRTCManagerCtx{
|
|
logger: log.With().Str("module", "webrtc").Logger(),
|
|
desktop: desktop,
|
|
capture: capture,
|
|
config: config,
|
|
participants: 0,
|
|
// TODO: Refactor.
|
|
curImgListeners: map[uintptr]*func(cur *types.CursorImage){},
|
|
curPosListeners: map[uintptr]*func(x, y int){},
|
|
}
|
|
}
|
|
|
|
type WebRTCManagerCtx struct {
|
|
mu sync.Mutex
|
|
logger zerolog.Logger
|
|
audioTrack *webrtc.TrackLocalStaticSample
|
|
audioStop func()
|
|
desktop types.DesktopManager
|
|
capture types.CaptureManager
|
|
config *config.WebRTC
|
|
participants uint32
|
|
// TODO: Refactor.
|
|
curImgListeners map[uintptr]*func(cur *types.CursorImage)
|
|
curPosListeners map[uintptr]*func(x, y int)
|
|
}
|
|
|
|
func (manager *WebRTCManagerCtx) Start() {
|
|
var err error
|
|
|
|
// create audio track
|
|
audio := manager.capture.Audio()
|
|
manager.audioTrack, err = webrtc.NewTrackLocalStaticSample(audio.Codec().Capability, "audio", "stream")
|
|
if err != nil {
|
|
manager.logger.Panic().Err(err).Msg("unable to create audio track")
|
|
}
|
|
|
|
audioListener := func(sample types.Sample) {
|
|
if err := manager.audioTrack.WriteSample(media.Sample(sample)); err != nil && err != io.ErrClosedPipe {
|
|
manager.logger.Warn().Err(err).Msg("audio pipeline failed to write")
|
|
}
|
|
}
|
|
|
|
audio.AddListener(&audioListener)
|
|
manager.audioStop = func() {
|
|
audio.RemoveListener(&audioListener)
|
|
}
|
|
|
|
manager.logger.Info().
|
|
Str("ice_lite", fmt.Sprintf("%t", manager.config.ICELite)).
|
|
Str("ice_trickle", fmt.Sprintf("%t", manager.config.ICETrickle)).
|
|
Str("ice_servers", strings.Join(manager.config.ICEServers, ",")).
|
|
Str("ephemeral_port_range", fmt.Sprintf("%d-%d", manager.config.EphemeralMin, manager.config.EphemeralMax)).
|
|
Str("nat_ips", strings.Join(manager.config.NAT1To1IPs, ",")).
|
|
Msgf("webrtc starting")
|
|
|
|
// TODO: Refactor.
|
|
manager.desktop.OnCursorChanged(func(serial uint64) {
|
|
cur := manager.desktop.GetCursorImage()
|
|
for _, emit := range manager.curImgListeners {
|
|
(*emit)(cur)
|
|
}
|
|
})
|
|
|
|
// TODO: Refactor.
|
|
manager.desktop.OnCursorPosition(func(x, y int) {
|
|
for _, emit := range manager.curPosListeners {
|
|
(*emit)(x, y)
|
|
}
|
|
})
|
|
}
|
|
|
|
func (manager *WebRTCManagerCtx) Shutdown() error {
|
|
manager.logger.Info().Msgf("webrtc shutting down")
|
|
|
|
manager.audioStop()
|
|
return nil
|
|
}
|
|
|
|
func (manager *WebRTCManagerCtx) ICELite() bool {
|
|
return manager.config.ICELite
|
|
}
|
|
|
|
func (manager *WebRTCManagerCtx) ICEServers() []string {
|
|
return manager.config.ICEServers
|
|
}
|
|
|
|
func (manager *WebRTCManagerCtx) CreatePeer(session types.Session, videoID string) (*webrtc.SessionDescription, error) {
|
|
logger := manager.logger.With().Str("id", session.ID()).Logger()
|
|
|
|
// Create MediaEngine
|
|
engine, err := manager.mediaEngine(videoID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
// Custom settings & configuration
|
|
settings := manager.apiSettings(logger)
|
|
configuration := manager.apiConfiguration()
|
|
|
|
// Create NewAPI with MediaEngine and SettingEngine
|
|
api := webrtc.NewAPI(webrtc.WithMediaEngine(engine), webrtc.WithSettingEngine(*settings))
|
|
|
|
connection, err := api.NewPeerConnection(*configuration)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
// Asynchronously send local ICE Candidates
|
|
if manager.config.ICETrickle {
|
|
connection.OnICECandidate(func(candidate *webrtc.ICECandidate) {
|
|
if candidate == nil {
|
|
logger.Debug().Msg("all local ice candidates sent")
|
|
return
|
|
}
|
|
|
|
ICECandidateInit := candidate.ToJSON()
|
|
err := session.Send(
|
|
message.SignalCandidate{
|
|
Event: event.SIGNAL_CANDIDATE,
|
|
ICECandidateInit: &ICECandidateInit,
|
|
})
|
|
|
|
if err != nil {
|
|
logger.Warn().Err(err).Msg("sending ice candidate failed")
|
|
}
|
|
})
|
|
}
|
|
|
|
// create video track
|
|
videoStream, ok := manager.capture.Video(videoID)
|
|
if !ok {
|
|
manager.logger.Warn().Str("videoID", videoID).Msg("video stream not found")
|
|
return nil, err
|
|
}
|
|
|
|
videoTrack, err := webrtc.NewTrackLocalStaticSample(videoStream.Codec().Capability, "video", "stream")
|
|
if err != nil {
|
|
manager.logger.Warn().Err(err).Msg("unable to create video track")
|
|
return nil, err
|
|
}
|
|
|
|
videoListener := func(sample types.Sample) {
|
|
if err := videoTrack.WriteSample(media.Sample(sample)); err != nil && err != io.ErrClosedPipe {
|
|
manager.logger.Warn().Err(err).Msg("video pipeline failed to write")
|
|
}
|
|
}
|
|
|
|
manager.mu.Lock()
|
|
|
|
// should be stream started
|
|
if videoStream.ListenersCount() == 0 {
|
|
if err := videoStream.Start(); err != nil {
|
|
manager.logger.Warn().Err(err).Msg("unable to start video pipeline")
|
|
return nil, err
|
|
}
|
|
}
|
|
|
|
videoStream.AddListener(&videoListener)
|
|
|
|
// start audio, when first participant connects
|
|
if !manager.capture.Audio().Started() {
|
|
if err := manager.capture.Audio().Start(); err != nil {
|
|
manager.logger.Panic().Err(err).Msg("unable to start audio stream")
|
|
}
|
|
}
|
|
|
|
manager.participants = manager.participants + 1
|
|
manager.mu.Unlock()
|
|
|
|
changeVideo := func(videoID string) error {
|
|
newVideoStream, ok := manager.capture.Video(videoID)
|
|
if !ok {
|
|
return fmt.Errorf("video stream not found")
|
|
}
|
|
|
|
// should be new stream started
|
|
if newVideoStream.ListenersCount() == 0 {
|
|
if err := newVideoStream.Start(); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
// switch videoListeners
|
|
videoStream.RemoveListener(&videoListener)
|
|
newVideoStream.AddListener(&videoListener)
|
|
|
|
// should be old stream stopped
|
|
if videoStream.ListenersCount() == 0 {
|
|
videoStream.Stop()
|
|
}
|
|
|
|
videoStream = newVideoStream
|
|
return nil
|
|
}
|
|
|
|
_, err = connection.AddTrack(manager.audioTrack)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
_, err = connection.AddTrack(videoTrack)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
dataChannel, err := connection.CreateDataChannel("data", nil)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
offer, err := connection.CreateOffer(nil)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if !manager.config.ICETrickle {
|
|
// Create channel that is blocked until ICE Gathering is complete
|
|
gatherComplete := webrtc.GatheringCompletePromise(connection)
|
|
|
|
if err := connection.SetLocalDescription(offer); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
<-gatherComplete
|
|
} else {
|
|
if err := connection.SetLocalDescription(offer); err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
|
|
peer := &WebRTCPeerCtx{
|
|
api: api,
|
|
connection: connection,
|
|
changeVideo: changeVideo,
|
|
dataChannel: dataChannel,
|
|
}
|
|
|
|
cursorChange := func(cur *types.CursorImage) {
|
|
if err := peer.SendCursorImage(cur); err != nil {
|
|
manager.logger.Warn().Err(err).Msg("could not send cursor image")
|
|
}
|
|
}
|
|
|
|
cursorPosition := func(x, y int) {
|
|
if session.IsHost() {
|
|
return
|
|
}
|
|
|
|
if err := peer.SendCursorPosition(x, y); err != nil {
|
|
manager.logger.Warn().Err(err).Msg("could not send cursor position")
|
|
}
|
|
}
|
|
|
|
// TODO: Refactor.
|
|
cursorChangePtr := reflect.ValueOf(&cursorChange).Pointer()
|
|
cursorPositionPtr := reflect.ValueOf(&cursorPosition).Pointer()
|
|
|
|
connection.OnConnectionStateChange(func(state webrtc.PeerConnectionState) {
|
|
switch state {
|
|
case webrtc.PeerConnectionStateConnected:
|
|
session.SetWebRTCConnected(peer, true)
|
|
case webrtc.PeerConnectionStateDisconnected:
|
|
fallthrough
|
|
case webrtc.PeerConnectionStateFailed:
|
|
connection.Close()
|
|
case webrtc.PeerConnectionStateClosed:
|
|
manager.mu.Lock()
|
|
|
|
session.SetWebRTCConnected(peer, false)
|
|
videoStream.RemoveListener(&videoListener)
|
|
|
|
// should be stream stopped
|
|
if videoStream.ListenersCount() == 0 {
|
|
videoStream.Stop()
|
|
}
|
|
|
|
// decrease participants
|
|
manager.participants = manager.participants - 1
|
|
|
|
// stop audio, if last participant disonnects
|
|
if manager.participants <= 0 {
|
|
manager.participants = 0
|
|
|
|
if manager.capture.Audio().Started() {
|
|
manager.capture.Audio().Stop()
|
|
}
|
|
}
|
|
|
|
// TODO: Refactor.
|
|
delete(manager.curImgListeners, cursorChangePtr)
|
|
delete(manager.curPosListeners, cursorPositionPtr)
|
|
|
|
manager.mu.Unlock()
|
|
}
|
|
})
|
|
|
|
dataChannel.OnOpen(func() {
|
|
// TODO: Refactor.
|
|
manager.curImgListeners[cursorChangePtr] = &cursorChange
|
|
manager.curPosListeners[cursorPositionPtr] = &cursorPosition
|
|
|
|
// send initial cursor image
|
|
cur := manager.desktop.GetCursorImage()
|
|
cursorChange(cur)
|
|
|
|
// send initial cursor position
|
|
x, y := manager.desktop.GetCursorPosition()
|
|
manager.logger.Warn().Int("x", x).Int("y", y).Msg("got cursor position")
|
|
cursorPosition(x, y)
|
|
})
|
|
|
|
dataChannel.OnMessage(func(message webrtc.DataChannelMessage) {
|
|
if !session.IsHost() {
|
|
return
|
|
}
|
|
|
|
if err = manager.handle(message); err != nil {
|
|
logger.Warn().Err(err).Msg("data handle failed")
|
|
}
|
|
})
|
|
|
|
session.SetWebRTCPeer(peer)
|
|
return connection.LocalDescription(), nil
|
|
}
|
|
|
|
func (manager *WebRTCManagerCtx) mediaEngine(videoID string) (*webrtc.MediaEngine, error) {
|
|
engine := &webrtc.MediaEngine{}
|
|
|
|
// all videos must have the same codec
|
|
video, ok := manager.capture.Video(videoID)
|
|
if !ok {
|
|
return nil, fmt.Errorf("default video track not found")
|
|
}
|
|
|
|
videoCodec := video.Codec()
|
|
if err := videoCodec.Register(engine); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
audioCodec := manager.capture.Audio().Codec()
|
|
if err := audioCodec.Register(engine); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return engine, nil
|
|
}
|
|
|
|
func (manager *WebRTCManagerCtx) apiSettings(logger zerolog.Logger) *webrtc.SettingEngine {
|
|
settings := &webrtc.SettingEngine{
|
|
LoggerFactory: loggerFactory{
|
|
logger: logger,
|
|
},
|
|
}
|
|
|
|
//nolint
|
|
settings.SetEphemeralUDPPortRange(manager.config.EphemeralMin, manager.config.EphemeralMax)
|
|
settings.SetNAT1To1IPs(manager.config.NAT1To1IPs, webrtc.ICECandidateTypeHost)
|
|
settings.SetLite(manager.config.ICELite)
|
|
|
|
return settings
|
|
}
|
|
|
|
func (manager *WebRTCManagerCtx) apiConfiguration() *webrtc.Configuration {
|
|
if manager.config.ICELite {
|
|
return &webrtc.Configuration{
|
|
SDPSemantics: webrtc.SDPSemanticsUnifiedPlanWithFallback,
|
|
}
|
|
}
|
|
|
|
return &webrtc.Configuration{
|
|
ICEServers: []webrtc.ICEServer{
|
|
{
|
|
URLs: manager.config.ICEServers,
|
|
},
|
|
},
|
|
SDPSemantics: webrtc.SDPSemanticsUnifiedPlanWithFallback,
|
|
}
|
|
}
|