Refactor: Rename NanoKVM to BatchuKVM and update server URL

server/service/stream/direct/h264.go

@@ -0,0 +1,44 @@
+package direct
+
+import (
+	"net/http"
+	"time"
+
+	"github.com/gin-gonic/gin"
+	"github.com/gorilla/websocket"
+	log "github.com/sirupsen/logrus"
+)
+
+var (
+	streamer = newStreamer()
+	upgrader = websocket.Upgrader{
+		CheckOrigin: func(r *http.Request) bool {
+			return true
+		},
+	}
+)
+
+func Connect(c *gin.Context) {
+	ws, err := upgrader.Upgrade(c.Writer, c.Request, nil)
+	if err != nil {
+		log.Errorf("failed to upgrade to websocket: %s", err)
+		return
+	}
+	defer func() {
+		_ = ws.Close()
+		log.Debugf("h264 websocket disconnected: %s", ws.RemoteAddr())
+	}()
+	log.Debugf("h264 websocket connected: %s", ws.RemoteAddr())
+
+	_ = ws.SetReadDeadline(time.Time{})
+
+	streamer.addClient(ws)
+	defer streamer.removeClient(ws)
+
+	for {
+		if _, _, err := ws.ReadMessage(); err != nil {
+			log.Debugf("failed to read message (client disconnected): %s", err)
+			return
+		}
+	}
+}

server/service/stream/direct/pool.go

@@ -0,0 +1,12 @@
+package direct
+
+import (
+	"bytes"
+	"sync"
+)
+
+var BufferPool = sync.Pool{
+	New: func() interface{} {
+		return new(bytes.Buffer)
+	},
+}

server/service/stream/direct/streamer.go

@@ -0,0 +1,123 @@
+package direct
+
+import (
+	"NanoKVM-Server/common"
+	"NanoKVM-Server/service/stream"
+	"bytes"
+	"encoding/binary"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/gorilla/websocket"
+	log "github.com/sirupsen/logrus"
+)
+
+type Streamer struct {
+	mutex   sync.RWMutex
+	clients map[*websocket.Conn]bool
+	running int32
+}
+
+func newStreamer() *Streamer {
+	return &Streamer{
+		clients: make(map[*websocket.Conn]bool),
+	}
+}
+
+func (s *Streamer) addClient(ws *websocket.Conn) {
+	s.mutex.Lock()
+	s.clients[ws] = true
+	s.mutex.Unlock()
+
+	if atomic.CompareAndSwapInt32(&s.running, 0, 1) {
+		go s.run()
+		log.Debug("h264 stream started")
+	}
+}
+
+func (s *Streamer) removeClient(ws *websocket.Conn) {
+	s.mutex.Lock()
+	delete(s.clients, ws)
+	s.mutex.Unlock()
+
+	log.Debugf("h264 websocket disconnected, remaining clients: %d", len(s.clients))
+}
+
+func (s *Streamer) getClientCount() int {
+	s.mutex.RLock()
+	defer s.mutex.RUnlock()
+
+	return len(s.clients)
+}
+
+func (s *Streamer) run() {
+	defer atomic.StoreInt32(&s.running, 0)
+
+	duration := time.Second / time.Duration(120)
+	ticker := time.NewTicker(duration)
+	defer ticker.Stop()
+
+	screen := common.GetScreen()
+	vision := common.GetKvmVision()
+	startTime := time.Now()
+
+	for range ticker.C {
+		if s.getClientCount() == 0 {
+			log.Debug("h264 stream stopped due to no clients")
+			return
+		}
+
+		data, result := vision.ReadH264(screen.Width, screen.Height, screen.BitRate)
+		if result < 0 || len(data) == 0 {
+			continue
+		}
+
+		isKeyFrame := byte(0)
+		if result == 3 {
+			isKeyFrame = byte(1)
+		}
+
+		timestamp := time.Since(startTime).Microseconds()
+
+		if err := s.send(isKeyFrame, timestamp, data); err != nil {
+			continue
+		}
+
+		stream.GetFrameRateCounter().Update()
+	}
+}
+
+func (s *Streamer) send(isKeyFrame byte, timestamp int64, data []byte) error {
+	buf := BufferPool.Get().(*bytes.Buffer)
+	defer BufferPool.Put(buf)
+
+	buf.Reset()
+
+	if err := buf.WriteByte(isKeyFrame); err != nil {
+		log.Errorf("failed to write keyframe flag: %s", err)
+		return err
+	}
+
+	tsBytes := make([]byte, 8)
+	binary.LittleEndian.PutUint64(tsBytes, uint64(timestamp))
+	if _, err := buf.Write(tsBytes); err != nil {
+		log.Errorf("failed to write timestamp: %s", err)
+		return err
+	}
+
+	if _, err := buf.Write(data); err != nil {
+		log.Errorf("failed to write h264 data: %s", err)
+		return err
+	}
+
+	for client := range s.clients {
+		if err := client.WriteMessage(websocket.BinaryMessage, buf.Bytes()); err != nil {
+			log.Errorf("failed to write message to client %s: %s.", client.RemoteAddr(), err)
+
+			s.removeClient(client)
+		}
+	}
+
+	return nil
+}

server/service/stream/frame_rate.go

@@ -0,0 +1,63 @@
+package stream
+
+import (
+	"fmt"
+	"os"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	log "github.com/sirupsen/logrus"
+)
+
+var (
+	counter     *FrameRateCounter
+	counterOnce sync.Once
+)
+
+type FrameRateCounter struct {
+	frameCount int32
+	fps        int32
+	mutex      sync.Mutex
+}
+
+func GetFrameRateCounter() *FrameRateCounter {
+	counterOnce.Do(func() {
+		counter = &FrameRateCounter{}
+
+		go func() {
+			ticker := time.NewTicker(3 * time.Second)
+			defer ticker.Stop()
+
+			for range ticker.C {
+				counter.mutex.Lock()
+
+				currentCount := atomic.LoadInt32(&counter.frameCount)
+
+				counter.fps = currentCount / 3
+				atomic.StoreInt32(&counter.frameCount, 0)
+
+				counter.mutex.Unlock()
+
+				data := fmt.Sprintf("%d", counter.fps)
+				err := os.WriteFile("/kvmapp/kvm/now_fps", []byte(data), 0o666)
+				if err != nil {
+					log.Errorf("failed to write fps: %s", err)
+				}
+			}
+		}()
+	})
+
+	return counter
+}
+
+func (f *FrameRateCounter) Update() {
+	atomic.AddInt32(&f.frameCount, 1)
+}
+
+func (f *FrameRateCounter) GetFPS() int32 {
+	f.mutex.Lock()
+	defer f.mutex.Unlock()
+
+	return f.fps
+}

server/service/stream/h264/client.go

@@ -0,0 +1,167 @@
+package h264
+
+import (
+	"encoding/json"
+	"sync"
+
+	"github.com/gorilla/websocket"
+	"github.com/pion/webrtc/v4"
+	log "github.com/sirupsen/logrus"
+)
+
+type Client struct {
+	ws    *websocket.Conn
+	pc    *webrtc.PeerConnection
+	mutex sync.Mutex
+}
+
+type Message struct {
+	Event string `json:"event"`
+	Data  string `json:"data"`
+}
+
+// add video track
+func (c *Client) addTrack() {
+	videoTrack, err := webrtc.NewTrackLocalStaticSample(
+		webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeH264},
+		"video",
+		"pion",
+	)
+	if err != nil {
+		log.Errorf("failed to create video track: %s", err)
+		return
+	}
+
+	_, err = c.pc.AddTrack(videoTrack)
+	if err != nil {
+		log.Errorf("failed to add video track: %s", err)
+		return
+	}
+
+	trackMap[c.ws] = videoTrack
+}
+
+// register callback events
+func (c *Client) register() {
+	// new ICE candidate found
+	c.pc.OnICECandidate(func(candidate *webrtc.ICECandidate) {
+		if candidate == nil {
+			return
+		}
+
+		candidateByte, err := json.Marshal(candidate.ToJSON())
+		if err != nil {
+			log.Errorf("failed to marshal candidate: %s", err)
+			return
+		}
+
+		_ = c.sendMessage("candidate", string(candidateByte))
+	})
+
+	// ICE connection state has changed
+	c.pc.OnICEConnectionStateChange(func(state webrtc.ICEConnectionState) {
+		if state == webrtc.ICEConnectionStateConnected && !isSending {
+			// start sending h264 data
+			go send()
+			isSending = true
+		}
+
+		log.Debugf("ice connection state has changed to %s", state.String())
+	})
+}
+
+// read websocket message
+func (c *Client) readMessage() {
+	message := &Message{}
+
+	for {
+		_, raw, err := c.ws.ReadMessage()
+		if err != nil {
+			delete(trackMap, c.ws)
+			if isSending && len(trackMap) == 0 {
+				// stop sending when all websocket connections are closed
+				isSending = false
+			}
+
+			log.Debugf("failed to read message: %s", err)
+			return
+		}
+
+		if err := json.Unmarshal(raw, &message); err != nil {
+			log.Errorf("failed to unmarshal message: %s", err)
+			continue
+		}
+
+		log.Debugf("receive message event: %s", message.Event)
+
+		switch message.Event {
+		case "offer":
+			offer := webrtc.SessionDescription{}
+			if err := json.Unmarshal([]byte(message.Data), &offer); err != nil {
+				log.Errorf("failed to unmarshal offer message: %s", err)
+				return
+			}
+
+			if err := c.pc.SetRemoteDescription(offer); err != nil {
+				log.Errorf("failed to set remote description: %s", err)
+				return
+			}
+
+			answer, answerErr := c.pc.CreateAnswer(nil)
+			if answerErr != nil {
+				log.Errorf("failed to create answer: %s", answerErr)
+				return
+			}
+
+			if err := c.pc.SetLocalDescription(answer); err != nil {
+				log.Errorf("failed to set local description: %s", err)
+				return
+			}
+
+			answerByte, answerByteErr := json.Marshal(answer)
+			if answerByteErr != nil {
+				log.Errorf("failed to marshal answer: %s", answerByteErr)
+				return
+			}
+
+			_ = c.sendMessage("answer", string(answerByte))
+
+		case "candidate":
+			candidate := webrtc.ICECandidateInit{}
+			if err := json.Unmarshal([]byte(message.Data), &candidate); err != nil {
+				log.Errorf("failed to unmarshal candidate message: %s", err)
+				return
+			}
+
+			if err := c.pc.AddICECandidate(candidate); err != nil {
+				log.Errorf("failed to add ICE candidate: %s", err)
+				return
+			}
+
+		case "heartbeat":
+			_ = c.sendMessage("heartbeat", "")
+
+		default:
+			log.Debugf("unhandled message event: %s", message.Event)
+		}
+	}
+}
+
+// send websocket message
+func (c *Client) sendMessage(event string, data string) error {
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
+
+	message := &Message{
+		Event: event,
+		Data:  data,
+	}
+
+	if err := c.ws.WriteJSON(message); err != nil {
+		log.Errorf("failed to send message %s: %s", event, err)
+		return err
+	}
+
+	log.Debugf("send message %s", message.Event)
+	return nil
+}

server/service/stream/h264/h264.go

@@ -0,0 +1,80 @@
+package h264
+
+import (
+	"NanoKVM-Server/config"
+	"net/http"
+	"sync"
+	"time"
+
+	"github.com/gin-gonic/gin"
+	"github.com/gorilla/websocket"
+	"github.com/pion/webrtc/v4"
+	log "github.com/sirupsen/logrus"
+)
+
+var (
+	upgrader = websocket.Upgrader{
+		CheckOrigin: func(r *http.Request) bool {
+			return true
+		},
+	}
+	trackMap  = make(map[*websocket.Conn]*webrtc.TrackLocalStaticSample)
+	isSending = false
+)
+
+func Connect(c *gin.Context) {
+	wsConn, err := upgrader.Upgrade(c.Writer, c.Request, nil)
+	if err != nil {
+		log.Errorf("failed to create websocket: %s", err)
+		return
+	}
+
+	defer func() {
+		_ = wsConn.Close()
+		log.Debugf("h264 websocket disconnected")
+	}()
+
+	var zeroTime time.Time
+	_ = wsConn.SetReadDeadline(zeroTime)
+
+	conf := config.GetInstance()
+
+	var iceServers []webrtc.ICEServer
+
+	if conf.Stun != "" && conf.Stun != "disable" {
+		iceServers = append(iceServers, webrtc.ICEServer{
+			URLs: []string{"stun:" + conf.Stun},
+		})
+	}
+
+	if conf.Turn.TurnAddr != "" && conf.Turn.TurnUser != "" && conf.Turn.TurnCred != "" {
+		iceServers = append(iceServers, webrtc.ICEServer{
+			URLs:       []string{"turn:" + conf.Turn.TurnAddr},
+			Username:   conf.Turn.TurnUser,
+			Credential: conf.Turn.TurnCred,
+		})
+	}
+
+	peerConn, err := webrtc.NewPeerConnection(webrtc.Configuration{
+		ICEServers: iceServers,
+	})
+	if err != nil {
+		log.Errorf("failed to create PeerConnection: %s", err)
+		return
+	}
+
+	defer func() {
+		_ = peerConn.Close()
+		log.Debugf("PeerConnection disconnected")
+	}()
+
+	client := &Client{
+		ws:    wsConn,
+		pc:    peerConn,
+		mutex: sync.Mutex{},
+	}
+
+	client.addTrack()
+	client.register()
+	client.readMessage()
+}

server/service/stream/h264/sender.go

@@ -0,0 +1,49 @@
+package h264
+
+import (
+	"NanoKVM-Server/common"
+	"time"
+
+	"github.com/pion/webrtc/v4/pkg/media"
+	log "github.com/sirupsen/logrus"
+)
+
+func send() {
+	screen := common.GetScreen()
+	common.CheckScreen()
+
+	fps := screen.FPS
+	duration := time.Second / time.Duration(fps)
+
+	ticker := time.NewTicker(duration)
+	defer ticker.Stop()
+
+	vision := common.GetKvmVision()
+	for range ticker.C {
+		if !isSending && len(trackMap) == 0 {
+			return
+		}
+
+		data, result := vision.ReadH264(screen.Width, screen.Height, screen.BitRate)
+		if result < 0 || len(data) == 0 {
+			continue
+		}
+
+		sample := media.Sample{
+			Data:     data,
+			Duration: duration,
+		}
+
+		for _, track := range trackMap {
+			if err := track.WriteSample(sample); err != nil {
+				log.Errorf("failed to send h264 data: %s", err)
+			}
+		}
+
+		if screen.FPS != fps {
+			fps = screen.FPS
+			duration = time.Second / time.Duration(fps)
+			ticker.Reset(duration)
+		}
+	}
+}

server/service/stream/mjpeg/frame-detect.go

@@ -0,0 +1,55 @@
+package mjpeg
+
+import (
+	"NanoKVM-Server/common"
+	"NanoKVM-Server/proto"
+	"time"
+
+	"github.com/gin-gonic/gin"
+	log "github.com/sirupsen/logrus"
+)
+
+const FrameDetectInterval uint8 = 60
+
+func UpdateFrameDetect(c *gin.Context) {
+	var req proto.UpdateFrameDetectReq
+	var rsp proto.Response
+
+	if err := proto.ParseFormRequest(c, &req); err != nil {
+		rsp.ErrRsp(c, -1, "invalid parameters")
+		return
+	}
+
+	var frame uint8 = 0
+	if req.Enabled {
+		frame = FrameDetectInterval
+	}
+
+	common.GetKvmVision().SetFrameDetect(frame)
+
+	rsp.OkRsp(c)
+	log.Debugf("update frame detect: %t", req.Enabled)
+}
+
+func StopFrameDetect(c *gin.Context) {
+	var req proto.StopFrameDetectReq
+	var rsp proto.Response
+
+	if err := proto.ParseFormRequest(c, &req); err != nil {
+		rsp.ErrRsp(c, -1, "invalid parameters")
+		return
+	}
+
+	duration := 10 * time.Second
+	if req.Duration > 0 {
+		duration = time.Duration(req.Duration) * time.Second
+	}
+
+	vision := common.GetKvmVision()
+
+	vision.SetFrameDetect(0)
+	time.Sleep(duration)
+	vision.SetFrameDetect(FrameDetectInterval)
+
+	rsp.OkRsp(c)
+}

server/service/stream/mjpeg/mjpeg.go

@@ -0,0 +1,22 @@
+package mjpeg
+
+import (
+	"time"
+
+	"github.com/gin-gonic/gin"
+)
+
+var streamer = NewStreamer()
+
+func Connect(c *gin.Context) {
+	c.Header("Content-Type", "multipart/x-mixed-replace; boundary=frame")
+	c.Header("Cache-Control", "no-cache")
+	c.Header("Connection", "keep-alive")
+	c.Header("Pragma", "no-cache")
+	c.Header("X-Server-Date", time.Now().Format(time.RFC1123))
+
+	streamer.AddClient(c)
+	defer streamer.RemoveClient(c)
+
+	<-c.Request.Context().Done()
+}

server/service/stream/mjpeg/streamer.go

@@ -0,0 +1,131 @@
+package mjpeg
+
+import (
+	"NanoKVM-Server/common"
+	"NanoKVM-Server/service/stream"
+	"fmt"
+	"strconv"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/gin-gonic/gin"
+	log "github.com/sirupsen/logrus"
+)
+
+type Streamer struct {
+	mutex   sync.RWMutex
+	clients map[*gin.Context]bool
+	running int32
+}
+
+func NewStreamer() *Streamer {
+	return &Streamer{
+		clients: make(map[*gin.Context]bool),
+	}
+}
+
+func (s *Streamer) AddClient(c *gin.Context) {
+	s.mutex.Lock()
+	s.clients[c] = true
+	s.mutex.Unlock()
+
+	if atomic.CompareAndSwapInt32(&s.running, 0, 1) {
+		go s.run()
+		log.Debug("mjpeg stream started")
+	}
+}
+
+func (s *Streamer) RemoveClient(c *gin.Context) {
+	s.mutex.Lock()
+	delete(s.clients, c)
+	s.mutex.Unlock()
+
+	log.Debugf("mjpeg connection removed, remaining clients: %d", len(s.clients))
+}
+
+func (s *Streamer) getClients() []*gin.Context {
+	s.mutex.RLock()
+	defer s.mutex.RUnlock()
+
+	clients := make([]*gin.Context, 0, len(s.clients))
+	for c := range s.clients {
+		clients = append(clients, c)
+	}
+
+	return clients
+}
+
+func (s *Streamer) getClientCount() int {
+	s.mutex.RLock()
+	defer s.mutex.RUnlock()
+
+	return len(s.clients)
+}
+
+func (s *Streamer) run() {
+	defer atomic.StoreInt32(&s.running, 0)
+
+	screen := common.GetScreen()
+	common.CheckScreen()
+	fps := screen.FPS
+
+	vision := common.GetKvmVision()
+
+	ticker := time.NewTicker(time.Second / time.Duration(fps))
+	defer ticker.Stop()
+
+	for range ticker.C {
+		if s.getClientCount() == 0 {
+			log.Debug("mjpeg stream stopped due to no clients")
+			return
+		}
+
+		data, result := vision.ReadMjpeg(screen.Width, screen.Height, screen.Quality)
+		if result < 0 || result == 5 || len(data) == 0 {
+			continue
+		}
+
+		clients := s.getClients()
+		for _, client := range clients {
+			if err := writeFrame(client, data); err != nil {
+				log.Errorf("failed to write mjpeg frame for client %s: %s", client.Request.RemoteAddr, err)
+				s.RemoveClient(client)
+			}
+		}
+
+		if screen.FPS != fps && screen.FPS != 0 {
+			fps = screen.FPS
+			ticker.Reset(time.Second / time.Duration(fps))
+		}
+
+		stream.GetFrameRateCounter().Update()
+	}
+}
+
+func writeFrame(c *gin.Context, data []byte) (err error) {
+	defer func() {
+		if r := recover(); r != nil {
+			err = c.Request.Context().Err()
+			if err == nil {
+				err = fmt.Errorf("panic recovered in writeFrame: %v", r)
+			}
+		}
+	}()
+
+	header := "--frame\r\nContent-Type: image/jpeg\r\nContent-Length: " + strconv.Itoa(len(data)) + "\r\n\r\n"
+	if _, err = c.Writer.WriteString(header); err != nil {
+		return err
+	}
+
+	if _, err = c.Writer.Write(data); err != nil {
+		return err
+	}
+
+	if _, err = c.Writer.Write([]byte("\r\n")); err != nil {
+		return err
+	}
+
+	c.Writer.Flush()
+	return nil
+}

server/service/stream/webrtc/client.go

@@ -0,0 +1,112 @@
+package webrtc
+
+import (
+	"encoding/json"
+	"errors"
+
+	"github.com/gorilla/websocket"
+	"github.com/pion/rtp"
+	"github.com/pion/rtp/codecs"
+	"github.com/pion/webrtc/v4"
+	log "github.com/sirupsen/logrus"
+
+	"sync"
+)
+
+func NewClient(ws *websocket.Conn, videoConn *webrtc.PeerConnection) *Client {
+	return &Client{
+		ws:    ws,
+		video: videoConn,
+		mutex: sync.Mutex{},
+	}
+}
+
+func (c *Client) WriteMessage(event string, data string) error {
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
+
+	message := &Message{
+		Event: event,
+		Data:  data,
+	}
+
+	if err := c.ws.WriteJSON(message); err != nil {
+		log.Errorf("failed to send message %s: %v", event, err)
+		return err
+	}
+
+	log.Debugf("sent message %s", event)
+	return nil
+}
+
+func (c *Client) ReadMessage() (*Message, error) {
+	_, raw, err := c.ws.ReadMessage()
+	if err != nil {
+		log.Errorf("failed to read message: %v", err)
+		return nil, err
+	}
+
+	var message Message
+	if err := json.Unmarshal(raw, &message); err != nil {
+		log.Errorf("failed to unmarshal message: %v", err)
+		return nil, nil
+	}
+
+	return &message, nil
+}
+
+func (c *Client) AddTrack() error {
+	// video track
+	videoTrack, err := webrtc.NewTrackLocalStaticRTP(
+		webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeH264},
+		"video",
+		"pion-video",
+	)
+	if err != nil {
+		log.Errorf("failed to create video track: %s", err)
+		return err
+	}
+
+	videoPacketizer := rtp.NewPacketizer(
+		1200,
+		100,
+		0x1234ABCD,
+		&codecs.H264Payloader{},
+		rtp.NewRandomSequencer(),
+		90000,
+	)
+	if videoPacketizer == nil {
+		err := errors.New("failed to create rtp packetizer")
+		log.Error(err)
+		return err
+	}
+
+	videoSender, err := c.video.AddTrack(videoTrack)
+	if err != nil {
+		log.Errorf("failed to add video track: %s", err)
+		return err
+	}
+	go startRTCPReader(videoSender)
+
+	track := &Track{
+		videoPacketizer: videoPacketizer,
+		video:           videoTrack,
+	}
+	track.updateExtension()
+
+	c.mutex.Lock()
+	c.track = track
+	c.mutex.Unlock()
+
+	return nil
+}
+
+func startRTCPReader(sender *webrtc.RTPSender) {
+	rtcpBuf := make([]byte, 1500)
+	for {
+		if _, _, err := sender.Read(rtcpBuf); err != nil {
+			log.Debugf("RTCP reader error: %v", err)
+			return
+		}
+	}
+}

server/service/stream/webrtc/h264.go

@@ -0,0 +1,158 @@
+package webrtc
+
+import (
+	"NanoKVM-Server/config"
+	"net/http"
+	"sync"
+	"time"
+
+	"github.com/gin-gonic/gin"
+	"github.com/gorilla/websocket"
+	"github.com/pion/dtls/v3"
+	"github.com/pion/webrtc/v4"
+	log "github.com/sirupsen/logrus"
+)
+
+var (
+	upgrader = websocket.Upgrader{
+		CheckOrigin: func(r *http.Request) bool {
+			return true
+		},
+	}
+	globalManager *WebRTCManager
+	managerOnce   sync.Once
+)
+
+func getManager() *WebRTCManager {
+	managerOnce.Do(func() {
+		globalManager = NewWebRTCManager()
+	})
+	return globalManager
+}
+
+func Connect(c *gin.Context) {
+	// create WebSocket connection
+	wsConn, err := upgrader.Upgrade(c.Writer, c.Request, nil)
+	if err != nil {
+		log.Errorf("failed to create h264 websocket: %s", err)
+		return
+	}
+	defer func() {
+		_ = wsConn.Close()
+		log.Debugf("h264 websocket disconnected: %s", c.ClientIP())
+	}()
+	log.Debugf("h264 websocket connected: %s", c.ClientIP())
+
+	var zeroTime time.Time
+	_ = wsConn.SetReadDeadline(zeroTime)
+
+	// create video connection
+	iceServers := createICEServers()
+
+	mediaEngine, err := createMediaEngine()
+	if err != nil {
+		log.Errorf("failed to create h264 media engine: %s", err)
+		return
+	}
+
+	videoConn, err := createPeerConnection(iceServers, mediaEngine)
+	if err != nil {
+		log.Errorf("failed to create h264 video peer connection: %s", err)
+		return
+	}
+	defer func() {
+		_ = videoConn.Close()
+		log.Debugf("h264 video peer disconnected: %s", c.ClientIP())
+	}()
+
+	// create client
+	client := NewClient(wsConn, videoConn)
+	if err := client.AddTrack(); err != nil {
+		log.Errorf("failed to add track: %s", err)
+		return
+	}
+
+	manager := getManager()
+	manager.AddClient(wsConn, client)
+	defer manager.RemoveClient(wsConn)
+
+	// handle signaling
+	signalingHandler := NewSignalingHandler(client)
+	signalingHandler.RegisterCallbacks()
+
+	// read and wait
+	for {
+		message, err := client.ReadMessage()
+		if err != nil {
+			return
+		}
+		if message != nil {
+			if err := signalingHandler.HandleMessage(message); err != nil {
+				log.Errorf("failed to handle signaling message: %s", err)
+			}
+		}
+	}
+}
+
+func createICEServers() []webrtc.ICEServer {
+	var iceServers []webrtc.ICEServer
+
+	conf := config.GetInstance()
+
+	if conf.Stun != "" && conf.Stun != "disable" {
+		iceServers = append(iceServers, webrtc.ICEServer{
+			URLs: []string{"stun:" + conf.Stun},
+		})
+	}
+
+	if conf.Turn.TurnAddr != "" && conf.Turn.TurnUser != "" && conf.Turn.TurnCred != "" {
+		iceServers = append(iceServers, webrtc.ICEServer{
+			URLs:       []string{"turn:" + conf.Turn.TurnAddr},
+			Username:   conf.Turn.TurnUser,
+			Credential: conf.Turn.TurnCred,
+		})
+	}
+
+	return iceServers
+}
+
+func createMediaEngine() (*webrtc.MediaEngine, error) {
+	mediaEngine := &webrtc.MediaEngine{}
+
+	if err := mediaEngine.RegisterDefaultCodecs(); err != nil {
+		log.Errorf("failed to register default codecs: %s", err)
+		return nil, err
+	}
+
+	if err := mediaEngine.RegisterHeaderExtension(
+		webrtc.RTPHeaderExtensionCapability{URI: "http://www.webrtc.org/experiments/rtp-hdrext/playout-delay"},
+		webrtc.RTPCodecTypeVideo,
+	); err != nil {
+		log.Errorf("failed to register header extension: %s", err)
+		return nil, err
+	}
+
+	return mediaEngine, nil
+}
+
+func createPeerConnection(iceServers []webrtc.ICEServer, mediaEngine *webrtc.MediaEngine) (*webrtc.PeerConnection, error) {
+	settingEngine := webrtc.SettingEngine{}
+	settingEngine.SetSRTPProtectionProfiles(
+		dtls.SRTP_AEAD_AES_128_GCM,
+		dtls.SRTP_AES128_CM_HMAC_SHA1_80,
+	)
+
+	apiOptions := []func(api *webrtc.API){
+		webrtc.WithSettingEngine(settingEngine),
+	}
+	if mediaEngine != nil {
+		apiOptions = append(apiOptions, webrtc.WithMediaEngine(mediaEngine))
+	}
+
+	api := webrtc.NewAPI(apiOptions...)
+
+	return api.NewPeerConnection(webrtc.Configuration{
+		ICEServers:   iceServers,
+		SDPSemantics: webrtc.SDPSemanticsUnifiedPlan,
+	})
+}

server/service/stream/webrtc/manager.go

@@ -0,0 +1,96 @@
+package webrtc
+
+import (
+	"NanoKVM-Server/common"
+	"NanoKVM-Server/service/stream"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/gorilla/websocket"
+	"github.com/pion/webrtc/v4/pkg/media"
+	log "github.com/sirupsen/logrus"
+)
+
+func NewWebRTCManager() *WebRTCManager {
+	return &WebRTCManager{
+		clients:      make(map[*websocket.Conn]*Client),
+		videoSending: 0,
+		mutex:        sync.RWMutex{},
+	}
+}
+
+func (m *WebRTCManager) AddClient(ws *websocket.Conn, client *Client) {
+	client.track.updateExtension()
+
+	m.mutex.Lock()
+	m.clients[ws] = client
+	m.mutex.Unlock()
+
+	log.Debugf("added client %s, total clients: %d", ws.RemoteAddr(), len(m.clients))
+}
+
+func (m *WebRTCManager) RemoveClient(ws *websocket.Conn) {
+	m.mutex.Lock()
+	delete(m.clients, ws)
+	m.mutex.Unlock()
+
+	log.Debugf("removed client %s, total clients: %d", ws.RemoteAddr(), len(m.clients))
+}
+
+func (m *WebRTCManager) GetClientCount() int {
+	m.mutex.RLock()
+	defer m.mutex.RUnlock()
+
+	return len(m.clients)
+}
+
+func (m *WebRTCManager) StartVideoStream() {
+	if atomic.CompareAndSwapInt32(&m.videoSending, 0, 1) {
+		go m.sendVideoStream()
+		log.Debugf("start sending h264 stream")
+	}
+}
+
+func (m *WebRTCManager) sendVideoStream() {
+	defer atomic.StoreInt32(&m.videoSending, 0)
+
+	screen := common.GetScreen()
+	common.CheckScreen()
+	fps := screen.FPS
+	duration := time.Second / time.Duration(fps)
+
+	vision := common.GetKvmVision()
+
+	ticker := time.NewTicker(duration)
+	defer ticker.Stop()
+
+	for range ticker.C {
+		if m.GetClientCount() == 0 {
+			log.Debugf("stop sending h264 stream")
+			return
+		}
+
+		data, result := vision.ReadH264(screen.Width, screen.Height, screen.BitRate)
+		if result < 0 || len(data) == 0 {
+			continue
+		}
+
+		sample := media.Sample{
+			Data:     data,
+			Duration: duration,
+		}
+
+		for _, client := range m.clients {
+			client.track.writeVideo(sample)
+		}
+
+		if screen.FPS != fps && screen.FPS != 0 {
+			fps = screen.FPS
+			duration = time.Second / time.Duration(fps)
+			ticker.Reset(duration)
+		}
+
+		stream.GetFrameRateCounter().Update()
+	}
+}

server/service/stream/webrtc/signaling.go

@@ -0,0 +1,149 @@
+package webrtc
+
+import (
+	"encoding/json"
+	"errors"
+
+	"github.com/pion/webrtc/v4"
+	log "github.com/sirupsen/logrus"
+)
+
+func NewSignalingHandler(client *Client) *SignalingHandler {
+	return &SignalingHandler{
+		client: client,
+	}
+}
+
+// RegisterCallbacks Register callback functions
+func (s *SignalingHandler) RegisterCallbacks() {
+	// video ICE candidate
+	s.client.video.OnICECandidate(func(candidate *webrtc.ICECandidate) {
+		if candidate == nil {
+			return
+		}
+
+		candidateByte, err := json.Marshal(candidate.ToJSON())
+		if err != nil {
+			log.Errorf("failed to marshal video candidate: %s", err)
+			return
+		}
+
+		if err := s.client.WriteMessage("video-candidate", string(candidateByte)); err != nil {
+			log.Errorf("failed to send video candidate: %s", err)
+		}
+	})
+
+	manager := getManager()
+
+	// video connection state change
+	s.client.video.OnICEConnectionStateChange(func(state webrtc.ICEConnectionState) {
+		if state == webrtc.ICEConnectionStateConnected {
+			manager.StartVideoStream()
+		}
+
+		log.Debugf("video connection state changed to %s", state.String())
+	})
+}
+
+// HandleMessage handle the received message
+func (s *SignalingHandler) HandleMessage(message *Message) error {
+	switch message.Event {
+	case "video-offer":
+		return s.handleVideoOffer(message.Data)
+	case "video-candidate":
+		return s.handleVideoCandidate(message.Data)
+	case "heartbeat":
+		return s.handleHeartbeat()
+	default:
+		log.Debugf("Unhandled message event: %s", message.Event)
+		return nil
+	}
+}
+
+func (s *SignalingHandler) handleVideoOffer(data string) error {
+	if s.client.video.SignalingState() != webrtc.SignalingStateStable {
+		err := errors.New("video signaling is not stable")
+		log.Error(err)
+		return err
+	}
+
+	offer := webrtc.SessionDescription{}
+	if err := json.Unmarshal([]byte(data), &offer); err != nil {
+		log.Errorf("failed to unmarshal video offer: %s", err)
+		return err
+	}
+
+	if err := s.client.video.SetRemoteDescription(offer); err != nil {
+		log.Errorf("failed to set remote description: %s", err)
+		return err
+	}
+
+	answer, err := s.client.video.CreateAnswer(nil)
+	if err != nil {
+		log.Errorf("failed to create answer: %s", err)
+		return err
+	}
+
+	if err := s.client.video.SetLocalDescription(answer); err != nil {
+		log.Errorf("failed to set local description: %s", err)
+		return err
+	}
+
+	if err := s.updateHeaderExtensionID(); err != nil {
+		log.Errorf("could not update header extension ID: %v", err)
+		return err
+	}
+
+	answerByte, err := json.Marshal(answer)
+	if err != nil {
+		log.Errorf("failed to marshal answer: %s", err)
+		return err
+	}
+
+	return s.client.WriteMessage("video-answer", string(answerByte))
+}
+
+// set extension ID
+func (s *SignalingHandler) updateHeaderExtensionID() error {
+	receivers := s.client.video.GetReceivers()
+	if len(receivers) == 0 {
+		return errors.New("no RTP receiver found for video")
+	}
+
+	params := receivers[0].GetParameters()
+	if len(params.HeaderExtensions) == 0 {
+		return errors.New("no header extensions found in negotiated parameters")
+	}
+
+	for _, ext := range params.HeaderExtensions {
+		if ext.URI == "http://www.webrtc.org/experiments/rtp-hdrext/playout-delay" {
+			s.client.track.playoutDelayExtensionID = uint8(ext.ID)
+			log.Debugf("found and set playout delay extension ID to: %d", ext.ID)
+			return nil
+		}
+	}
+
+	log.Warnf("no track extension found in negotiated parameters, use default value 5")
+	return nil
+}
+
+// handle video candidate
+func (s *SignalingHandler) handleVideoCandidate(data string) error {
+	candidate := webrtc.ICECandidateInit{}
+	if err := json.Unmarshal([]byte(data), &candidate); err != nil {
+		log.Errorf("failed to unmarshal candidate: %s", err)
+		return err
+	}
+
+	if err := s.client.video.AddICECandidate(candidate); err != nil {
+		log.Errorf("failed to add ICECandidate: %s", err)
+		return err
+	}
+
+	return nil
+}
+
+// handle heartbeat
+func (s *SignalingHandler) handleHeartbeat() error {
+	return s.client.WriteMessage("heartbeat", "")
+}

server/service/stream/webrtc/track.go

@@ -0,0 +1,53 @@
+package webrtc
+
+import (
+	"github.com/pion/rtp"
+	"github.com/pion/webrtc/v4/pkg/media"
+	log "github.com/sirupsen/logrus"
+)
+
+func (t *Track) updateExtension() {
+	if t.playoutDelayExtensionID == 0 {
+		t.playoutDelayExtensionID = 5
+	}
+
+	if t.playoutDelayExtensionData == nil || len(t.playoutDelayExtensionData) == 0 {
+		playoutDelay := &rtp.PlayoutDelayExtension{
+			MinDelay: 0,
+			MaxDelay: 0,
+		}
+		playoutDelayExtensionData, err := playoutDelay.Marshal()
+		if err == nil {
+			t.playoutDelayExtensionData = playoutDelayExtensionData
+		}
+	}
+}
+
+func (t *Track) writeVideoSample(sample media.Sample) error {
+	samples := uint32(sample.Duration.Seconds() * 90000)
+	packets := t.videoPacketizer.Packetize(sample.Data, samples)
+
+	for _, p := range packets {
+		p.Header.Extension = true
+		p.Header.ExtensionProfile = 0xBEDE
+
+		if err := p.Header.SetExtension(t.playoutDelayExtensionID, t.playoutDelayExtensionData); err != nil {
+			log.Errorf("Failed to set extension: %v", err)
+			return err
+		}
+
+		if err := t.video.WriteRTP(p); err != nil {
+			log.Errorf("failed to write RTP: %v", err)
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (t *Track) writeVideo(sample media.Sample) {
+	err := t.writeVideoSample(sample)
+	if err != nil {
+		log.Errorf("failed to write h264 video: %s", err)
+	}
+}

server/service/stream/webrtc/types.go

@@ -0,0 +1,38 @@
+package webrtc
+
+import (
+	"sync"
+
+	"github.com/gorilla/websocket"
+	"github.com/pion/rtp"
+	"github.com/pion/webrtc/v4"
+)
+
+type WebRTCManager struct {
+	clients      map[*websocket.Conn]*Client
+	videoSending int32
+	mutex        sync.RWMutex
+}
+
+type Client struct {
+	ws    *websocket.Conn
+	video *webrtc.PeerConnection
+	track *Track
+	mutex sync.Mutex
+}
+
+type SignalingHandler struct {
+	client *Client
+}
+
+type Track struct {
+	playoutDelayExtensionID   uint8
+	playoutDelayExtensionData []byte
+	videoPacketizer           rtp.Packetizer
+	video                     *webrtc.TrackLocalStaticRTP
+}
+
+type Message struct {
+	Event string `json:"event"`
+	Data  string `json:"data"`
+}