diff options
Diffstat (limited to 'main/src/plugins/rtp/plugin.vala')
| -rw-r--r-- | main/src/plugins/rtp/plugin.vala | 504 |
1 files changed, 504 insertions, 0 deletions
diff --git a/main/src/plugins/rtp/plugin.vala b/main/src/plugins/rtp/plugin.vala new file mode 100644 index 00000000..d09b7edf --- /dev/null +++ b/main/src/plugins/rtp/plugin.vala @@ -0,0 +1,504 @@ +using Gee; +using Xmpp; +using Xmpp.Xep; + +public class Dino.Plugins.Rtp.Plugin : RootInterface, VideoCallPlugin, Object { + public Dino.Application app { get; private set; } + public CodecUtil codec_util { get; private set; } + public Gst.DeviceMonitor? device_monitor { get; private set; } + public Gst.Pipeline? pipe { get; private set; } + public Gst.Bin? rtpbin { get; private set; } + public Gst.Element? echoprobe { get; private set; } + + private Gee.List<Stream> streams = new ArrayList<Stream>(); + private Gee.List<Device> devices = new ArrayList<Device>(); + private int64 last_devices_refresh = 0; + // private Gee.List<Participant> participants = new ArrayList<Participant>(); + + public void registered(Dino.Application app) { + this.app = app; + this.codec_util = new CodecUtil(); + app.add_option_group(Gst.init_get_option_group()); + app.stream_interactor.module_manager.initialize_account_modules.connect((account, list) => { + list.add(new Module(this)); + }); + app.plugin_registry.video_call_plugin = this; + } + + private int pause_count = 0; + public void pause() { +// if (pause_count == 0) { +// debug("Pausing pipe for modifications"); +// pipe.set_state(Gst.State.PAUSED); +// } + pause_count++; + } + public void unpause() { + pause_count--; + if (pause_count == 0) { + debug("Continue pipe after modifications"); + pipe.set_state(Gst.State.PLAYING); + } + if (pause_count < 0) warning("Pause count below zero!"); + } + + private void handle_existing_devices(Gst.DeviceMonitor device_monitor) { + var new_devices = new ArrayList<Device>(); + foreach (Gst.Device device in device_monitor.get_devices()) { + if (device.properties == null) continue; + if (device.properties.has_name("pipewire-proplist") && device.has_classes("Audio")) continue; + if (device.properties.get_string("device.class") == "monitor") continue; + var pre_device = devices.first_match((it) => it.matches(device)); + if (pre_device != null) { + if (!new_devices.contains(pre_device)) new_devices.add(pre_device); + continue; + } + var new_device = new Device(this, device); + devices.add(new_device); + new_devices.add(new_device); + } + devices.retain_all(new_devices); + } + + private void refresh_devices() { + if (device_monitor != null) return; // No manual refresh needed, we're monitoring actively. + if (last_devices_refresh > 0 && last_devices_refresh + 5000000 > get_monotonic_time()) return; // Already current + var device_monitor = new Gst.DeviceMonitor(); + device_monitor.show_all = true; + device_monitor.start(); + handle_existing_devices(device_monitor); + device_monitor.stop(); + last_devices_refresh = get_monotonic_time(); + } + + private void start_device_monitor() { + if (device_monitor != null) return; + device_monitor = new Gst.DeviceMonitor(); + device_monitor.show_all = true; + device_monitor.get_bus().add_watch(Priority.DEFAULT, on_device_monitor_message); + device_monitor.start(); + handle_existing_devices(device_monitor); + } + + private void stop_device_monitor() { + if (device_monitor == null) return; + device_monitor.stop(); + device_monitor = null; + } + + private void init_call_pipe() { + if (pipe != null) return; + debug("Creating call pipe."); + start_device_monitor(); + pipe = new Gst.Pipeline(null); + + // RTP + rtpbin = Gst.ElementFactory.make("rtpbin", null) as Gst.Bin; + if (rtpbin == null) { + warning("RTP not supported"); + pipe = null; + return; + } + rtpbin.pad_added.connect(on_rtp_pad_added); + rtpbin.@set("latency", 100); + rtpbin.@set("do-lost", true); +// rtpbin.@set("do-sync-event", true); + rtpbin.@set("drop-on-latency", true); + rtpbin.connect("signal::request-pt-map", request_pt_map, this); + pipe.add(rtpbin); + +#if WITH_VOICE_PROCESSOR + // Audio echo probe + echoprobe = new EchoProbe(); + if (echoprobe != null) pipe.add(echoprobe); +#endif + + // Pipeline + pipe.auto_flush_bus = true; + pipe.bus.add_watch(GLib.Priority.DEFAULT, (_, message) => { + on_pipe_bus_message(message); + return true; + }); + pipe.set_state(Gst.State.PLAYING); + } + + public void destroy_call_pipe_if_unused() { + if (streams.is_empty && !VideoWidget.has_instances()) { + destroy_call_pipe(); + } + } + + private void destroy_call_pipe() { + if (pipe == null) return; + pipe.set_state(Gst.State.NULL); + rtpbin = null; +#if WITH_VOICE_PROCESSOR + echoprobe = null; +#endif + pipe = null; + stop_device_monitor(); + debug("Call pipe destroyed"); + } + + private static Gst.Caps? request_pt_map(Gst.Element rtpbin, uint session, uint pt, Plugin plugin) { + debug("request-pt-map"); + return null; + } + + private void on_rtp_pad_added(Gst.Pad pad) { + debug("pad added: %s", pad.name); + if (pad.name.has_prefix("recv_rtp_src_")) { + string[] split = pad.name.split("_"); + uint8 rtpid = (uint8)int.parse(split[3]); + foreach (Stream stream in streams) { + if (stream.rtpid == rtpid) { + stream.on_ssrc_pad_added((uint32) split[4].to_uint64(), pad); + } + } + } + if (pad.name.has_prefix("send_rtp_src_")) { + string[] split = pad.name.split("_"); + uint8 rtpid = (uint8)int.parse(split[3]); + debug("pad %s for stream %hhu", pad.name, rtpid); + foreach (Stream stream in streams) { + if (stream.rtpid == rtpid) { + stream.on_send_rtp_src_added(pad); + } + } + } + } + + private void on_pipe_bus_message(Gst.Message message) { + switch (message.type) { + case Gst.MessageType.ERROR: + Error error; + string str; + message.parse_error(out error, out str); + warning("Error in pipeline: %s", error.message); + debug(str); + break; + case Gst.MessageType.WARNING: + Error error; + string str; + message.parse_warning(out error, out str); + warning("Warning in pipeline: %s", error.message); + debug(str); + break; + case Gst.MessageType.CLOCK_LOST: + debug("Clock lost. Restarting"); + pipe.set_state(Gst.State.READY); + pipe.set_state(Gst.State.PLAYING); + break; + case Gst.MessageType.STATE_CHANGED: + // Ignore + break; + case Gst.MessageType.STREAM_STATUS: + Gst.StreamStatusType status; + Gst.Element owner; + message.parse_stream_status(out status, out owner); + if (owner != null) { + debug("%s stream changed status to %s", owner.name, status.to_string()); + } + break; + case Gst.MessageType.ELEMENT: + unowned Gst.Structure struc = message.get_structure(); + if (struc != null && message.src is Gst.Element) { + debug("Message from %s in pipeline: %s", ((Gst.Element)message.src).name, struc.to_string()); + } + break; + case Gst.MessageType.NEW_CLOCK: + debug("New clock."); + break; + case Gst.MessageType.TAG: + // Ignore + break; + case Gst.MessageType.QOS: + // Ignore + break; + case Gst.MessageType.LATENCY: + if (message.src != null && message.src.name != null && message.src is Gst.Element) { + Gst.Query latency_query = new Gst.Query.latency(); + if (((Gst.Element)message.src).query(latency_query)) { + bool live; + Gst.ClockTime min_latency, max_latency; + latency_query.parse_latency(out live, out min_latency, out max_latency); + debug("Latency message from %s: live=%s, min_latency=%s, max_latency=%s", message.src.name, live.to_string(), min_latency.to_string(), max_latency.to_string()); + } + } + break; + default: + debug("Pipe bus message: %s", message.type.to_string()); + break; + } + } + + private bool on_device_monitor_message(Gst.Bus bus, Gst.Message message) { + Gst.Device? old_gst_device = null; + Gst.Device? gst_device = null; + Device? device = null; + switch (message.type) { + case Gst.MessageType.DEVICE_ADDED: + message.parse_device_added(out gst_device); + if (devices.any_match((it) => it.matches(gst_device))) return Source.CONTINUE; + device = new Device(this, gst_device); + devices.add(device); + break; + case Gst.MessageType.DEVICE_CHANGED: + message.parse_device_changed(out gst_device, out old_gst_device); + device = devices.first_match((it) => it.matches(old_gst_device)); + if (device != null) device.update(gst_device); + break; + case Gst.MessageType.DEVICE_REMOVED: + message.parse_device_removed(out gst_device); + device = devices.first_match((it) => it.matches(gst_device)); + if (device != null) devices.remove(device); + break; + default: + break; + } + if (device != null) { + devices_changed(device.media, device.is_sink); + } + return Source.CONTINUE; + } + + public uint8 next_free_id() { + uint8 rtpid = 0; + while (streams.size < 100 && streams.any_match((stream) => stream.rtpid == rtpid)) { + rtpid++; + } + return rtpid; + } + + // public Participant get_participant(Jid full_jid, bool self) { +// foreach (Participant participant in participants) { +// if (participant.full_jid.equals(full_jid)) { +// return participant; +// } +// } +// Participant participant; +// if (self) { +// participant = new SelfParticipant(pipe, full_jid); +// } else { +// participant = new Participant(pipe, full_jid); +// } +// participants.add(participant); +// return participant; +// } + + public Stream open_stream(Xmpp.Xep.Jingle.Content content) { + init_call_pipe(); + var content_params = content.content_params as Xmpp.Xep.JingleRtp.Parameters; + if (content_params == null) return null; + Stream stream; + if (content_params.media == "video") { + stream = new VideoStream(this, content); + } else { + stream = new Stream(this, content); + } + streams.add(stream); + return stream; + } + + public void close_stream(Stream stream) { + streams.remove(stream); + stream.destroy(); + destroy_call_pipe_if_unused(); + } + + public void shutdown() { + stop_device_monitor(); + destroy_call_pipe(); + Gst.deinit(); + } + + public bool supported() { + return codec_util.is_element_supported("rtpbin"); + } + + public VideoCallWidget? create_widget(WidgetType type) { + init_call_pipe(); + if (type == WidgetType.GTK4) { + return new VideoWidget(this); + } + return null; + } + + public Gee.List<MediaDevice> get_devices(string media, bool incoming) { + refresh_devices(); + Gee.List<MediaDevice> devices; + if (media == "video" && !incoming) { + devices = get_video_sources(); + } else if (media == "audio") { + devices = get_audio_devices(incoming); + } else { + devices = new ArrayList<MediaDevice>(); + devices.add_all_iterator(this.devices.filter(it => it.media == media && (incoming && it.is_sink || !incoming && it.is_source) && !it.is_monitor)); + } + devices.sort((media_left, media_right) => { + return strcmp(media_left.id, media_right.id); + }); + + return devices; + } + + public Gee.List<MediaDevice> get_audio_devices(bool incoming) { + ArrayList<MediaDevice> pulse_devices = new ArrayList<MediaDevice>(); + ArrayList<MediaDevice> other_devices = new ArrayList<MediaDevice>(); + + foreach (Device device in devices) { + if (device.media != "audio") continue; + if (incoming && !device.is_sink || !incoming && !device.is_source) continue; + + // Skip monitors + if (device.is_monitor) continue; + + if (device.protocol == DeviceProtocol.PULSEAUDIO) { + pulse_devices.add(device); + } else { + other_devices.add(device); + } + } + + // If we have any pulseaudio devices, present only those. Don't want duplicated devices from pipewire and pulseaudio. + return pulse_devices.size > 0 ? pulse_devices : other_devices; + } + + public Gee.List<MediaDevice> get_video_sources() { + ArrayList<MediaDevice> pipewire_devices = new ArrayList<MediaDevice>(); + ArrayList<MediaDevice> other_devices = new ArrayList<MediaDevice>(); + + foreach (Device device in devices) { + if (device.media != "video") continue; + if (device.is_sink) continue; + + // Skip monitors + if (device.is_monitor) continue; + + bool is_color = false; + for (int i = 0; i < device.device.caps.get_size(); i++) { + unowned Gst.Structure structure = device.device.caps.get_structure(i); + if (!structure.has_field("format")) continue; + // "format" might be an array and get_string() will then return null. We just assume arrays to be fine. + string? format = structure.get_string("format"); + if (format == null || !format.has_prefix("GRAY")) { + is_color = true; + } + } + + // Don't allow grey-scale devices + if (!is_color) continue; + + if (device.protocol == DeviceProtocol.PIPEWIRE) { + pipewire_devices.add(device); + } else { + other_devices.add(device); + } + } + + // If we have any pipewire devices, present only those. Don't want duplicated devices from pipewire and video for linux. + return pipewire_devices.size > 0 ? pipewire_devices : other_devices; + } + + private int get_max_fps(Device device) { + int fps = 0; + for (int i = 0; i < device.device.caps.get_size(); i++) { + unowned Gst.Structure structure = device.device.caps.get_structure(i); + + if (structure.has_field("framerate")) { + Value framerate = structure.get_value("framerate"); + if (framerate.type() == typeof(Gst.Fraction)) { + int num = Gst.Value.get_fraction_numerator(framerate); + int den = Gst.Value.get_fraction_denominator(framerate); + fps = int.max(fps, num / den); + } else if (framerate.type() == typeof(Gst.ValueList)) { + for(uint j = 0; j < Gst.ValueList.get_size(framerate); j++) { + Value fraction = Gst.ValueList.get_value(framerate, j); + int num = Gst.Value.get_fraction_numerator(fraction); + int den = Gst.Value.get_fraction_denominator(fraction); + fps = int.max(fps, num / den); + } + } else { + debug("Unknown type for framerate %s on device %s", framerate.type_name(), device.display_name); + } + } + } + + debug("Max framerate for device %s: %d", device.display_name, fps); + return fps; + } + + public MediaDevice? get_preferred_device(string media, bool incoming) { + Gee.List<Device> devices = new ArrayList<Device>(); + foreach (MediaDevice media_device in get_devices(media, incoming)) { + if (media_device is Device) devices.add((Device)media_device); + } + if (devices.is_empty) { + warning("No preferred device for %s %s. Media will not be processed.", incoming ? "incoming" : "outgoing", media); + return null; + } + + // Take default if present + foreach (Device device in devices) { + if (device.is_default) { + debug("Using %s for %s %s as it's default", device.display_name, incoming ? "incoming" : "outgoing", media); + return device; + } + } + + if (media == "video") { + // Pick best FPS + int max_fps = -1; + Device? max_fps_device = null; + foreach (Device device in devices) { + int fps = get_max_fps(device); + if (fps > max_fps || max_fps_device == null) { + max_fps = fps; + max_fps_device = device; + } + } + debug("Using %s for %s %s as it has max FPS (%d)", max_fps_device.display_name, incoming ? "incoming" : "outgoing", media, max_fps); + return max_fps_device; + } else { + // Pick any + Device? device = devices.first(); + debug("Using %s for %s %s as it's first pick", device.display_name, incoming ? "incoming" : "outgoing", media); + return device; + } + } + + public MediaDevice? get_device(Xmpp.Xep.JingleRtp.Stream? stream, bool incoming) { + Stream? plugin_stream = stream as Stream?; + if (plugin_stream == null) return null; + MediaDevice? device = incoming ? plugin_stream.output_device : plugin_stream.input_device; + return device ?? get_preferred_device(stream.media, incoming); + } + + public void dump_dot() { + if (pipe == null) return; + string name = @"pipe-$(pipe.clock.get_time())-$(pipe.current_state)"; + Gst.Debug.bin_to_dot_file(pipe, Gst.DebugGraphDetails.ALL, name); + print(@"Stored pipe details as $name\n"); + } + + public void set_pause(Xmpp.Xep.JingleRtp.Stream? stream, bool pause) { + Stream? plugin_stream = stream as Stream?; + if (plugin_stream == null) return; + if (pause) { + plugin_stream.pause(); + } else { + plugin_stream.unpause(); + } + } + + public void set_device(Xmpp.Xep.JingleRtp.Stream? stream, MediaDevice? device) { + Device? real_device = device as Device?; + Stream? plugin_stream = stream as Stream?; + if (real_device == null || plugin_stream == null) return; + if (real_device.is_source) { + plugin_stream.input_device = real_device; + } else if (real_device.is_sink) { + plugin_stream.output_device = real_device; + } + } +} |