helper.go

package main

import (
	"encoding/json"
	"fmt"
	"io"
	"log"
	"net"
	"net/http"

	"github.com/google/uuid"
	"github.com/pion/interceptor"
	"github.com/pion/webrtc/v4"
)

type Offer struct {
	SDP string `json:"sdp"`
}

type Answer struct {
	SDP string `json:"sdp"`
}

func doProcessOffer(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
		return
	}

	body, err := io.ReadAll(r.Body)
	if err != nil {
		http.Error(w, "Failed to read request body", http.StatusBadRequest)
		return
	}

	var offer Offer
	if err := json.Unmarshal(body, &offer); err != nil {
		http.Error(w, "Failed to parse JSON", http.StatusBadRequest)
		return
	}

	config := webrtc.Configuration{
		ICEServers: []webrtc.ICEServer{
			{
				URLs: []string{"stun:stun.l.google.com:19302"},
			},
		},
	}

	settingEngine := webrtc.SettingEngine{}
	mediaEngine := webrtc.MediaEngine{}
	interceptorRegistry := interceptor.Registry{}

	videoRTCPFeedback := []webrtc.RTCPFeedback{{"goog-remb", ""}, {"ccm", "fir"}, {"nack", ""}, {"nack", "pli"}}
	for _, codec := range []webrtc.RTPCodecParameters{
		{
			RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeAV1, ClockRate: 90000, SDPFmtpLine: "level-idx=5;profile=0;tier=0", RTCPFeedback: videoRTCPFeedback},
			PayloadType:        45,
		},
		{
			RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeH264, ClockRate: 90000, SDPFmtpLine: "level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=42e01f", RTCPFeedback: videoRTCPFeedback},
			PayloadType:        96,
		},
	} {
		if err := mediaEngine.RegisterCodec(codec, webrtc.RTPCodecTypeVideo); err != nil {
			http.Error(w, "Failed to register video: "+err.Error(), http.StatusBadRequest)
			return
		}
	}

	_ = webrtc.RegisterDefaultInterceptors(&mediaEngine, &interceptorRegistry)

	api := webrtc.NewAPI(
		webrtc.WithInterceptorRegistry(&interceptorRegistry),
		webrtc.WithMediaEngine(&mediaEngine),
		webrtc.WithSettingEngine(settingEngine),
	)

	peerConnection, err := api.NewPeerConnection(config)
	if err != nil {
		http.Error(w, "Failed to create peer connection", http.StatusInternalServerError)
		return
	}

	outBountTrack, err := webrtc.NewTrackLocalStaticRTP(webrtc.RTPCodecCapability{
		MimeType:  webrtc.MimeTypeH264,
		ClockRate: 90000,
	}, uuid.NewString(), uuid.NewString())
	if err != nil {
		http.Error(w, "Failed to create track", http.StatusInternalServerError)
		return
	}

	_, err = peerConnection.AddTransceiverFromTrack(outBountTrack, webrtc.RTPTransceiverInit{
		Direction: webrtc.RTPTransceiverDirectionSendonly,
	})
	if err != nil {
		http.Error(w, "Failed to add transceiver", http.StatusInternalServerError)
		return
	}

	if err := peerConnection.SetRemoteDescription(webrtc.SessionDescription{
		Type: webrtc.SDPTypeOffer,
		SDP:  offer.SDP,
	}); err != nil {
		http.Error(w, "Failed to set remote description", http.StatusBadRequest)
		return
	}

	//peerConnection.OnConnectionStateChange(func(state webrtc.PeerConnectionState) {
	//	fmt.Printf("PeerConnectionStateChange: %v \n", state)
	//})
	//
	//peerConnection.OnICEConnectionStateChange(func(state webrtc.ICEConnectionState) {
	//	fmt.Printf("ICEConnectionStateChange: %v \n", state)
	//})
	//
	//peerConnection.OnICEGatheringStateChange(func(state webrtc.ICEGatheringState) {
	//	fmt.Printf("ICEGatheringStateChange: %v \n", state)
	//})

	peerConnection.OnTrack(func(track *webrtc.TrackRemote, receiver *webrtc.RTPReceiver) {
		go doProcessRemoteTrack(track, receiver, outBountTrack)
	})

	answer, err := peerConnection.CreateAnswer(nil)
	if err != nil {
		http.Error(w, "Failed to create answer", http.StatusInternalServerError)
		return
	}

	if err := peerConnection.SetLocalDescription(answer); err != nil {
		http.Error(w, "Failed to set local description", http.StatusInternalServerError)
		return
	}

	<-webrtc.GatheringCompletePromise(peerConnection)

	response := Answer{SDP: peerConnection.LocalDescription().SDP}
	w.Header().Set("Content-Type", "application/json")
	if err := json.NewEncoder(w).Encode(response); err != nil {
		http.Error(w, "Failed to encode response", http.StatusInternalServerError)
		return
	}
}

func doProcessRemoteTrack(t *webrtc.TrackRemote, r *webrtc.RTPReceiver, h264OutTrack *webrtc.TrackLocalStaticRTP) {
	fmt.Printf("Remote track %s and codec %v \n", t.Kind(), t.Codec().MimeType)

	udpConn, err := net.Dial("udp", "127.0.0.1:5000")
	if err != nil {
		panic(err)
	}
	defer udpConn.Close()

	buff := make([]byte, 1490)
	go func() {
		_, _, err := r.Read(buff)
		if err != nil {
			return
		}
	}()

	go doProcessSink(h264OutTrack)

	pktBuff := make([]byte, 1490)
	for {
		n, _, err := t.Read(pktBuff)
		if err != nil {
			fmt.Println("Track read error:", err)
			return
		}

		if _, err := udpConn.Write(pktBuff[:n]); err != nil {
			fmt.Println("UDP write error:", err)
			return
		}

	}
}

func doProcessSink(h264OutTrack *webrtc.TrackLocalStaticRTP) {
	addr, err := net.ResolveUDPAddr("udp", ":5001")
	if err != nil {
		log.Fatal(err)
	}
	conn, err := net.ListenUDP("udp", addr)
	if err != nil {
		log.Fatal(err)
	}
	defer conn.Close()

	//If no data arrives within 30s, a timeout error occurs, which can be handled later.
	//_ = conn.SetReadDeadline(time.Now().Add(30 * time.Second))

	buffer := make([]byte, 1490)
	for {
		n, _, err := conn.ReadFromUDP(buffer)
		if err != nil {
			//var netErr net.Error
			//if errors.As(err, &netErr) && netErr.Timeout() {
			//	fmt.Println("Read timeout, retrying...")
			//	_ = conn.SetReadDeadline(time.Now().Add(30 * time.Second))
			//	continue
			//}
			//log.Printf("Read error: %v", err)
			continue
		}

		_, err = h264OutTrack.Write(buffer[:n])
		if err != nil {
			log.Printf("Write error: %v", err)
			continue
		}
	}
}