feat: add audio stream and fix only current parcel id can plays audio (…

…#61)

* feat: add audio stream and fix only current parcel id can plays audio

* fix format

* fix audio stream and video player, set the same logic

* format

* add local videos

* format

* apply review

godot/src/decentraland_components/audio_streaming.gd → godot/src/decentraland_components/audio_stream.gd

@@ -1,4 +1,4 @@
-extends AudioStreamPlayer
+extends DclAudioStream
 
 
 func stream_buffer(data: PackedVector2Array):
@@ -8,9 +8,9 @@ func stream_buffer(data: PackedVector2Array):
 	self.get_stream_playback().push_buffer(data)
 
 
-func init(frame_rate, frames, length, format, bit_rate, frame_size, channels):
+func init_audio(frame_rate, frames, length, format, bit_rate, frame_size, channels):
 	print(
-		"audio_streaming debug init ",
+		"audio_stream debug init ",
 		frame_rate,
 		" - ",
 		frames,

godot/src/decentraland_components/audio_streaming.tscn → godot/src/decentraland_components/audio_stream.tscn

@@ -1,6 +1,6 @@
 [gd_scene load_steps=2 format=3 uid="uid://de7563ke1agyk"]
 
-[ext_resource type="Script" path="res://src/decentraland_components/audio_streaming.gd" id="1_oqq6w"]
+[ext_resource type="Script" path="res://src/decentraland_components/audio_stream.gd" id="1_oqq6w"]
 
-[node name="audio_streaming" type="AudioStreamPlayer"]
+[node name="audio_stream" type="DclAudioStream"]
 script = ExtResource("1_oqq6w")

godot/src/decentraland_components/avatar.gd

@@ -173,8 +173,11 @@ func try_to_set_body_shape(body_shape_hash):
 	if skeleton == null:
 		return
 
+	var animation_player_parent = animation_player.get_parent()
+	if animation_player_parent != null:
+		animation_player_parent.remove_child(animation_player)
+
 	if body_shape_root != null:
-		body_shape_root.remove_child(animation_player)
 		remove_child(body_shape_root)
 		_free_old_skeleton.call_deferred(body_shape_root)
 

godot/src/decentraland_components/video_player.gd

@@ -0,0 +1,60 @@
+extends DclVideoPlayer
+
+var file_hash: String = ""
+
+
+func stream_buffer(data: PackedVector2Array):
+	if not self.playing:
+		self.play()
+
+	self.get_stream_playback().push_buffer(data)
+
+
+func request_video(_file_hash):
+	var content_mapping = Global.scene_runner.get_scene_content_mapping(dcl_scene_id)
+	self.file_hash = _file_hash
+
+	var fetching_resource = Global.content_manager.fetch_video(file_hash, content_mapping)
+	if not fetching_resource:
+		self._on_video_loaded(self.file_hash)
+	else:
+		Global.content_manager.content_loading_finished.connect(
+			self._content_manager_resource_loaded
+		)
+
+
+func _content_manager_resource_loaded(resource_hash: String):
+	_on_video_loaded(resource_hash, true)
+
+
+func _on_video_loaded(resource_hash: String, from_signal: bool = false):
+	if resource_hash != file_hash:
+		return
+
+	if from_signal:
+		Global.content_manager.content_loading_finished.disconnect(
+			self._content_manager_resource_loaded
+		)
+
+	var local_video_path = "user://content/" + file_hash
+	var absolute_file_path = local_video_path.replace("user:/", OS.get_user_data_dir())
+	self.resolve_resource(absolute_file_path)
+
+
+func init_audio(frame_rate, frames, length, format, bit_rate, frame_size, channels):
+	print(
+		"audio_stream debug init ",
+		frame_rate,
+		" - ",
+		frames,
+		" - ",
+		length,
+		" - ",
+		format,
+		" - ",
+		bit_rate,
+		" - ",
+		frame_size,
+		" - ",
+		channels
+	)

godot/src/decentraland_components/video_player.tscn

@@ -0,0 +1,6 @@
+[gd_scene load_steps=2 format=3 uid="uid://de7563ke1agyk"]
+
+[ext_resource type="Script" path="res://src/decentraland_components/video_player.gd" id="1_oqq6w"]
+
+[node name="video_player" type="DclVideoPlayer"]
+script = ExtResource("1_oqq6w")

godot/src/logic/content_manager.gd

@@ -12,7 +12,8 @@ enum ContentType {
 	CT_WEARABLE_EMOTE = 3,
 	CT_MESHES_MATERIAL = 4,
 	CT_INSTACE_GLTF = 5,
-	CT_AUDIO = 6
+	CT_AUDIO = 6,
+	CT_VIDEO = 7
 }
 
 var loading_content: Array[Dictionary] = []
@@ -204,6 +205,29 @@ func fetch_audio(file_path: String, content_mapping: Dictionary):
 	return true
 
 
+# Public function
+# @returns true if the resource was added to queue to fetch, false if it had already been fetched
+func fetch_video(file_hash: String, content_mapping: Dictionary):
+	var content_cached = content_cache_map.get(file_hash)
+	if content_cached != null:
+		return not content_cached.get("loaded")
+
+	content_cache_map[file_hash] = {
+		"loaded": false,
+	}
+
+	pending_content.push_back(
+		{
+			"content_mapping": content_mapping,
+			"file_hash": file_hash,
+			"content_type": ContentType.CT_VIDEO,
+			"stage": 0
+		}
+	)
+
+	return true
+
+
 func _process(_dt: float) -> void:
 	_th_poll()
 
@@ -251,6 +275,10 @@ func _th_poll():
 				if not _process_instance_gltf(content):
 					_th_to_delete.push_back(content)
 
+			ContentType.CT_VIDEO:
+				if not _process_loading_video(content, _th_finished_downloads):
+					_th_to_delete.push_back(content)
+
 			_:
 				printerr("Fetching invalid content type ", _th_content_type)
 
@@ -628,6 +656,59 @@ func _process_loading_audio(
 	return true
 
 
+func _process_loading_video(
+	content: Dictionary, finished_downloads: Array[RequestResponse]
+) -> bool:
+	var content_mapping = content.get("content_mapping")
+	var file_hash: String = content.get("file_hash")
+	var base_url: String = content_mapping.get("base_url", "")
+	var local_video_path = "user://content/" + file_hash
+	var stage = content.get("stage", 0)
+
+	match stage:
+		# Stage 0 => request png file
+		0:
+			if FileAccess.file_exists(local_video_path):
+				content["stage"] = 2
+			else:
+				if file_hash.is_empty() or base_url.is_empty():
+					printerr("hash or base_url is empty")
+					return false
+
+				var absolute_file_path = local_video_path.replace("user:/", OS.get_user_data_dir())
+				content["stage"] = 1
+				content["request_id"] = http_requester.request_file(
+					0, base_url + file_hash, absolute_file_path
+				)
+
+		# Stage 1 => wait for the file
+		1:
+			for item in finished_downloads:
+				if item.id() == content["request_id"]:
+					if item.is_error():
+						printerr("video download is_error() == true!")
+						return false
+					else:
+						content["stage"] = 2
+
+		# Stage 2 => process texture
+		2:
+			var file := FileAccess.open(local_video_path, FileAccess.READ)
+			if file == null:
+				printerr("video download fails")
+				return false
+
+			content_cache_map[file_hash]["loaded"] = true
+			content_cache_map[file_hash]["stage"] = 3
+			self.emit_signal.call_deferred("content_loading_finished", file_hash)
+			return false
+		_:
+			printerr("unknown stage ", file_hash)
+			return false
+
+	return true
+
+
 func split_animations(_gltf_node: Node) -> void:
 	pass
 

rust/decentraland-godot-lib/src/av/audio_context.rs

@@ -10,7 +10,6 @@ use super::stream_processor::AVCommand;
 use super::stream_processor::FfmpegContext;
 
 pub struct AudioSink {
-    pub volume: f32,
     pub command_sender: tokio::sync::mpsc::Sender<AVCommand>,
 }
 
@@ -160,7 +159,7 @@ impl AudioContext {
         );
 
         audio_stream_player.call_deferred(
-            "init".into(),
+            "init_audio".into(),
             &[
                 frame_rate.to_variant(),
                 input_stream.frames().to_variant(),

rust/decentraland-godot-lib/src/av/video_stream.rs

@@ -17,11 +17,10 @@ pub enum AudioDecoderError {
     StreamClosed,
     Other(String),
 }
-
 pub struct VideoSink {
     pub source: String,
     pub command_sender: tokio::sync::mpsc::Sender<AVCommand>,
-    pub tex: Gd<ImageTexture>,
+    pub texture: Gd<ImageTexture>,
     pub size: (u32, u32),
     pub current_time: f64,
     pub length: Option<f64>,
@@ -30,21 +29,20 @@ pub struct VideoSink {
 
 pub fn av_sinks(
     source: String,
-    tex: Gd<ImageTexture>,
+    texture: Option<Gd<ImageTexture>>,
     audio_stream_player: Gd<AudioStreamPlayer>,
-    volume: f32,
     playing: bool,
     repeat: bool,
-) -> (VideoSink, AudioSink) {
+    wait_for_resource: Option<tokio::sync::oneshot::Receiver<String>>,
+) -> (Option<VideoSink>, AudioSink) {
     let (command_sender, command_receiver) = tokio::sync::mpsc::channel(10);
 
     spawn_av_thread(
         command_receiver,
-        // video_sender,
-        // audio_sender,
         source.clone(),
-        tex.clone(),
+        texture.clone(),
         audio_stream_player,
+        wait_for_resource,
     );
 
     if playing {
@@ -55,29 +53,27 @@ pub fn av_sinks(
         .unwrap();
 
     (
-        VideoSink {
+        texture.map(|texture| VideoSink {
             source,
             command_sender: command_sender.clone(),
             size: (0, 0),
-            tex,
+            texture,
             current_time: 0.0,
             length: None,
             rate: None,
-        },
-        AudioSink {
-            volume,
-            command_sender,
-        },
+        }),
+        AudioSink { command_sender },
     )
 }
 
 pub fn spawn_av_thread(
     commands: tokio::sync::mpsc::Receiver<AVCommand>,
     path: String,
-    tex: Gd<ImageTexture>,
+    tex: Option<Gd<ImageTexture>>,
     audio_stream_player: Gd<AudioStreamPlayer>,
+    wait_for_resource: Option<tokio::sync::oneshot::Receiver<String>>,
 ) {
-    let video_instance_id = tex.instance_id();
+    let video_instance_id = tex.map(|value| value.instance_id());
     let audio_stream_player_instance_id = audio_stream_player.instance_id();
     std::thread::Builder::new()
         .name("av thread".to_string())
@@ -87,22 +83,23 @@ pub fn spawn_av_thread(
                 path,
                 video_instance_id,
                 audio_stream_player_instance_id,
+                wait_for_resource,
             )
         })
         .unwrap();
 }
 
 fn av_thread(
     commands: tokio::sync::mpsc::Receiver<AVCommand>,
-    // frames: tokio::sync::mpsc::Sender<VideoData>,
-    // audio: tokio::sync::mpsc::Sender<StreamingSoundData<AudioDecoderError>>,
     path: String,
-    tex: InstanceId,
+    tex: Option<InstanceId>,
     audio_stream: InstanceId,
+    wait_for_resource: Option<tokio::sync::oneshot::Receiver<String>>,
 ) {
-    let tex = Gd::from_instance_id(tex);
+    let tex = tex.map(Gd::from_instance_id);
     let audio_stream_player: Gd<AudioStreamPlayer> = Gd::from_instance_id(audio_stream);
-    if let Err(error) = av_thread_inner(commands, path, tex, audio_stream_player) {
+    if let Err(error) = av_thread_inner(commands, path, tex, audio_stream_player, wait_for_resource)
+    {
         warn!("av error: {error}");
     } else {
         debug!("av closed");
@@ -111,22 +108,39 @@ fn av_thread(
 
 pub fn av_thread_inner(
     commands: tokio::sync::mpsc::Receiver<AVCommand>,
-    path: String,
-    tex: Gd<ImageTexture>,
+    mut path: String,
+    texture: Option<Gd<ImageTexture>>,
     audio_stream_player: Gd<AudioStreamPlayer>,
+    wait_for_resource: Option<tokio::sync::oneshot::Receiver<String>>,
 ) -> Result<(), String> {
+    if let Some(wait_for_resource_receiver) = wait_for_resource {
+        match wait_for_resource_receiver.blocking_recv() {
+            Ok(file_source) => {
+                if file_source.is_empty() {
+                    return Err(format!("failed to get resource: {:?}", path));
+                }
+                path = file_source;
+            }
+            Err(err) => return Err(format!("failed to get resource: {:?}", err)),
+        }
+    }
+
     let input_context = input(&path).map_err(|e| format!("{:?} on line {}", e, line!()))?;
 
     // try and get a video context
     let video_context: Option<VideoContext> = {
-        match VideoContext::init(&input_context, tex) {
-            Ok(vc) => Some(vc),
-            Err(VideoError::BadPixelFormat) => {
-                return Err("bad pixel format".to_string());
+        if let Some(texture) = texture {
+            match VideoContext::init(&input_context, texture) {
+                Ok(vc) => Some(vc),
+                Err(VideoError::BadPixelFormat) => {
+                    return Err("bad pixel format".to_string());
+                }
+                Err(VideoError::NoStream) => None,
+                Err(VideoError::Failed(ffmpeg_err)) => return Err(ffmpeg_err.to_string()),
+                Err(VideoError::ChannelClosed) => return Ok(()),
             }
-            Err(VideoError::NoStream) => None,
-            Err(VideoError::Failed(ffmpeg_err)) => return Err(ffmpeg_err.to_string()),
-            Err(VideoError::ChannelClosed) => return Ok(()),
+        } else {
+            None
         }
     };