ericek111/ts3j
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

Add a PipeWire audio backend and stereo capture/playback

Browse Source 
Capture and playback now go through AudioCapture/AudioPlayback interfaces
with two implementations: PipeWire, used for the system default and for
every pw: device when a session is reachable, and Java Sound for the raw
ALSA devices and other platforms. The PipeWire binding is FFM-based
(PipeWireLibrary, PipeWireSession, PipeWireStream, SpaPod) and enumerates
the graph's real devices, so each stream shows up separately in volume
mixers.

Lines are opened with a negotiated channel count instead of a fixed mono
format: OPUS_MUSIC transmits the stereo capture as-is, OPUS_VOICE keeps
transmitting the mono downmix so the pre-processing chain and VAD still
see a single channel. Java Sound enumeration asks only for "a mixer that
can capture/play" so the PipeWire and PulseAudio ALSA plugins are no
longer hidden by their narrow format lists.

The Java Sound classes are renamed to Desktop* to match what they now
are, and JUnit plus a SpaPod round-trip test are wired into the build.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
ericek111
2026-08-14 05:30:46 +00:00
parent d5018729d8
commit 0333b92ce5
23 changed files with 2056 additions and 242 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
ts3-client/desktop/pom.xml
View File

@@ -12,7 +12,7 @@
<artifactId>ts3-client-desktop</artifactId>
<name>TS3J Client Desktop Audio</name>
<description>Desktop audio backend: Java Sound capture/playback and native Opus via the FFM API</description>
<description>Desktop audio backend: PipeWire/Java Sound capture/playback and native Opus via the FFM API</description>
<dependencies>
<dependency>
@@ -23,6 +23,12 @@
<groupId>com.github.manevolent</groupId>
<artifactId>ts3j</artifactId>
</dependency>
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter</artifactId>
<version>5.13.4</version>
<scope>test</scope>
</dependency>
</dependencies>
<build>

26
ts3-client/desktop/src/main/java/com/ts3client/audio/desktop/AudioCapture.java Normal file
View File

@@ -0,0 +1,26 @@
package com.ts3client.audio.desktop;
/**
* An open microphone line, delivering 16-bit little-endian PCM at
* {@link AudioDevices#SAMPLE_RATE}.
*
* <p>Modelled on {@link javax.sound.sampled.TargetDataLine} so the capture pipeline does
* not care whether the audio comes from PipeWire or Java Sound.
*/
public interface AudioCapture extends AutoCloseable {
/** Channel count actually negotiated with the device. */
int channels();
/** Begins capturing; audio read before this call is not delivered. */
void start();
/**
* Blocks until {@code length} bytes are captured. Returns fewer bytes only when the
* line is closing or the calling thread was interrupted.
*/
int read(byte[] buffer, int offset, int length);
@Override
void close();
}

242
ts3-client/desktop/src/main/java/com/ts3client/audio/desktop/AudioDevices.java
View File

@@ -1,9 +1,12 @@
package com.ts3client.audio.desktop;
import com.ts3client.audio.desktop.pipewire.PipeWire;
import javax.sound.sampled.AudioFormat;
import javax.sound.sampled.AudioSystem;
import javax.sound.sampled.DataLine;
import javax.sound.sampled.Line;
import javax.sound.sampled.LineUnavailableException;
import javax.sound.sampled.Mixer;
import javax.sound.sampled.SourceDataLine;
import javax.sound.sampled.TargetDataLine;
@@ -11,74 +14,209 @@ import java.util.ArrayList;
import java.util.List;
/**
* Helpers for enumerating and opening capture/playback lines by mixer name.
* Enumerates audio devices and opens capture/playback lines on them.
*
* <p>The TS3 protocol uses 48&nbsp;kHz, 16-bit, mono/stereo signed little-endian
* PCM for Opus. We standardise on that format everywhere.
* <p>The TS3 protocol uses 48&nbsp;kHz 16-bit signed little-endian PCM for Opus, mono for
* {@code OPUS_VOICE} and stereo for {@code OPUS_MUSIC}, so the channel count is
* negotiated per line rather than fixed.
*
* <p>Two backends sit behind {@link AudioCapture} / {@link AudioPlayback}:
* <ul>
* <li>{@link PipeWire} — used for the system default and for every {@code pw:} device
* when a PipeWire session is reachable. It sees the graph's real devices and each
* stream shows up separately in volume mixers.</li>
* <li>Java Sound — everywhere else, and for the raw ALSA devices which stay in the
* list because they still work without PipeWire and can be opened exclusively.</li>
* </ul>
*
* <p>Java Sound enumeration deliberately asks only for "a mixer that can do
* capture/playback" instead of "a mixer that can do exactly this format": the PipeWire
* and PulseAudio ALSA plugins advertise a narrow set of formats (typically stereo only),
* so a format-specific query hides them entirely. The desired format is negotiated in
* {@link #openCapture} / {@link #openPlayback}, which fall back through the supported
* channel counts.
*/
public final class AudioDevices {
public static final int SAMPLE_RATE = 48_000;
public static final int FRAME_SIZE = 960; // 20 ms @ 48 kHz
public static final AudioFormat CAPTURE_FORMAT =
new AudioFormat(SAMPLE_RATE, 16, 1, true, false); // mono capture
public static final AudioFormat PLAYBACK_FORMAT =
new AudioFormat(SAMPLE_RATE, 16, 1, true, false); // mono playback
public static final int MAX_CHANNELS = 2;
/** Buffer size, in 20 ms frames, requested when opening a Java Sound line. */
private static final int BUFFER_FRAMES = 8;
/** Marks a device id as a PipeWire node name rather than a Java Sound mixer name. */
private static final String PIPEWIRE_PREFIX = "pw:";
/** A selectable audio device. {@code id} is what gets persisted in the settings. */
public record Device(String id, String label) {
/** The implicit entry that lets the platform (PipeWire, Pulse, ALSA) choose. */
public static final Device DEFAULT = new Device("", "(System default)");
@Override
public String toString() {
return label;
}
}
private AudioDevices() {
}
/** Names of mixers that can provide microphone (capture) lines. */
public static List<String> inputDeviceNames() {
return deviceNames(new DataLine.Info(TargetDataLine.class, CAPTURE_FORMAT));
/** 48 kHz signed 16-bit little-endian PCM with the given channel count. */
public static AudioFormat format(int channels) {
return new AudioFormat(SAMPLE_RATE, 16, channels, true, false);
}
/** Names of mixers that can provide speaker (playback) lines. */
public static List<String> outputDeviceNames() {
return deviceNames(new DataLine.Info(SourceDataLine.class, PLAYBACK_FORMAT));
/** Devices that can provide microphone (capture) lines, system default first. */
public static List<Device> inputDevices() {
return devices(TargetDataLine.class, false);
}
private static List<String> deviceNames(Line.Info lineInfo) {
List<String> names = new ArrayList<>();
/** Devices that can provide speaker (playback) lines, system default first. */
public static List<Device> outputDevices() {
return devices(SourceDataLine.class, true);
}
private static List<Device> devices(Class<?> lineClass, boolean sinks) {
List<Device> devices = new ArrayList<>();
devices.add(Device.DEFAULT);
for (PipeWire.Node node : PipeWire.nodes()) {
if (node.sink() == sinks) {
devices.add(new Device(PIPEWIRE_PREFIX + node.name(), node.description()));
}
}
Line.Info anyFormat = new Line.Info(lineClass);
for (Mixer.Info mi : AudioSystem.getMixerInfo()) {
String name = mi.getName();
if (name == null || name.isEmpty()) continue;
// The ALSA "default" PCM is what the system-default entry already opens.
if (isDefaultMixer(name)) continue;
if (!AudioSystem.getMixer(mi).isLineSupported(anyFormat)) continue;
if (devices.stream().anyMatch(d -> d.id().equals(name))) continue;
devices.add(new Device(name, "ALSA: " + name));
}
return devices;
}
/**
* Java Sound names its ALSA {@code default} mixer after the PCM it opened, which under
* pipewire-alsa is our own stream node (e.g. {@code alsa_playback.java [default]}), so
* only the bracketed device part is stable.
*/
private static boolean isDefaultMixer(String mixerName) {
return mixerName.endsWith("[default]");
}
/**
* Opens a capture line, preferring {@code preferredChannels} and falling back to the
* other channel count if the device won't take it. Inspect {@link AudioCapture#channels()}
* for what was actually opened.
*/
public static AudioCapture openCapture(String deviceId, int preferredChannels)
throws LineUnavailableException {
if (usePipeWire(deviceId)) {
try {
return PipeWire.openCapture("TS3J Microphone", pipeWireNode(deviceId),
SAMPLE_RATE, clampChannels(preferredChannels));
} catch (Exception e) {
if (isPipeWireDevice(deviceId)) throw unavailable(true, deviceId, e);
// The default device: Java Sound may still reach it through ALSA.
}
}
return new JavaSoundCapture(open(TargetDataLine.class, deviceId, preferredChannels));
}
/** Opens a playback line; see {@link #openCapture} for the channel negotiation. */
public static AudioPlayback openPlayback(String deviceId, int preferredChannels)
throws LineUnavailableException {
if (usePipeWire(deviceId)) {
try {
return PipeWire.openPlayback("TS3J Playback", pipeWireNode(deviceId),
SAMPLE_RATE, clampChannels(preferredChannels));
} catch (Exception e) {
if (isPipeWireDevice(deviceId)) throw unavailable(false, deviceId, e);
}
}
return new JavaSoundPlayback(open(SourceDataLine.class, deviceId, preferredChannels));
}
/**
* PipeWire handles its own devices and, when a session is running, the system default
* too — a Java Sound "default" would only reach the same graph through the ALSA
* compatibility plugin, with no way to pick the device.
*/
private static boolean usePipeWire(String deviceId) {
if (!isPipeWireDevice(deviceId) && !isDefault(deviceId)) return false;
return PipeWire.isAvailable();
}
private static boolean isDefault(String deviceId) {
return deviceId == null || deviceId.isEmpty();
}
private static boolean isPipeWireDevice(String deviceId) {
return deviceId != null && deviceId.startsWith(PIPEWIRE_PREFIX);
}
/** The PipeWire node name behind a device id; empty for "let the server choose". */
private static String pipeWireNode(String deviceId) {
return isPipeWireDevice(deviceId) ? deviceId.substring(PIPEWIRE_PREFIX.length()) : "";
}
/** PipeWire converts freely, so any channel count works; keep it in the range we handle. */
private static int clampChannels(int channels) {
return Math.max(1, Math.min(MAX_CHANNELS, channels));
}
private static <T extends DataLine> T open(Class<T> lineClass, String deviceId, int preferredChannels)
throws LineUnavailableException {
Mixer mixer = findMixer(deviceId, lineClass); // null -> platform default
Exception failure = null;
for (int channels : channelOrder(preferredChannels)) {
AudioFormat fmt = format(channels);
DataLine.Info info = new DataLine.Info(lineClass, fmt);
try {
T line = lineClass.cast(
(mixer != null) ? mixer.getLine(info) : AudioSystem.getLine(info));
openLine(line, fmt, FRAME_SIZE * 2 * fmt.getChannels() * BUFFER_FRAMES);
return line;
} catch (Exception e) {
failure = e;
}
}
throw unavailable(lineClass == TargetDataLine.class, deviceId, failure);
}
private static LineUnavailableException unavailable(boolean capture, String deviceId,
Exception failure) {
LineUnavailableException e = new LineUnavailableException(
"No usable " + (capture ? "capture" : "playback") + " line for "
+ (isDefault(deviceId) ? "the system default" : deviceId));
if (failure != null) e.initCause(failure);
return e;
}
/** {@code open(format, bufferSize)} lives on the two line interfaces, not on {@code DataLine}. */
private static void openLine(DataLine line, AudioFormat fmt, int bufferSize)
throws LineUnavailableException {
if (line instanceof TargetDataLine target) target.open(fmt, bufferSize);
else ((SourceDataLine) line).open(fmt, bufferSize);
}
private static int[] channelOrder(int preferred) {
return preferred >= 2 ? new int[]{2, 1} : new int[]{1, 2};
}
private static Mixer findMixer(String deviceId, Class<?> lineClass) {
if (isDefault(deviceId) || isPipeWireDevice(deviceId)) return null;
Line.Info anyFormat = new Line.Info(lineClass);
for (Mixer.Info mi : AudioSystem.getMixerInfo()) {
if (!deviceId.equals(mi.getName())) continue;
Mixer mixer = AudioSystem.getMixer(mi);
if (mixer.isLineSupported(lineInfo)) {
String name = mi.getName();
if (name != null && !names.contains(name)) {
names.add(name);
}
}
if (mixer.isLineSupported(anyFormat)) return mixer;
}
return names;
}
public static TargetDataLine openCapture(String deviceName) throws Exception {
DataLine.Info info = new DataLine.Info(TargetDataLine.class, CAPTURE_FORMAT);
Mixer.Info mixerInfo = findMixer(deviceName, info);
TargetDataLine line = (mixerInfo != null)
? (TargetDataLine) AudioSystem.getMixer(mixerInfo).getLine(info)
: (TargetDataLine) AudioSystem.getLine(info);
line.open(CAPTURE_FORMAT, FRAME_SIZE * 2 * 8); // ~8 frame buffer
return line;
}
public static SourceDataLine openPlayback(String deviceName) throws Exception {
DataLine.Info info = new DataLine.Info(SourceDataLine.class, PLAYBACK_FORMAT);
Mixer.Info mixerInfo = findMixer(deviceName, info);
SourceDataLine line = (mixerInfo != null)
? (SourceDataLine) AudioSystem.getMixer(mixerInfo).getLine(info)
: (SourceDataLine) AudioSystem.getLine(info);
line.open(PLAYBACK_FORMAT, FRAME_SIZE * 2 * 8);
return line;
}
private static Mixer.Info findMixer(String deviceName, Line.Info lineInfo) {
if (deviceName == null || deviceName.isEmpty()) return null;
for (Mixer.Info mi : AudioSystem.getMixerInfo()) {
if (deviceName.equals(mi.getName()) && AudioSystem.getMixer(mi).isLineSupported(lineInfo)) {
return mi;
}
}
return null; // fall back to system default
return null; // device went away: fall back to the system default
}
}

26
ts3-client/desktop/src/main/java/com/ts3client/audio/desktop/AudioPlayback.java Normal file
View File

@@ -0,0 +1,26 @@
package com.ts3client.audio.desktop;
/**
* An open speaker line, accepting 16-bit little-endian PCM at
* {@link AudioDevices#SAMPLE_RATE}.
*
* <p>Modelled on {@link javax.sound.sampled.SourceDataLine} so the playback pipeline does
* not care whether the audio goes to PipeWire or Java Sound.
*/
public interface AudioPlayback extends AutoCloseable {
/** Channel count actually negotiated with the device. */
int channels();
/** Begins playing whatever is written from now on. */
void start();
/** Blocks until the audio has been queued for playback. */
void write(byte[] buffer, int offset, int length);
/** Blocks until everything already written has been played out. */
void drain();
@Override
void close();
}

42
ts3-client/desktop/src/main/java/com/ts3client/audio/desktop/DesktopAudioBackend.java Normal file
View File

@@ -0,0 +1,42 @@
package com.ts3client.audio.desktop;
import com.ts3client.audio.AudioBackend;
import com.ts3client.audio.VoiceInput;
import com.ts3client.audio.VoiceOutput;
import com.ts3client.audio.desktop.pipewire.PipeWire;
import com.ts3client.config.Settings;
/**
* Desktop audio backend: PipeWire for capture/playback where it is running, Java Sound
* everywhere else, and native Opus (via the FFM API) for the codec.
*/
public final class DesktopAudioBackend implements AudioBackend {
@Override
public VoiceInput createInput(Settings settings) {
return new DesktopVoiceInput(settings);
}
@Override
public VoiceOutput createOutput(Settings settings) {
return new DesktopVoiceOutput(settings.outputDevice);
}
@Override
public String description() {
return codec() + ", " + audioSystem();
}
private static String codec() {
try {
return "Opus " + Opus.getVersionString();
} catch (Throwable t) {
return "Opus (native library unavailable)";
}
}
private static String audioSystem() {
String pipeWire = PipeWire.version();
return pipeWire != null ? "PipeWire " + pipeWire : "Java Sound";
}
}

190
ts3-client/desktop/src/main/java/com/ts3client/audio/desktop/JavaSoundVoiceInput.java → ts3-client/desktop/src/main/java/com/ts3client/audio/desktop/DesktopVoiceInput.java
View File

@@ -7,21 +7,26 @@ import com.ts3client.audio.SpeechDetector;
import com.ts3client.audio.VoiceInput;
import com.ts3client.config.Settings;
import javax.sound.sampled.TargetDataLine;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
/**
* Java Sound {@link VoiceInput}: captures the microphone, applies voice-activation
* Desktop {@link VoiceInput}: captures the microphone, applies voice-activation
* or push-to-talk gating, and Opus-encodes 20&nbsp;ms frames.
*
* <p>The capture line comes from {@link AudioDevices}, which picks PipeWire or Java
* Sound; it is opened in stereo when the device offers it. Voice
* ({@code OPUS_VOICE}) is transmitted mono, from the downmix, so the pre-processing
* chain and VAD see a single channel; the music codec ({@code OPUS_MUSIC}) transmits
* the stereo capture as-is.
*
* <p>A dedicated capture thread fills a small packet queue while ts3j polls
* {@link #isReady()} / {@link #provide()} on its own timer, decoupling capture from
* network sending. When the gate closes and the queue drains {@link #isReady()}
* returns {@code false}, prompting ts3j to emit the terminating empty voice packet.
*/
public final class JavaSoundVoiceInput implements VoiceInput {
public final class DesktopVoiceInput implements VoiceInput {
private static final int HANGOVER_FRAMES = 15; // ~300 ms of tail after level drops
@@ -49,16 +54,19 @@ public final class JavaSoundVoiceInput implements VoiceInput {
private final Object encoderLock = new Object();
private volatile OpusParameters params;
private OpusParameters appliedParams;
private int encoderApplication = -1;
private volatile int encoderChannels = 1;
private volatile int captureChannels = 1;
private Thread captureThread;
private TargetDataLine line;
private AudioCapture line;
private OpusEncoder encoder;
private int hangover;
private boolean lastTransmitting;
public JavaSoundVoiceInput(Settings settings) {
public DesktopVoiceInput(Settings settings) {
this.deviceName = settings.inputDevice;
this.mode = settings.inputMode;
this.vadMode = settings.vadMode;
@@ -67,7 +75,7 @@ public final class JavaSoundVoiceInput implements VoiceInput {
this.vadOverPtt = settings.vadOverPtt;
this.inputGain = settings.inputVolume;
this.params = OpusParameters.from(settings);
this.codec = params.music ? CodecType.OPUS_MUSIC : CodecType.OPUS_VOICE;
this.codec = codecFor(params);
enhancer.setNoiseSuppression(settings.denoise);
enhancer.setDenoiserLevel(settings.denoiserLevel);
enhancer.setTypingAttenuation(settings.typingAttenuation);
@@ -117,25 +125,16 @@ public final class JavaSoundVoiceInput implements VoiceInput {
enhancer.setAgc(enabled);
}
/**
* Stages new encoder settings; the capture thread picks them up on the next frame
* (or {@link #start()} applies them directly when idle).
*/
public void setOpusParameters(OpusParameters p) {
this.params = p;
this.codec = p.music ? CodecType.OPUS_MUSIC : CodecType.OPUS_VOICE;
synchronized (encoderLock) {
if (encoder == null) return;
int desiredApplication = applicationFor(p);
if (desiredApplication != encoderApplication) {
// The Opus application (VOIP vs AUDIO) can only be chosen at
// creation, so switching voice<->music requires a new encoder.
OpusEncoder replacement = new OpusEncoder(
AudioDevices.SAMPLE_RATE, AudioDevices.FRAME_SIZE, 1, desiredApplication);
configureEncoder(replacement, p);
OpusEncoder previous = encoder;
encoder = replacement;
encoderApplication = desiredApplication;
previous.close();
} else {
configureEncoder(encoder, p);
}
// While capturing, the codec flag flips together with the encoder swap so
// no packet is ever tagged with a codec it wasn't encoded for.
if (encoder == null) this.codec = codecFor(p);
}
}
@@ -143,6 +142,40 @@ public final class JavaSoundVoiceInput implements VoiceInput {
return p.music ? Opus.OPUS_APPLICATION_AUDIO : Opus.OPUS_APPLICATION_VOIP;
}
private static CodecType codecFor(OpusParameters p) {
return p.music ? CodecType.OPUS_MUSIC : CodecType.OPUS_VOICE;
}
/**
* Channels to transmit: TeamSpeak's {@code OPUS_MUSIC} stream is stereo,
* {@code OPUS_VOICE} is mono. A mono-only capture device caps this at one.
*/
private int channelsFor(OpusParameters p) {
return (p.music && captureChannels >= 2) ? 2 : 1;
}
/** Creates or reconfigures the encoder to match {@code p}. Call under {@link #encoderLock}. */
private void applyParams(OpusParameters p) {
int application = applicationFor(p);
int channels = channelsFor(p);
// Application (VOIP vs AUDIO) and channel count are fixed at creation, so
// switching voice<->music means building a new encoder.
if (encoder == null || application != encoderApplication || channels != encoderChannels) {
OpusEncoder replacement = new OpusEncoder(
AudioDevices.SAMPLE_RATE, AudioDevices.FRAME_SIZE, channels, application);
configureEncoder(replacement, p);
OpusEncoder previous = encoder;
encoder = replacement;
encoderApplication = application;
encoderChannels = channels;
if (previous != null) previous.close();
} else {
configureEncoder(encoder, p);
}
codec = codecFor(p);
appliedParams = p;
}
private static void configureEncoder(OpusEncoder enc, OpusParameters p) {
enc.setBitrate(p.bitrate);
enc.setComplexity(p.complexity);
@@ -173,13 +206,10 @@ public final class JavaSoundVoiceInput implements VoiceInput {
public synchronized void start() {
if (running.get()) return;
try {
line = AudioDevices.openCapture(deviceName);
OpusParameters p = params;
line = AudioDevices.openCapture(deviceName, AudioDevices.MAX_CHANNELS);
captureChannels = line.channels();
synchronized (encoderLock) {
encoderApplication = applicationFor(p);
encoder = new OpusEncoder(
AudioDevices.SAMPLE_RATE, AudioDevices.FRAME_SIZE, 1, encoderApplication);
configureEncoder(encoder, p);
applyParams(params);
}
} catch (Throwable t) {
cleanup();
@@ -206,11 +236,7 @@ public final class JavaSoundVoiceInput implements VoiceInput {
private void cleanup() {
if (line != null) {
try {
line.stop();
line.close();
} catch (Exception ignored) {
}
line.close();
line = null;
}
synchronized (encoderLock) {
@@ -221,32 +247,30 @@ public final class JavaSoundVoiceInput implements VoiceInput {
}
encoder = null;
encoderApplication = -1;
appliedParams = null;
}
}
}
private void captureLoop() {
final int frameSamples = AudioDevices.FRAME_SIZE;
final byte[] buf = new byte[frameSamples * 2];
final float[] pcm = new float[frameSamples];
final int channels = captureChannels;
final byte[] buf = new byte[frameSamples * 2 * channels];
final float[] pcm = new float[frameSamples * channels]; // interleaved capture
// Analysis (level, VAD) always runs on a mono downmix; with a mono device that
// is the capture buffer itself, so nothing is copied.
final float[] mono = (channels == 1) ? pcm : new float[frameSamples];
line.start();
// Held locally so a concurrent stop() closing the line cannot null it mid-loop.
final AudioCapture capture = line;
capture.start();
while (running.get()) {
int read = 0;
try {
while (read < buf.length) {
int n = line.read(buf, read, buf.length - read);
if (n <= 0) break;
read += n;
}
} catch (Exception e) {
break;
}
if (read < buf.length) continue;
// A short read only happens once the line is closing, or on interruption.
if (capture.read(buf, 0, buf.length) < buf.length) break;
// 16-bit LE -> float, with input gain
for (int i = 0; i < frameSamples; i++) {
for (int i = 0; i < pcm.length; i++) {
int lo = buf[2 * i] & 0xFF;
int hi = buf[2 * i + 1];
short s = (short) ((hi << 8) | lo);
@@ -255,36 +279,50 @@ public final class JavaSoundVoiceInput implements VoiceInput {
else if (f < -1f) f = -1f;
pcm[i] = f;
}
// Denoise / typing attenuation feed the level meter, VAD and encoder alike.
enhancer.process(pcm, frameSamples);
double sumSq = 0;
for (int i = 0; i < frameSamples; i++) {
sumSq += (double) pcm[i] * pcm[i];
}
double rms = Math.sqrt(sumSq / frameSamples);
double db = (rms <= 1e-9) ? -100.0 : 20.0 * Math.log10(rms);
Consumer<Double> ll = levelListener;
if (ll != null) ll.accept(db);
boolean open = decideGate(db, pcm);
setTransmitting(open);
if (open && !muted.get()) {
try {
byte[] packet;
synchronized (encoderLock) {
packet = encoder != null ? encoder.encode(pcm) : null;
}
if (packet != null && packet.length > 0) {
queue.offer(packet);
// Guard against unbounded growth if the network stalls.
while (queue.size() > 10) queue.poll();
}
} catch (Exception ignored) {
if (channels > 1) {
for (int i = 0; i < frameSamples; i++) {
float sum = 0;
for (int c = 0; c < channels; c++) sum += pcm[i * channels + c];
mono[i] = sum / channels;
}
}
byte[] packet = null;
synchronized (encoderLock) {
OpusParameters p = params;
if (p != appliedParams && encoder != null) applyParams(p);
boolean stereo = encoderChannels == 2;
// Denoise / typing attenuation are voice-chain stages: they feed the
// level meter, VAD and encoder alike on the mono path. A stereo (music)
// stream bypasses them and is transmitted as captured.
if (!stereo) enhancer.process(mono, frameSamples);
double sumSq = 0;
for (int i = 0; i < frameSamples; i++) {
sumSq += (double) mono[i] * mono[i];
}
double rms = Math.sqrt(sumSq / frameSamples);
double db = (rms <= 1e-9) ? -100.0 : 20.0 * Math.log10(rms);
Consumer<Double> ll = levelListener;
if (ll != null) ll.accept(db);
boolean open = decideGate(db, mono);
setTransmitting(open);
if (open && !muted.get() && encoder != null) {
try {
packet = encoder.encode(stereo ? pcm : mono);
} catch (Exception ignored) {
}
}
}
if (packet != null && packet.length > 0) {
queue.offer(packet);
// Guard against unbounded growth if the network stalls.
while (queue.size() > 10) queue.poll();
}
}
}

109
ts3-client/desktop/src/main/java/com/ts3client/audio/desktop/JavaSoundVoiceOutput.java → ts3-client/desktop/src/main/java/com/ts3client/audio/desktop/DesktopVoiceOutput.java
View File

@@ -1,10 +1,10 @@
package com.ts3client.audio.desktop;
import com.github.manevolent.ts3j.enums.CodecType;
import com.github.manevolent.ts3j.protocol.packet.PacketBody0Voice;
import com.github.manevolent.ts3j.protocol.packet.PacketBody1VoiceWhisper;
import com.ts3client.audio.VoiceOutput;
import javax.sound.sampled.SourceDataLine;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
@@ -13,26 +13,35 @@ import java.util.concurrent.Executors;
import java.util.function.BiConsumer;
/**
* Java Sound {@link VoiceOutput}: decodes and plays incoming voice per speaker.
* Desktop {@link VoiceOutput}: decodes and plays incoming voice per speaker.
* Each client gets its own Opus decoder, playback line and worker thread, so
* simultaneous speakers are mixed by the OS and one slow decode never blocks
* another (or the network thread).
* simultaneous speakers are mixed by the audio server and one slow decode never
* blocks another (or the network thread). Lines come from {@link AudioDevices},
* so on a PipeWire desktop every speaker is a separate stream in the mixer.
*/
public final class JavaSoundVoiceOutput implements VoiceOutput {
public final class DesktopVoiceOutput implements VoiceOutput {
/** Longest Opus frame (120 ms @ 48 kHz) a packet may decode to, per channel. */
private static final int MAX_FRAME = 5760;
/** One speaker's decode + playback pipeline. */
private final class ClientStream {
final int clientId;
final OpusDecoder decoder;
final SourceDataLine line;
final AudioPlayback line;
final int lineChannels;
final ExecutorService worker;
/** Decoder scratch and byte buffer, touched only by {@link #worker}. */
final float[] pcm = new float[MAX_FRAME * AudioDevices.MAX_CHANNELS];
byte[] out = new byte[0];
OpusDecoder decoder;
int decoderChannels;
volatile boolean talking;
volatile long lastPacketNanos;
ClientStream(int clientId) throws Exception {
this.clientId = clientId;
this.decoder = new OpusDecoder(AudioDevices.SAMPLE_RATE, AudioDevices.FRAME_SIZE, 1);
this.line = AudioDevices.openPlayback(outputDevice);
this.line = AudioDevices.openPlayback(outputDevice, AudioDevices.MAX_CHANNELS);
this.lineChannels = line.channels();
this.line.start();
this.worker = Executors.newSingleThreadExecutor(r -> {
Thread t = new Thread(r, "ts3j-play-" + clientId);
@@ -41,14 +50,24 @@ public final class JavaSoundVoiceOutput implements VoiceOutput {
});
}
/**
* Returns a decoder matching the stream's channel count, rebuilding it when a
* speaker switches between the mono {@code OPUS_VOICE} and stereo
* {@code OPUS_MUSIC} codecs.
*/
OpusDecoder decoderFor(int channels) {
if (decoder == null || decoderChannels != channels) {
if (decoder != null) decoder.close();
decoder = new OpusDecoder(AudioDevices.SAMPLE_RATE, MAX_FRAME, channels);
decoderChannels = channels;
}
return decoder;
}
void close() {
worker.shutdownNow();
try {
line.stop();
line.close();
} catch (Exception ignored) {
}
decoder.close();
line.close();
if (decoder != null) decoder.close();
}
}
@@ -62,7 +81,7 @@ public final class JavaSoundVoiceOutput implements VoiceOutput {
/** Notified (clientId, talking) on the EDT-agnostic worker thread when a speaker starts/stops. */
private volatile BiConsumer<Integer, Boolean> talkListener;
public JavaSoundVoiceOutput(String outputDevice) {
public DesktopVoiceOutput(String outputDevice) {
this.outputDevice = outputDevice;
}
@@ -103,15 +122,20 @@ public final class JavaSoundVoiceOutput implements VoiceOutput {
/** Entry point wired into {@code client.setVoiceHandler(...)}. */
public void handleVoice(PacketBody0Voice voice) {
route(voice.getClientId(), voice.getCodecData());
route(voice.getClientId(), voice.getCodecType(), voice.getCodecData());
}
/** Entry point wired into {@code client.setWhisperHandler(...)}. */
public void handleWhisper(PacketBody1VoiceWhisper whisper) {
route(whisper.getClientId(), whisper.getCodecData());
route(whisper.getClientId(), whisper.getCodecType(), whisper.getCodecData());
}
private void route(int clientId, byte[] data) {
/** TeamSpeak streams {@code OPUS_MUSIC} in stereo and everything else in mono. */
private static int channelsFor(CodecType codec) {
return codec == CodecType.OPUS_MUSIC ? 2 : 1;
}
private void route(int clientId, CodecType codec, byte[] data) {
if (deafened) return;
if (mutedClients.contains(clientId)) return;
@@ -139,32 +163,51 @@ public final class JavaSoundVoiceOutput implements VoiceOutput {
target.line.drain();
} catch (Exception ignored) {
}
target.decoder.reset();
if (target.decoder != null) target.decoder.reset();
markTalking(target, false);
});
return;
}
markTalking(target, true);
target.worker.submit(() -> decodeAndPlay(target, data));
final int channels = channelsFor(codec);
target.worker.submit(() -> decodeAndPlay(target, channels, data));
}
private void decodeAndPlay(ClientStream stream, byte[] data) {
private void decodeAndPlay(ClientStream stream, int channels, byte[] data) {
try {
float[] pcm = new float[AudioDevices.FRAME_SIZE];
int samples = stream.decoder.decode(data, pcm);
float[] pcm = stream.pcm;
int frames = stream.decoderFor(channels).decode(data, pcm);
double vol = masterVolume;
byte[] out = new byte[samples * 2];
for (int i = 0; i < samples; i++) {
double v = pcm[i] * vol;
if (v > 1.0) v = 1.0;
else if (v < -1.0) v = -1.0;
short s = (short) Math.round(v * 32767.0);
out[2 * i] = (byte) (s & 0xFF);
out[2 * i + 1] = (byte) ((s >> 8) & 0xFF);
// Match the decoded stream to the line: duplicate mono across a stereo
// line, fold a stereo (music) stream down onto a mono-only line.
int lineChannels = stream.lineChannels;
int bytes = frames * lineChannels * 2;
if (stream.out.length < bytes) stream.out = new byte[bytes];
byte[] out = stream.out;
for (int i = 0, k = 0; i < frames; i++) {
for (int c = 0; c < lineChannels; c++, k += 2) {
double v;
if (channels == lineChannels) {
v = pcm[i * channels + c];
} else if (channels == 1) {
v = pcm[i];
} else {
double sum = 0;
for (int s = 0; s < channels; s++) sum += pcm[i * channels + s];
v = sum / channels;
}
v *= vol;
if (v > 1.0) v = 1.0;
else if (v < -1.0) v = -1.0;
short s = (short) Math.round(v * 32767.0);
out[k] = (byte) (s & 0xFF);
out[k + 1] = (byte) ((s >> 8) & 0xFF);
}
}
stream.line.write(out, 0, out.length);
stream.line.write(out, 0, bytes);
} catch (Exception ignored) {
}
}

32
ts3-client/desktop/src/main/java/com/ts3client/audio/desktop/JavaSoundAudioBackend.java
View File

@@ -1,32 +0,0 @@
package com.ts3client.audio.desktop;
import com.ts3client.audio.AudioBackend;
import com.ts3client.audio.VoiceInput;
import com.ts3client.audio.VoiceOutput;
import com.ts3client.config.Settings;
/**
* Desktop audio backend using Java Sound for capture/playback and native Opus
* (via JNA) for the codec.
*/
public final class JavaSoundAudioBackend implements AudioBackend {
@Override
public VoiceInput createInput(Settings settings) {
return new JavaSoundVoiceInput(settings);
}
@Override
public VoiceOutput createOutput(Settings settings) {
return new JavaSoundVoiceOutput(settings.outputDevice);
}
@Override
public String description() {
try {
return "Opus " + Opus.getVersionString();
} catch (Throwable t) {
return "Opus (native library unavailable)";
}
}
}

49
ts3-client/desktop/src/main/java/com/ts3client/audio/desktop/JavaSoundCapture.java Normal file
View File

@@ -0,0 +1,49 @@
package com.ts3client.audio.desktop;
import javax.sound.sampled.TargetDataLine;
/** {@link AudioCapture} backed by a Java Sound {@link TargetDataLine}. */
final class JavaSoundCapture implements AudioCapture {
private final TargetDataLine line;
JavaSoundCapture(TargetDataLine line) {
this.line = line;
}
@Override
public int channels() {
return line.getFormat().getChannels();
}
@Override
public void start() {
line.start();
}
/** {@code TargetDataLine.read} may return early, so it is repeated until satisfied. */
@Override
public int read(byte[] buffer, int offset, int length) {
int read = 0;
while (read < length) {
int n;
try {
n = line.read(buffer, offset + read, length - read);
} catch (Exception e) {
break;
}
if (n <= 0) break;
read += n;
}
return read;
}
@Override
public void close() {
try {
line.stop();
line.close();
} catch (Exception ignored) {
}
}
}

49
ts3-client/desktop/src/main/java/com/ts3client/audio/desktop/JavaSoundPlayback.java Normal file
View File

@@ -0,0 +1,49 @@
package com.ts3client.audio.desktop;
import javax.sound.sampled.SourceDataLine;
/** {@link AudioPlayback} backed by a Java Sound {@link SourceDataLine}. */
final class JavaSoundPlayback implements AudioPlayback {
private final SourceDataLine line;
JavaSoundPlayback(SourceDataLine line) {
this.line = line;
}
@Override
public int channels() {
return line.getFormat().getChannels();
}
@Override
public void start() {
line.start();
}
@Override
public void write(byte[] buffer, int offset, int length) {
try {
line.write(buffer, offset, length);
} catch (Exception ignored) {
// The line went away (device unplugged); the caller sees silence.
}
}
@Override
public void drain() {
try {
line.drain();
} catch (Exception ignored) {
}
}
@Override
public void close() {
try {
line.stop();
line.close();
} catch (Exception ignored) {
}
}
}

6
ts3-client/desktop/src/main/java/com/ts3client/audio/desktop/OpusDecoder.java
View File

@@ -7,9 +7,9 @@ import java.lang.foreign.ValueLayout;
/**
* Thin wrapper around a native Opus decoder.
*
* <p>Created as mono at 48&nbsp;kHz; Opus transparently down-mixes stereo streams
* (e.g. music-bot audio) to the requested channel count, so a single mono decoder
* copes with both {@code OPUS_VOICE} and {@code OPUS_MUSIC} payloads.
* <p>Fixed at 48&nbsp;kHz. The channel count should match the stream being decoded
* (mono for {@code OPUS_VOICE}, stereo for {@code OPUS_MUSIC}); Opus will up-/down-mix
* a mismatched stream, which costs the stereo image.
*/
public final class OpusDecoder implements AutoCloseable {

141
ts3-client/desktop/src/main/java/com/ts3client/audio/desktop/pipewire/PcmRing.java Normal file
View File

@@ -0,0 +1,141 @@
package com.ts3client.audio.desktop.pipewire;
import java.lang.foreign.MemorySegment;
/**
* Byte ring buffer between the application threads and PipeWire's loop thread.
*
* <p>The stream callbacks must never block — a late buffer is a glitch for the whole
* graph — so the loop side ({@link #put}/{@link #take}) always returns immediately and
* absorbs the mismatch: a capture overrun drops the oldest audio, a playback underrun is
* reported as a short read and padded with silence by the caller. The application side
* ({@link #write}/{@link #read}) blocks instead, which is what {@code SourceDataLine} and
* {@code TargetDataLine} do and what the voice pipeline expects.
*/
final class PcmRing {
private final byte[] data;
private final MemorySegment view;
private int start; // index of the oldest byte
private int count; // bytes currently held
private boolean closed;
PcmRing(int capacity) {
this.data = new byte[capacity];
this.view = MemorySegment.ofArray(data);
}
int capacity() {
return data.length;
}
synchronized int available() {
return count;
}
synchronized void clear() {
start = 0;
count = 0;
notifyAll();
}
/** Unblocks every waiter; subsequent reads report end-of-stream and writes are dropped. */
synchronized void close() {
closed = true;
notifyAll();
}
synchronized boolean isClosed() {
return closed;
}
// ---- loop-thread side (never blocks) ----
/** Appends up to {@code length} bytes, dropping the oldest audio if that overruns. */
synchronized void put(MemorySegment source, long offset, int length) {
if (closed) return;
int overflow = count + length - data.length;
if (overflow > 0) skip(overflow);
int copied = Math.min(length, data.length);
long from = offset + (length - copied);
int end = (start + count) % data.length;
int firstChunk = Math.min(copied, data.length - end);
MemorySegment.copy(source, from, view, end, firstChunk);
MemorySegment.copy(source, from + firstChunk, view, 0, copied - firstChunk);
count += copied;
notifyAll();
}
/** Removes up to {@code length} bytes into {@code destination}; returns how many. */
synchronized int take(MemorySegment destination, long offset, int length) {
int copied = Math.min(length, count);
int firstChunk = Math.min(copied, data.length - start);
MemorySegment.copy(view, start, destination, offset, firstChunk);
MemorySegment.copy(view, 0, destination, offset + firstChunk, copied - firstChunk);
skip(copied);
notifyAll();
return copied;
}
// ---- application side (blocks) ----
/**
* Blocks until {@code length} bytes have been read, the ring is closed, or the calling
* thread is interrupted. Returns the number of bytes actually read.
*/
synchronized int read(byte[] destination, int offset, int length) {
MemorySegment target = MemorySegment.ofArray(destination);
int read = 0;
while (read < length) {
if (count == 0) {
if (closed || !await()) break;
continue;
}
read += take(target, offset + read, length - read);
}
return read;
}
/**
* Blocks until {@code length} bytes have been queued, the ring is closed, or the
* calling thread is interrupted. Returns the number of bytes actually written.
*/
synchronized int write(byte[] source, int offset, int length) {
MemorySegment target = MemorySegment.ofArray(source);
int written = 0;
while (written < length && !closed) {
int free = data.length - count;
if (free == 0) {
if (!await()) break;
continue;
}
int chunk = Math.min(free, length - written);
put(target, offset + written, chunk);
written += chunk;
}
return written;
}
/** Blocks until the ring has drained (or was closed / the wait was interrupted). */
synchronized void drain() {
while (count > 0 && !closed) {
if (!await()) return;
}
}
/** Waits for a state change; {@code false} when interrupted. */
private boolean await() {
try {
wait(100);
return true;
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return false;
}
}
private void skip(int bytes) {
start = (start + bytes) % data.length;
count -= bytes;
}
}

78
ts3-client/desktop/src/main/java/com/ts3client/audio/desktop/pipewire/PipeWire.java Normal file
View File

@@ -0,0 +1,78 @@
package com.ts3client.audio.desktop.pipewire;
import com.ts3client.audio.desktop.AudioCapture;
import com.ts3client.audio.desktop.AudioPlayback;
import java.util.List;
/**
* PipeWire audio backend.
*
* <p>On a PipeWire desktop this replaces Java Sound entirely: devices are enumerated
* from the graph — with their real names, including Bluetooth and virtual devices,
* which Java Sound's ALSA-only provider never sees — and capture and playback run as
* ordinary {@code pw_stream}s that appear individually in volume mixers.
*
* <p>Everything here is a no-op on a machine without PipeWire: {@link #isAvailable()}
* reports {@code false} and {@link com.ts3client.audio.desktop.AudioDevices} falls back
* to Java Sound.
*/
public final class PipeWire {
/** A selectable PipeWire device. {@code name} is the node name used to target it. */
public record Node(String name, String description, boolean sink) {
}
private PipeWire() {
}
/** {@code true} when libpipewire loaded and a session is reachable. */
public static boolean isAvailable() {
return PipeWireSession.get() != null;
}
/** The version of the loaded libpipewire, or {@code null} when it is not available. */
public static String version() {
if (!isAvailable()) return null;
try {
return PipeWireLibrary.libraryVersion();
} catch (Throwable t) {
return null;
}
}
/** The audio sinks and sources currently in the graph; empty without PipeWire. */
public static List<Node> nodes() {
PipeWireSession session = PipeWireSession.get();
if (session == null) return List.of();
return session.nodes().stream()
.map(node -> new Node(node.name(), node.description(), node.sink()))
.toList();
}
/**
* Opens a capture stream on {@code targetNode} (empty or {@code null} for the default
* source).
*
* @throws IllegalStateException if PipeWire is unavailable or the stream is refused
*/
public static AudioCapture openCapture(String name, String targetNode, int rate, int channels) {
return new PipeWireCapture(session(), name, targetNode, rate, channels);
}
/**
* Opens a playback stream on {@code targetNode} (empty or {@code null} for the default
* sink).
*
* @throws IllegalStateException if PipeWire is unavailable or the stream is refused
*/
public static AudioPlayback openPlayback(String name, String targetNode, int rate, int channels) {
return new PipeWirePlayback(session(), name, targetNode, rate, channels);
}
private static PipeWireSession session() {
PipeWireSession session = PipeWireSession.get();
if (session == null) throw new IllegalStateException("PipeWire is not available");
return session;
}
}

28
ts3-client/desktop/src/main/java/com/ts3client/audio/desktop/pipewire/PipeWireCapture.java Normal file
View File

@@ -0,0 +1,28 @@
package com.ts3client.audio.desktop.pipewire;
import com.ts3client.audio.desktop.AudioCapture;
import java.lang.foreign.MemorySegment;
import static com.ts3client.audio.desktop.pipewire.PipeWireLibrary.INT;
/** A PipeWire capture stream ({@code PW_DIRECTION_INPUT}). */
final class PipeWireCapture extends PipeWireStream implements AudioCapture {
PipeWireCapture(PipeWireSession session, String name, String targetNode, int rate, int channels) {
super(session, name, targetNode, rate, channels, true);
}
/** Copies the valid part of the buffer — which need not start at zero — into the ring. */
@Override
void process(MemorySegment header, MemorySegment audio, MemorySegment chunk, int maxSize) {
int offset = Math.min(chunk.get(INT, PipeWireLibrary.SPA_CHUNK_OFFSET), maxSize);
int size = Math.min(chunk.get(INT, PipeWireLibrary.SPA_CHUNK_SIZE), maxSize - offset);
if (size > 0) ring.put(audio, offset, size);
}
@Override
public int read(byte[] buffer, int offset, int length) {
return ring.read(buffer, offset, length);
}
}

409
ts3-client/desktop/src/main/java/com/ts3client/audio/desktop/pipewire/PipeWireLibrary.java Normal file
View File

@@ -0,0 +1,409 @@
package com.ts3client.audio.desktop.pipewire;
import java.lang.foreign.AddressLayout;
import java.lang.foreign.Arena;
import java.lang.foreign.FunctionDescriptor;
import java.lang.foreign.Linker;
import java.lang.foreign.MemorySegment;
import java.lang.foreign.SymbolLookup;
import java.lang.foreign.ValueLayout;
import java.lang.invoke.MethodHandle;
/**
* Raw binding to {@code libpipewire-0.3} through the Foreign Function &amp; Memory API.
*
* <p>This class is the only place that knows about the C ABI: the exported entry points,
* the SPA/PipeWire constants, and the offsets of the few structures that are read or
* written directly. Everything above it works with Java types.
*
* <p>Loading is lazy and failure is expected: on a machine without PipeWire (or without
* a session running) {@link #isAvailable()} returns {@code false} and the caller falls
* back to Java Sound. The class initialiser is what fails, so both the first
* {@link ExceptionInInitializerError} and the later {@link NoClassDefFoundError} are
* caught.
*
* <h2>Interface methods</h2>
* The registry API ({@code pw_core_get_registry}, {@code pw_registry_add_listener},
* {@code pw_core_sync}, ...) is declared {@code static inline} in the headers. PipeWire
* only began exporting compiled copies of those helpers in 1.1; to work on older
* versions too they are instead called the way the inline code does, through the
* object's {@code spa_interface} method table — see {@link #interfaceMethod}.
*
* <p>Constants and structure offsets below were taken from the pipewire 1.4 headers and
* are part of the frozen 0.3 ABI.
*/
final class PipeWireLibrary {
private PipeWireLibrary() {
}
static final AddressLayout PTR = ValueLayout.ADDRESS;
static final ValueLayout.OfInt INT = ValueLayout.JAVA_INT;
static final ValueLayout.OfLong LONG = ValueLayout.JAVA_LONG;
static final ValueLayout.OfBoolean BOOL = ValueLayout.JAVA_BOOLEAN;
// ---- enum pw_direction ----
static final int PW_DIRECTION_INPUT = 0;
static final int PW_DIRECTION_OUTPUT = 1;
// ---- well-known object ids ----
static final int PW_ID_CORE = 0;
static final int PW_ID_ANY = 0xFFFFFFFF;
// ---- interface versions we ask for / implement ----
static final int PW_VERSION_REGISTRY = 3;
static final int PW_VERSION_CORE_EVENTS = 1;
static final int PW_VERSION_REGISTRY_EVENTS = 0;
static final int PW_VERSION_STREAM_EVENTS = 2;
// ---- enum pw_stream_flags ----
static final int PW_STREAM_FLAG_AUTOCONNECT = 1 << 0;
static final int PW_STREAM_FLAG_INACTIVE = 1 << 1;
static final int PW_STREAM_FLAG_MAP_BUFFERS = 1 << 2;
// ---- enum pw_stream_state ----
static final int PW_STREAM_STATE_ERROR = -1;
static final int PW_STREAM_STATE_UNCONNECTED = 0;
static final int PW_STREAM_STATE_CONNECTING = 1;
static final int PW_STREAM_STATE_PAUSED = 2;
static final int PW_STREAM_STATE_STREAMING = 3;
// ---- property keys (pipewire/keys.h) ----
static final String KEY_APP_NAME = "application.name";
static final String KEY_MEDIA_TYPE = "media.type";
static final String KEY_MEDIA_CATEGORY = "media.category";
static final String KEY_MEDIA_ROLE = "media.role";
static final String KEY_MEDIA_CLASS = "media.class";
static final String KEY_NODE_NAME = "node.name";
static final String KEY_NODE_DESCRIPTION = "node.description";
static final String KEY_NODE_NICK = "node.nick";
static final String KEY_NODE_LATENCY = "node.latency";
static final String KEY_NODE_RATE = "node.rate";
static final String KEY_TARGET_OBJECT = "target.object";
static final String TYPE_INTERFACE_NODE = "PipeWire:Interface:Node";
// ---- struct offsets (x86-64/aarch64 LP64; all fields are naturally aligned) ----
/** {@code struct spa_interface { const char *type; uint32_t version; struct spa_callbacks cb; }} */
private static final long SPA_INTERFACE_SIZE = 32;
private static final long SPA_INTERFACE_CB_FUNCS = 16;
private static final long SPA_INTERFACE_CB_DATA = 24;
/** {@code struct spa_dict { uint32_t flags, n_items; const struct spa_dict_item *items; }} */
static final long SPA_DICT_SIZE = 16;
static final long SPA_DICT_N_ITEMS = 4;
static final long SPA_DICT_ITEMS = 8;
/** {@code struct spa_dict_item { const char *key, *value; }} */
static final long SPA_DICT_ITEM_SIZE = 16;
static final long SPA_DICT_ITEM_VALUE = 8;
/** {@code struct pw_buffer { struct spa_buffer *buffer; void *user_data; uint64_t size, requested, time; }} */
static final long PW_BUFFER_BUFFER = 0;
static final long PW_BUFFER_SIZE = 16;
static final long PW_BUFFER_REQUESTED = 24;
static final long PW_BUFFER_SIZEOF = 40;
/** {@code struct spa_buffer { uint32_t n_metas, n_datas; struct spa_meta *metas; struct spa_data *datas; }} */
static final long SPA_BUFFER_N_DATAS = 4;
static final long SPA_BUFFER_DATAS = 16;
static final long SPA_BUFFER_SIZEOF = 24;
/**
* {@code struct spa_data { uint32_t type, flags; int64_t fd; uint32_t mapoffset, maxsize;
* void *data; struct spa_chunk *chunk; }}
*/
static final long SPA_DATA_MAXSIZE = 20;
static final long SPA_DATA_DATA = 24;
static final long SPA_DATA_CHUNK = 32;
static final long SPA_DATA_SIZEOF = 40;
/** {@code struct spa_chunk { uint32_t offset, size; int32_t stride, flags; }} */
static final long SPA_CHUNK_OFFSET = 0;
static final long SPA_CHUNK_SIZE = 4;
static final long SPA_CHUNK_STRIDE = 8;
static final long SPA_CHUNK_SIZEOF = 16;
/** Slot (pointer index) of a method inside {@code struct pw_core_methods}. */
static final int CORE_METHOD_ADD_LISTENER = 1;
static final int CORE_METHOD_SYNC = 3;
static final int CORE_METHOD_GET_REGISTRY = 6;
/** Slot of a method inside {@code struct pw_registry_methods}. */
static final int REGISTRY_METHOD_ADD_LISTENER = 1;
/** Field offsets in {@code struct pw_core_events} (version at 0, then one pointer each). */
static final long CORE_EVENTS_SIZEOF = 80;
static final long CORE_EVENTS_DONE = 16;
/** Field offsets in {@code struct pw_registry_events}. */
static final long REGISTRY_EVENTS_SIZEOF = 24;
static final long REGISTRY_EVENTS_GLOBAL = 8;
static final long REGISTRY_EVENTS_GLOBAL_REMOVE = 16;
/** Field offsets in {@code struct pw_stream_events}. */
static final long STREAM_EVENTS_SIZEOF = 96;
static final long STREAM_EVENTS_STATE_CHANGED = 16;
static final long STREAM_EVENTS_PROCESS = 64;
/** {@code struct spa_hook}; over-allocated so a future growth cannot corrupt anything. */
static final long SPA_HOOK_SIZEOF = 128;
static final Linker LINKER = Linker.nativeLinker();
/** SONAMEs to try, newest packaging first. */
private static final String[] LIBRARY_NAMES = {"libpipewire-0.3.so.0", "libpipewire-0.3.so"};
private static final class Handles {
static final SymbolLookup LOOKUP = load();
static final MethodHandle PW_INIT =
downcall("pw_init", FunctionDescriptor.ofVoid(PTR, PTR));
static final MethodHandle PW_GET_LIBRARY_VERSION =
downcall("pw_get_library_version", FunctionDescriptor.of(PTR));
static final MethodHandle PW_THREAD_LOOP_NEW =
downcall("pw_thread_loop_new", FunctionDescriptor.of(PTR, PTR, PTR));
static final MethodHandle PW_THREAD_LOOP_DESTROY =
downcall("pw_thread_loop_destroy", FunctionDescriptor.ofVoid(PTR));
static final MethodHandle PW_THREAD_LOOP_GET_LOOP =
downcall("pw_thread_loop_get_loop", FunctionDescriptor.of(PTR, PTR));
static final MethodHandle PW_THREAD_LOOP_START =
downcall("pw_thread_loop_start", FunctionDescriptor.of(INT, PTR));
static final MethodHandle PW_THREAD_LOOP_STOP =
downcall("pw_thread_loop_stop", FunctionDescriptor.ofVoid(PTR));
static final MethodHandle PW_THREAD_LOOP_LOCK =
downcall("pw_thread_loop_lock", FunctionDescriptor.ofVoid(PTR));
static final MethodHandle PW_THREAD_LOOP_UNLOCK =
downcall("pw_thread_loop_unlock", FunctionDescriptor.ofVoid(PTR));
static final MethodHandle PW_THREAD_LOOP_SIGNAL =
downcall("pw_thread_loop_signal", FunctionDescriptor.ofVoid(PTR, BOOL));
static final MethodHandle PW_THREAD_LOOP_TIMED_WAIT =
downcall("pw_thread_loop_timed_wait", FunctionDescriptor.of(INT, PTR, INT));
static final MethodHandle PW_CONTEXT_NEW =
downcall("pw_context_new", FunctionDescriptor.of(PTR, PTR, PTR, LONG));
static final MethodHandle PW_CONTEXT_CONNECT =
downcall("pw_context_connect", FunctionDescriptor.of(PTR, PTR, PTR, LONG));
static final MethodHandle PW_CONTEXT_DESTROY =
downcall("pw_context_destroy", FunctionDescriptor.ofVoid(PTR));
static final MethodHandle PW_CORE_DISCONNECT =
downcall("pw_core_disconnect", FunctionDescriptor.of(INT, PTR));
static final MethodHandle PW_PROXY_DESTROY =
downcall("pw_proxy_destroy", FunctionDescriptor.ofVoid(PTR));
static final MethodHandle PW_PROPERTIES_NEW_DICT =
downcall("pw_properties_new_dict", FunctionDescriptor.of(PTR, PTR));
static final MethodHandle PW_STREAM_NEW_SIMPLE =
downcall("pw_stream_new_simple", FunctionDescriptor.of(PTR, PTR, PTR, PTR, PTR, PTR));
static final MethodHandle PW_STREAM_CONNECT =
downcall("pw_stream_connect", FunctionDescriptor.of(INT, PTR, INT, INT, INT, PTR, INT));
static final MethodHandle PW_STREAM_DISCONNECT =
downcall("pw_stream_disconnect", FunctionDescriptor.of(INT, PTR));
static final MethodHandle PW_STREAM_DESTROY =
downcall("pw_stream_destroy", FunctionDescriptor.ofVoid(PTR));
static final MethodHandle PW_STREAM_SET_ACTIVE =
downcall("pw_stream_set_active", FunctionDescriptor.of(INT, PTR, BOOL));
static final MethodHandle PW_STREAM_FLUSH =
downcall("pw_stream_flush", FunctionDescriptor.of(INT, PTR, BOOL));
static final MethodHandle PW_STREAM_DEQUEUE_BUFFER =
downcall("pw_stream_dequeue_buffer", FunctionDescriptor.of(PTR, PTR));
static final MethodHandle PW_STREAM_QUEUE_BUFFER =
downcall("pw_stream_queue_buffer", FunctionDescriptor.of(INT, PTR, PTR));
private static SymbolLookup load() {
IllegalArgumentException last = null;
for (String name : LIBRARY_NAMES) {
try {
return SymbolLookup.libraryLookup(name, Arena.global());
} catch (IllegalArgumentException e) {
last = e;
}
}
throw (last != null) ? last : new IllegalArgumentException("libpipewire not found");
}
private static MethodHandle downcall(String symbol, FunctionDescriptor descriptor) {
return LINKER.downcallHandle(
LOOKUP.find(symbol).orElseThrow(() ->
new UnsatisfiedLinkError("libpipewire: unresolved symbol " + symbol)),
descriptor);
}
}
/** {@code true} when libpipewire could be loaded and every symbol we need resolved. */
static boolean isAvailable() {
try {
return Handles.LOOKUP != null;
} catch (Throwable t) {
return false;
}
}
// ---- library ----
static void init() {
invokeVoid(Handles.PW_INIT, MemorySegment.NULL, MemorySegment.NULL);
}
static String libraryVersion() {
MemorySegment version = (MemorySegment) invoke(Handles.PW_GET_LIBRARY_VERSION);
return cString(version);
}
// ---- thread loop ----
static MemorySegment threadLoopNew(MemorySegment name) {
return (MemorySegment) invoke(Handles.PW_THREAD_LOOP_NEW, name, MemorySegment.NULL);
}
static void threadLoopDestroy(MemorySegment loop) {
invokeVoid(Handles.PW_THREAD_LOOP_DESTROY, loop);
}
static MemorySegment threadLoopGetLoop(MemorySegment loop) {
return (MemorySegment) invoke(Handles.PW_THREAD_LOOP_GET_LOOP, loop);
}
static int threadLoopStart(MemorySegment loop) {
return (int) invoke(Handles.PW_THREAD_LOOP_START, loop);
}
static void threadLoopStop(MemorySegment loop) {
invokeVoid(Handles.PW_THREAD_LOOP_STOP, loop);
}
static void threadLoopLock(MemorySegment loop) {
invokeVoid(Handles.PW_THREAD_LOOP_LOCK, loop);
}
static void threadLoopUnlock(MemorySegment loop) {
invokeVoid(Handles.PW_THREAD_LOOP_UNLOCK, loop);
}
static void threadLoopSignal(MemorySegment loop, boolean waitForAccept) {
invokeVoid(Handles.PW_THREAD_LOOP_SIGNAL, loop, waitForAccept);
}
/** Waits for a {@code signal}, returning {@code -ETIMEDOUT} when {@code seconds} elapse. */
static int threadLoopTimedWait(MemorySegment loop, int seconds) {
return (int) invoke(Handles.PW_THREAD_LOOP_TIMED_WAIT, loop, seconds);
}
// ---- context / core ----
static MemorySegment contextNew(MemorySegment loop, MemorySegment properties) {
return (MemorySegment) invoke(Handles.PW_CONTEXT_NEW, loop, properties, 0L);
}
static MemorySegment contextConnect(MemorySegment context, MemorySegment properties) {
return (MemorySegment) invoke(Handles.PW_CONTEXT_CONNECT, context, properties, 0L);
}
static void contextDestroy(MemorySegment context) {
invokeVoid(Handles.PW_CONTEXT_DESTROY, context);
}
static int coreDisconnect(MemorySegment core) {
return (int) invoke(Handles.PW_CORE_DISCONNECT, core);
}
static void proxyDestroy(MemorySegment proxy) {
invokeVoid(Handles.PW_PROXY_DESTROY, proxy);
}
static MemorySegment propertiesNewDict(MemorySegment dict) {
return (MemorySegment) invoke(Handles.PW_PROPERTIES_NEW_DICT, dict);
}
// ---- stream ----
static MemorySegment streamNewSimple(MemorySegment loop, MemorySegment name,
MemorySegment properties, MemorySegment events,
MemorySegment data) {
return (MemorySegment) invoke(Handles.PW_STREAM_NEW_SIMPLE, loop, name, properties, events, data);
}
static int streamConnect(MemorySegment stream, int direction, int targetId, int flags,
MemorySegment params, int paramCount) {
return (int) invoke(Handles.PW_STREAM_CONNECT, stream, direction, targetId, flags, params, paramCount);
}
static int streamDisconnect(MemorySegment stream) {
return (int) invoke(Handles.PW_STREAM_DISCONNECT, stream);
}
static void streamDestroy(MemorySegment stream) {
invokeVoid(Handles.PW_STREAM_DESTROY, stream);
}
static int streamSetActive(MemorySegment stream, boolean active) {
return (int) invoke(Handles.PW_STREAM_SET_ACTIVE, stream, active);
}
static int streamFlush(MemorySegment stream, boolean drain) {
return (int) invoke(Handles.PW_STREAM_FLUSH, stream, drain);
}
static MemorySegment streamDequeueBuffer(MemorySegment stream) {
return (MemorySegment) invoke(Handles.PW_STREAM_DEQUEUE_BUFFER, stream);
}
static int streamQueueBuffer(MemorySegment stream, MemorySegment buffer) {
return (int) invoke(Handles.PW_STREAM_QUEUE_BUFFER, stream, buffer);
}
// ---- spa_interface method dispatch ----
/** The {@code cb.data} pointer an interface's methods expect as their first argument. */
static MemorySegment interfaceSelf(MemorySegment iface) {
return iface.reinterpret(SPA_INTERFACE_SIZE).get(PTR, SPA_INTERFACE_CB_DATA);
}
/**
* Binds slot {@code slot} of an interface's method table as a downcall. The table is
* a {@code uint32_t version} followed by one function pointer per method, so the slot
* is the pointer index (1 for the first method).
*/
static MethodHandle interfaceMethod(MemorySegment iface, int slot, FunctionDescriptor descriptor) {
MemorySegment methods = iface.reinterpret(SPA_INTERFACE_SIZE).get(PTR, SPA_INTERFACE_CB_FUNCS);
MemorySegment function = methods.reinterpret(8L * (slot + 1)).get(PTR, 8L * slot);
if (function.address() == 0) {
throw new UnsatisfiedLinkError("libpipewire: interface method slot " + slot + " is null");
}
return LINKER.downcallHandle(function, descriptor);
}
// ---- helpers ----
/** Reads a NUL-terminated C string from an unbounded pointer; {@code null} for NULL. */
static String cString(MemorySegment pointer) {
if (pointer == null || pointer.address() == 0) return null;
return pointer.reinterpret(Long.MAX_VALUE).getString(0);
}
/** Gives a pointer returned by C a known extent so its fields can be read. */
static MemorySegment at(MemorySegment pointer, long size) {
return pointer.reinterpret(size);
}
static Object invoke(MethodHandle handle, Object... arguments) {
try {
return handle.invokeWithArguments(arguments);
} catch (Throwable t) {
throw wrap(t);
}
}
static void invokeVoid(MethodHandle handle, Object... arguments) {
invoke(handle, arguments);
}
private static RuntimeException wrap(Throwable t) {
if (t instanceof RuntimeException re) return re;
if (t instanceof Error e) throw e;
return new IllegalStateException("libpipewire call failed", t);
}
}

47
ts3-client/desktop/src/main/java/com/ts3client/audio/desktop/pipewire/PipeWirePlayback.java Normal file
View File

@@ -0,0 +1,47 @@
package com.ts3client.audio.desktop.pipewire;
import com.ts3client.audio.desktop.AudioPlayback;
import java.lang.foreign.MemorySegment;
import static com.ts3client.audio.desktop.pipewire.PipeWireLibrary.INT;
import static com.ts3client.audio.desktop.pipewire.PipeWireLibrary.LONG;
/** A PipeWire playback stream ({@code PW_DIRECTION_OUTPUT}). */
final class PipeWirePlayback extends PipeWireStream implements AudioPlayback {
PipeWirePlayback(PipeWireSession session, String name, String targetNode, int rate, int channels) {
super(session, name, targetNode, rate, channels, false);
}
/**
* Fills the buffer with as much queued audio as there is, padding with silence on an
* underrun so the graph always gets a full quantum.
*/
@Override
void process(MemorySegment header, MemorySegment audio, MemorySegment chunk, int maxSize) {
int stride = stride();
long requested = header.get(LONG, PipeWireLibrary.PW_BUFFER_REQUESTED);
int bytes = maxSize;
if (requested > 0) bytes = (int) Math.min(bytes, requested * stride);
bytes -= bytes % stride;
int filled = ring.take(audio, 0, bytes);
if (filled < bytes) audio.asSlice(filled, bytes - filled).fill((byte) 0);
chunk.set(INT, PipeWireLibrary.SPA_CHUNK_OFFSET, 0);
chunk.set(INT, PipeWireLibrary.SPA_CHUNK_STRIDE, stride);
chunk.set(INT, PipeWireLibrary.SPA_CHUNK_SIZE, bytes);
header.set(LONG, PipeWireLibrary.PW_BUFFER_SIZE, bytes / stride);
}
@Override
public void write(byte[] buffer, int offset, int length) {
ring.write(buffer, offset, length);
}
@Override
public void drain() {
ring.drain();
}
}

287
ts3-client/desktop/src/main/java/com/ts3client/audio/desktop/pipewire/PipeWireSession.java Normal file
View File

@@ -0,0 +1,287 @@
package com.ts3client.audio.desktop.pipewire;
import java.lang.foreign.Arena;
import java.lang.foreign.FunctionDescriptor;
import java.lang.foreign.MemorySegment;
import java.lang.invoke.MethodHandle;
import java.lang.invoke.MethodHandles;
import java.lang.invoke.MethodType;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import static com.ts3client.audio.desktop.pipewire.PipeWireLibrary.INT;
import static com.ts3client.audio.desktop.pipewire.PipeWireLibrary.PTR;
/**
* The process-wide connection to the PipeWire daemon.
*
* <p>Owns the thread loop that every stream runs on, and keeps a live view of the
* graph's audio nodes by listening to the registry — so the device list is always
* current without polling. The session is created on first use and then kept: the
* loop is a native thread that does not hold the JVM open, and re-connecting per
* device query would cost a round trip each time.
*
* <p>Every libpipewire call from outside the loop thread must hold the loop lock; the
* {@code lock}/{@code unlock} pairs below are that lock, not a Java monitor.
*/
final class PipeWireSession {
/** An audio sink (playback device) or source (capture device) in the graph. */
record Node(String name, String description, boolean sink) {
}
private static final int ROUNDTRIP_TIMEOUT_SECONDS = 2;
private static volatile PipeWireSession instance;
private static volatile boolean failed;
private final Arena arena = Arena.ofShared();
private final MemorySegment threadLoop;
private final MemorySegment loop;
private final MemorySegment context;
private final MemorySegment core;
private final MemorySegment registry;
private final MethodHandle coreSync;
/** Nodes by global id, in the order the registry announced them. Guarded by itself. */
private final Map<Integer, Node> nodes = new LinkedHashMap<>();
// Written on the caller's thread, read on the loop thread; the native loop mutex
// that orders them is invisible to the Java memory model.
private volatile int pendingSync;
private volatile boolean syncDone;
/**
* Returns the shared session, or {@code null} when PipeWire is not usable here —
* no library, no daemon, or a connection that was refused. The first failure is
* remembered so an unavailable system is not retried on every device query.
*/
static PipeWireSession get() {
PipeWireSession existing = instance;
if (existing != null || failed) return existing;
synchronized (PipeWireSession.class) {
if (instance == null && !failed) {
try {
instance = new PipeWireSession();
} catch (Throwable t) {
failed = true;
}
}
return instance;
}
}
private PipeWireSession() {
if (!PipeWireLibrary.isAvailable()) {
throw new UnsupportedOperationException("libpipewire is not available");
}
PipeWireLibrary.init();
threadLoop = nonNull(PipeWireLibrary.threadLoopNew(arena.allocateFrom("ts3-client")),
"pw_thread_loop_new");
loop = PipeWireLibrary.threadLoopGetLoop(threadLoop);
context = nonNull(PipeWireLibrary.contextNew(loop, MemorySegment.NULL), "pw_context_new");
if (PipeWireLibrary.threadLoopStart(threadLoop) < 0) {
throw new IllegalStateException("pw_thread_loop_start failed");
}
PipeWireLibrary.threadLoopLock(threadLoop);
try {
core = nonNull(PipeWireLibrary.contextConnect(context, MemorySegment.NULL),
"pw_context_connect");
coreSync = PipeWireLibrary.interfaceMethod(core, PipeWireLibrary.CORE_METHOD_SYNC,
FunctionDescriptor.of(INT, PTR, INT, INT));
addCoreListener();
registry = nonNull(getRegistry(), "pw_core_get_registry");
addRegistryListener();
roundtrip();
} catch (Throwable t) {
PipeWireLibrary.threadLoopUnlock(threadLoop);
close();
throw t;
}
PipeWireLibrary.threadLoopUnlock(threadLoop);
}
// ---- registry / core wiring ----
private MemorySegment getRegistry() {
MethodHandle getRegistry = PipeWireLibrary.interfaceMethod(core,
PipeWireLibrary.CORE_METHOD_GET_REGISTRY,
FunctionDescriptor.of(PTR, PTR, INT, PipeWireLibrary.LONG));
return (MemorySegment) PipeWireLibrary.invoke(getRegistry,
PipeWireLibrary.interfaceSelf(core), PipeWireLibrary.PW_VERSION_REGISTRY, 0L);
}
private void addCoreListener() {
// Arena allocations are zero-filled, so every callback we do not set stays NULL.
MemorySegment events = arena.allocate(PipeWireLibrary.CORE_EVENTS_SIZEOF, 8);
events.set(INT, 0, PipeWireLibrary.PW_VERSION_CORE_EVENTS);
events.set(PTR, PipeWireLibrary.CORE_EVENTS_DONE,
upcall("onDone", MethodType.methodType(void.class, MemorySegment.class, int.class, int.class),
FunctionDescriptor.ofVoid(PTR, INT, INT)));
MethodHandle addListener = PipeWireLibrary.interfaceMethod(core,
PipeWireLibrary.CORE_METHOD_ADD_LISTENER,
FunctionDescriptor.of(INT, PTR, PTR, PTR, PTR));
PipeWireLibrary.invoke(addListener, PipeWireLibrary.interfaceSelf(core),
hook(), events, MemorySegment.NULL);
}
private void addRegistryListener() {
MemorySegment events = arena.allocate(PipeWireLibrary.REGISTRY_EVENTS_SIZEOF, 8);
events.set(INT, 0, PipeWireLibrary.PW_VERSION_REGISTRY_EVENTS);
events.set(PTR, PipeWireLibrary.REGISTRY_EVENTS_GLOBAL,
upcall("onGlobal", MethodType.methodType(void.class, MemorySegment.class, int.class,
int.class, MemorySegment.class, int.class, MemorySegment.class),
FunctionDescriptor.ofVoid(PTR, INT, INT, PTR, INT, PTR)));
events.set(PTR, PipeWireLibrary.REGISTRY_EVENTS_GLOBAL_REMOVE,
upcall("onGlobalRemove", MethodType.methodType(void.class, MemorySegment.class, int.class),
FunctionDescriptor.ofVoid(PTR, INT)));
MethodHandle addListener = PipeWireLibrary.interfaceMethod(registry,
PipeWireLibrary.REGISTRY_METHOD_ADD_LISTENER,
FunctionDescriptor.of(INT, PTR, PTR, PTR, PTR));
PipeWireLibrary.invoke(addListener, PipeWireLibrary.interfaceSelf(registry),
hook(), events, MemorySegment.NULL);
}
/** A zeroed {@code struct spa_hook} for a listener; it must outlive the object it hooks. */
private MemorySegment hook() {
return arena.allocate(PipeWireLibrary.SPA_HOOK_SIZEOF, 8);
}
private MemorySegment upcall(String method, MethodType type, FunctionDescriptor descriptor) {
try {
MethodHandle handle = MethodHandles.lookup()
.bind(this, method, type)
.asType(descriptor.toMethodType());
return PipeWireLibrary.LINKER.upcallStub(handle, descriptor, arena);
} catch (ReflectiveOperationException e) {
throw new IllegalStateException("cannot bind PipeWire callback " + method, e);
}
}
// ---- callbacks, all invoked on the loop thread ----
@SuppressWarnings("unused") // bound as an upcall
private void onDone(MemorySegment data, int id, int seq) {
if (id != PipeWireLibrary.PW_ID_CORE || seq != pendingSync) return;
syncDone = true;
PipeWireLibrary.threadLoopSignal(threadLoop, false);
}
@SuppressWarnings("unused") // bound as an upcall
private void onGlobal(MemorySegment data, int id, int permissions, MemorySegment type,
int version, MemorySegment props) {
try {
if (!PipeWireLibrary.TYPE_INTERFACE_NODE.equals(PipeWireLibrary.cString(type))) return;
if (props.address() == 0) return;
Map<String, String> properties = readDict(props);
String mediaClass = properties.get(PipeWireLibrary.KEY_MEDIA_CLASS);
boolean sink = "Audio/Sink".equals(mediaClass);
if (!sink && !"Audio/Source".equals(mediaClass)) return;
String name = properties.get(PipeWireLibrary.KEY_NODE_NAME);
if (name == null || name.isBlank()) return;
String description = firstNonBlank(
properties.get(PipeWireLibrary.KEY_NODE_DESCRIPTION),
properties.get(PipeWireLibrary.KEY_NODE_NICK),
name);
synchronized (nodes) {
nodes.put(id, new Node(name, description, sink));
}
} catch (Throwable ignored) {
// An upcall must not throw: a node we cannot read is simply not listed.
}
}
@SuppressWarnings("unused") // bound as an upcall
private void onGlobalRemove(MemorySegment data, int id) {
synchronized (nodes) {
nodes.remove(id);
}
}
/** Copies a {@code struct spa_dict} into a Java map. */
private static Map<String, String> readDict(MemorySegment dict) {
MemorySegment header = PipeWireLibrary.at(dict, PipeWireLibrary.SPA_DICT_SIZE);
int itemCount = header.get(INT, PipeWireLibrary.SPA_DICT_N_ITEMS);
MemorySegment items = header.get(PTR, PipeWireLibrary.SPA_DICT_ITEMS);
Map<String, String> properties = new LinkedHashMap<>();
if (items.address() == 0 || itemCount <= 0) return properties;
MemorySegment array = PipeWireLibrary.at(items, PipeWireLibrary.SPA_DICT_ITEM_SIZE * itemCount);
for (int i = 0; i < itemCount; i++) {
long offset = PipeWireLibrary.SPA_DICT_ITEM_SIZE * i;
String key = PipeWireLibrary.cString(array.get(PTR, offset));
String value = PipeWireLibrary.cString(
array.get(PTR, offset + PipeWireLibrary.SPA_DICT_ITEM_VALUE));
if (key != null) properties.put(key, value);
}
return properties;
}
private static String firstNonBlank(String... candidates) {
for (String candidate : candidates) {
if (candidate != null && !candidate.isBlank()) return candidate;
}
return null;
}
// ---- public surface ----
/** The audio sinks and sources currently in the graph. */
List<Node> nodes() {
synchronized (nodes) {
return new ArrayList<>(nodes.values());
}
}
/** The loop streams are created on; only touch it under {@link #lock()}. */
MemorySegment loop() {
return loop;
}
void lock() {
PipeWireLibrary.threadLoopLock(threadLoop);
}
void unlock() {
PipeWireLibrary.threadLoopUnlock(threadLoop);
}
/**
* Waits until the server has processed everything sent so far, so that the initial
* burst of registry events has been delivered. Call with the loop lock held.
*/
private void roundtrip() {
pendingSync = (int) PipeWireLibrary.invoke(coreSync,
PipeWireLibrary.interfaceSelf(core), PipeWireLibrary.PW_ID_CORE, pendingSync + 1);
syncDone = false;
while (!syncDone) {
if (PipeWireLibrary.threadLoopTimedWait(threadLoop, ROUNDTRIP_TIMEOUT_SECONDS) != 0) break;
}
}
private void close() {
if (threadLoop != null) PipeWireLibrary.threadLoopStop(threadLoop);
if (registry != null) PipeWireLibrary.proxyDestroy(registry);
if (core != null) PipeWireLibrary.coreDisconnect(core);
if (context != null) PipeWireLibrary.contextDestroy(context);
if (threadLoop != null) PipeWireLibrary.threadLoopDestroy(threadLoop);
}
private static MemorySegment nonNull(MemorySegment pointer, String what) {
if (pointer == null || pointer.address() == 0) {
throw new IllegalStateException(what + " failed");
}
return pointer;
}
}

222
ts3-client/desktop/src/main/java/com/ts3client/audio/desktop/pipewire/PipeWireStream.java Normal file
View File

@@ -0,0 +1,222 @@
package com.ts3client.audio.desktop.pipewire;
import java.lang.foreign.Arena;
import java.lang.foreign.FunctionDescriptor;
import java.lang.foreign.MemorySegment;
import java.lang.invoke.MethodHandle;
import java.lang.invoke.MethodHandles;
import java.lang.invoke.MethodType;
import java.util.LinkedHashMap;
import java.util.Map;
import static com.ts3client.audio.desktop.pipewire.PipeWireLibrary.INT;
import static com.ts3client.audio.desktop.pipewire.PipeWireLibrary.PTR;
/**
* One {@code pw_stream} plus the ring buffer that decouples it from the application.
*
* <p>PipeWire is pull-based: it calls {@code process} on the loop thread whenever a
* buffer can be filled (playback) or has arrived (capture). The voice pipeline, on the
* other hand, wants to block on a line. {@link PcmRing} bridges the two — see there for
* what happens when the two sides drift apart.
*
* <p>{@code PW_STREAM_FLAG_RT_PROCESS} is deliberately not used: it would run the
* callback on a real-time thread, where a garbage collection pause would be an xrun.
* Without it PipeWire schedules the callback on the loop thread instead, which is a
* normal thread and can safely run Java code.
*/
abstract class PipeWireStream {
/** Ring capacity, as a multiple of the 20 ms frame the pipeline works in. */
private static final int RING_FRAMES = 20;
private static final int QUANTUM = 960; // 20 ms at 48 kHz
private final PipeWireSession session;
private final Arena arena = Arena.ofShared();
private final int channels;
private final int rate;
private final int stride;
protected final PcmRing ring;
private volatile MemorySegment stream;
private volatile boolean started;
private volatile boolean closed;
PipeWireStream(PipeWireSession session, String name, String targetNode,
int rate, int channels, boolean capture) {
this.session = session;
this.rate = rate;
this.channels = channels;
this.stride = channels * 2; // S16, interleaved
this.ring = new PcmRing(QUANTUM * stride * RING_FRAMES);
session.lock();
try {
stream = PipeWireLibrary.streamNewSimple(session.loop(), arena.allocateFrom(name),
properties(name, targetNode, capture), events(), MemorySegment.NULL);
if (stream.address() == 0) {
throw new IllegalStateException("pw_stream_new_simple failed");
}
MemorySegment params = arena.allocate(PTR, 1);
params.set(PTR, 0, SpaPod.audioFormat(arena, rate, channels));
int flags = PipeWireLibrary.PW_STREAM_FLAG_AUTOCONNECT
| PipeWireLibrary.PW_STREAM_FLAG_MAP_BUFFERS
| PipeWireLibrary.PW_STREAM_FLAG_INACTIVE;
int result = PipeWireLibrary.streamConnect(stream,
capture ? PipeWireLibrary.PW_DIRECTION_INPUT : PipeWireLibrary.PW_DIRECTION_OUTPUT,
PipeWireLibrary.PW_ID_ANY, flags, params, 1);
if (result < 0) {
throw new IllegalStateException("pw_stream_connect failed: " + result);
}
} catch (Throwable t) {
session.unlock();
close();
throw t;
}
session.unlock();
}
/**
* Properties the server uses to place and label the stream: the target device, the
* name shown in volume mixers, and the quantum we would like to be called with.
*/
private MemorySegment properties(String name, String targetNode, boolean capture) {
Map<String, String> properties = new LinkedHashMap<>();
properties.put(PipeWireLibrary.KEY_MEDIA_TYPE, "Audio");
properties.put(PipeWireLibrary.KEY_MEDIA_CATEGORY, capture ? "Capture" : "Playback");
properties.put(PipeWireLibrary.KEY_MEDIA_ROLE, "Communication");
properties.put(PipeWireLibrary.KEY_APP_NAME, "TS3J");
properties.put(PipeWireLibrary.KEY_NODE_NAME, name);
properties.put(PipeWireLibrary.KEY_NODE_LATENCY, QUANTUM + "/" + rate);
properties.put(PipeWireLibrary.KEY_NODE_RATE, "1/" + rate);
if (targetNode != null && !targetNode.isEmpty()) {
properties.put(PipeWireLibrary.KEY_TARGET_OBJECT, targetNode);
}
return PipeWireLibrary.propertiesNewDict(dict(properties));
}
/** Builds a temporary {@code struct spa_dict}; {@code pw_properties_new_dict} copies it. */
private MemorySegment dict(Map<String, String> properties) {
MemorySegment items = arena.allocate(
PipeWireLibrary.SPA_DICT_ITEM_SIZE * properties.size(), 8);
int index = 0;
for (Map.Entry<String, String> entry : properties.entrySet()) {
long offset = PipeWireLibrary.SPA_DICT_ITEM_SIZE * index++;
items.set(PTR, offset, arena.allocateFrom(entry.getKey()));
items.set(PTR, offset + PipeWireLibrary.SPA_DICT_ITEM_VALUE,
arena.allocateFrom(entry.getValue()));
}
MemorySegment dict = arena.allocate(PipeWireLibrary.SPA_DICT_SIZE, 8);
dict.set(INT, PipeWireLibrary.SPA_DICT_N_ITEMS, properties.size());
dict.set(PTR, PipeWireLibrary.SPA_DICT_ITEMS, items);
return dict;
}
private MemorySegment events() {
MemorySegment events = arena.allocate(PipeWireLibrary.STREAM_EVENTS_SIZEOF, 8);
events.set(INT, 0, PipeWireLibrary.PW_VERSION_STREAM_EVENTS);
events.set(PTR, PipeWireLibrary.STREAM_EVENTS_PROCESS,
upcall("onProcess", MethodType.methodType(void.class, MemorySegment.class),
FunctionDescriptor.ofVoid(PTR)));
events.set(PTR, PipeWireLibrary.STREAM_EVENTS_STATE_CHANGED,
upcall("onStateChanged", MethodType.methodType(void.class, MemorySegment.class,
int.class, int.class, MemorySegment.class),
FunctionDescriptor.ofVoid(PTR, INT, INT, PTR)));
return events;
}
private MemorySegment upcall(String method, MethodType type, FunctionDescriptor descriptor) {
try {
MethodHandle handle = MethodHandles.lookup().bind(this, method, type);
return PipeWireLibrary.LINKER.upcallStub(handle, descriptor, arena);
} catch (ReflectiveOperationException e) {
throw new IllegalStateException("cannot bind PipeWire callback " + method, e);
}
}
// ---- callbacks, invoked on the loop thread with the loop lock held ----
@SuppressWarnings("unused") // bound as an upcall
private void onStateChanged(MemorySegment data, int previous, int current, MemorySegment message) {
// A stream that errored will never be scheduled again, so unblock the pipeline
// instead of letting it wait for audio that cannot arrive.
if (current == PipeWireLibrary.PW_STREAM_STATE_ERROR) ring.close();
}
@SuppressWarnings("unused") // bound as an upcall
private void onProcess(MemorySegment data) {
try {
MemorySegment pwBuffer = PipeWireLibrary.streamDequeueBuffer(stream);
if (pwBuffer.address() == 0) return; // out of buffers; skip a cycle
try {
MemorySegment header = PipeWireLibrary.at(pwBuffer, PipeWireLibrary.PW_BUFFER_SIZEOF);
MemorySegment spaBuffer = PipeWireLibrary.at(
header.get(PTR, PipeWireLibrary.PW_BUFFER_BUFFER), PipeWireLibrary.SPA_BUFFER_SIZEOF);
if (spaBuffer.get(INT, PipeWireLibrary.SPA_BUFFER_N_DATAS) < 1) return;
MemorySegment datum = PipeWireLibrary.at(
spaBuffer.get(PTR, PipeWireLibrary.SPA_BUFFER_DATAS), PipeWireLibrary.SPA_DATA_SIZEOF);
MemorySegment audio = datum.get(PTR, PipeWireLibrary.SPA_DATA_DATA);
if (audio.address() == 0) return; // not mapped (e.g. a DmaBuf)
int maxSize = datum.get(INT, PipeWireLibrary.SPA_DATA_MAXSIZE);
MemorySegment chunk = PipeWireLibrary.at(
datum.get(PTR, PipeWireLibrary.SPA_DATA_CHUNK), PipeWireLibrary.SPA_CHUNK_SIZEOF);
process(header, PipeWireLibrary.at(audio, maxSize), chunk, maxSize);
} finally {
PipeWireLibrary.streamQueueBuffer(stream, pwBuffer);
}
} catch (Throwable ignored) {
// An upcall must not throw: drop the cycle rather than take the JVM down.
}
}
/**
* Moves one buffer between {@code audio} and the ring.
*
* @param header the {@code struct pw_buffer}, for {@code requested} and {@code size}
* @param audio the mapped payload, {@code maxSize} bytes long
* @param chunk the {@code struct spa_chunk} describing the valid part of {@code audio}
*/
abstract void process(MemorySegment header, MemorySegment audio, MemorySegment chunk, int maxSize);
// ---- lifecycle ----
public int channels() {
return channels;
}
final int stride() {
return stride;
}
public synchronized void start() {
if (started || closed) return;
started = true;
session.lock();
try {
PipeWireLibrary.streamSetActive(stream, true);
} finally {
session.unlock();
}
}
public synchronized void close() {
if (closed) return;
closed = true;
ring.close();
if (stream != null && stream.address() != 0) {
session.lock();
try {
PipeWireLibrary.streamDestroy(stream); // disconnects and drops the listeners
} finally {
session.unlock();
}
stream = null;
}
// Only safe once the loop can no longer reach the upcall stubs, i.e. after destroy.
arena.close();
}
}

161
ts3-client/desktop/src/main/java/com/ts3client/audio/desktop/pipewire/SpaPod.java Normal file
View File

@@ -0,0 +1,161 @@
package com.ts3client.audio.desktop.pipewire;
import java.lang.foreign.MemorySegment;
import java.lang.foreign.SegmentAllocator;
import java.lang.foreign.ValueLayout;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
/**
* Builder for the SPA "plain old data" objects PipeWire uses to describe formats.
*
* <p>A POD is a length-prefixed tree: every value is an 8-byte header
* ({@code uint32_t size; uint32_t type;}) followed by {@code size} bytes of body, padded
* out to a multiple of 8. An object adds {@code uint32_t type; uint32_t id;} in front of
* its properties, and each property is {@code uint32_t key; uint32_t flags;} followed by
* one value POD.
*
* <p>libpipewire only offers {@code spa_pod_builder} as inline C, so the bytes are
* assembled here instead. {@code SpaPodTest} pins the output against the bytes the real
* {@code spa_format_audio_raw_build()} produces.
*/
final class SpaPod {
// ---- enum spa_type ----
private static final int SPA_TYPE_ID = 3;
private static final int SPA_TYPE_INT = 4;
private static final int SPA_TYPE_ARRAY = 13;
private static final int SPA_TYPE_OBJECT = 15;
private static final int SPA_TYPE_OBJECT_FORMAT = 0x0004_0003;
// ---- enum spa_param_type ----
static final int SPA_PARAM_ENUM_FORMAT = 3;
// ---- enum spa_format (object property keys) ----
private static final int SPA_FORMAT_MEDIA_TYPE = 1;
private static final int SPA_FORMAT_MEDIA_SUBTYPE = 2;
private static final int SPA_FORMAT_AUDIO_FORMAT = 0x0001_0001;
private static final int SPA_FORMAT_AUDIO_RATE = 0x0001_0003;
private static final int SPA_FORMAT_AUDIO_CHANNELS = 0x0001_0004;
private static final int SPA_FORMAT_AUDIO_POSITION = 0x0001_0005;
private static final int SPA_MEDIA_TYPE_AUDIO = 1;
private static final int SPA_MEDIA_SUBTYPE_RAW = 1;
/** {@code SPA_AUDIO_FORMAT_S16_LE} — interleaved signed 16-bit, what our pipeline uses. */
static final int SPA_AUDIO_FORMAT_S16_LE = 0x103;
// ---- enum spa_audio_channel ----
private static final int SPA_AUDIO_CHANNEL_UNKNOWN = 0;
private static final int SPA_AUDIO_CHANNEL_MONO = 2;
private static final int SPA_AUDIO_CHANNEL_FL = 3;
private static final int SPA_AUDIO_CHANNEL_FR = 4;
private final ByteBuffer buffer;
private SpaPod(int capacity) {
this.buffer = ByteBuffer.allocate(capacity).order(ByteOrder.LITTLE_ENDIAN);
}
/**
* Builds the single {@code EnumFormat} parameter a stream offers when connecting:
* interleaved S16 audio at one fixed rate and channel count. Offering exactly one
* format lets the server put its own converter in front of us, so the negotiated
* format is always the one asked for here.
*/
static MemorySegment audioFormat(SegmentAllocator allocator, int rate, int channels) {
SpaPod pod = new SpaPod(256);
pod.beginObject(SPA_TYPE_OBJECT_FORMAT, SPA_PARAM_ENUM_FORMAT);
pod.id(SPA_FORMAT_MEDIA_TYPE, SPA_MEDIA_TYPE_AUDIO);
pod.id(SPA_FORMAT_MEDIA_SUBTYPE, SPA_MEDIA_SUBTYPE_RAW);
pod.id(SPA_FORMAT_AUDIO_FORMAT, SPA_AUDIO_FORMAT_S16_LE);
pod.integer(SPA_FORMAT_AUDIO_RATE, rate);
pod.integer(SPA_FORMAT_AUDIO_CHANNELS, channels);
pod.idArray(SPA_FORMAT_AUDIO_POSITION, positions(channels));
pod.endObject();
return pod.toSegment(allocator);
}
/** Channel map for the layouts we use; anything wider is left for the server to assign. */
private static int[] positions(int channels) {
return switch (channels) {
case 1 -> new int[]{SPA_AUDIO_CHANNEL_MONO};
case 2 -> new int[]{SPA_AUDIO_CHANNEL_FL, SPA_AUDIO_CHANNEL_FR};
default -> {
int[] unknown = new int[channels];
java.util.Arrays.fill(unknown, SPA_AUDIO_CHANNEL_UNKNOWN);
yield unknown;
}
};
}
// ---- writing ----
/** Offset of the size field of the object being built, patched by {@link #endObject}. */
private int objectSizeOffset = -1;
private void beginObject(int type, int id) {
objectSizeOffset = buffer.position();
buffer.putInt(0); // size, patched in endObject
buffer.putInt(SPA_TYPE_OBJECT);
buffer.putInt(type);
buffer.putInt(id);
}
private void endObject() {
int bodyStart = objectSizeOffset + 8;
buffer.putInt(objectSizeOffset, buffer.position() - bodyStart);
}
private void id(int key, int value) {
property(key, SPA_TYPE_ID, 4);
buffer.putInt(value);
pad();
}
private void integer(int key, int value) {
property(key, SPA_TYPE_INT, 4);
buffer.putInt(value);
pad();
}
/** An array POD: a child header describing one element, then the elements. */
private void idArray(int key, int[] values) {
property(key, SPA_TYPE_ARRAY, 8 + 4 * values.length);
buffer.putInt(4);
buffer.putInt(SPA_TYPE_ID);
for (int value : values) {
buffer.putInt(value);
}
pad();
}
/** Writes {@code key/flags} plus the value header; the caller writes the body. */
private void property(int key, int type, int size) {
buffer.putInt(key);
buffer.putInt(0); // flags
buffer.putInt(size);
buffer.putInt(type);
}
/** Bodies are padded to 8 bytes; the padding is not counted in the declared size. */
private void pad() {
while ((buffer.position() & 7) != 0) {
buffer.put((byte) 0);
}
}
private MemorySegment toSegment(SegmentAllocator allocator) {
MemorySegment segment = allocator.allocate(buffer.position(), 8);
MemorySegment.copy(MemorySegment.ofBuffer(buffer.flip()), 0, segment, 0, segment.byteSize());
return segment;
}
/** The bytes this builder produced, for tests. */
static byte[] audioFormatBytes(int rate, int channels) {
try (java.lang.foreign.Arena arena = java.lang.foreign.Arena.ofConfined()) {
MemorySegment pod = audioFormat(arena, rate, channels);
return pod.toArray(ValueLayout.JAVA_BYTE);
}
}
}

44
ts3-client/desktop/src/test/java/com/ts3client/audio/desktop/pipewire/SpaPodTest.java Normal file
View File

@@ -0,0 +1,44 @@
package com.ts3client.audio.desktop.pipewire;
import org.junit.jupiter.api.Test;
import java.util.HexFormat;
import static org.junit.jupiter.api.Assertions.assertEquals;
/**
* Pins {@link SpaPod} to the bytes libpipewire itself produces.
*
* <p>The expected values are the output of {@code spa_format_audio_raw_build()} (pipewire
* 1.4, {@code SPA_AUDIO_FORMAT_S16_LE} at 48 kHz), captured from a C program linked
* against the real library. Since the POD layout is what the server parses, a mismatch
* here is a stream that will fail to negotiate.
*/
class SpaPodTest {
private static final String MONO_48K =
"a00000000f000000030004000300000001000000000000000400000003000000"
+ "0100000000000000020000000000000004000000030000000100000000000000"
+ "0100010000000000040000000300000003010000000000000300010000000000"
+ "040000000400000080bb00000000000004000100000000000400000004000000"
+ "010000000000000005000100000000000c0000000d0000000400000003000000"
+ "0200000000000000";
private static final String STEREO_48K =
"a00000000f000000030004000300000001000000000000000400000003000000"
+ "0100000000000000020000000000000004000000030000000100000000000000"
+ "0100010000000000040000000300000003010000000000000300010000000000"
+ "040000000400000080bb00000000000004000100000000000400000004000000"
+ "02000000000000000500010000000000100000000d0000000400000003000000"
+ "0300000004000000";
@Test
void buildsTheFormatLibpipewireExpects() {
assertEquals(MONO_48K, hex(SpaPod.audioFormatBytes(48_000, 1)));
assertEquals(STEREO_48K, hex(SpaPod.audioFormatBytes(48_000, 2)));
}
private static String hex(byte[] bytes) {
return HexFormat.of().formatHex(bytes);
}
}