Request 5813 (accepted)

pipewire-aptx.changes Changed

pipewire-aptx.spec Changed

pipewire-0.3.74.tar.gz/NEWS -> pipewire-0.3.76.tar.gz/NEWS Changed

@@ -1,3 +1,94 @@
+# PipeWire 0.3.76 (2023-07-28)
+
+This is a quick bugfix release that is API and ABI compatible with previous
+0.3.x releases.
+
+## Highlights
+  - Fix a regression that would cause the MPV pipewire backend to fail because
+    of a spurious thread-loop signal.
+  - Fix a crash when DBus is not found.
+  - ALSA hires timestamps are now disabled by default.
+  - Some more fixes and improvements.
+
+## PipeWire
+  - A new option was added to pw-thread-loop to signal when the thread starts.
+    This is only used in module-rt to avoid regressions in mpv. (#3374)
+  - Fix a compilation problem.
+  - Stream flags now only set the properties when not already set. This fixes
+    a regression with node autoconnect. (#3382)
+
+## Tools
+  - pw-cat will now stop when the stream is disconnected. (#2731)
+  - Improve the pw-cat man page, mention that stdin/stdout handling is only
+    on raw data.
+
+## modules
+  - module-rt will now not crash when dbus is not available but error out as
+    before.
+  - A new VBAN (vb-audio.com) sender and receiver was added. (#3380)
+
+## SPA
+  - Add an option in audioconvert to disable volume updates. (#3361)
+  - ALSA hires timestamps are disabled by default because many drivers seem to
+    give wrong timestamps and cause extra delay.
+
+## bluetooth
+  - LE Audio support is now enabled by default when liblc3 is available now that
+    bluez has support for detecting the hardware features.
+
+Older versions:
+
+
+# PipeWire 0.3.75 (2023-07-21)
+
+This is a bugfix release that is API and ABI compatible with previous
+0.3.x releases.
+
+## Highlights
+  - Link permissions between nodes are now enforced. This avoids potential portal
+    managed screencast nodes to link to the camera even though it was not assigned
+    permissions to do so by the session manager.
+  - Libcamera and v4l2 devices now have properties so that duplicates can be
+    filtered out by the session manager.
+  - A bug with draining was fixed where a buffer would be marked EMPTY and would not
+    play when it contained drained samples. (#3365)
+  - Many fixes and improvements.
+
+
+## PipeWire
+  - Permissions for links between nodes are now enforced. The link will now check
+    that the owner clients of the nodes can see eachother before allowing the link.
+    This avoids screensharing clients to accidentally being linked to the camera
+    nodes by the session manager. A side effect is that patchbay tools will no longer
+    be able to link portal managed screencast nodes to the camera, for this we need
+    a new permission for those patchbay clients. (wireplumber#218)
+  - The stream.rules/filter.rules are now evaluated when connecting the stream/filter
+    so that more properties can be matched. (#3355)
+  - Move some internal events from the context to the nodes to better handle per-node
+    threads in the future.
+  - The thread-loop will now signal when the thread is started.
+
+## modules
+  - A timestamp workaround in module-raop was reverted because it does not work
+    in all cases. Instead latency was increased to 1.5 seconds, which also makes
+    the problematic device in question work. (#3247)
+  - The profiler module was reworked a bit to use the new node realtime events. It
+    should now also handle dynamically added and removed drivers.
+  - The module-rt now does the rtkit calls from a separate thread so that it does
+    not block the main thread. This could cause deadlocks during startup in some
+    cases. (#3357)
+
+## SPA
+  - Atomic operation macros were move from internal pipewire API to public API.
+  - The video-info structure now has a new SPA_VIDEO_FLAG_MODIFIER_FIXATION_REQUIRED
+    flag to instruct the application to fixate the modifiers. This simplifies some
+    logic in applications a lot.
+  - The libcamera and v4l2 nodes now have properties to enumerate the device id
+    they are using. This can be used to match v4l2 devices and libcamera devices
+    and filter out duplicates.
+  - A bug with draining was fixed where a buffer would be marked EMPTY and would not
+    play when it contained drained samples. (#3365)
+
 # PipeWire 0.3.74 (2023-07-12)
 
 This is a quick bugfix release that is API and ABI compatible with previous
@@ -33,9 +124,6 @@
   - Merge scope based cleanup macros.
   - Add ratelimit function.
 
-Older versions:
-
-
 # PipeWire 0.3.73 (2023-07-06)
 
 This is a bugfix release that is API and ABI compatible with previous

pipewire-0.3.74.tar.gz/doc/pipewire-modules.dox -> pipewire-0.3.76.tar.gz/doc/pipewire-modules.dox Changed

pipewire-0.3.74.tar.gz/man/pw-cat.1.rst.in -> pipewire-0.3.76.tar.gz/man/pw-cat.1.rst.in Changed

pipewire-0.3.74.tar.gz/meson.build -> pipewire-0.3.76.tar.gz/meson.build Changed

pipewire-0.3.74.tar.gz/meson_options.txt -> pipewire-0.3.76.tar.gz/meson_options.txt Changed

pipewire-0.3.74.tar.gz/pipewire-alsa/alsa-plugins/pcm_pipewire.c -> pipewire-0.3.76.tar.gz/pipewire-alsa/alsa-plugins/pcm_pipewire.c Changed

@@ -19,20 +19,12 @@
 #include <spa/param/audio/format-utils.h>
 #include <spa/debug/types.h>
 #include <spa/param/props.h>
+#include <spa/utils/atomic.h>
 #include <spa/utils/result.h>
 #include <spa/utils/string.h>
 
 #include <pipewire/pipewire.h>
 
-#define ATOMIC_INC(s)                   __atomic_add_fetch(&(s), 1, __ATOMIC_SEQ_CST)
-#define ATOMIC_LOAD(s)                  __atomic_load_n(&(s), __ATOMIC_SEQ_CST)
-
-#define SEQ_WRITE(s)                    ATOMIC_INC(s)
-#define SEQ_WRITE_SUCCESS(s1,s2)        ((s1) + 1 == (s2) && ((s2) & 1) == 0)
-
-#define SEQ_READ(s)                     ATOMIC_LOAD(s)
-#define SEQ_READ_SUCCESS(s1,s2)         ((s1) == (s2) && ((s2) & 1) == 0)
-
 PW_LOG_TOPIC_STATIC(alsa_log_topic, "alsa.pcm");
 #define PW_LOG_TOPIC_DEFAULT alsa_log_topic
 
@@ -223,7 +215,7 @@
 	int64_t diff;
 
 	do {
-		seq1 = SEQ_READ(pw->seq);
+		seq1 = SPA_SEQ_READ(pw->seq);
 
 		delay = pw->delay + pw->transfered;
 		now = pw->now;
@@ -232,8 +224,8 @@
 		else
 			avail = snd_pcm_ioplug_avail(io, pw->hw_ptr, io->appl_ptr);
 
-		seq2 = SEQ_READ(pw->seq);
-	} while (!SEQ_READ_SUCCESS(seq1, seq2));
+		seq2 = SPA_SEQ_READ(pw->seq);
+	} while (!SPA_SEQ_READ_SUCCESS(seq1, seq2));
 
 	if (now != 0 && (io->state == SND_PCM_STATE_RUNNING ||
 	    io->state == SND_PCM_STATE_DRAINING)) {
@@ -437,7 +429,7 @@
 
 	want = b->requested ? b->requested : hw_avail;
 
-	SEQ_WRITE(pw->seq);
+	SPA_SEQ_WRITE(pw->seq);
 
 	if (pw->now != pwt.now) {
 		pw->transfered = pw->buffered;
@@ -455,7 +447,7 @@
 	pw->buffered = (want == 0 || pw->transfered < want) ?  0 : (pw->transfered % want);
 
 	pw->now = pwt.now;
-	SEQ_WRITE(pw->seq);
+	SPA_SEQ_WRITE(pw->seq);
 
 	pw_log_trace("%p: avail-before:%lu avail:%lu want:%lu xfer:%lu hw:%lu appl:%lu",
 			pw, before, hw_avail, want, xfer, pw->hw_ptr, io->appl_ptr);

pipewire-0.3.74.tar.gz/pipewire-jack/src/pipewire-jack.c -> pipewire-0.3.76.tar.gz/pipewire-jack/src/pipewire-jack.c Changed

@@ -1737,7 +1737,7 @@
 		pw_log_trace_fp("%p: link %p %p %d/%d", c, l, state,
 				state->pending, state->required);
 
-		if (pw_node_activation_state_dec(state, 1)) {
+		if (pw_node_activation_state_dec(state)) {
 			l->activation->status = PW_NODE_ACTIVATION_TRIGGERED;
 			l->activation->signal_time = nsec;
 
@@ -1898,18 +1898,18 @@
 	pw_log_debug("%p: activation %p", c, a);
 
 	/* was ok */
-	owner = ATOMIC_LOAD(a->segment_owner0);
+	owner = SPA_ATOMIC_LOAD(a->segment_owner0);
 	if (owner == c->node_id)
 		return 0;
 
 	/* try to become owner */
 	if (c->timeowner_conditional) {
-		if (!ATOMIC_CAS(a->segment_owner0, 0, c->node_id)) {
+		if (!SPA_ATOMIC_CAS(a->segment_owner0, 0, c->node_id)) {
 			pw_log_debug("%p: owner:%u id:%u", c, owner, c->node_id);
 			return -EBUSY;
 		}
 	} else {
-		ATOMIC_STORE(a->segment_owner0, c->node_id);
+		SPA_ATOMIC_STORE(a->segment_owner0, c->node_id);
 	}
 
 	pw_log_debug("%p: timebase installed for id:%u", c, c->node_id);
@@ -3536,6 +3536,7 @@
 	uint32_t n_support;
 	const char *str;
 	struct spa_cpu *cpu_iface;
+	const struct pw_properties *props;
 	va_list ap;
 
         if (getenv("PIPEWIRE_NOJACK") != NULL ||
@@ -3607,14 +3608,16 @@
 	client->notify_buffer = calloc(1, NOTIFY_BUFFER_SIZE + sizeof(struct notify));
 	spa_ringbuffer_init(&client->notify_ring);
 
-	client->allow_mlock = client->context.context->settings.mem_allow_mlock;
-	client->warn_mlock = client->context.context->settings.mem_warn_mlock;
-
 	pw_context_conf_update_props(client->context.context,
 			"jack.properties", client->props);
 
+	props = pw_context_get_properties(client->context.context);
+
+	client->allow_mlock = pw_properties_get_bool(props, "mem.allow-mlock", true);
+	client->warn_mlock = pw_properties_get_bool(props, "mem.warn-mlock", false);
+
 	pw_context_conf_section_match_rules(client->context.context, "jack.rules",
-			&client->context.context->properties->dict, execute_match, client);
+			&props->dict, execute_match, client);
 
 	support = pw_context_get_support(client->context.context, &n_support);
 
@@ -6192,7 +6195,7 @@
 	if ((a = c->driver_activation) == NULL)
 		return -EIO;
 
-	if (!ATOMIC_CAS(a->segment_owner0, c->node_id, 0))
+	if (!SPA_ATOMIC_CAS(a->segment_owner0, c->node_id, 0))
 		return -EINVAL;
 
 	c->timebase_callback = NULL;
@@ -6364,7 +6367,7 @@
 	na->reposition.duration = 0;
 	na->reposition.position = pos->frame;
 	na->reposition.rate = 1.0;
-	ATOMIC_STORE(a->reposition_owner, c->node_id);
+	SPA_ATOMIC_STORE(a->reposition_owner, c->node_id);
 
 	return 0;
 }
@@ -6374,7 +6377,7 @@
 	struct pw_node_activation *a = c->rt.driver_activation;
 	if (!a)
 		return;
-	ATOMIC_STORE(a->command, command);
+	SPA_ATOMIC_STORE(a->command, command);
 }
 
 SPA_EXPORT

pipewire-0.3.74.tar.gz/pipewire-v4l2/src/pipewire-v4l2.c -> pipewire-0.3.76.tar.gz/pipewire-v4l2/src/pipewire-v4l2.c Changed

@@ -19,6 +19,7 @@
 
 #include "pipewire-v4l2.h"
 
+#include <spa/utils/atomic.h>
 #include <spa/utils/result.h>
 #include <spa/pod/iter.h>
 #include <spa/pod/parser.h>
@@ -248,8 +249,6 @@
 		}
 	}
 }
-#define ATOMIC_DEC(s)                   __atomic_sub_fetch(&(s), 1, __ATOMIC_SEQ_CST)
-#define ATOMIC_INC(s)                   __atomic_add_fetch(&(s), 1, __ATOMIC_SEQ_CST)
 
 static struct file *make_file(void)
 {
@@ -301,7 +300,7 @@
 static void unref_file(struct file *file)
 {
 	pw_log_debug("file:%d ref:%d", file->fd, file->ref);
-	if (ATOMIC_DEC(file->ref) <= 0)
+	if (SPA_ATOMIC_DEC(file->ref) <= 0)
 		free_file(file);
 }
 
@@ -314,7 +313,7 @@
 		map->fd = fd;
 		map->flags = flags;
 		map->file = file;
-		ATOMIC_INC(file->ref);
+		SPA_ATOMIC_INC(file->ref);
 		pw_log_debug("fd:%d -> file:%d ref:%d", fd, file->fd, file->ref);
 	}
 	pthread_mutex_unlock(&globals.lock);
@@ -349,7 +348,7 @@
 	struct fd_map *map;
 	pw_array_for_each(map, &globals.fd_maps) {
 		if (map->fd == fd) {
-			ATOMIC_INC(map->file->ref);
+			SPA_ATOMIC_INC(map->file->ref);
 			pw_log_debug("fd:%d find:%d ref:%d", map->fd, fd, map->file->ref);
 			return map;
 		}
@@ -384,7 +383,7 @@
 		if (tmp->file->dev_id == dev) {
 			if (tmp->file->closed)
 				tmp->file->fd = tmp->fd;
-			ATOMIC_INC(tmp->file->ref);
+			SPA_ATOMIC_INC(tmp->file->ref);
 			map = tmp;
 			pw_log_debug("dev:%d find:%d ref:%d",
 					tmp->file->dev_id, dev, tmp->file->ref);

pipewire-0.3.74.tar.gz/spa/include/spa/graph/graph.h -> pipewire-0.3.76.tar.gz/spa/include/spa/graph/graph.h Changed

pipewire-0.3.74.tar.gz/spa/include/spa/monitor/device.h -> pipewire-0.3.76.tar.gz/spa/include/spa/monitor/device.h Changed

pipewire-0.3.74.tar.gz/spa/include/spa/param/props.h -> pipewire-0.3.76.tar.gz/spa/include/spa/param/props.h Changed

@@ -59,13 +59,16 @@
 	SPA_PROP_START_Audio	= 0x10000,	/**< audio related properties */
 	SPA_PROP_waveType,
 	SPA_PROP_frequency,
-	SPA_PROP_volume,				/**< a volume (Float), 0.0 silence, 1.0 normal */
+	SPA_PROP_volume,			/**< a volume (Float), 0.0 silence, 1.0 no attenutation */
 	SPA_PROP_mute,				/**< mute (Bool) */
 	SPA_PROP_patternType,
 	SPA_PROP_ditherType,
 	SPA_PROP_truncate,
-	SPA_PROP_channelVolumes,		/**< a volume array, one volume per
-						  *  channel (Array of Float) */
+	SPA_PROP_channelVolumes,		/**< a volume array, one volume per channel
+						  * (Array of Float). 0.0 is silence, 1.0 is
+						  *  without attenuation. This is the effective volume
+						  *  that is applied. It can result in a hardware volume
+						  *  and software volume (see softVolumes) */
 	SPA_PROP_volumeBase,			/**< a volume base (Float) */
 	SPA_PROP_volumeStep,			/**< a volume step (Float) */
 	SPA_PROP_channelMap,			/**< a channelmap array
@@ -74,9 +77,11 @@
 	SPA_PROP_monitorVolumes,		/**< a volume array, one volume per
 						  *  channel (Array of Float) */
 	SPA_PROP_latencyOffsetNsec,		/**< delay adjustment */
-	SPA_PROP_softMute,			/**< mute (Bool) */
-	SPA_PROP_softVolumes,			/**< a volume array, one volume per
-						  *  channel (Array of Float) */
+	SPA_PROP_softMute,			/**< mute (Bool) applied in software */
+	SPA_PROP_softVolumes,			/**< a volume array, one volume per channel
+						  * (Array of Float). 0.0 is silence, 1.0 is without
+						  * attenuation. This is the volume applied in software,
+						  * there might be a part applied in hardware. */
 
 	SPA_PROP_iec958Codecs,			/**< enabled IEC958 (S/PDIF) codecs,
 						  *  (Array (Id enum spa_audio_iec958_codec) */

pipewire-0.3.74.tar.gz/spa/include/spa/param/video/dsp-utils.h -> pipewire-0.3.76.tar.gz/spa/include/spa/param/video/dsp-utils.h Changed

pipewire-0.3.74.tar.gz/spa/include/spa/param/video/raw-utils.h -> pipewire-0.3.76.tar.gz/spa/include/spa/param/video/raw-utils.h Changed

pipewire-0.3.74.tar.gz/spa/include/spa/param/video/raw.h -> pipewire-0.3.76.tar.gz/spa/include/spa/param/video/raw.h Changed

pipewire-0.3.76.tar.gz/spa/include/spa/utils/atomic.h Added

@@ -0,0 +1,35 @@
+/* Atomic operations */
+/* SPDX-FileCopyrightText: Copyright © 2023 Wim Taymans */
+/* SPDX-License-Identifier: MIT */
+
+#ifndef SPA_ATOMIC_H
+#define SPA_ATOMIC_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#define SPA_ATOMIC_CAS(v,ov,nv)						\
+({									\
+	__typeof__(v) __ov = (ov);					\
+	__atomic_compare_exchange_n(&(v), &__ov, (nv),			\
+			0, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST);		\
+})
+
+#define SPA_ATOMIC_DEC(s)		__atomic_sub_fetch(&(s), 1, __ATOMIC_SEQ_CST)
+#define SPA_ATOMIC_INC(s)		__atomic_add_fetch(&(s), 1, __ATOMIC_SEQ_CST)
+#define SPA_ATOMIC_LOAD(s)		__atomic_load_n(&(s), __ATOMIC_SEQ_CST)
+#define SPA_ATOMIC_STORE(s,v)		__atomic_store_n(&(s), (v), __ATOMIC_SEQ_CST)
+#define SPA_ATOMIC_XCHG(s,v)		__atomic_exchange_n(&(s), (v), __ATOMIC_SEQ_CST)
+
+#define SPA_SEQ_WRITE(s)		SPA_ATOMIC_INC(s)
+#define SPA_SEQ_WRITE_SUCCESS(s1,s2)	((s1) + 1 == (s2) && ((s2) & 1) == 0)
+
+#define SPA_SEQ_READ(s)			SPA_ATOMIC_LOAD(s)
+#define SPA_SEQ_READ_SUCCESS(s1,s2)	((s1) == (s2) && ((s2) & 1) == 0)
+
+#ifdef __cplusplus
+} /* extern "C" */
+#endif
+
+#endif /* SPA_ATOMIC_H */

pipewire-0.3.74.tar.gz/spa/meson.build -> pipewire-0.3.76.tar.gz/spa/meson.build Changed

@@ -80,7 +80,6 @@
     summary({'Opus': opus_dep.found()}, bool_yn: true, section: 'Bluetooth audio codecs')
     lc3_dep = dependency('lc3', required : get_option('bluez5-codec-lc3'))
     summary({'LC3': lc3_dep.found()}, bool_yn: true, section: 'Bluetooth audio codecs')
-    cdata.set('HAVE_BLUETOOTH_BAP', get_option('bluez5-codec-lc3').allowed() and lc3_dep.found())
     if get_option('bluez5-backend-hsp-native').allowed() or get_option('bluez5-backend-hfp-native').allowed()
       mm_dep = dependency('ModemManager', version : '>= 1.10.0', required : get_option('bluez5-backend-native-mm'))
       summary({'ModemManager': mm_dep.found()}, bool_yn: true, section: 'Bluetooth backends')
@@ -94,6 +93,7 @@
 
   libcamera_dep = dependency('libcamera', required: get_option('libcamera'))
   summary({'libcamera': libcamera_dep.found()}, bool_yn: true, section: 'Backend')
+  cdata.set('HAVE_LIBCAMERA_SYSTEM_DEVICES', libcamera_dep.version().version_compare('>= 0.1.0'))
 
   compress_offload_option = get_option('compress-offload')
   summary({'Compress-Offload': compress_offload_option.allowed()}, bool_yn: true, section: 'Backend')

pipewire-0.3.74.tar.gz/spa/plugins/alsa/alsa-pcm.c -> pipewire-0.3.76.tar.gz/spa/plugins/alsa/alsa-pcm.c Changed

pipewire-0.3.74.tar.gz/spa/plugins/audioconvert/audioconvert.c -> pipewire-0.3.76.tar.gz/spa/plugins/audioconvert/audioconvert.c Changed

@@ -89,6 +89,7 @@
 	unsigned int resample_quality;
 	double rate;
 	char wav_path512;
+	unsigned int lock_volumes:1;
 };
 
 static void props_reset(struct props *props)
@@ -109,6 +110,7 @@
 	props->resample_quality = RESAMPLE_DEFAULT_QUALITY;
 	props->rate = 1.0;
 	spa_zero(props->wav_path);
+	props->lock_volumes = false;
 }
 
 struct buffer {
@@ -695,6 +697,14 @@
 				SPA_PROP_INFO_type, SPA_POD_String(p->wav_path),
 				SPA_PROP_INFO_params, SPA_POD_Bool(true));
 			break;
+		case 27:
+			param = spa_pod_builder_add_object(&b,
+				SPA_TYPE_OBJECT_PropInfo, id,
+				SPA_PROP_INFO_name, SPA_POD_String("channelmix.lock-volumes"),
+				SPA_PROP_INFO_description, SPA_POD_String("Disable volume updates"),
+				SPA_PROP_INFO_type, SPA_POD_CHOICE_Bool(p->lock_volumes),
+				SPA_PROP_INFO_params, SPA_POD_Bool(true));
+			break;
 		default:
 			return 0;
 		}
@@ -773,6 +783,8 @@
 			spa_pod_builder_string(&b, dither_method_infothis->dir1.conv.method.label);
 			spa_pod_builder_string(&b, "debug.wav-path");
 			spa_pod_builder_string(&b, p->wav_path);
+			spa_pod_builder_string(&b, "channelmix.lock-volumes");
+			spa_pod_builder_bool(&b, p->lock_volumes);
 			spa_pod_builder_pop(&b, &f1);
 			param = spa_pod_builder_pop(&b, &f0);
 			break;
@@ -854,6 +866,8 @@
 		spa_scnprintf(this->props.wav_path,
 				sizeof(this->props.wav_path), "%s", s ? s : "");
 	}
+	else if (spa_streq(k, "channelmix.lock-volumes"))
+		this->props.lock_volumes = spa_atob(s);
 	else
 		return 0;
 	return 1;
@@ -1049,13 +1063,15 @@
 		case SPA_PROP_volume:
 			p->prev_volume = p->volume;
 
-			if (spa_pod_get_float(&prop->value, &p->volume) == 0) {
+			if (!p->lock_volumes &&
+			    spa_pod_get_float(&prop->value, &p->volume) == 0) {
 				spa_log_debug(this->log, "%p new volume %f", this, p->volume);
 				changed++;
 			}
 			break;
 		case SPA_PROP_mute:
-			if (spa_pod_get_bool(&prop->value, &p->channel.mute) == 0) {
+			if (!p->lock_volumes &&
+			    spa_pod_get_bool(&prop->value, &p->channel.mute) == 0) {
 				have_channel_volume = true;
 				changed++;
 			}
@@ -1124,7 +1140,8 @@
 			}
 			break;
 		case SPA_PROP_channelVolumes:
-			if ((n = spa_pod_copy_array(&prop->value, SPA_TYPE_Float,
+			if (!p->lock_volumes &&
+			    (n = spa_pod_copy_array(&prop->value, SPA_TYPE_Float,
 					p->channel.volumes, SPA_AUDIO_MAX_CHANNELS)) > 0) {
 				have_channel_volume = true;
 				p->channel.n_volumes = n;
@@ -1139,13 +1156,15 @@
 			}
 			break;
 		case SPA_PROP_softMute:
-			if (spa_pod_get_bool(&prop->value, &p->soft.mute) == 0) {
+			if (!p->lock_volumes &&
+			    spa_pod_get_bool(&prop->value, &p->soft.mute) == 0) {
 				have_soft_volume = true;
 				changed++;
 			}
 			break;
 		case SPA_PROP_softVolumes:
-			if ((n = spa_pod_copy_array(&prop->value, SPA_TYPE_Float,
+			if (!p->lock_volumes &&
+			    (n = spa_pod_copy_array(&prop->value, SPA_TYPE_Float,
 					p->soft.volumes, SPA_AUDIO_MAX_CHANNELS)) > 0) {
 				have_soft_volume = true;
 				p->soft.n_volumes = n;
@@ -1187,7 +1206,7 @@
 		set_volume(this);
 	}
 
-	if (vol_ramp_params_changed) {
+	if (!p->lock_volumes && vol_ramp_params_changed) {
 		void *sequence = NULL;
 		if (p->volume == p->prev_volume)
 			spa_log_error(this->log, "no change in volume, cannot ramp volume");
@@ -2609,7 +2628,8 @@
 	struct dir *dir;
 	int tmp = 0, res = 0;
 	bool in_passthrough, mix_passthrough, resample_passthrough, out_passthrough;
-	bool in_avail = false, flush_in = false, flush_out = false, draining = false, in_empty = true;
+	bool in_avail = false, flush_in = false, flush_out = false;
+	bool draining = false, in_empty = this->out_offset == 0;
 	struct spa_io_buffers *io, *ctrlio = NULL;
 	const struct spa_pod_sequence *ctrl = NULL;

pipewire-0.3.74.tar.gz/spa/plugins/audiomixer/mixer-dsp.c -> pipewire-0.3.76.tar.gz/spa/plugins/audiomixer/mixer-dsp.c Changed

pipewire-0.3.74.tar.gz/spa/plugins/bluez5/backend-hsphfpd.c -> pipewire-0.3.76.tar.gz/spa/plugins/bluez5/backend-hsphfpd.c Changed

@@ -18,6 +18,7 @@
 #include <spa/param/audio/raw.h>
 
 #include "defs.h"
+#include "dbus-helpers.h"
 
 static struct spa_log_topic log_topic = SPA_LOG_TOPIC(0, "spa.bluez5.hsphfpd");
 #undef SPA_LOG_TOPIC_DEFAULT
@@ -764,7 +765,6 @@
 {
 	const char *interface_name = HSPHFPD_AUDIO_AGENT_INTERFACE;
 	DBusMessageIter object, array, entry, dict, codec, data;
-	char *str = "AgentCodec";
 
 	dbus_message_iter_open_container(iter, DBUS_TYPE_DICT_ENTRY, NULL, &object);
 	dbus_message_iter_append_basic(&object, DBUS_TYPE_OBJECT_PATH, &endpoint);
@@ -777,7 +777,7 @@
 	dbus_message_iter_open_container(&entry, DBUS_TYPE_ARRAY, "{sv}", &dict);
 
 	dbus_message_iter_open_container(&dict, DBUS_TYPE_DICT_ENTRY, NULL, &codec);
-	dbus_message_iter_append_basic(&codec, DBUS_TYPE_STRING, &str);
+	dbus_message_iter_append_basic(&codec, DBUS_TYPE_STRING, &(const char *) { "AgentCodec" });
 	dbus_message_iter_open_container(&codec, DBUS_TYPE_VARIANT, "s", &data);
 	dbus_message_iter_append_basic(&data, DBUS_TYPE_STRING, &agent_codec);
 	dbus_message_iter_close_container(&codec, &data);
@@ -847,7 +847,7 @@
 
 	backend->acquire_in_progress = false;
 
-	r = dbus_pending_call_steal_reply(pending);
+	r = steal_reply_and_unref(&pending);
 	if (r == NULL)
 		return;
 
@@ -890,7 +890,6 @@
 
 finish:
 	dbus_message_unref(r);
-	dbus_pending_call_unref(pending);
 
 	if (ret < 0)
 		spa_bt_transport_set_state(transport, SPA_BT_TRANSPORT_STATE_ERROR);
@@ -906,7 +905,6 @@
 	const char *air_codec = HSPHFP_AIR_CODEC_CVSD;
 	const char *agent_codec = HSPHFP_AGENT_CODEC_PCM;
 	DBusPendingCall *call;
-	DBusError err;
 
 	spa_log_debug(backend->log, "transport %p: Acquire %s",
 			transport, transport->path);
@@ -927,8 +925,6 @@
 		return -ENOMEM;
 	dbus_message_append_args(m, DBUS_TYPE_STRING, &air_codec, DBUS_TYPE_STRING, &agent_codec, DBUS_TYPE_INVALID);
 
-	dbus_error_init(&err);
-
 	dbus_connection_send_with_reply(backend->conn, m, &call, -1);
 	dbus_pending_call_set_notify(call, hsphfpd_audio_acquire_reply, transport, NULL);
 	dbus_message_unref(m);
@@ -1183,7 +1179,7 @@
 	DBusMessage *r;
 	DBusMessageIter i, array_i;
 
-	r = dbus_pending_call_steal_reply(pending);
+	r = steal_reply_and_unref(&pending);
 	if (r == NULL)
 		return;
 
@@ -1216,7 +1212,6 @@
 
 finish:
 	dbus_message_unref(r);
-	dbus_pending_call_unref(pending);
 }
 
 static int backend_hsphfpd_register(void *data)
@@ -1303,9 +1298,6 @@
 {
 	const char *sender;
 	struct impl *backend = user_data;
-	DBusError err;
-
-	dbus_error_init(&err);
 
 	sender = dbus_message_get_sender(m);

pipewire-0.3.74.tar.gz/spa/plugins/bluez5/backend-native.c -> pipewire-0.3.76.tar.gz/spa/plugins/bluez5/backend-native.c Changed

@@ -33,6 +33,7 @@
 #include <libusb.h>
 #endif
 
+#include "dbus-helpers.h"
 #include "modemmanager.h"
 #include "upower.h"
 
@@ -1163,10 +1164,9 @@
 			return true;
 		}
 	} else if (spa_strstartswith(buf, "AT+VTS=")) {
-		char *dtmf;
+		char dtmf2;
 		enum cmee_error error;
 
-		dtmf = calloc(1, 2);
 		if (sscanf(buf, "AT+VTS=%1s", dtmf) != 1) {
 			spa_log_debug(backend->log, "Failed to parse AT+VTS: \"%s\"", buf);
 			rfcomm_send_error(rfcomm, CMEE_AG_FAILURE);
@@ -2196,7 +2196,7 @@
 {
 	struct impl *backend = userdata;
 	DBusMessage *r;
-	DBusMessageIter it5;
+	DBusMessageIter it;
 	const char *handler, *path;
 	enum spa_bt_profile profile;
 	struct rfcomm *rfcomm;
@@ -2216,8 +2216,8 @@
 		return DBUS_HANDLER_RESULT_NOT_YET_HANDLED;
 	}
 
-	dbus_message_iter_init(m, &it0);
-	dbus_message_iter_get_basic(&it0, &path);
+	dbus_message_iter_init(m, &it);
+	dbus_message_iter_get_basic(&it, &path);
 
 	d = spa_bt_device_find(backend->monitor, path);
 	if (d == NULL || d->adapter == NULL) {
@@ -2226,8 +2226,8 @@
 	}
 	spa_bt_device_add_profile(d, profile);
 
-	dbus_message_iter_next(&it0);
-	dbus_message_iter_get_basic(&it0, &fd);
+	dbus_message_iter_next(&it);
+	dbus_message_iter_get_basic(&it, &fd);
 
 	spa_log_debug(backend->log, "NewConnection path=%s, fd=%d, profile %s", path, fd, handler);
 
@@ -2414,7 +2414,7 @@
 	struct impl *backend = user_data;
 	DBusMessage *r;
 
-	r = dbus_pending_call_steal_reply(pending);
+	r = steal_reply_and_unref(&pending);
 	if (r == NULL)
 		return;
 
@@ -2432,9 +2432,8 @@
 		goto finish;
 	}
 
-      finish:
+finish:
 	dbus_message_unref(r);
-        dbus_pending_call_unref(pending);
 }
 
 static int register_profile(struct impl *backend, const char *profile, const char *uuid)

pipewire-0.3.74.tar.gz/spa/plugins/bluez5/backend-ofono.c -> pipewire-0.3.76.tar.gz/spa/plugins/bluez5/backend-ofono.c Changed

pipewire-0.3.74.tar.gz/spa/plugins/bluez5/bluez5-dbus.c -> pipewire-0.3.76.tar.gz/spa/plugins/bluez5/bluez5-dbus.c Changed

@@ -35,6 +35,7 @@
 
 #include "config.h"
 #include "codec-loader.h"
+#include "dbus-helpers.h"
 #include "player.h"
 #include "iso-io.h"
 #include "defs.h"
@@ -225,12 +226,7 @@
 	DBusMessage *m;
 	const char *interface;
 
-	if (device->battery_pending_call) {
-		spa_log_debug(device->monitor->log, "Cancelling and freeing pending battery provider register call");
-		dbus_pending_call_cancel(device->battery_pending_call);
-		dbus_pending_call_unref(device->battery_pending_call);
-		device->battery_pending_call = NULL;
-	}
+	cancel_and_unref(&device->battery_pending_call);
 
 	if (!device->adapter || !device->adapter->has_battery_provider || !device->has_battery)
 		return;
@@ -355,10 +351,8 @@
 	DBusMessage *reply;
 	struct spa_bt_device *device = data;
 
-	reply = dbus_pending_call_steal_reply(pending_call);
-	dbus_pending_call_unref(pending_call);
-
-	device->battery_pending_call = NULL;
+	spa_assert(device->battery_pending_call == pending_call);
+	reply = steal_reply_and_unref(&device->battery_pending_call);
 
 	if (dbus_message_get_type(reply) == DBUS_MESSAGE_TYPE_ERROR) {
 		spa_log_error(device->monitor->log, "Failed to register battery provider. Error: %s", dbus_message_get_error_name(reply));
@@ -422,9 +416,7 @@
 		    device->battery_pending_call, on_battery_provider_registered,
 		    device, NULL)) {
 		spa_log_error(device->monitor->log, "Failed to register battery provider");
-		dbus_pending_call_cancel(device->battery_pending_call);
-		dbus_pending_call_unref(device->battery_pending_call);
-		device->battery_pending_call = NULL;
+		cancel_and_unref(&device->battery_pending_call);
 	}
 }
 
@@ -2340,7 +2332,7 @@
 		if (j >= size) {
 			const struct media_codec **p;
 			size = size * 2;
-#ifdef HAVE_REALLOCARRRAY
+#ifdef HAVE_REALLOCARRAY
 			p = reallocarray(supported_codecs, size, sizeof(const struct media_codec *));
 #else
 			p = realloc(supported_codecs, size * sizeof(const struct media_codec *));
@@ -2634,17 +2626,8 @@
 
 	spa_bt_transport_destroy(transport);
 
-	if (transport->acquire_call) {
-		dbus_pending_call_cancel(transport->acquire_call);
-		dbus_pending_call_unref(transport->acquire_call);
-		transport->acquire_call = NULL;
-	}
-
-	if (transport->volume_call) {
-		dbus_pending_call_cancel(transport->volume_call);
-		dbus_pending_call_unref(transport->volume_call);
-		transport->volume_call = NULL;
-	}
+	cancel_and_unref(&transport->acquire_call);
+	cancel_and_unref(&transport->volume_call);
 
 	if (transport->fd >= 0) {
 		spa_bt_player_set_state(transport->device->adapter->dummy_player, SPA_BT_PLAYER_STOPPED);
@@ -2669,6 +2652,7 @@
 
 	spa_list_remove(&transport->bap_transport_linked);
 
+	free(transport->configuration);
 	free(transport->endpoint_path);
 	free(transport->path);
 	free(transport);
@@ -3242,11 +3226,8 @@
 	DBusError err = DBUS_ERROR_INIT;
 	DBusMessage *r;
 
-	r = dbus_pending_call_steal_reply(pending);
-
 	spa_assert(transport->volume_call == pending);
-	dbus_pending_call_unref(pending);
-	transport->volume_call = NULL;
+	r = steal_reply_and_unref(&transport->volume_call);
 
 	if (dbus_set_error_from_message(&err, r)) {
 		spa_log_info(monitor->log, "transport %p: set volume failed for transport %s: %s",
@@ -3270,11 +3251,7 @@
 	int res = 0;
 	dbus_bool_t ret;
 
-	if (transport->volume_call) {
-		dbus_pending_call_cancel(transport->volume_call);
-		dbus_pending_call_unref(transport->volume_call);
-		transport->volume_call = NULL;
-	}
+	cancel_and_unref(&transport->volume_call);
 
 	m = dbus_message_new_method_call(BLUEZ_SERVICE,
 					 transport->path,
@@ -3400,11 +3377,8 @@
 	DBusMessage *r;
 	struct spa_bt_transport *t, *t_linked;
 
-	r = dbus_pending_call_steal_reply(pending);
-
 	spa_assert(transport->acquire_call == pending);
-	dbus_pending_call_unref(pending);
-	transport->acquire_call = NULL;
+	r = steal_reply_and_unref(&transport->acquire_call);
 
 	spa_bt_device_update_last_bluez_action_time(device);
 
@@ -3613,11 +3587,7 @@
 
 	spa_bt_transport_set_state(transport, SPA_BT_TRANSPORT_STATE_IDLE);
 
-	if (transport->acquire_call) {
-		dbus_pending_call_cancel(transport->acquire_call);
-		dbus_pending_call_unref(transport->acquire_call);
-		transport->acquire_call = NULL;
-	}
+	cancel_and_unref(&transport->acquire_call);
 
 	if (transport->iso_io) {
 		spa_log_debug(monitor->log, "transport %p: remove ISO IO", transport);
@@ -3742,10 +3712,7 @@
 
 	media_codec_switch_stop_timer(sw);
 
-	if (sw->pending != NULL) {
-		dbus_pending_call_cancel(sw->pending);
-		dbus_pending_call_unref(sw->pending);
-	}
+	cancel_and_unref(&sw->pending);
 
 	if (sw->device != NULL)
 		spa_list_remove(&sw->device_link);
@@ -3992,11 +3959,8 @@
 	struct spa_bt_device *device = sw->device;
 	DBusMessage *r;
 
-	r = dbus_pending_call_steal_reply(pending);
-
 	spa_assert(sw->pending == pending);
-	dbus_pending_call_unref(pending);
-	sw->pending = NULL;
+	r = steal_reply_and_unref(&sw->pending);
 
 	spa_bt_device_update_last_bluez_action_time(device);
 
@@ -4475,9 +4439,7 @@
 	struct spa_bt_monitor *monitor = adapter->monitor;
 	DBusMessage *r;
 
-	r = dbus_pending_call_steal_reply(pending);
-	dbus_pending_call_unref(pending);
-
+	r = steal_reply_and_unref(&pending);
 	if (r == NULL)
 		return;
 
@@ -4776,9 +4738,7 @@
 	DBusMessage *r;
 	bool fallback = true;
 
-	r = dbus_pending_call_steal_reply(pending);
-	dbus_pending_call_unref(pending);
-
+	r = steal_reply_and_unref(&pending);
 	if (r == NULL)
 		return;
 
@@ -4809,9 +4769,7 @@
 	struct spa_bt_monitor *monitor = adapter->monitor;
 	DBusMessage *r;
 
-	r = dbus_pending_call_steal_reply(pending);
-	dbus_pending_call_unref(pending);
-
+	r = steal_reply_and_unref(&pending);
 	if (r == NULL)
 		return;
 
@@ -5225,12 +5183,8 @@
 	DBusMessage *r;
 	DBusMessageIter it6;
 
-	spa_assert(pending == monitor->get_managed_objects_call);
-	monitor->get_managed_objects_call = NULL;
-
-	r = dbus_pending_call_steal_reply(pending);
-	dbus_pending_call_unref(pending);
-
+	spa_assert(monitor->get_managed_objects_call == pending);
+	r = steal_reply_and_unref(&monitor->get_managed_objects_call);
 	if (r == NULL)
 		return;
 
@@ -5625,10 +5579,7 @@
 		monitor->filters_added = false;
 	}
 
-	if (monitor->get_managed_objects_call) {
-		dbus_pending_call_cancel(monitor->get_managed_objects_call);
-		dbus_pending_call_unref(monitor->get_managed_objects_call);
-	}
+	cancel_and_unref(&monitor->get_managed_objects_call);
 
 	spa_list_consume(t, &monitor->transport_list, link)
 		spa_bt_transport_free(t);

pipewire-0.3.76.tar.gz/spa/plugins/bluez5/dbus-helpers.h Added

pipewire-0.3.74.tar.gz/spa/plugins/bluez5/modemmanager.c -> pipewire-0.3.76.tar.gz/spa/plugins/bluez5/modemmanager.c Changed

@@ -7,6 +7,7 @@
 
 #include <ModemManager.h>
 
+#include "dbus-helpers.h"
 #include "modemmanager.h"
 
 #define DBUS_INTERFACE_OBJECTMANAGER "org.freedesktop.DBus.ObjectManager"
@@ -56,8 +57,7 @@
 	ret = dbus_pending_call_set_notify(pending_call, function, user_data, NULL);
 	if (!ret) {
 		spa_log_debug(this->log, "dbus set notify failure");
-		dbus_pending_call_cancel(pending_call);
-		dbus_pending_call_unref(pending_call);
+		cancel_and_unref(&pending_call);
 		goto out;
 	}
 
@@ -125,10 +125,7 @@
 	MMCallState state;
 
 	spa_assert(call->pending == pending);
-	call->pending = NULL;
-
-	r = dbus_pending_call_steal_reply(pending);
-	dbus_pending_call_unref(pending);
+	r = steal_reply_and_unref(&call->pending);
 	if (r == NULL)
 		return;
 
@@ -424,10 +421,7 @@
 	DBusMessageIter i, array_i;
 
 	spa_assert(this->pending == pending);
-	this->pending = NULL;
-
-	r = dbus_pending_call_steal_reply(pending);
-	dbus_pending_call_unref(pending);
+	r = steal_reply_and_unref(&this->pending);
 	if (r == NULL)
 		return;
 
@@ -455,13 +449,11 @@
 	dbus_message_unref(r);
 }
 
-static void call_free(struct call *call) {
+static void call_free(struct call *call)
+{
 	spa_list_remove(&call->link);
 
-	if (call->pending != NULL) {
-		dbus_pending_call_cancel(call->pending);
-		dbus_pending_call_unref(call->pending);
-	}
+	cancel_and_unref(&call->pending);
 
 	if (call->number)
 		free(call->number);
@@ -477,10 +469,7 @@
 	spa_list_consume(call, &this->call_list, link)
 		call_free(call);
 
-	if (this->voice_pending != NULL) {
-		dbus_pending_call_cancel(this->voice_pending);
-		dbus_pending_call_unref(this->voice_pending);
-	}
+	cancel_and_unref(&this->voice_pending);
 
 	if (this->ops->set_call_setup)
 		this->ops->set_call_setup(CIND_CALLSETUP_NONE, this->user_data);
@@ -787,10 +776,7 @@
 	free(data);
 
 	spa_assert(call->pending == pending);
-	call->pending = NULL;
-
-	r = dbus_pending_call_steal_reply(pending);
-	dbus_pending_call_unref(pending);
+	r = steal_reply_and_unref(&call->pending);
 	if (r == NULL)
 		return;
 
@@ -820,10 +806,7 @@
 	free(data);
 
 	spa_assert(this->voice_pending == pending);
-	this->voice_pending = NULL;
-
-	r = dbus_pending_call_steal_reply(pending);
-	dbus_pending_call_unref(pending);
+	r = steal_reply_and_unref(&this->voice_pending);
 	if (r == NULL)
 		return;
 
@@ -1158,10 +1141,7 @@
 {
 	struct impl *this = data;
 
-	if (this->pending != NULL) {
-		dbus_pending_call_cancel(this->pending);
-		dbus_pending_call_unref(this->pending);
-	}
+	cancel_and_unref(&this->pending);
 
 	mm_clean_voice(this);
 	mm_clean_modem3gpp(this);

pipewire-0.3.74.tar.gz/spa/plugins/bluez5/upower.c -> pipewire-0.3.76.tar.gz/spa/plugins/bluez5/upower.c Changed

@@ -5,6 +5,7 @@
 #include <errno.h>
 #include <spa/utils/string.h>
 
+#include "dbus-helpers.h"
 #include "upower.h"
 
 #define UPOWER_SERVICE "org.freedesktop.UPower"
@@ -17,6 +18,8 @@
 	struct spa_log *log;
 	DBusConnection *conn;
 
+	DBusPendingCall *pending_get_call;
+
 	bool filters_added;
 
 	void *user_data;
@@ -43,8 +46,8 @@
 	DBusMessage *r;
 	DBusMessageIter i, variant_i;
 
-	r = dbus_pending_call_steal_reply(pending);
-	dbus_pending_call_unref(pending);
+	spa_assert(backend->pending_get_call == pending);
+	r = steal_reply_and_unref(&backend->pending_get_call);
 	if (r == NULL)
 		return;
 
@@ -66,6 +69,31 @@
 	dbus_message_unref(r);
 }
 
+static int update_battery_percentage(struct impl *this)
+{
+	cancel_and_unref(&this->pending_get_call);
+
+	DBusMessage *m = dbus_message_new_method_call(UPOWER_SERVICE,
+						      UPOWER_DISPLAY_DEVICE_OBJECT,
+						      DBUS_INTERFACE_PROPERTIES,
+						      "Get");
+	if (!m)
+		return -ENOMEM;
+
+	dbus_message_append_args(m,
+				 DBUS_TYPE_STRING, &(const char *){ UPOWER_DEVICE_INTERFACE },
+				 DBUS_TYPE_STRING, &(const char *){ "Percentage" },
+				 DBUS_TYPE_INVALID);
+	dbus_message_set_auto_start(m, false);
+
+	dbus_connection_send_with_reply(this->conn, m, &this->pending_get_call, -1);
+	dbus_pending_call_set_notify(this->pending_get_call, upower_get_percentage_properties_reply, this, NULL);
+
+	dbus_message_unref(m);
+
+	return 0;
+}
+
 static void upower_clean(struct impl *this)
 {
 	this->set_battery_level(0, this->user_data);
@@ -99,20 +127,8 @@
 			}
 
 			if (new_owner && *new_owner) {
-				DBusPendingCall *call;
-				static const char* upower_device_interface = UPOWER_DEVICE_INTERFACE;
-				static const char* percentage_property = "Percentage";
-
 				spa_log_debug(this->log, "UPower daemon appeared (%s)", new_owner);
-
-				m = dbus_message_new_method_call(UPOWER_SERVICE, UPOWER_DISPLAY_DEVICE_OBJECT, DBUS_INTERFACE_PROPERTIES, "Get");
-				if (m == NULL)
-					goto finish;
-				dbus_message_append_args(m, DBUS_TYPE_STRING, &upower_device_interface,
-						         DBUS_TYPE_STRING, &percentage_property, DBUS_TYPE_INVALID);
-				dbus_connection_send_with_reply(this->conn, m, &call, -1);
-				dbus_pending_call_set_notify(call, upower_get_percentage_properties_reply, this, NULL);
-				dbus_message_unref(m);
+				update_battery_percentage(this);
 			}
 		}
 	} else if (dbus_message_is_signal(m, DBUS_INTERFACE_PROPERTIES, DBUS_SIGNAL_PROPERTIES_CHANGED)) {
@@ -210,26 +226,12 @@
 	this->set_battery_level = set_battery_level;
 	this->user_data = user_data;
 
-	if (add_filters(this) < 0) {
+	if (add_filters(this) < 0)
 		goto fail4;
-	}
-
-	DBusMessage *m;
-	DBusPendingCall *call;
 
-	m = dbus_message_new_method_call(UPOWER_SERVICE, UPOWER_DISPLAY_DEVICE_OBJECT, DBUS_INTERFACE_PROPERTIES, "Get");
-	if (m == NULL)
+	if (update_battery_percentage(this) < 0)
 		goto fail4;
 
-	dbus_message_append_args(m,
-				 DBUS_TYPE_STRING, &(const char *){ UPOWER_DEVICE_INTERFACE },
-				 DBUS_TYPE_STRING, &(const char *){ "Percentage" },
-				 DBUS_TYPE_INVALID);
-	dbus_message_set_auto_start(m, false);
-	dbus_connection_send_with_reply(this->conn, m, &call, -1);
-	dbus_pending_call_set_notify(call, upower_get_percentage_properties_reply, this, NULL);
-	dbus_message_unref(m);
-
 	return this;
 
 fail4:
@@ -241,6 +243,8 @@
 {
 	struct impl *this = data;
 
+	cancel_and_unref(&this->pending_get_call);
+
 	if (this->filters_added) {
 		dbus_connection_remove_filter(this->conn, upower_filter_cb, this);
 		this->filters_added = false;

pipewire-0.3.74.tar.gz/spa/plugins/libcamera/libcamera-device.cpp -> pipewire-0.3.76.tar.gz/spa/plugins/libcamera/libcamera-device.cpp Changed

@@ -4,6 +4,8 @@
 /* SPDX-FileCopyrightText: Copyright © 2021 Wim Taymans <wim.taymans@gmail.com> */
 /* SPDX-License-Identifier: MIT */
 
+#include "config.h"
+
 #include <stddef.h>
 #include <sys/types.h>
 #include <sys/stat.h>
@@ -29,6 +31,7 @@
 
 #include <libcamera/camera.h>
 #include <libcamera/property_ids.h>
+#include <libcamera/base/span.h>
 
 using namespace libcamera;
 
@@ -55,6 +58,19 @@
 
 }
 
+static const libcamera::Span<const int64_t> cameraDevice(
+			const Camera *camera)
+{
+#ifdef HAVE_LIBCAMERA_SYSTEM_DEVICES
+	const ControlList &props = camera->properties();
+
+	if (auto devices = props.get(properties::SystemDevices))
+		return devices.value();
+#endif
+
+	return {};
+}
+
 static std::string cameraModel(const Camera *camera)
 {
 	const ControlList &props = camera->properties();
@@ -90,7 +106,8 @@
 	uint32_t n_items = 0;
 	struct spa_device_info info;
 	struct spa_param_info params2;
-	char path256, model256, name256;
+	char path256, model256, name256, devices_str128;
+	struct spa_strbuf buf;
 
 	info = SPA_DEVICE_INFO_INIT();
 
@@ -111,6 +128,19 @@
 	ADD_ITEM(SPA_KEY_DEVICE_DESCRIPTION, model);
 	snprintf(name, sizeof(name), "libcamera_device.%s", impl->device_id.c_str());
 	ADD_ITEM(SPA_KEY_DEVICE_NAME, name);
+
+	auto device_numbers = cameraDevice(impl->camera.get());
+
+	if (!device_numbers.empty()) {
+		spa_strbuf_init(&buf, devices_str, sizeof(devices_str));
+
+		/* created a space separated string of all the device numbers */
+		for (int64_t device_number : device_numbers)
+			spa_strbuf_append(&buf, "%" PRId64 " ", device_number);
+
+		ADD_ITEM(SPA_KEY_DEVICE_DEVIDS, devices_str);
+	}
+
 #undef ADD_ITEM
 
 	dict = SPA_DICT_INIT(items, n_items);

pipewire-0.3.74.tar.gz/spa/plugins/libcamera/meson.build -> pipewire-0.3.76.tar.gz/spa/plugins/libcamera/meson.build Changed

pipewire-0.3.74.tar.gz/spa/plugins/support/dbus.c -> pipewire-0.3.76.tar.gz/spa/plugins/support/dbus.c Changed

pipewire-0.3.74.tar.gz/spa/plugins/v4l2/v4l2-device.c -> pipewire-0.3.76.tar.gz/spa/plugins/v4l2/v4l2-device.c Changed

@@ -29,6 +29,7 @@
 
 struct props {
 	char device64;
+	char devnum32;
 	char product_id6;
 	char vendor_id6;
 	int device_fd;
@@ -55,11 +56,11 @@
 static int emit_info(struct impl *this, bool full)
 {
 	int res;
-	struct spa_dict_item items12;
+	struct spa_dict_item items13;
 	uint32_t n_items = 0;
 	struct spa_device_info info;
 	struct spa_param_info params2;
-        char path128, version16, capabilities16, device_caps16;
+	char path128, version16, capabilities16, device_caps16;
 
 	if ((res = spa_v4l2_open(&this->dev, this->props.device)) < 0)
 		return res;
@@ -78,6 +79,7 @@
 	if (this->props.vendor_id0)
 		ADD_ITEM(SPA_KEY_DEVICE_VENDOR_ID, this->props.vendor_id);
 	ADD_ITEM(SPA_KEY_API_V4L2_PATH, (char *)this->props.device);
+	ADD_ITEM(SPA_KEY_DEVICE_DEVIDS, (char *)this->props.devnum);
 	ADD_ITEM(SPA_KEY_API_V4L2_CAP_DRIVER, (char *)this->dev.cap.driver);
 	ADD_ITEM(SPA_KEY_API_V4L2_CAP_CARD, (char *)this->dev.cap.card);
 	ADD_ITEM(SPA_KEY_API_V4L2_CAP_BUS_INFO, (char *)this->dev.cap.bus_info);
@@ -233,6 +235,8 @@
 
 	if (info && (str = spa_dict_lookup(info, SPA_KEY_API_V4L2_PATH)))
 		strncpy(this->props.device, str, 63);
+	if (info && (str = spa_dict_lookup(info, SPA_KEY_DEVICE_DEVIDS)))
+		strncpy(this->props.devnum, str, 31);
 	if (info && (str = spa_dict_lookup(info, SPA_KEY_DEVICE_PRODUCT_ID)))
 		strncpy(this->props.product_id, str, 5);
 	if (info && (str = spa_dict_lookup(info, SPA_KEY_DEVICE_VENDOR_ID)))

pipewire-0.3.74.tar.gz/spa/plugins/v4l2/v4l2-udev.c -> pipewire-0.3.76.tar.gz/spa/plugins/v4l2/v4l2-udev.c Changed

pipewire-0.3.74.tar.gz/src/modules/meson.build -> pipewire-0.3.76.tar.gz/src/modules/meson.build Changed

@@ -38,6 +38,8 @@
   'module-rtp-session.c',
   'module-rtp-source.c',
   'module-rtp-sink.c',
+  'module-vban-recv.c',
+  'module-vban-send.c',
   'module-session-manager.c',
   'module-zeroconf-discover.c',
   'module-roc-source.c',
@@ -659,6 +661,26 @@
   dependencies : mathlib, dl_lib, rt_lib, pipewire_dep,
 )
 
+pipewire_module_vban_send = shared_library('pipewire-module-vban-send',
+   'module-vban-send.c',
+    'module-vban/stream.c' ,
+  include_directories : configinc,
+  install : true,
+  install_dir : modules_install_dir,
+  install_rpath: modules_install_dir,
+  dependencies : mathlib, dl_lib, rt_lib, pipewire_dep,
+)
+
+pipewire_module_vban_recv = shared_library('pipewire-module-vban-recv',
+   'module-vban-recv.c',
+    'module-vban/stream.c' ,
+  include_directories : configinc,
+  install : true,
+  install_dir : modules_install_dir,
+  install_rpath: modules_install_dir,
+  dependencies : mathlib, dl_lib, rt_lib, pipewire_dep,
+)
+
 build_module_roc = roc_dep.found()
 if build_module_roc
   pipewire_module_roc_sink = shared_library('pipewire-module-roc-sink',

pipewire-0.3.74.tar.gz/src/modules/module-client-node/client-node.c -> pipewire-0.3.76.tar.gz/src/modules/module-client-node/client-node.c Changed

pipewire-0.3.74.tar.gz/src/modules/module-client-node/remote-node.c -> pipewire-0.3.76.tar.gz/src/modules/module-client-node/remote-node.c Changed

@@ -45,7 +45,6 @@
 
 struct node_data {
 	struct pw_context *context;
-	struct spa_hook context_listener;
 
 	struct pw_loop *data_loop;
 	struct spa_system *data_system;
@@ -61,6 +60,7 @@
 
 	struct pw_impl_node *node;
 	struct spa_hook node_listener;
+	struct spa_hook node_rt_listener;
 	unsigned int do_free:1;
 	unsigned int have_transport:1;
 	unsigned int allow_mlock:1;
@@ -889,8 +889,9 @@
 {
 	pw_log_debug("port %p: mix clear %d.%d", mix->port, mix->port->port_id, mix->mix.id);
 
-	spa_node_port_set_io(mix->port->mix, mix->mix.port.direction,
-			mix->mix.port.port_id, SPA_IO_Buffers, NULL, 0);
+	if (mix->mix.id != SPA_ID_INVALID)
+		spa_node_port_set_io(mix->port->mix, mix->mix.port.direction,
+				mix->mix.port.port_id, SPA_IO_Buffers, NULL, 0);
 
 	spa_list_remove(&mix->link);
 
@@ -1129,11 +1130,10 @@
 	spa_hook_remove(&data->proxy_client_node_listener);
 	spa_hook_remove(&data->client_node_listener);
 
-	pw_context_driver_remove_listener(data->context,
-			&data->context_listener);
-
 	if (data->node) {
 		spa_hook_remove(&data->node_listener);
+		pw_impl_node_remove_rt_listener(data->node,
+				&data->node_rt_listener);
 		pw_impl_node_set_state(data->node, PW_NODE_STATE_SUSPENDED);
 
 		clean_node(data);
@@ -1168,12 +1168,13 @@
 	.bound_props = client_node_bound_props,
 };
 
-static void context_complete(void *data, struct pw_impl_node *node)
+static void node_rt_complete(void *data)
 {
 	struct node_data *d = data;
+	struct pw_impl_node *node = d->node;
 	struct spa_system *data_system = d->data_system;
 
-	if (node != d->node || !node->driving ||
+	if (!node->driving ||
 	    !SPA_FLAG_IS_SET(node->rt.target.activation->flags, PW_NODE_ACTIVATION_FLAG_PROFILER))
 		return;
 
@@ -1181,9 +1182,9 @@
 		pw_log_warn("node %p: write failed %m", node);
 }
 
-static const struct pw_context_driver_events context_events = {
-	PW_VERSION_CONTEXT_DRIVER_EVENTS,
-	.complete = context_complete,
+static const struct pw_impl_node_rt_events node_rt_events = {
+	PW_VERSION_IMPL_NODE_RT_EVENTS,
+	.complete = node_rt_complete,
 };
 
 static struct pw_proxy *node_export(struct pw_core *core, void *object, bool do_free,
@@ -1238,14 +1239,13 @@
 			&proxy_client_node_events, data);
 
 	pw_impl_node_add_listener(node, &data->node_listener, &node_events, data);
+	pw_impl_node_add_rt_listener(node, &data->node_rt_listener,
+			&node_rt_events, data);
 
 	pw_client_node_add_listener(data->client_node,
 					  &data->client_node_listener,
 					  &client_node_events,
 					  data);
-	pw_context_driver_add_listener(data->context,
-			&data->context_listener,
-			&context_events, data);
 
 	do_node_init(data);

pipewire-0.3.74.tar.gz/src/modules/module-profiler.c -> pipewire-0.3.76.tar.gz/src/modules/module-profiler.c Changed

@@ -52,10 +52,8 @@
 #define PW_LOG_TOPIC_DEFAULT mod_topic
 
 #define TMP_BUFFER		(16 * 1024)
-#define MAX_BUFFER		(8 * 1024 * 1024)
-#define MIN_FLUSH		(16 * 1024)
-#define DEFAULT_IDLE		5
-#define DEFAULT_INTERVAL	1
+#define DATA_BUFFER		(32 * 1024)
+#define FLUSH_BUFFER		(8 * 1024 * 1024)
 
 int pw_protocol_native_ext_profiler_init(struct pw_context *context);
 
@@ -71,6 +69,21 @@
 	{ PW_KEY_MODULE_VERSION, PACKAGE_VERSION },
 };
 
+struct node {
+	struct spa_list link;
+	struct impl *impl;
+
+	struct pw_impl_node *node;
+	struct spa_hook node_rt_listener;
+
+	int64_t count;
+	struct spa_ringbuffer buffer;
+	uint8_t tmpTMP_BUFFER;
+	uint8_t dataDATA_BUFFER;
+
+	unsigned enabled:1;
+};
+
 struct impl {
 	struct pw_context *context;
 	struct pw_properties *properties;
@@ -84,18 +97,13 @@
 	struct pw_global *global;
 	struct spa_hook global_listener;
 
-	int64_t count;
+	struct spa_list node_list;
+
 	uint32_t busy;
-	uint32_t empty;
-	struct spa_source *flush_timeout;
-	unsigned int flushing:1;
+	struct spa_source *flush_event;
 	unsigned int listening:1;
 
-	struct spa_ringbuffer buffer;
-	uint8_t tmpTMP_BUFFER;
-	uint8_t dataMAX_BUFFER;
-
-	uint8_t flushMAX_BUFFER + sizeof(struct spa_pod_struct);
+	uint8_t flushFLUSH_BUFFER + sizeof(struct spa_pod_struct);
 };
 
 struct resource_data {
@@ -105,69 +113,45 @@
 	struct spa_hook resource_listener;
 };
 
-static void start_flush(struct impl *impl)
-{
-	struct timespec value, interval;
-
-	value.tv_sec = 0;
-        value.tv_nsec = 1;
-	interval.tv_sec = DEFAULT_INTERVAL;
-        interval.tv_nsec = 0;
-        pw_loop_update_timer(impl->main_loop,
-			impl->flush_timeout, &value, &interval, false);
-	impl->flushing = true;
-}
-
-static void stop_flush(struct impl *impl)
-{
-	struct timespec value, interval;
-
-	if (!impl->flushing)
-		return;
-
-	value.tv_sec = 0;
-        value.tv_nsec = 0;
-	interval.tv_sec = 0;
-        interval.tv_nsec = 0;
-        pw_loop_update_timer(impl->main_loop,
-			impl->flush_timeout, &value, &interval, false);
-	impl->flushing = false;
-}
-
-static void flush_timeout(void *data, uint64_t expirations)
+static void do_flush_event(void *data, uint64_t count)
 {
 	struct impl *impl = data;
-	int32_t avail;
-	uint32_t idx;
-	struct spa_pod_struct *p;
 	struct pw_resource *resource;
+	struct node *n;
+	uint32_t total = 0;
+	struct spa_pod_struct *p;
 
-	avail = spa_ringbuffer_get_read_index(&impl->buffer, &idx);
+	p = (struct spa_pod_struct *)impl->flush;
 
-	pw_log_trace("%p avail %d", impl, avail);
+	spa_list_for_each(n, &impl->node_list, link) {
+		int32_t avail;
+		uint32_t idx;
 
-	if (avail <= 0) {
-		if (++impl->empty == DEFAULT_IDLE)
-			stop_flush(impl);
-		return;
-	}
-	impl->empty = 0;
+		avail = spa_ringbuffer_get_read_index(&n->buffer, &idx);
 
-	p = (struct spa_pod_struct *)impl->flush;
-	*p = SPA_POD_INIT_Struct(avail);
+		pw_log_trace("%p avail %d", impl, avail);
+
+		if (avail > 0) {
+			spa_ringbuffer_read_data(&n->buffer, n->data, DATA_BUFFER,
+					idx % DATA_BUFFER,
+					SPA_PTROFF(p, sizeof(struct spa_pod_struct) + total, void),
+					avail);
+			spa_ringbuffer_read_update(&n->buffer, idx + avail);
+			total += avail;
+		}
+	}
 
-	spa_ringbuffer_read_data(&impl->buffer, impl->data, MAX_BUFFER,
-			idx % MAX_BUFFER,
-			SPA_PTROFF(p, sizeof(struct spa_pod_struct), void), avail);
-	spa_ringbuffer_read_update(&impl->buffer, idx + avail);
+	*p = SPA_POD_INIT_Struct(total);
 
 	spa_list_for_each(resource, &impl->global->resource_list, link)
 		pw_profiler_resource_profile(resource, &p->pod);
 }
 
-static void context_do_profile(void *data, struct pw_impl_node *node)
+static void context_do_profile(void *data)
 {
-	struct impl *impl = data;
+	struct node *n = data;
+	struct pw_impl_node *node = n->node;
+	struct impl *impl = n->impl;
 	struct spa_pod_builder b;
 	struct spa_pod_frame f2;
 	uint32_t id = node->info.id;
@@ -180,13 +164,13 @@
 	if (SPA_FLAG_IS_SET(pos->clock.flags, SPA_IO_CLOCK_FLAG_FREEWHEEL))
 		return;
 
-	spa_pod_builder_init(&b, impl->tmp, sizeof(impl->tmp));
+	spa_pod_builder_init(&b, n->tmp, sizeof(n->tmp));
 	spa_pod_builder_push_object(&b, &f0,
 			SPA_TYPE_OBJECT_Profiler, 0);
 
 	spa_pod_builder_prop(&b, SPA_PROFILER_info, 0);
 	spa_pod_builder_add_struct(&b,
-			SPA_POD_Long(impl->count),
+			SPA_POD_Long(n->count),
 			SPA_POD_Float(a->cpu_load0),
 			SPA_POD_Float(a->cpu_load1),
 			SPA_POD_Float(a->cpu_load2),
@@ -253,42 +237,107 @@
 	}
 	spa_pod_builder_pop(&b, &f0);
 
-	if (b.state.offset > sizeof(impl->tmp))
+	if (b.state.offset > sizeof(n->tmp))
 		goto done;
 
-	filled = spa_ringbuffer_get_write_index(&impl->buffer, &idx);
-	if (filled < 0 || filled > MAX_BUFFER) {
+	filled = spa_ringbuffer_get_write_index(&n->buffer, &idx);
+	if (filled < 0 || filled > DATA_BUFFER) {
 		pw_log_warn("%p: queue xrun %d", impl, filled);
 		goto done;
 	}
-	avail = MAX_BUFFER - filled;
+	avail = DATA_BUFFER - filled;
 	if (avail < b.state.offset) {
 		pw_log_warn("%p: queue full %d < %d", impl, avail, b.state.offset);
 		goto done;
 	}
-	spa_ringbuffer_write_data(&impl->buffer,
-			impl->data, MAX_BUFFER,
-			idx % MAX_BUFFER,
+	spa_ringbuffer_write_data(&n->buffer,
+			n->data, DATA_BUFFER,
+			idx % DATA_BUFFER,
 			b.data, b.state.offset);
-	spa_ringbuffer_write_update(&impl->buffer, idx + b.state.offset);
+	spa_ringbuffer_write_update(&n->buffer, idx + b.state.offset);
 
-	if (!impl->flushing || filled + b.state.offset > MIN_FLUSH)
-		start_flush(impl);
+	pw_loop_signal_event(impl->main_loop, impl->flush_event);
 done:
-	impl->count++;
+	n->count++;
 }
 
-static const struct pw_context_driver_events context_events = {
-	PW_VERSION_CONTEXT_DRIVER_EVENTS,
-	.incomplete = context_do_profile,
+static struct pw_impl_node_rt_events node_rt_events = {
+	PW_VERSION_IMPL_NODE_RT_EVENTS,
 	.complete = context_do_profile,
+	.incomplete = context_do_profile,
+};
+
+static void enable_node_profiling(struct node *n, bool enabled)
+{
+	if (enabled && !n->enabled) {
+		SPA_FLAG_SET(n->node->rt.target.activation->flags, PW_NODE_ACTIVATION_FLAG_PROFILER);
+		pw_impl_node_add_rt_listener(n->node, &n->node_rt_listener, &node_rt_events, n);
+	} else if (!enabled && n->enabled) {
+		SPA_FLAG_CLEAR(n->node->rt.target.activation->flags, PW_NODE_ACTIVATION_FLAG_PROFILER);
+		pw_impl_node_remove_rt_listener(n->node, &n->node_rt_listener);
+	}
+	n->enabled = enabled;
+}
+
+static void enable_profiling(struct impl *impl, bool enabled)
+{
+	struct node *n;
+	spa_list_for_each(n, &impl->node_list, link)
+		enable_node_profiling(n, enabled);
+}
+
+static void context_driver_added(void *data, struct pw_impl_node *node)
+{
+	struct impl *impl = data;
+	struct node *n;
+
+	n = calloc(1, sizeof(*n));
+	if (n == NULL)
+		return;
+
+	n->impl = impl;
+	n->node = node;
+	spa_list_append(&impl->node_list, &n->link);
+	spa_ringbuffer_init(&n->buffer);
+
+	if (impl->busy > 0)
+		enable_node_profiling(n, true);
+}
+
+static struct node *find_node(struct impl *impl, struct pw_impl_node *node)
+{
+	struct node *n;
+	spa_list_for_each(n, &impl->node_list, link) {
+		if (n->node == node)
+			return n;
+	}
+	return NULL;
+}
+
+static void context_driver_removed(void *data, struct pw_impl_node *node)
+{
+	struct impl *impl = data;
+	struct node *n;
+
+	n = find_node(impl, node);
+	if (n == NULL)
+		return;
+
+	enable_node_profiling(n, false);
+	spa_list_remove(&n->link);
+	free(n);
+}
+
+static const struct pw_context_events context_events = {
+	PW_VERSION_CONTEXT_EVENTS,
+	.driver_added = context_driver_added,
+	.driver_removed = context_driver_removed,
 };
 
 static void stop_listener(struct impl *impl)
 {
 	if (impl->listening) {
-		pw_context_driver_remove_listener(impl->context,
-			&impl->context_listener);
+		enable_profiling(impl, false);
 		impl->listening = false;
 	}
 }
@@ -331,9 +380,7 @@
 
 	if (++impl->busy == 1) {
 		pw_log_info("%p: starting profiler", impl);
-		pw_context_driver_add_listener(impl->context,
-			&impl->context_listener,
-			&context_events, impl);
+		enable_profiling(impl, true);
 		impl->listening = true;
 	}
 	return 0;
@@ -346,11 +393,12 @@
 	if (impl->global != NULL)
 		pw_global_destroy(impl->global);
 
+	spa_hook_remove(&impl->context_listener);
 	spa_hook_remove(&impl->module_listener);
 
 	pw_properties_free(impl->properties);
 
-	pw_loop_destroy_source(impl->main_loop, impl->flush_timeout);
+	pw_loop_destroy_source(impl->main_loop, impl->flush_event);
 
 	free(impl);
 }
@@ -365,7 +413,6 @@
 	struct impl *impl = data;
 
 	stop_listener(impl);
-	stop_flush(impl);
 
 	spa_hook_remove(&impl->global_listener);
 	impl->global = NULL;
@@ -393,6 +440,7 @@
 	if (impl == NULL)
 		return -errno;
 
+	spa_list_init(&impl->node_list);
 	pw_protocol_native_ext_profiler_init(context);
 
 	pw_log_debug("module %p: new %s", impl, args);
@@ -407,8 +455,6 @@
 	impl->main_loop = pw_context_get_main_loop(impl->context);
 	impl->data_loop = pw_data_loop_get_loop(pw_context_get_data_loop(impl->context));
 
-	spa_ringbuffer_init(&impl->buffer);
-
 	impl->global = pw_global_new(context,
 			PW_TYPE_INTERFACE_Profiler,
 			PW_VERSION_PROFILER,
@@ -422,7 +468,7 @@
 	pw_properties_setf(impl->properties, PW_KEY_OBJECT_SERIAL, "%"PRIu64,
 			pw_global_get_serial(impl->global));
 
-	impl->flush_timeout = pw_loop_add_timer(impl->main_loop, flush_timeout, impl);
+	impl->flush_event = pw_loop_add_event(impl->main_loop, do_flush_event, impl);
 
 	pw_global_update_keys(impl->global, &impl->properties->dict, keys);
 
@@ -430,6 +476,10 @@
 
 	pw_impl_module_update_properties(module, &SPA_DICT_INIT_ARRAY(module_props));
 
+	pw_context_add_listener(impl->context,
+			&impl->context_listener,
+			&context_events, impl);
+
 	pw_global_register(impl->global);
 
 	pw_global_add_listener(impl->global, &impl->global_listener, &global_events, impl);

pipewire-0.3.74.tar.gz/src/modules/module-protocol-pulse.c -> pipewire-0.3.76.tar.gz/src/modules/module-protocol-pulse.c Changed

pipewire-0.3.74.tar.gz/src/modules/module-protocol-pulse/manager.c -> pipewire-0.3.76.tar.gz/src/modules/module-protocol-pulse/manager.c Changed

pipewire-0.3.74.tar.gz/src/modules/module-protocol-pulse/manager.h -> pipewire-0.3.76.tar.gz/src/modules/module-protocol-pulse/manager.h Changed

pipewire-0.3.74.tar.gz/src/modules/module-protocol-pulse/modules/module-switch-on-connect.c -> pipewire-0.3.76.tar.gz/src/modules/module-protocol-pulse/modules/module-switch-on-connect.c Changed

pipewire-0.3.74.tar.gz/src/modules/module-protocol-pulse/modules/module-zeroconf-publish.c -> pipewire-0.3.76.tar.gz/src/modules/module-protocol-pulse/modules/module-zeroconf-publish.c Changed

pipewire-0.3.74.tar.gz/src/modules/module-protocol-pulse/pulse-server.c -> pipewire-0.3.76.tar.gz/src/modules/module-protocol-pulse/pulse-server.c Changed

pipewire-0.3.74.tar.gz/src/modules/module-raop-discover.c -> pipewire-0.3.76.tar.gz/src/modules/module-raop-discover.c Changed

pipewire-0.3.74.tar.gz/src/modules/module-raop-sink.c -> pipewire-0.3.76.tar.gz/src/modules/module-raop-sink.c Changed

@@ -124,7 +124,7 @@
 #define FRAMES_PER_UDP_PACKET 352
 
 #define RAOP_LATENCY_MIN	11025u
-#define DEFAULT_LATENCY_MS	"1000"
+#define DEFAULT_LATENCY_MS	"1500"
 
 #define DEFAULT_TCP_AUDIO_PORT   6000
 #define DEFAULT_UDP_AUDIO_PORT   6000
@@ -318,7 +318,7 @@
 		pkt0 |= htonl(0x10000000);
 	pkt1 = htonl(rtptime - latency);
 	transmitted = ntp_now();
-	pkt2 = htonl((transmitted >> 32) & 0x0000ffff);
+	pkt2 = htonl(transmitted >> 32);
 	pkt3 = htonl(transmitted & 0xffffffff);
 	pkt4 = htonl(rtptime);
 
@@ -827,7 +827,7 @@
 
 static int rtsp_send(struct impl *impl, const char *method,
 		const char *content_type, const char *content,
-		int (*reply) (void *data, int status, const struct spa_dict *headers))
+		int (*reply) (void *data, int status, const struct spa_dict *headers, const struct pw_array *content))
 {
 	int res;
 
@@ -838,9 +838,9 @@
 	return res;
 }
 
-static int rtsp_flush_reply(void *data, int status, const struct spa_dict *headers)
+static int rtsp_log_reply_status(void *data, int status, const struct spa_dict *headers, const struct pw_array *content)
 {
-	pw_log_info("reply %d", status);
+	pw_log_info("reply status: %d", status);
 	return 0;
 }
 
@@ -857,7 +857,7 @@
 
 	impl->recording = false;
 
-	res = rtsp_send(impl, "FLUSH", NULL, NULL, rtsp_flush_reply);
+	res = rtsp_send(impl, "FLUSH", NULL, NULL, rtsp_log_reply_status);
 
 	pw_properties_set(impl->headers, "Range", NULL);
 	pw_properties_set(impl->headers, "RTP-Info", NULL);
@@ -873,10 +873,10 @@
 	char header128, volstr64;
 	snprintf(header, sizeof(header), "volume: %s\r\n",
 			spa_dtoa(volstr, sizeof(volstr), impl->volume));
-	return rtsp_send(impl, "SET_PARAMETER", "text/parameters", header, NULL);
+	return rtsp_send(impl, "SET_PARAMETER", "text/parameters", header, rtsp_log_reply_status);
 }
 
-static int rtsp_record_reply(void *data, int status, const struct spa_dict *headers)
+static int rtsp_record_reply(void *data, int status, const struct spa_dict *headers, const struct pw_array *content)
 {
 	struct impl *impl = data;
 	const char *str;
@@ -887,7 +887,7 @@
 	struct spa_latency_info latency;
 	char progress128;
 
-	pw_log_info("reply %d", status);
+	pw_log_info("record status: %d", status);
 
 	if ((str = spa_dict_lookup(headers, "Audio-Latency")) != NULL) {
 		uint32_t l;
@@ -913,7 +913,7 @@
 	rtsp_send_volume(impl);
 
 	snprintf(progress, sizeof(progress), "progress: %s/%s/%s\r\n", "0", "0", "0");
-	return rtsp_send(impl, "SET_PARAMETER", "text/parameters", progress, NULL);
+	return rtsp_send(impl, "SET_PARAMETER", "text/parameters", progress, rtsp_log_reply_status);
 }
 
 static int rtsp_do_record(struct impl *impl)
@@ -966,7 +966,7 @@
 	pw_loop_update_io(impl->loop, impl->server_source, 0);
 }
 
-static int rtsp_setup_reply(void *data, int status, const struct spa_dict *headers)
+static int rtsp_setup_reply(void *data, int status, const struct spa_dict *headers, const struct pw_array *content)
 {
 	struct impl *impl = data;
 	const char *str, *state = NULL, *s;
@@ -975,7 +975,7 @@
 	uint16_t control_port, timing_port;
 	int res;
 
-	pw_log_info("reply %d", status);
+	pw_log_info("setup status: %d", status);
 
 	if ((str = spa_dict_lookup(headers, "Session")) == NULL) {
 		pw_log_error("missing Session header");
@@ -1105,11 +1105,11 @@
 	return -EIO;
 }
 
-static int rtsp_announce_reply(void *data, int status, const struct spa_dict *headers)
+static int rtsp_announce_reply(void *data, int status, const struct spa_dict *headers, const struct pw_array *content)
 {
 	struct impl *impl = data;
 
-	pw_log_info("reply %d", status);
+	pw_log_info("announce status: %d", status);
 
 	pw_properties_set(impl->headers, "Apple-Challenge", NULL);
 
@@ -1294,11 +1294,11 @@
 	return rtsp_send(impl, "ANNOUNCE", "application/sdp", sdp, rtsp_announce_reply);
 }
 
-static int rtsp_auth_setup_reply(void *data, int status, const struct spa_dict *headers)
+static int rtsp_auth_setup_reply(void *data, int status, const struct spa_dict *headers, const struct pw_array *content)
 {
 	struct impl *impl = data;
 
-	pw_log_info("reply %d", status);
+	pw_log_info("auth-setup status: %d", status);
 
 	return rtsp_do_announce(impl);
 }
@@ -1315,12 +1315,12 @@
 				       rtsp_auth_setup_reply, impl);
 }
 
-static int rtsp_auth_reply(void *data, int status, const struct spa_dict *headers)
+static int rtsp_auth_reply(void *data, int status, const struct spa_dict *headers, const struct pw_array *content)
 {
 	struct impl *impl = data;
 	int res = 0;
 
-	pw_log_info("auth %d", status);
+	pw_log_info("auth status: %d", status);
 
 	switch (status) {
 	case 200:
@@ -1385,12 +1385,12 @@
 	return rtsp_send(impl, "OPTIONS", NULL, NULL, rtsp_auth_reply);
 }
 
-static int rtsp_options_reply(void *data, int status, const struct spa_dict *headers)
+static int rtsp_options_reply(void *data, int status, const struct spa_dict *headers, const struct pw_array *content)
 {
 	struct impl *impl = data;
 	int res = 0;
 
-	pw_log_info("options %d", status);
+	pw_log_info("options status: %d", status);
 
 	switch (status) {
 	case 401:
@@ -1547,12 +1547,12 @@
 	return pw_rtsp_client_connect(impl->rtsp, hostname, atoi(port), impl->session_id);
 }
 
-static int rtsp_teardown_reply(void *data, int status, const struct spa_dict *headers)
+static int rtsp_teardown_reply(void *data, int status, const struct spa_dict *headers, const struct pw_array *content)
 {
 	struct impl *impl = data;
 	const char *str;
 
-	pw_log_info("reply");
+	pw_log_info("teardown status: %d", status);
 
 	connection_cleanup(impl);

pipewire-0.3.74.tar.gz/src/modules/module-raop/rtsp-client.c -> pipewire-0.3.76.tar.gz/src/modules/module-raop/rtsp-client.c Changed

@@ -25,7 +25,7 @@
 	size_t len;
 	size_t offset;
 	uint32_t cseq;
-	int (*reply) (void *user_data, int status, const struct spa_dict *headers);
+	int (*reply) (void *user_data, int status, const struct spa_dict *headers, const struct pw_array *content);
 	void *user_data;
 };
 
@@ -60,6 +60,7 @@
 	char line_buf1024;
 	size_t line_pos;
 	struct pw_properties *headers;
+	struct pw_array content;
 	size_t content_length;
 
 	uint32_t cseq;
@@ -89,6 +90,7 @@
 	spa_list_init(&client->pending);
 	spa_hook_list_init(&client->listener_list);
 	client->headers = pw_properties_new(NULL, NULL);
+	pw_array_init(&client->content, 4096);
 	client->recv_state = CLIENT_RECV_NONE;
 
 	pw_log_info("new client %p", client);
@@ -105,6 +107,7 @@
 	pw_properties_free(client->headers);
 	pw_properties_free(client->props);
 	spa_hook_list_clean(&client->listener_list);
+	pw_array_clear(&client->content);
 	free(client);
 }
 
@@ -277,7 +280,7 @@
 
 	msg = find_pending(client, cseq);
 	if (msg) {
-		res = msg->reply(msg->user_data, client->status, &client->headers->dict);
+		res = msg->reply(msg->user_data, client->status, &client->headers->dict, &client->content);
 		spa_list_remove(&msg->link);
 		free(msg);
 
@@ -288,6 +291,8 @@
 	else {
 		pw_rtsp_client_emit_message(client, client->status, &client->headers->dict);
 	}
+
+	pw_array_reset(&client->content);
 }
 
 static void process_received_message(struct pw_rtsp_client *client)
@@ -328,7 +333,7 @@
 
 static int process_content(struct pw_rtsp_client *client)
 {
-	char buf1024;
+	uint8_t buf4096;
 
 	while (client->content_length > 0) {
 		const size_t max_recv = SPA_MIN(sizeof(buf), client->content_length);
@@ -345,6 +350,9 @@
 			return res;
 		}
 
+		void *p = pw_array_add(&client->content, res);
+		memcpy(p, buf, res);
+
 		spa_assert((size_t) res <= client->content_length);
 		client->content_length -= res;
 	}
@@ -556,7 +564,7 @@
 int pw_rtsp_client_url_send(struct pw_rtsp_client *client, const char *url,
 		const char *cmd, const struct spa_dict *headers,
 		const char *content_type, const void *content, size_t content_length,
-		int (*reply) (void *user_data, int status, const struct spa_dict *headers),
+		int (*reply) (void *user_data, int status, const struct spa_dict *headers, const struct pw_array *content),
 		void *user_data)
 {
 	FILE *f;
@@ -610,7 +618,7 @@
 int pw_rtsp_client_send(struct pw_rtsp_client *client,
 		const char *cmd, const struct spa_dict *headers,
 		const char *content_type, const char *content,
-		int (*reply) (void *user_data, int status, const struct spa_dict *headers),
+		int (*reply) (void *user_data, int status, const struct spa_dict *headers, const struct pw_array *content),
 		void *user_data)
 {
 	const size_t content_length = content ? strlen(content) : 0;

pipewire-0.3.74.tar.gz/src/modules/module-raop/rtsp-client.h -> pipewire-0.3.76.tar.gz/src/modules/module-raop/rtsp-client.h Changed

pipewire-0.3.74.tar.gz/src/modules/module-rt.c -> pipewire-0.3.76.tar.gz/src/modules/module-rt.c Changed

@@ -173,6 +173,8 @@
 
 	struct spa_thread_utils thread_utils;
 
+	pid_t main_pid;
+	struct rlimit rl;
 	int nice_level;
 	int rt_prio;
 	rlim_t rt_time_soft;
@@ -191,6 +193,10 @@
 	const char* object_path;
 	const char* interface;
 	struct pw_rtkit_bus *rtkit_bus;
+	struct pw_thread_loop *thread_loop;
+	int max_rtprio;
+	int min_nice_level;
+	rlim_t rttime_max;
 
 	/* These are only for the RTKit implementation to fill in the `thread`
 	 * struct. Since there's barely any overhead here we'll do this
@@ -383,41 +389,12 @@
 	return ret;
 }
 
-static int pw_rtkit_get_max_realtime_priority(struct impl *impl)
-{
-	long long retval;
-	int err;
-
-	err = rtkit_get_int_property(impl, "MaxRealtimePriority", &retval);
-	return err < 0 ? err : retval;
-}
-
-static int pw_rtkit_get_min_nice_level(struct impl *impl, int *min_nice_level)
-{
-	long long retval;
-	int err;
-
-	err = rtkit_get_int_property(impl, "MinNiceLevel", &retval);
-	if (err >= 0)
-		*min_nice_level = retval;
-	return err;
-}
-
-static long long pw_rtkit_get_rttime_usec_max(struct impl *impl)
-{
-	long long retval;
-	int err;
-
-	err = rtkit_get_int_property(impl, "RTTimeUSecMax", &retval);
-	return err < 0 ? err : retval;
-}
-
 static int pw_rtkit_make_realtime(struct impl *impl, pid_t thread, int priority)
 {
-	DBusMessage *m = NULL, *r = NULL;
+	DBusMessage *m = NULL;
 	dbus_uint64_t pid;
 	dbus_uint64_t u64;
-	dbus_uint32_t u32;
+	dbus_uint32_t u32, serial;
 	DBusError error;
 	int ret;
 	struct pw_rtkit_bus *connection = impl->rtkit_bus;
@@ -446,17 +423,10 @@
 		goto finish;
 	}
 
-	if (!(r = dbus_connection_send_with_reply_and_block(connection->bus, m, -1, &error))) {
+	if (!dbus_connection_send(connection->bus, m, &serial)) {
 		ret = translate_error(error.name);
 		goto finish;
 	}
-
-
-	if (dbus_set_error_from_message(&error, r)) {
-		ret = translate_error(error.name);
-		goto finish;
-	}
-
 	ret = 0;
 
 finish:
@@ -464,26 +434,20 @@
 	if (m)
 		dbus_message_unref(m);
 
-	if (r)
-		dbus_message_unref(r);
-
-	dbus_error_free(&error);
-
 	return ret;
 }
 
+
 static int pw_rtkit_make_high_priority(struct impl *impl, pid_t thread, int nice_level)
 {
-	DBusMessage *m = NULL, *r = NULL;
+	DBusMessage *m = NULL;
 	dbus_uint64_t pid;
 	dbus_uint64_t u64;
 	dbus_int32_t s32;
-	DBusError error;
+	dbus_uint32_t serial;
 	int ret;
 	struct pw_rtkit_bus *connection = impl->rtkit_bus;
 
-	dbus_error_init(&error);
-
 	if (thread == 0)
 		thread = _gettid();
 
@@ -505,20 +469,10 @@
 		ret = -ENOMEM;
 		goto finish;
 	}
-
-
-
-	if (!(r = dbus_connection_send_with_reply_and_block(connection->bus, m, -1, &error))) {
-		ret = translate_error(error.name);
-		goto finish;
-	}
-
-
-	if (dbus_set_error_from_message(&error, r)) {
-		ret = translate_error(error.name);
+	if (!dbus_connection_send(connection->bus, m, &serial)) {
+		ret = -EIO;
 		goto finish;
 	}
-
 	ret = 0;
 
 finish:
@@ -526,11 +480,6 @@
 	if (m)
 		dbus_message_unref(m);
 
-	if (r)
-		dbus_message_unref(r);
-
-	dbus_error_free(&error);
-
 	return ret;
 }
 #endif /* HAVE_DBUS */
@@ -542,6 +491,8 @@
 	spa_hook_remove(&impl->module_listener);
 
 #ifdef HAVE_DBUS
+	if (impl->thread_loop)
+		pw_thread_loop_destroy(impl->thread_loop);
 	if (impl->rtkit_bus)
 		pw_rtkit_bus_free(impl->rtkit_bus);
 #endif
@@ -637,9 +588,9 @@
 	return false;
 }
 
-static int sched_set_nice(int nice_level)
+static int sched_set_nice(pid_t pid, int nice_level)
 {
-	if (setpriority(PRIO_PROCESS, _gettid(), nice_level) == 0)
+	if (setpriority(PRIO_PROCESS, pid, nice_level) == 0)
 		return 0;
 	else
 		return -errno;
@@ -651,30 +602,27 @@
 
 #ifdef HAVE_DBUS
 	if (impl->use_rtkit) {
-		int min_nice = nice_level;
-		pw_rtkit_get_min_nice_level(impl, &min_nice);
-		if (nice_level < min_nice) {
+		if (nice_level < impl->min_nice_level) {
 			pw_log_info("clamped nice level %d to %d",
-					nice_level, min_nice);
-			nice_level = min_nice;
+					nice_level, impl->min_nice_level);
+			nice_level = impl->min_nice_level;
 		}
-		res = pw_rtkit_make_high_priority(impl, 0, nice_level);
+		res = pw_rtkit_make_high_priority(impl, impl->main_pid, nice_level);
 	}
-	else if (impl->rlimits_enabled)
-		res = sched_set_nice(nice_level);
 	else
-		res = -ENOTSUP;
-#else
+#endif
 	if (impl->rlimits_enabled)
-		res = sched_set_nice(nice_level);
+		res = sched_set_nice(impl->main_pid, nice_level);
 	else
 		res = -ENOTSUP;
-#endif
 
 	if (res < 0) {
 		if (warn)
 			pw_log_warn("could not set nice-level to %d: %s",
 					nice_level, spa_strerror(res));
+	} else if (res > 0) {
+		pw_log_info("main thread setting nice level to %d: %s",
+				nice_level, spa_strerror(-res));
 	} else {
 		pw_log_info("main thread nice level set to %d",
 				nice_level);
@@ -684,37 +632,16 @@
 
 static int set_rlimit(struct impl *impl)
 {
-	struct rlimit rl;
 	int res = 0;
 
-	spa_zero(rl);
-	rl.rlim_cur = impl->rt_time_soft;
-	rl.rlim_max = impl->rt_time_hard;
-
-#ifdef HAVE_DBUS
-	if (impl->use_rtkit) {
-		long long rttime;
-		rttime = pw_rtkit_get_rttime_usec_max(impl);
-		if (rttime >= 0) {
-			if ((rlim_t)rttime < rl.rlim_cur) {
-				pw_log_debug("clamping rt.time.soft from %llu to %lld because of RTKit",
-					     (long long)rl.rlim_cur, rttime);
-			}
-
-			rl.rlim_cur = SPA_MIN(rl.rlim_cur, (rlim_t)rttime);
-			rl.rlim_max = SPA_MIN(rl.rlim_max, (rlim_t)rttime);
-		}
-	}
-#endif
-
-	if (setrlimit(RLIMIT_RTTIME, &rl) < 0)
+	if (setrlimit(RLIMIT_RTTIME, &impl->rl) < 0)
 		res = -errno;
 
 	if (res < 0)
 		pw_log_debug("setrlimit() failed: %s", spa_strerror(res));
 	else
 		pw_log_debug("rt.time.soft:%"PRIi64" rt.time.hard:%"PRIi64,
-				(int64_t)rl.rlim_cur, (int64_t)rl.rlim_max);
+				(int64_t)impl->rl.rlim_cur, (int64_t)impl->rl.rlim_max);
 
 	return res;
 }
@@ -840,8 +767,7 @@
 	if (min)
 		*min = 1;
 	if (max) {
-		if ((*max = pw_rtkit_get_max_realtime_priority(impl)) < 0)
-			return *max;
+		*max = impl->max_rtprio;
 		if (*max < 1)
 			*max = 1;
 	}
@@ -860,62 +786,72 @@
 	return res;
 }
 
-static pid_t impl_gettid(struct impl *impl, pthread_t pt)
-{
-	struct thread *thr;
+struct rt_params {
 	pid_t pid;
+	int priority;
+};
 
-	pthread_mutex_lock(&impl->lock);
-	if ((thr = find_thread_by_pt(impl, pt)) != NULL)
-		pid = thr->pid;
-	else
-		pid = _gettid();
-	pthread_mutex_unlock(&impl->lock);
+static int do_make_realtime(struct spa_loop *loop, bool async, uint32_t seq,
+		const void *data, size_t size, void *user_data)
+{
+	struct impl *impl = user_data;
+	const struct rt_params *params = data;
+	int err, min, max, priority = params->priority;
+	pid_t pid = params->pid;
+
+	pw_log_debug("rtkit realtime");
+
+	if ((err = get_rtkit_priority_range(impl, &min, &max)) < 0)
+		return err;
+
+	if (priority < min || priority > max) {
+		pw_log_info("clamping requested priority %d for thread %d "
+				"between %d  and %d", priority, pid, min, max);
+		priority = SPA_CLAMP(priority, min, max);
+	}
+
+	if ((err = pw_rtkit_make_realtime(impl, pid, priority)) < 0) {
+		pw_log_warn("could not make thread %d realtime using RTKit: %s", pid, spa_strerror(err));
+		return err;
+	}
 
-	return pid;
+	pw_log_info("acquired realtime priority %d for thread %d using RTKit", priority, pid);
+	return 0;
 }
 
 static int impl_acquire_rt(void *object, struct spa_thread *thread, int priority)
 {
 	struct impl *impl = object;
 	struct sched_param sp;
-	int err;
 	pthread_t pt = (pthread_t)thread;
-	pid_t pid;
+	int res;
 
 	/* See the docstring on `spa_thread_utils_methods::acquire_rt` */
 	if (priority == -1) {
 		priority = impl->rt_prio;
 	}
-
 	if (impl->use_rtkit) {
-		int min, max;
-
-		if ((err = get_rtkit_priority_range(impl, &min, &max)) < 0)
-			return err;
-
-		pid = impl_gettid(impl, pt);
-
-		if (priority < min || priority > max) {
-			pw_log_info("clamping requested priority %d for thread %d "
-					"between %d  and %d", priority, pid, min, max);
-			priority = SPA_CLAMP(priority, min, max);
-		}
+		struct rt_params params;
+		struct thread *thr;
 
 		spa_zero(sp);
-		sp.sched_priority = priority;
-
 		if (pthread_setschedparam(pt, SCHED_OTHER | PW_SCHED_RESET_ON_FORK, &sp) == 0) {
 			pw_log_debug("SCHED_OTHER|SCHED_RESET_ON_FORK worked.");
 		}
 
-		if ((err = pw_rtkit_make_realtime(impl, pid, priority)) < 0) {
-			pw_log_warn("could not make thread %d realtime using RTKit: %s", pid, spa_strerror(err));
-			return err;
-		}
+		params.priority = priority;
 
-		pw_log_info("acquired realtime priority %d for thread %d using RTKit", priority, pid);
-		return 0;
+		pthread_mutex_lock(&impl->lock);
+		if ((thr = find_thread_by_pt(impl, pt)) != NULL)
+			params.pid = thr->pid;
+		else
+			params.pid = _gettid();
+
+		res = pw_loop_invoke(pw_thread_loop_get_loop(impl->thread_loop),
+				do_make_realtime, 0, &params, sizeof(params), false, impl);
+		pthread_mutex_unlock(&impl->lock);
+
+		return res;
 	} else {
 		return acquire_rt_sched(thread, priority);
 	}
@@ -986,6 +922,93 @@
 
 	return 0;
 }
+
+static int rtkit_get_bus(struct impl *impl)
+{
+	int res;
+
+	pw_log_debug("enter rtkit get bus");
+
+	/* Checking xdg-desktop-portal. It works fine in all situations. */
+	if (impl->rtportal_enabled)
+		impl->rtkit_bus = pw_rtkit_bus_get_session();
+	else
+		pw_log_info("Portal Realtime disabled");
+
+	if (impl->rtkit_bus != NULL) {
+		if (pw_rtkit_check_xdg_portal(impl->rtkit_bus)) {
+			impl->service_name = XDG_PORTAL_SERVICE_NAME;
+			impl->object_path = XDG_PORTAL_OBJECT_PATH;
+			impl->interface = XDG_PORTAL_INTERFACE;
+		} else {
+			pw_log_info("found session bus but no portal, trying RTKit fallback");
+			pw_rtkit_bus_free(impl->rtkit_bus);
+			impl->rtkit_bus = NULL;
+		}
+	}
+	/* Failed to get xdg-desktop-portal, try to use rtkit. */
+	if (impl->rtkit_bus == NULL) {
+		if (impl->rtkit_enabled)
+			impl->rtkit_bus = pw_rtkit_bus_get_system();
+		else
+			pw_log_info("RTkit disabled");
+
+		if (impl->rtkit_bus != NULL) {
+			impl->service_name = RTKIT_SERVICE_NAME;
+			impl->object_path = RTKIT_OBJECT_PATH;
+			impl->interface = RTKIT_INTERFACE;
+		} else {
+			res = -errno;
+			pw_log_warn("Realtime scheduling disabled: unsufficient realtime privileges, "
+				"Portal not found on session bus, and no system bus for RTKit: %m");
+			return res;
+		}
+	}
+
+	return 0;
+}
+
+static int do_rtkit_setup(struct spa_loop *loop, bool async, uint32_t seq,
+		const void *data, size_t size, void *user_data)
+{
+	struct impl *impl = user_data;
+	long long retval;
+
+	pw_log_debug("enter rtkit setup");
+
+	/* get some properties */
+	if (rtkit_get_int_property(impl, "MaxRealtimePriority", &retval) < 0) {
+		retval = 1;
+		pw_log_warn("RTKit does not give us MaxRealtimePriority, using %lld", retval);
+	}
+	impl->max_rtprio = retval;
+	if (rtkit_get_int_property(impl, "MinNiceLevel", &retval) < 0) {
+		retval = 0;
+		pw_log_warn("RTKit does not give us MinNiceLevel, using %lld", retval);
+	}
+	impl->min_nice_level = retval;
+	if (rtkit_get_int_property(impl, "RTTimeUSecMax", &retval) < 0) {
+		retval = impl->rl.rlim_cur;
+		pw_log_warn("RTKit does not give us RTTimeUSecMax, using %lld", retval);
+	}
+	impl->rttime_max = retval;
+
+	/* Retry set_nice with rtkit */
+	if (IS_VALID_NICE_LEVEL(impl->nice_level))
+		set_nice(impl, impl->nice_level, true);
+
+	/* Set rlimit with rtkit limits */
+	if (impl->rttime_max < impl->rl.rlim_cur) {
+		pw_log_debug("clamping rt.time.soft from %llu to %lld because of RTKit",
+			     (long long)impl->rl.rlim_cur, (long long)impl->rttime_max);
+	}
+	impl->rl.rlim_cur = SPA_MIN(impl->rl.rlim_cur, impl->rttime_max);
+	impl->rl.rlim_max = SPA_MIN(impl->rl.rlim_max, impl->rttime_max);
+
+	set_rlimit(impl);
+
+	return 0;
+}
 #endif /* HAVE_DBUS */
 
 SPA_EXPORT
@@ -1019,6 +1042,10 @@
 	impl->rtportal_enabled = pw_properties_get_bool(props, "rtportal.enabled", true);
 	impl->rtkit_enabled = pw_properties_get_bool(props, "rtkit.enabled", true);
 
+	impl->rl.rlim_cur = impl->rt_time_soft;
+	impl->rl.rlim_max = impl->rt_time_hard;
+	impl->main_pid = _gettid();
+
 	bool can_use_rtkit = false, use_rtkit = false;
 
 	if (!IS_VALID_NICE_LEVEL(impl->nice_level)) {
@@ -1034,6 +1061,7 @@
 
 	if ((res = check_rtkit(impl, context, &can_use_rtkit)) < 0)
 		goto error;
+
 #endif
 	/* If the user has permissions to use regular realtime scheduling, as well as
 	 * the nice level we want, then we'll use that instead of RTKit */
@@ -1051,50 +1079,39 @@
 		if (set_nice(impl, impl->nice_level, !can_use_rtkit) < 0)
 			use_rtkit = can_use_rtkit;
 	}
+	if (!use_rtkit)
+		set_rlimit(impl);
 
 #ifdef HAVE_DBUS
 	impl->use_rtkit = use_rtkit;
 	if (impl->use_rtkit) {
-		/* Checking xdg-desktop-portal. It works fine in all situations. */
-		if (impl->rtportal_enabled)
-			impl->rtkit_bus = pw_rtkit_bus_get_session();
-		else
-			pw_log_info("Portal Realtime disabled");
-		if (impl->rtkit_bus != NULL) {
-			if (pw_rtkit_check_xdg_portal(impl->rtkit_bus)) {
-				impl->service_name = XDG_PORTAL_SERVICE_NAME;
-				impl->object_path = XDG_PORTAL_OBJECT_PATH;
-				impl->interface = XDG_PORTAL_INTERFACE;
-			} else {
-				pw_log_info("found session bus but no portal, trying RTKit fallback");
-				pw_rtkit_bus_free(impl->rtkit_bus);
-				impl->rtkit_bus = NULL;
-			}
-		}
-		/* Failed to get xdg-desktop-portal, try to use rtkit. */
-		if (impl->rtkit_bus == NULL) {
-			if (impl->rtkit_enabled)
-				impl->rtkit_bus = pw_rtkit_bus_get_system();
-			else
-				pw_log_info("RTkit disabled");
-
-			if (impl->rtkit_bus != NULL) {
-				impl->service_name = RTKIT_SERVICE_NAME;
-				impl->object_path = RTKIT_OBJECT_PATH;
-				impl->interface = RTKIT_INTERFACE;
-			} else {
-				res = -errno;
-				pw_log_warn("Realtime scheduling disabled: unsufficient realtime privileges, "
-					"Portal not found on session bus, and no system bus for RTKit: %m");
-				goto error;
-			}
+		struct spa_dict_item items = {
+			{ "thread-loop.start-signal", "true" }
+		};
+		if ((res = rtkit_get_bus(impl)) < 0)
+			goto error;
+
+		impl->thread_loop = pw_thread_loop_new("module-rt",
+			&SPA_DICT_INIT_ARRAY(items));
+		if (impl->thread_loop == NULL) {
+			res = -errno;
+			goto error;
 		}
-		/* Retry set_nice with rtkit */
-		if (IS_VALID_NICE_LEVEL(impl->nice_level))
-			set_nice(impl, impl->nice_level, true);
+		pw_thread_loop_lock(impl->thread_loop);
+		pw_thread_loop_start(impl->thread_loop);
+		pw_thread_loop_wait(impl->thread_loop);
+		pw_thread_loop_unlock(impl->thread_loop);
+
+		pw_loop_invoke(pw_thread_loop_get_loop(impl->thread_loop),
+				do_rtkit_setup, 0, NULL, 0, false, impl);
+
+		pw_log_debug("initialized using RTKit");
+	} else {
+		pw_log_debug("initialized using regular realtime scheduling");
 	}
+#else
+	pw_log_debug("initialized using regular realtime scheduling");
 #endif
-	set_rlimit(impl);
 
 	impl->thread_utils.iface = SPA_INTERFACE_INIT(
 			SPA_TYPE_INTERFACE_ThreadUtils,
@@ -1109,20 +1126,12 @@
 	pw_impl_module_update_properties(module, &SPA_DICT_INIT_ARRAY(module_props));
 	pw_impl_module_update_properties(module, &props->dict);
 
-#ifdef HAVE_DBUS
-	if (impl->use_rtkit) {
-		pw_log_debug("initialized using RTKit");
-	} else {
-		pw_log_debug("initialized using regular realtime scheduling");
-	}
-#else
-	pw_log_debug("initialized using regular realtime scheduling");
-#endif
-
 	goto done;
 
 error:
 #ifdef HAVE_DBUS
+	if (impl->thread_loop)
+		pw_thread_loop_destroy(impl->thread_loop);
 	if (impl->rtkit_bus)
 		pw_rtkit_bus_free(impl->rtkit_bus);
 #endif

pipewire-0.3.76.tar.gz/src/modules/module-vban Added

pipewire-0.3.76.tar.gz/src/modules/module-vban-recv.c Added

@@ -0,0 +1,572 @@
+/* PipeWire */
+/* SPDX-FileCopyrightText: Copyright © 2023 Wim Taymans <wim.taymans@gmail.com> */
+/* SPDX-License-Identifier: MIT */
+
+#include "config.h"
+
+#include <limits.h>
+#include <string.h>
+#include <unistd.h>
+#include <sys/stat.h>
+#include <sys/socket.h>
+#include <sys/ioctl.h>
+#include <arpa/inet.h>
+#include <netinet/in.h>
+#include <net/if.h>
+#include <ctype.h>
+
+#include <spa/utils/hook.h>
+#include <spa/utils/result.h>
+#include <spa/utils/ringbuffer.h>
+#include <spa/utils/defs.h>
+#include <spa/utils/dll.h>
+#include <spa/utils/json.h>
+#include <spa/param/audio/format-utils.h>
+#include <spa/control/control.h>
+#include <spa/debug/types.h>
+#include <spa/debug/mem.h>
+
+#include <pipewire/pipewire.h>
+#include <pipewire/impl.h>
+
+#include <module-vban/stream.h>
+
+#ifdef __FreeBSD__
+#define ifr_ifindex ifr_index
+#endif
+
+/** \page page_module_vban_recv PipeWire Module: VBAN receiver
+ *
+ * The `vban-recv` module creates a PipeWire source that receives audio
+ * and midi VBAN(https://vb-audio.com) packets.
+ *
+ * ## Module Options
+ *
+ * Options specific to the behavior of this module
+ *
+ * - `local.ifname = <str>`: interface name to use
+ * - `source.ip = <str>`: the source ip address, default 127.0.0.1
+ * - `source.port = <int>`: the source port
+ * - `node.always-process = <bool>`: true to receive even when not running
+ * - `sess.latency.msec = <str>`: target network latency in milliseconds, default 100
+ * - `sess.ignore-ssrc = <bool>`: ignore SSRC, default false
+ * - `sess.media = <string>`: the media type audio|midi|opus, default audio
+ * - `stream.props = {}`: properties to be passed to the stream
+ *
+ * ## General options
+ *
+ * Options with well-known behavior:
+ *
+ * - \ref PW_KEY_REMOTE_NAME
+ * - \ref PW_KEY_AUDIO_FORMAT
+ * - \ref PW_KEY_AUDIO_RATE
+ * - \ref PW_KEY_AUDIO_CHANNELS
+ * - \ref SPA_KEY_AUDIO_POSITION
+ * - \ref PW_KEY_MEDIA_NAME
+ * - \ref PW_KEY_MEDIA_CLASS
+ * - \ref PW_KEY_NODE_NAME
+ * - \ref PW_KEY_NODE_DESCRIPTION
+ * - \ref PW_KEY_NODE_GROUP
+ * - \ref PW_KEY_NODE_LATENCY
+ * - \ref PW_KEY_NODE_VIRTUAL
+ *
+ * ## Example configuration
+ *\code{.unparsed}
+ * context.modules = 
+ * {   name = libpipewire-module-vban-recv
+ *     args = {
+ *         #local.ifname = eth0
+ *         #source.ip = 127.0.0.1
+ *         #source.port = 6980
+ *         sess.latency.msec = 100
+ *         #sess.ignore-ssrc = false
+ *         #node.always-process = false
+ *         #sess.media = "audio"
+ *         #audio.format = "S16LE"
+ *         #audio.rate = 44100
+ *         #audio.channels = 2
+ *         #audio.position =  FL FR 
+ *         stream.props = {
+ *            #media.class = "Audio/Source"
+ *            node.name = "vban-receiver"
+ *         }
+ *     }
+ * }
+ * 
+ *\endcode
+ *
+ * \since 0.3.76
+ */
+
+#define NAME "vban-recv"
+
+PW_LOG_TOPIC_STATIC(mod_topic, "mod." NAME);
+#define PW_LOG_TOPIC_DEFAULT mod_topic
+
+#define DEFAULT_CLEANUP_SEC		60
+#define DEFAULT_SOURCE_IP		"127.0.0.1"
+#define DEFAULT_SOURCE_PORT		6980
+
+#define USAGE   "( local.ifname=<local interface name to use> ) "						\
+		"( source.ip=<source IP address, default:"DEFAULT_SOURCE_IP"> ) "				\
+ 		"( source.port=<int, source port, default:"SPA_STRINGIFY(DEFAULT_SOURCE_PORT)"> "		\
+		"( sess.latency.msec=<target network latency, default "SPA_STRINGIFY(DEFAULT_SESS_LATENCY)"> ) "\
+ 		"( sess.media=<string, the media type audio|midi, default audio> ) "				\
+		"( audio.format=<format, default:"DEFAULT_FORMAT"> ) "						\
+		"( audio.rate=<sample rate, default:"SPA_STRINGIFY(DEFAULT_RATE)"> ) "				\
+		"( audio.channels=<number of channels, default:"SPA_STRINGIFY(DEFAULT_CHANNELS)"> ) "		\
+		"( audio.position=<channel map, default:"DEFAULT_POSITION"> ) "					\
+		"( stream.props= { key=value ... } ) "
+
+static const struct spa_dict_item module_info = {
+	{ PW_KEY_MODULE_AUTHOR, "Wim Taymans <wim.taymans@gmail.com>" },
+	{ PW_KEY_MODULE_DESCRIPTION, "VBAN Receiver" },
+	{ PW_KEY_MODULE_USAGE,	USAGE },
+	{ PW_KEY_MODULE_VERSION, PACKAGE_VERSION },
+};
+
+struct impl {
+	struct pw_impl_module *module;
+	struct spa_hook module_listener;
+	struct pw_properties *props;
+	struct pw_context *context;
+
+	struct pw_loop *loop;
+	struct pw_loop *data_loop;
+
+	struct pw_core *core;
+	struct spa_hook core_listener;
+	struct spa_hook core_proxy_listener;
+	unsigned int do_disconnect:1;
+
+	char *ifname;
+	bool always_process;
+	uint32_t cleanup_interval;
+
+	struct spa_source *timer;
+
+	struct pw_properties *stream_props;
+	struct vban_stream *stream;
+
+	uint16_t src_port;
+	struct sockaddr_storage src_addr;
+	socklen_t src_len;
+	struct spa_source *source;
+
+	unsigned receiving:1;
+};
+
+static void
+on_vban_io(void *data, int fd, uint32_t mask)
+{
+	struct impl *impl = data;
+	ssize_t len;
+	uint8_t buffer2048;
+
+	if (mask & SPA_IO_IN) {
+		if ((len = recv(fd, buffer, sizeof(buffer), 0)) < 0)
+			goto receive_error;
+
+		if (len < 12)
+			goto short_packet;
+
+		if (SPA_LIKELY(impl->stream))
+			vban_stream_receive_packet(impl->stream, buffer, len);
+
+		impl->receiving = true;
+	}
+	return;
+
+receive_error:
+	pw_log_warn("recv error: %m");
+	return;
+short_packet:
+	pw_log_warn("short packet received");
+	return;
+}
+
+static int parse_address(const char *address, uint16_t port,
+		struct sockaddr_storage *addr, socklen_t *len)
+{
+	struct sockaddr_in *sa4 = (struct sockaddr_in*)addr;
+	struct sockaddr_in6 *sa6 = (struct sockaddr_in6*)addr;
+
+	if (inet_pton(AF_INET, address, &sa4->sin_addr) > 0) {
+		sa4->sin_family = AF_INET;
+		sa4->sin_port = htons(port);
+		*len = sizeof(*sa4);
+	} else if (inet_pton(AF_INET6, address, &sa6->sin6_addr) > 0) {
+		sa6->sin6_family = AF_INET6;
+		sa6->sin6_port = htons(port);
+		*len = sizeof(*sa6);
+	} else
+		return -EINVAL;
+
+	return 0;
+}
+
+static int make_socket(const struct sockaddr* sa, socklen_t salen, char *ifname)
+{
+	int af, fd, val, res;
+	struct ifreq req;
+
+	af = sa->sa_family;
+	if ((fd = socket(af, SOCK_DGRAM | SOCK_CLOEXEC | SOCK_NONBLOCK, 0)) < 0) {
+		pw_log_error("socket failed: %m");
+		return -errno;
+	}
+#ifdef SO_TIMESTAMP
+	val = 1;
+	if (setsockopt(fd, SOL_SOCKET, SO_TIMESTAMP, &val, sizeof(val)) < 0) {
+		res = -errno;
+		pw_log_error("setsockopt failed: %m");
+		goto error;
+	}
+#endif
+	val = 1;
+	if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &val, sizeof(val)) < 0) {
+		res = -errno;
+		pw_log_error("setsockopt failed: %m");
+		goto error;
+	}
+
+	spa_zero(req);
+	if (ifname) {
+		snprintf(req.ifr_name, sizeof(req.ifr_name), "%s", ifname);
+		res = ioctl(fd, SIOCGIFINDEX, &req);
+	        if (res < 0)
+	                pw_log_warn("SIOCGIFINDEX %s failed: %m", ifname);
+	}
+	res = 0;
+	if (af == AF_INET) {
+		static const uint32_t ipv4_mcast_mask = 0xe0000000;
+		struct sockaddr_in *sa4 = (struct sockaddr_in*)sa;
+		if ((ntohl(sa4->sin_addr.s_addr) & ipv4_mcast_mask) == ipv4_mcast_mask) {
+			struct ip_mreqn mr4;
+			memset(&mr4, 0, sizeof(mr4));
+			mr4.imr_multiaddr = sa4->sin_addr;
+			mr4.imr_ifindex = req.ifr_ifindex;
+			res = setsockopt(fd, IPPROTO_IP, IP_ADD_MEMBERSHIP, &mr4, sizeof(mr4));
+		} else {
+			sa4->sin_addr.s_addr = INADDR_ANY;
+		}
+	} else if (af == AF_INET6) {
+		struct sockaddr_in6 *sa6 = (struct sockaddr_in6*)sa;
+		if (sa6->sin6_addr.s6_addr0 == 0xff) {
+			struct ipv6_mreq mr6;
+			memset(&mr6, 0, sizeof(mr6));
+			mr6.ipv6mr_multiaddr = sa6->sin6_addr;
+			mr6.ipv6mr_interface = req.ifr_ifindex;
+			res = setsockopt(fd, IPPROTO_IPV6, IPV6_JOIN_GROUP, &mr6, sizeof(mr6));
+		} else {
+		        sa6->sin6_addr = in6addr_any;
+		}
+	} else {
+		res = -EINVAL;
+		goto error;
+	}
+
+	if (res < 0) {
+		res = -errno;
+		pw_log_error("join mcast failed: %m");
+		goto error;
+	}
+
+	if (bind(fd, sa, salen) < 0) {
+		res = -errno;
+		pw_log_error("bind() failed: %m");
+		goto error;
+	}
+	return fd;
+error:
+	close(fd);
+	return res;
+}
+
+static int stream_start(struct impl *impl)
+{
+	int fd;
+
+	if (impl->source != NULL)
+		return 0;
+
+	pw_log_info("starting VBAN listener");
+
+	if ((fd = make_socket((const struct sockaddr *)&impl->src_addr,
+					impl->src_len, impl->ifname)) < 0) {
+		pw_log_error("failed to create socket: %m");
+		return fd;
+	}
+
+	impl->source = pw_loop_add_io(impl->data_loop, fd,
+				SPA_IO_IN, true, on_vban_io, impl);
+	if (impl->source == NULL) {
+		pw_log_error("can't create io source: %m");
+		close(fd);
+		return -errno;
+	}
+	return 0;
+}
+
+static void stream_stop(struct impl *impl)
+{
+	if (!impl->source)
+		return;
+
+	pw_log_info("stopping VBAN listener");
+
+	pw_loop_destroy_source(impl->data_loop, impl->source);
+	impl->source = NULL;
+}
+
+static void stream_destroy(void *d)
+{
+	struct impl *impl = d;
+	impl->stream = NULL;
+}
+
+static void stream_state_changed(void *data, bool started, const char *error)
+{
+	struct impl *impl = data;
+
+	if (error) {
+		pw_log_error("stream error: %s", error);
+		pw_impl_module_schedule_destroy(impl->module);
+	} else if (started) {
+		if ((errno = -stream_start(impl)) < 0)
+			pw_log_error("failed to start VBAN stream: %m");
+	} else {
+		if (!impl->always_process)
+			stream_stop(impl);
+	}
+}
+
+static const struct vban_stream_events stream_events = {
+	VBAN_VERSION_STREAM_EVENTS,
+	.destroy = stream_destroy,
+	.state_changed = stream_state_changed,
+};
+
+static void on_timer_event(void *data, uint64_t expirations)
+{
+	struct impl *impl = data;
+
+	if (!impl->receiving) {
+		pw_log_info("timeout, inactive VBAN source");
+		//pw_impl_module_schedule_destroy(impl->module);
+	} else {
+		pw_log_debug("timeout, keeping active VBAN source");
+	}
+	impl->receiving = false;
+}
+
+static void core_destroy(void *d)
+{
+	struct impl *impl = d;
+	spa_hook_remove(&impl->core_listener);
+	impl->core = NULL;
+	pw_impl_module_schedule_destroy(impl->module);
+}
+
+static const struct pw_proxy_events core_proxy_events = {
+	.destroy = core_destroy,
+};
+
+static void impl_destroy(struct impl *impl)
+{
+	if (impl->stream)
+		vban_stream_destroy(impl->stream);
+	if (impl->source)
+		pw_loop_destroy_source(impl->data_loop, impl->source);
+
+	if (impl->core && impl->do_disconnect)
+		pw_core_disconnect(impl->core);
+
+	if (impl->timer)
+		pw_loop_destroy_source(impl->loop, impl->timer);
+
+	pw_properties_free(impl->stream_props);
+	pw_properties_free(impl->props);
+
+	free(impl->ifname);
+	free(impl);
+}
+
+static void module_destroy(void *d)
+{
+	struct impl *impl = d;
+	spa_hook_remove(&impl->module_listener);
+	impl_destroy(impl);
+}
+
+static const struct pw_impl_module_events module_events = {
+	PW_VERSION_IMPL_MODULE_EVENTS,
+	.destroy = module_destroy,
+};
+
+static void on_core_error(void *d, uint32_t id, int seq, int res, const char *message)
+{
+	struct impl *impl = d;
+
+	pw_log_error("error id:%u seq:%d res:%d (%s): %s",
+			id, seq, res, spa_strerror(res), message);
+
+	if (id == PW_ID_CORE && res == -EPIPE)
+		pw_impl_module_schedule_destroy(impl->module);
+}
+
+static const struct pw_core_events core_events = {
+	PW_VERSION_CORE_EVENTS,
+	.error = on_core_error,
+};
+
+static void copy_props(struct impl *impl, struct pw_properties *props, const char *key)
+{
+	const char *str;
+	if ((str = pw_properties_get(props, key)) != NULL) {
+		if (pw_properties_get(impl->stream_props, key) == NULL)
+			pw_properties_set(impl->stream_props, key, str);
+	}
+}
+
+SPA_EXPORT
+int pipewire__module_init(struct pw_impl_module *module, const char *args)
+{
+	struct pw_context *context = pw_impl_module_get_context(module);
+	struct impl *impl;
+	const char *str, *sess_name;
+	struct timespec value, interval;
+	struct pw_properties *props, *stream_props;
+	int res = 0;
+
+	PW_LOG_TOPIC_INIT(mod_topic);
+
+	impl = calloc(1, sizeof(struct impl));
+	if (impl == NULL)
+		return -errno;
+
+	if (args == NULL)
+		args = "";
+
+	props = impl->props = pw_properties_new_string(args);
+	stream_props = impl->stream_props = pw_properties_new(NULL, NULL);
+	if (props == NULL || stream_props == NULL) {
+		res = -errno;
+		pw_log_error( "can't create properties: %m");
+		goto out;
+	}
+
+	impl->module = module;
+	impl->context = context;
+	impl->loop = pw_context_get_main_loop(context);
+	impl->data_loop = pw_data_loop_get_loop(pw_context_get_data_loop(context));
+
+	if ((sess_name = pw_properties_get(props, "sess.name")) == NULL)
+		sess_name = pw_get_host_name();
+
+	if (pw_properties_get(props, PW_KEY_NODE_NAME) == NULL)
+		pw_properties_setf(props, PW_KEY_NODE_NAME, "vban_session.%s", sess_name);
+	if (pw_properties_get(props, PW_KEY_NODE_DESCRIPTION) == NULL)
+		pw_properties_setf(props, PW_KEY_NODE_DESCRIPTION, "%s", sess_name);
+	if (pw_properties_get(props, PW_KEY_MEDIA_NAME) == NULL)
+		pw_properties_setf(props, PW_KEY_MEDIA_NAME, "VBAN Session with %s",
+				sess_name);
+
+	if ((str = pw_properties_get(props, "stream.props")) != NULL)
+		pw_properties_update_string(stream_props, str, strlen(str));
+
+	copy_props(impl, props, PW_KEY_AUDIO_FORMAT);
+	copy_props(impl, props, PW_KEY_AUDIO_RATE);
+	copy_props(impl, props, PW_KEY_AUDIO_CHANNELS);
+	copy_props(impl, props, SPA_KEY_AUDIO_POSITION);
+	copy_props(impl, props, PW_KEY_NODE_NAME);
+	copy_props(impl, props, PW_KEY_NODE_DESCRIPTION);
+	copy_props(impl, props, PW_KEY_NODE_GROUP);
+	copy_props(impl, props, PW_KEY_NODE_LATENCY);
+	copy_props(impl, props, PW_KEY_NODE_VIRTUAL);
+	copy_props(impl, props, PW_KEY_NODE_CHANNELNAMES);
+	copy_props(impl, props, PW_KEY_MEDIA_NAME);
+	copy_props(impl, props, PW_KEY_MEDIA_CLASS);
+	copy_props(impl, props, "net.mtu");
+	copy_props(impl, props, "sess.media");
+	copy_props(impl, props, "sess.name");
+	copy_props(impl, props, "sess.min-ptime");
+	copy_props(impl, props, "sess.max-ptime");
+	copy_props(impl, props, "sess.latency.msec");
+
+	str = pw_properties_get(props, "local.ifname");
+	impl->ifname = str ? strdup(str) : NULL;
+
+	impl->src_port = pw_properties_get_uint32(props, "source.port", DEFAULT_SOURCE_PORT);
+	if (impl->src_port == 0) {
+		pw_log_error("invalid source.port");
+		goto out;
+	}
+	if ((str = pw_properties_get(props, "source.ip")) == NULL)
+		str = DEFAULT_SOURCE_IP;
+	if ((res = parse_address(str, impl->src_port, &impl->src_addr, &impl->src_len)) < 0) {
+		pw_log_error("invalid source.ip %s: %s", str, spa_strerror(res));
+		goto out;
+	}
+
+	impl->always_process = pw_properties_get_bool(stream_props,
+			PW_KEY_NODE_ALWAYS_PROCESS, true);
+
+	impl->cleanup_interval = pw_properties_get_uint32(props,
+			"cleanup.sec", DEFAULT_CLEANUP_SEC);
+
+	impl->core = pw_context_get_object(impl->context, PW_TYPE_INTERFACE_Core);
+	if (impl->core == NULL) {
+		str = pw_properties_get(props, PW_KEY_REMOTE_NAME);
+		impl->core = pw_context_connect(impl->context,
+				pw_properties_new(
+					PW_KEY_REMOTE_NAME, str,
+					NULL),
+				0);
+		impl->do_disconnect = true;
+	}
+	if (impl->core == NULL) {
+		res = -errno;
+		pw_log_error("can't connect: %m");
+		goto out;
+	}
+
+	pw_proxy_add_listener((struct pw_proxy*)impl->core,
+			&impl->core_proxy_listener,
+			&core_proxy_events, impl);
+	pw_core_add_listener(impl->core,
+			&impl->core_listener,
+			&core_events, impl);
+
+	impl->timer = pw_loop_add_timer(impl->loop, on_timer_event, impl);
+	if (impl->timer == NULL) {
+		res = -errno;
+		pw_log_error("can't create timer source: %m");
+		goto out;
+	}
+	value.tv_sec = impl->cleanup_interval;
+	value.tv_nsec = 0;
+	interval.tv_sec = impl->cleanup_interval;
+	interval.tv_nsec = 0;
+	pw_loop_update_timer(impl->loop, impl->timer, &value, &interval, false);
+
+	impl->stream = vban_stream_new(impl->core,
+			PW_DIRECTION_OUTPUT, pw_properties_copy(stream_props),
+			&stream_events, impl);
+	if (impl->stream == NULL) {
+		res = -errno;
+		pw_log_error("can't create stream: %m");
+		goto out;
+	}
+
+	pw_impl_module_add_listener(module, &impl->module_listener, &module_events, impl);
+
+	pw_impl_module_update_properties(module, &SPA_DICT_INIT_ARRAY(module_info));
+
+	pw_log_info("Successfully loaded module-vban-recv");
+
+	return 0;
+out:
+	impl_destroy(impl);
+	return res;
+}

pipewire-0.3.76.tar.gz/src/modules/module-vban-send.c Added

@@ -0,0 +1,534 @@
+/* PipeWire */
+/* SPDX-FileCopyrightText: Copyright © 2023 Wim Taymans <wim.taymans@gmail.com> */
+/* SPDX-License-Identifier: MIT */
+
+#include "config.h"
+
+#include <limits.h>
+#include <unistd.h>
+#include <sys/stat.h>
+#include <sys/socket.h>
+#include <sys/ioctl.h>
+#include <arpa/inet.h>
+#include <netinet/ip.h>
+#include <netinet/in.h>
+#include <net/if.h>
+#include <ctype.h>
+
+#include <spa/utils/hook.h>
+#include <spa/utils/result.h>
+#include <spa/utils/ringbuffer.h>
+#include <spa/utils/json.h>
+#include <spa/param/audio/format-utils.h>
+#include <spa/debug/types.h>
+
+#include <pipewire/pipewire.h>
+#include <pipewire/impl.h>
+
+#include <module-vban/stream.h>
+
+#ifndef IPTOS_DSCP
+#define IPTOS_DSCP_MASK 0xfc
+#define IPTOS_DSCP(x) ((x) & IPTOS_DSCP_MASK)
+#endif
+
+/** \page page_module_vban_send PipeWire Module: VBAN sender
+ *
+ * The `vban-send` module creates a PipeWire sink that sends
+ * audio and midi VBAN(https://vb-audio.com) packets.
+ *
+ * ## Module Options
+ *
+ * Options specific to the behavior of this module
+ *
+ * - `source.ip =<str>`: source IP address, default "0.0.0.0"
+ * - `destination.ip =<str>`: destination IP address, default "127.0.0.1"
+ * - `destination.port =<int>`: destination port, default 6980
+ * - `local.ifname = <str>`: interface name to use
+ * - `net.mtu = <int>`: MTU to use, default 1500
+ * - `net.ttl = <int>`: TTL to use, default 1
+ * - `net.loop = <bool>`: loopback multicast, default false
+ * - `sess.min-ptime = <int>`: minimum packet time in milliseconds, default 2
+ * - `sess.max-ptime = <int>`: maximum packet time in milliseconds, default 20
+ * - `sess.name = <str>`: a session name
+ * - `sess.media = <string>`: the media type audio|midi, default audio
+ * - `stream.props = {}`: properties to be passed to the stream
+ *
+ * ## General options
+ *
+ * Options with well-known behavior:
+ *
+ * - \ref PW_KEY_REMOTE_NAME
+ * - \ref PW_KEY_AUDIO_FORMAT
+ * - \ref PW_KEY_AUDIO_RATE
+ * - \ref PW_KEY_AUDIO_CHANNELS
+ * - \ref SPA_KEY_AUDIO_POSITION
+ * - \ref PW_KEY_NODE_NAME
+ * - \ref PW_KEY_NODE_DESCRIPTION
+ * - \ref PW_KEY_MEDIA_NAME
+ * - \ref PW_KEY_NODE_GROUP
+ * - \ref PW_KEY_NODE_LATENCY
+ * - \ref PW_KEY_NODE_VIRTUAL
+ * - \ref PW_KEY_MEDIA_CLASS
+ *
+ * ## Example configuration
+ *\code{.unparsed}
+ * context.modules = 
+ * {   name = libpipewire-module-vban-send
+ *     args = {
+ *         #local.ifname = "eth0"
+ *         #source.ip = "0.0.0.0"
+ *         #destination.ip = "127.0.0.1"
+ *         #destination.port = 6980
+ *         #net.mtu = 1500
+ *         #net.ttl = 1
+ *         #net.loop = false
+ *         #sess.min-ptime = 2
+ *         #sess.max-ptime = 20
+ *         #sess.name = "PipeWire VBAN stream"
+ *         #sess.media = "audio"
+ *         #audio.format = "S16LE"
+ *         #audio.rate = 44100
+ *         #audio.channels = 2
+ *         #audio.position =  FL FR 
+ *         stream.props = {
+ *             node.name = "vban-sender"
+ *         }
+ *     }
+ *}
+ *
+ *\endcode
+ *
+ * \since 0.3.76
+ */
+
+#define NAME "vban-send"
+
+PW_LOG_TOPIC_STATIC(mod_topic, "mod." NAME);
+#define PW_LOG_TOPIC_DEFAULT mod_topic
+
+#define DEFAULT_PORT		6980
+#define DEFAULT_SOURCE_IP	"0.0.0.0"
+#define DEFAULT_DESTINATION_IP	"127.0.0.1"
+#define DEFAULT_TTL		1
+#define DEFAULT_LOOP		false
+#define DEFAULT_DSCP		34 /* Default to AES-67 AF41 (34) */
+
+#define USAGE	"( source.ip=<source IP address, default:"DEFAULT_SOURCE_IP"> ) "			\
+		"( destination.ip=<destination IP address, default:"DEFAULT_DESTINATION_IP"> ) "	\
+ 		"( destination.port=<int, default:"SPA_STRINGIFY(DEFAULT_PORT)"> ) "			\
+		"( local.ifname=<local interface name to use> ) "					\
+		"( net.mtu=<desired MTU, default:"SPA_STRINGIFY(DEFAULT_MTU)"> ) "			\
+		"( net.ttl=<desired TTL, default:"SPA_STRINGIFY(DEFAULT_TTL)"> ) "			\
+		"( net.loop=<desired loopback, default:"SPA_STRINGIFY(DEFAULT_LOOP)"> ) "		\
+		"( net.dscp=<desired DSCP, default:"SPA_STRINGIFY(DEFAULT_DSCP)"> ) "			\
+		"( sess.name=<a name for the session> ) "						\
+		"( sess.min-ptime=<minimum packet time in milliseconds, default:2> ) "			\
+		"( sess.max-ptime=<maximum packet time in milliseconds, default:20> ) "			\
+ 		"( sess.media=<string, the media type audio|midi, default audio> ) "			\
+		"( audio.format=<format, default:"DEFAULT_FORMAT"> ) "					\
+		"( audio.rate=<sample rate, default:"SPA_STRINGIFY(DEFAULT_RATE)"> ) "			\
+		"( audio.channels=<number of channels, default:"SPA_STRINGIFY(DEFAULT_CHANNELS)"> ) "	\
+		"( audio.position=<channel map, default:"DEFAULT_POSITION"> ) "				\
+		"( stream.props= { key=value ... } ) "
+
+static const struct spa_dict_item module_info = {
+	{ PW_KEY_MODULE_AUTHOR, "Wim Taymans <wim.taymans@gmail.com>" },
+	{ PW_KEY_MODULE_DESCRIPTION, "VBAN Sender" },
+	{ PW_KEY_MODULE_USAGE, USAGE },
+	{ PW_KEY_MODULE_VERSION, PACKAGE_VERSION },
+};
+
+struct impl {
+	struct pw_context *context;
+
+	struct pw_impl_module *module;
+	struct spa_hook module_listener;
+	struct pw_properties *props;
+
+	struct pw_loop *loop;
+
+	struct pw_core *core;
+	struct spa_hook core_listener;
+	struct spa_hook core_proxy_listener;
+
+	struct pw_properties *stream_props;
+	struct vban_stream *stream;
+
+	unsigned int do_disconnect:1;
+
+	char *ifname;
+	char *session_name;
+	uint32_t ttl;
+	bool mcast_loop;
+	uint32_t dscp;
+
+	struct sockaddr_storage src_addr;
+	socklen_t src_len;
+
+	uint16_t dst_port;
+	struct sockaddr_storage dst_addr;
+	socklen_t dst_len;
+
+	int vban_fd;
+};
+
+static void stream_destroy(void *d)
+{
+	struct impl *impl = d;
+	impl->stream = NULL;
+}
+
+static void stream_send_packet(void *data, struct iovec *iov, size_t iovlen)
+{
+	struct impl *impl = data;
+	struct msghdr msg;
+	ssize_t n;
+
+	spa_zero(msg);
+	msg.msg_iov = iov;
+	msg.msg_iovlen = iovlen;
+	msg.msg_control = NULL;
+	msg.msg_controllen = 0;
+	msg.msg_flags = 0;
+
+	n = sendmsg(impl->vban_fd, &msg, MSG_NOSIGNAL);
+	if (n < 0)
+		pw_log_debug("sendmsg() failed: %m");
+}
+
+static void stream_state_changed(void *data, bool started, const char *error)
+{
+	struct impl *impl = data;
+
+	if (error) {
+		pw_log_error("stream error: %s", error);
+		pw_impl_module_schedule_destroy(impl->module);
+	}
+}
+
+static const struct vban_stream_events stream_events = {
+	VBAN_VERSION_STREAM_EVENTS,
+	.destroy = stream_destroy,
+	.state_changed = stream_state_changed,
+	.send_packet = stream_send_packet,
+};
+
+static int parse_address(const char *address, uint16_t port,
+		struct sockaddr_storage *addr, socklen_t *len)
+{
+	struct sockaddr_in *sa4 = (struct sockaddr_in*)addr;
+	struct sockaddr_in6 *sa6 = (struct sockaddr_in6*)addr;
+
+	if (inet_pton(AF_INET, address, &sa4->sin_addr) > 0) {
+		sa4->sin_family = AF_INET;
+		sa4->sin_port = htons(port);
+		*len = sizeof(*sa4);
+	} else if (inet_pton(AF_INET6, address, &sa6->sin6_addr) > 0) {
+		sa6->sin6_family = AF_INET6;
+		sa6->sin6_port = htons(port);
+		*len = sizeof(*sa6);
+	} else
+		return -EINVAL;
+
+	return 0;
+}
+
+static bool is_multicast(struct sockaddr *sa, socklen_t salen)
+{
+	if (sa->sa_family == AF_INET) {
+		static const uint32_t ipv4_mcast_mask = 0xe0000000;
+		struct sockaddr_in *sa4 = (struct sockaddr_in*)sa;
+		return (ntohl(sa4->sin_addr.s_addr) & ipv4_mcast_mask) == ipv4_mcast_mask;
+	} else if (sa->sa_family == AF_INET6) {
+		struct sockaddr_in6 *sa6 = (struct sockaddr_in6*)sa;
+		return sa6->sin6_addr.s6_addr0 == 0xff;
+	}
+	return false;
+}
+
+static int make_socket(struct sockaddr_storage *src, socklen_t src_len,
+		struct sockaddr_storage *dst, socklen_t dst_len,
+		bool loop, int ttl, int dscp)
+{
+	int af, fd, val, res;
+
+	af = src->ss_family;
+	if ((fd = socket(af, SOCK_DGRAM | SOCK_CLOEXEC | SOCK_NONBLOCK, 0)) < 0) {
+		pw_log_error("socket failed: %m");
+		return -errno;
+	}
+	if (bind(fd, (struct sockaddr*)src, src_len) < 0) {
+		res = -errno;
+		pw_log_error("bind() failed: %m");
+		goto error;
+	}
+	if (connect(fd, (struct sockaddr*)dst, dst_len) < 0) {
+		res = -errno;
+		pw_log_error("connect() failed: %m");
+		goto error;
+	}
+	if (is_multicast((struct sockaddr*)dst, dst_len)) {
+		val = loop;
+		if (setsockopt(fd, IPPROTO_IP, IP_MULTICAST_LOOP, &val, sizeof(val)) < 0)
+			pw_log_warn("setsockopt(IP_MULTICAST_LOOP) failed: %m");
+
+		val = ttl;
+		if (setsockopt(fd, IPPROTO_IP, IP_MULTICAST_TTL, &val, sizeof(val)) < 0)
+			pw_log_warn("setsockopt(IP_MULTICAST_TTL) failed: %m");
+	}
+#ifdef SO_PRIORITY
+	val = 6;
+	if (setsockopt(fd, SOL_SOCKET, SO_PRIORITY, &val, sizeof(val)) < 0)
+		pw_log_warn("setsockopt(SO_PRIORITY) failed: %m");
+#endif
+	if (dscp > 0) {
+		val = IPTOS_DSCP(dscp << 2);
+		if (setsockopt(fd, IPPROTO_IP, IP_TOS, &val, sizeof(val)) < 0)
+			pw_log_warn("setsockopt(IP_TOS) failed: %m");
+	}
+
+
+	return fd;
+error:
+	close(fd);
+	return res;
+}
+
+static int get_ip(const struct sockaddr_storage *sa, char *ip, size_t len)
+{
+	if (sa->ss_family == AF_INET) {
+		struct sockaddr_in *in = (struct sockaddr_in*)sa;
+		inet_ntop(sa->ss_family, &in->sin_addr, ip, len);
+	} else if (sa->ss_family == AF_INET6) {
+		struct sockaddr_in6 *in = (struct sockaddr_in6*)sa;
+		inet_ntop(sa->ss_family, &in->sin6_addr, ip, len);
+	} else
+		return -EIO;
+	return 0;
+}
+
+static void core_destroy(void *d)
+{
+	struct impl *impl = d;
+	spa_hook_remove(&impl->core_listener);
+	impl->core = NULL;
+	pw_impl_module_schedule_destroy(impl->module);
+}
+
+static const struct pw_proxy_events core_proxy_events = {
+	.destroy = core_destroy,
+};
+
+static void impl_destroy(struct impl *impl)
+{
+	if (impl->stream)
+		vban_stream_destroy(impl->stream);
+
+	if (impl->core && impl->do_disconnect)
+		pw_core_disconnect(impl->core);
+
+	if (impl->vban_fd != -1)
+		close(impl->vban_fd);
+
+	pw_properties_free(impl->stream_props);
+	pw_properties_free(impl->props);
+
+	free(impl->ifname);
+	free(impl->session_name);
+	free(impl);
+}
+
+static void module_destroy(void *d)
+{
+	struct impl *impl = d;
+	spa_hook_remove(&impl->module_listener);
+	impl_destroy(impl);
+}
+
+static const struct pw_impl_module_events module_events = {
+	PW_VERSION_IMPL_MODULE_EVENTS,
+	.destroy = module_destroy,
+};
+
+static void on_core_error(void *d, uint32_t id, int seq, int res, const char *message)
+{
+	struct impl *impl = d;
+
+	pw_log_error("error id:%u seq:%d res:%d (%s): %s",
+			id, seq, res, spa_strerror(res), message);
+
+	if (id == PW_ID_CORE && res == -EPIPE)
+		pw_impl_module_schedule_destroy(impl->module);
+}
+
+static const struct pw_core_events core_events = {
+	PW_VERSION_CORE_EVENTS,
+	.error = on_core_error,
+};
+
+static void copy_props(struct impl *impl, struct pw_properties *props, const char *key)
+{
+	const char *str;
+	if ((str = pw_properties_get(props, key)) != NULL) {
+		if (pw_properties_get(impl->stream_props, key) == NULL)
+			pw_properties_set(impl->stream_props, key, str);
+	}
+}
+
+SPA_EXPORT
+int pipewire__module_init(struct pw_impl_module *module, const char *args)
+{
+	struct pw_context *context = pw_impl_module_get_context(module);
+	struct impl *impl;
+	struct pw_properties *props = NULL, *stream_props = NULL;
+	char addr64;
+	const char *str, *sess_name;
+	int res = 0;
+
+	PW_LOG_TOPIC_INIT(mod_topic);
+
+	impl = calloc(1, sizeof(struct impl));
+	if (impl == NULL)
+		return -errno;
+
+	impl->vban_fd = -1;
+
+	if (args == NULL)
+		args = "";
+
+	props = pw_properties_new_string(args);
+	if (props == NULL) {
+		res = -errno;
+		pw_log_error( "can't create properties: %m");
+		goto out;
+	}
+	impl->props = props;
+
+	stream_props = pw_properties_new(NULL, NULL);
+	if (stream_props == NULL) {
+		res = -errno;
+		pw_log_error( "can't create properties: %m");
+		goto out;
+	}
+	impl->stream_props = stream_props;
+
+	impl->module = module;
+	impl->context = context;
+	impl->loop = pw_context_get_main_loop(context);
+
+	if ((sess_name = pw_properties_get(props, "sess.name")) == NULL)
+		sess_name = pw_get_host_name();
+
+	if (pw_properties_get(props, PW_KEY_NODE_NAME) == NULL)
+		pw_properties_setf(props, PW_KEY_NODE_NAME, "vban_session.%s", sess_name);
+	if (pw_properties_get(props, PW_KEY_NODE_DESCRIPTION) == NULL)
+		pw_properties_setf(props, PW_KEY_NODE_DESCRIPTION, "%s", sess_name);
+	if (pw_properties_get(props, PW_KEY_MEDIA_NAME) == NULL)
+		pw_properties_setf(props, PW_KEY_MEDIA_NAME, "VBAN Session with %s",
+				sess_name);
+
+	if ((str = pw_properties_get(props, "stream.props")) != NULL)
+		pw_properties_update_string(stream_props, str, strlen(str));
+
+	copy_props(impl, props, PW_KEY_AUDIO_FORMAT);
+	copy_props(impl, props, PW_KEY_AUDIO_RATE);
+	copy_props(impl, props, PW_KEY_AUDIO_CHANNELS);
+	copy_props(impl, props, SPA_KEY_AUDIO_POSITION);
+	copy_props(impl, props, PW_KEY_NODE_NAME);
+	copy_props(impl, props, PW_KEY_NODE_DESCRIPTION);
+	copy_props(impl, props, PW_KEY_NODE_GROUP);
+	copy_props(impl, props, PW_KEY_NODE_LATENCY);
+	copy_props(impl, props, PW_KEY_NODE_VIRTUAL);
+	copy_props(impl, props, PW_KEY_NODE_CHANNELNAMES);
+	copy_props(impl, props, PW_KEY_MEDIA_NAME);
+	copy_props(impl, props, PW_KEY_MEDIA_CLASS);
+	copy_props(impl, props, "net.mtu");
+	copy_props(impl, props, "sess.media");
+	copy_props(impl, props, "sess.name");
+	copy_props(impl, props, "sess.min-ptime");
+	copy_props(impl, props, "sess.max-ptime");
+	copy_props(impl, props, "sess.latency.msec");
+	copy_props(impl, props, "sess.ts-refclk");
+
+	str = pw_properties_get(props, "local.ifname");
+	impl->ifname = str ? strdup(str) : NULL;
+
+	if ((str = pw_properties_get(props, "source.ip")) == NULL)
+		str = DEFAULT_SOURCE_IP;
+	if ((res = parse_address(str, 0, &impl->src_addr, &impl->src_len)) < 0) {
+		pw_log_error("invalid source.ip %s: %s", str, spa_strerror(res));
+		goto out;
+	}
+
+	impl->dst_port = pw_properties_get_uint32(props, "destination.port", DEFAULT_PORT);
+	if ((str = pw_properties_get(props, "destination.ip")) == NULL)
+		str = DEFAULT_DESTINATION_IP;
+	if ((res = parse_address(str, impl->dst_port, &impl->dst_addr, &impl->dst_len)) < 0) {
+		pw_log_error("invalid destination.ip %s: %s", str, spa_strerror(res));
+		goto out;
+	}
+
+	impl->ttl = pw_properties_get_uint32(props, "net.ttl", DEFAULT_TTL);
+	impl->mcast_loop = pw_properties_get_bool(props, "net.loop", DEFAULT_LOOP);
+	impl->dscp = pw_properties_get_uint32(props, "net.dscp", DEFAULT_DSCP);
+
+	get_ip(&impl->src_addr, addr, sizeof(addr));
+	pw_properties_set(stream_props, "vban.source.ip", addr);
+	get_ip(&impl->dst_addr, addr, sizeof(addr));
+	pw_properties_set(stream_props, "vban.destination.ip", addr);
+	pw_properties_setf(stream_props, "vban.destination.port", "%u", impl->dst_port);
+	pw_properties_setf(stream_props, "vban.ttl", "%u", impl->ttl);
+	pw_properties_setf(stream_props, "vban.dscp", "%u", impl->dscp);
+
+	impl->core = pw_context_get_object(impl->context, PW_TYPE_INTERFACE_Core);
+	if (impl->core == NULL) {
+		str = pw_properties_get(props, PW_KEY_REMOTE_NAME);
+		impl->core = pw_context_connect(impl->context,
+				pw_properties_new(
+					PW_KEY_REMOTE_NAME, str,
+					NULL),
+				0);
+		impl->do_disconnect = true;
+	}
+	if (impl->core == NULL) {
+		res = -errno;
+		pw_log_error("can't connect: %m");
+		goto out;
+	}
+
+	pw_proxy_add_listener((struct pw_proxy*)impl->core,
+			&impl->core_proxy_listener,
+			&core_proxy_events, impl);
+	pw_core_add_listener(impl->core,
+			&impl->core_listener,
+			&core_events, impl);
+
+	if ((res = make_socket(&impl->src_addr, impl->src_len,
+					&impl->dst_addr, impl->dst_len,
+					impl->mcast_loop, impl->ttl, impl->dscp)) < 0) {
+		pw_log_error("can't make socket: %s", spa_strerror(res));
+		goto out;
+	}
+	impl->vban_fd = res;
+
+	impl->stream = vban_stream_new(impl->core,
+			PW_DIRECTION_INPUT, pw_properties_copy(stream_props),
+			&stream_events, impl);
+	if (impl->stream == NULL) {
+		res = -errno;
+		pw_log_error("can't create stream: %m");
+		goto out;
+	}
+
+	pw_impl_module_add_listener(module, &impl->module_listener, &module_events, impl);
+
+	pw_impl_module_update_properties(module, &SPA_DICT_INIT_ARRAY(module_info));
+
+	pw_log_info("Successfully loaded module-vban-send");
+
+	return 0;
+out:
+	impl_destroy(impl);
+	return res;
+}

pipewire-0.3.76.tar.gz/src/modules/module-vban/audio.c Added

@@ -0,0 +1,281 @@
+/* PipeWire */
+/* SPDX-FileCopyrightText: Copyright © 2022 Wim Taymans <wim.taymans@gmail.com> */
+/* SPDX-License-Identifier: MIT */
+
+static void vban_audio_process_playback(void *data)
+{
+	struct impl *impl = data;
+	struct pw_buffer *buf;
+	struct spa_data *d;
+	uint32_t wanted, timestamp, target_buffer, stride, maxsize;
+	int32_t avail;
+
+	if ((buf = pw_stream_dequeue_buffer(impl->stream)) == NULL) {
+		pw_log_debug("Out of stream buffers: %m");
+		return;
+	}
+	d = buf->buffer->datas;
+
+	stride = impl->stride;
+
+	maxsize = d0.maxsize / stride;
+	wanted = buf->requested ? SPA_MIN(buf->requested, maxsize) : maxsize;
+
+	avail = spa_ringbuffer_get_read_index(&impl->ring, &timestamp);
+
+	target_buffer = impl->target_buffer;
+
+	if (avail < (int32_t)wanted) {
+		enum spa_log_level level;
+		memset(d0.data, 0, wanted * stride);
+		if (impl->have_sync) {
+			impl->have_sync = false;
+			level = SPA_LOG_LEVEL_WARN;
+		} else {
+			level = SPA_LOG_LEVEL_DEBUG;
+		}
+		pw_log(level, "underrun %d/%u < %u",
+					avail, target_buffer, wanted);
+	} else {
+		float error, corr;
+		if (impl->first) {
+			if ((uint32_t)avail > target_buffer) {
+				uint32_t skip = avail - target_buffer;
+				pw_log_debug("first: avail:%d skip:%u target:%u",
+							avail, skip, target_buffer);
+				timestamp += skip;
+				avail = target_buffer;
+			}
+			impl->first = false;
+		} else if (avail > (int32_t)SPA_MIN(target_buffer * 8, BUFFER_SIZE / stride)) {
+			pw_log_warn("overrun %u > %u", avail, target_buffer * 8);
+			timestamp += avail - target_buffer;
+			avail = target_buffer;
+		}
+		/* try to adjust our playback rate to keep the
+		 * requested target_buffer bytes in the ringbuffer */
+		error = (float)target_buffer - (float)avail;
+		error = SPA_CLAMP(error, -impl->max_error, impl->max_error);
+
+		corr = spa_dll_update(&impl->dll, error);
+
+		pw_log_debug("avail:%u target:%u error:%f corr:%f", avail,
+				target_buffer, error, corr);
+
+		if (impl->io_rate_match) {
+			SPA_FLAG_SET(impl->io_rate_match->flags,
+					SPA_IO_RATE_MATCH_FLAG_ACTIVE);
+			impl->io_rate_match->rate = 1.0f / corr;
+		}
+		spa_ringbuffer_read_data(&impl->ring,
+				impl->buffer,
+				BUFFER_SIZE,
+				(timestamp * stride) & BUFFER_MASK,
+				d0.data, wanted * stride);
+
+		timestamp += wanted;
+		spa_ringbuffer_read_update(&impl->ring, timestamp);
+	}
+	d0.chunk->size = wanted * stride;
+	d0.chunk->stride = stride;
+	d0.chunk->offset = 0;
+	buf->size = wanted;
+
+	pw_stream_queue_buffer(impl->stream, buf);
+}
+
+static int vban_audio_receive(struct impl *impl, uint8_t *buffer, ssize_t len)
+{
+	struct vban_header *hdr;
+	ssize_t hlen, plen;
+	uint32_t n_frames, timestamp, samples, write, expected_write;
+	uint32_t stride = impl->stride;
+	int32_t filled;
+
+	if (len < VBAN_HEADER_SIZE)
+		goto short_packet;
+
+	hdr = (struct vban_header*)buffer;
+	if (strncmp(hdr->vban, "VBAN", 3))
+		goto invalid_version;
+
+	impl->receiving = true;
+
+	hlen = VBAN_HEADER_SIZE;
+	plen = len - hlen;
+	samples = SPA_MIN(hdr->format_nbs, plen / stride);
+
+	n_frames = hdr->n_frames;
+	if (impl->have_sync && impl->n_frames != n_frames) {
+		pw_log_info("unexpected frame (%d != %d)",
+				n_frames, impl->n_frames);
+		impl->have_sync = false;
+	}
+	impl->n_frames = n_frames + 1;
+
+	timestamp = impl->timestamp;
+	impl->timestamp += samples;
+
+	filled = spa_ringbuffer_get_write_index(&impl->ring, &expected_write);
+
+	/* we always write to timestamp + delay */
+	write = timestamp + impl->target_buffer;
+
+	if (!impl->have_sync) {
+		pw_log_info("sync to timestamp:%u target:%u",
+				timestamp, impl->target_buffer);
+
+		/* we read from timestamp, keeping target_buffer of data
+		 * in the ringbuffer. */
+		impl->ring.readindex = timestamp;
+		impl->ring.writeindex = write;
+		filled = impl->target_buffer;
+
+		spa_dll_init(&impl->dll);
+		spa_dll_set_bw(&impl->dll, SPA_DLL_BW_MIN, 128, impl->rate);
+		memset(impl->buffer, 0, BUFFER_SIZE);
+		impl->have_sync = true;
+	} else if (expected_write != write) {
+		pw_log_debug("unexpected write (%u != %u)",
+				write, expected_write);
+	}
+
+	if (filled + samples > BUFFER_SIZE / stride) {
+		pw_log_debug("capture overrun %u + %u > %u", filled, samples,
+				BUFFER_SIZE / stride);
+		impl->have_sync = false;
+	} else {
+		pw_log_debug("got samples:%u", samples);
+		spa_ringbuffer_write_data(&impl->ring,
+				impl->buffer,
+				BUFFER_SIZE,
+				(write * stride) & BUFFER_MASK,
+				&bufferhlen, (samples * stride));
+		write += samples;
+		spa_ringbuffer_write_update(&impl->ring, write);
+	}
+	return 0;
+
+short_packet:
+	pw_log_warn("short packet received");
+	return -EINVAL;
+invalid_version:
+	pw_log_warn("invalid VBAN version");
+	spa_debug_mem(0, buffer, len);
+	return -EPROTO;
+}
+
+static inline void
+set_iovec(struct spa_ringbuffer *rbuf, void *buffer, uint32_t size,
+		uint32_t offset, struct iovec *iov, uint32_t len)
+{
+	iov0.iov_len = SPA_MIN(len, size - offset);
+	iov0.iov_base = SPA_PTROFF(buffer, offset, void);
+	iov1.iov_len = len - iov0.iov_len;
+	iov1.iov_base = buffer;
+}
+
+static void vban_audio_flush_packets(struct impl *impl)
+{
+	int32_t avail, tosend;
+	uint32_t stride, timestamp;
+	struct iovec iov3;
+	struct vban_header header;
+
+	avail = spa_ringbuffer_get_read_index(&impl->ring, &timestamp);
+	tosend = impl->psamples;
+
+	if (avail < tosend)
+		return;
+
+	stride = impl->stride;
+
+	header = impl->header;
+	header.format_nbs = tosend - 1;
+	header.format_nbc = impl->stream_info.info.raw.channels - 1;
+
+	iov0.iov_base = &header;
+	iov0.iov_len = sizeof(header);
+
+	while (avail >= tosend) {
+		set_iovec(&impl->ring,
+			impl->buffer, BUFFER_SIZE,
+			(timestamp * stride) & BUFFER_MASK,
+			&iov1, tosend * stride);
+
+		pw_log_trace("sending %d timestamp:%08x", tosend, timestamp);
+
+		vban_stream_emit_send_packet(impl, iov, 3);
+
+		timestamp += tosend;
+		avail -= tosend;
+		impl->header.n_frames++;
+	}
+	spa_ringbuffer_read_update(&impl->ring, timestamp);
+}
+
+static void vban_audio_process_capture(void *data)
+{
+	struct impl *impl = data;
+	struct pw_buffer *buf;
+	struct spa_data *d;
+	uint32_t offs, size, timestamp, expected_timestamp, stride;
+	int32_t filled, wanted;
+
+	if ((buf = pw_stream_dequeue_buffer(impl->stream)) == NULL) {
+		pw_log_debug("Out of stream buffers: %m");
+		return;
+	}
+	d = buf->buffer->datas;
+
+	offs = SPA_MIN(d0.chunk->offset, d0.maxsize);
+	size = SPA_MIN(d0.chunk->size, d0.maxsize - offs);
+	stride = impl->stride;
+	wanted = size / stride;
+
+	filled = spa_ringbuffer_get_write_index(&impl->ring, &expected_timestamp);
+
+	if (SPA_LIKELY(impl->io_position)) {
+		uint32_t rate = impl->io_position->clock.rate.denom;
+		timestamp = impl->io_position->clock.position * impl->rate / rate;
+	} else
+		timestamp = expected_timestamp;
+
+	if (!impl->have_sync) {
+		pw_log_info("sync to timestamp:%u", timestamp);
+		impl->ring.readindex = impl->ring.writeindex = timestamp;
+		memset(impl->buffer, 0, BUFFER_SIZE);
+		impl->have_sync = true;
+		expected_timestamp = timestamp;
+	} else {
+		if (SPA_ABS((int32_t)expected_timestamp - (int32_t)timestamp) > 32) {
+			pw_log_warn("expected %u != timestamp %u", expected_timestamp, timestamp);
+			impl->have_sync = false;
+		} else if (filled + wanted > (int32_t)(BUFFER_SIZE / stride)) {
+			pw_log_warn("overrun %u + %u > %u", filled, wanted, BUFFER_SIZE / stride);
+			impl->have_sync = false;
+		}
+	}
+
+	spa_ringbuffer_write_data(&impl->ring,
+			impl->buffer,
+			BUFFER_SIZE,
+			(expected_timestamp * stride) & BUFFER_MASK,
+			SPA_PTROFF(d0.data, offs, void), wanted * stride);
+	expected_timestamp += wanted;
+	spa_ringbuffer_write_update(&impl->ring, expected_timestamp);
+
+	pw_stream_queue_buffer(impl->stream, buf);
+
+	vban_audio_flush_packets(impl);
+}
+
+static int vban_audio_init(struct impl *impl, enum spa_direction direction)
+{
+	if (direction == SPA_DIRECTION_INPUT)
+		impl->stream_events.process = vban_audio_process_capture;
+	else
+		impl->stream_events.process = vban_audio_process_playback;
+	impl->receive_vban = vban_audio_receive;
+	return 0;
+}

pipewire-0.3.76.tar.gz/src/modules/module-vban/stream.c Added

@@ -0,0 +1,479 @@
+/* PipeWire */
+/* SPDX-FileCopyrightText: Copyright © 2023 Wim Taymans <wim.taymans@gmail.com> */
+/* SPDX-License-Identifier: MIT */
+
+#include <sys/socket.h>
+#include <arpa/inet.h>
+
+#include <spa/utils/result.h>
+#include <spa/utils/json.h>
+#include <spa/utils/ringbuffer.h>
+#include <spa/utils/dll.h>
+#include <spa/param/audio/format-utils.h>
+#include <spa/control/control.h>
+#include <spa/debug/types.h>
+#include <spa/debug/mem.h>
+
+#include "config.h"
+
+#include <pipewire/pipewire.h>
+#include <pipewire/impl.h>
+
+#include <module-vban/vban.h>
+#include <module-vban/stream.h>
+
+#define BUFFER_SIZE			(1u<<22)
+#define BUFFER_MASK			(BUFFER_SIZE-1)
+#define BUFFER_SIZE2			(BUFFER_SIZE>>1)
+#define BUFFER_MASK2			(BUFFER_SIZE2-1)
+
+#define vban_stream_emit(s,m,v,...)		spa_hook_list_call(&s->listener_list, \
+							struct vban_stream_events, m, v, ##__VA_ARGS__)
+#define vban_stream_emit_destroy(s)		vban_stream_emit(s, destroy, 0)
+#define vban_stream_emit_state_changed(s,n,e)	vban_stream_emit(s, state_changed,0,n,e)
+#define vban_stream_emit_send_packet(s,i,l)	vban_stream_emit(s, send_packet,0,i,l)
+#define vban_stream_emit_send_feedback(s,seq)	vban_stream_emit(s, send_feedback,0,seq)
+
+struct impl {
+	struct spa_audio_info info;
+	struct spa_audio_info stream_info;
+
+	struct pw_stream *stream;
+	struct spa_hook stream_listener;
+	struct pw_stream_events stream_events;
+
+	struct spa_hook_list listener_list;
+	struct spa_hook listener;
+
+	const struct format_info *format_info;
+
+	void *stream_data;
+
+	uint32_t rate;
+	uint32_t stride;
+	uint32_t psamples;
+	uint32_t mtu;
+
+	struct vban_header header;
+	uint32_t timestamp;
+	uint32_t n_frames;
+
+	struct spa_ringbuffer ring;
+	uint8_t bufferBUFFER_SIZE;
+
+	struct spa_io_rate_match *io_rate_match;
+	struct spa_io_position *io_position;
+	struct spa_dll dll;
+	double corr;
+	uint32_t target_buffer;
+	float max_error;
+
+	float last_timestamp;
+	float last_time;
+
+	unsigned always_process:1;
+	unsigned started:1;
+	unsigned have_sync:1;
+	unsigned receiving:1;
+	unsigned first:1;
+
+	int (*receive_vban)(struct impl *impl, uint8_t *buffer, ssize_t len);
+};
+
+#include "module-vban/audio.c"
+//#include "module-vban/midi.c"
+
+struct format_info {
+	uint32_t media_subtype;
+	uint32_t format;
+	uint32_t size;
+	uint8_t format_bit;
+};
+
+static const struct format_info audio_format_info = {
+	{ SPA_MEDIA_SUBTYPE_raw, SPA_AUDIO_FORMAT_U8, 1, VBAN_DATATYPE_U8, },
+	{ SPA_MEDIA_SUBTYPE_raw, SPA_AUDIO_FORMAT_S16_LE, 2, VBAN_DATATYPE_INT16, },
+	{ SPA_MEDIA_SUBTYPE_raw, SPA_AUDIO_FORMAT_S24_LE, 3, VBAN_DATATYPE_INT24, },
+	{ SPA_MEDIA_SUBTYPE_raw, SPA_AUDIO_FORMAT_S32_LE, 4, VBAN_DATATYPE_INT32, },
+	{ SPA_MEDIA_SUBTYPE_raw, SPA_AUDIO_FORMAT_F32_LE, 4, VBAN_DATATYPE_FLOAT32, },
+	{ SPA_MEDIA_SUBTYPE_raw, SPA_AUDIO_FORMAT_F64_LE, 8, VBAN_DATATYPE_FLOAT64, },
+	{ SPA_MEDIA_SUBTYPE_control, 0, 1, },
+};
+
+static void stream_io_changed(void *data, uint32_t id, void *area, uint32_t size)
+{
+	struct impl *impl = data;
+	switch (id) {
+	case SPA_IO_RateMatch:
+		impl->io_rate_match = area;
+		break;
+	case SPA_IO_Position:
+		impl->io_position = area;
+		break;
+	}
+}
+
+static void stream_destroy(void *d)
+{
+	struct impl *impl = d;
+	spa_hook_remove(&impl->stream_listener);
+	impl->stream = NULL;
+}
+
+static int stream_start(struct impl *impl)
+{
+	if (impl->started)
+		return 0;
+
+	vban_stream_emit_state_changed(impl, true, NULL);
+
+	impl->started = true;
+	return 0;
+}
+
+static int stream_stop(struct impl *impl)
+{
+	if (!impl->started)
+		return 0;
+
+	vban_stream_emit_state_changed(impl, false, NULL);
+
+	impl->started = false;
+	return 0;
+}
+
+static void on_stream_state_changed(void *d, enum pw_stream_state old,
+		enum pw_stream_state state, const char *error)
+{
+	struct impl *impl = d;
+
+	switch (state) {
+		case PW_STREAM_STATE_UNCONNECTED:
+			pw_log_info("stream disconnected");
+			break;
+		case PW_STREAM_STATE_ERROR:
+			pw_log_error("stream error: %s", error);
+			vban_stream_emit_state_changed(impl, false, error);
+			break;
+		case PW_STREAM_STATE_STREAMING:
+			if ((errno = -stream_start(impl)) < 0)
+				pw_log_error("failed to start RTP stream: %m");
+			break;
+		case PW_STREAM_STATE_PAUSED:
+			if (!impl->always_process)
+				stream_stop(impl);
+			impl->have_sync = false;
+			break;
+		default:
+			break;
+	}
+}
+
+static const struct pw_stream_events stream_events = {
+	PW_VERSION_STREAM_EVENTS,
+	.destroy = stream_destroy,
+	.state_changed = on_stream_state_changed,
+	.io_changed = stream_io_changed,
+};
+
+static const struct format_info *find_audio_format_info(const struct spa_audio_info *info)
+{
+	SPA_FOR_EACH_ELEMENT_VAR(audio_format_info, f)
+		if (f->media_subtype == info->media_subtype &&
+		    (f->format == 0 || f->format == info->info.raw.format))
+			return f;
+	return NULL;
+}
+
+static inline uint32_t format_from_name(const char *name, size_t len)
+{
+	int i;
+	for (i = 0; spa_type_audio_formati.name; i++) {
+		if (strncmp(name, spa_debug_type_short_name(spa_type_audio_formati.name), len) == 0)
+			return spa_type_audio_formati.type;
+	}
+	return SPA_AUDIO_FORMAT_UNKNOWN;
+}
+
+static uint32_t channel_from_name(const char *name)
+{
+	int i;
+	for (i = 0; spa_type_audio_channeli.name; i++) {
+		if (spa_streq(name, spa_debug_type_short_name(spa_type_audio_channeli.name)))
+			return spa_type_audio_channeli.type;
+	}
+	return SPA_AUDIO_CHANNEL_UNKNOWN;
+}
+
+static void parse_position(struct spa_audio_info_raw *info, const char *val, size_t len)
+{
+	struct spa_json it2;
+	char v256;
+
+	spa_json_init(&it0, val, len);
+        if (spa_json_enter_array(&it0, &it1) <= 0)
+                spa_json_init(&it1, val, len);
+
+	info->channels = 0;
+	while (spa_json_get_string(&it1, v, sizeof(v)) > 0 &&
+	    info->channels < SPA_AUDIO_MAX_CHANNELS) {
+		info->positioninfo->channels++ = channel_from_name(v);
+	}
+}
+
+static void parse_audio_info(const struct pw_properties *props, struct spa_audio_info_raw *info)
+{
+	const char *str;
+
+	spa_zero(*info);
+	if ((str = pw_properties_get(props, PW_KEY_AUDIO_FORMAT)) == NULL)
+		str = DEFAULT_FORMAT;
+	info->format = format_from_name(str, strlen(str));
+
+	info->rate = pw_properties_get_uint32(props, PW_KEY_AUDIO_RATE, info->rate);
+	if (info->rate == 0)
+		info->rate = DEFAULT_RATE;
+
+	info->channels = pw_properties_get_uint32(props, PW_KEY_AUDIO_CHANNELS, info->channels);
+	info->channels = SPA_MIN(info->channels, SPA_AUDIO_MAX_CHANNELS);
+	if ((str = pw_properties_get(props, SPA_KEY_AUDIO_POSITION)) != NULL)
+		parse_position(info, str, strlen(str));
+	if (info->channels == 0)
+		parse_position(info, DEFAULT_POSITION, strlen(DEFAULT_POSITION));
+}
+
+static uint32_t msec_to_samples(struct impl *impl, uint32_t msec)
+{
+	return msec * impl->rate / 1000;
+}
+
+struct vban_stream *vban_stream_new(struct pw_core *core,
+		enum pw_direction direction, struct pw_properties *props,
+		const struct vban_stream_events *events, void *data)
+{
+	struct impl *impl;
+	const char *str;
+	uint8_t buffer1024;
+	struct spa_pod_builder b;
+	uint32_t n_params, min_samples, max_samples;
+	float min_ptime, max_ptime;
+	const struct spa_pod *params1;
+	enum pw_stream_flags flags;
+	int latency_msec;
+	int res;
+
+	impl = calloc(1, sizeof(*impl));
+	if (impl == NULL) {
+		res = -errno;
+		goto out;
+		return NULL;
+	}
+	impl->first = true;
+	spa_hook_list_init(&impl->listener_list);
+	impl->stream_events = stream_events;
+
+	if ((str = pw_properties_get(props, "sess.media")) == NULL)
+		str = "audio";
+
+	if (spa_streq(str, "audio")) {
+		impl->info.media_type = SPA_MEDIA_TYPE_audio;
+		impl->info.media_subtype = SPA_MEDIA_SUBTYPE_raw;
+	}
+	else if (spa_streq(str, "midi")) {
+		impl->info.media_type = SPA_MEDIA_TYPE_application;
+		impl->info.media_subtype = SPA_MEDIA_SUBTYPE_control;
+	}
+	else {
+		pw_log_error("unsupported media type:%s", str);
+		res = -EINVAL;
+		goto out;
+	}
+	memcpy(impl->header.vban, "VBAN", 4);
+	if ((str = pw_properties_get(props, "sess.name")) == NULL)
+		str = "Stream1";
+	strcpy(impl->header.stream_name, str);
+
+	switch (impl->info.media_subtype) {
+	case SPA_MEDIA_SUBTYPE_raw:
+		parse_audio_info(props, &impl->info.info.raw);
+		impl->stream_info = impl->info;
+		impl->format_info = find_audio_format_info(&impl->info);
+		if (impl->format_info == NULL) {
+			pw_log_error("unsupported audio format:%d channels:%d",
+					impl->stream_info.info.raw.format,
+					impl->stream_info.info.raw.channels);
+			res = -EINVAL;
+			goto out;
+		}
+		impl->stride = impl->format_info->size * impl->stream_info.info.raw.channels;
+		impl->rate = impl->stream_info.info.raw.rate;
+		impl->header.format_SR = vban_sr_index(impl->rate);
+		if (impl->header.format_SR == VBAN_SR_MAXNUMBER) {
+			pw_log_error("unsupported audio rate:%u", impl->rate);
+			res = -EINVAL;
+			goto out;
+		}
+		impl->header.format_bit = impl->format_info->format_bit;
+		break;
+	case SPA_MEDIA_SUBTYPE_control:
+		impl->stream_info = impl->info;
+		impl->format_info = find_audio_format_info(&impl->info);
+		if (impl->format_info == NULL) {
+			res = -EINVAL;
+			goto out;
+		}
+		pw_properties_set(props, PW_KEY_FORMAT_DSP, "8 bit raw midi");
+		impl->stride = impl->format_info->size;
+		impl->rate = pw_properties_get_uint32(props, "midi.rate", 10000);
+		if (impl->rate == 0)
+			impl->rate = 10000;
+		break;
+	default:
+		spa_assert_not_reached();
+		break;
+	}
+
+	if (pw_properties_get(props, PW_KEY_NODE_VIRTUAL) == NULL)
+		pw_properties_set(props, PW_KEY_NODE_VIRTUAL, "true");
+	if (pw_properties_get(props, PW_KEY_NODE_NETWORK) == NULL)
+		pw_properties_set(props, PW_KEY_NODE_NETWORK, "true");
+
+	impl->mtu = pw_properties_get_uint32(props, "net.mtu", DEFAULT_MTU);
+
+	str = pw_properties_get(props, "sess.min-ptime");
+	if (!spa_atof(str, &min_ptime))
+		min_ptime = DEFAULT_MIN_PTIME;
+	str = pw_properties_get(props, "sess.max-ptime");
+	if (!spa_atof(str, &max_ptime))
+		max_ptime = DEFAULT_MAX_PTIME;
+
+	min_samples = min_ptime * impl->rate / 1000;
+	max_samples = SPA_MIN(256, max_ptime * impl->rate / 1000);
+
+	float ptime = 0;
+	if ((str = pw_properties_get(props, "vban.ptime")) != NULL)
+		if (!spa_atof(str, &ptime))
+			ptime = 0.0;
+
+	if (ptime) {
+		impl->psamples = ptime * impl->rate / 1000;
+	} else {
+		impl->psamples = impl->mtu / impl->stride;
+		impl->psamples = SPA_CLAMP(impl->psamples, min_samples, max_samples);
+		if (direction == PW_DIRECTION_OUTPUT)
+			pw_properties_setf(props, "vban.ptime", "%f",
+					impl->psamples * 1000.0 / impl->rate);
+	}
+	latency_msec = pw_properties_get_uint32(props,
+			"sess.latency.msec", DEFAULT_SESS_LATENCY);
+	impl->target_buffer = msec_to_samples(impl, latency_msec);
+	impl->max_error = msec_to_samples(impl, ERROR_MSEC);
+
+	pw_properties_setf(props, PW_KEY_NODE_RATE, "1/%d", impl->rate);
+	if (direction == PW_DIRECTION_INPUT) {
+		pw_properties_setf(props, PW_KEY_NODE_LATENCY, "%d/%d",
+				impl->psamples, impl->rate);
+	} else {
+		pw_properties_setf(props, PW_KEY_NODE_LATENCY, "%d/%d",
+				impl->target_buffer / 2, impl->rate);
+	}
+
+	pw_properties_setf(props, "net.mtu", "%u", impl->mtu);
+	pw_properties_setf(props, "vban.rate", "%u", impl->rate);
+	if (impl->info.info.raw.channels > 0)
+		pw_properties_setf(props, "vban.channels", "%u", impl->info.info.raw.channels);
+
+	spa_dll_init(&impl->dll);
+	spa_dll_set_bw(&impl->dll, SPA_DLL_BW_MIN, 128, impl->rate);
+	impl->corr = 1.0;
+
+	impl->stream = pw_stream_new(core, "vban-session", props);
+	props = NULL;
+	if (impl->stream == NULL) {
+		res = -errno;
+		pw_log_error("can't create stream: %m");
+		goto out;
+	}
+
+	n_params = 0;
+	spa_pod_builder_init(&b, buffer, sizeof(buffer));
+
+	flags = PW_STREAM_FLAG_MAP_BUFFERS | PW_STREAM_FLAG_RT_PROCESS;
+
+
+	switch (impl->info.media_subtype) {
+	case SPA_MEDIA_SUBTYPE_raw:
+		paramsn_params++ = spa_format_audio_build(&b,
+				SPA_PARAM_EnumFormat, &impl->stream_info);
+		flags |= PW_STREAM_FLAG_AUTOCONNECT;
+		vban_audio_init(impl, direction);
+		break;
+	case SPA_MEDIA_SUBTYPE_control:
+		paramsn_params++ = spa_pod_builder_add_object(&b,
+                                SPA_TYPE_OBJECT_Format, SPA_PARAM_EnumFormat,
+                                SPA_FORMAT_mediaType,           SPA_POD_Id(SPA_MEDIA_TYPE_application),
+                                SPA_FORMAT_mediaSubtype,        SPA_POD_Id(SPA_MEDIA_SUBTYPE_control));
+//		vban_midi_init(impl, direction);
+		break;
+	default:
+		res = -EINVAL;
+		goto out;
+	}
+
+	pw_stream_add_listener(impl->stream,
+			&impl->stream_listener,
+			&impl->stream_events, impl);
+
+	if ((res = pw_stream_connect(impl->stream,
+			direction,
+			PW_ID_ANY,
+			flags,
+			params, n_params)) < 0) {
+		pw_log_error("can't connect stream: %s", spa_strerror(res));
+		goto out;
+	}
+
+	if (impl->always_process &&
+		(res = stream_start(impl)) < 0)
+		goto out;
+
+	spa_hook_list_append(&impl->listener_list, &impl->listener, events, data);
+
+	return (struct vban_stream*)impl;
+out:
+	pw_properties_free(props);
+	errno = -res;
+	return NULL;
+}
+
+void vban_stream_destroy(struct vban_stream *s)
+{
+	struct impl *impl = (struct impl*)s;
+
+	vban_stream_emit_destroy(impl);
+
+	if (impl->stream)
+		pw_stream_destroy(impl->stream);
+
+	spa_hook_list_clean(&impl->listener_list);
+	free(impl);
+}
+
+int vban_stream_receive_packet(struct vban_stream *s, uint8_t *buffer, size_t len)
+{
+	struct impl *impl = (struct impl*)s;
+	return impl->receive_vban(impl, buffer, len);
+}
+
+uint64_t vban_stream_get_time(struct vban_stream *s, uint64_t *rate)
+{
+	struct impl *impl = (struct impl*)s;
+	struct spa_io_position *pos = impl->io_position;
+
+	if (pos == NULL)
+		return -EIO;
+
+	*rate = impl->rate;
+	return pos->clock.position * impl->rate *
+		pos->clock.rate.num / pos->clock.rate.denom;
+}

pipewire-0.3.76.tar.gz/src/modules/module-vban/stream.h Added

@@ -0,0 +1,54 @@
+/* PipeWire */
+/* SPDX-FileCopyrightText: Copyright © 2023 Wim Taymans <wim.taymans@gmail.com> */
+/* SPDX-License-Identifier: MIT */
+
+#ifndef PIPEWIRE_VBAN_STREAM_H
+#define PIPEWIRE_VBAN_STREAM_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+struct vban_stream;
+
+#define DEFAULT_FORMAT		"S16LE"
+#define DEFAULT_RATE		44100
+#define DEFAULT_CHANNELS	2
+#define DEFAULT_POSITION	" FL FR "
+
+#define ERROR_MSEC		2
+#define DEFAULT_SESS_LATENCY	100
+
+#define DEFAULT_MTU		VBAN_PROTOCOL_MAX_SIZE
+#define DEFAULT_MIN_PTIME	2
+#define DEFAULT_MAX_PTIME	20
+
+struct vban_stream_events {
+#define VBAN_VERSION_STREAM_EVENTS        0
+	uint32_t version;
+
+	void (*destroy) (void *data);
+
+	void (*state_changed) (void *data, bool started, const char *error);
+
+	void (*send_packet) (void *data, struct iovec *iov, size_t iovlen);
+
+	void (*send_feedback) (void *data, uint32_t senum);
+};
+
+struct vban_stream *vban_stream_new(struct pw_core *core,
+		enum pw_direction direction, struct pw_properties *props,
+		const struct vban_stream_events *events, void *data);
+
+void vban_stream_destroy(struct vban_stream *s);
+
+int vban_stream_receive_packet(struct vban_stream *s, uint8_t *buffer, size_t len);
+
+uint64_t vban_stream_get_time(struct vban_stream *s, uint64_t *rate);
+
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /* PIPEWIRE_VBAN_STREAM_H */

pipewire-0.3.76.tar.gz/src/modules/module-vban/vban.h Added

@@ -0,0 +1,60 @@
+/* PipeWire */
+/* SPDX-FileCopyrightText: Copyright © 2023 Wim Taymans <wim.taymans@gmail.com> */
+/* SPDX-License-Identifier: MIT */
+
+#ifndef PIPEWIRE_VBAN_H
+#define PIPEWIRE_VBAN_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#define VBAN_HEADER_SIZE	(4 + 4 + 16 + 4)
+#define VBAN_STREAM_NAME_SIZE	16
+#define VBAN_PROTOCOL_MAX_SIZE	1464
+#define VBAN_DATA_MAX_SIZE	(VBAN_PROTOCOL_MAX_SIZE - VBAN_HEADER_SIZE)
+#define VBAN_CHANNELS_MAX_NB	256
+#define VBAN_SAMPLES_MAX_NB	256
+
+struct vban_header {
+	char vban4;			/* contains 'V' 'B', 'A', 'N' */
+	uint8_t format_SR;			/* SR index */
+	uint8_t format_nbs;			/* nb sample per frame (1 to 256) */
+	uint8_t format_nbc;			/* nb channel (1 to 256) */
+	uint8_t format_bit;			/* bit format */
+	char stream_nameVBAN_STREAM_NAME_SIZE;	/* stream name */
+	uint32_t n_frames;			/* growing frame number. */
+} __attribute__ ((packed));
+
+#define VBAN_SR_MAXNUMBER	21
+
+static uint32_t const vban_SRVBAN_SR_MAXNUMBER = {
+    6000, 12000, 24000, 48000, 96000, 192000, 384000,
+    8000, 16000, 32000, 64000, 128000, 256000, 512000,
+    11025, 22050, 44100, 88200, 176400, 352800, 705600
+};
+
+static inline uint8_t vban_sr_index(uint32_t rate)
+{
+	uint8_t i;
+	for (i = 0; i < SPA_N_ELEMENTS(vban_SR); i++) {
+		if (vban_SRi == rate)
+			return i;
+	}
+	return VBAN_SR_MAXNUMBER;
+}
+
+#define VBAN_DATATYPE_U8	0x00
+#define VBAN_DATATYPE_INT16	0x01
+#define VBAN_DATATYPE_INT24	0x02
+#define VBAN_DATATYPE_INT32	0x03
+#define VBAN_DATATYPE_FLOAT32	0x04
+#define VBAN_DATATYPE_FLOAT64	0x05
+#define VBAN_DATATYPE_12BITS	0x06
+#define VBAN_DATATYPE_10BITS	0x07
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /* PIPEWIRE_VBAN_H */

pipewire-0.3.74.tar.gz/src/pipewire/context.c -> pipewire-0.3.76.tar.gz/src/pipewire/context.c Changed

@@ -17,6 +17,7 @@
 #include <spa/support/plugin.h>
 #include <spa/support/plugin-loader.h>
 #include <spa/node/utils.h>
+#include <spa/utils/atomic.h>
 #include <spa/utils/names.h>
 #include <spa/utils/string.h>
 #include <spa/debug/types.h>
@@ -478,61 +479,6 @@
 	spa_hook_list_append(&context->listener_list, listener, events, data);
 }
 
-struct listener_data {
-	struct spa_hook *listener;
-	const struct pw_context_driver_events *events;
-	void *data;
-};
-
-static int
-do_add_listener(struct spa_loop *loop,
-		bool async, uint32_t seq, const void *data, size_t size, void *user_data)
-{
-	struct pw_context *context = user_data;
-	const struct listener_data *d = data;
-	spa_hook_list_append(&context->driver_listener_list,
-			d->listener, d->events, d->data);
-	return 0;
-}
-
-SPA_EXPORT
-void pw_context_driver_add_listener(struct pw_context *context,
-			  struct spa_hook *listener,
-			  const struct pw_context_driver_events *events,
-			  void *data)
-{
-	struct listener_data d = {
-		.listener = listener,
-		.events = events,
-		.data = data };
-	struct pw_impl_node *n;
-	spa_list_for_each(n, &context->driver_list, driver_link) {
-		SPA_FLAG_SET(n->rt.target.activation->flags, PW_NODE_ACTIVATION_FLAG_PROFILER);
-	}
-	pw_loop_invoke(context->data_loop,
-                       do_add_listener, SPA_ID_INVALID, &d, sizeof(d), false, context);
-}
-
-static int do_remove_listener(struct spa_loop *loop,
-		bool async, uint32_t seq, const void *data, size_t size, void *user_data)
-{
-	struct spa_hook *listener = user_data;
-	spa_hook_remove(listener);
-	return 0;
-}
-
-SPA_EXPORT
-void pw_context_driver_remove_listener(struct pw_context *context,
-			  struct spa_hook *listener)
-{
-	struct pw_impl_node *n;
-	spa_list_for_each(n, &context->driver_list, driver_link) {
-		SPA_FLAG_CLEAR(n->rt.target.activation->flags, PW_NODE_ACTIVATION_FLAG_PROFILER);
-	}
-	pw_loop_invoke(context->data_loop,
-                       do_remove_listener, SPA_ID_INVALID, NULL, 0, true, listener);
-}
-
 SPA_EXPORT
 const struct spa_support *pw_context_get_support(struct pw_context *context, uint32_t *n_support)
 {
@@ -1534,10 +1480,10 @@
 			pw_log_debug("%p: apply duration:%"PRIu64" rate:%u/%u", context,
 					n->target_quantum, n->target_rate.num,
 					n->target_rate.denom);
-			SEQ_WRITE(n->rt.position->clock.target_seq);
+			SPA_SEQ_WRITE(n->rt.position->clock.target_seq);
 			n->rt.position->clock.target_duration = n->target_quantum;
 			n->rt.position->clock.target_rate = n->target_rate;
-			SEQ_WRITE(n->rt.position->clock.target_seq);
+			SPA_SEQ_WRITE(n->rt.position->clock.target_seq);
 
 			if (n->info.state < PW_NODE_STATE_RUNNING) {
 				n->rt.position->clock.duration = n->target_quantum;

pipewire-0.3.74.tar.gz/src/pipewire/context.h -> pipewire-0.3.76.tar.gz/src/pipewire/context.h Changed

pipewire-0.3.74.tar.gz/src/pipewire/filter.c -> pipewire-0.3.76.tar.gz/src/pipewire/filter.c Changed

@@ -1222,7 +1222,6 @@
 	struct filter *impl;
 	struct pw_filter *this;
 	const char *str;
-	struct match match;
 	int res;
 
 	ensure_loop(context->main_loop, return NULL);
@@ -1250,28 +1249,6 @@
 	spa_hook_list_init(&impl->hooks);
 	this->properties = props;
 
-	pw_context_conf_update_props(context, "filter.properties", props);
-
-	match = MATCH_INIT(this);
-	pw_context_conf_section_match_rules(context, "filter.rules",
-		&this->properties->dict, execute_match, &match);
-
-	if ((str = getenv("PIPEWIRE_PROPS")) != NULL)
-		pw_properties_update_string(props, str, strlen(str));
-	if ((str = getenv("PIPEWIRE_QUANTUM")) != NULL) {
-		struct spa_fraction q;
-		if (sscanf(str, "%u/%u", &q.num, &q.denom) == 2 && q.denom != 0) {
-			pw_properties_setf(props, PW_KEY_NODE_RATE,
-					"1/%u", q.denom);
-			pw_properties_setf(props, PW_KEY_NODE_LATENCY,
-					"%u/%u", q.num, q.denom);
-		}
-	}
-	if ((str = getenv("PIPEWIRE_LATENCY")) != NULL)
-		pw_properties_set(props, PW_KEY_NODE_LATENCY, str);
-	if ((str = getenv("PIPEWIRE_RATE")) != NULL)
-		pw_properties_set(props, PW_KEY_NODE_RATE, str);
-
 	if (pw_properties_get(props, PW_KEY_NODE_NAME) == NULL && extra) {
 		str = pw_properties_get(extra, PW_KEY_APP_NAME);
 		if (str == NULL)
@@ -1281,6 +1258,11 @@
 		pw_properties_set(props, PW_KEY_NODE_NAME, str);
 	}
 
+	if ((pw_properties_get(props, PW_KEY_NODE_WANT_DRIVER) == NULL))
+		pw_properties_set(props, PW_KEY_NODE_WANT_DRIVER, "true");
+
+	pw_context_conf_update_props(context, "filter.properties", props);
+
 	this->name = name ? strdup(name) : NULL;
 	this->node_id = SPA_ID_INVALID;
 
@@ -1605,6 +1587,8 @@
 {
 	struct filter *impl = SPA_CONTAINER_OF(filter, struct filter, this);
 	struct pw_properties *props = NULL;
+	const char *str;
+	struct match match;
 	int res;
 	uint32_t i;
 
@@ -1658,13 +1642,31 @@
 
 	if (flags & PW_FILTER_FLAG_DRIVER)
 		pw_properties_set(filter->properties, PW_KEY_NODE_DRIVER, "true");
-	if ((pw_properties_get(filter->properties, PW_KEY_NODE_WANT_DRIVER) == NULL))
-		pw_properties_set(filter->properties, PW_KEY_NODE_WANT_DRIVER, "true");
 	if (flags & PW_FILTER_FLAG_TRIGGER) {
 		pw_properties_set(filter->properties, PW_KEY_NODE_TRIGGER, "true");
 		impl->trigger = true;
 	}
 
+	match = MATCH_INIT(filter);
+	pw_context_conf_section_match_rules(impl->context, "filter.rules",
+		&filter->properties->dict, execute_match, &match);
+
+	if ((str = getenv("PIPEWIRE_PROPS")) != NULL)
+		pw_properties_update_string(filter->properties, str, strlen(str));
+	if ((str = getenv("PIPEWIRE_QUANTUM")) != NULL) {
+		struct spa_fraction q;
+		if (sscanf(str, "%u/%u", &q.num, &q.denom) == 2 && q.denom != 0) {
+			pw_properties_setf(filter->properties, PW_KEY_NODE_RATE,
+					"1/%u", q.denom);
+			pw_properties_setf(filter->properties, PW_KEY_NODE_LATENCY,
+					"%u/%u", q.num, q.denom);
+		}
+	}
+	if ((str = getenv("PIPEWIRE_LATENCY")) != NULL)
+		pw_properties_set(filter->properties, PW_KEY_NODE_LATENCY, str);
+	if ((str = getenv("PIPEWIRE_RATE")) != NULL)
+		pw_properties_set(filter->properties, PW_KEY_NODE_RATE, str);
+
 	if (filter->core == NULL) {
 		filter->core = pw_context_connect(impl->context,
 				pw_properties_copy(filter->properties), 0);

pipewire-0.3.74.tar.gz/src/pipewire/impl-client.c -> pipewire-0.3.76.tar.gz/src/pipewire/impl-client.c Changed

pipewire-0.3.74.tar.gz/src/pipewire/impl-link.c -> pipewire-0.3.76.tar.gz/src/pipewire/impl-link.c Changed

@@ -1130,18 +1130,63 @@
 	}
 }
 
+static int check_owner_permissions(struct pw_context *context,
+		struct pw_impl_node *node, uint32_t id, uint32_t permissions)
+{
+	const char *str;
+	struct pw_impl_client *client;
+	struct pw_global *global;
+	uint32_t perms;
+	uint32_t client_id;
+
+	str = pw_properties_get(node->properties, PW_KEY_CLIENT_ID);
+	if (str == NULL)
+		/* node not owned by client */
+		return 0;
+
+	if (!spa_atou32(str, &client_id, 0))
+		/* invalid client_id, something is wrong */
+		return -EIO;
+	if ((global = pw_context_find_global(context, client_id)) == NULL)
+		/* current client can't see the owner client */
+		return -errno;
+	if (!pw_global_is_type(global, PW_TYPE_INTERFACE_Client) ||
+	    (client = global->object) == NULL)
+		/* not the right object, something wrong */
+		return -EIO;
+
+	if ((global = pw_context_find_global(context, id)) == NULL)
+		/* current client can't see node id */
+		return -errno;
+
+	perms = pw_global_get_permissions(global, client);
+	if ((perms & permissions) != permissions)
+		/* owner client can't see other node */
+		return -EPERM;
+
+	return 0;
+}
+
 static int
 check_permission(struct pw_context *context,
 		 struct pw_impl_port *output,
 		 struct pw_impl_port *input,
 		 struct pw_properties *properties)
 {
+	int res;
+	if ((res = check_owner_permissions(context, output->node,
+					input->node->info.id, PW_PERM_R)) < 0)
+		return res;
+	if ((res = check_owner_permissions(context, input->node,
+					output->node->info.id, PW_PERM_R)) < 0)
+		return res;
 	return 0;
 }
 
 static void permissions_changed(struct pw_impl_link *this, struct pw_impl_port *other,
 		struct pw_impl_client *client, uint32_t old, uint32_t new)
 {
+	int res;
 	uint32_t perm;
 
 	perm = pw_global_get_permissions(other->global, client);
@@ -1149,17 +1194,36 @@
 	new &= perm;
 	pw_log_debug("%p: permissions changed %08x -> %08x", this, old, new);
 
-	if (check_permission(this->context, this->output, this->input, this->properties) < 0) {
+	if ((res = check_permission(this->context, this->output, this->input, this->properties)) < 0) {
+		pw_log_info("%p: link permissions removed: %s", this, spa_strerror(res));
 		pw_impl_link_destroy(this);
 	} else if (this->global != NULL) {
 		pw_global_update_permissions(this->global, client, old, new);
 	}
 }
 
+static bool is_port_owner(struct pw_impl_client *client, struct pw_impl_port *port)
+{
+	const char *str;
+	uint32_t client_id;
+
+	str = pw_properties_get(port->node->properties, PW_KEY_CLIENT_ID);
+	if (str == NULL)
+		return false;
+
+	if (!spa_atou32(str, &client_id, 0))
+		return false;
+
+	return client_id == client->info.id;
+}
+
 static void output_permissions_changed(void *data,
 		struct pw_impl_client *client, uint32_t old, uint32_t new)
 {
 	struct pw_impl_link *this = data;
+	if (!is_port_owner(client, this->output) &&
+	    !is_port_owner(client, this->input))
+		return;
 	permissions_changed(this, this->input, client, old, new);
 }
 
@@ -1172,6 +1236,9 @@
 		struct pw_impl_client *client, uint32_t old, uint32_t new)
 {
 	struct pw_impl_link *this = data;
+	if (!is_port_owner(client, this->output) &&
+	    !is_port_owner(client, this->input))
+		return;
 	permissions_changed(this, this->output, client, old, new);
 }

pipewire-0.3.74.tar.gz/src/pipewire/impl-node.c -> pipewire-0.3.76.tar.gz/src/pipewire/impl-node.c Changed

@@ -698,6 +698,13 @@
 			break;
 	}
 	spa_list_append(&n->driver_link, &node->driver_link);
+	pw_context_emit_driver_added(context, node);
+}
+
+static inline void remove_driver(struct pw_context *context, struct pw_impl_node *node)
+{
+	spa_list_remove(&node->driver_link);
+	pw_context_emit_driver_removed(context, node);
 }
 
 static void update_io(struct pw_impl_node *node)
@@ -846,8 +853,8 @@
 static void remove_segment_owner(struct pw_impl_node *driver, uint32_t node_id)
 {
 	struct pw_node_activation *a = driver->rt.target.activation;
-	ATOMIC_CAS(a->segment_owner0, node_id, 0);
-	ATOMIC_CAS(a->segment_owner1, node_id, 0);
+	SPA_ATOMIC_CAS(a->segment_owner0, node_id, 0);
+	SPA_ATOMIC_CAS(a->segment_owner1, node_id, 0);
 }
 
 SPA_EXPORT
@@ -941,8 +948,9 @@
 		if (node->registered) {
 			if (driver)
 				insert_driver(context, node);
-			else
-				spa_list_remove(&node->driver_link);
+			else {
+				remove_driver(context, node);
+			}
 		}
 		if (driver && node->driver_node == node)
 			node->driving = true;
@@ -1158,7 +1166,7 @@
 		pw_log_trace_fp("%p: (%s-%u) state:%p pending:%d/%d", t->node,
 				t->name, t->id, state, state->pending, state->required);
 
-		if (pw_node_activation_state_dec(state, 1)) {
+		if (pw_node_activation_state_dec(state)) {
 			a->status = PW_NODE_ACTIVATION_TRIGGERED;
 			a->signal_time = nsec;
 			if (SPA_UNLIKELY(spa_system_eventfd_write(t->system, t->fd, 1) < 0))
@@ -1249,11 +1257,12 @@
 		/* calculate CPU time when finished */
 		a->signal_time = this->driver_start;
 		calculate_stats(this, a);
-		pw_context_driver_emit_complete(this->context, this);
+		pw_impl_node_rt_emit_complete(this);
+//		pw_context_driver_emit_complete(this->context, this);
 	}
 
 	if (SPA_UNLIKELY(status & SPA_STATUS_DRAINED))
-		pw_context_driver_emit_drained(this->context, this);
+		pw_impl_node_rt_emit_drained(this);
 
 	return status;
 }
@@ -1263,7 +1272,7 @@
 	struct pw_node_activation *a = node->rt.target.activation;
 	struct pw_node_activation_state *state = &a->state0;
 
-	if (pw_node_activation_state_dec(state, 1)) {
+	if (pw_node_activation_state_dec(state)) {
 		uint64_t nsec = get_time_ns(node->data_system);
 		a->status = PW_NODE_ACTIVATION_TRIGGERED;
 		a->signal_time = nsec;
@@ -1394,6 +1403,7 @@
 	spa_list_init(&this->peer_list);
 
 	spa_hook_list_init(&this->listener_list);
+	spa_hook_list_init(&this->rt_listener_list);
 
 	this->info.state = PW_NODE_STATE_CREATING;
 	this->info.props = &this->properties->dict;
@@ -1661,8 +1671,8 @@
 	if (SPA_UNLIKELY(a->position.offset == INT64_MIN))
 		a->position.offset = a->position.clock.position;
 
-	command = ATOMIC_XCHG(a->command, PW_NODE_ACTIVATION_COMMAND_NONE);
-	*reposition_owner = ATOMIC_XCHG(a->reposition_owner, 0);
+	command = SPA_ATOMIC_XCHG(a->command, PW_NODE_ACTIVATION_COMMAND_NONE);
+	*reposition_owner = SPA_ATOMIC_XCHG(a->reposition_owner, 0);
 
 	if (SPA_UNLIKELY(command != PW_NODE_ACTIVATION_COMMAND_NONE)) {
 		pw_log_debug("%p: update command:%u", node, command);
@@ -1726,7 +1736,7 @@
 			pw_log_warn("(%s-%u) sync timeout, going to RUNNING",
 					node->name, node->info.id);
 			check_states(node, nsec);
-			pw_context_driver_emit_timeout(node->context, node);
+			pw_impl_node_rt_emit_timeout(node);
 			all_ready = true;
 		}
 		if (all_ready)
@@ -1775,7 +1785,7 @@
 					state, a->position.clock.duration,
 					state->pending, state->required);
 			check_states(node, nsec);
-			pw_context_driver_emit_incomplete(node->context, node);
+			pw_impl_node_rt_emit_incomplete(node);
 		}
 
 		/* This update is done too late, the driver should do this
@@ -1790,8 +1800,8 @@
 		}
 
 		sync_type = check_updates(node, &reposition_owner);
-		owner0 = ATOMIC_LOAD(a->segment_owner0);
-		owner1 = ATOMIC_LOAD(a->segment_owner1);
+		owner0 = SPA_ATOMIC_LOAD(a->segment_owner0);
+		owner1 = SPA_ATOMIC_LOAD(a->segment_owner1);
 again:
 		all_ready = sync_type == SYNC_CHECK;
 		update_sync = !all_ready;
@@ -1841,7 +1851,7 @@
 
 		update_position(node, all_ready, nsec);
 
-		pw_context_driver_emit_start(node->context, node);
+		pw_impl_node_rt_emit_start(node);
 	}
 	/* this should not happen, driver nodes that are not currently driving
 	 * should not emit the ready callback */
@@ -1904,7 +1914,7 @@
 				missed);
 	}
 
-	pw_context_driver_emit_xrun(this->context, this);
+	pw_impl_node_rt_emit_xrun(this);
 
 	return 0;
 }
@@ -1954,6 +1964,50 @@
 	spa_hook_list_append(&node->listener_list, listener, events, data);
 }
 
+struct listener_data {
+	struct spa_hook *listener;
+	const struct pw_impl_node_rt_events *events;
+	void *data;
+};
+
+static int
+do_add_rt_listener(struct spa_loop *loop,
+		bool async, uint32_t seq, const void *data, size_t size, void *user_data)
+{
+	struct pw_impl_node *node = user_data;
+	const struct listener_data *d = data;
+	spa_hook_list_append(&node->rt_listener_list,
+			d->listener, d->events, d->data);
+	return 0;
+}
+
+SPA_EXPORT
+void pw_impl_node_add_rt_listener(struct pw_impl_node *node,
+			   struct spa_hook *listener,
+			   const struct pw_impl_node_rt_events *events,
+			   void *data)
+{
+	struct listener_data d = { .listener = listener, .events = events, .data = data };
+	pw_loop_invoke(node->data_loop,
+                       do_add_rt_listener, SPA_ID_INVALID, &d, sizeof(d), false, node);
+}
+
+static int do_remove_listener(struct spa_loop *loop,
+		bool async, uint32_t seq, const void *data, size_t size, void *user_data)
+{
+	struct spa_hook *listener = user_data;
+	spa_hook_remove(listener);
+	return 0;
+}
+
+SPA_EXPORT
+void pw_impl_node_remove_rt_listener(struct pw_impl_node *node,
+			  struct spa_hook *listener)
+{
+	pw_loop_invoke(node->data_loop,
+                       do_remove_listener, SPA_ID_INVALID, NULL, 0, true, listener);
+}
+
 /** Destroy a node
  * \param node a node to destroy
  *
@@ -1998,7 +2052,7 @@
 	if (node->registered) {
 		spa_list_remove(&node->link);
 		if (node->driver)
-			spa_list_remove(&node->driver_link);
+			remove_driver(context, node);
 	}
 
 	if (node->node) {

pipewire-0.3.74.tar.gz/src/pipewire/impl-node.h -> pipewire-0.3.76.tar.gz/src/pipewire/impl-node.h Changed

@@ -75,6 +75,23 @@
 	void (*peer_removed) (void *data, struct pw_impl_node *peer);
 };
 
+struct pw_impl_node_rt_events {
+#define PW_VERSION_IMPL_NODE_RT_EVENTS	0
+	uint32_t version;
+	/** the node is drained */
+	void (*drained) (void *data);
+	/** the node had an xrun */
+	void (*xrun) (void *data);
+	/** the driver node starts processing */
+	void (*start) (void *data);
+	/** the driver node completed processing */
+	void (*complete) (void *data);
+	/** the driver node did not complete processing */
+	void (*incomplete) (void *data);
+	/** the node had */
+	void (*timeout) (void *data);
+};
+
 /** Create a new node */
 struct pw_impl_node *
 pw_context_create_node(struct pw_context *context,	/**< the context */
@@ -118,6 +135,14 @@
 			  const struct pw_impl_node_events *events,
 			  void *data);
 
+/** Add an rt_event listener */
+void pw_impl_node_add_rt_listener(struct pw_impl_node *node,
+			  struct spa_hook *listener,
+			  const struct pw_impl_node_rt_events *events,
+			  void *data);
+void pw_impl_node_remove_rt_listener(struct pw_impl_node *node,
+			  struct spa_hook *listener);
+
 /** Iterate the ports in the given direction. The callback should return
  * 0 to fetch the next item, any other value stops the iteration and returns
  * the value. When all callbacks return 0, this function returns 0 when all

pipewire-0.3.74.tar.gz/src/pipewire/pipewire.c -> pipewire-0.3.76.tar.gz/src/pipewire/pipewire.c Changed

pipewire-0.3.74.tar.gz/src/pipewire/private.h -> pipewire-0.3.76.tar.gz/src/pipewire/private.h Changed

@@ -17,6 +17,7 @@
 #include <spa/support/plugin.h>
 #include <spa/pod/builder.h>
 #include <spa/param/latency-utils.h>
+#include <spa/utils/atomic.h>
 #include <spa/utils/ratelimit.h>
 #include <spa/utils/result.h>
 #include <spa/utils/type-info.h>
@@ -359,39 +360,6 @@
 	}										\
 })
 
-#define pw_context_driver_emit(c,m,v,...) spa_hook_list_call_simple(&c->driver_listener_list, struct pw_context_driver_events, m, v, ##__VA_ARGS__)
-#define pw_context_driver_emit_start(c,n)	pw_context_driver_emit(c, start, 0, n)
-#define pw_context_driver_emit_xrun(c,n)	pw_context_driver_emit(c, xrun, 0, n)
-#define pw_context_driver_emit_incomplete(c,n)	pw_context_driver_emit(c, incomplete, 0, n)
-#define pw_context_driver_emit_timeout(c,n)	pw_context_driver_emit(c, timeout, 0, n)
-#define pw_context_driver_emit_drained(c,n)	pw_context_driver_emit(c, drained, 0, n)
-#define pw_context_driver_emit_complete(c,n)	pw_context_driver_emit(c, complete, 0, n)
-
-struct pw_context_driver_events {
-#define PW_VERSION_CONTEXT_DRIVER_EVENTS	0
-	uint32_t version;
-
-	/** The driver graph is started */
-	void (*start) (void *data, struct pw_impl_node *node);
-	/** The driver under/overruns */
-	void (*xrun) (void *data, struct pw_impl_node *node);
-	/** The driver could not complete the graph */
-	void (*incomplete) (void *data, struct pw_impl_node *node);
-	/** The driver got a sync timeout */
-	void (*timeout) (void *data, struct pw_impl_node *node);
-	/** a node drained */
-	void (*drained) (void *data, struct pw_impl_node *node);
-	/** The driver completed the graph */
-	void (*complete) (void *data, struct pw_impl_node *node);
-};
-
-void pw_context_driver_add_listener(struct pw_context *context,
-			  struct spa_hook *listener,
-			  const struct pw_context_driver_events *events,
-			  void *data);
-void pw_context_driver_remove_listener(struct pw_context *context,
-			  struct spa_hook *listener);
-
 #define pw_registry_resource(r,m,v,...) pw_resource_call(r, struct pw_registry_events,m,v,##__VA_ARGS__)
 #define pw_registry_resource_global(r,...)        pw_registry_resource(r,global,0,__VA_ARGS__)
 #define pw_registry_resource_global_remove(r,...) pw_registry_resource(r,global_remove,0,__VA_ARGS__)
@@ -403,6 +371,8 @@
 #define pw_context_emit_check_access(c,cl)	pw_context_emit(c, check_access, 0, cl)
 #define pw_context_emit_global_added(c,g)	pw_context_emit(c, global_added, 0, g)
 #define pw_context_emit_global_removed(c,g)	pw_context_emit(c, global_removed, 0, g)
+#define pw_context_emit_driver_added(c,n)	pw_context_emit(c, driver_added, 1, n)
+#define pw_context_emit_driver_removed(c,n)	pw_context_emit(c, driver_removed, 1, n)
 
 struct pw_context {
 	struct pw_impl_core *core;		/**< core object */
@@ -549,7 +519,7 @@
         state->pending = state->required;
 }
 
-#define pw_node_activation_state_dec(s,c) (__atomic_sub_fetch(&(s)->pending, c, __ATOMIC_SEQ_CST) == 0)
+#define pw_node_activation_state_dec(s) (SPA_ATOMIC_DEC(s->pending) == 0)
 
 struct pw_node_target {
 	struct spa_list link;
@@ -631,25 +601,6 @@
 							 * to update wins */
 };
 
-#define ATOMIC_CAS(v,ov,nv)						\
-({									\
-	__typeof__(v) __ov = (ov);					\
-	__atomic_compare_exchange_n(&(v), &__ov, (nv),			\
-			0, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST);		\
-})
-
-#define ATOMIC_DEC(s)			__atomic_sub_fetch(&(s), 1, __ATOMIC_SEQ_CST)
-#define ATOMIC_INC(s)			__atomic_add_fetch(&(s), 1, __ATOMIC_SEQ_CST)
-#define ATOMIC_LOAD(s)			__atomic_load_n(&(s), __ATOMIC_SEQ_CST)
-#define ATOMIC_STORE(s,v)		__atomic_store_n(&(s), (v), __ATOMIC_SEQ_CST)
-#define ATOMIC_XCHG(s,v)		__atomic_exchange_n(&(s), (v), __ATOMIC_SEQ_CST)
-
-#define SEQ_WRITE(s)			ATOMIC_INC(s)
-#define SEQ_WRITE_SUCCESS(s1,s2)	((s1) + 1 == (s2) && ((s2) & 1) == 0)
-
-#define SEQ_READ(s)			ATOMIC_LOAD(s)
-#define SEQ_READ_SUCCESS(s1,s2)		((s1) == (s2) && ((s2) & 1) == 0)
-
 #define pw_impl_node_emit(o,m,v,...) spa_hook_list_call(&o->listener_list, struct pw_impl_node_events, m, v, ##__VA_ARGS__)
 #define pw_impl_node_emit_destroy(n)			pw_impl_node_emit(n, destroy, 0)
 #define pw_impl_node_emit_free(n)			pw_impl_node_emit(n, free, 0)
@@ -669,6 +620,14 @@
 #define pw_impl_node_emit_peer_added(n,p)		pw_impl_node_emit(n, peer_added, 0, p)
 #define pw_impl_node_emit_peer_removed(n,p)		pw_impl_node_emit(n, peer_removed, 0, p)
 
+#define pw_impl_node_rt_emit(o,m,v,...) spa_hook_list_call(&o->rt_listener_list, struct pw_impl_node_rt_events, m, v, ##__VA_ARGS__)
+#define pw_impl_node_rt_emit_drained(n)			pw_impl_node_rt_emit(n, drained, 0)
+#define pw_impl_node_rt_emit_xrun(n)			pw_impl_node_rt_emit(n, xrun, 0)
+#define pw_impl_node_rt_emit_start(n)			pw_impl_node_rt_emit(n, start, 0)
+#define pw_impl_node_rt_emit_complete(n)		pw_impl_node_rt_emit(n, complete, 0)
+#define pw_impl_node_rt_emit_incomplete(n)		pw_impl_node_rt_emit(n, incomplete, 0)
+#define pw_impl_node_rt_emit_timeout(n)			pw_impl_node_rt_emit(n, timeout, 0)
+
 struct pw_impl_node {
 	struct pw_context *context;		/**< context object */
 	struct spa_list link;		/**< link in context node_list */
@@ -737,6 +696,7 @@
 	struct pw_map output_port_map;		/**< map from port_id to port */
 
 	struct spa_hook_list listener_list;
+	struct spa_hook_list rt_listener_list;
 
 	struct pw_loop *data_loop;		/**< the data loop for this node */
 	struct spa_system *data_system;
@@ -1085,6 +1045,7 @@
 
 	struct pw_impl_node *node;
 	struct spa_hook node_listener;
+	struct spa_hook node_rt_listener;
 
 	struct spa_list controls;
 };
@@ -1321,8 +1282,6 @@
 int pw_settings_expose(struct pw_context *context);
 void pw_settings_clean(struct pw_context *context);
 
-pthread_attr_t *pw_thread_fill_attr(const struct spa_dict *props, pthread_attr_t *attr);
-
 /** \endcond */
 
 #ifdef __cplusplus

pipewire-0.3.74.tar.gz/src/pipewire/stream.c -> pipewire-0.3.76.tar.gz/src/pipewire/stream.c Changed

@@ -17,7 +17,6 @@
 #include <spa/pod/filter.h>
 #include <spa/pod/dynamic.h>
 #include <spa/debug/types.h>
-#include <spa/debug/dict.h>
 
 #define PW_ENABLE_DEPRECATED
 
@@ -84,7 +83,6 @@
 	const char *path;
 
 	struct pw_context *context;
-	struct spa_hook context_listener;
 
 	struct pw_loop *main_loop;
 	struct pw_loop *data_loop;
@@ -436,7 +434,7 @@
 		buffer->this.requested = impl->quantum;
 		res = 1;
 	}
-	pw_log_trace_fp("%p: update buffer:%u size:%"PRIu64, impl, id, buffer->this.requested);
+	pw_log_trace_fp("%p: update buffer:%u req:%"PRIu64, impl, id, buffer->this.requested);
 	return res;
 }
 
@@ -480,6 +478,7 @@
 
 static void call_drained(struct stream *impl)
 {
+	pw_log_info("%p: drained", impl);
 	pw_loop_invoke(impl->main_loop,
 		do_call_drained, 1, NULL, 0, false, impl);
 }
@@ -621,7 +620,7 @@
 {
 	struct spa_io_position *p = impl->rt.position;
 
-	SEQ_WRITE(impl->seq);
+	SPA_SEQ_WRITE(impl->seq);
 	if (SPA_LIKELY(p != NULL)) {
 		impl->time.now = p->clock.nsec;
 		impl->time.rate = p->clock.rate;
@@ -636,7 +635,7 @@
 	}
 	if (SPA_LIKELY(impl->rate_match != NULL))
 		impl->rate_queued = impl->rate_match->delay;
-	SEQ_WRITE(impl->seq);
+	SPA_SEQ_WRITE(impl->seq);
 }
 
 static int impl_send_command(void *object, const struct spa_command *command)
@@ -859,7 +858,7 @@
 
 		while ((b = queue_pop(impl, &impl->dequeued))) {
 			if (b->busy)
-				ATOMIC_DEC(b->busy->count);
+				SPA_ATOMIC_DEC(b->busy->count);
 		}
 	} else
 		clear_queue(impl, &impl->dequeued);
@@ -1039,7 +1038,7 @@
 		pw_log_trace_fp("%p: push %d %p", stream, b->id, io);
 		if (queue_push(impl, &impl->dequeued, b) == 0) {
 			if (b->busy)
-				ATOMIC_INC(b->busy->count);
+				SPA_ATOMIC_INC(b->busy->count);
 		}
 	}
 	if (!queue_is_empty(impl, &impl->dequeued)) {
@@ -1389,6 +1388,7 @@
 	struct pw_stream *stream = data;
 	struct stream *impl = SPA_CONTAINER_OF(stream, struct stream, this);
 	spa_hook_remove(&stream->node_listener);
+	pw_impl_node_remove_rt_listener(stream->node, &stream->node_rt_listener);
 	stream->node = NULL;
 	impl->data_loop = NULL;
 }
@@ -1440,11 +1440,9 @@
 	.error = on_core_error,
 };
 
-static void context_drained(void *data, struct pw_impl_node *node)
+static void node_drained(void *data)
 {
 	struct stream *impl = data;
-	if (impl->this.node != node)
-		return;
 	if (impl->draining && impl->drained) {
 		impl->draining = false;
 		if (impl->io != NULL)
@@ -1453,9 +1451,9 @@
 	}
 }
 
-static const struct pw_context_driver_events context_events = {
-	PW_VERSION_CONTEXT_DRIVER_EVENTS,
-	.drained = context_drained,
+static const struct pw_impl_node_rt_events node_rt_events = {
+	PW_VERSION_IMPL_NODE_RT_EVENTS,
+	.drained = node_drained,
 };
 
 struct match {
@@ -1481,7 +1479,6 @@
 	struct stream *impl;
 	struct pw_stream *this;
 	const char *str;
-	struct match match;
 	int res;
 
 	ensure_loop(context->main_loop, return NULL);
@@ -1513,28 +1510,6 @@
 	spa_hook_list_init(&impl->hooks);
 	this->properties = props;
 
-	pw_context_conf_update_props(context, "stream.properties", props);
-
-	match = MATCH_INIT(this);
-	pw_context_conf_section_match_rules(context, "stream.rules",
-			&this->properties->dict, execute_match, &match);
-
-	if ((str = getenv("PIPEWIRE_PROPS")) != NULL)
-		pw_properties_update_string(props, str, strlen(str));
-	if ((str = getenv("PIPEWIRE_QUANTUM")) != NULL) {
-		struct spa_fraction q;
-		if (sscanf(str, "%u/%u", &q.num, &q.denom) == 2 && q.denom != 0) {
-			pw_properties_setf(props, PW_KEY_NODE_RATE,
-					"1/%u", q.denom);
-			pw_properties_setf(props, PW_KEY_NODE_LATENCY,
-					"%u/%u", q.num, q.denom);
-		}
-	}
-	if ((str = getenv("PIPEWIRE_LATENCY")) != NULL)
-		pw_properties_set(props, PW_KEY_NODE_LATENCY, str);
-	if ((str = getenv("PIPEWIRE_RATE")) != NULL)
-		pw_properties_set(props, PW_KEY_NODE_RATE, str);
-
 	if (pw_properties_get(props, PW_KEY_STREAM_IS_LIVE) == NULL)
 		pw_properties_set(props, PW_KEY_STREAM_IS_LIVE, "true");
 	if (pw_properties_get(props, PW_KEY_NODE_NAME) == NULL && extra) {
@@ -1545,6 +1520,10 @@
 			str = name;
 		pw_properties_set(props, PW_KEY_NODE_NAME, str);
 	}
+	if ((pw_properties_get(props, PW_KEY_NODE_WANT_DRIVER) == NULL))
+		pw_properties_set(props, PW_KEY_NODE_WANT_DRIVER, "true");
+
+	pw_context_conf_update_props(context, "stream.properties", props);
 
 	this->name = name ? strdup(name) : NULL;
 	this->node_id = SPA_ID_INVALID;
@@ -1562,9 +1541,6 @@
 	impl->allow_mlock = context->settings.mem_allow_mlock;
 	impl->warn_mlock = context->settings.mem_warn_mlock;
 
-	pw_context_driver_add_listener(impl->context,
-			&impl->context_listener,
-			&context_events, impl);
 	return impl;
 
 error_properties:
@@ -1730,9 +1706,6 @@
 	spa_hook_list_clean(&impl->hooks);
 	spa_hook_list_clean(&stream->listener_list);
 
-	pw_context_driver_remove_listener(impl->context,
-			&impl->context_listener);
-
 	if (impl->data.context)
 		pw_context_destroy(impl->data.context);
 
@@ -1914,6 +1887,7 @@
 	struct pw_impl_factory *factory;
 	struct pw_properties *props = NULL;
 	const char *str;
+	struct match match;
 	uint32_t i;
 	int res;
 
@@ -2007,49 +1981,69 @@
 	impl->using_trigger = false;
 	stream_set_state(stream, PW_STREAM_STATE_CONNECTING, 0, NULL);
 
-	if ((str = getenv("PIPEWIRE_NODE")) != NULL)
-		pw_properties_set(stream->properties, PW_KEY_TARGET_OBJECT, str);
-	else if (target_id != PW_ID_ANY)
+	if (target_id != PW_ID_ANY)
 		/* XXX this is deprecated but still used by the portal and its apps */
-		pw_properties_setf(stream->properties, PW_KEY_NODE_TARGET, "%d", target_id);
+		if (pw_properties_get(stream->properties, PW_KEY_NODE_TARGET) == NULL)
+			pw_properties_setf(stream->properties, PW_KEY_NODE_TARGET, "%d", target_id);
+	if (flags & PW_STREAM_FLAG_AUTOCONNECT)
+		if (pw_properties_get(stream->properties, PW_KEY_NODE_AUTOCONNECT) == NULL)
+			pw_properties_set(stream->properties, PW_KEY_NODE_AUTOCONNECT, "true");
+	if (flags & PW_STREAM_FLAG_EXCLUSIVE)
+		if (pw_properties_get(stream->properties, PW_KEY_NODE_EXCLUSIVE) == NULL)
+			pw_properties_set(stream->properties, PW_KEY_NODE_EXCLUSIVE, "true");
+	if (flags & PW_STREAM_FLAG_DONT_RECONNECT)
+		if (pw_properties_get(stream->properties, PW_KEY_NODE_DONT_RECONNECT) == NULL)
+			pw_properties_set(stream->properties, PW_KEY_NODE_DONT_RECONNECT, "true");
 
-	if ((str = getenv("PIPEWIRE_AUTOCONNECT")) != NULL)
-		pw_properties_set(stream->properties,
-				PW_KEY_NODE_AUTOCONNECT, spa_atob(str) ? "true" : "false");
-	else if ((flags & PW_STREAM_FLAG_AUTOCONNECT) &&
-	    pw_properties_get(stream->properties, PW_KEY_NODE_AUTOCONNECT) == NULL) {
-		pw_properties_set(stream->properties, PW_KEY_NODE_AUTOCONNECT, "true");
-	}
 	if (flags & PW_STREAM_FLAG_DRIVER)
 		pw_properties_set(stream->properties, PW_KEY_NODE_DRIVER, "true");
-	if ((pw_properties_get(stream->properties, PW_KEY_NODE_WANT_DRIVER) == NULL))
-		pw_properties_set(stream->properties, PW_KEY_NODE_WANT_DRIVER, "true");
-
-	if (flags & PW_STREAM_FLAG_EXCLUSIVE)
-		pw_properties_set(stream->properties, PW_KEY_NODE_EXCLUSIVE, "true");
-	if (flags & PW_STREAM_FLAG_DONT_RECONNECT)
-		pw_properties_set(stream->properties, PW_KEY_NODE_DONT_RECONNECT, "true");
 	if (flags & PW_STREAM_FLAG_TRIGGER) {
 		pw_properties_set(stream->properties, PW_KEY_NODE_TRIGGER, "true");
 		impl->trigger = true;
 	}
-
-	if ((str = pw_properties_get(stream->properties, "mem.warn-mlock")) != NULL)
-		impl->warn_mlock = pw_properties_parse_bool(str);
-
 	if ((pw_properties_get(stream->properties, PW_KEY_MEDIA_CLASS) == NULL)) {
 		const char *media_type = pw_properties_get(stream->properties, PW_KEY_MEDIA_TYPE);
 		pw_properties_setf(stream->properties, PW_KEY_MEDIA_CLASS, "Stream/%s/%s",
 				direction == PW_DIRECTION_INPUT ? "Input" : "Output",
 				media_type ? media_type : get_media_class(impl));
 	}
-
 	if ((str = pw_properties_get(stream->properties, PW_KEY_FORMAT_DSP)) != NULL)
 		pw_properties_set(impl->port_props, PW_KEY_FORMAT_DSP, str);
 	else if (impl->media_type == SPA_MEDIA_TYPE_application &&
 	    impl->media_subtype == SPA_MEDIA_SUBTYPE_control)
 		pw_properties_set(impl->port_props, PW_KEY_FORMAT_DSP, "8 bit raw midi");
 
+	match = MATCH_INIT(stream);
+	pw_context_conf_section_match_rules(impl->context, "stream.rules",
+			&stream->properties->dict, execute_match, &match);
+
+	if ((str = getenv("PIPEWIRE_NODE")) != NULL)
+		pw_properties_set(stream->properties, PW_KEY_TARGET_OBJECT, str);
+	if ((str = getenv("PIPEWIRE_AUTOCONNECT")) != NULL)
+		pw_properties_set(stream->properties,
+				PW_KEY_NODE_AUTOCONNECT, spa_atob(str) ? "true" : "false");
+
+	if ((str = getenv("PIPEWIRE_PROPS")) != NULL)
+		pw_properties_update_string(stream->properties, str, strlen(str));
+	if ((str = getenv("PIPEWIRE_QUANTUM")) != NULL) {
+		struct spa_fraction q;
+		if (sscanf(str, "%u/%u", &q.num, &q.denom) == 2 && q.denom != 0) {
+			pw_properties_setf(stream->properties, PW_KEY_NODE_RATE,
+					"1/%u", q.denom);
+			pw_properties_setf(stream->properties, PW_KEY_NODE_LATENCY,
+					"%u/%u", q.num, q.denom);
+		}
+	}
+	if ((str = getenv("PIPEWIRE_LATENCY")) != NULL)
+		pw_properties_set(stream->properties, PW_KEY_NODE_LATENCY, str);
+	if ((str = getenv("PIPEWIRE_RATE")) != NULL)
+		pw_properties_set(stream->properties, PW_KEY_NODE_RATE, str);
+
+	if ((str = pw_properties_get(stream->properties, "mem.warn-mlock")) != NULL)
+		impl->warn_mlock = pw_properties_parse_bool(str);
+	if ((str = pw_properties_get(stream->properties, "mem.allow-mlock")) != NULL)
+		impl->allow_mlock = pw_properties_parse_bool(str);
+
 	impl->port_info.props = &impl->port_props->dict;
 
 	if (stream->core == NULL) {
@@ -2125,6 +2119,8 @@
 	pw_proxy_add_listener(stream->proxy, &stream->proxy_listener, &proxy_events, stream);
 
 	pw_impl_node_add_listener(stream->node, &stream->node_listener, &node_events, stream);
+	pw_impl_node_add_rt_listener(stream->node, &stream->node_rt_listener,
+			&node_rt_events, stream);
 
 	return 0;
 
@@ -2340,12 +2336,12 @@
 	uint32_t buffered, quantum, index;
 
 	do {
-		seq1 = SEQ_READ(impl->seq);
+		seq1 = SPA_SEQ_READ(impl->seq);
 		memcpy(time, &impl->time, SPA_MIN(size, sizeof(struct pw_time)));
 		buffered = impl->rate_queued;
 		quantum = impl->quantum;
-		seq2 = SEQ_READ(impl->seq);
-	} while (!SEQ_READ_SUCCESS(seq1, seq2));
+		seq2 = SPA_SEQ_READ(impl->seq);
+	} while (!SPA_SEQ_READ_SUCCESS(seq1, seq2));
 
 	if (impl->direction == SPA_DIRECTION_INPUT)
 		time->queued = (int64_t)(time->queued - impl->dequeued.outcount);
@@ -2394,11 +2390,12 @@
 		errno = -res;
 		return NULL;
 	}
-	pw_log_trace_fp("%p: dequeue buffer %d size:%"PRIu64, stream, b->id, b->this.size);
+	pw_log_trace_fp("%p: dequeue buffer %d size:%"PRIu64" req:%"PRIu64,
+			stream, b->id, b->this.size, b->this.requested);
 
 	if (b->busy && impl->direction == SPA_DIRECTION_OUTPUT) {
-		if (ATOMIC_INC(b->busy->count) > 1) {
-			ATOMIC_DEC(b->busy->count);
+		if (SPA_ATOMIC_INC(b->busy->count) > 1) {
+			SPA_ATOMIC_DEC(b->busy->count);
 			queue_push(impl, &impl->dequeued, b);
 			pw_log_trace_fp("%p: buffer busy", stream);
 			errno = EBUSY;
@@ -2416,9 +2413,10 @@
 	int res;
 
 	if (b->busy)
-		ATOMIC_DEC(b->busy->count);
+		SPA_ATOMIC_DEC(b->busy->count);
 
-	pw_log_trace_fp("%p: queue buffer %d", stream, b->id);
+	pw_log_trace_fp("%p: queue buffer %d size:%"PRIu64, stream, b->id,
+			b->this.size);
 	if ((res = queue_push(impl, &impl->queued, b)) < 0)
 		return res;

pipewire-0.3.74.tar.gz/src/pipewire/thread-loop.c -> pipewire-0.3.76.tar.gz/src/pipewire/thread-loop.c Changed

pipewire-0.3.74.tar.gz/src/pipewire/thread.c -> pipewire-0.3.76.tar.gz/src/pipewire/thread.c Changed

pipewire-0.3.74.tar.gz/src/pipewire/thread.h -> pipewire-0.3.76.tar.gz/src/pipewire/thread.h Changed

pipewire-0.3.74.tar.gz/src/pipewire/version.h.in -> pipewire-0.3.76.tar.gz/src/pipewire/version.h.in Changed

pipewire-0.3.74.tar.gz/src/tools/pw-cat.c -> pipewire-0.3.76.tar.gz/src/tools/pw-cat.c Changed

pipewire-0.3.74.tar.gz/test/test-context.c -> pipewire-0.3.76.tar.gz/test/test-context.c Changed

@@ -27,6 +27,8 @@
 		void (*check_access) (void *data, struct pw_impl_client *client);
 		void (*global_added) (void *data, struct pw_global *global);
 		void (*global_removed) (void *data, struct pw_global *global);
+		void (*driver_added) (void *data, struct pw_impl_node *node);
+		void (*driver_removed) (void *data, struct pw_impl_node *node);
 	} test = { PW_VERSION_CONTEXT_EVENTS, NULL };
 
 	pw_init(0, NULL);
@@ -36,8 +38,10 @@
 	TEST_FUNC(ev, test, check_access);
 	TEST_FUNC(ev, test, global_added);
 	TEST_FUNC(ev, test, global_removed);
+	TEST_FUNC(ev, test, driver_added);
+	TEST_FUNC(ev, test, driver_removed);
 
-	pwtest_int_eq(PW_VERSION_CONTEXT_EVENTS, 0);
+	pwtest_int_eq(PW_VERSION_CONTEXT_EVENTS, 1);
 	pwtest_int_eq(sizeof(ev), sizeof(test));
 
 	pw_deinit();
@@ -65,7 +69,14 @@
 {
 	pwtest_fail_if_reached();
 }
-
+static void context_driver_added_error(void *data, struct pw_impl_node *node)
+{
+	pwtest_fail_if_reached();
+}
+static void context_driver_removed_error(void *data, struct pw_impl_node *node)
+{
+	pwtest_fail_if_reached();
+}
 static const struct pw_context_events context_events_error =
 {
 	PW_VERSION_CONTEXT_EVENTS,
@@ -74,6 +85,8 @@
 	.check_access = context_check_access_error,
 	.global_added = context_global_added_error,
 	.global_removed = context_global_removed_error,
+	.driver_added = context_driver_added_error,
+	.driver_removed = context_driver_removed_error,
 };
 
 static int destroy_count = 0;

Overview

Request 5813 (accepted)

Request History