Update main.go

Fix / WebRTC sample timing

README.md

@@ -413,7 +413,7 @@ Read more about this [at the FAQ](#faq) below.
 
 ## Contributors
 
-This project exists thanks to all the people who contribute.
+This project exists thanks to all the people who contribute. Bravo!
 
 <a href="https://github.com/kerberos-io/agent/graphs/contributors">
   <img src="https://contrib.rocks/image?repo=kerberos-io/agent" />

machinery/src/cloud/Cloud.go

@@ -824,7 +824,6 @@ func HandleRealtimeProcessing(processingCursor *packets.QueueCursor, configurati
 					}
 					payload, err := models.PackageMQTTMessage(configuration, message)
 					if err == nil {
-
 						mqttClient.Publish(realtimeProcessingTopic, 0, false, payload)
 					} else {
 						log.Log.Info("cloud.RealtimeProcessing(): something went wrong while sending acknowledge config to hub: " + string(payload))

machinery/src/computervision/main.go

@@ -150,7 +150,7 @@ func ProcessMotion(motionCursor *packets.QueueCursor, configuration *models.Conf
 										}
 										payload, err := models.PackageMQTTMessage(configuration, message)
 										if err == nil {
-											mqttClient.Publish("kerberos/hub/"+hubKey, 0, false, payload)
+											mqttClient.Publish("kerberos/hub/"+hubKey, 2, false, payload)
 										} else {
 											log.Log.Info("computervision.main.ProcessMotion(): failed to package MQTT message: " + err.Error())
 										}

machinery/src/routers/mqtt/main.go

@@ -341,7 +341,7 @@ func HandleGetPTZPosition(mqttClient mqtt.Client, hubKey string, payload models.
 			}
 			payload, err := models.PackageMQTTMessage(configuration, message)
 			if err == nil {
-				mqttClient.Publish("kerberos/hub/"+hubKey, 0, false, payload)
+				mqttClient.Publish("kerberos/hub/"+hubKey, 2, false, payload)
 			} else {
 				log.Log.Info("routers.mqtt.main.HandlePTZPosition(): something went wrong while sending position to hub: " + string(payload))
 			}
@@ -412,7 +412,7 @@ func HandleRequestConfig(mqttClient mqtt.Client, hubKey string, payload models.P
 			}
 			payload, err := models.PackageMQTTMessage(configuration, message)
 			if err == nil {
-				mqttClient.Publish("kerberos/hub/"+hubKey, 0, false, payload)
+				mqttClient.Publish("kerberos/hub/"+hubKey, 2, false, payload)
 			} else {
 				log.Log.Info("routers.mqtt.main.HandleRequestConfig(): something went wrong while sending config to hub: " + string(payload))
 			}
@@ -451,7 +451,7 @@ func HandleUpdateConfig(mqttClient mqtt.Client, hubKey string, payload models.Pa
 			}
 			payload, err := models.PackageMQTTMessage(configuration, message)
 			if err == nil {
-				mqttClient.Publish("kerberos/hub/"+hubKey, 0, false, payload)
+				mqttClient.Publish("kerberos/hub/"+hubKey, 2, false, payload)
 			} else {
 				log.Log.Info("routers.mqtt.main.HandleUpdateConfig(): something went wrong while sending acknowledge config to hub: " + string(payload))
 			}

machinery/src/utils/main.go

@@ -23,7 +23,7 @@ import (
 	"github.com/kerberos-io/agent/machinery/src/models"
 )
 
-const VERSION = "3.2.6"
+const VERSION = "3.3.4"
 
 const letterBytes = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ"
 

machinery/src/webrtc/main.go

@@ -341,8 +341,8 @@ func WriteToTrack(livestreamCursor *packets.QueueCursor, configuration *models.C
 
 		var cursorError error
 		var pkt packets.Packet
-		//var previousTimeVideo int64
-		//var previousTimeAudio int64
+		var previousTimeVideo int64
+		var previousTimeAudio int64
 
 		start := false
 		receivedKeyFrame := false
@@ -401,16 +401,17 @@ func WriteToTrack(livestreamCursor *packets.QueueCursor, configuration *models.C
 			if pkt.IsVideo {
 
 				// Calculate the difference
-				//bufferDuration := pkt.Time - previousTimeVideo
-				//previousTimeVideo = pkt.Time
+				bufferDuration := int64(pkt.TimeLegacy) - previousTimeVideo
+				previousTimeVideo = pkt.Time
 
 				// Start at the first keyframe
 				if pkt.IsKeyFrame {
 					start = true
 				}
 				if start {
-					//bufferDurationCasted := time.Duration(bufferDuration) * time.Millisecond
-					sample := pionMedia.Sample{Data: pkt.Data, PacketTimestamp: uint32(pkt.Time)}
+					bufferDurationCasted := time.Duration(bufferDuration) * time.Millisecond
+					sample := pionMedia.Sample{Data: pkt.Data, Duration: bufferDurationCasted}
+					//sample = pionMedia.Sample{Data: pkt.Data, Duration: time.Second}
 					if config.Capture.ForwardWebRTC == "true" {
 						// We will send the video to a remote peer
 						// TODO..
@@ -433,12 +434,13 @@ func WriteToTrack(livestreamCursor *packets.QueueCursor, configuration *models.C
 				}
 
 				// Calculate the difference
-				//bufferDuration := pkt.Time - previousTimeAudio
-				//previousTimeAudio = pkt.Time
+				bufferDuration := int64(pkt.TimeLegacy) - previousTimeAudio
+				previousTimeAudio = pkt.Time
 
 				// We will send the audio
-				//bufferDurationCasted := time.Duration(bufferDuration) * time.Millisecond
-				sample := pionMedia.Sample{Data: pkt.Data, PacketTimestamp: uint32(pkt.Time)}
+				bufferDurationCasted := time.Duration(bufferDuration) * time.Millisecond
+				sample := pionMedia.Sample{Data: pkt.Data, Duration: bufferDurationCasted}
+				//sample = pionMedia.Sample{Data: pkt.Data, Duration: time.Second}
 				if err := audioTrack.WriteSample(sample); err != nil && err != io.ErrClosedPipe {
 					log.Log.Error("webrtc.main.WriteToTrack(): something went wrong while writing sample: " + err.Error())
 				}