diff --git a/bundles/org.openhab.core.audio/src/main/java/org/openhab/core/audio/internal/javasound/JavaSoundAudioSource.java b/bundles/org.openhab.core.audio/src/main/java/org/openhab/core/audio/internal/javasound/JavaSoundAudioSource.java index cabb2ce39..792beb380 100644 --- a/bundles/org.openhab.core.audio/src/main/java/org/openhab/core/audio/internal/javasound/JavaSoundAudioSource.java +++ b/bundles/org.openhab.core.audio/src/main/java/org/openhab/core/audio/internal/javasound/JavaSoundAudioSource.java @@ -12,9 +12,16 @@ */ package org.openhab.core.audio.internal.javasound; -import java.util.HashSet; +import java.io.IOException; +import java.io.InputStream; +import java.io.InterruptedIOException; +import java.io.PipedInputStream; +import java.io.PipedOutputStream; import java.util.Locale; import java.util.Set; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.Future; +import java.util.concurrent.ScheduledExecutorService; import javax.sound.sampled.AudioSystem; import javax.sound.sampled.TargetDataLine; @@ -25,7 +32,10 @@ import org.openhab.core.audio.AudioException; import org.openhab.core.audio.AudioFormat; import org.openhab.core.audio.AudioSource; import org.openhab.core.audio.AudioStream; +import org.openhab.core.common.ThreadPoolManager; import org.osgi.service.component.annotations.Component; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * This is an AudioSource from an input channel of the host. @@ -39,6 +49,8 @@ import org.osgi.service.component.annotations.Component; @Component(service = AudioSource.class, immediate = true) public class JavaSoundAudioSource implements AudioSource { + private final Logger logger = LoggerFactory.getLogger(JavaSoundAudioSource.class); + /** * Java Sound audio format */ @@ -61,14 +73,22 @@ public class JavaSoundAudioSource implements AudioSource { private @Nullable TargetDataLine microphone; /** - * Set for control microphone close on Windows OS + * Set for control microphone sharing on Windows OS */ - private final Set openStreamRefs = new HashSet<>(); + private final ConcurrentLinkedQueue openStreamRefs = new ConcurrentLinkedQueue<>(); + + /** + * Task for writing microphone data to each of the open sources on Windows OS + */ + private @Nullable Future pipeWriteTask; + + private final ScheduledExecutorService executor; /** * Constructs a JavaSoundAudioSource */ public JavaSoundAudioSource() { + executor = ThreadPoolManager.getScheduledPool("OH-core-javasound-source"); } private TargetDataLine initMicrophone(javax.sound.sampled.AudioFormat format) throws AudioException { @@ -82,34 +102,107 @@ public class JavaSoundAudioSource implements AudioSource { } @Override - public synchronized AudioStream getInputStream(AudioFormat expectedFormat) throws AudioException { + public AudioStream getInputStream(AudioFormat expectedFormat) throws AudioException { if (!expectedFormat.isCompatible(audioFormat)) { throw new AudioException("Cannot produce streams in format " + expectedFormat); } // on OSs other than windows we can open multiple lines for the microphone if (!windowsOS) { - return new JavaSoundInputStream(initMicrophone(format), audioFormat); + TargetDataLine microphone = initMicrophone(format); + var inputStream = new JavaSoundInputStream((InputStream) microphone, audioFormat); + microphone.start(); + return inputStream; } // on Windows OS we share the microphone line - var ref = new Object(); - TargetDataLine microphoneDataLine; synchronized (openStreamRefs) { - microphoneDataLine = this.microphone; - if (microphoneDataLine == null) { - microphoneDataLine = initMicrophone(format); - this.microphone = microphoneDataLine; + TargetDataLine microphone = this.microphone; + if (microphone == null) { + microphone = initMicrophone(format); + this.microphone = microphone; } - openStreamRefs.add(ref); + var pipedOutputStream = new PipedOutputStream(); + PipedInputStream pipedInputStream; + try { + pipedInputStream = new PipedInputStream(pipedOutputStream, 1024 * 10) { + @Override + public void close() throws IOException { + unregisterPipe(pipedOutputStream); + super.close(); + } + }; + } catch (IOException ie) { + throw new AudioException("Cannot open stream pipe: " + ie.getMessage()); + } + openStreamRefs.add(pipedOutputStream); + var inputStream = new JavaSoundInputStream(pipedInputStream, audioFormat); + microphone.start(); + startPipeWrite(); + return inputStream; } - return new JavaSoundInputStream(microphoneDataLine, audioFormat, () -> { - synchronized (openStreamRefs) { - var microphone = this.microphone; - if (openStreamRefs.remove(ref) && openStreamRefs.isEmpty() && microphone != null) { + } + + private void startPipeWrite() { + if (this.pipeWriteTask == null) { + this.pipeWriteTask = executor.submit(() -> { + int lengthRead; + byte[] buffer = new byte[1024]; + while (!openStreamRefs.isEmpty()) { + TargetDataLine stream = this.microphone; + if (stream != null) { + try { + lengthRead = stream.read(buffer, 0, buffer.length); + for (PipedOutputStream output : openStreamRefs) { + try { + output.write(buffer, 0, lengthRead); + if (openStreamRefs.contains(output)) { + output.flush(); + } + } catch (IOException e) { + if (e instanceof InterruptedIOException && openStreamRefs.isEmpty()) { + // task has been ended while writing + return; + } + logger.warn("IOException while writing to source pipe: {}", e.getMessage()); + } catch (RuntimeException e) { + logger.warn("RuntimeException while writing to source pipe: {}", e.getMessage()); + } + } + } catch (RuntimeException e) { + logger.warn("RuntimeException while reading from JavaSound source: {}", e.getMessage()); + } + } else { + logger.warn("Unable access to microphone stream"); + } + } + this.pipeWriteTask = null; + }); + } + } + + private void unregisterPipe(PipedOutputStream pipedOutputStream) { + synchronized (openStreamRefs) { + openStreamRefs.remove(pipedOutputStream); + try { + Thread.sleep(0); + } catch (InterruptedException ignored) { + } + if (openStreamRefs.isEmpty()) { + Future pipeWriteTask = this.pipeWriteTask; + if (pipeWriteTask != null) { + pipeWriteTask.cancel(true); + this.pipeWriteTask = null; + } + TargetDataLine microphone = this.microphone; + if (microphone != null) { microphone.close(); this.microphone = null; } } - }); + try { + pipedOutputStream.close(); + } catch (IOException ignored) { + } + } } @Override diff --git a/bundles/org.openhab.core.audio/src/main/java/org/openhab/core/audio/internal/javasound/JavaSoundInputStream.java b/bundles/org.openhab.core.audio/src/main/java/org/openhab/core/audio/internal/javasound/JavaSoundInputStream.java index b0c3e02d5..e1a119fce 100644 --- a/bundles/org.openhab.core.audio/src/main/java/org/openhab/core/audio/internal/javasound/JavaSoundInputStream.java +++ b/bundles/org.openhab.core.audio/src/main/java/org/openhab/core/audio/internal/javasound/JavaSoundInputStream.java @@ -13,8 +13,7 @@ package org.openhab.core.audio.internal.javasound; import java.io.IOException; - -import javax.sound.sampled.TargetDataLine; +import java.io.InputStream; import org.eclipse.jdt.annotation.NonNullByDefault; import org.eclipse.jdt.annotation.Nullable; @@ -31,32 +30,19 @@ import org.openhab.core.audio.AudioStream; public class JavaSoundInputStream extends AudioStream { /** - * TargetDataLine for the input + * InputStream for the input */ - private final TargetDataLine input; + private final InputStream input; private final AudioFormat format; - private final @Nullable JavaSoundInputStreamCloseHandler closeHandler; - - /** - * Constructs a JavaSoundInputStream with the passed input - * - * @param input The mic which data is pulled from - */ - public JavaSoundInputStream(TargetDataLine input, AudioFormat format) { - this(input, format, null); - } /** * Constructs a JavaSoundInputStream with the passed input and a close handler. * * @param input The mic which data is pulled from */ - public JavaSoundInputStream(TargetDataLine input, AudioFormat format, - @Nullable JavaSoundInputStreamCloseHandler closeHandler) { + public JavaSoundInputStream(InputStream input, AudioFormat format) { this.format = format; - this.closeHandler = closeHandler; this.input = input; - this.input.start(); } @Override @@ -86,20 +72,11 @@ public class JavaSoundInputStream extends AudioStream { @Override public void close() throws IOException { - var closeHandler = this.closeHandler; - if (closeHandler != null) { - closeHandler.onStreamClosed(); - } else { - input.close(); - } + input.close(); } @Override public AudioFormat getFormat() { return format; } - - public interface JavaSoundInputStreamCloseHandler { - void onStreamClosed(); - } }