plugin-webrtc/batcher.go

package webrtc

import (
	"encoding/json"
	"fmt"
	"time"

	"github.com/pion/rtcp"
	. "github.com/pion/webrtc/v3"
	"go.uber.org/zap"
	. "m7s.live/engine/v4"
	. "m7s.live/engine/v4/track"
)

type Signal struct {
	Type       string   `json:"type"`
	StreamList []string `json:"streamList"`
	Offer      string   `json:"offer"`
	Answer     string   `json:"answer"`
	StreamPath string   `json:"streamPath"`
}
type BatchUplink struct {
	Publisher
	StreamPath string
}
type WebRTCBatcher struct {
	WebRTCIO
	PageSize      int
	PageNum       int
	subscribers   []*WebRTCBatchSubscriber
	signalChannel *DataChannel
	BatchUplink
}

func (suber *WebRTCBatcher) Start() (err error) {
	suber.OnICECandidate(func(ice *ICECandidate) {
		if ice != nil {
			WebRTCPlugin.Info(ice.ToJSON().Candidate)
		}
	})
	suber.OnDataChannel(func(d *DataChannel) {
		WebRTCPlugin.Info("OnDataChannel:" + d.Label())
		suber.signalChannel = d
		suber.signalChannel.OnMessage(suber.Signal)
	})
	if err = suber.SetRemoteDescription(SessionDescription{Type: SDPTypeOffer, SDP: suber.SDP}); err != nil {
		return
	}
	suber.OnConnectionStateChange(func(pcs PeerConnectionState) {
		WebRTCPlugin.Info("Connection State has changed:" + pcs.String())
		switch pcs {
		case PeerConnectionStateConnected:
			suber.OnTrack(func(track *TrackRemote, receiver *RTPReceiver) {
				if suber.Publisher.Stream == nil {
					WebRTCPlugin.Publish(suber.StreamPath, &suber.BatchUplink)
				}
				if suber.Publisher.Stream == nil {
					return
				}
				puber := &suber.Publisher
				if codec := track.Codec(); track.Kind() == RTPCodecTypeAudio {
					if puber.AudioTrack == nil {
						switch codec.PayloadType {
						case 8:
							puber.AudioTrack = NewG711(puber.Stream, true)
						case 0:
							puber.AudioTrack = NewG711(puber.Stream, false)
						default:
							puber.AudioTrack = nil
							return
						}
					}
					for {
						b := make([]byte, 1460)
						if i, _, err := track.Read(b); err == nil {
							puber.AudioTrack.WriteRTP(b[:i])
						} else {
							return
						}
					}
				} else {
					go func() {
						ticker := time.NewTicker(time.Millisecond * webrtcConfig.PLI)
						for {
							select {
							case <-ticker.C:
								if rtcpErr := suber.WriteRTCP([]rtcp.Packet{&rtcp.PictureLossIndication{MediaSSRC: uint32(track.SSRC())}}); rtcpErr != nil {
									fmt.Println(rtcpErr)
								}
							case <-puber.Done():
								return
							}
						}
					}()
					puber.VideoTrack = NewH264(puber.Stream)
					for {
						b := make([]byte, 1460)
						if i, _, err := track.Read(b); err == nil {
							puber.VideoTrack.WriteRTP(b[:i])
						} else {
							return
						}
					}
				}
			})
		case PeerConnectionStateDisconnected, PeerConnectionStateFailed:
			for _, sub := range suber.subscribers {
				go sub.Stop()
			}
			suber.PeerConnection.Close()
		}
	})
	return
}

func (suber *WebRTCBatcher) Signal(msg DataChannelMessage) {
	var s Signal
	var removeMap = map[string]string{"type": "remove", "streamPath": ""}
	// var offer SessionDescription
	if err := json.Unmarshal(msg.Data, &s); err != nil {
		WebRTCPlugin.Error("Signal", zap.Error(err))
	} else {
		switch s.Type {
		case "streamPath":
			suber.StreamPath = s.StreamPath
		case "subscribe":
			if err = suber.SetRemoteDescription(SessionDescription{Type: SDPTypeOffer, SDP: s.Offer}); err != nil {
				WebRTCPlugin.Error("Signal SetRemoteDescription", zap.Error(err))
				return
			}
			for _, streamPath := range s.StreamList {
				sub := &WebRTCBatchSubscriber{}
				sub.WebRTCIO = suber.WebRTCIO
				if err = WebRTCPlugin.SubscribeExist(streamPath, sub); err == nil {
					suber.subscribers = append(suber.subscribers, sub)
					go func() {
						sub.PlayRTP()
						if sub.audioSender != nil {
							suber.RemoveTrack(sub.audioSender)
						}
						if sub.videoSender != nil {
							suber.RemoveTrack(sub.videoSender)
						}
						if sub.DC != nil {
							sub.DC.Close()
						}
						removeMap["streamPath"] = streamPath
						b, _ := json.Marshal(removeMap)
						suber.signalChannel.SendText(string(b))
					}()
				} else {
					removeMap["streamPath"] = streamPath
					b, _ := json.Marshal(removeMap)
					suber.signalChannel.SendText(string(b))
				}
			}
			var answer string
			if answer, err = suber.GetAnswer(); err == nil {
				b, _ := json.Marshal(map[string]string{"type": "answer", "sdp": answer})
				err = suber.signalChannel.SendText(string(b))
			}
			if err != nil {
				WebRTCPlugin.Error("Signal GetAnswer", zap.Error(err))
				return
			}
		// if offer, err = suber.CreateOffer(nil); err == nil {
		// 	b, _ := json.Marshal(offer)
		// 	err = suber.signalChannel.SendText(string(b))
		// 	suber.SetLocalDescription(offer)
		// }
		case "publish":
			if err = suber.SetRemoteDescription(SessionDescription{Type: SDPTypeOffer, SDP: s.Offer}); err != nil {
				WebRTCPlugin.Error("Signal SetRemoteDescription", zap.Error(err))
				return
			}
			var answer string
			if answer, err = suber.GetAnswer(); err == nil {
				b, _ := json.Marshal(map[string]string{"type": "answer", "sdp": answer})
				err = suber.signalChannel.SendText(string(b))
			}
			if err != nil {
				WebRTCPlugin.Error("Signal GetAnswer", zap.Error(err))
				return
			}
		case "answer":
			if err = suber.SetRemoteDescription(SessionDescription{Type: SDPTypeAnswer, SDP: s.Answer}); err != nil {
				WebRTCPlugin.Error("Signal SetRemoteDescription", zap.Error(err))
				return
			}
		}
	}
}