RemoteCraft 05 - WebRTC Pion!

2021.07.23 - [projects.] - RemoteCraft 04 - WebRTC 소개

RemoteCraft 04 - WebRTC 소개

 

WebRTC 클라이언트 개발

Go 커뮤니티에는 정말 없는 것 빼고 다 있다. 다행히 내가 필요한 것은 모두 있어서 고생을 많이 덜었다.. 난 Server-to-Peer 연결을 하는 것인데 이걸 써도 되는 건가..? 싶었지만 곰곰이 생각해 보니 결국 Server가 또 다른 클라이언트가 되면 Peer-to-Peer랑 다른 것이 없지 않은가! 하고 바로 examples를 실험해보고 만지작거렸다. 지금은 잘 작동하는 이쁜 Pion WebRTC Client! 그 개발 이야기를 써보려 한다.

Pion Go!

Pion은 Go언어로 쓰여진 WebRTC API 오픈소스 프로젝트다. 현시점에도 계속 활발하게 이슈가 올라오고 수정되고 있으며... 슬랙 채널에서도 많은 도움을 받았다. RemoteCraft 개발을 시작할 때만 해도 v2 였는데... v3이 나왔다. 릴리즈 노트를 확인하는데 breaking change가 보인다.. 부서질까 봐 두려워 아직 포팅 작업은 하지 못했기에.. 아쉽지만 v2를 기준으로 작성한다!

 

WebRTC 클라이언트 설계

RemoteCraft의 WebRTC 설계 

개발을 중간에 계속 뒤엎는 행위를 하다보니... 전체 설계도 전에 WebRTC의 설계를 먼저 보고 넘어가자!

 

RemoteCraft의 WebRTC설계는 크게 3 덩이로 나누어져 있다.

1. 브라우저 : 말 그대로 사용자가 서버로부터 게임영상을 스트리밍 받고 키보드 및 컨트롤 인풋을 서버로 보내게 될 Peer이다.
2. 로비 : Signaling 서버의 역할이다. 브라우저와 워커와 Websocket 연결을 유지하며 브라우저에 할당된 워커에게 WebRTC의 Session control messages를 relay하는 역할을 한다.
3. 브라우저와 WebRTC연결을 하는 상대 Peer가 되겠다. 쉽게 생각하여 카메라와 마이크 대신에 게임의 비디오와 오디오를 전달하는 WebRTC클라이언트가 있다고 생각하면 된다.

연결을 위한 여정을 순차적으로 적자면

1. 브라우저가 lobby에 GET /users_ws로 연결을 시도하면 websocket연결로 업그레이드한다.
2. 로비에도 유저와 소통하는 websocket클라이언트(UserClient)가 생성되고 유저의 세션을 기억하기 위해 세션 아이디를 생성한다.
3. 생성된 UserClient는 가용 가능한 WorkerClient에 매핑되며 매핑된 세션 및 워커+게임 인스턴스의 상태는 모두 로비에서 관리한다.
4. 유저가 websocket을 통해  init_webrtc 메시지를 워커에게 전달
5. 워커는 그 메시지의 응답으로 SDP Offer를 생성해 유저에게 전달한다.
6. 브라우저는 워커로부터 받은 SDP Offer를 setRemoteDescription 함수를 통해 PeerConnection 객체에 저장한다.
7. 브라우저는 자신의 SDP Answer를 생성해 워커에게 전달한다.
8. 각각 워커와 브라우저의 WebRTC의 PeerConnection객체는 인스턴스화 되면서 구글의 STUN서버를 통해 각자의 public IP 및 port를 websocket 메시지를  통해 교환한다. (이 부분 때문에 HTTP request만으로 구현하기보단 웹소켓을 활용하는 것이 더 나을 것 같다고 판단했다.)
9. 서로를 찾을 수 있다면 연결이 완료되고 스트리밍이 시작된다!

 

중요한 정보 하나... SDP 및 ICE Candidate정보는 모두 base64로 인코딩하여 전송해야 보내진다!

 

여기서 의문 하나...

Offer를 어디서 먼저 보내야 하는가? 는 아직도 계속 질문 중이다. 미디어의 주인이 되는 워커가 돼야 하는 것이 설계적으로 합리적인 것 같기도 하고... 브라우저뿐만 아니라 다른 애플리케이션 또는 기기를 지원한다고 했을 때(예를 들어 모바일폰!) 그에 따라 달라지는 SDP를 미리 서버에 전달하고 그에 맞는 MediaStream 객체를 생성하는 것도 합리적인 것 같고.. 아리송 아리송한 가운데.. 현재는 모든 WebRTC-compatible 브라우저에서 지원하는 코덱만 인코더가 완성되어 있고.. 또 개발을 브라우저의 WebRTC 클라이언트보다 워커의 WebRTC클라이언트를 먼저 개발했기 때문에 디버깅의 편의상 자연스럽게 offer를 만드는 쪽은 워커가 돼버렸다.

 

 

워커의 websocket 메시지 핸들러 및 webRTC 초기화 코드이다.

 

api2.go

// init_webrtc 메세지에 실행되는 핸들러
func (wc *Controller) handleInitWebRTC(inMessage ws.WSPacket) ws.WSPacket {
	log.Println("[WORKER][handleInitWebRTC] is called")
	webRtcInstance := webRTC.InitWebRTC().WithConfig(
		webrtcConfig.Config{Encoder: wc.config.Encoder, Webrtc: wc.config.Webrtc.Webrtc},
	)
	localSession, err := webRtcInstance.StartClient(
    	// ice_cb 함수. ICECandidate정보를 브라우저에 보낼때 실행된다.
		func(candidate string) {
			wc.lClient.Send(ws.WSPacket{
				Title:     api.ICE_CANDIDATE,
				Data:      candidate,
				SessionID: inMessage.SessionID},nil)
		})

	if err != nil {
		log.Println("[WORKER][handleInitWebRTC] failed to create offer\n")
		return ws.EmptyPacket
	}

	session := &Session{
		rtc: webRtcInstance,
	}
	// 워커에 등록될 세션 매핑
	wc.sessions[inMessage.SessionID] = session

	// 서버의 SDP 전달
	return ws.WSPacket{Title: api.SDP_OFFER, Data: localSession}
}

// sdp_answer 메세지에 실행되는 핸들러
// 브라우저로부터 SDP를 전달 받고, SDP를 저장한다.
func (wc *Controller) handleSDPAnswer(inMessage ws.WSPacket) ws.WSPacket {
	log.Println("[WORKER][handleSDPAnswer] is called")
	session := wc.getSession(inMessage.SessionID)
	if session != nil {
		err := session.rtc.SetRemoteSDP(inMessage.Data)
		if err != nil {
			log.Printf("[WORKER][handleSDPAnswer] failed to save SDP Answer from user: %+v\n", err)
			return ws.EmptyPacket
		}
	} else {
		log.Printf("[WORKER][handleSDPAnswer]  No session for Title: %s\n", inMessage.SessionID)
	}
	return ws.EmptyPacket
}

// ice_candidate 메세지에 실행되는 핸들러
// 브라우저로부터 ICE Candidate 정보를 전달 받고, 이를 저장한다.
func (wc *Controller) handleICECandidate(inMessage ws.WSPacket) ws.WSPacket {
	log.Println("[WORKER][handleICECandidate] is called")
	session := wc.getSession(inMessage.SessionID)
	if session != nil {
		err := session.rtc.AddCandidate(inMessage.Data)
		if err != nil {
			log.Println("[WORKER][handleICECandidate] failed to add ice candidate to rtcSession")
		}
	} else {
		log.Printf("[WORKER][handleICECandidate]  No session for Title: %s\n", inMessage.SessionID)
	}
	return ws.EmptyPacket
}

 

webRTC.go 

type WebRTC struct {
	ID          	string
	config      	webrtcConfig.Config
	connection  	*pion.PeerConnection

	videoTrack		*pion.Track
	audioTrack		*pion.Track

	isConnected 	bool

	FrameChannel 	chan RtcFrame
	AudioChannel	chan []byte
	InputChannel 	chan []byte

	RoomID 			string
	PlayerIndex		int
}


func (rtc *WebRTC) WithConfig(conf webrtcConfig.Config) *WebRTC {
	rtc.config = conf
	return rtc
}

func InitWebRTC() *WebRTC {
	rtc := &WebRTC{
		ID: uuid.Must(uuid.NewV4()).String(),
		FrameChannel: make(chan RtcFrame, 42),
		AudioChannel: make(chan []byte, 42),
		InputChannel: make(chan []byte, 100),
		config: webrtcConfig.Config{},
	}
	return rtc
}

func (rtc *WebRTC) SetRemoteSDP(remoteSDP string) error {
	sdp := pion.SessionDescription{}
	log.Println("DecodeBase64 browser sdp....")
	err := DecodeBase64(remoteSDP, &sdp)
	if err != nil {
		log.Println("Decode remote sdp from peer failed")
		return err
	}
	err = rtc.connection.SetRemoteDescription(sdp)
	if err != nil {
		log.Println("Set remote description from peer failed")
		return err
	}
	return nil
}

func NewPionPeerConnection(conf webrtcConfig.Webrtc) (*pion.PeerConnection, error) {
	m := pion.MediaEngine{}
	m.RegisterDefaultCodecs()


	peerConf := pion.Configuration{ICEServers: []pion.ICEServer{}}
	for _, server := range conf.IceServers {
		peerConf.ICEServers = append(peerConf.ICEServers, pion.ICEServer{
			URLs:       []string{server.Url},
		})
	}
	api := pion.NewAPI(pion.WithMediaEngine(m))
    
    // RTCPeerConnection생성!
	return api.NewPeerConnection(peerConf)
}


// init_webrtc 메세지를 워커가 받고 실행하는 함수
func (rtc *WebRTC) StartClient(ice_cb OnICECandidateCallback) (string, error) {
	defer func() {
		if err := recover(); err != nil {
			log.Println(err)
			rtc.StopClient()
		}
	}()


	log.Println("Start webrtc client...")

	var err error
	
    // RTCPeerConnection생성 함수!
	rtc.connection, err = NewPionPeerConnection(rtc.config.Webrtc)
	if err != nil {
		return "", err
	}

	// MediaStream 추가
	err = rtc.SetUpVideoTrack()
	if err != nil {
		return "", err
	}
	log.Println("Successfully Add video Output")

	err = rtc.SetUpAudioTrack()
	if err != nil {
		return "", err
	}
	log.Println("Successfully Add audio Output")

	err = rtc.setUpDataChannel()
	if err != nil {
		return "", err
	}


	// 연결 상태에 따라 실행되는 핸들러 등록.
    // 상태에 따른 핸들러 함수를 작성하여 Pion에게 전달해주면 내부의 로직은 알아서 처리해준다.
	rtc.connection.OnICEConnectionStateChange(func(connectionState pion.ICEConnectionState) {
		log.Printf("Connection State has changed [ %s ]\n", connectionState.String())
		if connectionState == pion.ICEConnectionStateConnected {
			go func() {
				rtc.isConnected = true
				log.Printf("ConnectionState: [ %s ]\n", connectionState.String())
				rtc.BeginStreaming()
			}()
		}
		if connectionState == pion.ICEConnectionStateFailed ||
			connectionState == pion.ICEConnectionStateDisconnected ||
			connectionState == pion.ICEConnectionStateClosed {
			rtc.StopClient()
		}
	})

	// STUN서버로부터 ICE Candidate정보를 받으면 상황에 실행되는 핸들러 등록.
    // ice_cb는 브라우저에게 바뀐 ICECandidate정보를 websocket으로 보내는 함수이다.
	rtc.connection.OnICECandidate(func(iceCandidate *pion.ICECandidate) {
		if iceCandidate != nil {
			log.Println("OnIceCandidate:", iceCandidate.ToJSON().Candidate)
			candidate, err := EncodeBase64(iceCandidate.ToJSON())
			if err != nil {
				log.Println("Encode IceCandidate failed: " + iceCandidate.ToJSON().Candidate)
				return
			}
			ice_cb(candidate)
		} else {
			ice_cb("")
		}

	})


	// 위의 정보를 바탕으로 SDP Offer생성
	sessionOffer, err := rtc.connection.CreateOffer(nil)
	if err != nil {
		return "", err
	}
	err = rtc.connection.SetLocalDescription(sessionOffer)
	if err != nil {
		return "", err
	}
	localSession, err := EncodeBase64(sessionOffer)
	if err != nil {
		return "", err
	}
    
    //SDP Offer를 리턴 받은 handleInitWebRTC는 브라우저에 SDP를 전송한다.
	return localSession, nil
}


// 연결이 완료됐을때 OnICEConnectionStateChange에서 커넥션이 성공적으로 되었을 때
// 실행되는 스트리밍 시작 고루틴
func (rtc *WebRTC) BeginStreaming() {
	go func() {
		log.Println("Open WebRTC videoTrack channel...")
		for data := range rtc.FrameChannel {
			if err := rtc.videoTrack.WriteSample(media.Sample{Data: data.Data, Samples: 1}); err != nil {
				log.Println("[ERROR][WebRTC] videoTrack.WriteSample: ", err)
				break
			}
		}
	}()

	go func() {
		log.Println("Open WebRTC audioTrack channel...")
		samples := uint32(rtc.config.Encoder.Audio.SampleRate * rtc.config.Encoder.Audio.Frame / 1000)
		for data := range rtc.AudioChannel {
			if err := rtc.audioTrack.WriteSample(media.Sample{Data: data, Samples: samples}); err != nil {
				log.Println("[ERROR][WebRTC]: audioTrack.WriteSample: ", err)
				break
			}
		}
	}()
}

// 브라우저로부터 SDP Answer를 전달 받았을때 실행되는 함수.
// base64 디코딩 후 RTCPeerConnection 인스턴스에 등록한다.
func (rtc *WebRTC) SetRemoteSDP(remoteSDP string) error {
	sdp := pion.SessionDescription{}
	log.Println("DecodeBase64 browser sdp....")
	err := DecodeBase64(remoteSDP, &sdp)
	if err != nil {
		log.Println("Decode remote sdp from peer failed")
		return err
	}
	err = rtc.connection.SetRemoteDescription(sdp)
	if err != nil {
		log.Println("Set remote description from peer failed")
		return err
	}
	return nil
}

// 브라우저로부터 ICECandidate 변경내용이 왔을때 실행되는 함수
func (rtc *WebRTC) AddCandidate(candidate string) error {
	var iceCandidate pion.ICECandidateInit
	err := DecodeBase64(candidate, &iceCandidate)
	if err != nil {
		log.Println("Decode ICE Candidate from user failed")
		return err
	}

	err = rtc.connection.AddICECandidate(iceCandidate)
	if err != nil {
		log.Println("Add ICE Candidate from user failed")
		return err
	}
	
	log.Println("Add ICE Candidate complete: " + iceCandidate.Candidate)
	return nil
}

 

 

 

일기)

글을 쓰다가 느낀 건데.. 코드를 기능별로 잘게 쪼개 놔서 디버깅할 때는 편했는데 글로 풀어 쓰려하니 한눈에 흐름이 들어오지 않는 단점이 있다..

 

TODO.. 클라이언트에서 이렇게 보내면... 트랙 등록을 하지 않아도 sdp Offer에 코덱 정보가 들어가는 것 같다..! 바꿀 수 있겠다..

pc.createOffer({offerToReceiveVideo: true, offerToReceiveAudio: true})