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

          上篇文章说了风头正健的pion生态之livekit,现在轮到pion生态第一个sfu ion,这个由国内大佬鱼大等主持开发两年多开源项目，为国人乃至开源社区普及pion起了至关重要的作用，得到了Sean-Der的大力支持，也汇集了众多高手加盟，livekit的风格也深受其影响，下面是发布ion的习作，望各位大佬指正

    1. package livekitclient
    2.  
    3. import (
    4. 	"fmt"
    5. 	"strings"
    6. 	"time"
    7.  
    8. 	// "github.com/livekit/server-sdk-go/pkg/media/ivfwriter"
    9. 	"github.com/livekit/server-sdk-go/pkg/samplebuilder"
    10. 	ionsdk "github.com/pion/ion-sdk-go"
    11. 	"github.com/pion/rtp"
    12. 	"github.com/pion/webrtc/v3"
    13. 	"github.com/pion/webrtc/v3/pkg/media"
    14. 	"github.com/pion/webrtc/v3/pkg/media/h264writer"
    15. 	"github.com/pion/webrtc/v3/pkg/media/ivfwriter"
    16. 	"github.com/pion/webrtc/v3/pkg/media/oggwriter"
    17. 	"github.com/xiangxud/rtmp_webrtc_server/identity"
    18. 	"github.com/xiangxud/rtmp_webrtc_server/log"
    19. 	// "github.com/livekit/server-sdk-go/pkg/samplebuilder"
    20. )
    21.  
    22. const (
    23. // sid = "ion"
    24. // uid = ionsdk.RandomKey(6)
    25. )
    26.  
    27. func (t *LocalTrackPublication) saveToDisk(track *webrtc.TrackRemote, receiver *webrtc.RTPReceiver) {
    28. 	codec := track.Codec()
    29. 	var fileWriter media.Writer
    30. 	var err error
    31. 	if strings.EqualFold(codec.MimeType, webrtc.MimeTypeOpus) {
    32. 		log.Infof("Got Opus track, saving to disk as ogg (48 kHz, 2 channels)")
    33. 		fileWriter, err = oggwriter.New(fmt.Sprintf("%d_%d.ogg", codec.PayloadType, track.SSRC()), 48000, 2)
    34. 	} else if strings.EqualFold(codec.MimeType, webrtc.MimeTypeVP8) {
    35. 		log.Infof("Got VP8 track, saving to disk as ivf")
    36. 		fileWriter, err = ivfwriter.New(fmt.Sprintf("%d_%d.ivf", codec.PayloadType, track.SSRC()))
    37. 	} else if strings.EqualFold(codec.MimeType, webrtc.MimeTypeH264) {
    38. 		log.Infof("Got H264 track, saving to disk as h264")
    39. 		fileWriter, err = h264writer.New(fmt.Sprintf("%d_%d.h264", codec.PayloadType, track.SSRC()))
    40. 	}
    41. 	if err != nil {
    42. 		log.Errorf("error: %v", err)
    43. 		fileWriter.Close()
    44. 		return
    45. 	}
    46.  
    47. 	for {
    48. 		rtpPacket, _, err := track.ReadRTP()
    49. 		if err != nil {
    50. 			log.Warnf("track.ReadRTP error: %v", err)
    51. 			break
    52. 		}
    53. 		if err := fileWriter.WriteRTP(rtpPacket); err != nil {
    54. 			log.Warnf("fileWriter.WriteRTP error: %v", err)
    55. 			break
    56. 		}
    57. 	}
    58. }
    59. func (t *LocalTrackPublication) INORoomRTCJoin(r *Room, streamname, identify string) (*ionsdk.RTC, error) {
    60.  
    61. 	// join room
    62. 	uid := streamname + ":" + identify
    63. 	err := r.IONRoom.Join(
    64. 		ionsdk.JoinInfo{
    65. 			Sid:         identify,
    66. 			Uid:         uid,
    67. 			DisplayName: uid,
    68. 			Role:        ionsdk.Role_Host,
    69. 			Protocol:    ionsdk.Protocol_WebRTC,
    70. 			Direction:   ionsdk.Peer_BILATERAL,
    71. 		},
    72. 	)
    73.  
    74. 	if err != nil {
    75. 		log.Errorf("Join error: %v", err)
    76. 		return nil, err
    77. 	}
    78. 	// new sdk engine
    79. 	config := ionsdk.RTCConfig{
    80. 		WebRTC: ionsdk.WebRTCTransportConfig{
    81. 			VideoMime: ionsdk.MimeTypeH264,
    82. 		},
    83. 	}
    84. 	joinedch := make(chan struct{})
    85. 	r.IONRoom.OnJoin = func(success bool, info ionsdk.RoomInfo, err error) {
    86. 		// THIS IS ROOM SINGAL API
    87. 		// ===============================
    88. 		rtc, err1 := ionsdk.NewRTC(r.IONConnector, config)
    89. 		if err1 != nil {
    90. 			log.Error(err1)
    91. 			return
    92. 		}
    93.  
    94. 		// user define receiving rtp
    95. 		rtc.OnTrack = t.saveToDisk
    96. 		rtc.GetPubTransport().GetPeerConnection().OnICEConnectionStateChange(func(state webrtc.ICEConnectionState) {
    97. 			log.Infof("Connection state changed: %s", state)
    98. 		})
    99. 		rtc.OnDataChannel = func(dc *webrtc.DataChannel) {
    100. 			log.Infof("dc: %v", dc.Label())
    101. 		}
    102.  
    103. 		rtc.OnError = func(err error) {
    104. 			log.Errorf("err: %v", err)
    105. 		}
    106. 		configjoin := ionsdk.JoinConfig{}
    107. 		configjoin.SetNoPublish()
    108. 		configjoin.SetNoSubscribe()
    109. 		configjoin.SetNoAutoSubscribe()
    110. 		uid = ionsdk.RandomKey(6)
    111. 		err1 = rtc.Join(identify, uid, &configjoin)
    112. 		if err1 != nil {
    113. 			log.Errorf("error: %v", err1)
    114. 			return
    115. 		}
    116. 		log.Infof("rtc.Join ok sid=%v username =%v", identify, uid)
    117.  
    118. 		// err = rtc.Join(session, ionsdk.RandomKey(4))
    119. 		// if err != nil {
    120. 		// 	log.Errorf("error: %v", err)
    121. 		// 	return
    122. 		// }
    123. 		t.IONRtc = rtc
    124. 		joinedch <- struct{}{}
    125. 	}
    126. 	<-joinedch
    127. 	return t.IONRtc, nil
    128. }
    129. func (r *Room) CreateIonRoom(addr, session string) (*ionsdk.Room, error) {
    130. 	log.Debug("CreateIonRoom: ", addr, "  session: ", session)
    131. 	r.HostIon = addr
    132. 	connector := ionsdk.NewConnector(addr)
    133. 	// uid := ionsdk.RandomKey(6)
    134. 	room := ionsdk.NewRoom(connector)
    135. 	peers := room.GetPeers(session)
    136. 	if len(peers) != 0 {
    137. 		log.Debug("room is exit peers :", peers)
    138. 		// err := room.Join(
    139. 		// 	ionsdk.JoinInfo{
    140. 		// 		Sid:         session,
    141. 		// 		Uid:         uid,
    142. 		// 		DisplayName: uid,
    143. 		// 		Role:        ionsdk.Role_Host,
    144. 		// 		Protocol:    ionsdk.Protocol_WebRTC,
    145. 		// 		Direction:   ionsdk.Peer_BILATERAL,
    146. 		// 	},
    147. 		// )
    148.  
    149. 		// if err != nil {
    150. 		// 	log.Errorf("Join error: %v", err)
    151. 		// 	return nil, err
    152. 		// }
    153. 		r.IONRoom = room
    154. 		r.IONConnector = connector
    155. 		return room, nil
    156. 	}
    157. 	// THIS IS ROOM MANAGEMENT API
    158. 	// ==========================
    159. 	// create room
    160. 	err := room.CreateRoom(ionsdk.RoomInfo{Sid: session})
    161. 	if err != nil {
    162. 		log.Errorf("error:%v", err)
    163. 		return nil, err
    164. 	}
    165.  
    166. 	// // new sdk engine
    167. 	// config := ionsdk.RTCConfig{
    168. 	// 	WebRTC: ionsdk.WebRTCTransportConfig{
    169. 	// 		VideoMime: ionsdk.MimeTypeH264,
    170. 	// 	},
    171. 	// }
    172. 	// // THIS IS ROOM SINGAL API
    173. 	// // ===============================
    174. 	// rtc, err := ionsdk.NewRTC(connector, config)
    175. 	// if err != nil {
    176. 	// 	log.Error(err)
    177. 	// 	// return err
    178. 	// }
    179.  
    180. 	// // user define receiving rtp
    181. 	// rtc.OnTrack = r.saveToDisk
    182.  
    183. 	// rtc.OnDataChannel = func(dc *webrtc.DataChannel) {
    184. 	// 	log.Infof("dc: %v", dc.Label())
    185. 	// }
    186.  
    187. 	// rtc.OnError = func(err error) {
    188. 	// 	log.Errorf("err: %v", err)
    189. 	// }
    190. 	// err = rtc.Join(session, ionsdk.RandomKey(4))
    191. 	// if err != nil {
    192. 	// 	log.Errorf("error: %v", err)
    193. 	// 	return nil, err
    194. 	// }
    195. 	// log.Infof("rtc.Join ok sid=%v", session)
    196. 	// // err = rtc.Join(session, ionsdk.RandomKey(4))
    197. 	// // if err != nil {
    198. 	// // 	log.Errorf("error: %v", err)
    199. 	// // 	return
    200. 	// // }
    201. 	// r.IONRtc = rtc
    202. 	room.OnJoin = func(success bool, info ionsdk.RoomInfo, err error) {
    203. 		log.Infof("OnJoin success = %v, info = %v, err = %v", success, info, err)
    204.  
    205. 	}
    206. 	room.OnLeave = func(success bool, err error) {
    207. 		log.Infof("OnLeave success = %v err = %v", success, err)
    208. 	}
    209.  
    210. 	room.OnPeerEvent = func(state ionsdk.PeerState, peer ionsdk.PeerInfo) {
    211. 		log.Infof("OnPeerEvent state = %v, peer = %v", state, peer)
    212. 	}
    213.  
    214. 	room.OnMessage = func(from string, to string, data map[string]interface{}) {
    215. 		log.Infof("OnMessage from = %v, to = %v, data = %v", from, to, data)
    216. 	}
    217.  
    218. 	room.OnDisconnect = func(sid, reason string) {
    219. 		log.Infof("OnDisconnect sid = %v, reason = %v", sid, reason)
    220. 	}
    221.  
    222. 	room.OnRoomInfo = func(info ionsdk.RoomInfo) {
    223. 		log.Infof("OnRoomInfo info=%v", info)
    224. 	}
    225.  
    226. 	// join room
    227. 	// err = room.Join(
    228. 	// 	ionsdk.JoinInfo{
    229. 	// 		Sid:         session,
    230. 	// 		Uid:         uid,
    231. 	// 		DisplayName: uid,
    232. 	// 		Role:        ionsdk.Role_Host,
    233. 	// 		Protocol:    ionsdk.Protocol_WebRTC,
    234. 	// 		Direction:   ionsdk.Peer_BILATERAL,
    235. 	// 	},
    236. 	// )
    237.  
    238. 	// if err != nil {
    239. 	// 	log.Errorf("Join error: %v", err)
    240. 	// 	return nil, err
    241. 	// }
    242. 	r.IONRoom = room
    243. 	r.IONConnector = connector
    244. 	return room, nil
    245. }
    246.  
    247. // func (t *LocalTrackPublication) ConnectRoomIon(host,  identity string) error {
    248. // 	// host := ""
    249. // 	// apiKey := "api-key"
    250. // 	// apiSecret := "api-secret"
    251. // 	// roomName := "myroom"
    252. // 	// identity := "botuser"
    253. // 	room, err := lksdk.ConnectToRoom(host, lksdk.ConnectInfo{
    254. // 		APIKey:              apikey,
    255. // 		APISecret:           apisecret,
    256. // 		RoomName:            roomname,
    257. // 		ParticipantIdentity: identity,
    258. // 	})
    259. // 	if err != nil {
    260. // 		log.Debug(err)
    261. // 		return err
    262. // 	}
    263. // 	t.LiveKitRoomConnect = room
    264. // 	room.Callback.OnTrackSubscribed = t.TrackSubscribed
    265. // 	return nil
    266. // 	// room.Disconnect()
    267. // }
    268. func (r *Room) TrackPublished_to_ION(streamname string) error {
    269. 	// - `in` implements io.ReadCloser, such as buffer or file
    270. 	// - `mime` has to be one of webrtc.MimeType...
    271. 	// videoTrack, err := lksdk.NewLocalSampleTrack(webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeH264})
    272. 	r.ionlock.Lock()
    273. 	defer r.ionlock.Unlock()
    274. 	if r.Ctx != nil && r.IONRoom == nil {
    275. 		var err error
    276. 		sn, _ := identity.GetSN()
    277. 		r.IONRoom, err = r.CreateIonRoom(r.HostIon, sn)
    278. 		if err != nil {
    279. 			log.Debug("room->", sn, "create room ok", r)
    280. 			return err
    281. 		}
    282.  
    283. 	}
    284. 	t := r.Localtracks[streamname]
    285. 	if t == nil {
    286. 		t = &LocalTrackPublication{Streamname: streamname}
    287. 		t.INORoomRTCJoin(r, streamname, r.Identity)
    288. 		log.Debug("ion track->", streamname, "<-is nil ,Connect room", t, r)
    289. 	} else {
    290. 		if !t.IONRtc.Connected() {
    291. 			t.INORoomRTCJoin(r, streamname, r.Identity)
    292. 			log.Debug("ion track->", streamname, "<-is nil ,re Connect room", t, r)
    293. 		}
    294. 	}
    295. 	if t.IONSfuTrack.VideoTrack == nil && t.Videopub == nil && t.IONSfuTrack.AudioTrack == nil && t.Audiopub == nil {
    296. 		videoTrack, err := webrtc.NewTrackLocalStaticSample(webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeH264}, streamname+"-video", streamname)
    297. 		if err != nil {
    298. 			panic(err)
    299. 		}
    300. 		// r.RoomClient.MutePublishedTrack(r.Ctx,)
    301. 		// var local_video *lksdk.LocalTrackPublication
    302. 		if _, err = t.IONRtc.Publish(videoTrack); err != nil {
    303. 			log.Debug("Error publishing video track->", err)
    304. 			return err
    305. 		}
    306. 		t.IONSfuTrack.VideoTrack = videoTrack
    307. 		// r.Localtracks[streamname] = &LocalTrackPublication{p: local_video, Track: videoTrack, Trackname: streamname + "-video"}
    308. 		log.Debug("[TrackPublished_to_ION]", "published video track -> ", streamname)
    309.  
    310. 		if t.IONSfuTrack.AudioTrack == nil {
    311. 			audioTrack, err := webrtc.NewTrackLocalStaticSample(webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeOpus}, streamname+"-audio", streamname)
    312. 			//audioTrack, err := lksdk.NewLocalSampleTrack(webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeOpus})
    313. 			if err != nil {
    314. 				panic(err)
    315. 			}
    316. 			// var local_audio *lksdk.LocalTrackPublication
    317.  
    318. 			if _, err = t.IONRtc.Publish(audioTrack); err != nil {
    319. 				log.Debug("Error publishing audio track->", err)
    320. 				return err
    321. 			}
    322. 			t.IONSfuTrack.AudioTrack = audioTrack
    323. 			log.Debug("[TrackPublished_to_ION]", "published audio track -> ", streamname)
    324. 		}
    325.  
    326. 		r.Localtracks[streamname] = t
    327.  
    328. 	} else {
    329. 		log.Debug(streamname, "is exit publish")
    330. 	}
    331. 	return nil
    332. }
    333. func (r *Room) RTPTrackPublished_to_ION(trackRemote []*webrtc.TrackRemote, streamname string) error {
    334. 	// - `in` implements io.ReadCloser, such as buffer or file
    335. 	// - `mime` has to be one of webrtc.MimeType...
    336. 	// videoTrack, err := lksdk.NewLocalSampleTrack(webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeH264})
    337. 	// r.mux.lock()
    338. 	r.ionlock.Lock()
    339. 	defer r.ionlock.Unlock()
    340. 	if r.Ctx != nil && r.IONRoom == nil {
    341. 		var err error
    342. 		sn, _ := identity.GetSN()
    343. 		r.IONRoom, err = r.CreateIonRoom(r.HostIon, sn)
    344. 		if err != nil {
    345. 			log.Debug("room->", sn, "create room ok", r)
    346. 			return err
    347. 		}
    348. 	}
    349. 	t := r.Localtracks[streamname]
    350. 	if t == nil {
    351. 		t = &LocalTrackPublication{Streamname: streamname}
    352. 		t.INORoomRTCJoin(r, streamname, r.Identity)
    353. 		log.Debug("track->", streamname, "<-is nil ,Connect room", t, r)
    354. 	} else {
    355. 		if !t.IONRtc.Connected() {
    356. 			t.INORoomRTCJoin(r, streamname, r.Identity)
    357. 			log.Debug("ion track->", streamname, "<-is nil ,re Connect room", t, r)
    358. 		}
    359. 	}
    360. 	for _, v := range trackRemote {
    361. 		if v.Kind().String() == "video" {
    362. 			if t.IONSfuTrack.VideoRTPTrack == nil {
    363. 				if strings.Contains(v.Codec().MimeType, "video") {
    364. 					videoRTPTrack, err := webrtc.NewTrackLocalStaticRTP(v.Codec().RTPCodecCapability, streamname+"-video", streamname)
    365. 					if err != nil {
    366. 						panic(err)
    367. 					}
    368. 					var pub []*webrtc.RTPSender
    369. 					log.Debug("ion track publish", videoRTPTrack)
    370. 					if pub, err = t.IONRtc.Publish(videoRTPTrack); err != nil {
    371. 						log.Debug("Error publishing video RTP track->", err)
    372. 						return err
    373. 					}
    374. 					t.IONVideopub = pub[0]
    375. 					t.IONSfuTrack.VideoRTPTrack = videoRTPTrack
    376. 					r.Localtracks[streamname] = t
    377. 					log.Debug("[RTPTrackPublished_to_ION]", "published video track -> ", streamname)
    378. 				}
    379. 			}
    380. 		} else {
    381. 			if v.Kind().String() == "audio" {
    382. 				if t.IONSfuTrack.AudioRTPTrack == nil {
    383. 					if strings.Contains(v.Codec().MimeType, "audio") {
    384. 						audioRTPTrack, err := webrtc.NewTrackLocalStaticRTP(v.Codec().RTPCodecCapability, streamname+"-audio", streamname)
    385. 						if err != nil {
    386. 							panic(err)
    387. 						}
    388. 						// var local_audio *lksdk.LocalTrackPublication
    389. 						var pub []*webrtc.RTPSender
    390. 						log.Debug("ion track publish", audioRTPTrack)
    391. 						if pub, err = t.IONRtc.Publish(audioRTPTrack); err != nil {
    392. 							log.Debug("Error publishing audio track", err)
    393. 							return err
    394. 						}
    395. 						t.IONAudiopub = pub[0]
    396. 						t.IONSfuTrack.AudioRTPTrack = audioRTPTrack
    397. 						r.Localtracks[streamname] = t
    398. 						log.Debug("[RTPTrackPublished_to_ION]", "published audio track -> ", streamname)
    399. 					}
    400. 				}
    401. 			}
    402. 		}
    403. 	}
    404. 	// r.Localtracks[streamname] = t
    405.  
    406. 	// } else {
    407. 	// 	log.Debug(streamname, "is exit publish")
    408. 	// }
    409. 	return nil
    410. }
    411. func (r *Room) TrackSendIonRtpPackets(trackname, kind string, data []byte) (n int, err error) {
    412. 	if trackname == "" {
    413. 		log.Debug("Track name is null")
    414. 		return 0, fmt.Errorf("input trackname is null")
    415. 	}
    416. 	// var t *webrtc.TrackLocalStaticSample
    417. 	var t *webrtc.TrackLocalStaticRTP
    418. 	track := r.Localtracks[trackname]
    419. 	if track == nil {
    420. 		log.Debug("TrackSendIonRtpPackets:", "Track is nil ->", trackname, "<- no to publish")
    421. 		return 0, fmt.Errorf(" track is null,no to publish")
    422. 	}
    423. 	if kind == "video" {
    424. 		t = track.IONSfuTrack.VideoRTPTrack
    425. 	} else if kind == "audio" {
    426. 		t = track.IONSfuTrack.AudioRTPTrack
    427. 	}
    428. 	if t == nil {
    429. 		log.Debug("TrackSendIonRtpPackets:", "t is nil ->", trackname, "<- no to publish")
    430. 		return 0, fmt.Errorf(" track is null,no to publish")
    431. 	}
    432. 	var sb *samplebuilder.SampleBuilder
    433. 	packets := &rtp.Packet{}
    434. 	if err := packets.Unmarshal(data); err != nil {
    435. 		return 0, err
    436. 	}
    437. 	sb.Push(packets)
    438. 	for _, p := range sb.PopPackets() {
    439. 		err = t.WriteRTP(p)
    440. 		if err != nil {
    441. 			log.Debug("[TrackSendIonRtpPackets] error", err)
    442. 			return 0, err
    443. 		}
    444. 	}
    445. 	//n, err = t.Write(data)
    446. 	return len(data), nil
    447.  
    448. }
    449. func (r *Room) TrackSendIonData(trackname, kind string, data []byte, duration time.Duration) error {
    450. 	if trackname == "" {
    451. 		log.Debug("Track name is null")
    452. 		return fmt.Errorf("input trackname is null")
    453. 	}
    454. 	var t *webrtc.TrackLocalStaticSample
    455. 	track := r.Localtracks[trackname]
    456. 	if track == nil {
    457. 		log.Debug("Track is nil ->", trackname, "<- no to publish")
    458. 		return fmt.Errorf(" track is null,no to publish")
    459. 	}
    460. 	if kind == "video" {
    461. 		t = track.IONSfuTrack.VideoTrack
    462. 	} else if kind == "audio" {
    463. 		t = track.IONSfuTrack.AudioTrack
    464. 	}
    465. 	if t == nil {
    466. 		log.Debug("Track is nil ->", trackname, "<- no to publish")
    467. 		return fmt.Errorf(" track is null,no to publish")
    468. 	}
    469.  
    470. 	if videoErr := t.WriteSample(media.Sample{
    471. 		Data:     data,
    472. 		Duration: duration,
    473. 	}); videoErr != nil {
    474. 		log.Debug("WriteSample err", videoErr)
    475. 		// r.ConnectRoom()
    476. 		return nil //fmt.Errorf("WriteSample err %s", vedioErr)
    477. 	}
    478.  
    479. 	return nil
    480. }
    481. func (r *Room) TrackCloseION(streamname string) error {
    482. 	if t := r.Localtracks[streamname]; t != nil {
    483. 		var pub []*webrtc.RTPSender
    484. 		if r.Localtracks[streamname].IONVideopub != nil {
    485. 			pub = append(pub, r.Localtracks[streamname].IONVideopub)
    486. 		}
    487. 		if r.Localtracks[streamname].IONAudiopub != nil {
    488. 			pub = append(pub, r.Localtracks[streamname].IONAudiopub)
    489. 		}
    490. 		if pub != nil {
    491. 			t.IONRtc.UnPublish(pub...)
    492. 			t.LiveKitRoomConnect.Disconnect()
    493. 		}
    494. 		r.Localtracks[streamname] = nil
    495. 		log.Debug("track ", streamname, "lost ,now removed", r)
    496. 		//r.RoomConnect.LocalParticipant.UnpublishTrack(r.RoomConnect.LocalParticipant.SID())
    497. 		// r.Localtracks[streamname+"-video"]
    498. 	}
    499.  
    500. 	return nil
    501. }

     

     

  • 相关阅读:
    qrcode 生成二维码（jQuery、vue）
    operator=中自我赋值和异常安全问题
    【Git】IDEA 集成 GitHub
    linux rsyslog日志采集格式设定四
    accelerate 分布式技巧（一）
    D. Number into Sequence【1300】
    设计模式-行为型-状态模式
    【数据结构】单链表定义的介绍及增删查改的实现
    Seata 与三大平台携手编程之夏，百万奖金等你来拿
    TypeScript（4）接口
  • 原文地址：https://blog.csdn.net/superxxd/article/details/126297780