Merge pull request #25 from mbattista/dev

removes small lags from video live streamings

server/internal/gst/gst.go

@@ -90,7 +90,7 @@ func CreateAppPipeline(codecName string, pipelineDevice string, pipelineSrc stri
 		if pipelineSrc != "" {
 			pipelineStr = fmt.Sprintf(pipelineSrc+pipelineStr, pipelineDevice)
 		} else {
-			pipelineStr = fmt.Sprintf(videoSrc+"vp8enc target-bitrate=%d cpu-used=-5 threads=4 deadline=1 error-resilient=partitions keyframe-max-dist=30 auto-alt-ref=true"+pipelineStr, pipelineDevice, fps, bitrate*1000)
+			pipelineStr = fmt.Sprintf(videoSrc+"vp8enc target-bitrate=%d cpu-used=4 end-usage=cbr threads=4 deadline=1 undershoot=95 buffer-size=%d buffer-initial-size=%d buffer-optimal-size=%d keyframe-max-dist=180 min-quantizer=3 max-quantizer=40"+pipelineStr, pipelineDevice, fps, bitrate*1000, bitrate*6, bitrate*4, bitrate*5)
 		}
 	case "VP9":
 		// https://gstreamer.freedesktop.org/documentation/vpx/vp9enc.html?gi-language=c
@@ -131,7 +131,11 @@ func CreateAppPipeline(codecName string, pipelineDevice string, pipelineSrc stri
 			return nil, err
 		}
 
-		pipelineStr = fmt.Sprintf(videoSrc+"video/x-raw,format=I420 ! x264enc threads=4 bitrate=%d byte-stream=true tune=zerolatency speed-preset=veryfast ! video/x-h264,stream-format=byte-stream"+pipelineStr, pipelineDevice, fps, bitrate)
+		vbvbuf := uint(1000)
+		if bitrate > 1000 {
+			vbvbuf = bitrate
+		}
+		pipelineStr = fmt.Sprintf(videoSrc+"video/x-raw,format=NV12 ! x264enc threads=4 bitrate=%d key-int-max=60 vbv-buf-capacity=%d byte-stream=true tune=zerolatency speed-preset=veryfast ! video/x-h264,stream-format=byte-stream"+pipelineStr, pipelineDevice, fps, bitrate, vbvbuf)
 	case "Opus":
 		// https://gstreamer.freedesktop.org/documentation/opus/opusenc.html
 		// gstreamer1.0-plugins-base

server/internal/webrtc/peer.go

@@ -28,7 +28,7 @@ func (peer *Peer) WriteData(v interface{}) error {
 }
 
 func (peer *Peer) Destroy() error {
-	if peer.connection != nil && peer.connection.ConnectionState() == webrtc.PeerConnectionStateConnected {
+	if peer.connection != nil && peer.connection.ConnectionState() != webrtc.PeerConnectionStateClosed {
 		if err := peer.connection.Close(); err != nil {
 			return err
 		}

server/internal/webrtc/webrtc.go

@@ -146,11 +146,13 @@ func (manager *WebRTCManager) CreatePeer(id string, session types.Session) (stri
 			Msg("connection state has changed")
 	})
 
-	if _, err = connection.AddTrack(manager.videoTrack); err != nil {
+	rtpVideo, err := connection.AddTrack(manager.videoTrack);
+	if err != nil {
 		return "", manager.config.ICELite, manager.config.ICEServers, err
 	}
 
-	if _, err = connection.AddTrack(manager.audioTrack); err != nil {
+	rtpAudio, err := connection.AddTrack(manager.audioTrack);
+	if err != nil {
 		return "", manager.config.ICELite, manager.config.ICEServers, err
 	}
 
@@ -167,17 +169,20 @@ func (manager *WebRTCManager) CreatePeer(id string, session types.Session) (stri
 	connection.OnConnectionStateChange(func(state webrtc.PeerConnectionState) {
 		switch state {
 		case webrtc.PeerConnectionStateDisconnected:
-		case webrtc.PeerConnectionStateFailed:
 			manager.logger.Info().Str("id", id).Msg("peer disconnected")
 			manager.sessions.Destroy(id)
-			break
+		case webrtc.PeerConnectionStateFailed:
+			manager.logger.Warn().Str("id", id).Msg("peer failed")
+			manager.sessions.Destroy(id)
+		case webrtc.PeerConnectionStateClosed:
+			manager.logger.Info().Str("id", id).Msg("peer closed")
+			manager.sessions.Destroy(id)
 		case webrtc.PeerConnectionStateConnected:
 			manager.logger.Info().Str("id", id).Msg("peer connected")
 			if err = session.SetConnected(true); err != nil {
 				manager.logger.Warn().Err(err).Msg("unable to set connected on peer")
 				manager.sessions.Destroy(id)
 			}
-			break
 		}
 	})
 
@@ -211,6 +216,25 @@ func (manager *WebRTCManager) CreatePeer(id string, session types.Session) (stri
 		return "", manager.config.ICELite, manager.config.ICEServers, err
 	}
 
+	go func() {
+		rtcpBuf := make([]byte, 1500)
+		for {
+			if _, _, rtcpErr := rtpVideo.Read(rtcpBuf); rtcpErr != nil {
+				return
+			}
+		}
+	}()
+
+	go func() {
+		rtcpBuf := make([]byte, 1500)
+		for {
+			if _, _, rtcpErr := rtpAudio.Read(rtcpBuf); rtcpErr != nil {
+				return
+			}
+		}
+	}()
+
+
 	return description.SDP, manager.config.ICELite, manager.config.ICEServers, nil
 }