Merge 9d12dd1e54 into 4e90883bf8

audio.go

@@ -0,0 +1,81 @@
+package kvm
+
+import (
+	"fmt"
+	"net"
+	"os/exec"
+	"sync"
+	"syscall"
+	"time"
+)
+
+func startFFmpeg() (cmd *exec.Cmd, err error) {
+	binaryPath := "/userdata/jetkvm/bin/ffmpeg"
+	// Run the binary in the background
+	cmd = exec.Command(binaryPath,
+		"-f", "alsa",
+		"-channels", "2",
+		"-sample_rate", "48000",
+		"-i", "hw:1,0",
+		"-c:a", "libopus",
+		"-b:a", "64k", // ought to be enough for anybody
+		"-vbr", "off",
+		"-frame_duration", "20",
+		"-compression_level", "2",
+		"-f", "rtp",
+		"rtp://127.0.0.1:3333")
+
+	nativeOutputLock := sync.Mutex{}
+	nativeStdout := &nativeOutput{
+		mu:     &nativeOutputLock,
+		logger: nativeLogger.Info().Str("pipe", "stdout"),
+	}
+	nativeStderr := &nativeOutput{
+		mu:     &nativeOutputLock,
+		logger: nativeLogger.Info().Str("pipe", "stderr"),
+	}
+
+	// Redirect stdout and stderr to the current process
+	cmd.Stdout = nativeStdout
+	cmd.Stderr = nativeStderr
+
+	// Set the process group ID so we can kill the process and its children when this process exits
+	cmd.SysProcAttr = &syscall.SysProcAttr{
+		Setpgid:   true,
+		Pdeathsig: syscall.SIGKILL,
+	}
+
+	// Start the command
+	if err := cmd.Start(); err != nil {
+		return nil, fmt.Errorf("failed to start binary: %w", err)
+	}
+
+	return
+}
+
+func StartNtpAudioServer(handleClient func(net.Conn)) {
+	scopedLogger := nativeLogger.With().
+		Logger()
+
+	listener, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 3333})
+	if err != nil {
+		scopedLogger.Warn().Err(err).Msg("failed to start server")
+		return
+	}
+
+	scopedLogger.Info().Msg("server listening")
+
+	go func() {
+		for {
+			cmd, err := startFFmpeg()
+			if err != nil {
+				scopedLogger.Error().Err(err).Msg("failed to start ffmpeg")
+			}
+			err = cmd.Wait()
+			scopedLogger.Error().Err(err).Msg("ffmpeg exited, restarting")
+			time.Sleep(2 * time.Second)
+		}
+	}()
+
+	go handleClient(listener)
+}

config.go

@@ -125,6 +125,7 @@ var defaultConfig = &Config{
 		RelativeMouse: true,
 		Keyboard:      true,
 		MassStorage:   true,
+		Audio:         true,
 	},
 	NetworkConfig:   &network.NetworkConfig{},
 	DefaultLogLevel: "INFO",

internal/usbgadget/config.go

@@ -59,6 +59,23 @@ var defaultGadgetConfig = map[string]gadgetConfigItem{
 	// mass storage
 	"mass_storage_base": massStorageBaseConfig,
 	"mass_storage_lun0": massStorageLun0Config,
+	// audio
+	"audio": {
+		order:      4000,
+		device:     "uac1.usb0",
+		path:       []string{"functions", "uac1.usb0"},
+		configPath: []string{"uac1.usb0"},
+		attrs: gadgetAttributes{
+			"p_chmask":         "3",
+			"p_srate":          "48000",
+			"p_ssize":          "2",
+			"p_volume_present": "0",
+			"c_chmask":         "3",
+			"c_srate":          "48000",
+			"c_ssize":          "2",
+			"c_volume_present": "0",
+		},
+	},
 }
 
 func (u *UsbGadget) isGadgetConfigItemEnabled(itemKey string) bool {
@@ -73,6 +90,8 @@ func (u *UsbGadget) isGadgetConfigItemEnabled(itemKey string) bool {
 		return u.enabledDevices.MassStorage
 	case "mass_storage_lun0":
 		return u.enabledDevices.MassStorage
+	case "audio":
+		return u.enabledDevices.Audio
 	default:
 		return true
 	}

internal/usbgadget/usbgadget.go

@@ -18,6 +18,7 @@ type Devices struct {
 	RelativeMouse bool `json:"relative_mouse"`
 	Keyboard      bool `json:"keyboard"`
 	MassStorage   bool `json:"mass_storage"`
+	Audio         bool `json:"audio"`
 }
 
 // Config is a struct that represents the customizations for a USB gadget.

main.go

@@ -76,6 +76,7 @@ func Main() {
 	}()
 
 	initUsbGadget()
+	StartNtpAudioServer(handleAudioClient)
 
 	if err := setInitialVirtualMediaState(); err != nil {
 		logger.Warn().Err(err).Msg("failed to set initial virtual media state")

native.go

@@ -12,6 +12,7 @@ import (
 	"time"
 
 	"github.com/jetkvm/kvm/resource"
+	"github.com/pion/rtp"
 
 	"github.com/pion/webrtc/v4/pkg/media"
 )
@@ -215,7 +216,7 @@ func handleVideoClient(conn net.Conn) {
 
 	scopedLogger.Info().Msg("native video socket client connected")
 
-	inboundPacket := make([]byte, maxFrameSize)
+	inboundPacket := make([]byte, maxVideoFrameSize)
 	lastFrame := time.Now()
 	for {
 		n, err := conn.Read(inboundPacket)
@@ -235,6 +236,44 @@ func handleVideoClient(conn net.Conn) {
 	}
 }
 
+func handleAudioClient(conn net.Conn) {
+	defer conn.Close()
+	scopedLogger := nativeLogger.With().
+		Str("type", "audio").
+		Logger()
+
+	scopedLogger.Info().Msg("native audio socket client connected")
+	inboundPacket := make([]byte, maxAudioFrameSize)
+	var timestamp uint32
+	var packet rtp.Packet
+	for {
+		n, err := conn.Read(inboundPacket)
+		if err != nil {
+			scopedLogger.Warn().Err(err).Msg("error during read")
+			return
+		}
+
+		if currentSession != nil {
+			if err := packet.Unmarshal(inboundPacket[:n]); err != nil {
+				scopedLogger.Warn().Err(err).Msg("error unmarshalling audio socket packet")
+				continue
+			}
+
+			timestamp += 960
+			packet.Header.Timestamp = timestamp
+			buf, err := packet.Marshal()
+			if err != nil {
+				scopedLogger.Warn().Err(err).Msg("error marshalling packet")
+				continue
+			}
+
+			if _, err := currentSession.AudioTrack.Write(buf); err != nil {
+				scopedLogger.Warn().Err(err).Msg("error writing sample")
+			}
+		}
+	}
+}
+
 func ExtractAndRunNativeBin() error {
 	binaryPath := "/userdata/jetkvm/bin/jetkvm_native"
 	if err := ensureBinaryUpdated(binaryPath); err != nil {

ui/src/components/WebRTCVideo.tsx

@@ -683,7 +683,7 @@ export default function WebRTCVideo() {
                           controls={false}
                           onPlaying={onVideoPlaying}
                           onPlay={onVideoPlaying}
-                          muted={true}
+                          muted={false}
                           playsInline
                           disablePictureInPicture
                           controlsList="nofullscreen"

ui/src/routes/devices.$id.tsx

@@ -478,6 +478,8 @@ export default function KvmIdRoute() {
     };
 
     setTransceiver(pc.addTransceiver("video", { direction: "recvonly" }));
+    // Add audio transceiver to receive audio from the server
+    pc.addTransceiver("audio", { direction: "recvonly" });
 
     const rpcDataChannel = pc.createDataChannel("rpc");
     rpcDataChannel.onopen = () => {

video.go

@@ -5,7 +5,8 @@ import (
 )
 
 // max frame size for 1080p video, specified in mpp venc setting
-const maxFrameSize = 1920 * 1080 / 2
+const maxVideoFrameSize = 1920 * 1080 / 2
+const maxAudioFrameSize = 1500
 
 func writeCtrlAction(action string) error {
 	actionMessage := map[string]string{

webrtc.go

@@ -18,6 +18,7 @@ import (
 type Session struct {
 	peerConnection           *webrtc.PeerConnection
 	VideoTrack               *webrtc.TrackLocalStaticSample
+	AudioTrack               *webrtc.TrackLocalStaticRTP
 	ControlChannel           *webrtc.DataChannel
 	RPCChannel               *webrtc.DataChannel
 	HidChannel               *webrtc.DataChannel
@@ -136,7 +137,17 @@ func newSession(config SessionConfig) (*Session, error) {
 		return nil, err
 	}
 
-	rtpSender, err := peerConnection.AddTrack(session.VideoTrack)
+	session.AudioTrack, err = webrtc.NewTrackLocalStaticRTP(webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeOpus}, "audio", "kvm")
+	if err != nil {
+		return nil, err
+	}
+
+	videoRtpSender, err := peerConnection.AddTrack(session.VideoTrack)
+	if err != nil {
+		return nil, err
+	}
+
+	audioRtpSender, err := peerConnection.AddTrack(session.AudioTrack)
 	if err != nil {
 		return nil, err
 	}
@@ -144,14 +155,9 @@ func newSession(config SessionConfig) (*Session, error) {
 	// Read incoming RTCP packets
 	// Before these packets are returned they are processed by interceptors. For things
 	// like NACK this needs to be called.
-	go func() {
+	go drainRtpSender(videoRtpSender)
-		rtcpBuf := make([]byte, 1500)
+	go drainRtpSender(audioRtpSender)
-		for {
+
-			if _, _, rtcpErr := rtpSender.Read(rtcpBuf); rtcpErr != nil {
-				return
-			}
-		}
-	}()
 	var isConnected bool
 
 	peerConnection.OnICECandidate(func(candidate *webrtc.ICECandidate) {
@@ -203,6 +209,15 @@ func newSession(config SessionConfig) (*Session, error) {
 	return session, nil
 }
 
+func drainRtpSender(rtpSender *webrtc.RTPSender) {
+	rtcpBuf := make([]byte, 1500)
+	for {
+		if _, _, err := rtpSender.Read(rtcpBuf); err != nil {
+			return
+		}
+	}
+}
+
 var actionSessions = 0
 
 func onActiveSessionsChanged() {