基于pion生态的SFU实时音视频发布服务（一）

pion是google 大佬Sean-Der开源在github.com上的性能优异的基于golang开发的webrtc协议栈，同时他也是aws kvs的代码主要维护人，目前有两大sfu基于此构建ion、livekit，都有完善的sdk开发包，国内大佬的开源项目flutterwebrtc也比较深度的嵌入这些开源项目，可以为初入门的开发者节省大量的时间成本并提供经验借鉴。

在多媒体服务器的开发中，传统的c、c++版非常高效，而且稳定可靠，被大量采用比如srs,mediasoup,janus等，但对c/c++的技巧要求比较高，开发语言带来的复杂度让人望而生畏。pion的出现如一缕春风，让普通程序员都能快速入门高大上的sfu开发行业。并发性能，网络吞吐能力，以及SDK的完善度均可以支撑一般规模的应用，个人认为小团队创业公司首选路径就是采用好入门的生态完整的，社区活跃的开源代码库进行二次开发，享受生态带来的红利，在自己的实践中深入了解底层原理，然后再根据自己业务的需要逐步更改为自己的模块，不失为一条稳妥高效的技术路线图。

在实践过程中，由于webrtc在p2p实现后，有一部分需求是类似视频会议类的交互，srs等优秀的c++ 类sfu媒体服务器是非常好的选择，除此之外也可以自己动手做一个自己的sfu转发服务，讲IPC的p2p流转发至服务器，实现快捷的多人视频会议模式。以下是我采用go编写的发布至livekit/ion sfu的相关代码，你会发现利用go sdk实现起来是非常简单的事，话不多说，直接上代码，望大佬们多指正

package livekitclientimport ("context""errors""fmt""strings""sync""time"livekit "github.com/livekit/protocol/livekit"lksdk "github.com/livekit/server-sdk-go""github.com/livekit/server-sdk-go/pkg/samplebuilder"// "github.com/livekit/server-sdk-go/pkg/media/ivfwriter"// "github.com/livekit/server-sdk-go/pkg/samplebuilder"ionsdk "github.com/pion/ion-sdk-go""github.com/pion/rtp""github.com/pion/rtp/codecs""github.com/pion/webrtc/v3""github.com/pion/webrtc/v3/pkg/media""github.com/pion/webrtc/v3/pkg/media/h264writer""github.com/pion/webrtc/v3/pkg/media/ivfwriter""github.com/pion/webrtc/v3/pkg/media/oggwriter"// "github.com/xiangxud/rtmp_webrtc_server/config""github.com/xiangxud/rtmp_webrtc_server/identity""github.com/xiangxud/rtmp_webrtc_server/log"
)type SfuTrack struct {VideoRTPTrack *webrtc.TrackLocalStaticRTPAudioRTPTrack *webrtc.TrackLocalStaticRTPVideoTrack    *webrtc.TrackLocalStaticSampleAudioTrack    *webrtc.TrackLocalStaticSample
}
type LocalTrackPublication struct {LiveKitRoomConnect *lksdk.RoomIONRtc             *ionsdk.RTC// IONRoomConnect     *ionsdk.RoomVideopub        *lksdk.LocalTrackPublicationAudiopub        *lksdk.LocalTrackPublicationIONVideopub     *webrtc.RTPSenderIONAudiopub     *webrtc.RTPSenderLiveKitSfuTrack SfuTrackIONSfuTrack     SfuTracklivekitsb       *samplebuilder.SampleBuilder// publication *lksdk.RemoteTrackPublication// pliWriter       lksdk.PLIWriter// VideoRTPTrack *webrtc.TrackLocalStaticRTP// AudioRTPTrack *webrtc.TrackLocalStaticRTP// VideoTrack    *webrtc.TrackLocalStaticSample// AudioTrack    *webrtc.TrackLocalStaticSample// RemoteSDPOffer webrtc.SessionDescription `json:"sdpoffer"`// LocalSDPanswer webrtc.SessionDescription `json:"answer"`Streamname string// Trackname string
}
type Room struct {TokenCtx          context.ContextRoomClient   *lksdk.RoomServiceClientLiveKitRoom  *livekit.Roomlivekitlock  sync.Mutexionlock      sync.MutexIONRoom      *ionsdk.RoomIONConnector *ionsdk.Connector// IONRtc       *ionsdk.RTCLocaltracks map[string]*LocalTrackPublication
}func NewRoom(ctx context.Context, token *Token) *Room { //host, apiKey, apiSecret, roomName, identity string) *Room {return &Room{Ctx:         ctx,Token:       *token,Localtracks: make(map[string]*LocalTrackPublication),}
}func (r *Room) CreateliveKitRoom(roomName string) (*Room, error) {var err errorr.RoomClient = lksdk.NewRoomServiceClient(r.HostLiveKit, r.ApiKey, r.ApiSecret)r.RoomName = roomName// create a new roomif r.Ctx != nil {r.LiveKitRoom, err = r.RoomClient.CreateRoom(r.Ctx, &livekit.CreateRoomRequest{Name: roomName,})if err != nil {return nil, err}return r, nil}return nil, fmt.Errorf("context is invalid")
}
func (t *LocalTrackPublication) ConnectRoom(host, apikey, apisecret, roomname, identity string) error {// host := "<host>"// apiKey := "api-key"// apiSecret := "api-secret"// roomName := "myroom"// identity := "botuser"room, err := lksdk.ConnectToRoom(host, lksdk.ConnectInfo{APIKey:              apikey,APISecret:           apisecret,RoomName:            roomname,ParticipantIdentity: identity,}, &lksdk.RoomCallback{ParticipantCallback: lksdk.ParticipantCallback{OnTrackSubscribed: t.TrackSubscribed,},})if err != nil {panic(err)}// room, err := lksdk.ConnectToRoom(host, lksdk.ConnectInfo{//  APIKey:              apikey,//  APISecret:           apisecret,//   RoomName:            roomname,//    ParticipantIdentity: identity,// })// if err != nil {//    log.Debug(err)//    return err// }t.LiveKitRoomConnect = room// room.Callback.OnTrackSubscribed = t.TrackSubscribedreturn nil// room.Disconnect()
}// func (t *LocalTrackPublication) SetOffer(offer webrtc.SessionDescription) {
//  t.RemoteSDPOffer = offer
// }
// func (t *LocalTrackPublication) SetAnswer(answer webrtc.SessionDescription) {
//  t.LocalSDPanswer = answer
// }// func (t *Room) ConnectRoom(streamname string) error {
//  // host := "<host>"
//  // apiKey := "api-key"
//  // apiSecret := "api-secret"
//  // roomName := "myroom"
//  // identity := "botuser"
//  room, err := lksdk.ConnectToRoom(r.Host, lksdk.ConnectInfo{
//      APIKey:              r.ApiKey,
//      APISecret:           r.ApiSecret,
//      RoomName:            r.RoomName,
//      ParticipantIdentity: streamname,
//  })
//  if err != nil {
//      panic(err)
//  }
//  room.Callback.OnTrackSubscribed = r.TrackSubscribed
//  if t := r.Localtracks[streamname]; t != nil {
//      t.RoomConnect = room
//  }
//  return nil
//  // room.Disconnect()
// }
func (t *LocalTrackPublication) TrackSubscribed(track *webrtc.TrackRemote, publication *lksdk.RemoteTrackPublication, rp *lksdk.RemoteParticipant) {// }// func onTrackSubscribed(track *webrtc.TrackRemote, publication *lksdk.RemoteTrackPublication, rp *lksdk.RemoteParticipant) {fileName := fmt.Sprintf("%s-%s", rp.Identity(), track.ID())fmt.Println("write track to file ", fileName)NewTrackWriter(track, rp.WritePLI, fileName)// t.pliWriter=
}const (maxVideoLate = 1000 // nearly 2s for fhd videomaxAudioLate = 200  // 4s for audio
)type TrackWriter struct {sb     *samplebuilder.SampleBuilderwriter media.Writertrack  *webrtc.TrackRemote
}func NewTrackWriter(track *webrtc.TrackRemote, pliWriter lksdk.PLIWriter, fileName string) (*TrackWriter, error) {var (sb     *samplebuilder.SampleBuilderwriter media.Writererr    error)switch {case strings.EqualFold(track.Codec().MimeType, "video/vp8"):sb = samplebuilder.New(maxVideoLate, &codecs.VP8Packet{}, track.Codec().ClockRate, samplebuilder.WithPacketDroppedHandler(func() {pliWriter(track.SSRC())}))// ivfwriter use frame count as PTS, that might cause video played in a incorrect framerate(fast or slow)writer, err = ivfwriter.New(fileName + ".ivf")case strings.EqualFold(track.Codec().MimeType, "video/h264"):sb = samplebuilder.New(maxVideoLate, &codecs.H264Packet{}, track.Codec().ClockRate, samplebuilder.WithPacketDroppedHandler(func() {pliWriter(track.SSRC())}))writer, err = h264writer.New(fileName + ".h264")case strings.EqualFold(track.Codec().MimeType, "audio/opus"):sb = samplebuilder.New(maxAudioLate, &codecs.OpusPacket{}, track.Codec().ClockRate)writer, err = oggwriter.New(fileName+".ogg", 48000, track.Codec().Channels)default:return nil, errors.New("unsupported codec type")}if err != nil {return nil, err}t := &TrackWriter{sb:     sb,writer: writer,track:  track,}go t.start()return t, nil
}
func (t *TrackWriter) start() {defer t.writer.Close()for {pkt, _, err := t.track.ReadRTP()if err != nil {break}t.sb.Push(pkt)for _, p := range t.sb.PopPackets() {t.writer.WriteRTP(p)}}
}func (r *Room) TrackSendLivekitRtpPackets(trackname, kind string, data []byte) (n int, err error) {if trackname == "" {log.Debug("Track name is null")return 0, fmt.Errorf("input trackname is null")}// var t *webrtc.TrackLocalStaticSamplevar t *webrtc.TrackLocalStaticRTPtrack := r.Localtracks[trackname]if track == nil {log.Debug("TrackSendLivekitRtpPackets: ", "Track is nil ->", trackname, "<- no to publish")return 0, fmt.Errorf(" track is null,no to publish")}if kind == "video" {t = track.LiveKitSfuTrack.VideoRTPTrack} else if kind == "audio" {t = track.LiveKitSfuTrack.AudioRTPTrack}if t == nil {log.Debug("TrackSendLivekitRtpPackets: ", "t is nil ->", trackname, "<- no to publish")return 0, fmt.Errorf(" track is null,no to publish")}if kind == "video" {packets := &rtp.Packet{}if err := packets.Unmarshal(data); err != nil {return 0, err}track.livekitsb.Push(packets)for _, p := range track.livekitsb.PopPackets() {err = t.WriteRTP(p)if err != nil {log.Debug("[TrackSendIonRtpPackets] error", err)return 0, err}}//n, err = t.Write(data)return len(data), nil} else {n, err = t.Write(data)return n, err}
}
func (r *Room) TrackSendLivekitData(trackname, kind string, data []byte, duration time.Duration) error {if trackname == "" {log.Debug("Track name is null")return fmt.Errorf("input trackname is null")}var t *webrtc.TrackLocalStaticSampletrack := r.Localtracks[trackname]if track == nil {log.Debug("TrackSendLivekitData:", "Track is nil ->", trackname, "<- no to publish")return fmt.Errorf(" track is null,no to publish")}if kind == "video" {t = track.LiveKitSfuTrack.VideoTrack} else if kind == "audio" {t = track.LiveKitSfuTrack.AudioTrack}if t == nil {log.Debug("TrackSendLivekitData: ", "t is nil ->", trackname, "<- no to publish")return fmt.Errorf(" track is null,no to publish")}if videoErr := t.WriteSample(media.Sample{Data:     data,Duration: duration,}); videoErr != nil {log.Debug("WriteSample err", videoErr)// r.ConnectRoom()return nil //fmt.Errorf("WriteSample err %s", vedioErr)}return nil
}
func (r *Room) TrackClose(streamname string) error {if t := r.Localtracks[streamname]; t != nil {if t.IONRtc != nil {t.IONRtc.Close()}if t.LiveKitRoomConnect == nil || t.LiveKitRoomConnect.LocalParticipant == nil {return nil}t.LiveKitRoomConnect.LocalParticipant.UnpublishTrack(t.LiveKitRoomConnect.LocalParticipant.SID())t.LiveKitRoomConnect.Disconnect()r.Localtracks[streamname] = nillog.Debug("track ", streamname, "lost ,now removed", r)//r.RoomConnect.LocalParticipant.UnpublishTrack(r.RoomConnect.LocalParticipant.SID())// r.Localtracks[streamname+"-video"]}return nil
}
func (r *Room) TrackPublished(streamname string) error {// - `in` implements io.ReadCloser, such as buffer or file// - `mime` has to be one of webrtc.MimeType...// videoTrack, err := lksdk.NewLocalSampleTrack(webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeH264})r.livekitlock.Lock()defer r.livekitlock.Unlock()if r.Ctx != nil && r.LiveKitRoom == nil {var err errorsn, _ := identity.GetSN()r.LiveKitRoom, err = r.RoomClient.CreateRoom(r.Ctx, &livekit.CreateRoomRequest{Name: sn,})if err != nil {log.Debug("room->", sn, "create room ok", r)return err}}t := r.Localtracks[streamname]if t == nil {t = &LocalTrackPublication{Streamname: streamname}t.ConnectRoom(r.HostLiveKit, r.ApiKey, r.ApiSecret, r.RoomName, streamname+":"+r.Identity)log.Debug("track->", streamname, "<-is nil ,Connect room", t, r)} else {if t.LiveKitRoomConnect == nil {t.ConnectRoom(r.HostLiveKit, r.ApiKey, r.ApiSecret, r.RoomName, streamname+":"+r.Identity)log.Debug("track->", streamname, "<-is nil ,re Connect room", t, r)}}if t.LiveKitSfuTrack.VideoTrack == nil && t.Videopub == nil && t.LiveKitSfuTrack.AudioTrack == nil && t.Audiopub == nil {videoTrack, err := webrtc.NewTrackLocalStaticSample(webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeH264}, streamname+"-video", streamname)if err != nil {panic(err)}// r.RoomClient.MutePublishedTrack(r.Ctx,)// var local_video *lksdk.LocalTrackPublicationif t.Videopub, err = t.LiveKitRoomConnect.LocalParticipant.PublishTrack(videoTrack, &lksdk.TrackPublicationOptions{Name: streamname + "-video"}); err != nil {log.Debug("Error publishing video track->", err)return err}t.LiveKitSfuTrack.VideoTrack = videoTrack// r.Localtracks[streamname] = &LocalTrackPublication{p: local_video, Track: videoTrack, Trackname: streamname + "-video"}log.Debug("[TrackPublished]", "published video track -> ", streamname)if t.LiveKitSfuTrack.AudioTrack == nil {audioTrack, err := webrtc.NewTrackLocalStaticSample(webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeOpus}, streamname+"-audio", streamname)//audioTrack, err := lksdk.NewLocalSampleTrack(webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeOpus})if err != nil {panic(err)}// var local_audio *lksdk.LocalTrackPublicationif t.Audiopub, err = t.LiveKitRoomConnect.LocalParticipant.PublishTrack(audioTrack, &lksdk.TrackPublicationOptions{Name: streamname + "-audio"}); err != nil {log.Debug("Error publishing audio track->", err)return err}t.LiveKitSfuTrack.AudioTrack = audioTracklog.Debug("[TrackPublished]", "published audio track -> ", streamname)}r.Localtracks[streamname] = t} else {log.Debug(streamname, "is exit publish")}return nil
}//from call package media stream
func (r *Room) RTPTrackPublished(trackRemote []*webrtc.TrackRemote, streamname string) error {// - `in` implements io.ReadCloser, such as buffer or file// - `mime` has to be one of webrtc.MimeType...// videoTrack, err := lksdk.NewLocalSampleTrack(webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeH264})// r.mux.lock()r.livekitlock.Lock()defer r.livekitlock.Unlock()if r.Ctx != nil && r.LiveKitRoom == nil {var err errorsn, _ := identity.GetSN()r.LiveKitRoom, err = r.RoomClient.CreateRoom(r.Ctx, &livekit.CreateRoomRequest{Name: sn,})if err != nil {log.Debug("room->", sn, "create room ok", r)return err}}t := r.Localtracks[streamname]if t == nil {t = &LocalTrackPublication{Streamname: streamname}t.ConnectRoom(r.HostLiveKit, r.ApiKey, r.ApiSecret, r.RoomName, streamname+":"+r.Identity)log.Debug("track->", streamname, "<-is nil ,Connect room", t, r)} else {if t.LiveKitRoomConnect == nil {t.ConnectRoom(r.HostLiveKit, r.ApiKey, r.ApiSecret, r.RoomName, streamname+":"+r.Identity)log.Debug("track->", streamname, "<-is nil ,re Connect room", t, r)}}for _, v := range trackRemote {if v.Kind().String() == "video" {if t.LiveKitSfuTrack.VideoRTPTrack == nil {if strings.Contains(v.Codec().MimeType, "video") {if t.livekitsb == nil {t.livekitsb = samplebuilder.New(maxVideoLate, &codecs.H264Packet{}, v.Codec().ClockRate) //, samplebuilder.WithPacketDroppedHandler(func() {//     t.Videopub.WritePLI(trackRemote.SSRC())// }))// t.Videopub.WritePLI}videoRTPTrack, err := webrtc.NewTrackLocalStaticRTP(v.Codec().RTPCodecCapability, streamname+"-video"+v.ID(), streamname)if err != nil {panic(err)}// r.RoomClient.MutePublishedTrack(r.Ctx,)// var local_video *lksdk.LocalTrackPublicationif t.LiveKitRoomConnect != nil && t.LiveKitRoomConnect.LocalParticipant != nil {if t.Videopub, err = t.LiveKitRoomConnect.LocalParticipant.PublishTrack(videoRTPTrack, &lksdk.TrackPublicationOptions{Name: streamname + "-video" + v.ID()}); err != nil {log.Debug("Error publishing video RTP track->", err)return err}t.LiveKitSfuTrack.VideoRTPTrack = videoRTPTrackr.Localtracks[streamname] = t// r.Localtracks[streamname] = &LocalTrackPublication{p: local_video, Track: videoTrack, Trackname: streamname + "-video"}log.Debug("[RTPTrackPublished]", "published video track -> ", streamname)}}}} else {if v.Kind().String() == "audio" {if t.LiveKitSfuTrack.AudioRTPTrack == nil {if strings.Contains(v.Codec().MimeType, "audio") {audioRTPTrack, err := webrtc.NewTrackLocalStaticRTP(v.Codec().RTPCodecCapability, streamname+"-audio"+v.ID(), streamname)//audioTrack, err := lksdk.NewLocalSampleTrack(webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeOpus})if err != nil {panic(err)}// var local_audio *lksdk.LocalTrackPublicationif t.LiveKitRoomConnect != nil && t.LiveKitRoomConnect.LocalParticipant != nil {if t.Audiopub, err = t.LiveKitRoomConnect.LocalParticipant.PublishTrack(audioRTPTrack, &lksdk.TrackPublicationOptions{Name: streamname + "-audio" + v.ID()}); err != nil {log.Debug("Error publishing audio track", err)return err}t.LiveKitSfuTrack.AudioRTPTrack = audioRTPTrackr.Localtracks[streamname] = tlog.Debug("[RTPTrackPublished]", "published audio track -> ", streamname)}}}}}}// r.Localtracks[streamname] = t// } else {//   log.Debug(streamname, "is exit publish")// }return nil
}
func (r *Room) Close() {for _, t := range r.Localtracks {if t == nil {continue}if t.LiveKitRoomConnect != nil {t.LiveKitRoomConnect.Disconnect()}}if r.IONRoom != nil {r.IONRoom.Close()}
}

obs 推流至转发服务器，livekit视频会议端展现

浏览器通过信令系统推流webrtc流至视频会议

基于pion生态的SFU实时音视频发布服务（一）相关推荐

【金猿产品展】拍乐云——新一代实时音视频云服务，构建云上的每一次美好互动...
拍乐云产品本项目由拍乐云投递并参与"数据猿年度金猿策划活动--2021大数据产业创新服务产品榜单及奖项"评选. 数据智能产业创新服务媒体 --聚焦数智 · 改变商业拍乐云提供的 ...
七牛云 RTN：基于 WebRTC 零基础搭建实时音视频平台
近年来,在线教育.狼人杀.在线抓娃娃.线上 KTV 等多人视频互动模式不断涌现,实时音视频通信风头正劲,实时音视频技术 WebRTC 也因此受到了广泛关注.相关数据显示,2017-2021 年期间,全 ...
最右显示请求服务器不存在,修改合流任务_实时音视频 RTC_服务端API参考_合流任务管理_华为云...
响应示例状态码: 200 修改成功 { "jobs" : [ { "job_id" : "607824b4fa163e19fe301cc817dda ...
18个实时音视频开发中会用到开源项目
实时音视频的开发学习有很多可以参考的开源项目.一个实时音视频应用共包括几个环节:采集.编码.前后处理.传输.解码.缓冲.渲染等很多环节.每一个细分环节,还有更细分的技术模块.比如,前后处理环节有美颜. ...
如何零门槛搭建实时音视频通信平台
迅达云视频云产品全面更新,为用户带来全新的一站式服务体验. 迅达云全面拥抱下一代互联网音视频通信开放标准 WebRTC,依托团队多年行业经验积累,结合迅达云实时音视频通信 SDK,匠心打造了一站式实时 ...
实时音视频聊天中超低延迟架构的思考与技术实践
1.前言从直播在线上抓娃娃,不断变化的是玩法的创新,始终不变的是对超低延迟的苛求.实时架构是超低延迟的基石,如何在信源编码.信道编码和实时传输整个链条来构建实时架构?在实时架构的基础之上,如果通过优 ...
李幸原：看好实时音视频在教育与医疗的前景
LiveVideoStack采访了三体云实时视频高级工程师李幸原,从无线桌面.远程医疗到互联网直播,李幸原分析了这几种场景下的技术难点.在不可靠的公网上,三体云抛弃了传统的CDN+TCP,构建起全新的 ...
RTC Meetup | 这可能是年底最大、最有料的实时音视频开发者聚会交流
活动报名链接:http://gz.huodongxing.com/event/1418096252900 回首2017,从直播.狼人杀到在线抓娃娃,不断迭起的风口让我们看到,实时音视频技术已经不断地应 ...
【活动预告】即构受邀分享实时音视频服务架构实践
今年年初,受所服务的线上应用爆炸式增长的影响,即构作为底层音视频服务商,平台数据节节攀升,高达数千万的并发,日均音视频互动时长突破20亿分钟. 要扛住千万级的高并发,首先要有一个支持千万级并发的底层架 ...

基于pion生态的SFU实时音视频发布服务（一）

基于pion生态的SFU实时音视频发布服务（一）相关推荐

最新文章

热门文章